From 8c0fcfee8391fe9fcb54623905c91ab4245ff9a8 Mon Sep 17 00:00:00 2001 From: LittleGuo Date: Wed, 14 Oct 2026 19:07:35 +0800 Subject: [PATCH] =?UTF-8?q?chore:=20=E5=88=A0=E9=99=A4=205=20=E4=B8=AA?= =?UTF-8?q?=E9=81=97=E7=95=99=E8=B0=83=E8=AF=95=E8=84=9A=E6=9C=AC=E5=B9=B6?= =?UTF-8?q?=E5=90=8C=E6=AD=A5=E6=96=87=E6=A1=A3=E5=BC=95=E7=94=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 删除 agent_loop.py / dwg_review_agent.py / countOfCircle.py （硬编码 DeepSeek API Key）与 test.py / test_autoagent.py （重复实现；且 test_autoagent.py 名字与 pytest 的 test_*.py 收集规则冲突） - CODEBUDDY.md 测试章节改为「pytest 快层 + verify_* 慢层」 - review_checklist.json 描述改指 dwg_review 包 - ai.memory 同步引用与历史说明 安全：Key 已随提交 bdd77af 进入 git 历史并推到远程，删除文件不能清除历史， 需去 DeepSeek 控制台轮换该 Key（已在 ai.memory 备案）。 --- CODEBUDDY.md | 28 ++- agent_loop.py | 163 ------------- ai.memory/project_dwg_review_agent.md | 9 +- ai.memory/project_review_agent_impl.md | 7 +- ai.memory/reference_environment.md | 24 +- countOfCircle.py | 127 ---------- dwg_review_agent.py | 310 ------------------------- review_checklist.json | 2 +- test.py | 41 ---- test_autoagent.py | 43 ---- 10 files changed, 53 insertions(+), 701 deletions(-) delete mode 100644 agent_loop.py delete mode 100644 countOfCircle.py delete mode 100644 dwg_review_agent.py delete mode 100644 test.py delete mode 100644 test_autoagent.py diff --git a/CODEBUDDY.md b/CODEBUDDY.md index 8316754..8f15c80 100644 --- a/CODEBUDDY.md +++ b/CODEBUDDY.md @@ -27,16 +27,32 @@ cmake --build --preset R243 --config Release ### 测试 -工程内**没有单元测试框架**。所谓“测试”是根目录下的 Python 脚本，它们是 MCP 客户端的联调工具，**必须先启动已加载插件的 CAD，并确认 8080 端口在监听**： +分两层，**互不替代**： + +**快层 —— pytest 离线单元测试**（`tests/`，无需 CAD、不联网、秒级）： ```bash -python test.py # 最小示例：截图 -> 发给 DeepSeek 视觉模型 -> 回答 -python test_autoagent.py # 仅拉取一次视口截图并保存为 test_screenshot.png -python countOfCircle.py # 伪造一次 zoom_window_normalized 调用，验证 TCP 链路 -python agent_loop.py # 完整 Agent 循环（多回合 截图/缩放/结论），需 openai 包与 API Key +pip install -r requirements-dev.txt # 仅 pytest +python -m pytest # 期望全绿 ``` -脚本通过 TCP 连接 `127.0.0.1:8080` 收发 JSON-RPC。`agent_loop.py` / `test.py` 中硬编码了 DeepSeek API Key 与模型名，属于调试脚本。 +全部使用假 LLM + 假 MCP 基座（`tests/fakes.py`），锁定 `dwg_review` 流水线既有行为。 +配置见 `pytest.ini`（`testpaths = tests`）与根 `conftest.py`（把仓库根加入 `sys.path`， +并提供 `RENDER_WAIT=0` / `REPORT_DIR=tmp` 两个 fixture）。**注意**：`dwg_review.config` +的常量在导入期定值，测试改 `os.environ` 无效，须 `monkeypatch.setattr(config, ...)`。 + +**慢层 —— 端到端冒烟脚本**（仓库根 `verify_*.py`，**必须先启动已加载插件的 CAD，并确认 +8080 端口在监听**）： + +```bash +python verify_mcp_frame.py # 基座最小集：分帧 / id / isError 映射 +python verify_mcp_tcp_hardening.py # TCP 收发路径加固（超长帧 / 粘包 / 半关闭） +python verify_mcp_fault_injection.py # 工具异常保护（类型错误不再崩 CAD） +python verify_stdio_bridge.py # stdio ↔ TCP 桥（协议协商 / 转发 / 通知） +python verify_annotate.py # 批注回写链路 +``` + +这些脚本通过 TCP 连接 `127.0.0.1:8080` 收发 JSON-RPC。 ## 架构 diff --git a/agent_loop.py b/agent_loop.py deleted file mode 100644 index 0e66e39..0000000 --- a/agent_loop.py +++ /dev/null @@ -1,163 +0,0 @@ -import socket -import json -import time -from openai import OpenAI - -# 初始化 DeepSeek 客户端 -client = OpenAI( - api_key="sk-420190f448fe41158c4e2ccff90e35ce", - base_url="https://api.deepseek.com" -) -MODEL_NAME = "deepseek-chat" - -def fetch_and_translate_tools(): - """向 CAD 基座发送 tools/list，并转换为大模型认识的格式""" - try: - cad_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - cad_socket.connect(("127.0.0.1", 8080)) - - # 1. 向 MCP Server 发送标准的 tools/list 请求 - req = { - "jsonrpc": "2.0", - "id": 100, - "method": "tools/list" - } - cad_socket.sendall(json.dumps(req).encode('utf-8')) - - response_data = cad_socket.recv(8192) - cad_socket.close() - - mcp_response = json.loads(response_data.decode('utf-8')) - - if "result" not in mcp_response or "tools" not in mcp_response["result"]: - print("[Error] 无法从 CAD 获取工具列表") - return [] - - llm_tools = [] - # 2. 遍历 CAD 返回的工具，将其翻译为 DeepSeek/OpenAI 格式 - for mcp_tool in mcp_response["result"]["tools"]: - llm_tools.append({ - "type": "function", - "function": { - "name": mcp_tool["name"], - "description": mcp_tool["description"], - # 核心转换：MCP 的 inputSchema 等价于 LLM 的 parameters - "parameters": mcp_tool["inputSchema"] - } - }) - - print(f"[Init] 成功从 CAD 动态加载了 {len(llm_tools)} 个工具！") - return llm_tools - - except Exception as e: - print(f"[Error] 获取工具列表失败: {e}") - return [] - -def call_cad_mcp_server(tool_name, arguments): - """底层 TCP 通信保持不变""" - try: - cad_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - cad_socket.connect(("127.0.0.1", 8080)) - req = { - "jsonrpc": "2.0", "id": 1, "method": "tools/call", - "params": { "name": tool_name, "arguments": arguments } - } - cad_socket.sendall(json.dumps(req).encode('utf-8')) - response_data = b"" - while True: - chunk = cad_socket.recv(8192) - response_data += chunk - if len(chunk) < 8192: break - cad_socket.close() - return json.loads(response_data.decode('utf-8')) - except Exception as e: - print(f"[Error] 连接 CAD 基座失败: {e}") - return None - -def run_agent(user_prompt): - print("=== 🚀 CAD 具身智能 Agent 启动 ===") - - # 动态向 CAD 请求可用工具，彻底解耦！ - tools = fetch_and_translate_tools() - - if not tools: - print("没有可用的工具，Agent 退出。") - return - - # 核心心智设定：教它怎么形成视觉闭环 - system_prompt = """你是一个高阶 AutoCAD 视觉检查专家。 -请严格遵循以下工作流： -1. 第一步永远是调用 `get_viewport_screenshot` 观察当前画面。 -2. 仔细评估图像。如果目标物体（如图元、文字）太小导致无法确信细节，请调用 `zoom_window_normalized` 放大该局部区域。 -3. 【关键指令】：每次执行缩放操作（zoom）后，你必须在下一回合再次调用 `get_viewport_screenshot` 获取放大后的新截图！ -4. 反复观察和放大，直到你 100% 看清细节，再用自然语言向用户输出最终结论。""" - - messages = [ - {"role": "system", "content": system_prompt}, - {"role": "user", "content": user_prompt} - ] - - MAX_TURNS = 8 # 熔断机制：最多允许思考 8 个回合，防止无限套娃 - - for turn in range(MAX_TURNS): - print(f"\n--- [第 {turn + 1} 回合] 思考中 ---") - - response = client.chat.completions.create( - model=MODEL_NAME, - messages=messages, - tools=tools - ) - - assistant_message = response.choices[0].message - messages.append(assistant_message) # 记录思维路径 - - # 1. 检查是否需要执行物理动作 - if assistant_message.tool_calls: - for tool_call in assistant_message.tool_calls: - tool_name = tool_call.function.name - args = json.loads(tool_call.function.arguments) - - print(f"[动作决定] ⚡ 调用工具: {tool_name}") - if args: print(f" 参数: {args}") - - # 执行 CAD 通信 - cad_res = call_cad_mcp_server(tool_name, args) - time.sleep(0.5) # 给予 CAD 渲染刷新窗口的时间缓冲 - - # 处理执行结果，并准备发回给 LLM 的观测报告 - tool_result_content = [] - - if cad_res and "result" in cad_res: - for item in cad_res["result"]["content"]: - if item["type"] == "text": - tool_result_content.append({"type": "text", "text": item["text"]}) - elif item["type"] == "image": - base64_img = item["data"] - mime = item.get("mimeType", "image/png") - tool_result_content.append({ - "type": "image_url", - "image_url": {"url": f"data:{mime};base64,{base64_img}"} - }) - print("[观测反馈] 📸 已将最新 CAD 屏幕画面传回视觉中枢。") - else: - tool_result_content.append({"type": "text", "text": "CAD 工具执行失败或无响应。"}) - - # 必须向大模型提交 Tool 返回结果 - messages.append({ - "role": "tool", - "tool_call_id": tool_call.id, - "content": tool_result_content - }) - - # 2. 如果不调工具，说明任务已完成，输出最终自然语言结论 - else: - print("\n==================================") - print(f"🎯 [最终结论]:\n{assistant_message.content}") - print("==================================") - break - - else: - print("\n[警告] 达到最大思考回合数限制，Agent 强行中止。") - -if __name__ == "__main__": - run_agent("请仔细检查当前图纸，告诉我图纸中心那些小圆的内部，是否还包含了更小的同心圆？如果看不清，请务必放大确认。") \ No newline at end of file diff --git a/ai.memory/project_dwg_review_agent.md b/ai.memory/project_dwg_review_agent.md index f2b598c..aa706bf 100644 --- a/ai.memory/project_dwg_review_agent.md +++ b/ai.memory/project_dwg_review_agent.md @@ -100,8 +100,10 @@ TCP 线程只 `recv`，把请求封装成 `TaskData` 后 `PostMessage(WM_MCP_EXE ### 审查脚本与清单 -- `dwg_review_agent.py`：逐图框审查流程。`MODEL_NAME`（默认 `deepseek-flash`）、 - `CHECKLIST_PATH`（默认 `review_checklist.json`）。 +> 注：原单文件脚本 `dwg_review_agent.py` 已由 `dwg_review/` 包取代（见 +> [LLM 识图审图 Agent 架构决策](project_review_agent_impl.md)），并于 2026-10-14 删除。 + +- 逐图框审查流程：`MODEL_NAME`（默认 `deepseek-flash`）、`CHECKLIST_PATH`（默认 `review_checklist.json`）。 流程：`detect_drawing_frames` → 每框 `zoom_to_window` + `query_entities(window)` + `get_text_content(window)` + `get_viewport_screenshot` → LLM 审查 → 汇总。 - `review_checklist.json`：**可配置审查清单**，5 大类 @@ -144,6 +146,7 @@ TCP 线程只 `recv`，把请求封装成 `TaskData` 后 `PostMessage(WM_MCP_EXE ## 遗留 / 待办 -1. `dwg_review_agent.py` 仍硬编码 DeepSeek API Key（已推送远程），建议改为环境变量 `DEEPSEEK_API_KEY`。 +1. ~~`dwg_review_agent.py` 仍硬编码 DeepSeek API Key~~ —— 该脚本已于 2026-10-14 删除； + 但 Key 已随提交 `bdd77af` 进入 git 历史并推到远程，**必须去 DeepSeek 控制台轮换**。 2. `review_checklist.json` 可补充「检查方法 / 阈值」；可加跨图框统计工具；可把清单移入 MCP 侧。 3. 后期整理建筑/工业管道等审图规范并接入，让 LLM 参照规范审核。 diff --git a/ai.memory/project_review_agent_impl.md b/ai.memory/project_review_agent_impl.md index d3a8966..38392bd 100644 --- a/ai.memory/project_review_agent_impl.md +++ b/ai.memory/project_review_agent_impl.md @@ -40,7 +40,12 @@ type: project （唯一 TCP 封装：单飞锁 + 换行分帧 + 重试 + 兼容 `error`/`isError`）、`llm_client.py`、 `cad_tools.py`（动态工具表 + 只读白名单 `READ_ONLY_TOOLS`）、`checklist.py`、 `phases/{planning,evidence,review,report}.py`、`annotator.py`、`__main__.py`。 - **取代**根目录散落的 `dwg_review_agent.py` / `agent_loop.py` 等重复脚本。 + **取代**根目录散落的 `dwg_review_agent.py` / `agent_loop.py` 等重复脚本 + （这 5 个遗留脚本已于 2026-10-14 删除，见 [环境与外部资源](reference_environment.md)）。 +- `tests/`（pytest 回归网）：假 LLM + 假 MCP 基座的**离线快层**，`tests/fakes.py` 提供 + `FakeLlmClient` / `FakeMcpClient` / `make_ctx`。锁定 `dwg_review` 既有行为。 + **陷阱**：绝不构造真 `LlmClient`（无 key 抛错）、无参 `CadTools()` 或 `ReviewContext.create()` + （构造即 `list_tools()` 触网）；一律 `make_ctx()` 注入假件。 - `verify_mcp_frame.py`：基座改动验证脚本（分帧 / 大报文 / string id / 通知 / isError）。 ## 环境事实 diff --git a/ai.memory/reference_environment.md b/ai.memory/reference_environment.md index b97b48f..7a18f43 100644 --- a/ai.memory/reference_environment.md +++ b/ai.memory/reference_environment.md @@ -1,6 +1,6 @@ --- name: 环境与外部资源 -description: CadProject 的构建/运行要点、CAD SDK 路径、测试脚本等外部资源指引 +description: CadProject 的构建/运行要点、CAD SDK 路径、测试分层（pytest 快层 + verify_* 慢层）等外部资源指引 type: reference --- @@ -26,16 +26,28 @@ type: reference - 先启动 **已加载插件的 CAD**，确认 `127.0.0.1:8080` 在监听。 - 脚本通过 TCP 连接 `127.0.0.1:8080` 收发 JSON-RPC。 -## 测试脚本（根目录，非单元测试） +## 测试（两层，互不替代） + +**快层 —— pytest 离线单元测试**（`tests/`，假 LLM + 假基座，无需 CAD、不联网、秒级）： + +- `pip install -r requirements-dev.txt` → `python -m pytest`。 +- 配置：`pytest.ini`（`testpaths = tests`）、根 `conftest.py`（加 `sys.path` + `RENDER_WAIT=0` / `REPORT_DIR=tmp` fixture）。 +- 陷阱：`dwg_review.config` 常量导入期定值，改 `os.environ` 无效，须 `monkeypatch.setattr(config, ...)`。 + +**慢层 —— 端到端冒烟脚本**（根目录 `verify_*.py`，需 CAD 在线且 `127.0.0.1:8080` 监听）： | 脚本 | 作用 | | --- | --- | -| `test.py` | 截图 → 发 DeepSeek 视觉模型 → 回答 | -| `test_autoagent.py` | 拉一次视口截图存 `test_screenshot.png` | -| `countOfCircle.py` | 伪造 `zoom_window_normalized` 调用，验证 TCP 链路 | -| `agent_loop.py` | 完整 Agent 循环（多回合），需 openai 包与 API Key | +| `verify_mcp_frame.py` | 基座最小集：分帧 / id / isError 映射 | +| `verify_mcp_tcp_hardening.py` | TCP 收发路径加固（超长帧 / 粘包 / 半关闭） | +| `verify_mcp_fault_injection.py` | 工具异常保护（类型错误不再崩 CAD） | +| `verify_stdio_bridge.py` | stdio ↔ TCP 桥（协议协商 / 转发 / 通知） | | `verify_annotate.py` | 批注工具运行时验证（绘制/校验/改属性/清理）；`--keep` 保留批注 | +> 历史：根目录 `test.py` / `test_autoagent.py` / `countOfCircle.py` / `agent_loop.py` / +> `dwg_review_agent.py` 五个遗留调试脚本已于 2026-10-14 删除（前三个含硬编码 API Key， +> 后两个是重复实现）。`test_screenshot.png` 是 `test_autoagent.py` 的残留产物。 + ## 两个构建目标 | 目标 | 产物 | 作用 | diff --git a/countOfCircle.py b/countOfCircle.py deleted file mode 100644 index 6defc86..0000000 --- a/countOfCircle.py +++ /dev/null @@ -1,127 +0,0 @@ -import socket -import json -from openai import OpenAI - -# 1. 初始化 DeepSeek 客户端 -# 注意：务必确保已经执行过 pip install openai -client = OpenAI( - api_key="sk-420190f448fe41158c4e2ccff90e35ce", - base_url="https://api.deepseek.com" # 核心修改：将网关指向 DeepSeek 服务器 -) - -# 核心修改：使用 DeepSeek 的主模型。 -# (注: DeepSeek-V4/V4.1 的视觉支持已集成，具体模型名请以你 DeepSeek 后台显示的可用模型为准) -MODEL_NAME = "deepseek-flash" - -def call_cad_mcp_server(tool_name, arguments): - """封装与 AutoCAD MCP 基座的 TCP 通信""" - try: - cad_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - cad_socket.connect(("127.0.0.1", 8080)) - - req = { - "jsonrpc": "2.0", - "id": 1, - "method": "tools/call", - "params": { - "name": tool_name, - "arguments": arguments - } - } - cad_socket.sendall(json.dumps(req).encode('utf-8')) - - response_data = b"" - while True: - chunk = cad_socket.recv(8192) - response_data += chunk - if len(chunk) < 8192: - break - - cad_socket.close() - return json.loads(response_data.decode('utf-8')) - except Exception as e: - print(f"[Error] 连接 CAD 基座失败: {e}") - return None - -def main(): - print("=== DeepSeek CAD Vision Agent 启动 ===") - - # 2. 定义 MCP 工具 (Tool Calling) - tools = [ - { - "type": "function", - "function": { - "name": "get_viewport_screenshot", - "description": "获取当前 AutoCAD 视口的实时截图。当用户询问图纸上的视觉特征、数量或位置时，必须先调用此工具获取画面。" - } - } - ] - - # 3. 初始化历史 - messages = [ - {"role": "system", "content": "你是一个专业的 AutoCAD 视觉审查助手。必须先使用截图工具观察图纸，再回答用户问题。"}, - {"role": "user", "content": "请看看当前 CAD 屏幕上，我一共画了几个圆？"} - ] - - print("\n[DeepSeek] 正在思考如何完成任务...") - - # === 第一回合：DeepSeek 思考并决定调用工具 === - response = client.chat.completions.create( - model=MODEL_NAME, - messages=messages, - tools=tools - ) - - assistant_message = response.choices[0].message - messages.append(assistant_message) - - # 检查 DeepSeek 是否发起了 Tool Call - if assistant_message.tool_calls: - tool_call = assistant_message.tool_calls[0] - tool_name = tool_call.function.name - - print(f"[DeepSeek] 决定调用 CAD 工具: {tool_name}") - print(f"[CAD] 正在执行截图，请稍候...") - - # === 与 CAD 基座通信 === - cad_res = call_cad_mcp_server(tool_name, {}) - - if cad_res and "result" in cad_res: - content_array = cad_res["result"]["content"] - base64_img = "" - mime_type = "image/png" - for item in content_array: - if item["type"] == "image": - base64_img = item["data"] - mime_type = item.get("mimeType", "image/png") - - print("[CAD] 截图成功！正在将视觉数据传回大模型神经中枢...") - - # 4. 将 Base64 图片按标准多模态格式塞回历史 - messages.append({ - "role": "tool", - "tool_call_id": tool_call.id, - "content": [ - {"type": "text", "text": "这是 AutoCAD 当前界面的实时截图。请根据图片回答用户的问题。"}, - {"type": "image_url", "image_url": {"url": f"data:{mime_type};base64,{base64_img}"}} - ] - }) - - # === 第二回合：DeepSeek “看”图并回答 === - print("[DeepSeek] 正在进行视觉分析...") - final_response = client.chat.completions.create( - model=MODEL_NAME, - messages=messages - ) - - print("\n==================================") - print(f"[DeepSeek 最终回答]:\n{final_response.choices[0].message.content}") - print("==================================") - - else: - print("[Error] CAD 工具执行失败。") - else: - print(f"[DeepSeek 盲猜]: {assistant_message.content}") - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/dwg_review_agent.py b/dwg_review_agent.py deleted file mode 100644 index ed8461e..0000000 --- a/dwg_review_agent.py +++ /dev/null @@ -1,310 +0,0 @@ -""" -dwg_review_agent.py -==================== -逐图框图纸审查 Agent（联调脚本） - -流程： - 1. detect_drawing_frames 识别图纸中的所有图框 - 2. 对每个图框： - a. zoom_to_window 缩放到该图框范围 - b. query_entities 取图框内图元摘要（按 window 过滤） - c. get_text_content 取图框内文字 - d. get_viewport_screenshot 截取当前视口 - e. 交给 LLM 审查，产出该图框的审查意见 - 3. 汇总所有图框意见，生成总报告 - -依赖： - pip install openai -环境变量： - DEEPSEEK_API_KEY （不设置时回退到脚本内默认值） - DEEPSEEK_MODEL （默认 deepseek-chat） - -使用前提： - CAD 已打开并加载了 cad_mcp_frame.arx，8080 端口在监听。 -""" - -import os -import sys -import json -import time -import socket - -from openai import OpenAI - -HOST = "127.0.0.1" -PORT = 8080 - -# 与仓库既有脚本保持一致；生产环境请改为从环境变量注入 -API_KEY = os.environ.get("DEEPSEEK_API_KEY", "sk-420190f448fe41158c4e2ccff90e35ce") -BASE_URL = os.environ.get("DEEPSEEK_BASE_URL", "https://api.deepseek.com") -MODEL_NAME = os.environ.get("DEEPSEEK_MODEL", "deepseek-flash") # 该模型已确认支持视觉 - -# 每个图框审查时，取数据的条数上限 -FRAME_ENTITY_LIMIT = int(os.environ.get("FRAME_ENTITY_LIMIT", "200")) -# 最多审查多少个图框（0 = 全部） -MAX_FRAMES = int(os.environ.get("MAX_FRAMES", "0")) -# 审查清单文件路径 -CHECKLIST_PATH = os.environ.get("REVIEW_CHECKLIST", "review_checklist.json") - -client = OpenAI(api_key=API_KEY, base_url=BASE_URL) - - -# --------------------------------------------------------------------------- -# MCP / TCP 通信 -# --------------------------------------------------------------------------- -def call_tool(tool_name, arguments, recv_timeout=30.0): - """调用 CAD 基座的 tools/call，返回解析后的 JSON-RPC 响应（失败返回 None）""" - try: - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.settimeout(recv_timeout) - sock.connect((HOST, PORT)) - req = { - "jsonrpc": "2.0", - "id": 1, - "method": "tools/call", - "params": {"name": tool_name, "arguments": arguments}, - } - sock.sendall(json.dumps(req).encode("utf-8")) - - data = b"" - while True: - chunk = sock.recv(65536) - if not chunk: - break - data += chunk - # 基座是单次 send 返回整包；收到后尝试解析即可结束 - try: - json.loads(data.decode("utf-8")) - break - except json.JSONDecodeError: - continue - sock.close() - return json.loads(data.decode("utf-8")) - except Exception as e: - print(f"[Error] 调用工具 {tool_name} 失败: {e}") - return None - - -def mcp_text(resp): - """从 MCP 响应中取出文本内容；若文本是 JSON 则解析为对象返回""" - if not resp or "result" not in resp: - return None - for item in resp["result"].get("content", []): - if item.get("type") == "text": - text = item.get("text", "") - try: - return json.loads(text) - except (json.JSONDecodeError, TypeError): - return text - return None - - -def mcp_image(resp): - """从 MCP 响应中取出图片，返回 (base64, mimeType)""" - if not resp or "result" not in resp: - return None, None - for item in resp["result"].get("content", []): - if item.get("type") == "image": - return item.get("data"), item.get("mimeType", "image/png") - return None, None - - -def mcp_error(resp): - """若响应是错误，返回错误信息字符串，否则返回 None""" - if resp and "error" in resp: - err = resp["error"] - return f"code={err.get('code')} message={err.get('message')}" - return None - - -def bbox_to_window(bbox): - """[minx,miny,maxx,maxy] -> window 对象""" - return {"min_x": bbox[0], "min_y": bbox[1], "max_x": bbox[2], "max_y": bbox[3]} - - -# --------------------------------------------------------------------------- -# LLM 审查（审查清单可配置） -# --------------------------------------------------------------------------- -def load_checklist(path): - """加载审查清单 JSON；失败时回退到内置通用清单""" - try: - with open(path, "r", encoding="utf-8") as f: - data = json.load(f) - if isinstance(data, dict) and data.get("categories"): - return data - print(f"[Warn] 审查清单 {path} 结构异常，使用内置默认清单。") - except Exception as e: - print(f"[Warn] 无法加载审查清单 {path}: {e}，使用内置默认清单。") - - return { - "name": "内置通用清单", - "categories": [ - {"title": "图层规范性", "severity": "medium", - "checks": ["图元是否在正确图层", "是否存在空图层/命名异常图层"]}, - {"title": "文字", "severity": "medium", - "checks": ["空文字", "错别字", "字高异常", "文字重叠"]}, - {"title": "标注", "severity": "high", - "checks": ["关键尺寸缺失", "标注值异常"]}, - {"title": "几何", "severity": "medium", - "checks": ["零长度/退化实体", "重复重叠实体", "未闭合轮廓"]}, - {"title": "图面整体", "severity": "low", - "checks": ["图面完整性", "内容是否超出图框"]}, - ], - } - - -def build_system_prompt(checklist): - """根据审查清单动态生成 LLM 系统提示词""" - lines = [ - "你是一名资深建筑与工业厂房图纸审查专家。", - "你将逐张收到某个图框（一张图纸）的：结构化图元数据、文字内容，以及该图框的视口截图。", - "", - f"请依据审查清单《{checklist.get('name', '审查清单')}》逐项审查，" - "指出**具体问题**（尽量引用实体 handle 或文字内容），并标注严重程度：", - "", - ] - for cat in checklist.get("categories", []): - lines.append(f"【{cat.get('title', '')}】（严重度：{cat.get('severity', 'medium')}）") - for chk in cat.get("checks", []): - lines.append(f" - {chk}") - lines.append("") - lines += [ - "要求：", - "- 只报告你**有依据**的问题，不要臆测；数据与截图冲突时以数据为准。", - "- 按清单条目组织结论，未命中的条目可省略。", - "- 若该图框无任何明显问题，明确说“未发现明显问题”。", - "- 不要复述输入数据。", - ] - return "\n".join(lines) - - -def review_frame(idx, frame, entities, texts, img_b64, mime, system_prompt): - """对单个图框调用 LLM 审查，返回审查意见文本""" - sheet = frame.get("sheet_size") or "未知图幅" - kind = frame.get("kind") - layer = frame.get("layer") - - summary = { - "图框序号": idx, - "图幅": sheet, - "形态": kind, - "图层": layer, - "包围盒": frame.get("bbox"), - "图元总数": (entities or {}).get("total") if isinstance(entities, dict) else None, - "图元(截断)": (entities or {}).get("items") if isinstance(entities, dict) else None, - "文字": (texts or {}).get("items") if isinstance(texts, dict) else None, - } - - user_content = [ - {"type": "text", "text": "以下是该图框的审查输入数据(JSON)：\n" + - json.dumps(summary, ensure_ascii=False, indent=2)} - ] - if img_b64: - user_content.append({ - "type": "image_url", - "image_url": {"url": f"data:{mime};base64,{img_b64}"}, - }) - - messages = [ - {"role": "system", "content": system_prompt}, - {"role": "user", "content": user_content}, - ] - - try: - resp = client.chat.completions.create(model=MODEL_NAME, messages=messages) - return resp.choices[0].message.content - except Exception as e: - return f"[LLM 审查失败] {e}" - - -# --------------------------------------------------------------------------- -# 主流程 -# --------------------------------------------------------------------------- -def main(): - print("=== 🏗️ 逐图框图纸审查 Agent 启动 ===") - - # 0. 载入可配置审查清单 - checklist = load_checklist(CHECKLIST_PATH) - system_prompt = build_system_prompt(checklist) - print(f"[Init] 审查清单：{checklist.get('name')}" - f"（{len(checklist.get('categories', []))} 类），模型：{MODEL_NAME}\n") - - # 1. 识别图框 - resp = call_tool("detect_drawing_frames", {"min_confidence": 0.3}) - err = mcp_error(resp) - if err: - print(f"[Error] 识别图框失败: {err}") - return - frames_data = mcp_text(resp) - items = frames_data.get("items", []) if isinstance(frames_data, dict) else [] - if not items: - print("未识别到任何图框，请确认图纸中存在图框，或调整 detect_drawing_frames 参数。") - return - - total = len(items) - if MAX_FRAMES > 0: - items = items[:MAX_FRAMES] - print(f"识别到 {total} 个图框，本次审查 {len(items)} 个。\n") - - findings = [] - - # 2. 逐图框审查 - for idx, frame in enumerate(items, 1): - bbox = frame.get("bbox") - if not bbox or len(bbox) != 4: - print(f"[跳过] 图框 {idx} 缺少有效包围盒") - continue - - print(f"--- 图框 {idx}/{len(items)} kind={frame.get('kind')} " - f"sheet={frame.get('sheet_size')} bbox={bbox}") - - win = bbox_to_window(bbox) - - # a. 缩放到图框 - call_tool("zoom_to_window", dict(win, margin=0.02)) - time.sleep(0.6) # 给 CAD 渲染留出时间 - - # b/c. 取图框内数据 - entities = mcp_text(call_tool("query_entities", { - "window": win, "limit": FRAME_ENTITY_LIMIT - })) - texts = mcp_text(call_tool("get_text_content", { - "window": win, "limit": FRAME_ENTITY_LIMIT - })) - - # d. 截图 - shot = call_tool("get_viewport_screenshot", {}) - img_b64, mime = mcp_image(shot) - print(f" 图元 {(entities or {}).get('total')} 个，" - f"文字 {(texts or {}).get('total')} 条，" - f"截图 {'已获取' if img_b64 else '无'}") - - # e. LLM 审查 - finding = review_frame(idx, frame, entities, texts, img_b64, mime, system_prompt) - findings.append({"frame": idx, "sheet": frame.get("sheet_size"), "finding": finding}) - print(f" 审查结论：{finding}\n") - - # 3. 汇总总报告 - print("=== 📋 汇总总报告 ===") - report_input = "\n\n".join( - f"【图框 {f['frame']}（图幅 {f['sheet']}）】\n{f['finding']}" for f in findings - ) - try: - resp = client.chat.completions.create( - model=MODEL_NAME, - messages=[ - {"role": "system", "content": - "你是图纸审查负责人。请把下面各图框的审查意见汇总为一份" - "结构化的审查报告：先给整体结论，再按图框列出问题，最后给出整改建议。" - "语言简洁、条目化。"}, - {"role": "user", "content": report_input}, - ], - ) - print(resp.choices[0].message.content) - except Exception as e: - print(f"[汇总失败] {e}") - print(report_input) - - -if __name__ == "__main__": - main() diff --git a/review_checklist.json b/review_checklist.json index 70978c7..242bad5 100644 --- a/review_checklist.json +++ b/review_checklist.json @@ -1,7 +1,7 @@ { "version": "1.0.0", "name": "通用图纸审查清单", - "description": "审查清单（可替换）。dwg_review_agent.py 会把 categories[].checks[] 注入 LLM 审查提示词。后期可在此追加建筑标准、工业管道标准等规范条目。", + "description": "审查清单（可替换）。dwg_review 包会把 categories[].checks[] 注入 LLM 审查提示词。后期可在此追加建筑标准、工业管道标准等规范条目。", "categories": [ { "id": "layers", diff --git a/test.py b/test.py deleted file mode 100644 index 76bd46c..0000000 --- a/test.py +++ /dev/null @@ -1,41 +0,0 @@ -import socket -import json - -def test_fake_llm_zoom(): - print("=== 模拟 LLM 发起缩放指令测试 ===") - - try: - client = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - client.connect(("127.0.0.1", 8080)) - - # 伪造大模型生成的 tool_call 报文 - req = { - "jsonrpc": "2.0", - "id": 999, - "method": "tools/call", - "params": { - "name": "zoom_window_normalized", - "arguments": { - "x1": 0.4, - "y1": 0.4, - "x2": 0.6, - "y2": 0.6 - } - } - } - - print(f"\n[发送] 请求参数:\n{json.dumps(req['params'], indent=2)}") - client.sendall(json.dumps(req).encode('utf-8')) - - response_data = client.recv(4096) - - print(f"\n[接收] CAD 基座返回:") - print(json.dumps(json.loads(response_data.decode('utf-8')), indent=2, ensure_ascii=False)) - - client.close() - - except Exception as e: - print(f"测试失败: {e}") - -if __name__ == "__main__": - test_fake_llm_zoom() \ No newline at end of file diff --git a/test_autoagent.py b/test_autoagent.py deleted file mode 100644 index 0241a2f..0000000 --- a/test_autoagent.py +++ /dev/null @@ -1,43 +0,0 @@ -import socket -import json -import base64 - -def test_screenshot(): - client = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - client.connect(("127.0.0.1", 8080)) - - req = { - "jsonrpc": "2.0", - "id": 5, - "method": "tools/call", - "params": { - "name": "get_viewport_screenshot", - "arguments": {} - } - } - client.sendall(json.dumps(req).encode('utf-8')) - - # 图片 base64 报文可能很大，需要循环接收直到拿完 - response_data = b"" - while True: - chunk = client.recv(8192) - response_data += chunk - if len(chunk) < 8192: - break - - client.close() - - resp_json = json.loads(response_data.decode('utf-8')) - - # 解析并保存图片 - content_list = resp_json["result"]["content"] - for item in content_list: - if item["type"] == "image": - base64_data = item["data"] - image_bytes = base64.b64decode(base64_data) - with open("test_screenshot.png", "wb") as f: - f.write(image_bytes) - print("截图已保存为 test_screenshot.png，请在当前目录查看！") - -if __name__ == "__main__": - test_screenshot() \ No newline at end of file