cad-agent/dwg_review/phases/review.py

"""阶段 3：逐图框审查（LLM + 受控只读取证子循环）。

流程：
  1. 用清单生成系统提示，喂入结构化证据 + 截图。
  2. LLM 若请求只读取证工具 → 执行后把观测回灌，再问（有界循环）。
  3. LLM 不再请求工具时，其输出即为该图框的审查结论。
"""

import json

from ..cad_tools import READ_ONLY_TOOLS, is_write_tool
from ..llm_client import LlmClient, downsample_image_b64
from ..mcp_client import McpError, McpToolError, result_images, result_texts

# 单次审查回灌给 LLM 的图片数上限（控制视觉 token 成本）
_MAX_OBS_IMAGES = 3


def evidence_payload(evidence):
    """把证据整理成给 LLM 的 JSON（剔除 base64 图片，避免撑爆上下文）。"""
    frame = evidence.get("frame") or {}
    return {
        "frame_index": evidence.get("index"),
        "frame": {
            k: frame.get(k)
            for k in ("sheet_size", "kind", "layer", "bbox", "confidence")
        },
        "entities": evidence.get("entities"),
        "texts": evidence.get("texts"),
        "dimensions": evidence.get("dimensions"),
        "blocks": evidence.get("blocks"),
        "polylines": evidence.get("polylines"),
    }


def _run_readonly_tool(ctx, name, args, image_budget):
    """执行一个只读取证工具，返回回灌给 LLM 的 content 列表。"""
    if name not in READ_ONLY_TOOLS or is_write_tool(name):
        return [{"type": "text", "text": f"工具 {name} 不在只读取证白名单内，已拒绝。"}], image_budget

    try:
        resp = ctx.tools.call(name, args)
    except (McpError, McpToolError) as exc:
        return [{"type": "text", "text": f"工具 {name} 执行失败: {exc}"}], image_budget

    content = []
    for text in result_texts(resp):
        content.append({"type": "text", "text": text})
    for b64, mime in result_images(resp):
        if image_budget[0] <= 0:
            content.append({"type": "text", "text": "（后续截图已省略以控制成本）"})
            break
        image_budget[0] -= 1
        nb, nm = downsample_image_b64(b64, mime)
        content.append(LlmClient.image_content(nb, nm))
    if not content:
        content.append({"type": "text", "text": "（无返回内容）"})
    return content, image_budget


def review_frame(ctx, evidence, system_prompt, log=print):
    """审查单个图框，返回审查意见文本。"""
    payload = evidence_payload(evidence)
    user_content = [
        {
            "type": "text",
            "text": "以下是该图框的审查输入数据(JSON)：\n"
            + json.dumps(payload, ensure_ascii=False, indent=2),
        }
    ]
    image_b64 = evidence.get("image_b64")
    if image_b64:
        nb, nm = downsample_image_b64(image_b64, evidence.get("image_mime") or "image/png")
        user_content.append(LlmClient.image_content(nb, nm))

    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_content},
    ]

    tools_schema = ctx.tools.openai_tools(names=READ_ONLY_TOOLS)
    image_budget = [_MAX_OBS_IMAGES]

    for _ in range(max(0, ctx.max_iters)):
        resp = ctx.llm.chat(messages, tools=tools_schema)
        msg = resp.choices[0].message
        messages.append(msg)

        if not msg.tool_calls:
            return msg.content or ""

        for tool_call in msg.tool_calls:
            name = tool_call.function.name
            try:
                args = json.loads(tool_call.function.arguments or "{}")
            except json.JSONDecodeError:
                args = {}
            log(f"      [取证] {name} {args if args else ''}")
            content, image_budget = _run_readonly_tool(ctx, name, args, image_budget)
            messages.append(
                {"role": "tool", "tool_call_id": tool_call.id, "content": content}
            )

    # 达到回合上限：强制收口
    messages.append(
        {
            "role": "user",
            "content": "已达到取证回合上限，请基于现有信息直接给出该图框的审查结论。",
        }
    )
    resp = ctx.llm.chat(messages)
    return resp.choices[0].message.content or ""