8.5 自研 Agent 框架
承前：在 8.4 节中，我们建立了系统化的框架选型决策方法——场景驱动、团队能力匹配、生态评估、量化决策矩阵。但在某些场景下，团队可能会发现：无论怎么评估，现有的开源框架都无法完全满足自己的独特需求。这时候就面临一个重大决策：要不要自研 Agent 框架？本节将深入探讨这个问题。
8.5.1 你确定需要自研框架吗？
自研框架是一把双刃剑——它给你最大的自由度，也给你最大的维护负担。在决定自研之前，先问自己三个问题：
决策三问：
┌─────────────────────────────────────────────────────┐
│ │
│ Q1: 开源框架是否真的无法满足需求？ │
│ -> 你试过至少 2 个框架并做了原型验证吗？ │
│ │
│ Q2: 自研的长期维护成本是否可接受？ │
│ -> 你有至少 2 名工程师可以持续维护吗？ │
│ │
│ Q3: 自研带来的差异化价值是否足够大？ │
│ -> 这个框架是否是你们的核心竞争力？ │
│ │
└─────────────────────────────────────────────────────┘生活类比：自研框架就像自己盖房子而不是买现成的。盖房子给你最大的定制空间——想要几个房间、朝哪个方向、用什么材料都由你决定。但盖房子需要设计师、施工队、验收、长期维护。如果你只是需要一个住处，买现房是更好的选择。
如果三个问题中有任何一个回答"不确定"或"否"，建议先使用开源框架。只有三个问题都是肯定的"是"，自研才有充分的理由。
什么时候应该自研？
| 场景 | 自研理由 | 真实案例 |
|---|---|---|
| 深度行业定制 | 通用框架无法满足行业特有的合规、安全、流程需求 | 医疗 AI 需要 HIPAA 合规的 Agent 日志系统 |
| 性能极致要求 | 需要亚秒级延迟，通用框架的抽象层带来额外开销 | 高频交易 Agent 需要 < 100ms 响应 |
| 核心 IP 保护 | Agent 框架本身是公司的核心竞争力 | 某 AI 公司的 Agent 编排引擎是核心产品 |
| 技术栈锁定 | 公司技术栈（如 Rust/Go）与主流 Python 框架不兼容 | 嵌入式设备需要 C++ Agent 运行时 |
| 极简需求 | 只需要 Agent 循环 + 工具调用，不需要完整框架 | 一个简单的客服 Bot，用 LangChain 反而更复杂 |
什么时候不应该自研？
| 场景 | 不自研的理由 |
|---|---|
| 团队 < 5 人 | 维护框架会消耗大量精力，影响业务开发 |
| 需求不明确 | 先用开源框架快速验证，明确需求后再决定 |
| 没有框架维护经验 | 框架设计是专业领域，需要 API 设计、版本管理、文档等能力 |
| "觉得开源框架不好用" | 可能是学习成本问题，而非框架本身的问题 |
常见陷阱："觉得开源框架不好用"是最常见的错误自研理由。很多时候，"不好用"实际上是因为没有深入学习框架的设计理念和使用模式。在投入自研之前，建议先花一周时间深度学习 2-3 个开源框架。
8.5.2 最小可行 Agent 框架（MVP）设计
如果你确实决定自研，不要一上来就追求"大而全"。正确的策略是先实现一个最小可行框架（MVP），然后根据实际需求逐步扩展。
核心架构：三大模块
一个最小可行的 Agent 框架只需要三个核心模块：
┌─────────────────────────────────────────────────────┐
│ 最小可行 Agent 框架 │
├─────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Agent │ │ Tool │ │ Memory │ │
│ │ Loop │◄──▶│ System │ │ System │ │
│ │ (核心循环)│ │ (工具系统)│ │ (记忆系统)│ │
│ └──────────┘ └──────────┘ └──────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────────────────────────────────────┐ │
│ │ LLM Provider (可插拔) │ │
│ └──────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────┘生活类比：最小可行框架就像一辆基础款汽车——发动机（Agent 循环）、方向盘（工具系统）、油箱（记忆系统）三个核心部件就能上路。真皮座椅、导航仪、天窗等都是后续可以加装的选装件。
模块一：Agent 核心循环
Agent 循环是框架的心脏，遵循经典的 ReAct（Reasoning + Acting） 模式。这个模式由三个步骤循环组成：思考（Think）→ 行动（Act）→ 观察（Observe），直到任务完成。
┌──────────────────────────────┐
│ │
▼ │
┌──────────┐ ┌──────────┐
│ THINK │──────────────────▶│ ACT │
│ (思考) │ 决定使用工具 │ (行动) │
└──────────┘ └──────────┘
▲ │
│ ┌──────────┐ │
└──────│ OBSERVE │◀───────────┘
│ (观察) │ 工具返回结果
└──────────┘生活类比：ReAct 循环就像一个厨师做菜的过程。THINK 是看菜谱思考下一步做什么；ACT 是实际操作——切菜、炒菜、调味；OBSERVE 是尝味道、看火候，根据结果决定下一步。如果味道不够，就再加盐（回到 THINK）；如果菜熟了，就出锅（返回结果）。
下面是一个完整的最小可行 Agent 循环实现：
python
# 最小可行 Agent 循环实现
from abc import ABC, abstractmethod # 抽象基类支持
from dataclasses import dataclass, field # 数据类装饰器，自动生成 __init__ 等
from typing import Any, Callable # 类型注解
import json # JSON 解析（处理工具调用参数）
# ===== 1. 消息模型 =====
@dataclass
class Message:
"""统一的消息格式，兼容 OpenAI 消息格式"""
role: str # 角色：system/user/assistant/tool
content: str # 消息内容
tool_calls: list[dict] = field(default_factory=list) # 工具调用列表，默认为空
tool_call_id: str = None # 工具调用 ID，用于关联请求和响应
# ===== 2. 工具定义 =====
@dataclass
class Tool:
"""工具定义：名称、描述、参数 Schema、执行函数"""
name: str # 工具名称，LLM 通过此名称调用工具
description: str # 工具描述，告诉 LLM 这个工具能做什么
parameters: dict # JSON Schema，描述工具参数类型和结构
function: Callable # 实际执行的 Python 函数
def to_openai_format(self) -> dict:
"""转换为 OpenAI Function Calling 格式，供 LLM 识别"""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
}
}
# ===== 3. LLM Provider 接口 =====
class LLMProvider(ABC):
"""LLM 提供者抽象接口，支持可插拔切换"""
@abstractmethod
def chat(self, messages: list[Message], tools: list[dict] = None) -> Message:
"""调用 LLM，传入消息历史和可用工具，返回助手消息"""
...
class OpenAIProvider(LLMProvider):
"""OpenAI LLM 提供者实现"""
def __init__(self, model: str = "gpt-4o-mini", api_key: str = None):
from openai import OpenAI # 延迟导入，避免未使用时加载
self.client = OpenAI(api_key=api_key) # 创建 OpenAI 客户端
self.model = model # 保存模型名称
def chat(self, messages: list[Message], tools: list[dict] = None) -> Message:
"""调用 OpenAI Chat API"""
response = self.client.chat.completions.create(
model=self.model, # 指定模型
messages=[{"role": m.role, "content": m.content} for m in messages], # 转换消息格式
tools=tools, # 传入可用工具列表
tool_choice="auto" if tools else None, # 有工具时自动选择，无工具时不传
)
choice = response.choices[0].message # 取第一个回复
msg = Message(role="assistant", content=choice.content or "") # 构造返回消息
if choice.tool_calls: # 如果 LLM 决定调用工具
msg.tool_calls = [
{"id": tc.id, "name": tc.function.name, "arguments": tc.function.arguments}
for tc in choice.tool_calls # 提取每个工具调用的信息
]
return msg
# ===== 4. Agent 核心循环 =====
class SimpleAgent:
"""最小可行的 Agent 实现：ReAct 循环"""
def __init__(
self,
llm: LLMProvider, # LLM 提供者（可插拔）
tools: list[Tool] = None, # 工具列表
system_prompt: str = "你是一个有帮助的AI助手。", # 系统提示词
max_iterations: int = 10, # 最大循环次数，防止无限循环
):
self.llm = llm
self.tools = {t.name: t for t in (tools or [])} # 工具名到工具的映射字典
self.system_prompt = system_prompt
self.max_iterations = max_iterations
def run(self, user_input: str) -> str:
"""执行 Agent 循环：Think -> Act -> Observe，直到完成或达到上限"""
# 初始化消息列表，包含系统提示和用户输入
messages = [
Message(role="system", content=self.system_prompt),
Message(role="user", content=user_input),
]
# 将所有工具转换为 OpenAI 格式
tool_schemas = [t.to_openai_format() for t in self.tools.values()] if self.tools else None
for iteration in range(self.max_iterations):
# 1. THINK：调用 LLM，让模型思考下一步做什么
response = self.llm.chat(messages, tools=tool_schemas)
# 2. 如果 LLM 没有调用工具，说明任务完成，返回最终答案
if not response.tool_calls:
return response.content
# 3. ACT：LLM 决定调用工具，执行工具调用
messages.append(response) # 先将助手消息（含工具调用）加入历史
for tool_call in response.tool_calls:
tool_name = tool_call["name"] # 工具名称
tool_args = json.loads(tool_call["arguments"]) # 解析参数 JSON
tool = self.tools.get(tool_name) # 查找工具
if tool:
# 4. OBSERVE：执行工具并获取结果
try:
result = tool.function(**tool_args) # 调用工具函数
except Exception as e:
result = f"工具执行错误: {str(e)}" # 捕获异常，不让循环崩溃
else:
result = f"未知工具: {tool_name}" # 工具不存在时的兜底
# 将工具执行结果加入消息历史，供 LLM 下一轮参考
messages.append(Message(
role="tool",
content=str(result),
tool_call_id=tool_call["id"], # 关联到对应的工具调用
))
# 达到最大迭代次数仍未完成
return "已达到最大迭代次数，任务未完成。"
# ===== 5. 使用示例 =====
def search_web(query: str) -> str:
"""模拟网页搜索工具"""
return f"关于 '{query}' 的搜索结果：AI Agent框架正在快速发展..."
def get_weather(city: str) -> str:
"""模拟天气查询工具"""
return f"{city}今天晴天，25°C"
# 创建 Agent 实例
agent = SimpleAgent(
llm=OpenAIProvider(model="gpt-4o-mini", api_key="..."), # LLM 提供者
tools=[
Tool( # 搜索工具
name="search_web",
description="搜索互联网信息",
parameters={
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"}
},
"required": ["query"],
},
function=search_web,
),
Tool( # 天气工具
name="get_weather",
description="查询城市天气",
parameters={
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"}
},
"required": ["city"],
},
function=get_weather,
),
],
system_prompt="你是一个有帮助的助手，可以使用搜索和天气查询工具。",
)
# 运行 Agent：它会自动决定先查天气还是先搜索
result = agent.run("北京今天天气怎么样？最近AI Agent有什么新闻？")
print(result)
# Agent 执行流程：
# 1. THINK -> 决定调用 get_weather(city="北京")
# 2. OBSERVE -> "北京今天晴天，25°C"
# 3. THINK -> 决定调用 search_web(query="AI Agent 最新新闻")
# 4. OBSERVE -> "关于 'AI Agent 最新新闻' 的搜索结果..."
# 5. THINK -> 综合天气和搜索结果，生成最终回答代码要点：整个 Agent 循环只有约 60 行核心代码，但实现了完整的 ReAct 模式。
max_iterations是关键的安全阀——防止 LLM 陷入无限循环消耗 Token。tool_call_id的关联确保多工具调用时结果不会串台。
模块二：工具系统
生产环境的工具系统需要比 MVP 更健壮——支持重试、超时、并行执行和条件激活。
python
# 增强版工具系统
import asyncio # 异步编程支持
import time # 时间相关功能（重试延迟）
from functools import wraps # 装饰器工具
from enum import Enum # 枚举类型
class ToolExecutionMode(Enum):
"""工具执行模式枚举"""
SYNC = "sync" # 同步执行：阻塞等待结果
ASYNC = "async" # 异步执行：不阻塞，通过 await 获取结果
PARALLEL = "parallel" # 并行执行：多个工具同时执行
@dataclass
class ToolConfig:
"""工具配置：超时、重试策略"""
timeout: float = 30.0 # 超时时间（秒），防止工具卡死
max_retries: int = 2 # 最大重试次数，应对临时网络错误
retry_delay: float = 1.0 # 重试间隔（秒），避免密集重试
class RobustTool(Tool):
"""带重试和超时的增强版工具"""
def __init__(self, *args, config: ToolConfig = None, **kwargs):
super().__init__(*args, **kwargs) # 调用父类初始化
self.config = config or ToolConfig() # 使用传入配置或默认配置
def execute(self, **kwargs) -> str:
"""执行工具，带自动重试机制"""
last_error = None # 记录最后一次错误
for attempt in range(self.config.max_retries + 1): # 重试次数 + 1 次初始尝试
try:
result = self.function(**kwargs) # 调用实际工具函数
return str(result) # 成功则返回结果
except Exception as e:
last_error = e # 记录错误
if attempt < self.config.max_retries: # 还有重试机会
time.sleep(self.config.retry_delay) # 等待后重试
# 所有重试都失败
return f"工具执行失败（已重试{self.config.max_retries}次）: {last_error}"
class ToolRegistry:
"""工具注册中心，支持条件激活（只在特定场景下暴露某些工具）"""
def __init__(self):
self._tools: dict[str, RobustTool] = {} # 工具名到工具的映射
self._activation_rules: dict[str, Callable] = {} # 工具名到激活条件的映射
def register(self, tool: RobustTool, activation_rule: Callable = None):
"""注册工具，可附带激活条件（返回 bool 的函数）"""
self._tools[tool.name] = tool # 存入工具
if activation_rule: # 如果提供了激活条件
self._activation_rules[tool.name] = activation_rule # 存入激活规则
def get_active_tools(self, context: dict = None) -> list[RobustTool]:
"""返回当前上下文下激活的工具列表"""
active = []
for name, tool in self._tools.items():
if name in self._activation_rules: # 有激活规则的工具
if self._activation_rules[name](context): # 检查规则是否满足
active.append(tool) # 规则满足，加入激活列表
else:
active.append(tool) # 无规则的工具始终激活
return active
async def execute_parallel(self, calls: list[dict]) -> list[str]:
"""并行执行多个工具调用，提高吞吐量"""
tasks = []
for call in calls:
tool = self._tools.get(call["name"]) # 查找工具
if tool:
# asyncio.to_thread 将同步函数包装为异步执行
tasks.append(asyncio.to_thread(
tool.execute, **json.loads(call["arguments"])
))
# asyncio.gather 并行等待所有任务完成
return await asyncio.gather(*tasks, return_exceptions=True)条件激活的价值：不是所有工具都需要同时暴露给 LLM。比如财务 Agent 在"报销"场景下只需要发票识别和审批工具，在"报表"场景下只需要数据查询和图表生成工具。条件激活让 LLM 每次只看到相关工具，减少幻觉和误调用。
模块三：记忆系统
记忆系统让 Agent 从"金鱼记忆"进化为"长期记忆"。
python
# 层次化记忆系统
from collections import deque # 双端队列，高效从头尾操作
import hashlib, json # 生成记忆 ID
class ConversationMemory:
"""短期记忆：滑动窗口对话历史管理"""
def __init__(self, max_tokens: int = 4000):
self.max_tokens = max_tokens # Token 预算上限
self.messages: deque[Message] = deque() # 双端队列存储消息
def add(self, message: Message):
"""添加消息并自动裁剪"""
self.messages.append(message) # 从尾部添加新消息
self._trim() # 检查是否超出预算
def _trim(self):
"""按 Token 预算裁剪历史，保留 system prompt"""
total = sum(len(m.content) for m in self.messages) # 简化估算：字符数 ≈ Token × 4
while total > self.max_tokens * 4 and len(self.messages) > 2:
removed = self.messages.popleft() # 从头部移除最旧的消息
if removed.role == "system": # system prompt 不能被移除
self.messages.appendleft(removed) # 放回去
break
total = sum(len(m.content) for m in self.messages) # 重新计算
def get_context(self) -> list[Message]:
"""返回当前记忆中的所有消息"""
return list(self.messages)
class SemanticMemory:
"""长期记忆：基于向量存储的语义检索"""
def __init__(self, embedding_provider, vector_store):
self.embedding = embedding_provider # 嵌入模型提供者
self.store = vector_store # 向量存储后端
async def store(self, content: str, metadata: dict = None):
"""存储一段记忆到向量数据库"""
embedding = await self.embedding.embed(content) # 将文本转为向量
memory_id = hashlib.md5(content.encode()).hexdigest() # 用内容 MD5 作为 ID（去重）
await self.store.add(
id=memory_id, # 唯一标识
embedding=embedding, # 向量表示
metadata=metadata or {}, # 元数据（时间、来源等）
text=content, # 原始文本，检索时返回
)
async def retrieve(self, query: str, top_k: int = 5) -> list[str]:
"""根据查询检索相关记忆"""
embedding = await self.embedding.embed(query) # 将查询转为向量
results = await self.store.search(embedding, top_k=top_k) # 向量相似度搜索
return [r["text"] for r in results] # 返回最相似的 top_k 条记忆文本生活类比：
ConversationMemory就像你的工作记忆——记着最近几轮对话，但容量有限，太久远的会被遗忘。SemanticMemory就像你的笔记本——把重要的东西存下来，需要时通过关键词搜索找到。
8.5.3 从框架到平台的演进路径
自研框架不是终点，而是一个不断演进的起点。从项目内框架到内部平台，通常经历四个阶段。
演进阶段
阶段一：项目内框架（1-3个月）
┌─────────────────────────────────────┐
│ · 功能耦合在项目中 │
│ · 通过复制粘贴"复用" │
│ · 只有核心开发者能修改 │
└─────────────────────────────────────┘
│
▼
阶段二：内部库（3-6个月）
┌─────────────────────────────────────┐
│ · 抽取为独立 Python 包 │
│ · 有版本号和 CHANGELOG │
│ · 2-3 个项目在使用 │
└─────────────────────────────────────┘
│
▼
阶段三：内部平台（6-12个月）
┌─────────────────────────────────────┐
│ · 提供 CLI/Web 管理界面 │
│ · 有插件市场和工具注册中心 │
│ · 有文档、培训、on-call 支持 │
│ · 10+ 个项目在使用 │
└─────────────────────────────────────┘
│
▼
阶段四：开源/商业化（12个月+）
┌─────────────────────────────────────┐
│ · 开源发布，建立社区 │
│ · 提供托管服务（SaaS） │
│ · 企业版功能（SSO、审计、SLA） │
└─────────────────────────────────────┘生活类比：这就像一家餐厅的成长路径——阶段一是路边摊（一个人干），阶段二是连锁小店（标准化菜谱），阶段三是品牌餐厅（有管理系统和培训体系），阶段四是上市餐饮集团（开放加盟、提供供应链服务）。
内部平台化 Checklist
当你的框架被多个团队使用时，需要这些基础设施：
yaml
# 平台化需求清单
必须:
- 版本管理: 语义化版本，清晰的升级路径
- 文档: 快速开始、API 参考、最佳实践、FAQ
- 测试: 单元测试覆盖率 > 80%，集成测试
- CI/CD: 自动构建、测试、发布
- 变更日志: CHANGELOG.md，Breaking Changes 标注
建议:
- 开发者门户: Web UI 浏览工具、Agent 模板
- 沙箱环境: 在线试用，无需本地安装
- 监控面板: 使用量、成本、错误率
- 认证培训: 内部课程，降低支持成本
- 社区频道: Slack/Discord 频道，知识共享8.5.4 自研框架的常见陷阱与对策
| 陷阱 | 症状 | 对策 |
|---|---|---|
| 过度设计 | 第一版就支持10种LLM、20种存储后端 | 先支持1-2个，通过接口抽象预留扩展 |
| 重复造轮子 | 实现了 LangChain 已有功能，但更差 | 自研前先深度调研，明确差异化价值 |
| 缺乏文档 | 只有作者能看懂代码 | 从第一天起写文档，至少包括 README + 架构图 |
| API 不稳定 | 每个版本都有 Breaking Changes | 语义化版本 + 弃用警告（Deprecation Warning） |
| 忽视可观测性 | 出问题时无法定位 | 内置结构化日志 + 追踪（至少支持 OpenTelemetry） |
| 闭门造车 | 不与用户沟通，按自己想象开发 | 定期与使用者沟通，收集反馈 |
最危险的陷阱是"过度设计"。很多团队第一版就想支持所有 LLM Provider、所有向量存储后端、所有编排模式。结果是开发 3 个月还没出可用版本。正确做法是：先支持 1 个 LLM Provider（通常是 OpenAI），1 种存储（通常是内存或 Chroma），1 种编排模式（ReAct 循环），上线运行后再根据实际需求扩展。
8.5.5 常见误区
误区一：自研 = 重新发明一切
自研框架不意味着从零开始写所有代码。你可以复用开源组件（如 OpenAI SDK、Chroma 向量库），只在编排层和接口层做自研。关键差异化在于你的框架如何组织 Agent 协作和工具调度，而不是底层的 LLM 调用代码。
误区二：先写框架再做业务
正确的顺序是：先做业务，在业务中积累通用模式，再把这些模式抽取为框架。先写框架再做业务就像先造发动机再决定造什么车——你不知道发动机需要多大马力、什么尺寸。
误区三：忽视迁移策略
如果从开源框架迁移到自研框架，需要设计适配层（Adapter Pattern），让自研框架实现与原框架兼容的接口，逐步替换底层实现，保持上层业务代码不变。一次性"推倒重来"的迁移几乎总是失败的。
误区四：没有退出策略
自研框架可能因为团队变动、业务调整而失去维护者。设计框架时就要考虑"如果没人维护了怎么办"——核心逻辑应该足够简单，让任何有经验的 Python 开发者都能在一周内理解。
误区五：文档是"以后再写"的事情
没有文档的框架等于没有框架。从第一天起就应该写 README 和架构图。文档不是框架完成后的"收尾工作"，而是框架设计的一部分——如果你无法用文字解释清楚某个设计决策，说明这个设计本身可能有问题。
8.5.6 本节小结
本节探讨了自研 Agent 框架的决策边界、设计和演进：
- 自研框架不是"信仰"问题，而是"成本-收益"问题：先问三个问题（需求、维护成本、差异化价值），再决定是否自研
- 最小可行框架 = Agent 循环 + 工具系统 + 记忆系统：从这三个模块开始，逐步扩展
- Agent 循环遵循 ReAct 模式：Think（思考）→ Act（行动）→ Observe（观察）的循环，直到任务完成
- 工具系统要支持重试、超时、并行执行：生产环境的工具调用需要容错能力
- 从框架到平台是自然演进：项目内框架 → 内部库 → 内部平台 → 开源/商业化
- 避免过度设计：先支持 1-2 个 LLM Provider，通过接口抽象预留扩展空间
- 文档是第一优先级：没有文档的框架等于没有框架
启后：至此，第 8 章 Agent 框架的内容全部结束——从框架概览到核心能力对比，从工程结构设计到选型决策，再到自研框架的完整路径。然而，框架只是 Agent 系统的"骨架"，真正让 Agent 系统在生产环境中可靠运行，还需要系统级的设计与架构思考。第 9 章将从系统设计的视角，探讨 Agent 系统的高可用架构、性能优化、安全防护和成本控制等关键主题。
参考资料
- AgentSquare：模块化 Agent 设计框架 - 智源社区，清华大学团队提出的模块化 Agent 框架设计范式，展示了如何将 Agent 拆解为规划、推理、工具、记忆四大模块
- AgentSquare 论文 - arXiv，2024年，学术论文详细阐述了模块化 Agent 设计空间与自动搜索方法
- AgentSquare 开源代码 - GitHub，清华大学，模块化 Agent 框架的参考实现
- LangChain 1.0 Agent 循环设计 - LangChain 官方，2025年10月，LangChain 1.0 的 Agent 循环和中间件设计，可作为自研框架的参考
- Semantic Kernel Plugin 架构 - Microsoft Developer Blogs，插件化 Agent 编排的实现参考
- ReAct 论文：Synergizing Reasoning and Acting in Language Models - arXiv，2022年，Agent 循环的理论基础，ReAct 模式的原始论文