[!info] related notes

• 所属 MOC: Agent MOC
• 前置概念: Agent, Agentic Systems, Agent Runtime, Agent 执行闭环
• 并列概念: Agentic Workflow Patterns, 面向 Agent 的工具设计, AI Agent Memory Layer
• 关系笔记: AI Agent 创建实践, Python Agent API 服务, 多服务架构
• 参考来源: Anthropic - Building Effective Agents, OpenAI Agents SDK, Vercel AI SDK, Langfuse

Agent 项目系统架构设计

一句话定义

Agent 项目系统架构是把 LLM、工具、状态、记忆和可观测性组装成一个可运行、可维护、可演进的生产系统的分层设计方法——核心决策不是”用什么框架”，而是”从哪一档抽象开始”。

第一步决策：三档抽象选型

在画架构图之前，先回答一个问题：你的任务需要多复杂的 agent？

档位	抽象层级	代表实现	适用场景
Raw API	直接调用 LLM API + tool schema	OpenAI Responses API, Anthropic Messages API	单轮任务、最大控制权、快速验证
Managed Loop	框架管理执行循环 + 工具分发	OpenAI Agent, Vercel AI SDK	标准 tool-use agent、多轮对话
Graph	显式状态机 + 条件路由	LangGraph StateGraph	复杂分支、循环、human-in-the-loop

[!important] Anthropic 的核心建议 “Start by using LLM APIs directly: many patterns can be implemented in a few lines of code. Frameworks create extra layers of abstraction that obscure what is happening.”

大多数团队应该从 Raw API 开始，只有当 tool 数量超过 5 个、或需要多 agent 协作时，才升级到 Managed Loop。Graph 只在真正有环的状态机时才需要。

错误的起步：上来就引入 LangGraph/CrewAI，结果 80% 的时间在理解框架概念而非解决业务问题。

六层架构（当需要完整系统时）

一旦确定了抽象档位，完整的 agent 项目可以拆成六层：

┌─────────────────────────────────────────────────────┐
│                   API Gateway                        │
│         认证 · 限流 · 请求路由 · 响应格式化            │
├─────────────────────────────────────────────────────┤
│               Agent Orchestrator                     │
│       多 agent 路由 · 任务拆分 · 协作编排               │
├─────────────────────────────────────────────────────┤
│                 Agent Runtime                        │
│     执行循环 · 状态机 · 停止条件 · 审批检查点             │
├─────────────────────────────────────────────────────┤
│              LLM Provider Layer                      │
│     模型抽象 · 流式输出 · 重试 · token 计量             │
├─────────────────────────────────────────────────────┤
│           Tool & Integration Layer                   │
│      工具注册 · 执行沙箱 · 结果标准化 · 错误映射          │
├─────────────────────────────────────────────────────┤
│          State & Persistence Layer                   │
│    会话状态 · 工作记忆 · 对话历史 · 长期记忆 · 向量存储    │
└─────────────────────────────────────────────────────┘
         ↕ 所有层依赖 ↕
    Observability Layer (tracing · 指标 · 评估 · 成本)

关键约束：依赖方向从外到内，内层不依赖外层。Observability 是横切关注点，通过注入而非直接引用来接入每一层。

各层职责与设计要点

1. API Gateway

职责：对外暴露 HTTP/WebSocket/SSE 端点，处理认证、限流、请求分发。

设计要点：

• 认证在 gateway 层完成，下游只接收已验证的 user context，不再重复校验
• 流式响应（SSE）在 gateway 层透传，不要在这里缓冲完整响应再返回
• 请求体做 schema 校验，非法请求在入口就拒绝，不要让脏数据深入 runtime
• 每个请求带 request ID，贯穿所有下游日志

反模式：

• ❌ 把 agent 逻辑直接写在 HTTP handler 里
• ❌ gateway 层做业务决策（应该路由到哪个 agent）

2. Agent Orchestrator

职责：决定”这个任务交给谁”——单 agent 直接执行，多 agent 时做任务拆分和协作编排。

五种编排模式及选型：

