#!/usr/bin/env python3 """capstone_skeleton.py — Helios Assist 参考架构的无 key 骨架模拟。 把 reference-architecture.md 的编排循环用伪组件跑通一遍： 网关 → 路由分诊 → 三条路径（L1 单次+RAG / L2 固定 workflow+审批门 / L3 受控 agent）→ 工具执行（含策略引擎）→ 人审门 → 降级链 → 升级队列。 设计说明： - 全部组件为伪实现（MockClaudeClient 按剧本出牌），零网络、零依赖、可直接运行。 - 真实接入点只有一处：ClaudeClient 接口。生产实现时先设置环境变量 ANTHROPIC_API_KEY，再把 MockClaudeClient 换成真 SDK 客户端即可，其余 编排/策略/降级逻辑不变。 - 演示的架构不变式（见 decision-workbook D15）： 1) 预算之内才有资格重试；2) 高危动作冻结待审、绝不绕过； 3) 无证据不回答；4) 每次降级都是可观测事件（TraceEvent）。 用法：python3 capstone_skeleton.py """ from __future__ import annotations import random import sys import time from dataclasses import dataclass, field random.seed(13) # 固定种子：jitter/退避可复现 # --------------------------------------------------------------------------- # 基础设施：追踪事件（审计面，对应 D11） # --------------------------------------------------------------------------- @dataclass class TraceEvent: session: str layer: str event: str detail: str = "" path: str = "-" # L1 / L2 / L3 / GW / ESC ... TRACE: list[TraceEvent] = [] def trace(session: str, layer: str, event: str, detail: str = "", path: str = "-") -> None: ev = TraceEvent(session, layer, event, detail, path) TRACE.append(ev) print(f" [{session}|{path}] {layer:<12} {event:<22} {detail}") # --------------------------------------------------------------------------- # 伪组件 1：网关（D14/D09）——认证/限流/入口筛查/出口脱敏 # --------------------------------------------------------------------------- class Gateway: """生产中是 API Gateway：OIDC 认证、token bucket 限流、注入筛查。""" INJECTION_MARKERS = ("ignore previous", "忽略以上指令", "disregard all") def admission(self, session: str, user: str, text: str) -> bool: trace(session, "Gateway", "AUTH_OK", f"user={user}") low = text.lower() hit = [m for m in self.INJECTION_MARKERS if m in low] if hit: trace(session, "Gateway", "INPUT_FLAGGED", f"marker={hit[0]}") return False trace(session, "Gateway", "ADMITTED") return True # --------------------------------------------------------------------------- # 伪组件 2：路由分诊（D06/D01）——Haiku 意图分类的剧本版 # --------------------------------------------------------------------------- @dataclass class RouteResult: path: str # L1 / L2 / L3 intent: str confidence: float class Router: """生产中是 Haiku 4.5 + 结构化输出；不确定时保守走高档路径。""" def classify(self, session: str, text: str) -> RouteResult: if "退款" in text or "refund" in text.lower(): res = RouteResult("L2", "refund_request", 0.82) elif "排查" in text or "迁移后" in text: res = RouteResult("L3", "complex_case", 0.74) else: res = RouteResult("L1", "faq", 0.93) trace(session, "Router", "CLASSIFIED", f"intent={res.intent} conf={res.confidence:.2f} -> {res.path}") return res # --------------------------------------------------------------------------- # 伪组件 3：检索层（D03/D07）——hybrid+rerank+租户 pre-filter 的剧本版 # --------------------------------------------------------------------------- @dataclass class Chunk: doc: str tenant: str valid_until: str text: str CORPUS = [ Chunk("policy-refunds-v3", "public", "2027-03-31", "年付客户可在购买后 30 天内申请全额退款；超过 30 天按剩余月份折算。" "单笔超过 $200 的退款需人工审批（政策 P12）。"), Chunk("wiki-connectors-v9", "public", "2027-06-30", "TS-1042：源系统连接超时。检查重试策略配置（控制台 → Sources → Retry），" "指数退避建议初始 2s、上限 30s。"), Chunk("ticket-9001-summary", "tenant-acme", "2026-12-31", "Acme 上次工单：3 月迁移后 12 个管道任务失败，根因是旧密钥未轮换。"), Chunk("policy-refunds-v1", "public", "2025-12-31", # 已作废的旧政策 "旧政策：所有退款一律秒批，无需审批。"), # 被 valid_until pre-filter 拦截 ] class RetrievalIndex: """metadata pre-filter：tenant 隔离 + 作废文档过滤，在候选集形成前发生。""" def search(self, session: str, query: str, tenant: str, top_k: int = 2) -> list[Chunk]: today = "2026-08-30" candidates = [c for c in CORPUS if c.tenant in ("public", tenant) and c.valid_until >= today] dropped = [c.doc for c in CORPUS if c not in candidates] trace(session, "Retrieval", "PRE_FILTER", f"candidates={len(candidates)} dropped={dropped or '[]'}") # 剧本化的相关性排序（生产中是 embeddings+BM25 → RRF → rerank） hits = [c for c in candidates if any(k in query for k in ("退款", "refund", "TS-1042"))] return (hits or candidates)[:top_k] # --------------------------------------------------------------------------- # 伪组件 4：Claude 客户端（D02/D01/D13）——剧本 + 故障注入 # --------------------------------------------------------------------------- class ApiError(Exception): def __init__(self, kind: str, status: int | None = None): self.kind = kind # "overload_5xx" / "timeout" / "invalid_4xx" self.status = status super().__init__(kind) @dataclass class ModelReply: text: str = "" tool_name: str = "" tool_args: dict = field(default_factory=dict) class MockClaudeClient: """生产实现：先设置环境变量 ANTHROPIC_API_KEY，再换真 SDK 客户端。""" def __init__(self, script: dict[tuple[str, int], ModelReply | ApiError]): self._script = script self._turn: dict[str, int] = {} def complete(self, session: str, model: str) -> ModelReply: n = self._turn.get(session, 0) self._turn[session] = n + 1 out = self._script.get((session, n), ModelReply(text="(默认回复)")) if isinstance(out, ApiError): raise out trace(session, "Claude", "REPLY", f"model={model} turn={n} text={out.text[:28]!r}" + (f" tool={out.tool_name}" if out.tool_name else "")) return out def reset(self) -> None: self._turn = {} # --------------------------------------------------------------------------- # 伪组件 5：工具层 + 策略引擎 + 人审门（D04/D07/D08） # --------------------------------------------------------------------------- ORDERS = {"ORD-77": {"amount": 480.0, "plan": "annual", "tenant": "tenant-acme"}} class ToolExecutor: """任务导向最小工具面；执行类动作（真正打款）不在此层——模型够不到。""" def __init__(self) -> None: self.approvals: list[dict] = [] def call(self, session: str, name: str, args: dict, tenant: str) -> dict: if name == "get_order": order = ORDERS.get(args.get("order_id", "")) if order and order["tenant"] != tenant: trace(session, "Tools", "TOOL_BLOCKED", "跨租户访问被拒（行级安全）") return {"is_error": True, "message": "无权访问该订单"} trace(session, "Tools", "GET_ORDER", f"{args.get('order_id')} -> {order}") return {"is_error": False, "order": order} if name == "request_refund": amount = float(args.get("amount", 0)) policy_hit = "P12: amount>200 requires human approval" frozen = {"status": "FROZEN_PENDING_APPROVAL", "amount": amount, "reason": args.get("reason"), "policy": policy_hit} trace(session, "Tools", "REFUND_REQUESTED", f"amount={amount} policy={policy_hit} -> 冻结待审") return {"is_error": False, "refund": frozen} trace(session, "Tools", "TOOL_ERROR", f"unknown tool {name}") return {"is_error": True, "message": f"未知工具 {name}"} class ApprovalGate: """人审门：坐席看结构化卡片批准/拒绝。不可用时动作冻结排队，绝不放行。""" def __init__(self, reviewer_decision: str): self.reviewer_decision = reviewer_decision self.audit: list[dict] = [] def review(self, session: str, refund: dict, user: str) -> str: decision = self.reviewer_decision card = {"order": refund, "evidence": "policy-refunds-v3#P12", "suggested": "approve", "context": "年付 30 天内"} self.audit.append({"session": session, "actor": "agent-007", "action": "REFUND_DECISION", "decision": decision, "basis": refund.get("policy"), "card": card}) trace(session, "ApprovalGate", decision.upper(), f"reviewer=agent-007 basis={refund.get('policy')} user={user}") return decision # --------------------------------------------------------------------------- # 伪组件 6：升级队列（D08） # --------------------------------------------------------------------------- class EscalationQueue: def __init__(self) -> None: self.items: list[dict] = [] def push(self, session: str, reason: str, handoff: dict) -> None: self.items.append({"session": session, "reason": reason, **handoff}) trace(session, "Escalation", "ESCALATED", f"reason={reason} handoff_keys={sorted(handoff)}", path="ESC") def __len__(self) -> int: return len(self.items) # --------------------------------------------------------------------------- # 编排器：预算 / 分类重试 / 降级链（D06/D13） # --------------------------------------------------------------------------- MAIN_MODEL, FALLBACK_MODEL = "claude-sonnet-5 (pinned)", "claude-haiku-4-5 (pinned)" class Orchestrator: def __init__(self, claude: MockClaudeClient, tools: ToolExecutor, gate: ApprovalGate, esc: EscalationQueue): self.claude = claude self.tools = tools self.gate = gate self.esc = esc self.budget_s = 20.0 self.clock = 0.0 # 模拟时钟（秒） def _spend(self, seconds: float) -> None: self.clock += seconds # -- 分类重试：只有 429/5xx/超时才有资格；预算耗尽立即降级 -------------- def _call_with_retry(self, session: str, model: str) -> ModelReply | None: for attempt in range(3): if self.clock > self.budget_s: trace(session, "Orchestrator", "BUDGET_EXHAUSTED", f"clock={self.clock:.0f}s") return None try: self._spend(3.0) return self.claude.complete(session, model) except ApiError as e: if e.kind == "invalid_4xx": trace(session, "Orchestrator", "NO_RETRY_4XX", "修复调用方，不重试") return None backoff = min(2 ** attempt, 4) + random.uniform(0, 0.5) trace(session, "Orchestrator", "RETRY_BACKOFF", f"{e.kind} attempt={attempt + 1} backoff={backoff:.1f}s") self._spend(backoff) return None # -- 降级链：强模型 → 弱模型 → 模板+表单 → 人工 ------------------------ def _call_with_fallback(self, session: str, script_key: str, template: str) -> tuple[str, str]: for model in (MAIN_MODEL, FALLBACK_MODEL): reply = self._call_with_retry(session, model) if reply is not None: return "ok" if model == MAIN_MODEL else "degraded_model", reply.text trace(session, "Orchestrator", "TEMPLATE_FALLBACK", template) return "degraded_template", template # -- L1：单次调用 + RAG（D06）----------------------------------------- def run_l1(self, session: str, text: str, tenant: str) -> str: chunks = self.index.search(session, text, tenant) evidence = " | ".join(c.doc for c in chunks) if not chunks: self.esc.push(session, "no_evidence", {"summary": "检索无结果"}) return "escalated" # 不变式③：无证据不回答 status, answer = self._call_with_fallback(session, f"l1:{text}", "（模板）常见问题解答暂时受限，已为您转接人工。") if status == "degraded_template": self.esc.push(session, "degraded_chain_exhausted", {"summary": "Claude API 不可用", "evidence": evidence}) return "escalated" return status # -- L2：固定 workflow + 审批门（D06/D07/D08）------------------------- def run_l2(self, session: str, tenant: str, user: str) -> str: # step1：取订单（模型发起工具调用，应用层执行） step1 = self._call_with_retry(session, MAIN_MODEL) if step1 is None: self.esc.push(session, "budget_or_api", {"summary": "退款流程无法启动"}) return "escalated" if step1.tool_name == "get_order": r1 = self.tools.call(session, "get_order", step1.tool_args, tenant) if r1["is_error"]: self.esc.push(session, "tool_blocked", {"summary": "订单访问被拒"}) return "escalated" amount = r1["order"]["amount"] # step2：创建退款申请（冻结态） r2 = self.tools.call(session, "request_refund", {"amount": amount, "reason": "annual_cancel"}, tenant) refund = r2["refund"] # 人审门（政策 P12：>200 必审） decision = self.gate.review(session, refund, user) if decision == "approve": trace(session, "Orchestrator", "REFUND_EXECUTED", "执行通道在工具层代码内（模型不可达）") return "resolved_approved" self.esc.push(session, "approval_rejected", {"summary": "坐席拒绝退款，直接接管", "refund": refund}) return "escalated" return "resolved_no_tool" # -- L3：受控 agent（步数上限 + 检查点 + 工具预算）--------------------- def run_l3(self, session: str, max_steps: int = 4) -> str: steps = 0 while steps < max_steps: steps += 1 reply = self._call_with_retry(session, MAIN_MODEL) if reply is None: self.esc.push(session, "budget_or_api", {"summary": "agent 中断"}) return "escalated" if reply.tool_name: r = self.tools.call(session, reply.tool_name, reply.tool_args, "tenant-acme") trace(session, "Orchestrator", "CHECKPOINT", f"step={steps} tool_result_is_error={r['is_error']}") continue trace(session, "Orchestrator", "AGENT_STOP", f"steps={steps} 目标达成") return "resolved_agent" trace(session, "Orchestrator", "STEP_LIMIT", f"steps={steps} 触发停止条件") self.esc.push(session, "step_limit", {"summary": "步数耗尽，升级人工"}) return "escalated" # --------------------------------------------------------------------------- # 场景驱动 # --------------------------------------------------------------------------- def main() -> int: print("=" * 78) print("Helios Assist 架构骨架模拟（无 key · 伪组件 · 可复现）") print("=" * 78) esc = EscalationQueue() results: list[tuple[str, str]] = [] # 场景 1：L1 FAQ —— 正常路径，租户过滤 + 缓存命中示意 print("\n--- 场景 1：L1 单次调用+RAG（FAQ：TS-1042 怎么办）---") claude1 = MockClaudeClient({ ("s1", 0): ModelReply( text="TS-1042 是源系统连接超时：检查 Retry 策略（指数退避 2s→30s）…" "（引用 wiki-connectors-v9）"), }) orch = Orchestrator(claude1, ToolExecutor(), ApprovalGate("approve"), esc) Gateway().admission("s1", "user-alice", "任务失败 TS-1042 怎么处理？") Router().classify("s1", "任务失败 TS-1042 怎么处理？") orch.index = RetrievalIndex() # 简化注入（演示用） results.append(("S1 L1 FAQ", orch.run_l1("s1", "TS-1042 连接超时", "tenant-acme"))) # 场景 2：L2 退款 $480 —— 工具 + 策略引擎 + 人审门（批准） print("\n--- 场景 2：L2 固定 workflow（退款 $480 > $200 → 人审批准）---") claude2 = MockClaudeClient({ ("s2", 0): ModelReply(tool_name="get_order", tool_args={"order_id": "ORD-77"}), ("s2", 1): ModelReply(tool_name="request_refund", tool_args={"amount": 480, "reason": "annual_cancel"}), ("s2", 2): ModelReply(text="您的退款申请已由坐席批准并执行，单号 RF-2031…"), }) orch2 = Orchestrator(claude2, ToolExecutor(), ApprovalGate("approve"), esc) Gateway().admission("s2", "user-bob", "年付套餐我要退款") Router().classify("s2", "年付套餐我要退款") results.append(("S2 L2 退款→人审→批准", orch2.run_l2("s2", "tenant-acme", "user-bob"))) # 场景 3：L1 + API 连续故障 —— 分类重试 → 降级链 → 模板 → 升级人工 print("\n--- 场景 3：Claude API 连续 5xx/超时 → 降级链打穿 → 升级 ---") claude3 = MockClaudeClient({ ("s3", 0): ApiError("overload_5xx", 529), ("s3", 1): ApiError("timeout"), ("s3", 2): ApiError("overload_5xx", 529), ("s3", 3): ApiError("timeout"), # 备模型也挂 ("s3", 4): ApiError("overload_5xx", 529), ("s3", 5): ApiError("timeout"), }) orch3 = Orchestrator(claude3, ToolExecutor(), ApprovalGate("approve"), esc) Gateway().admission("s3", "user-carol", "你们的 SLA 承诺是什么？") Router().classify("s3", "你们的 SLA 承诺是什么？") orch3.index = RetrievalIndex() results.append(("S3 API 故障→降级→升级", orch3.run_l1("s3", "SLA 承诺", "tenant-acme"))) # 场景 4：L2 退款 $480 但坐席拒绝 —— 审批否决 + 接管 print("\n--- 场景 4：退款 $480 → 人审拒绝 → 坐席接管 ---") claude4 = MockClaudeClient({ ("s4", 0): ModelReply(tool_name="get_order", tool_args={"order_id": "ORD-77"}), ("s4", 1): ModelReply(tool_name="request_refund", tool_args={"amount": 480, "reason": "annual_cancel"}), }) orch4 = Orchestrator(claude4, ToolExecutor(), ApprovalGate("reject"), esc) Gateway().admission("s4", "user-dave", "我要退款，马上") Router().classify("s4", "我要退款，马上") results.append(("S4 退款→人审拒绝", orch4.run_l2("s4", "tenant-acme", "user-dave"))) # 场景 5：L3 疑难 agent —— 多步工具 + 正常停止 print("\n--- 场景 5：L3 受控 agent（迁移后任务全挂：查订单→查工单→收敛停止）---") claude5 = MockClaudeClient({ ("s5", 0): ModelReply(tool_name="get_order", tool_args={"order_id": "ORD-77"}), ("s5", 1): ModelReply(tool_name="get_order", tool_args={"order_id": "ORD-77"}), ("s5", 2): ModelReply(text="根因：迁移后旧密钥未轮换（见工单 9001）。" "已建工单 T-5521，建议轮换密钥后重跑。"), }) orch5 = Orchestrator(claude5, ToolExecutor(), ApprovalGate("approve"), esc) Gateway().admission("s5", "user-eve", "迁移后所有管道任务都失败了，帮我排查") Router().classify("s5", "迁移后所有管道任务都失败了，帮我排查") results.append(("S5 L3 agent 收敛", orch5.run_l3("s5"))) # 汇总 print("\n" + "=" * 78) print("结果汇总") print("=" * 78) for name, status in results: print(f" {name:<28} -> {status}") print(f" 升级队列条目: {len(esc)} | 追踪事件总数: {len(TRACE)}") ok = (results[0][1] == "ok" and results[1][1] == "resolved_approved" and results[2][1] == "escalated" and results[3][1] == "escalated" and results[4][1] == "resolved_agent") print(f"\n断言检查: {'PASS' if ok else 'FAIL'}" "（L1 ok / L2 批准 / S3 升级 / S4 拒绝升级 / L3 收敛）") return 0 if ok else 1 if __name__ == "__main__": sys.exit(main())