1. 项目概述
今天我想分享一个我自己在开发AI Agent系统时总结出来的最小但完整的复杂Agent系统模板。这个模板虽然只有200行左右的代码量,但完整实现了Agent系统的5个核心模块:任务规划(Planner)、工具调用(Tool Router)、记忆(Memory)、反思(Reflection)和执行循环(Agent Loop)。
这个模板的设计灵感来源于我在实际项目中观察到的几个主流Agent框架,包括LangGraph、Microsoft AutoGen和SWE-agent的核心架构。通过这个精简但功能完整的实现,我们可以清晰地理解一个复杂Agent系统的基本工作原理。
需要模型API调用? 免费领10W Token,多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 核心模块解析
2.1 Memory系统实现
Memory模块是Agent系统的"大脑皮层",负责存储和检索历史交互信息。在我们的实现中,我采用了最简单的列表结构来存储历史记录:
python复制class Memory:
def __init__(self):
self.history = []
def add(self, item):
self.history.append(item)
def get_context(self):
return "\n".join(self.history[-10:])
这里有几个设计考量:
- 使用列表而非数据库是为了保持最小实现
- 只保留最近10条记录是为了防止上下文过长
- 简单的字符串拼接作为上下文表示
在实际项目中,我通常会升级为:
- 使用向量数据库存储长期记忆
- 实现基于时间或重要性的记忆筛选
- 添加记忆分块和索引功能
2.2 Tool系统设计
Tool系统是Agent与外部世界交互的接口。我们的模板实现了两个基础工具:
python复制def search_tool(query: str):
return f"Search results for {query}"
def calculator(expression: str):
try:
return str(eval(expression))
except:
return "calculation error"
TOOLS = {
"search": search_tool,
"calculator": calculator
}
工具系统的关键点:
- 统一接口设计:所有工具都接受字符串输入,返回字符串输出
- 错误处理:特别是像calculator这样的工具需要有健壮的错误处理
- 工具注册机制:通过字典维护工具映射
在实际应用中,我通常会:
- 添加工具描述元数据
- 实现工具权限管理
- 开发工具组合调用能力
3. Agent核心逻辑
3.1 Planner模块
Planner负责将高层目标分解为可执行的任务序列:
python复制def planner(goal, context):
prompt = f"""
You are a planner.
Goal:
{goal}
Context:
{context}
Break the goal into steps in JSON:
example:
["step1","step2"]
"""
result = call_llm(prompt)
try:
steps = json.loads(result)
except:
steps = [goal]
return steps
Planner的设计要点:
- 使用LLM进行目标分解
- 结构化输出要求(JSON格式)
- 异常情况回退机制
我在实际项目中发现,好的Planner需要:
- 考虑任务依赖关系
- 评估任务可行性
- 支持多轮规划迭代
3.2 Tool Router实现
Tool Router决定哪个工具最适合处理当前任务:
python复制def choose_tool(task):
prompt = f"""
Choose tool for task:
task: {task}
Available tools:
search
calculator
Return tool name or NONE
"""
result = call_llm(prompt).strip()
if result in TOOLS:
return result
return None
Router的关键设计:
- 明确列出可用工具
- 简单直接的决策机制
- 清晰的返回约定
实际应用中,我会增强:
- 工具选择置信度评估
- 多工具协同调用
- 工具使用历史记录
4. 执行循环与反思机制
4.1 Agent主循环
Agent类封装了完整的执行逻辑:
python复制class Agent:
def __init__(self):
self.memory = Memory()
def execute_task(self, task):
tool = choose_tool(task)
if tool:
result = TOOLS[tool](task)
else:
result = call_llm(task)
self.memory.add(f"TASK:{task}")
self.memory.add(f"RESULT:{result}")
return result
def run(self, goal):
print("Goal:", goal)
context = self.memory.get_context()
tasks = planner(goal, context)
print("Plan:", tasks)
for task in tasks:
print("\nExecuting:", task)
result = self.execute_task(task)
print("Result:", result)
good = reflect(task, result)
if not good:
print("Reflection: retrying")
result = self.execute_task(task)
return "done"
循环的关键流程:
- 目标接收
- 上下文获取
- 任务规划
- 任务执行
- 结果反思
- 记忆更新
4.2 Reflection机制
Reflection评估任务执行质量:
python复制def reflect(task, result):
prompt = f"""
Evaluate the result.
task: {task}
result: {result}
Is it good? reply YES or NO
"""
decision = call_llm(prompt)
return "YES" in decision
反思机制的设计考虑:
- 简单二元评估
- 基于LLM的质量判断
- 触发重试机制
在实际系统中,我会扩展为:
- 多维度评估
- 错误原因分析
- 自适应学习机制
5. 项目扩展方向
5.1 多Agent协作
参考Microsoft AutoGen的设计,可以扩展为:
python复制class PlannerAgent:
def plan(self, goal):
pass
class CoderAgent:
def code(self, task):
pass
class ReviewerAgent:
def review(self, code):
pass
多Agent系统的关键点:
- 角色专业化
- 通信协议
- 协调机制
5.2 图工作流
借鉴LangGraph的思路,可以实现:
python复制class WorkflowNode:
def __init__(self, agent, condition):
self.agent = agent
self.condition = condition
class WorkflowGraph:
def add_edge(self, from_node, to_node):
pass
图工作流的优势:
- 复杂流程可视化
- 条件分支支持
- 并行执行能力
5.3 计算环境集成
类似OpenDevin,可以添加:
python复制def run_shell(command):
pass
def edit_file(path, content):
pass
def execute_python(code):
pass
环境集成的关键能力:
- 系统级操作
- 文件交互
- 代码执行
6. 实际应用建议
6.1 项目结构优化
建议的目录结构:
code复制agent/
├── agent.py # 主Agent类
├── planner.py # 规划逻辑
├── tools.py # 工具集合
├── memory.py # 记忆系统
├── reflection.py # 反思机制
└── llm.py # LLM封装
这种结构的优势:
- 模块化设计
- 职责分离
- 易于扩展
6.2 生产环境考量
从原型到生产需要添加:
- 监控系统:跟踪Agent行为
- 日志记录:问题诊断
- 性能优化:响应时间
- 安全控制:权限管理
6.3 性能优化技巧
基于我的实践经验:
- 缓存常用工具结果
- 并行化独立任务
- 优化提示词设计
- 限制递归深度
7. 开发经验分享
在开发这类系统时,我总结出几个关键经验:
- 保持核心简洁:初期不要过度设计
- 模块边界清晰:便于后续扩展
- 测试驱动开发:特别是工具集成
- 监控关键指标:成功率、响应时间等
一个常见的误区是过早优化。根据我的经验,应该:
- 先让核心流程跑通
- 收集真实使用数据
- 基于数据优化瓶颈点
另一个重要建议是建立完善的测试套件,特别是对于:
- 工具调用边界情况
- 规划器的任务分解
- 反思机制的质量判断
最后,文档和示例同样重要。我通常会:
- 为每个工具编写使用示例
- 记录典型工作流
- 维护常见问题解答