模式	机制	最佳场景	复杂度
Manager-as-tools	主 agent 把子 agent 当 tool 调用	层级任务、专家分工	低
Handoffs	agent A 主动移交控制权给 agent B	客服分流、多领域助手	中
Sequential pipeline	固定顺序，前一个的输出是后一个的输入	研究→起草→审核→发布	低
Orchestrator-workers	中央 LLM 动态拆解任务、分配给 worker	无法预先确定子任务	中
Graph	有向图 + 条件边，支持循环和分支	复杂状态机、HITL 检查点	高

[!tip] 选型建议

1. 大多数场景用 Manager-as-tools——最简单、最可调试、每个子 agent 独立可测
2. 需要 agent 间传递状态时用 Handoffs
3. 任务拆解不可预测时用 Orchestrator-workers
4. 只有真正有环的状态机才用 Graph
5. 固定顺序的 pipeline 不需要框架，一个 for 循环就够了

反模式：单 agent + 更好的工具就能解决的问题，不要上多 agent。一个有 15 个工具的 agent 几乎总是优于 3 个各有 5 个工具的 agent——因为单 agent 能整体推理工具间的交互。

什么时候不需要这层：

• 只有一个 agent，没有路由需求
• 用 Dify/LangGraph 等框架，框架本身承担了编排职责

3. Agent Runtime

职责：驱动单个 agent 的执行闭环——接收输入、调用 LLM、分发工具调用、收集结果、判断是否继续。

设计要点：

• runtime 是一个状态机，不是一段线性代码。核心循环：

async def run(self, input: AgentInput) -> AgentOutput:
    state = self.init_state(input)
    for turn in range(self.max_turns):
        llm_output = await self.call_llm(state.context)
        if llm_output.has_tool_calls:
            tool_results = await self.execute_tools(llm_output.tool_calls)
            state.context.append(tool_results)
        else:
            return self.finalize(llm_output)
    return self.force_stop(state)  # 硬上限，不依赖模型自行停止

• 停止条件必须显式定义：模型输出了最终回答？达到最大轮次？token 预算耗尽？触发安全阀？
• 审批检查点（approval checkpoints）在 tool 执行前拦截，不是在 tool 内部判断
• runtime 不应该知道具体工具的实现细节，只通过 Tool Layer 的统一接口调用
• streaming 不是可选的——从 LLM 到用户的流式输出必须是默认行为，不是后加的功能

关键接口：

class AgentRuntime:
    async def run(self, input: AgentInput) -> AgentOutput:
        """一次完整的 agent 执行，从输入到最终输出"""
        ...

    async def step(self, state: AgentState) -> AgentState:
        """单步执行：调 LLM → 处理输出 → 更新状态"""
        ...

4. LLM Provider Layer

职责：抽象底层模型调用，屏蔽不同 provider 的 API 差异。

设计要点：

• 统一接口：不管用 OpenAI、Anthropic 还是本地模型，调用方式一致
• 流式输出是一等公民，不要先完整拿到再返回——用户等待时间和首 token 时间差距巨大
• Function calling 的 tool_call 增量到达时，必须按 index 累积，不能只取最后一块
• 重试策略区分：429 限流可重试，400 参数错误不可重试，500 有限次数重试
• token 计量在这一层完成，向上层暴露 usage 信息，让 runtime 可以做上下文裁剪决策
• model fallback chain：主模型失败时自动降级到备用模型

关键接口：

class LLMProvider(Protocol):
    async def chat_stream(
        self,
        messages: list[Message],
        tools: list[ToolDef] | None = None,
    ) -> AsyncIterator[StreamChunk]:
        """流式返回，每个 chunk 可能是 text delta 或 tool_call delta"""
        ...

5. Tool & Integration Layer

职责：注册、执行、标准化所有外部工具的调用。

设计要点：

• 工具注册是声明式的：name、description、parameters schema、execute function
• 工具执行结果标准化为统一格式：{success, data, error, metadata}
• 每个工具有独立的超时和重试配置，不是全局一刀切
• 工具不知道 agent 的存在——它们是纯函数或独立服务调用，不依赖 agent 状态
• 危险工具（写文件、发邮件、执行命令）需要 approval checkpoint，这由 runtime 层控制，工具层只负责”被批准后执行”

[!warning] Tool 接口设计是 agent 的可靠性表面 Anthropic 在 SWE-bench 中发现：模型在目录切换后无法正确使用相对路径，切换到绝对路径后成功率显著提升。工具文档的质量直接决定 agent 的成功率。

