310 lines
11 KiB
Python
310 lines
11 KiB
Python
|
"""
|
|||
|
|
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()
|