架构不是画得漂亮,而是决定「当模型出错时,系统还能不能兜住」。
五层架构
一个生产级 Agent 可以拆成五层。越往下越稳定,越往上越贴近业务。
很多 Demo 只有「能力层 + 一个 while 循环」。生产系统的大部分代码在编排、状态和治理三层。
一次请求的完整生命周期
以「帮我处理这张工单」为例:
1. 接入
鉴权、租户识别、限流、PII 脱敏、输入分类
2. 初始化状态
创建 task_id,写入 goal、user_context、budget
3. 组装上下文
系统提示 + 可用工具 + 历史摘要 + 检索结果
4. 模型决策
直接回答 / 调用工具 / 请求人工 / 结束
5. 工具执行
参数校验 → 权限检查 → 执行 → 裁剪结果 → 记录 trace
6. 状态更新
写入观察、更新计划、扣减预算、检查终止条件
7. 循环或结束
未完成且未超限 → 回到 3;否则输出或转人工
8. 后处理
输出校验、敏感信息过滤、引用补全、审计落库注意第 8 步:模型输出不是最终产物。它需要经过与普通代码一样的校验和过滤,才能到达用户或下游系统。
状态模型:Agent 到底记住什么
一个可维护的 Agent 状态至少包含:
@dataclass
class AgentState:
task_id: str
goal: str
status: Literal["planning", "running", "waiting_human", "done", "failed"]
plan: list[Step]
current_step: int
facts: list[Fact] # 带来源的已确认事实
artifacts: dict[str, str] # 文件、草稿、报告
messages: list[Message] # 最近对话
budget: Budget # token / 金额 / 步数 / 时间
errors: list[ErrorRecord] # 失败尝试,避免重复
trace_id: str关键在于把状态显式化。如果所有信息都埋在消息列表里,你无法:
- 判断任务进行到哪;
- 从中断处恢复;
- 统计每一步的成本;
- 让人工审批者快速理解上下文;
- 在失败时只重跑某一步。
状态显式化之后,消息历史只是一种视图,而不是唯一真相。
控制平面与数据平面
借用网络架构的概念:
- 数据平面:模型调用、工具执行、检索、文件读写——真正干活的部分;
- 控制平面:决定谁在什么时候做什么——路由、预算、权限、重试、终止。
Agent 出问题,往往是控制平面太薄:
| 控制平面缺失 | 后果 |
|---|---|
| 没有最大步数 | 死循环、成本失控 |
| 没有超时 | 请求悬挂、连接耗尽 |
| 没有权限检查 | 越权读写 |
| 没有预算 | 单个任务烧掉整月额度 |
| 没有终止条件 | 模型「觉得」还没完成,一直查 |
| 没有人工出口 | 遇到无法处理的情况只能失败 |
把控制平面当成一等公民,架构才算完整。
三种编排形态
单循环
一个模型、一组工具、一个 while 循环。适合任务边界清晰、工具不多的场景。
while not done and steps < max:
action = model(context)
if action.is_final: break
observation = execute(action)
context.append(observation)优点是简单、好调试。缺点是所有逻辑挤在一个上下文里,长任务容易跑偏。
图 / 状态机
把流程拆成节点和边,模型只在某些节点内决策:
classify → retrieve → draft → validate → approve → send
↑ |
└── retry ───┘每个节点有明确的输入输出、重试策略和超时。适合有合规要求、需要人工审批、步骤较多的业务。
多智能体
主管拆解任务,子 Agent 在各自上下文里执行。适合可以并行、子任务差异大的工作。代价是协调与错误传播。
三者不是互斥的。常见组合是:外层状态机 + 内层单 Agent + 少数子 Agent。
同步、异步与持久化
用户能等多久,决定架构:
| 时长 | 交互方式 | 技术 |
|---|---|---|
| < 10 秒 | 同步流式 | SSE / WebSocket |
| 10 秒–2 分钟 | 异步任务 + 进度 | 队列 + 轮询/推送 |
| 数分钟–数天 | 工作流 | 持久化执行、检查点、通知 |
| 长期运行 | 后台 Agent | 定时调度、事件触发、人工巡检 |
一旦任务超过 30 秒,就应该考虑异步化:立即返回 task_id,前端展示进度,完成后通知。否则用户会以为页面卡死,连接也容易超时。
持久化执行要求每一步都可恢复:
- 状态落库,而不是只存在内存;
- 工具调用有幂等键;
- 检查点记录「已执行到哪」;
- 恢复时跳过已完成的副作用;
- 人工审批挂起时,进程可以安全退出。
失败边界
好的架构会明确「哪里失败会影响哪里」:
- 模型失败:重试、换模型、降级到规则;
- 工具失败:重试、换工具、返回可操作错误;
- 检索失败:明确说资料不足,而不是编;
- 校验失败:修复一次,仍失败则转人工;
- 预算耗尽:停止并返回已完成部分;
- 人工超时:按默认策略处理或关闭任务。
每条边界都要有对应的状态和用户可见的说明。不要把所有异常都变成一个「系统繁忙」。
一个最小参考实现
class Agent:
def __init__(self, tools, model, max_steps=8, budget=None):
self.tools = {t.name: t for t in tools}
self.model = model
self.max_steps = max_steps
self.budget = budget or Budget(tokens=60_000, seconds=120)
def run(self, goal: str, state: AgentState) -> AgentState:
state.goal = goal
for step in range(self.max_steps):
if state.budget.exceeded():
state.status = "failed"
state.errors.append(ErrorRecord("BUDGET_EXCEEDED", "预算耗尽"))
return state
ctx = build_context(state)
action = self.model.decide(ctx, tools=self.tools.values())
state.budget.consume(action.usage)
if action.kind == "final":
state.status = "done"
state.artifacts["answer"] = action.text
return state
if action.kind == "ask_human":
state.status = "waiting_human"
return state
result = self.execute(action, state)
state.observe(action, result)
state.status = "failed"
state.errors.append(ErrorRecord("MAX_STEPS", "超过最大步数"))
return state真正的工程代码还会加上:持久化、trace、并发控制、权限、幂等、输出校验。但骨架就是这些。
设计时先回答的八个问题
- 任务的完成标准是什么?能不能写成断言?
- 哪些动作是只读,哪些有副作用,哪些不可逆?
- 最长允许多少步、多少秒、多少 token?
- 失败时用户看到什么?能不能重试?
- 哪些情况必须转人工?审批者需要看到什么?
- 状态存在哪里?进程崩溃后能否恢复?
- 如何知道它「做对了」?评估信号从哪来?
- 出问题时,能否回放到具体一步?
这八个问题答不上来,架构图再漂亮也只是 Demo。
小结
- Agent 架构分交互、编排、能力、状态、治理五层,模型只是能力层的一部分。
- 状态要显式化,消息历史不是唯一真相。
- 控制平面决定系统能不能兜住失败:步数、超时、预算、权限、终止、人工出口。
- 单循环、状态机、多智能体可以组合,外层越确定越好。
- 超过 30 秒的任务应考虑异步与持久化。
下一篇拆开最核心的循环:ReAct 到底怎么跑,以及一个能运行的最小实现。