检验标准：如果一个人类工程师看不懂你的工具描述，模型也不可能稳定调用它。

错误处理分层：

错误类型	谁处理	怎么处理
参数校验失败	Tool Layer	返回结构化错误，让 LLM 自行修正
外部服务超时	Tool Layer	重试 → 返回超时错误给 LLM
权限不足	Tool Layer	返回明确错误信息
工具不存在	Runtime	停止执行，报告给用户

关键原则：tool 的错误应该回填到 context 中让 LLM 决定下一步，而不是直接 crash agent。LLM 在拿到结构化错误信息后，有很强的自我修正能力。

6. State & Persistence Layer

职责：管理 agent 运行所需的所有状态——短期会话、工作记忆、对话历史、长期记忆。

四层状态分离：

层级	内容	存储位置	生命周期	特征
会话状态	当前轮次的 tool call、中间结果	内存	一次 run	不可 checkpoint，run 结束即丢
工作记忆	任务进度、中间决策、runtimeContext	内存 + checkpoint	一个 session	可在任意节点快照，支持 resume
对话历史	多轮消息、agent 输出	数据库	跨 session	source of truth，不是内存 list
长期记忆	用户偏好、项目知识、实体记忆	向量库 + KV	永久	通过 tool 检索，不直接塞入 prompt

工作记忆的两种优雅设计：

Vercel AI SDK 模式（推荐）：

• runtimeContext：整个 agent loop 共享的状态（租户设置、request ID、任务进度），step 间可变
• toolsContext：每个 tool 有自己独立的 typed context，tool 之间互不可见——隔离防止状态污染

LangGraph 模式：

• typed StateGraph：每个节点读写共享状态对象，reducer 控制合并方式
• 每次节点转换都 checkpoint 整个状态——代价是存储，收益是任意节点可 resume

[!tip] 状态管理核心原则

1. 对话历史是 source of truth，不是内存中的 list
2. 长期记忆通过 tool 检索后才进入上下文——不要把所有记忆塞进 prompt
3. 上下文窗口管理在这一层做：截断旧消息 / 摘要压缩 / 滑动窗口
4. 不要用一个大 dict 传所有状态——typed + scoped 的状态容器防止隐式耦合

数据流：一次请求的完整路径

用户发送消息
    │
    ▼
API Gateway ── 认证、限流、schema 校验、注入 request ID
    │
    ▼
Orchestrator ── 判断路由（哪个 agent）
    │
    ▼
Runtime ── 加载对话历史 + 检索长期记忆 → 组装 context
    │
    ├──▶ LLM Provider ── 调用模型，流式返回
    │        │
    │        ▼
    │    模型输出：text / tool_call / handoff
    │        │
    │        ├── text → 检查停止条件 → 流式返回给用户
    │        │
    │        ├── tool_call → 执行前检查 approval → Tool Layer 执行
    │        │        │
    │        │        ▼
    │        │   结果（成功/错误）回填到 context
    │        │        │
    │        │        └── 回到 LLM Provider（下一轮）
    │        │
    │        └── handoff → 交给目标 agent（Orchestrator 层处理）
    │
    ▼
响应返回（SSE 流式 / 一次性）
    │
    ▼
State Layer ── checkpoint 工作记忆 + 持久化对话历史
    │
    ▼
Observability ── 异步发送 trace（LLM 调用、tool 调用、耗时、token、成本）

关键观察：

1. LLM 和 Tool 的调用是交替进行的，不是一次性的——runtime 必须支持循环
2. 流式输出是从 LLM 直通用户的，gateway 不缓冲——中间层只做透传和日志记录
3. tool 的错误回填 context 而非 crash——LLM 是最好的错误恢复器
4. trace 是异步发送的，不影响请求延迟

错误恢复与容错

Agent 系统的错误比普通 web 应用复杂得多，因为涉及 LLM 的不确定性。

分层容错策略

LLM 层：

• 模型超时 → 重试（指数退避），超过阈值后降级到备用模型（model fallback chain）
• 模型输出格式异常 → 重试一次（在 prompt 中强调格式要求），仍失败则报错
• 上下文溢出 → 触发上下文裁剪，重试
• 优雅降级：完整分析失败 → 返回摘要；复杂 tool call 失败 → 退回简单方案

