""" 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()