"""Isolated original-images → parametric graph → CAD → visual feedback trial. No company BOM enters inference. No business sessions or production services are mutated. A successful compile/render is NOT visual or pricing acceptance. """ from __future__ import annotations import argparse import base64 import importlib.util import json import os import pathlib import re import time from structure_graph import ( GraphError, SCHEMA_VERSION, apply_graph_patch, build_dimension_ledger, build_source_topology_contract, compile_graph, normalize_graph_patch_payload, ) from manufacturing_oracle import plan_manufacturing_features ROLE_FILENAME_ALIASES = { "front": ("front",), "side": ("side",), "third": ("third", "open", "three-quarter", "perspective"), "dimension": ("dimension", "dimensions", "dim", "drawing"), } def source_for_role(asset_dir: pathlib.Path, case: str, role: str) -> pathlib.Path | None: """Resolve test evidence by semantic role, not one fragile filename.""" aliases = ROLE_FILENAME_ALIASES[role] allowed = {".jpg", ".jpeg", ".png", ".webp"} candidates = [] for alias_index, alias in enumerate(aliases): for path in asset_dir.glob(f"{case}-{alias}.*"): if path.suffix.lower() in allowed: candidates.append((alias_index, path.name, path)) return min(candidates)[2] if candidates else None def _json_object_from_error(error: Exception | str) -> dict | None: """Extract the first structured CAD-kernel payload from an exception.""" text = str(error or "").strip() decoder = json.JSONDecoder() for index, character in enumerate(text): if character != "{": continue try: value, _ = decoder.raw_decode(text[index:]) except json.JSONDecodeError: continue if isinstance(value, dict): return value return None def structured_kernel_feedback(error: Exception | str) -> list[dict]: """Turn B-rep failures into stable-instance repair instructions.""" payload = _json_object_from_error(error) feedback = [] if payload: for clash in payload.get("clashes") or []: if not isinstance(clash, dict): continue first = str((clash.get("a") or {}).get("name") or "").strip() second = str((clash.get("b") or {}).get("name") or "").strip() bounds = clash.get("bounds") or {} minimum, maximum = bounds.get("min") or [], bounds.get("max") or [] overlap = { axis: round(float(maximum[position]) - float(minimum[position]), 4) for position, axis in enumerate("xyz") if position < len(minimum) and position < len(maximum) } feedback.append({ "code": "brep_interference", "severity": "blocking", "instance_ids": [value for value in (first, second) if value], "volume_mm3": clash.get("volume"), "overlap_size_mm": overlap, "bounds_mm": bounds, "correction": ( "回看原图后修改这两个实例的零件尺寸、所属总成或placement整轴约束，" "使实体没有正体积重叠；不得删除原图可见零件或缩小到错误尺寸。" ), "legal_patch_roots": [ f"/instances/{instance_id}/placement" for instance_id in (first, second) if instance_id ], }) for detail in payload.get("errors") or []: feedback.append({ "code": "cad_kernel_validation_error", "severity": "blocking", "detail": str(detail), }) if feedback: return feedback return [{ "code": "cad_kernel_error", "severity": "blocking", "detail": str(error or "CAD几何内核未通过")[-1800:], "correction": "只修改导致内核失败的图谱参数、零件几何或定位约束，然后重新构建。", }] def structured_patch_feedback(graph: dict, patches, error: Exception | str) -> list[dict]: """Explain a rejected JSON Patch without silently changing its semantics.""" issues = [] instances = list(graph.get("instances") or []) if isinstance(graph, dict) else [] for patch in patches if isinstance(patches, list) else []: path = str(patch.get("path") or "") if isinstance(patch, dict) else "" match = re.fullmatch(r"/instances/([^/]+)/placement/([xyz])/(value_mm|offset_mm)", path) if not match: continue token, axis, leaf = match.groups() instance = None if token.isdigit() and int(token) < len(instances): instance = instances[int(token)] else: instance = next((row for row in instances if str(row.get("id") or "") == token), None) if not isinstance(instance, dict): continue stable_id = str(instance.get("id") or token) current = ((instance.get("placement") or {}).get(axis) or {}) legal = [{ "op": "replace", "path": f"/instances/{stable_id}/placement/{axis}", "value_shape": {"mode": "absolute", "value_mm": "<毫米值>"}, }] if current.get("mode") == "align": legal.append({ "op": "add" if "offset_mm" not in current else "replace", "path": f"/instances/{stable_id}/placement/{axis}/offset_mm", "value_shape": "<毫米偏移值>", }) issues.append({ "code": "patch_application_error", "severity": "blocking", "attempted_path": path, "instance_id": stable_id, "axis": axis, "current_constraint": current, "legal_repair_examples": legal, "detail": str(error), "correction": ( f"当前{axis.upper()}轴没有可replace的{leaf}字段。使用稳定实例ID并替换整个轴约束；" "如果当前是align且只需平移，则对offset_mm使用add/replace。" ), }) return issues or [{ "code": "patch_application_error", "severity": "blocking", "detail": str(error), "attempted_patches": patches, "correction": "保留上一版图谱，根据现有字段和稳定ID重新生成合法局部修改。", }] GRAPH_PROMPT = """根据全部原始图片构建可编辑、可验证的家具装配图谱，不输出报价。 先理解产品正面、侧面和尺寸线方向，再分清每个物理板件及其所属总成。 不能把抽屉整体算成板件，固定横隔板不属于抽屉箱体；每个物理实例有独立ID。 侧板是否断开需要局部接缝证据，颜色变化不等于断开，层板边缘不等于侧板接缝。 多张图存在矛盾则在uncertainties中记录；未看见不能当作没有。 严格JSON结构： {"schema_version":"furniture-assembly-graph/1", "placement_solver":"anchored_axes_v1", "orientation_solver":"role_axis_templates_v1", "contact_derivation":"aabb_within_assembly_v1", "observations":[{"id":"e1","source_type":"image","image_id":"image1","confidence":0.0,"text":"原图观察，不是经验","region":[0,0,1,1]}], "parameters":{"parameter_name":{"kind":"length/count/scalar","origin":"observed/derived/estimated","value":0,"unit":"mm/cm/m","evidence_ids":["e1"],"confidence":0.0}}, "envelope_mm":["width","depth","height"], "assemblies":[{"id":"lower_module","role":"carcass","bounds_mm":{"x":[0,"width"],"y":[0,"depth"],"z":[0,"lower_height"]},"evidence_ids":["e1"]},{"id":"drawer_1","role":"drawer","bounds_mm":{"x":["side_thickness","width-side_thickness"],"y":[0,"drawer_depth"],"z":["drawer_z_min","drawer_z_max"]},"evidence_ids":["e1"]}], "parts":[{"id":"part_id","name":"中文部件名","shape":"box","functional_role":"side","thickness_axis":"x","size_mm":["参数或算式","参数或算式","参数或算式"],"observed_quantity":1,"evidence_ids":["e1"]}], "instances":[{"id":"物理实例ID","part_id":"part_id","assembly_id":"明确所属总成ID","placement":{"x":{"mode":"align","self_anchor":"min","target":"envelope","target_anchor":"min","offset_mm":0},"y":{"mode":"absolute","value_mm":0},"z":{"mode":"align","self_anchor":"max","target":"另一实例ID","target_anchor":"min","offset_mm":0}}}], "connections":[], "assembly_connections":[{"id":"drawer_slide","kind":"sliding","a":"drawer_1","b":"carcass","axis":"y","travel_limits_mm":[0,250],"evidence_ids":["e1"]}], "uncertainties":[]} 示例字段名仅是格式，不是产品答案。不要照抄示例观察。 参数：observed仅用于图上明确可读的标注，必须引用实际图片证据；照片比例、隐蔽板厚等用estimated且confidence<=0.55。 derived参数不写value/unit，改写expression为参数名和加减乘除的算式，如width-2*panel_thickness。 先定义变量，再引用；数字是无量纲系数，不要把mm和件数直接相加。零件size_mm/position_mm中数字字面量单位固定mm。 全图只使用一个产品坐标系：X=产品宽、Y=产品深、Z=产品高。固定写orientation_solver=role_axis_templates_v1。 size_mm必须直接按产品[X宽,Y深,Z高]填写；instances禁止输出axis_order，也禁止对零件做第二次轴交换。 固定写placement_solver=anchored_axes_v1，禁止输出position_mm或心算XYZ坐标。 每个part必须声明functional_role，只允许top/bottom/side/shelf/door/back/drawer_front/drawer_left/drawer_right/drawer_back/drawer_bottom/drawer_support/divider/leg/rail/stile/frame_member/hardware/other。 所有板件还必须声明产品坐标中的thickness_axis=x/y/z，它必须是size_mm最薄的一轴；不能根据中文名称猜方向。非板件可写null。 角色朝向规则：top/bottom/shelf/drawer_bottom/drawer_support厚度轴为z；side/drawer_left/drawer_right为x；door/back/drawer_front/drawer_back为y；divider只允许x或z。 例如侧板size_mm=[t,depth,height]且thickness_axis=x；层板size_mm=[inner_width,depth,t]且thickness_axis=z；门板size_mm=[door_width,t,door_height]且thickness_axis=y。 每个实例的placement必须有x/y/z三轴。每轴只允许absolute(value_mm)，或align(self_anchor,target,target_anchor,offset_mm)。 anchor只能是min/center/max；target只能是envelope、字面量assembly、当前实例自己的assembly_id，或先定义的实例ID。当前assembly_id与assembly等价，不能用其他总成ID做跨总成定位。每个部件优先在所属assembly局部边界内定位；例如模块右侧板X最大面对齐所属总成X最大面，应写self_anchor=max,target=assembly,target_anchor=max，实际坐标由求解器计算。只有部件确实位于整件产品最外边界时才能直接锚定envelope。 固定写contact_derivation=aabb_within_assembly_v1。AI只声明每个物理件所属总成，同一总成内的固定/支撑接触由编译器根据真实几何正面积相交自动推导，禁止AI猜固定连接面。 每个assembly必须至少有一个实例，不能创建空root。每个assembly必须给出严格bounds_mm，边界等于所属实体并集的真实包围盒，不能把整件产品边界复制给局部模块。尺寸图或多视图显示上下分段时，按真实结构分成上下模块；上部模块的Z最小值必须来自分段尺寸或可追溯派生参数，其侧板不得再次落到产品Z=0。 assembly.role只允许carcass/drawer/door/module/frame/hardware_group/other；门总成写door，禁止写door_assembly、sliding_door等自创角色。滑动或铰接性质写在assembly_connections.kind，不写进role。 若顶板、底板、侧板、背板或腿属于同一carcass，它的bounds_mm必须覆盖这些实体的完整并集；不能只用内部净空作为carcass边界。落地腿的Z最小面通常对齐产品envelope最小面，不能把腿的Z最大面对齐最小面而放到产品外。 抽屉的前/后/左/右帮板和底板属于同一drawer总成；柜体板属于carcass，不能把“抽屉”运动性质挂在某一块帮板上。层板、底板和抽屉帮板必须按相邻侧板之间的净空生成，不能按产品总宽横穿侧板；任何实体正体积穿插都必须修正。 每个会独立运动的抽屉、平开门或移门必须是独立assembly；单块门板也要建立包含一个实例的总成。不同总成之间用assembly_connections，a/b只能引用assemblies中的稳定ID，禁止引用零件实例ID：fixed必须在最终几何中真实接触；sliding必须有axis与travel_limits_mm；hinged必须有axis与angle_limits_deg。connections中不要生成fixed/support。 不同规格独立part；相同规格由多个instances引用，BOM数量只等于真实实例数，不额外产生总成板。 所有运动连接必须引用真实实例和真实总成，不能用大公差掩盖错位。 只画现有图像支持的结构；不要根据常见柜子想象层板数量。未标注内部尺寸先由AI按多图比例估算并标记estimated，生成评审草案； 只有原图完全不可见且对制造/成本高度敏感、经过局部复读与CAD反向复核仍不能判定的字段，才进入用户确认候选。 不得引用公司BOM或产品名称模板生成几何，不得生成source_type=user。 当前编译器只支持正交box板件；复杂曲面仍须保留uncertainties，不能宣称已精确复刻。 输出紧凑有效JSON，最多40个parts、100个instances，不输出油漆材料、成本或孔位模板。""" def render_multiview(ai3d, build_dir: pathlib.Path, step_path: str) -> list[dict]: """Render orthographic evidence so the reviewer is not trapped in one ISO view.""" python = str(ai3d.runtime_dir / "venv/bin/python") snapshot = str(ai3d.skill_dir / "scripts/snapshot") outputs = [ {"path": str(build_dir / f"cad-{camera}.png"), "camera": camera} for camera in ("front", "right", "top", "iso") ] ai3d._run_trusted_snapshot( build_dir, [python, snapshot, "--job", "-", "--json"], stdin=json.dumps({ "input": step_path, "mode": "view", "outputs": outputs, "render": {"viewLabels": True, "padding": 0.1, "sizeProfile": "diagnostic"}, }), timeout=ai3d.timeout, ) return [{"camera": item["camera"], "path": item["path"]} for item in outputs if pathlib.Path(item["path"]).is_file()] def cad_source(compiled, manufacturing_plan=None): lines = [ "from build123d import Align, Axis, Box, Cylinder, Pos, Compound, Color", "", "def gen_step():", " parts = []", ] colors = [(0.72, 0.83, 0.96), (0.96, 0.77, 0.45), (0.8, 0.65, 0.86), (0.5, 0.82, 0.67)] feature_map = (manufacturing_plan or {}).get("features_by_instance") or {} manufacturing_geometry = (manufacturing_plan or {}).get("manufacturing_geometry_by_instance") or {} for index, inst in enumerate(compiled["instances"]): adjusted = manufacturing_geometry.get(str(inst["id"])) or {} size = adjusted.get("size_mm") or inst["size_mm"] pos = adjusted.get("position_mm") or inst["position_mm"] center = [pos[axis] + size[axis] / 2 for axis in range(3)] lines.extend([ f" part = Box({size[0]!r}, {size[1]!r}, {size[2]!r}, align=(Align.CENTER, Align.CENTER, Align.CENTER)).moved(Pos({center[0]!r}, {center[1]!r}, {center[2]!r}))", ]) for feature_index, feature in enumerate((feature_map.get(str(inst["id"])) or {}).get("holes") or [], 1): if feature.get("validation_status") != "valid": continue world = feature["world_center_mm"] axis = feature["axis"] rotate = ".rotate(Axis.Y, 90)" if axis == "x" else ".rotate(Axis.X, 90)" if axis == "y" else "" lines.extend([ f" hole_{index}_{feature_index} = Cylinder({float(feature['diameter_mm']) / 2!r}, {float(feature['depth_mm'])!r}, align=(Align.CENTER, Align.CENTER, Align.CENTER)){rotate}.moved(Pos({world['x']!r}, {world['y']!r}, {world['z']!r}))", f" part = part - hole_{index}_{feature_index}", ]) for feature_index, feature in enumerate((feature_map.get(str(inst["id"])) or {}).get("grooves") or [], 1): if feature.get("validation_status") != "valid": continue world, groove_size = feature["world_center_mm"], feature["world_size_mm"] lines.extend([ f" groove_{index}_{feature_index} = Box({groove_size['x']!r}, {groove_size['y']!r}, {groove_size['z']!r}, align=(Align.CENTER, Align.CENTER, Align.CENTER)).moved(Pos({world['x']!r}, {world['y']!r}, {world['z']!r}))", f" part = part - groove_{index}_{feature_index}", ]) lines.extend([ f" part.label = {str(inst['id'])!r}", f" part.color = Color{colors[index % len(colors)]!r}", " parts.append(part)", ]) lines.append(" return Compound(children=parts, label='assembly_graph_draft')") return "\n".join(lines) + "\n" def main(): parser = argparse.ArgumentParser() parser.add_argument("--engine", required=True) parser.add_argument("--assets", required=True) parser.add_argument("--output", required=True) parser.add_argument("--cases", default="book,slide") parser.add_argument("--model", default="qwen3.5-plus") parser.add_argument("--rounds", type=int, default=2) parser.add_argument( "--material-family", default="engineered_panel", choices=("engineered_panel", "solid_wood", "metal", "plastic", "glass", "other"), help="Authoritative user-selected material family; never inferred from the image.", ) args = parser.parse_args() out = pathlib.Path(args.output) out.mkdir(parents=True, exist_ok=True) os.environ["DALONGXIA_SOLID_WOOD_COST_STATE_DIR"] = str(out / "isolated-cost-state") os.environ["DALONGXIA_AI_3D_STATE_DIR"] = str(out / "isolated-cad-state") os.environ["DALONGXIA_SOLID_WOOD_HISTORY_BOM_ENABLED"] = "0" spec = importlib.util.spec_from_file_location("graph_probe_engine", args.engine) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) engine = mod.SolidWoodCostEngine() engine.model = args.model ai3d = engine._load_ai3d() reports = [] for case in args.cases.split(","): case_dir = out / case case_dir.mkdir(exist_ok=True) images = [] source_assets = [] image_ids = set() for index, role in enumerate(("front", "side", "third", "dimension"), 1): source = source_for_role(pathlib.Path(args.assets), case, role) if source is None: raise RuntimeError(f"missing source view: {case}/{role}") image_id = f"image{index}" image_ids.add(image_id) mime = "image/png" if source.suffix == ".png" else "image/jpeg" data_url = f"data:{mime};base64," + base64.b64encode(source.read_bytes()).decode() images.extend([{"type": "text", "text": f"原始图ID={image_id}；候选视角={role}，需按图片内容复核。"}, {"type": "image_url", "image_url": {"url": data_url}}]) source_assets.append({ "asset_id": index, "role": role, "kind": "image_base64", "value": data_url, "filename": source.name, "validation_status": "accepted", }) report = {"case": case, "schema_version": SCHEMA_VERSION, "model": args.model, "material_family": args.material_family, "full_chat_api": False, "company_bom_used_in_inference": False, "rounds": [], "status": "not_validated"} print(json.dumps({"case": case, "phase": "ai_source_evidence_qualification"}), flush=True) source_structure, source_dimensions = {}, {} try: source_structure = engine._read_source_structure(source_assets) except Exception as exc: source_structure = {"status": "machine_reread_failed", "error": f"{type(exc).__name__}: {exc}"} try: source_dimensions = engine._read_source_dimension_labels( source_assets[-1], [source_assets[0], source_assets[1]] ) except Exception as exc: source_dimensions = {"status": "machine_reread_failed", "error": f"{type(exc).__name__}: {exc}"} dimension_ledger = build_dimension_ledger( source_dimensions.get("overall_dimension_labels") or [], source_image_id="image4" ) topology_contract = build_source_topology_contract(source_structure) report["source_qualification"] = { "structure": source_structure, "dimensions": source_dimensions, "dimension_ledger": dimension_ledger, "topology_contract": topology_contract, "policy": "AI先交叉识别和CAD复核；只有不可见且成本敏感的残余冲突才询问用户", } def checkpoint(current_row=None): snapshot = json.loads(json.dumps(report, ensure_ascii=False)) if current_row is not None: snapshot["rounds"] = list(snapshot.get("rounds") or []) + [current_row] (case_dir / "report.json").write_text( json.dumps(snapshot, ensure_ascii=False, indent=2), encoding="utf-8" ) checkpoint() missing_axes = [axis for axis in "xyz" if axis not in dimension_ledger] if missing_axes or dimension_ledger.get("conflicts"): report["status"] = "source_dimensions_not_qualified" report["source_qualification"]["blocking"] = { "code": "overall_dimension_ledger_incomplete", "missing_axes": missing_axes, "conflicts": dimension_ledger.get("conflicts") or [], "next_owner": "ai_focused_source_reread", } reports.append(report) (case_dir / "report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") (out / "report.json").write_text(json.dumps(reports, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps({ "case": case, "status": report["status"], "blocking": report["source_qualification"]["blocking"], }, ensure_ascii=False), flush=True) continue evidence_context = ( "\n以下是同一批原图经过两个独立AI专项读取后的候选证据。它们不是公司BOM，也不是最终答案；" "你必须回看原图。dimension_ledger中存在的X/Y/Z是已通过范围、方向和置信度门的整体尺寸，必须精确采用；" "structure.view_adjudication.accepted是跨视图成立的结构约束；conflicts和machine_followups应由本轮图像复核解决，" "不能直接转嫁给用户。\n" + json.dumps({ "structure": source_structure, "dimension_read": source_dimensions, "dimension_ledger": dimension_ledger, "topology_contract": topology_contract, }, ensure_ascii=False) ) graph, feedback = {}, [] accepted_compile_issue_count = None for turn in range(max(1, min(args.rounds, 5))): started = time.time() row = {"round": turn + 1} prompt = GRAPH_PROMPT + evidence_context if graph: prompt = ( "你在修正一份已存在的参数化家具结构图谱。必须回看本轮所有原图，并只返回局部JSON修改，不能返回整份图谱。" "每条操作为add/replace/remove和JSON Pointer路径；修复引用、参数、实例位置、接触面或遗漏部件。" "不能通过放大公差、删除有原图证据的部件/连接或改低置信度来逃避问题。" "一次最多24条patch，优先级固定为：原图整体尺寸冲突、总成局部边界错误、实体穿插、已定义却无实例的部件、总成归属/总成关系错误、几何未真实接触、其他外观差异。" "若零件observed_quantity大于实例数，必须为每个真实物理件补instances，不能只保留零件定义。" "同一总成的固定关系不准手写connections；必须修正部件尺寸、placement锚点或assembly_id，让求解器计算坐标并从真实接触面推导连接。" "anchored_axes_v1下严禁修改不存在的position_mm；必须修改总成bounds_mm、零件size_mm或placement锚点/offset_mm。" "禁止replace或remove整个/instances//placement；只修改反馈明确指出的/placement/x、/placement/y或/placement/z单轴，不能连带改坏已通过的轴。" "若仍返回完整placement，系统只会拆成实际发生变化的单轴修改并记录审计，不会允许绕过底层保护。" "assembly.role只允许carcass/drawer/door/module/frame/hardware_group/other；门总成必须写door，运动类型写assembly_connections.kind。" "图谱固定使用orientation_solver=role_axis_templates_v1；每个板件必须保留functional_role，size_mm直接表示产品X/Y/Z，thickness_axis也是产品轴。" "严禁新增或修改axis_order；若角色朝向冲突，必须修正part的size_mm与thickness_axis，不能对instance二次旋转。" "编译器会用这些字段区分面贴面、边贴面与边贴边，不允许用名称代替接口语义。" "修正定位时优先使用稳定实例ID。若某轴当前没有value_mm或offset_mm，不得replace不存在的叶子字段；" "必须replace整个轴对象，或对align轴用add新增offset_mm。" "合法示例：{\"patches\":[{\"op\":\"replace\",\"path\":\"/instances/door_left_inst/placement/x\"," "\"value\":{\"mode\":\"absolute\",\"value_mm\":20}}]}。" "路径可修改observations、parameters、envelope_mm、assemblies、parts、instances、connections、assembly_connections、uncertainties、placement_solver、orientation_solver、contact_derivation；" "数组追加使用/-。修改后仍需符合furniture-assembly-graph/1。\n上一版图谱：" + json.dumps(graph, ensure_ascii=False) + evidence_context + "\n编译器及原图视觉复核的结构化反馈：" + json.dumps(feedback, ensure_ascii=False) ) phase = "ai_graph_from_original_images" if not graph else "ai_local_graph_repair_from_feedback" print(json.dumps({"case": case, "round": turn + 1, "phase": phase}), flush=True) try: response = engine._request_model_json({ "model": args.model, "messages": [{"role": "user", "content": images + [{"type": "text", "text": prompt}]}], "response_format": {"type": "json_object"}, "temperature": 0.0, "enable_thinking": True, "thinking_budget": 2048, "vl_high_resolution_images": True, "max_tokens": 14000, }, "原图到参数化结构图谱失败", max_attempts=1, timeout_seconds=180, decoder=lambda text: json.loads(text)) if graph: row["patch_response"] = response previous_graph = graph try: normalized_patches, normalization_audit = normalize_graph_patch_payload(graph, response) row["patch_normalization_audit"] = normalization_audit row["normalized_patches"] = normalized_patches candidate_graph, patch_audit = apply_graph_patch(graph, normalized_patches) row["patch_audit"] = patch_audit candidate_compiled = compile_graph( candidate_graph, source_image_ids=image_ids, dimension_ledger=dimension_ledger, topology_contract=topology_contract, ) candidate_issue_count = len(candidate_compiled.get("issues") or []) if ( accepted_compile_issue_count is not None and candidate_issue_count > accepted_compile_issue_count ): graph = previous_graph row["patch_error"] = "repair_regression_rejected" row["patch_regression"] = { "previous_issue_count": accepted_compile_issue_count, "candidate_issue_count": candidate_issue_count, "candidate_issue_codes": dict(__import__("collections").Counter( item.get("code") for item in candidate_compiled.get("issues") or [] )), } feedback = [{ "code": "repair_regression_rejected", "detail": ( f"本轮局部修改使编译问题从{accepted_compile_issue_count}项增加到" f"{candidate_issue_count}项，已保留上一版图谱；下一轮必须缩小修改范围。" ), "correction": "逐条采用编译器给出的稳定ID和单轴路径，禁止整段覆盖placement或引入统一经验偏移。", }] + list(candidate_compiled.get("issues") or [])[:12] report["status"] = "regressive_patch_rejected" row["checkpoint_phase"] = "regressive_patch_rejected_graph_preserved" checkpoint(row) else: graph = candidate_graph except GraphError as exc: row["patch_error"] = f"{type(exc).__name__}: {exc}" raw_patches = response.get("patches") if isinstance(response, dict) else response feedback = structured_patch_feedback(graph, raw_patches, exc) row["patch_feedback"] = feedback report["status"] = "patch_not_applied" row["checkpoint_phase"] = "invalid_patch_rejected_graph_preserved" checkpoint(row) else: graph = response if not row.get("patch_error"): row["graph"] = graph compiled = compile_graph( graph, source_image_ids=image_ids, dimension_ledger=dimension_ledger, topology_contract=topology_contract, ) row["compiled"] = compiled accepted_compile_issue_count = len(compiled.get("issues") or []) feedback = list(compiled["issues"]) manufacturing_plan = plan_manufacturing_features( compiled, policy={"default_material_family": args.material_family}, ) row["manufacturing_plan"] = manufacturing_plan if compiled["ok"]: feedback += list(manufacturing_plan.get("issues") or []) row["checkpoint_phase"] = "graph_compiled" checkpoint(row) if compiled["instances"]: build_dir = case_dir / f"round-{turn + 1}" build_dir.mkdir(exist_ok=True) source = build_dir / "assembly.step.py" source.write_text(cad_source(compiled, manufacturing_plan), encoding="utf-8") ai3d._validate_source(source.read_text()) try: built = ai3d._build_and_validate(build_dir, source.name, requested_exports=["glb"]) except Exception as exc: row["cad_kernel_error"] = f"{type(exc).__name__}: {exc}" kernel_feedback = structured_kernel_feedback(exc) row["cad_kernel_feedback"] = kernel_feedback feedback += kernel_feedback report["status"] = ( "cad_kernel_not_passed" if compiled["ok"] else "graph_and_cad_kernel_not_passed" ) row["checkpoint_phase"] = "cad_kernel_rejected" checkpoint(row) else: built["multiview_previews"] = render_multiview(ai3d, build_dir, built["step_path"]) row["built" if compiled["ok"] else "diagnostic_built"] = built row["checkpoint_phase"] = "cad_built_before_visual_review" checkpoint(row) cad_views = [] for item in built["multiview_previews"]: view_path = pathlib.Path(item["path"]) cad_views.extend([ {"type": "text", "text": f"候选CAD {item['camera']}正交视图（不是原图证据）："}, {"type": "image_url", "image_url": {"url": "data:image/png;base64," + base64.b64encode(view_path.read_bytes()).decode()}}, ]) review = engine._request_model_json({ "model": args.model, "messages": [{"role": "user", "content": images + cad_views + [ {"type": "text", "text": ( "必须分别使用CAD正面、右侧、俯视和等轴视图，对照全部原始图核验CAD侧板分段、抽屉排列、" "所有可见板件数量、门型、部件方向、连接关系和外形比例。" "这是诊断草图，编译错误中的零件可能未渲染；不能因为画面简单就判正确。" "先由AI给出可执行correction；uncertain仅保留所有原图和四个CAD视图都不能消解的隐蔽信息。" "只返回JSON {ok:boolean,issues:[{code,component,source_image_id,evidence,correction}],uncertain:[]}。" "每个问题引用一张能证明问题的原图ID；没有原图证据的推测放uncertain。\n编译器问题：" + json.dumps(compiled["issues"], ensure_ascii=False) + "\n当前由物理实例生成的BOM：" + json.dumps(compiled["bom"], ensure_ascii=False) + "\n由真实接触面生成的孔槽与连接件计划：" + json.dumps(manufacturing_plan, ensure_ascii=False) )}, ]}], "response_format": {"type": "json_object"}, "temperature": 0.0, "enable_thinking": True, "thinking_budget": 2048, "vl_high_resolution_images": True, "max_tokens": 6000, }, "原图与结构图谱CAD复核失败", max_attempts=1, timeout_seconds=150, decoder=lambda text: json.loads(text)) row["visual_review"] = review feedback = list(compiled["issues"]) + list(review.get("issues") or []) + list(review.get("uncertain") or []) if compiled["ok"] and manufacturing_plan.get("ok") and review.get("ok") is True and not feedback and not graph.get("uncertainties"): report["status"] = "candidate_visual_pass_requires_independent_acceptance" else: report["status"] = "visual_not_passed" if compiled["ok"] else "graph_not_passed_with_diagnostic_render" else: report["status"] = "graph_not_passed" except Exception as exc: row["error"] = f"{type(exc).__name__}: {exc}" report["status"] = "failed" feedback = [{"code": "request_or_compile_error", "detail": row["error"]}] row["elapsed_seconds"] = round(time.time() - started, 2) row.pop("checkpoint_phase", None) report["rounds"].append(row) (case_dir / "report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps({"case": case, "round": turn + 1, "status": report["status"], "elapsed": row["elapsed_seconds"], "feedback": feedback[:8]}, ensure_ascii=False), flush=True) if report["status"].startswith("candidate_visual_pass"): break reports.append(report) (out / "report.json").write_text(json.dumps(reports, ensure_ascii=False, indent=2), encoding="utf-8") if __name__ == "__main__": main()