Agent AtlasAI Agent 知识库
03.01架构模式 · Architecture

Agent 总体架构

感知、规划、行动、记忆、治理五层架构,以及一次请求在系统内部的完整生命周期。

架构不是画得漂亮,而是决定「当模型出错时,系统还能不能兜住」。

五层架构

一个生产级 Agent 可以拆成五层。越往下越稳定,越往上越贴近业务。

图 1 · Agent 的五层架构
交互层对话、表单、审批界面、流式输出、通知
编排层状态机、路由、循环、子 Agent、人工介入、持久化
能力层工具、检索、记忆、代码执行、浏览器、模型调用
状态层任务状态、检查点、事件日志、缓存、向量库
治理层鉴权、预算、限流、审计、评估、监控、告警
模型属于能力层的一个依赖,不是架构的中心。

很多 Demo 只有「能力层 + 一个 while 循环」。生产系统的大部分代码在编排、状态和治理三层。

一次请求的完整生命周期

以「帮我处理这张工单」为例:

text
1. 接入
   鉴权、租户识别、限流、PII 脱敏、输入分类

2. 初始化状态
   创建 task_id,写入 goal、user_context、budget

3. 组装上下文
   系统提示 + 可用工具 + 历史摘要 + 检索结果

4. 模型决策
   直接回答 / 调用工具 / 请求人工 / 结束

5. 工具执行
   参数校验 → 权限检查 → 执行 → 裁剪结果 → 记录 trace

6. 状态更新
   写入观察、更新计划、扣减预算、检查终止条件

7. 循环或结束
   未完成且未超限 → 回到 3;否则输出或转人工

8. 后处理
   输出校验、敏感信息过滤、引用补全、审计落库

注意第 8 步:模型输出不是最终产物。它需要经过与普通代码一样的校验和过滤,才能到达用户或下游系统。

状态模型:Agent 到底记住什么

一个可维护的 Agent 状态至少包含:

python
@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 循环。适合任务边界清晰、工具不多的场景。

text
while not done and steps < max:
    action = model(context)
    if action.is_final: break
    observation = execute(action)
    context.append(observation)

优点是简单、好调试。缺点是所有逻辑挤在一个上下文里,长任务容易跑偏。

图 / 状态机

把流程拆成节点和边,模型只在某些节点内决策:

text
classify → retrieve → draft → validate → approve → send
             ↑            |
             └── retry ───┘

每个节点有明确的输入输出、重试策略和超时。适合有合规要求、需要人工审批、步骤较多的业务。

多智能体

主管拆解任务,子 Agent 在各自上下文里执行。适合可以并行、子任务差异大的工作。代价是协调与错误传播。

三者不是互斥的。常见组合是:外层状态机 + 内层单 Agent + 少数子 Agent。

同步、异步与持久化

用户能等多久,决定架构:

时长交互方式技术
< 10 秒同步流式SSE / WebSocket
10 秒–2 分钟异步任务 + 进度队列 + 轮询/推送
数分钟–数天工作流持久化执行、检查点、通知
长期运行后台 Agent定时调度、事件触发、人工巡检

一旦任务超过 30 秒,就应该考虑异步化:立即返回 task_id,前端展示进度,完成后通知。否则用户会以为页面卡死,连接也容易超时。

持久化执行要求每一步都可恢复:

  • 状态落库,而不是只存在内存;
  • 工具调用有幂等键;
  • 检查点记录「已执行到哪」;
  • 恢复时跳过已完成的副作用;
  • 人工审批挂起时,进程可以安全退出。

失败边界

好的架构会明确「哪里失败会影响哪里」:

  • 模型失败:重试、换模型、降级到规则;
  • 工具失败:重试、换工具、返回可操作错误;
  • 检索失败:明确说资料不足,而不是编;
  • 校验失败:修复一次,仍失败则转人工;
  • 预算耗尽:停止并返回已完成部分;
  • 人工超时:按默认策略处理或关闭任务。

每条边界都要有对应的状态和用户可见的说明。不要把所有异常都变成一个「系统繁忙」。

一个最小参考实现

python
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、并发控制、权限、幂等、输出校验。但骨架就是这些。

设计时先回答的八个问题

  1. 任务的完成标准是什么?能不能写成断言?
  2. 哪些动作是只读,哪些有副作用,哪些不可逆?
  3. 最长允许多少步、多少秒、多少 token?
  4. 失败时用户看到什么?能不能重试?
  5. 哪些情况必须转人工?审批者需要看到什么?
  6. 状态存在哪里?进程崩溃后能否恢复?
  7. 如何知道它「做对了」?评估信号从哪来?
  8. 出问题时,能否回放到具体一步?

这八个问题答不上来,架构图再漂亮也只是 Demo。

小结

  • Agent 架构分交互、编排、能力、状态、治理五层,模型只是能力层的一部分。
  • 状态要显式化,消息历史不是唯一真相。
  • 控制平面决定系统能不能兜住失败:步数、超时、预算、权限、终止、人工出口。
  • 单循环、状态机、多智能体可以组合,外层越确定越好。
  • 超过 30 秒的任务应考虑异步与持久化。

下一篇拆开最核心的循环:ReAct 到底怎么跑,以及一个能运行的最小实现。

读完了?进度会保存在本地浏览器。
内容会持续修订。发现错误或想补充,欢迎反馈。 遇到问题?查排查手册 →