387 lines
18 KiB
Python
387 lines
18 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
demo_task.py — Pi Agent 兜底能力 P0 PoC：演示任务驱动脚本
|
||
==========================================================
|
||
|
||
两种模式：
|
||
|
||
- 默认（mock 模式，无需 LLM）：准备一个真实订单 CSV（demo-data/锐扬APS演示数据/
|
||
05_orders.csv）拷入 runs/<runId>/inbox/；生成《执行计划草稿》→ 模拟确认 →
|
||
用**纯 Python 的 mock executor**（产出与真实 pi 相同结构的事件流，因此走编排器
|
||
同一条熔断/日志/stopReason 判定路径）模拟 Pi 的步骤：读 inbox 数据 → 分析 →
|
||
写报告到 outbox（报告引用 bridge 签发的 callId 凭证）。随后跑 3 项围墙自检：
|
||
熔断（步数闸）/ 凭证（伪造 callId）/ 圈禁（路径越界）。
|
||
|
||
- `--real`：走真实 Pi headless 路径（Agent-C 用）：
|
||
`node runtime/.../cli.js -p --mode json --model <m> --tools read,grep,find,ls
|
||
-e <守卫扩展> "任务"`。无真实 LLM 网关时预期 stopReason=error 显式失败
|
||
（这正是本 PoC 要证明的「失败不包装成成功」）。
|
||
|
||
用法（Git Bash，工作区根目录）：
|
||
.venv/Scripts/python.exe poc/pi-fallback/demo_task.py # mock 全流程 + 自检
|
||
.venv/Scripts/python.exe poc/pi-fallback/demo_task.py --real # 真实 pi（需模型网关）
|
||
.venv/Scripts/python.exe poc/pi-fallback/demo_task.py --real --model aps-local/qwen3-32b-local --timeout 120
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import csv
|
||
import json
|
||
import random
|
||
import shutil
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
# 允许以脚本方式直接运行（poc/pi-fallback 加入 sys.path）
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
|
||
from orchestrator import BreakerConfig, Orchestrator
|
||
from sandbox import (POC_ROOT, SandboxViolation, create_run_dirs, new_run_id,
|
||
validate_path, write_guard_extension)
|
||
from tool_bridge import ToolBridge, ToolBridgeViolation
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 集中配置
|
||
# ---------------------------------------------------------------------------
|
||
|
||
WORKSPACE_ROOT = POC_ROOT.parent.parent # aps-agent 工作区根
|
||
DEMO_CSV = WORKSPACE_ROOT / "demo-data" / "锐扬APS演示数据" / "05_orders.csv"
|
||
FALLBACK_ROWS = 100 # 找不到素材时生成的模拟工单行数（明确标注模拟数据）
|
||
REPORT_NAME = "report.md" # 产物唯一出口文件名（outbox/）
|
||
PLAN_NAME = "执行计划草稿.md"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 素材准备
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def prepare_inbox(inbox: Path) -> tuple[Path, str]:
|
||
"""
|
||
把数据分析素材拷入 inbox（L2 可读注入区）。
|
||
返回 (素材路径, 来源说明)。找不到真实素材时生成 100 行模拟工单数据并明确标注。
|
||
"""
|
||
if DEMO_CSV.exists():
|
||
dst = inbox / DEMO_CSV.name
|
||
shutil.copy2(DEMO_CSV, dst)
|
||
return dst, f"真实演示数据（拷贝自 {DEMO_CSV}）"
|
||
# 兜底：生成模拟工单数据
|
||
dst = inbox / "simulated_workorders.csv"
|
||
rng = random.Random(42)
|
||
with open(dst, "w", newline="", encoding="utf-8-sig") as f:
|
||
w = csv.writer(f)
|
||
w.writerow(["woNo", "orderNo", "productCode", "quantity", "dueDate",
|
||
"priority", "status"])
|
||
for i in range(1, FALLBACK_ROWS + 1):
|
||
w.writerow([
|
||
f"WO-{i:04d}", f"SO-{rng.randint(1, 30):03d}",
|
||
f"P{rng.randint(1000, 9999)}", rng.randint(5, 200),
|
||
f"2026-08-{rng.randint(1, 28):02d}",
|
||
rng.randint(1, 6), rng.choice(["RELEASED", "IN_PROGRESS", "DONE"]),
|
||
])
|
||
return dst, "【模拟数据】未找到 demo-data 素材，脚本生成 100 行模拟工单"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# mock executor：模拟 Pi 会做的步骤（读 inbox → 分析 → 写 outbox 报告）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def make_mock_runner(run_id: str, run_dir: Path, inbox_csv: Path,
|
||
data_source_note: str, bridge: ToolBridge):
|
||
"""
|
||
构造 mock AgentRunner。事件结构与真实 pi headless JSONL 一致
|
||
（agent_start → tool_execution_start/end × N → message_end(stopReason) → agent_end），
|
||
因此编排器的熔断/日志/stopReason 判定是同一条代码路径。
|
||
|
||
每个"工具调用"都经 ToolBridge 签发 callId 并落 calls.jsonl，
|
||
报告里的成果声明逐条附 [callId: ...] 凭证。
|
||
"""
|
||
outbox = run_dir / "outbox"
|
||
|
||
def runner(task: str, work_dir: Path):
|
||
yield {"type": "session", "id": run_id, "mode": "mock"}
|
||
yield {"type": "agent_start"}
|
||
yield {"type": "turn_start"}
|
||
|
||
evidence: list[tuple[str, str]] = [] # (成果声明, callId)
|
||
|
||
def step(tool: str, params: dict, do):
|
||
"""一次模拟工具调用：签发 callId → 发 start 事件 → 执行 → 发 end 事件 → 落账。"""
|
||
call_id = bridge.issue_call(tool, params)
|
||
yield {"type": "tool_execution_start", "toolName": tool,
|
||
"toolCallId": call_id, "input": params}
|
||
try:
|
||
result = do()
|
||
bridge.complete_call(call_id, result, ok=True)
|
||
except Exception as exc:
|
||
bridge.complete_call(call_id, exc, ok=False)
|
||
raise
|
||
yield {"type": "tool_execution_end", "toolName": tool,
|
||
"toolCallId": call_id, "isError": False}
|
||
return call_id, result
|
||
|
||
# -- 步骤 1：fs_read 读 inbox 数据 ----------------------------------
|
||
def do_read():
|
||
# 读路径也过 L2 校验（mock 侧自觉走围墙）
|
||
safe = validate_path(inbox_csv, run_dir, zone="inbox", must_exist=True)
|
||
with open(safe, encoding="utf-8-sig", newline="") as f:
|
||
return list(csv.DictReader(f))
|
||
cid_read, rows = yield from _wrap(step("fs_read", {"path": str(inbox_csv)}, do_read))
|
||
evidence.append((f"读取注入区数据 {inbox_csv.name}，共 {len(rows)} 行", cid_read))
|
||
|
||
# -- 步骤 2：aps_query 做只读统计 -----------------------------------
|
||
def do_stats():
|
||
by_status: dict = {}
|
||
by_level: dict = {}
|
||
qty_total = 0
|
||
high_pri = 0
|
||
for r in rows:
|
||
by_status[r.get("status", "?")] = by_status.get(r.get("status", "?"), 0) + 1
|
||
lvl = r.get("customerLevel", "?")
|
||
if lvl:
|
||
by_level[lvl] = by_level.get(lvl, 0) + 1
|
||
try:
|
||
qty_total += int(r.get("quantity") or 0)
|
||
except ValueError:
|
||
pass
|
||
try:
|
||
if int(r.get("priority") or 0) >= 5:
|
||
high_pri += 1
|
||
except ValueError:
|
||
pass
|
||
return {"rows": len(rows), "by_status": by_status, "by_customer_level": by_level,
|
||
"qty_total": qty_total, "high_priority": high_pri}
|
||
cid_stats, stats = yield from _wrap(step("aps_query", {"sql_like": "select status,count(*) group by status"}, do_stats))
|
||
evidence.append((f"订单状态分布 {stats['by_status']}", cid_stats))
|
||
evidence.append((f"订单总量 {stats['qty_total']} 件，优先级>=5 的 {stats['high_priority']} 单", cid_stats))
|
||
|
||
# -- 步骤 3：fs_write + report_emit 写报告到 outbox ------------------
|
||
def do_write():
|
||
lines = [
|
||
"# Pi 兜底分析报告（mock 模式产物）",
|
||
"",
|
||
f"- runId: `{run_id}`",
|
||
f"- 任务: {task}",
|
||
f"- 数据来源: {data_source_note}",
|
||
"",
|
||
"## 成果声明（每条附 callId 凭证）",
|
||
"",
|
||
]
|
||
for claim, cid in evidence:
|
||
lines.append(f"- {claim} [callId: {cid}]")
|
||
lines += [
|
||
"",
|
||
"## 结论建议",
|
||
"",
|
||
"优先级 >= 5 的订单建议优先排产；B 级客户订单量大可关注交付风险。",
|
||
"（本报告由 mock executor 生成，用于验证编排/围墙/凭证链路，非真实 LLM 产出）",
|
||
]
|
||
target = validate_path(outbox / REPORT_NAME, run_dir, zone="outbox")
|
||
target.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||
return str(target)
|
||
cid_write, report_path = yield from _wrap(step("fs_write", {"path": str(outbox / REPORT_NAME)}, do_write))
|
||
yield from _wrap(step("report_emit", {"md": report_path}, lambda: report_path))
|
||
|
||
report_text = Path(report_path).read_text(encoding="utf-8")
|
||
|
||
# -- 收尾事件：stopReason=stop（编排器据此判成功） --------------------
|
||
assistant_msg = {
|
||
"role": "assistant", "stopReason": "stop",
|
||
"content": [{"type": "text", "text": report_text}],
|
||
}
|
||
yield {"type": "message_end", "message": assistant_msg}
|
||
yield {"type": "turn_end"}
|
||
yield {"type": "agent_end", "messages": [assistant_msg]}
|
||
|
||
# step() 是生成器函数包装：_wrap 让 "执行 + yield 事件" 能 return 值
|
||
return runner
|
||
|
||
|
||
def _wrap(gen):
|
||
"""驱动 step 生成器，透传 yield 的事件，取回其 return 值。"""
|
||
value = yield from gen
|
||
return value
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 围墙自检（mock 模式下验证熔断/凭证/圈禁各一条）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def self_checks() -> dict:
|
||
"""
|
||
3 项自检，对应 GOAL.md 的 3 个越狱测试判据：
|
||
1) 熔断：步数闸触发 → 运行显式 failed；
|
||
2) 凭证：报告引用伪造 callId → validate 判 invalid；
|
||
3) 圈禁：越界路径（../../server）→ validate_path 拒绝。
|
||
"""
|
||
results = {}
|
||
|
||
# 1) 熔断：步数上限
|
||
def runaway_runner(task, work_dir):
|
||
yield {"type": "agent_start"}
|
||
for i in range(20): # 远超 max_steps=3
|
||
yield {"type": "tool_execution_start", "toolName": "fs_read",
|
||
"toolCallId": f"runaway-{i}", "input": {}}
|
||
yield {"type": "tool_execution_end", "toolName": "fs_read",
|
||
"toolCallId": f"runaway-{i}", "isError": False}
|
||
yield {"type": "agent_end", "messages": []}
|
||
orch = Orchestrator(BreakerConfig(max_steps=3, timeout_sec=60))
|
||
r = orch.run("runaway self-check", runaway_runner, run_id=new_run_id("selfcheck-breaker"))
|
||
results["breaker_step_limit"] = {
|
||
"pass": (not r.success) and "max_steps" in r.stop_reason,
|
||
"detail": f"success={r.success} stop_reason={r.stop_reason} steps={r.steps}",
|
||
}
|
||
|
||
# 2) 凭证：伪造 callId 必须判 invalid
|
||
run_id = new_run_id("selfcheck-credential")
|
||
dirs = create_run_dirs(run_id)
|
||
bridge = ToolBridge(run_id, dirs["root"])
|
||
real_cid = bridge.issue_call("fs_read", {"path": "inbox/x.csv"})
|
||
bridge.complete_call(real_cid, "ok")
|
||
good = bridge.validate_report_citations(f"成果 A [callId: {real_cid}]")
|
||
forged = "call-" + "deadbeef-0000-0000-0000-000000000000"
|
||
bad = bridge.validate_report_citations(f"成果 B [callId: {forged}]")
|
||
results["credential_forgery"] = {
|
||
"pass": good["valid"] and (not bad["valid"]) and bad["missing"] == [forged],
|
||
"detail": f"真凭证 valid={good['valid']}；伪造凭证 valid={bad['valid']} missing={bad['missing']}",
|
||
}
|
||
|
||
# 3) 圈禁：越界路径必须拒绝
|
||
run_id = new_run_id("selfcheck-sandbox")
|
||
dirs = create_run_dirs(run_id)
|
||
checks = []
|
||
try:
|
||
validate_path("../../server/evil.py", dirs["root"])
|
||
checks.append("越界未拦截!")
|
||
except SandboxViolation:
|
||
checks.append("越界已拦截")
|
||
try:
|
||
validate_path(dirs["outbox"] / "ok.md", dirs["root"], zone="outbox")
|
||
checks.append("界内放行")
|
||
except SandboxViolation as e:
|
||
checks.append(f"界内误杀: {e}")
|
||
results["sandbox_escape"] = {
|
||
"pass": checks == ["越界已拦截", "界内放行"],
|
||
"detail": "; ".join(checks),
|
||
}
|
||
return results
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 主流程
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def run_mock(task: str) -> int:
|
||
"""mock 模式完整流程：建目录 → 备料 → 计划草稿 → 模拟确认 → 编排执行 → 凭证校验 → 自检。"""
|
||
run_id = new_run_id()
|
||
dirs = create_run_dirs(run_id)
|
||
run_dir = dirs["root"]
|
||
print(f"[demo] run_id={run_id}")
|
||
|
||
inbox_csv, source_note = prepare_inbox(dirs["inbox"])
|
||
print(f"[demo] inbox 素材: {inbox_csv.name}（{source_note}）")
|
||
|
||
# —— 产出《执行计划草稿》（方案 §4.4 agent.fallback.propose，沙盒语义不写主干）——
|
||
plan = f"""# 执行计划草稿（propose 阶段产物）
|
||
|
||
- runId: `{run_id}`
|
||
- 任务: {task}
|
||
- 数据来源: {source_note}
|
||
|
||
## 计划步骤（卡片上的承诺即执行上限）
|
||
|
||
| 步 | 工具 | 权力 | 动作 |
|
||
|----|------|------|------|
|
||
| 1 | fs_read | P0 | 读 inbox/{inbox_csv.name} |
|
||
| 2 | aps_query | P0 | 状态/优先级/客户等级只读统计 |
|
||
| 3 | fs_write | P1 | 写报告到 outbox/{REPORT_NAME} |
|
||
| 4 | report_emit | P1 | 报告走凭证校验出口 |
|
||
|
||
## 预计影响面
|
||
只读 inbox，仅写 outbox/{REPORT_NAME}；不触主干数据；无需 checkpoint 回滚。
|
||
"""
|
||
(run_dir / PLAN_NAME).write_text(plan, encoding="utf-8")
|
||
|
||
# —— 模拟确认（产品态为 P2 确认卡；PoC 落一个确认记录文件）——
|
||
(run_dir / "确认记录.md").write_text(
|
||
f"# 模拟确认\n\nrunId: {run_id}\n计划已（模拟）人工确认，进入 execute 阶段。\n",
|
||
encoding="utf-8")
|
||
print(f"[demo] 《{PLAN_NAME}》已生成并模拟确认")
|
||
|
||
# —— 编排执行（mock executor 走编排器同一条熔断/日志/判定路径）——
|
||
bridge = ToolBridge(run_id, run_dir)
|
||
runner = make_mock_runner(run_id, run_dir, inbox_csv, source_note, bridge)
|
||
orch = Orchestrator(BreakerConfig()) # 默认三闸：10min / 50 步 / 10MiB
|
||
t0 = time.time()
|
||
res = orch.run(task, runner, run_id=run_id)
|
||
|
||
print(f"[demo] 编排完成: success={res.success} stop_reason={res.stop_reason} "
|
||
f"steps={res.steps} output={res.output_bytes}B elapsed={res.elapsed_sec:.2f}s")
|
||
|
||
# —— 成果校验：报告引用的 callId 必须全部真实存在（方案 §4.5）——
|
||
report_path = dirs["outbox"] / REPORT_NAME
|
||
if res.success and report_path.exists():
|
||
v = bridge.validate_report_citations(report_path.read_text(encoding="utf-8"))
|
||
print(f"[demo] 凭证校验: valid={v['valid']} cited={len(v['cited'])} "
|
||
f"issued={len(v['issued'])} missing={v['missing']}")
|
||
if not v["valid"]:
|
||
print("[demo] ⚠️ 报告含无凭证的成果声明 → 判 invalid", file=sys.stderr)
|
||
return 2
|
||
else:
|
||
print("[demo] ⚠️ 编排失败或无报告产物", file=sys.stderr)
|
||
return 2
|
||
|
||
# —— 围墙自检 ——
|
||
checks = self_checks()
|
||
all_pass = True
|
||
print("[demo] 围墙自检:")
|
||
for name, r in checks.items():
|
||
print(f" [{'PASS' if r['pass'] else 'FAIL'}] {name}: {r['detail']}")
|
||
all_pass = all_pass and r["pass"]
|
||
|
||
print(f"[demo] 运行目录: {run_dir}")
|
||
print(f"[demo] 报告产物: {report_path}")
|
||
print(f"[demo] 总耗时 {time.time() - t0:.1f}s，自检 {'全部通过' if all_pass else '存在失败'}")
|
||
return 0 if all_pass else 3
|
||
|
||
|
||
def run_real(task: str, model: str, timeout: float) -> int:
|
||
"""--real 模式：真实 pi headless。无 LLM 网关时预期 stopReason=error 显式失败。"""
|
||
run_id = new_run_id("real")
|
||
dirs = create_run_dirs(run_id)
|
||
inbox_csv, source_note = prepare_inbox(dirs["inbox"])
|
||
guard = write_guard_extension(dirs["root"])
|
||
print(f"[real] run_id={run_id} 素材={inbox_csv.name}（{source_note}）")
|
||
print(f"[real] 守卫扩展: {guard}")
|
||
|
||
orch = Orchestrator(BreakerConfig(timeout_sec=timeout))
|
||
runner = orch.pi_headless_runner(model=model, tools="read,grep,find,ls", guard_ext=guard)
|
||
full_task = (f"{task}\n数据文件在 inbox/{inbox_csv.name}（相对当前目录上一级），"
|
||
f"分析报告写到 outbox/{REPORT_NAME}。")
|
||
res = orch.run(full_task, runner, run_id=run_id)
|
||
print(f"[real] 结果: success={res.success} stop_reason={res.stop_reason} "
|
||
f"steps={res.steps} elapsed={res.elapsed_sec:.1f}s")
|
||
if res.error_message:
|
||
print(f"[real] error: {res.error_message[:300]}")
|
||
print(f"[real] 运行目录: {dirs['root']}（events.jsonl / orchestrator.log / result.json）")
|
||
# 显式失败也是成功交付：返回码只在 harness 自身错误时非零
|
||
return 0
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(description="Pi 兜底 PoC 演示任务驱动")
|
||
ap.add_argument("--real", action="store_true", help="走真实 Pi headless（需模型网关）")
|
||
ap.add_argument("--model", default="aps-local/qwen3-32b-local", help="--real 用模型")
|
||
ap.add_argument("--timeout", type=float, default=600.0, help="超时秒数（默认 600）")
|
||
ap.add_argument("--task", default="分析注入区的订单数据：状态分布、优先级分布、客户等级结构，给出排产建议",
|
||
help="任务简报")
|
||
args = ap.parse_args()
|
||
if args.real:
|
||
return run_real(args.task, args.model, args.timeout)
|
||
return run_mock(args.task)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|