从零设计一个 Agent 框架,最常犯的错是一上来就堆功能——记忆、规划、工具、监控、调试面板一齐上,结果是 6 个月后还跑不通一个真实任务。在做内部 Agent 框架”AgentKit”的第一个版本时,我把模块按”必需 → 加分 → 锦上添花”分了三档,先实现最底层的 7 个模块,6 周就交付了一个能跑生产任务的 MVP。下文把这 7 个模块的依赖关系、实现顺序和最小可用版本都讲清,重点是工程团队最关心的”先做哪个能最快打通闭环”。
一、模块依赖全景
一个 Agent 框架的核心模块,大致分三层:
- 基础层(必须先有,无它整个框架跑不起来):LLM 客户端、工具注册与执行、对话循环;
- 核心层(决定框架能跑多复杂的任务):记忆管理、规划与任务分解、可观测性;
- 增值层(让框架好用,延后实现影响也不大):调试 UI、评估体系、多 Agent 协作。
按依赖关系,前 7 个模块的优先级如下:
- LLM 客户端(必需)
- 工具注册与执行(必需)
- 对话循环(必需)
- 记忆管理(核心)
- 规划与任务分解(核心)
- 可观测性(核心)
- 错误处理与重试(核心)
二、模块 1:LLM 客户端
最底层,所有其它模块都依赖它。最小可用版本只需要支持:同步调用、流式输出、function calling、错误重试。
class LLMClient:
def __init__(self, model="gpt-4o", api_key=None):
self.model = model
self.client = OpenAI(api_key=api_key)
def chat(self, messages, tools=None, stream=False):
kwargs = {"model": self.model, "messages": messages}
if tools: kwargs["tools"] = tools
if stream: kwargs["stream"] = True
return self.client.chat.completions.create(**kwargs)实现要点:用统一的 ChatCompletion 接口封装 OpenAI、Anthropic、本地模型,让上层不关心模型细节。
三、模块 2:工具注册与执行
工具是 Agent 与外部世界交互的桥梁。最小可用版本需要:工具描述 schema、参数校验、调用超时、调用日志。
class ToolRegistry:
def __init__(self):
self.tools = {}
def register(self, name, func, description, parameters):
self.tools[name] = {
"func": func,
"schema": {
"type": "function",
"function": {
"name": name,
"description": description,
"parameters": parameters
}
}
}
def execute(self, name, arguments, timeout=30):
if name not in self.tools:
raise ValueError(f"Tool {name} not found")
# 参数校验(JSON Schema)
# jsonschema.validate(arguments, self.tools[name]["schema"]["function"]["parameters"])
return self.tools[name]["func"](**arguments)实现要点:工具描述必须严格遵循 OpenAI function calling 格式,这样 LLM 能直接消费。
四、模块 3:对话循环
Agent 的”心脏”:接收用户输入、调 LLM 决策、执行工具、把结果再喂回 LLM,直到任务完成或放弃。
class AgentLoop:
def __init__(self, llm, tools, max_iterations=10):
self.llm = llm
self.tools = tools
self.max_iterations = max_iterations
def run(self, user_input, system_prompt=""):
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_input}
]
for i in range(self.max_iterations):
resp = self.llm.chat(messages, tools=[t["schema"] for t in self.tools.tools.values()])
msg = resp.choices[0].message
messages.append(msg)
if not msg.tool_calls:
return msg.content
for tc in msg.tool_calls:
args = json.loads(tc.function.arguments)
try:
result = self.tools.execute(tc.function.name, args)
except Exception as e:
result = f"Error: {str(e)}"
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": str(result)
})
return "任务超时,请简化后重试"实现要点:max_iterations 必须设上限,防止死循环;每次循环要记录 trace,便于调试。
五、模块 4:记忆管理
让 Agent 跨任务/跨会话有记忆,是处理复杂任务的必备能力。最小可用版本支持:短期消息列表(上下文窗口)+ 长期向量库(按需检索)。
class Memory:
def __init__(self, collection_name="agent_memory"):
import chromadb
self.client = chromadb.PersistentClient(path="./memory")
self.collection = self.client.get_or_create_collection(collection_name)
def store(self, text, metadata=None):
import uuid
self.collection.add(documents=[text], ids=[str(uuid.uuid4())], metadatas=[metadata] or [{}])
def retrieve(self, query, top_k=3):
results = self.collection.query(query_texts=[query], n_results=top_k)
return "\n".join(results["documents"][0]) if results["documents"] else ""实现要点:短期记忆(对话历史)和长期记忆(向量库)双轨,Agent 每次循环把”相关长期记忆”注入上下文。
六、模块 5:规划与任务分解
复杂任务不能”一杆子捅到底”,需要拆成子任务。最小可用版本支持:Plan-and-Execute 模式(先生成完整计划,再逐步执行,执行中可重规划)。
class Planner:
def plan(self, goal, available_tools):
# 调 LLM 生成步骤列表
prompt = f"""把目标拆成 2-7 个步骤,每步用 available_tools 里的工具或子任务。
目标: {goal}
可用工具: {json.dumps(available_tools, ensure_ascii=False)}
返回 JSON 列表,每项: {{"step": 1, "action": "...", "tool": "...", "input": "..."}}"""
resp = self.llm.chat([{"role": "user", "content": prompt}])
return json.loads(resp.choices[0].message.content)实现要点:规划器本身也是一个 LLM 调用,通常用更便宜的模型(如 gpt-4o-mini)即可。
七、模块 6:可观测性
Agent 调试之所以难,是因为”决策”和”行动”分散在多步循环里。最小可用版本需要:trace_id 贯穿、每步记录 span、span 间能查因果。
import uuid
from datetime import datetime
class Tracer:
def __init__(self):
self.spans = []
def start(self, name, parent_id=None):
span = {
"id": str(uuid.uuid4()),
"parent_id": parent_id,
"name": name,
"start": datetime.now().isoformat()
}
self.spans.append(span)
return span["id"]
def end(self, span_id, output=None):
for s in self.spans:
if s["id"] == span_id:
s["end"] = datetime.now().isoformat()
s["output"] = output
break
def export(self):
return self.spans实现要点:span 设计要带 traceid(贯穿整个 Agent 任务)+ parentid(父子关系),便于画调用树。
八、模块 7:错误处理与重试
工具调用失败、LLM 速率限制、网络抖动……Agent 必须能优雅恢复。最小可用版本需要:工具调用失败重试(指数退避)+ LLM 速率限制处理 + 整体降级(工具调不通时改用 LLM 推理兜底)。
import time
class RetryPolicy:
def __init__(self, max_retries=3, base_delay=1):
self.max_retries = max_retries
self.base_delay = base_delay
def execute(self, func, *args, **kwargs):
for attempt in range(self.max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == self.max_retries - 1:
raise
delay = self.base_delay * (2 ** attempt)
time.sleep(delay)
return None实现要点:可重试错误(网络、超时)自动重试,不可重试错误(参数错、权限拒绝)直接失败转人工。
7 个模块的依赖关系可总结为下面这张表:
| 序号 | 模块 | 依赖 | 必需 | 估时(人天) |
|---|---|---|---|---|
| 1 | LLM 客户端 | 无 | 是 | 1-2 |
| 2 | 工具注册与执行 | 1 | 是 | 2-3 |
| 3 | 对话循环 | 1、2 | 是 | 2-3 |
| 4 | 记忆管理 | 1 | 推荐 | 2-3 |
| 5 | 规划与任务分解 | 1、2 | 推荐 | 3-5 |
| 6 | 可观测性 | 1、2、3 | 推荐 | 2-3 |
| 7 | 错误处理与重试 | 1、2 | 推荐 | 1-2 |
九、实施路线图
按下面 8 周时间表推进,可在 6 周交付 MVP:
- 第 1-2 周:模块 1-3(LLM 客户端 + 工具注册 + 对话循环),打通”用户问 → Agent 调工具 → 返回结果”闭环
- 第 3-4 周:模块 4-5(记忆 + 规划),让 Agent 能处理多步任务
- 第 5-6 周:模块 6-7(可观测性 + 错误处理),上生产前的最后一道门
- 第 7-8 周:接真实业务、收集问题、修 bug、加评估
十、不要先做的模块
几个常见的”过早优化”,应该延后:
- 多 Agent 协作:等单 Agent 跑稳再上,否则复杂度爆炸;
- 调试 UI:用 trace 日志先凑合,UI 后期再做;
- 评估体系:先靠人工事后抽审,有真实流量后再上自动化评估;
- 可视化流程编排:大多数任务用代码定义 Agent 就够,可视化是后期优化项。
常见问题(FAQ)
Q1：为什么不用 LangGraph / CrewAI / AutoGen 这些现成框架?
业务有特殊需求(比如必须走内部审批流)时自研更合适;否则先用现成框架,等真碰到瓶颈再考虑自研。
Q2：7 个模块必须按这个顺序吗?
基本是,前 3 个有强依赖(没 LLM 客户端就没对话循环),后 4 个可并行。
Q3：小项目也要做这 7 个吗?
不必。前 3 个必需,后 4 个按需——单 Agent 短任务可能不需要规划也不需要长期记忆。
