136 lines
7.1 KiB
Python
136 lines
7.1 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
tests/real_llm_chain.py — 真实 LLM 全链路验证（Agent-C 任务 1）
|
||
==============================================================
|
||
|
||
链路：计划草稿生成 →（模拟）确认 → 真实 Pi headless 沙箱执行（真实 Kimi K2 模型）
|
||
→ 报告落 outbox（带 callId 凭证）→ validate_report_citations 校验。
|
||
|
||
凭证机制（真实模式）：编排器 on_tool_event 回调把 pi 事件流里真实发生的每次
|
||
tool_execution_start/end 登记进 ToolBridge（pi toolCallId ↔ 桥 callId 双向映射），
|
||
报告附录中的证据链引用这些真实 callId —— 与 mock 模式同一套校验代码。
|
||
|
||
用法（Git Bash，工作区根）：
|
||
.venv/Scripts/python.exe poc/pi-fallback/tests/real_llm_chain.py [--timeout 600]
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
from _common import load_llm_env, patch_child_env, POC_ROOT
|
||
|
||
sys.path.insert(0, str(POC_ROOT))
|
||
|
||
from orchestrator import BreakerConfig, Orchestrator # noqa: E402
|
||
from sandbox import create_run_dirs, new_run_id, validate_path, write_guard_extension # noqa: E402
|
||
from tool_bridge import ToolBridge # noqa: E402
|
||
from demo_task import prepare_inbox, REPORT_NAME, PLAN_NAME # noqa: E402
|
||
|
||
# pi 内置只读工具 → 桥注册表工具的映射（真实模式下 pi 侧工具名是 read/grep/find/ls）
|
||
PI_TOOL_MAP = {"read": "fs_read", "grep": "fs_read", "find": "fs_read", "ls": "fs_read"}
|
||
|
||
TASK = (
|
||
"你是一个数据分析 agent。当前工作目录的上一级有 inbox/05_orders.csv（订单数据，"
|
||
"列为 orderNo,productCode,quantity,dueDate,priority,customerName,customerLevel,status）。"
|
||
"请用 read 工具读取它（路径 ../inbox/05_orders.csv），分析："
|
||
"1) 订单状态(status)分布；2) 优先级(priority)分布；3) 客户等级(customerLevel)结构。"
|
||
"然后把完整的 Markdown 分析报告作为你的最后一条消息输出（不要尝试写文件，你没有写权限），"
|
||
"报告包含：数据概览、三项分布的具体数字、排产建议。"
|
||
)
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--timeout", type=float, default=600.0)
|
||
ap.add_argument("--model", default="moonshot/kimi-k2-0711-preview")
|
||
args = ap.parse_args()
|
||
|
||
creds = load_llm_env()
|
||
if not creds["LLM_API_KEY"]:
|
||
print("[real-chain] 未在 .env 找到 LLM_API_KEY，无法跑真实链路", file=sys.stderr)
|
||
return 2
|
||
patch_child_env(creds["LLM_API_KEY"])
|
||
print(f"[real-chain] 端点={creds['LLM_BASE_URL']} 模型={args.model} key=已注入子进程环境(未落盘)")
|
||
|
||
# —— propose：计划草稿 + 模拟确认 ——
|
||
run_id = new_run_id("realchain")
|
||
dirs = create_run_dirs(run_id)
|
||
run_dir = dirs["root"]
|
||
inbox_csv, source_note = prepare_inbox(dirs["inbox"])
|
||
(run_dir / PLAN_NAME).write_text(
|
||
f"# 执行计划草稿（propose 阶段产物）\n\n- runId: `{run_id}`\n- 任务: 真实 LLM 分析 inbox 订单数据\n"
|
||
f"- 数据来源: {source_note}\n- 模型: {args.model}\n\n"
|
||
"| 步 | 工具 | 权力 | 动作 |\n|----|------|------|------|\n"
|
||
"| 1 | read(→fs_read) | P0 | 读 inbox/05_orders.csv |\n"
|
||
"| 2 | (LLM 推理) | - | 状态/优先级/客户等级统计 |\n"
|
||
f"| 3 | fs_write(编排器侧) | P1 | 报告落 outbox/{REPORT_NAME} 并附证据链 |\n",
|
||
encoding="utf-8")
|
||
(run_dir / "确认记录.md").write_text(
|
||
f"# 模拟确认\n\nrunId: {run_id}\n计划已（模拟）人工确认，进入 execute 阶段。\n",
|
||
encoding="utf-8")
|
||
print(f"[real-chain] run_id={run_id} 素材={inbox_csv.name}（{source_note}）")
|
||
|
||
# —— execute：真实 pi headless，桥凭证随真实工具事件登记 ——
|
||
guard = write_guard_extension(run_dir)
|
||
bridge = ToolBridge(run_id, run_dir)
|
||
id_map: dict[str, str] = {} # pi toolCallId -> bridge callId
|
||
|
||
def on_tool_event(ev: dict) -> None:
|
||
tcid = ev.get("toolCallId", "")
|
||
if ev.get("type") == "tool_execution_start":
|
||
mapped = PI_TOOL_MAP.get(ev.get("toolName", ""), "fs_read")
|
||
id_map[tcid] = bridge.issue_call(mapped, ev.get("input") or {})
|
||
elif ev.get("type") == "tool_execution_end":
|
||
cid = id_map.get(tcid)
|
||
if cid:
|
||
bridge.complete_call(cid, {"isError": ev.get("isError", False)},
|
||
ok=not ev.get("isError", False))
|
||
|
||
orch = Orchestrator(BreakerConfig(timeout_sec=args.timeout))
|
||
runner = orch.pi_headless_runner(model=args.model, tools="read,grep,find,ls", guard_ext=guard)
|
||
res = orch.run(TASK, runner, run_id=run_id, on_tool_event=on_tool_event)
|
||
print(f"[real-chain] 编排结果: success={res.success} stop_reason={res.stop_reason} "
|
||
f"steps={res.steps} output={res.output_bytes}B elapsed={res.elapsed_sec:.1f}s")
|
||
if res.error_message:
|
||
print(f"[real-chain] error: {res.error_message[:300]}")
|
||
|
||
# —— 报告落账 + 证据链 + 凭证校验 ——
|
||
verdict = {"run_id": run_id, "success": res.success, "stop_reason": res.stop_reason,
|
||
"steps": res.steps, "elapsed_sec": round(res.elapsed_sec, 1)}
|
||
if res.success and res.final_text.strip():
|
||
evidence_lines = ["", "", "---", "", "## 证据链（桥层凭证，由编排器据真实事件流登记）", ""]
|
||
for tcid, cid in id_map.items():
|
||
evidence_lines.append(f"- 工具调用 `{tcid}` → 凭证 [callId: {cid}]")
|
||
cid_w = bridge.issue_call("fs_write", {"path": f"outbox/{REPORT_NAME}"})
|
||
target = validate_path(dirs["outbox"] / REPORT_NAME, run_dir, zone="outbox")
|
||
target.write_text(res.final_text + "\n".join(evidence_lines) + "\n", encoding="utf-8")
|
||
bridge.complete_call(cid_w, str(target))
|
||
cid_e = bridge.issue_call("report_emit", {"md": REPORT_NAME})
|
||
bridge.complete_call(cid_e, str(target))
|
||
|
||
v = bridge.validate_report_citations(target.read_text(encoding="utf-8"))
|
||
verdict.update({"report": str(target), "citation_valid": v["valid"],
|
||
"cited": len(v["cited"]), "issued": len(v["issued"]),
|
||
"missing": v["missing"]})
|
||
print(f"[real-chain] 报告: {target}")
|
||
print(f"[real-chain] 凭证校验: valid={v['valid']} cited={len(v['cited'])} "
|
||
f"issued={len(v['issued'])} missing={v['missing']}")
|
||
else:
|
||
verdict["report"] = None
|
||
print("[real-chain] 未产出报告（success=false 或 final_text 为空）", file=sys.stderr)
|
||
|
||
blocked = run_dir / "guard-blocked-calls.jsonl"
|
||
verdict["guard_blocked"] = blocked.read_text(encoding="utf-8").count("\n") if blocked.exists() else 0
|
||
(run_dir / "verdict.json").write_text(json.dumps(verdict, ensure_ascii=False, indent=2),
|
||
encoding="utf-8")
|
||
print(f"[real-chain] verdict: {json.dumps({k: v for k, v in verdict.items() if k != 'missing'}, ensure_ascii=False)}")
|
||
return 0 if (res.success and verdict.get("citation_valid")) else 2
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|