Tool 层：

• 工具超时 → 返回超时错误给 LLM，让模型决定下一步（换工具 / 告知用户）
• 工具返回异常 → 结构化错误信息回填 context，模型可以理解并调整
• 工具不可用 → 注册时标记 degraded，runtime 跳过不可用工具
• 超时预算按 tool 粒度设置，不是整个 run 一刀切

Runtime 层：

• 超过最大轮次 → 强制停止，返回当前最佳结果 + 提示”任务未完成”
• token 预算耗尽 → 强制停止（比轮次限制更精确，因为某些轮次 token 很少）
• 审批被拒绝 → 记录拒绝原因，让模型调整策略

Session 层：

• 对话历史丢失 → 从数据库重建，最多丢失最后一轮未持久化的消息
• 长期记忆不可用 → 降级为无记忆模式，不阻塞主流程

死循环防护

最常见的 agent 失效模式是死循环：模型反复调用同一个工具、反复得到同样的错误、反复重试。

三层防护：

1. 硬上限：max_turns（runtime 层，不依赖模型自觉）
2. 工具去重：连续 N 次相同 tool + 相同参数 → 强制中断
3. Stagnation 检测：如果 agent 在 K 轮后的输出与 K-2 轮无实质差异 → 终止

[!tip] Stagnation 检测（Evaluator 模式） 用一个轻量 evaluator 比较 agent 近 N 轮的输出。如果内容没有实质推进（可以用简单的文本相似度或 token 重叠率），就判定为 stagnation。这比单纯计数轮次更精确——agent 可能在 20 轮内都在有意义地工作，也可能在第 3 轮就陷入了循环。

OpenAI Agents SDK 的 Guardrail 模式：guardrail 和 agent 并行执行，guardrail 检测到违规时可以 fail-fast 中断 agent，不阻塞主流程。

可观测性

[!important] 最重要的可观测工具是 Traces，不是 Logs，不是 Metrics

Trace 模型（Langfuse 参考架构）

Session（多轮对话）
  └── Trace（一次完整请求）
        ├── Observation: LLM 调用（prompt, completion, model, tokens, latency, cost）
        ├── Observation: Tool 执行（name, args, result, duration, success/fail）
        ├── Observation: 路由决策（为什么选这个 tool/agent）
        └── Observation: 状态快照（每个决策点的 context）

关键设计：

• tracing 是异步的——SDK 在后台批量发送，零延迟影响
• 每个 observation 记录输入、输出、耗时、成本
• session 维度聚合，可以回溯整个多轮对话的决策链

必须 trace 的 5 件事

1. 每次 LLM 调用（prompt, completion, model, tokens, latency）
2. 每次 tool 调用（name, arguments, result, duration, success/failure）
3. 路由决策（为什么选这个 tool/agent）
4. 每个决策点的完整状态快照（支持 replay）
5. 成本累积（per-trace cost tracking）

可观测性 → 评估闭环

最好的系统把 tracing 和评估连起来：

• LLM-as-a-Judge：自动评估 agent 输出质量
• Prompt 管理：prompt 变更触发 eval suite
• 成本告警：异常成本自动通知
• A/B 测试：对比不同 prompt / model 的效果

工具选型：Langfuse（开源，可自托管）或 LangSmith（LangChain 生态）。不要自己造 tracing 系统。通用 APM 工具不够用——它们不理解 token usage、prompt/completion pairs、evaluation scores 这些 LLM 特有概念。

Debug 工作流

1. 从 trace 复现失败
2. 检查每个决策点的状态
3. 检查 tool 接口是否模糊（Anthropic 的发现：大多数 agent 失败其实是 tool 接口失败）
4. 检查 prompt 是否充分
5. 添加测试用例到 eval suite

从 Prototype 到 Production 的架构演进

阶段一：单文件 Prototype（Raw API）

main.py
├── prompt + LLM API 调用
├── 几个 tool 函数
└── print 输出

目标：验证 agent 能不能完成核心任务。

关键原则：Anthropic 建议从 Raw API 开始。“很多 pattern 用几行代码就能实现。” 不要在这个阶段引入任何框架。

退出条件：agent 能稳定完成 3 个以上典型任务。

