chore: 删除 5 个遗留调试脚本并同步文档引用

- 删除 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 备案)。
This commit is contained in:
gjm committed 2026-10-14 19:07:35 +08:00
1 parent 0656397a68
commit 8c0fcfee83
10 files changed
+53 -701

No files matched your search

+22 -6
View File
@@ -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。
## 架构
-163
View File
@@ -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("请仔细检查当前图纸,告诉我图纸中心那些小圆的内部,是否还包含了更小的同心圆?如果看不清,请务必放大确认。")
+6 -3
View File
@@ -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 参照规范审核。
+6 -1
View File
@@ -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)。
## 环境事实
+18 -6
View File
@@ -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` 的残留产物。
## 两个构建目标
| 目标 | 产物 | 作用 |
-127
View File
@@ -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()
-310
View File
@@ -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()
+1 -1
View File
@@ -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",
-41
View File
@@ -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()
-43
View File
@@ -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()