不要倒着建 AI Agent：从 Demo 到生产系统的组件化教程

用户请求
  ↓
任务入口：校验输入、定义目标、生成 request_id
  ↓
上下文准备器：读取必要资料，裁剪历史，形成 context_packet
  ↓
决策层 LLM：只输出结构化 decision
  ↓
编排器：根据 decision 调工具、重试、超时、路由
  ↓
工具执行层：窄工具、明确输入输出、无隐藏副作用
  ↓
状态存储：记录当前事实、版本、过期时间、读写人
  ↓
验证与可观测：trace、指标、人工确认、回滚依据

用户说：帮我处理退款。
Agent 读取聊天历史 → 自己判断是否退款 → 调 refund_api → 回复用户。

request_id: ticket-20260529-001
intent: refund_request
context_packet:
  user_id: u_123
  order_id: o_456
  policy_version: refund-policy-2026-05
  order_status: delivered
  risk_flags: []
decision_contract:
  allowed_actions: [ask_more_info, approve_refund_candidate, reject_with_reason]
  forbidden_actions: [direct_refund_without_policy_check]

{
  "action": "approve_refund_candidate",
  "reason": "order is within refund window and no risk flags were found",
  "needs_human_confirmation": true
}

用户问：某公司是否值得投资？
Agent 自己搜索网页 → 自己筛选 → 自己总结 → 自己给建议。

evidence_packet:
  question: 某公司是否值得投资？
  sources:
    - type: official_filing
      url: https://example.com/10q
      extracted_facts:
        - revenue_growth: "12% YoY"
        - operating_margin: "18%"
    - type: news
      url: https://example.com/news
      extracted_facts:
        - risk: "regulatory investigation opened"
  missing_evidence:
    - latest cash flow statement
    - management guidance transcript

{
  "answer_type": "insufficient_evidence",
  "supported_claims": [
    "Revenue grew 12% YoY in the extracted filing",
    "A regulatory investigation is reported by the cited news source"
  ],
  "unsupported_claims": [
    "No conclusion on valuation because current cash flow data is missing"
  ],
  "next_step": "fetch_latest_cash_flow_statement"
}

用户说：服务器好像卡了。
Agent 直接执行清理缓存、重启服务、删除日志。

uptime
free -h
df -h
systemctl --failed
journalctl -p err -n 50 --no-pager

diagnosis_card:
  host: app-01
  observed_at: 2026-05-29T20:00:00+08:00
  symptoms:
    - load_average_high
    - disk_usage_92_percent
  proposed_actions:
    - action: rotate_logs
      risk: medium
      requires_confirmation: true
    - action: restart_service
      risk: high
      requires_confirmation: true
  forbidden_without_confirmation:
    - rm -rf
    - systemctl restart production_service
    - database_delete_or_drop

from __future__ import annotations

from dataclasses import dataclass, asdict
import json


@dataclass
class Decision:
    action: str
    reason: str


@dataclass
class TraceEvent:
    step: str
    status: str
    detail: str


def prepare_context(user_text: str) -> dict[str, str]:
    return {
        "request": user_text,
        "policy": "read-only inspection before write actions",
    }


def model_decide(context: dict[str, str]) -> Decision:
    if "delete" in context["request"].lower():
        return Decision(action="deny", reason="destructive action requires explicit approval")
    return Decision(action="inspect", reason="safe read-only first step")


def run_tool(action: str) -> dict[str, str]:
    if action == "inspect":
        return {"result": "system looks healthy", "risk": "low"}
    if action == "deny":
        return {"result": "blocked", "risk": "high"}
    raise ValueError(f"unknown action: {action}")


def orchestrate(user_text: str) -> dict[str, object]:
    trace: list[TraceEvent] = []
    context = prepare_context(user_text)
    trace.append(TraceEvent("prepare_context", "ok", "context packet created"))

    decision = model_decide(context)
    trace.append(TraceEvent("model_decide", "ok", decision.reason))

    tool_result = run_tool(decision.action)
    trace.append(TraceEvent("run_tool", "ok", tool_result["result"]))

    return {
        "decision": asdict(decision),
        "tool_result": tool_result,
        "trace": [asdict(event) for event in trace],
    }


if __name__ == "__main__":
    print(json.dumps(orchestrate("please inspect disk usage"), ensure_ascii=False, indent=2))

python3 agent_orchestrator_demo.py

{
  "decision": {
    "action": "inspect",
    "reason": "safe read-only first step"
  },
  "tool_result": {
    "result": "system looks healthy",
    "risk": "low"
  },
  "trace": [
    {
      "step": "prepare_context",
      "status": "ok",
      "detail": "context packet created"
    }
  ]
}

上下文由谁准备？
状态由谁保存？
工具由谁调用？
重试由谁控制？
失败由谁处理？
验证由谁完成？
回滚由谁负责？

{
  "action": "inspect | ask_more_info | propose_write | stop",
  "reason": "why this action is justified",
  "required_evidence": ["evidence item"],
  "risk_level": "low | medium | high",
  "needs_human_confirmation": true
}

何时使用：
何时不要使用：
输入参数：
返回字段：
失败语义：
是否有外部副作用：

state_card:
  current_goal: "diagnose slow API responses"
  verified_facts:
    - "p95 latency increased from 300ms to 1200ms"
  assumptions:
    - "database may be slow"
  open_questions:
    - "is cache hit rate lower than usual?"
  next_allowed_steps:
    - "read metrics"
    - "read logs"
  forbidden_steps:
    - "restart production services without approval"

trace_event:
  request_id: "..."
  context_version: "..."
  decision: "..."
  tool_call: "..."
  tool_result: "..."
  validation_result: "..."
  final_status: "ok | stopped | needs_human"