阶段二：加入 Runtime 和 API（Managed Loop）

app/
├── api/          # FastAPI / Gin 路由
├── runtime/      # 执行循环 + 停止条件
├── tools/        # 工具注册 + 执行
└── llm/          # 模型调用抽象

目标：能通过 HTTP 调用，有基本的错误处理和流式输出。

新增：LLM Provider 抽象、Tool 注册机制、Runtime 执行循环。

阶段三：加入状态管理和可观测性

app/
├── api/
├── runtime/
├── tools/
├── llm/
├── state/          # 会话管理 + 工作记忆 checkpoint + 对话历史持久化
├── memory/         # 长期记忆 + 向量检索（通过 tool 接入）
└── observability/  # tracing + 成本追踪 + eval

目标：支持多轮对话，重启后能恢复上下文，出问题能回溯。

[!tip] 状态管理和可观测性应该同阶段引入 没有 tracing 的状态管理是盲的——你看不到 context 是怎么膨胀的、tool 调用为什么失败、成本花在哪里。反过来，没有持久化状态的 tracing 也只能看到单次请求。

阶段四：生产化

这个阶段解决的是真实痛点，不是架构完整性：

痛点	解法
Tool 接口失败率高	投入 ACI 设计：用人类测试工具描述、用绝对路径、用模型训练数据中见过的格式
成本失控	per-trace 成本追踪、请求级 token 预算、easy task 降级到便宜模型
延迟太高	streaming 是必须的、并行 tool 执行、prompt caching、面向用户的 agent 严格限制 max turns
输出不确定	temperature=0（生产环境）、structured outputs（JSON mode）、eval suite 跑在每次 prompt 变更后
状态丢失	外部状态存储（Redis/Postgres）、checkpoint 工作记忆

阶段五：多 Agent 和扩展

app/
├── gateway/        # API Gateway
├── orchestrator/   # 多 agent 编排（Manager-as-tools 优先）
├── agents/         # 多个独立 agent
│   ├── research/
│   ├── coding/
│   └── review/
├── shared/         # 共享工具、LLM Provider、状态层
└── ...

关键：到这一步才需要 Orchestrator 层。过早引入多 agent 编排是过度设计。

常见反模式

反模式	问题	正确做法
一锅炖	runtime、工具、API 逻辑全在一个文件	分层，每层只做一件事
工具依赖 agent 状态	工具函数内部读 agent 的全局变量	工具是纯函数，所有依赖通过参数传入
硬编码模型	换模型要改所有调用处	LLM Provider 抽象，运行时可切换
无限循环靠模型自觉	模型说”我停止了”就停	runtime 设硬上限 + stagnation 检测
对话历史存在内存里	重启丢数据，多实例不共享	持久化到数据库，内存只是缓存
日志只打 print	上线后无法排查问题	结构化 tracing + request ID 贯穿
过早引入框架	LangGraph/CrewAI 的概念比你的需求复杂	先用 Raw API 跑通，复杂度到了再引入框架
单大 dict 传状态	任何 tool 都能读写任何状态，隐式耦合	typed + scoped 状态容器（runtimeContext / toolsContext）
多 agent 解决单 agent 问题	增加复杂度但效果没提升	一个 agent + 更好的工具 > 多个 agent + 各自的工具

边界与易混淆点

• 这不是微服务架构：六层是逻辑分层，不是物理分层。单体部署时它们是模块，分布式部署时它们才变成服务
• Orchestrator 不是必须的：单 agent 项目只需要 Runtime + Tool Layer + LLM Provider，不要为了架构完整性硬加编排层
• Runtime ≠ 框架：LangGraph、CrewAI 是框架，它们帮你实现 runtime。但你需要先理解 runtime 该做什么，才能判断框架是否合适
• 记忆 ≠ 上下文：上下文是发给模型的 messages list，记忆是跨 session 持久化的知识。记忆经过检索后才进入上下文
• 工作记忆 ≠ 对话历史：对话历史是消息序列（source of truth），工作记忆是 agent 的任务进度和中间决策（可 checkpoint、可 resume）
• Guardrail ≠ Stopping condition：guardrail 是安全护栏（拦截危险操作），stopping condition 是终止条件（agent 完成任务或达到上限）。两者独立运行