第九章 系统设计
在第八章中，我们完整地讨论了 Agent 框架的设计与实现——从 Agent 的核心循环到工具调用的编排，再到记忆系统的构建。那些内容回答的是一个"怎么想、怎么做"的问题：Agent 在逻辑层面如何推理、如何决策。
但当你把这些框架推向生产环境，面对真实用户和真实流量时，一个全新的问题层就会浮现：系统怎么扛得住？ 一个能跑通 Demo 的 Agent，和能稳定服务一千个用户的 Agent，之间的差距可能比从零搭建框架还要大。
本章将视角从"Agent 内部逻辑"切换到"Agent 系统工程"，围绕六个核心主题展开：
- 9.1 Agent 系统架构——从四层架构到同步/异步选型，搭建系统的骨架
- 9.2 高并发处理——请求队列、连接池、限流与异步任务管道
- 9.3 缓存策略——从 Prefix Cache 到语义缓存，四层缓存各显神通
- 9.4 降级与容错——模型切换、熔断器、重试策略，让系统在风雨中站稳
- 9.5 成本控制——Token 计费、模型分层、预算告警，管好每一分钱
- 9.6 流式输出 Streaming——SSE 与 WebSocket，让用户在等待中看见进展
这些主题之间环环相扣：架构决定了扩展方式，高并发依赖队列和限流，缓存缓解并发压力，降级保证可用性，成本控制贯穿始终，流式输出则是用户体验的最后一公里。
9.1 Agent 系统架构
9.1.1 为什么 Agent 系统需要专门的架构设计？
很多团队在开发 Agent 应用时，最初的架构非常简单：一个 FastAPI 服务，直接调用 OpenAI API，然后把结果返回给前端。这在原型阶段完全没问题——十几个内部用户，偶尔超时也能接受。但当用户量从 10 人涨到 1000 人时，问题就集中爆发了。
打个生活中的比方：这就像你在家做饭招待三五好友，随手炒几个菜绰绰有余；但如果要开一家每天接待 1000 位客人的餐厅，你需要的就不再是一口锅和一个灶台，而是一套完整的后厨系统——备料区、烹饪区、出餐口、洗碗间，各司其职。
Agent 系统与传统 CRUD（增删改查）应用有本质区别：
- 高延迟：一次 LLM 调用可能需要 2~30 秒，而不是毫秒级
- 高成本：每次调用都在消耗 Token（即消耗金钱）
- 非确定性：同样的输入可能产生不同的输出
- 多轮交互：Agent 需要多步推理、工具调用、自我纠错
- 资源密集：GPU 资源昂贵且稀缺
这些特性决定了 Agent 系统不能简单地套用传统 Web 应用的架构模式，而是需要专门的设计。
9.1.2 Agent 系统四层架构
一个成熟的生产级 Agent 系统，通常采用四层架构：
┌─────────────────────────────────────────────────────────┐
│ 前端 / 客户端 │
│ (Web, Mobile, API Client) │
├─────────────────────────────────────────────────────────┤
│ API Gateway 层 │
│ (认证鉴权、限流、路由、请求/响应转换、日志审计) │
├─────────────────────────────────────────────────────────┤
│ Agent 引擎层 │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌─────────────┐ │
│ │ 意图识别 │ │ 规划调度 │ │ 工具调用 │ │ 记忆管理 │ │
│ └──────────┘ └──────────┘ └──────────┘ └─────────────┘ │
├─────────────────────────────────────────────────────────┤
│ LLM 服务层 │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌─────────────┐ │
│ │ GPT-4 │ │ Claude │ │ DeepSeek │ │ 本地模型 │ │
│ └──────────┘ └──────────┘ └──────────┘ └─────────────┘ │
└─────────────────────────────────────────────────────────┘继续用餐厅的比方来理解：前端就是大堂里点餐的顾客，API Gateway 是门口的迎宾和保安（检查你有没有预约、限流），Agent 引擎是后厨的厨师长（决定做什么菜、先做哪道），LLM 服务层则是灶台本身（真正"烹饪"的地方）。
第一层：前端/客户端
负责用户交互，包括 Web 界面、移动端 App、或第三方 API 调用。前端需要支持流式渲染（Streaming），因为 Agent 的响应往往是逐步生成的——就像看直播和看录播的区别，流式渲染让用户在等待中就能看到内容逐渐浮现。
第二层：API Gateway
这是系统的"守门人"，负责：
- 认证与鉴权：JWT Token 验证、API Key 管理
- 限流（Rate Limiting）：防止单个用户过度调用
- 请求路由：根据请求类型分发到不同的 Agent 服务
- 日志与审计：记录所有 API 调用，用于成本分析和安全审计
- 请求/响应转换：统一不同 LLM 提供商的 API 格式
在实际工程中，常用的 API Gateway 方案包括：
- Kong / APISIX：功能全面的开源 API 网关
- Nginx + OpenResty：高性能，适合对延迟敏感的场景
- FastAPI 中间件：轻量级方案，适合中小规模应用
AI Gateway 是近年来兴起的一个新概念。InfoQ 在 2025 年的一篇文章中提出了 AI Gateway 的参考设计，包含四个核心组件：Traffic Interceptor（流量拦截器）、Policy Engine（策略引擎）、Routing & Cost Manager（路由与成本管理）、Observability Layer（可观测性层）。它的定位是在传统 API Gateway 之上，增加一层"AI 感知"——知道什么是 Token、什么是 Prompt、什么算贵。
第三层：Agent 引擎
这是 Agent 系统的核心，负责：
- 意图识别：理解用户真正想做什么
- 规划与调度：制定执行计划（ReAct、Plan-and-Execute 等）
- 工具调用：管理 Function Calling 的生命周期
- 记忆管理：维护短期和长期记忆
第八章讨论的框架代码，主要就运行在这一层。
第四层：LLM 服务
负责实际的模型推理。可以是：
- 第三方 API（OpenAI、Anthropic、DeepSeek 等）
- 自部署模型（vLLM、TGI 等推理框架）
- 混合方案（根据任务复杂度动态路由）
9.1.3 同步架构 vs 异步架构
这是 Agent 系统设计中最关键的架构决策之一。
同步架构（Request-Response）
Client -> POST /chat -> Server -> LLM API -> Response -> Client特点是客户端发起请求后，阻塞等待完整响应。就像打电话：你拨通号码后必须在线等着对方接听和回答，中间挂断就什么也得不到。
- 实现简单，适合简单问答场景
- 缺点：长时间等待时用户体验差，连接容易超时
- 适用场景：后台批处理、简单对话、API 集成
异步架构（消息队列 / 事件驱动）
Client -> POST /chat -> Server -> Message Queue -> Worker -> LLM API
↓
Client ← SSE/WebSocket ← Server ← Result Queue ← Worker特点是请求立即返回任务 ID，客户端通过轮询或 WebSocket 获取结果。就像寄快递：你把包裹交给快递站，拿到一个运单号，然后随时可以查物流状态，不用在柜台干等。
- 系统各组件解耦，可独立扩缩容
- 适合长时间运行的 Agent 任务（多步推理、大量工具调用）
- 缺点：架构复杂度增加，需要额外的基础设施
异步架构的代码示例（FastAPI + Celery）：
# app.py - FastAPI 主服务
from fastapi import FastAPI, BackgroundTasks # 导入 FastAPI 框架和后台任务支持
from celery_app import run_agent_task # 导入 Celery 异步任务
from pydantic import BaseModel # 导入数据模型基类
import uuid # 导入 UUID 生成器
app = FastAPI() # 创建 FastAPI 应用实例
# 存储任务状态（生产环境应使用 Redis）
task_store = {} # 用字典暂存任务状态，仅限单机演示
class ChatRequest(BaseModel): # 定义请求体模型
message: str # 用户消息内容
user_id: str # 用户标识
@app.post("/api/chat") # 定义 POST 接口
async def chat(request: ChatRequest):
task_id = str(uuid.uuid4()) # 生成唯一任务 ID
# 将任务发送到 Celery 异步执行
run_agent_task.delay(task_id, request.message, request.user_id) # .delay() 将任务推入队列
task_store[task_id] = {"status": "pending"} # 记录初始状态
return {"task_id": task_id, "status": "pending"} # 立即返回任务 ID
@app.get("/api/chat/{task_id}/status") # 状态查询接口
async def get_status(task_id: str):
task = task_store.get(task_id) # 从存储中查找任务
if not task:
return {"error": "Task not found"} # 任务不存在
return {"task_id": task_id, **task} # 返回任务状态# celery_app.py - Celery Worker
from celery import Celery # 导入 Celery 分布式任务队列
import openai # 导入 OpenAI SDK
import redis # 导入 Redis 客户端
celery_app = Celery( # 创建 Celery 实例
'agent_tasks', # 应用名称
broker='redis://localhost:6379/0', # 消息代理（Redis）
backend='redis://localhost:6379/1' # 结果存储（Redis）
)
redis_client = redis.Redis(host='localhost', port=6379, db=2) # 状态存储
@celery_app.task(bind=True) # 注册为 Celery 任务
def run_agent_task(self, task_id: str, message: str, user_id: str):
"""异步执行 Agent 任务"""
try:
# 更新状态为运行中
redis_client.hset(task_id, mapping={ # 写入 Redis 哈希
"status": "running", # 状态标记
"progress": "0" # 进度初始化
})
# 调用 LLM
client = openai.OpenAI() # 创建 OpenAI 客户端
response = client.chat.completions.create( # 发起补全请求
model="gpt-4", # 指定模型
messages=[{"role": "user", "content": message}], # 构造消息
stream=False # 非流式（完整返回）
)
result = response.choices[0].message.content # 提取回复文本
# 存储结果到 Redis
redis_client.hset(task_id, mapping={ # 更新最终状态
"status": "completed", # 标记完成
"result": result, # 存储结果
"progress": "100" # 进度 100%
})
return result # 返回结果给 Celery
except Exception as e:
# 发生异常时记录失败状态
redis_client.hset(task_id, mapping={
"status": "failed", # 标记失败
"error": str(e) # 记录错误信息
})
raise # 重新抛出异常9.1.4 有状态 vs 无状态 Agent
无状态 Agent——每次请求都是独立的，Agent 不保留任何上下文。
打个比方，无状态 Agent 就像快餐店的点餐窗口：每次去都要从头报一遍你要什么，店员不会记得你上次点了什么。
优点：水平扩展极其简单（任意实例可以处理任意请求）；无状态同步问题；重启不会丢失数据。
缺点：每次请求需要传递完整上下文，Token 消耗大；无法在服务端维护会话状态。
@app.post("/api/stateless/chat")
async def stateless_chat(request: ChatRequest):
"""无状态 Agent：上下文完全由客户端维护"""
# 客户端在 messages 中传递完整历史
response = await llm_client.chat(messages=request.messages) # 直接用传入的消息调用
return {"reply": response} # 返回回复有状态 Agent——Agent 在服务端维护会话状态，包括对话历史、工具调用状态、中间推理结果等。
就像常去的小酒馆，老板记得你上次聊到哪了，你只要接着说就行。
优点：客户端只需传递最新的用户消息，节省带宽；可以在服务端做更复杂的上下文管理；支持"暂停-恢复"模式。
缺点：扩展复杂（需要会话亲和性或多节点状态同步）；需要持久化存储（Redis、PostgreSQL 等）；状态管理增加了系统复杂度。
import redis.asyncio as aioredis # 导入异步 Redis 客户端
redis = await aioredis.from_url("redis://localhost") # 连接 Redis
@app.post("/api/stateful/chat")
async def stateful_chat(session_id: str, message: str):
"""有状态 Agent：服务端维护会话历史"""
# 从 Redis 加载历史对话
history_json = await redis.get(f"session:{session_id}:history") # 读取历史
history = json.loads(history_json) if history_json else [] # 解析 JSON
# 追加新消息
history.append({"role": "user", "content": message}) # 添加用户消息
# 调用 LLM
response = await llm_client.chat(messages=history) # 带历史调用
# 保存更新后的历史
history.append({"role": "assistant", "content": response}) # 添加 AI 回复
await redis.set( # 写回 Redis
f"session:{session_id}:history", # 键名
json.dumps(history, ensure_ascii=False), # 序列化
ex=3600 # 1小时过期
)
return {"reply": response}混合方案（推荐）：在生产实践中，通常采用混合方案——核心状态（如会话 ID、用户身份）由客户端持有，扩展状态（如对话历史、工具调用中间结果）在服务端维护，通过会话 ID 关联。这就像你带着身份证（核心身份），但看病时医院的系统帮你记着病历（扩展状态）。
9.1.5 AI Gateway：Agent 系统的流量中枢
AI Gateway 是近年来出现的重要架构模式。它位于 Agent 引擎和 LLM 服务之间，提供统一的能力：
- 统一 API：屏蔽不同 LLM 提供商的 API 差异
- 智能路由：根据请求特征（复杂度、成本预算、延迟要求）自动选择最合适的模型
- 成本控制：实时追踪 Token 消耗，设置预算上限
- 安全防护：Prompt 注入检测、敏感信息过滤
- 可观测性：全链路追踪、延迟分析、错误监控
# AI Gateway 的简化实现
class AIGateway:
def __init__(self):
# 模型配置表：记录各模型的成本
self.models = {
"gpt-4": {"cost_per_1k_input": 0.03, "cost_per_1k_output": 0.06},
"gpt-4o-mini": {"cost_per_1k_input": 0.00015, "cost_per_1k_output": 0.0006},
"claude-3-opus": {"cost_per_1k_input": 0.015, "cost_per_1k_output": 0.075},
}
# 路由规则：根据复杂度选择模型
self.routing_rules = [
{"condition": "complexity == 'high'", "model": "gpt-4"},
{"condition": "complexity == 'low'", "model": "gpt-4o-mini"},
{"condition": "default", "model": "gpt-4o-mini"},
]
async def route_and_call(self, messages: list, complexity: str = "low"):
"""根据复杂度路由到合适的模型"""
model = self._select_model(complexity) # 按规则选择模型
response = await self._call_llm(model, messages) # 调用 LLM
self._track_cost(model, response.usage) # 记录成本
return response # 返回响应9.1.6 完整示例：搭建基础 Agent 服务架构
下面的代码展示了如何用 FastAPI 搭建一个包含 Gateway 层和 Agent 引擎层的服务：
from fastapi import FastAPI, Depends, HTTPException, Header # 导入 FastAPI 组件
from pydantic import BaseModel # 导入数据模型
import asyncio # 导入异步库
import time # 导入时间模块
import hashlib # 导入哈希库
app = FastAPI(title="Agent Service") # 创建应用
# ============ Gateway 层：认证中间件 ============
API_KEYS = {"demo-key-123": "demo-user"} # API Key 映射表
async def verify_api_key(x_api_key: str = Header(...)): # 从请求头读取 Key
"""API Key 认证中间件"""
if x_api_key not in API_KEYS: # 校验 Key
raise HTTPException(status_code=401, detail="Invalid API Key")
return API_KEYS[x_api_key] # 返回用户 ID
# ============ Gateway 层：限流中间件 ============
rate_limit_store = {} # 限流记录存储
async def rate_limiter(user_id: str, max_requests: int = 10, window: int = 60):
"""简单的滑动窗口限流"""
now = time.time() # 当前时间戳
key = f"rate:{user_id}" # 限流键
if key not in rate_limit_store: # 初始化
rate_limit_store[key] = []
# 清理过期记录（超出时间窗口的）
rate_limit_store[key] = [t for t in rate_limit_store[key] if now - t < window]
if len(rate_limit_store[key]) >= max_requests: # 超过上限
raise HTTPException(status_code=429, detail="Too many requests")
rate_limit_store[key].append(now) # 记录本次请求
# ============ Agent 引擎层 ============
class AgentEngine:
"""简化的 Agent 引擎"""
def __init__(self):
self.tools = { # 注册工具
"search": self.search_tool,
"calculate": self.calculate_tool,
}
async def search_tool(self, query: str) -> str:
"""模拟搜索工具"""
await asyncio.sleep(0.5) # 模拟网络延迟
return f"搜索结果：关于 '{query}' 的信息..."
async def calculate_tool(self, expression: str) -> str:
"""模拟计算工具"""
try:
result = eval(expression) # 计算表达式（生产环境不要用 eval）
return f"计算结果：{expression} = {result}"
except:
return "计算错误"
async def run(self, user_message: str, user_id: str) -> dict:
"""执行 Agent 任务"""
start_time = time.time() # 记录开始时间
# 简单的意图识别
if "搜索" in user_message or "search" in user_message.lower():
query = user_message.replace("搜索", "").strip() # 提取搜索词
result = await self.tools["search"](query) # 调用搜索工具
action = "search"
elif any(op in user_message for op in ["+", "-", "*", "/"]):
result = await self.tools["calculate"](user_message) # 调用计算工具
action = "calculate"
else:
result = f"收到你的消息：{user_message}（模拟回复）" # 默认对话
action = "chat"
elapsed = time.time() - start_time # 计算耗时
return {
"reply": result, # 回复内容
"action": action, # 执行的动作
"elapsed_ms": round(elapsed * 1000, 2), # 毫秒耗时
"user_id": user_id # 用户 ID
}
agent = AgentEngine() # 实例化 Agent 引擎
# ============ API 路由 ============
class ChatRequest(BaseModel):
message: str # 用户消息
stream: bool = False # 是否流式输出
@app.post("/api/v1/chat")
async def chat(
request: ChatRequest,
user_id: str = Depends(verify_api_key) # 依赖注入：自动认证
):
"""聊天接口 - 经过认证和限流"""
await rate_limiter(user_id) # 限流检查
result = await agent.run(request.message, user_id) # 执行 Agent
return result # 返回结果
@app.get("/api/v1/health")
async def health():
"""健康检查接口"""
return {
"status": "healthy", # 服务状态
"timestamp": time.time(), # 当前时间
"tools": list(agent.tools.keys()) # 可用工具列表
}
# 启动命令：uvicorn main:app --reload常见误区
"先用同步架构，以后再改异步"——这是最常见的陷阱。同步架构到异步架构的改造往往涉及全链路，包括前端、API 层、Worker 层，改动成本极高。如果预期用户量会增长，从一开始就用异步架构更省事。
"无状态一定比有状态好"——无状态确实更易于扩展，但对于多轮对话场景，每次传递完整历史的 Token 开销可能非常大。混合方案才是最优解。
"API Gateway 是可选的"——在原型阶段确实可以省略，但一旦上生产，没有 Gateway 意味着没有认证、没有限流、没有审计，任何一个恶意用户都能打垮你的系统。
"AI Gateway 就是 API Gateway 换个名字"——两者有本质区别。传统 API Gateway 不理解 Token、不关心模型成本、不做 Prompt 安全检查。AI Gateway 是为 LLM 流量专门设计的。
把所有逻辑塞进一个服务——很多团队把意图识别、工具调用、LLM 调用全放在一个进程里。短期没问题，但一旦需要独立扩缩容（比如工具调用量大但 LLM 调用量小），就会受限。
本节小结
| 要点 | 说明 |
|---|---|
| 四层架构 | 前端 → API Gateway → Agent 引擎 → LLM 服务，各层职责清晰，独立扩缩容 |
| 同步 vs 异步 | 简单问答用同步，长时间 Agent 任务用异步（消息队列） |
| 有状态 vs 无状态 | 混合方案最佳：客户端持有会话 ID，服务端维护扩展状态 |
| AI Gateway | 统一 API、智能路由、成本控制、安全防护的核心枢纽 |
| FastAPI + asyncio | Python Agent 服务的主流技术栈，天然支持异步 |
本节搭建了 Agent 系统的架构骨架——四层分层、同步异步选型、有状态无状态的权衡，以及 AI Gateway 的定位。但这只是"画了蓝图"。当真正的用户流量涌来时，骨架能扛住吗？下一节我们将深入高并发处理的细节：请求队列如何削峰填谷，连接池怎么调参，Rate Limit 的三种算法各有什么优劣。
下一节：9.2 高并发处理——从请求队列到异步任务管道，让系统在流量洪峰下稳如磐石。