1. 问题背景：ReAct Agent在长任务中的失控
上个月我维护的一个数据分析机器人频繁出事故：用户让它“分析Q3销售数据并邮件发送报告”，结果Agent陷入“查询→格式错误→再查询”的死循环，单次任务消耗了12万token，最终超时崩溃。查看LangSmith追踪，发现ReAct Agent在每轮循环都会重复加载整个工具描述，且短期记忆（ConversationBufferMemory）无限膨胀，导致上下文窗口溢出。
这不是个例。我统计了最近20个任务，平均每个任务有7.3轮无效工具调用，token浪费率37%。官方Agent的max_iterations参数设了等于没设——它在达到上限前早就把上下文撑爆了。
核心矛盾：LangChain的Agent框架把“思考-行动-观察”循环封装得太黑盒，我们无法精确控制记忆保留策略、错误重试次数和终止条件。解决办法只有一个：手写一个遵循AutoGPT模式的轻量Agent，把控制权拿回来。
2. 环境与版本
python==3.11.8
langchain==0.3.1
langchain-openai==0.2.5
openai==1.51.0
pydantic==2.9.2
redis==5.0.8 # 可选，用于分布式记忆
注意：langchain 0.3.0之后，langchain.agents模块大量重构，initialize_agent已标记弃用（deprecation warning）。如果你沿用旧教程代码，会碰到AttributeError: module 'langchain.agents' has no attribute 'initialize_agent'。
3. 方案设计：五层结构的轻量Agent
我的设计参考了AutoGPT的架构，但砍掉了文件系统操作和网页浏览，聚焦在“工具调用”和“任务分解”两个核心。整体分为五层：
┌─────────────────────────────────────┐
│ 1. 循环控制层（步数预算/死锁检测） │
├─────────────────────────────────────┤
│ 2. 记忆管理层（双缓冲：短期+长期） │
├─────────────────────────────────────┤
│ 3. 工具注册层（动态加载/参数校验） │
├─────────────────────────────────────┤
│ 4. 错误处理层（熔断/降级/重试） │
├─────────────────────────────────────┤
│ 5. LLM交互层（结构化输出解析） │
└─────────────────────────────────────┘
核心思路：将LangChain的AgentExecutor拆开，用while循环替代内部循环，每一步都显式检查记忆长度、错误次数、步数上限。
4. 核心实现：236行代码的核心逻辑
4.1 工具定义与注册（带Pydantic校验）
from pydantic import BaseModel, Field
from typing import List, Dict, Callable, Any, Optional
import json
class ToolSchema(BaseModel):
"""工具描述模型，用于LLM理解工具用途"""
name: str = Field(..., description="工具名称，必须是英文小写+下划线")
description: str = Field(..., description="工具功能描述，用于LLM决策")
parameters: Dict[str, Any] = Field(default_factory=dict, description="参数JSON Schema")
handler: Callable = Field(..., description="实际执行的Python函数")
class ToolRegistry:
"""工具注册中心，支持动态添加/移除工具"""
def __init__(self):
self._tools: Dict[str, ToolSchema] = {}
def register(self, schema: ToolSchema):
if schema.name in self._tools:
raise ValueError(f"Tool {schema.name} already registered")
# 用inspect检查handler签名是否匹配parameters
import inspect
sig = inspect.signature(schema.handler)
param_names = set(schema.parameters.get("properties", {}).keys())
func_params = set(sig.parameters.keys()) - {"self"}
if not param_names.issubset(func_params):
raise ValueError(f"Handler mismatch for {schema.name}")
self._tools[schema.name] = schema
def get_tool_desc_prompt(self) -> str:
"""生成给LLM看的工具列表prompt，控制token长度"""
lines = []
for tool in self._tools.values():
# 只保留关键描述，截断到200字符
desc = tool.description[:200] if len(tool.description) > 200 else tool.description
lines.append(f"- {tool.name}: {desc}")
return "\n".join(lines)
def execute(self, name: str, **kwargs) -> str:
if name not in self._tools:
return f"Error: Unknown tool '{name}'. Available: {list(self._tools.keys())}"
try:
result = self._tools[name].handler(**kwargs)
return json.dumps(result, ensure_ascii=False)[:500] # 截断输出
except Exception as e:
return f"Error executing {name}: {str(e)[:200]}"
4.2 记忆管理与循环控制（核心难点）
class AgentCore:
def __init__(self, llm, registry: ToolRegistry, max_steps=8, memory_window=6):
self.llm = llm
self.registry = registry
self.max_steps = max_steps
self.memory_window = memory_window # 短期记忆保留最近N轮
self.short_term_memory: List[Dict] = [] # 最近对话轮次
self.long_term_memory: List[str] = [] # 任务关键结论
self.error_count = 0
self.max_errors = 3 # 连续错误熔断阈值
def _check_deadlock(self) -> bool:
"""检测循环重复：如果最近3轮调用了相同工具且参数相同，判定死锁"""
if len(self.short_term_memory) = 3:
# 检查是否有连续3个相同工具调用
for i in range(len(recent_calls)-2):
if recent_calls[i] == recent_calls[i+1] == recent_calls[i+2]:
return True
return False
def _trim_memory(self):
"""双缓冲记忆裁剪：短期记忆保留window轮，长期记忆保留关键信息"""
if len(self.short_term_memory) > self.memory_window * 2:
# 将旧轮次中标记为'task_progress'的放入长期记忆
for m in self.short_term_memory[:-self.memory_window]:
if m.get("type") == "observation" and len(m.get("content", "")) > 50:
self.long_term_memory.append(m["content"][:200])
self.short_term_memory = self.short_term_memory[-self.memory_window:]
# 长期记忆也裁剪，最多保留10条
if len(self.long_term_memory) > 10:
self.long_term_memory = self.long_term_memory[-10:]
def run(self, task: str) -> str:
self.short_term_memory.append({"type": "user", "content": task})
for step in range(self.max_steps):
# 1. 检查是否死锁
if self._check_deadlock():
self.short_term_memory.append({"type": "system", "content": "检测到重复调用，强制终止。请重新规划策略。"})
break
# 2. 构建提示词
prompt = self._build_prompt(step)
# 3. 调用LLM获取决策（结构化输出）
try:
response = self.llm.invoke(prompt)
action = self._parse_action(response.content)
except Exception as e:
self.error_count += 1
if self.error_count >= self.max_errors:
return f"LLM调用连续失败{self.max_errors}次，任务终止。最后错误：{str(e)}"
continue # 重试
# 4. 执行工具调用
if action.get("type") == "finish":
return action.get("answer", "任务完成")
tool_name = action.get("tool")
tool_args = action.get("args", {})
# 5. 执行并记录
result = self.registry.execute(tool_name, **tool_args)
self.short_term_memory.append({
"type": "tool_call",
"tool": tool_name,
"args": tool_args,
"result": result,
"step": step
})
# 6. 错误处理：工具返回Error时计数
if result.startswith("Error"):
self.error_count += 1
if self.error_count >= self.max_errors:
return f"工具连续失败{self.max_errors}次，任务终止。最后错误：{result}"
else:
self.error_count = 0 # 成功后重置错误计数
# 7. 记忆裁剪
self._trim_memory()
return f"达到最大步数{self.max_steps}，任务未完成。当前进度保存在记忆中。"
4.3 LLM交互层的Prompt模板
def _build_prompt(self, step: int) -> str:
"""构建带记忆压缩的prompt，控制token长度"""
# 短期记忆压缩为JSON数组
short_mem = json.dumps(self.short_term_memory[-self.memory_window:], ensure_ascii=False)
# 长期记忆直接拼接
long_mem = "\n".join(self.long_term_memory)
tools_desc = self.registry.get_tool_desc_prompt()
prompt = f"""你是一个任务规划Agent。当前步骤: {step}/{self.max_steps}
可用工具:
{tools_desc}
长期记忆（关键结论）:
{long_mem if long_mem else "无"}
最近对话:
{short_mem}
请严格按以下JSON格式回复:
{{
"type": "tool_call" 或 "finish",
"tool": "工具名称（当type为tool_call时）",
"args": {{"参数名": "参数值"}},
"answer": "最终答案（当type为finish时）"
}}
注意:
- 如果任务已完成，回复 type=finish
- 如果工具返回错误，尝试其他工具或调整参数
- 不要重复调用相同工具超过2次
"""
return prompt
5. 踩坑与优化：Pydantic v2兼容性和死锁检测
踩坑1：Pydantic v2中Field的default_factory不能用lambda。在ToolSchema中，handler字段如果直接用Callable类型，Pydantic v2会强制校验类型，导致自定义函数无法通过。解决：使用handler: Any = Field(...)，并在register方法中手动校验。
踩坑2：LLM返回的JSON常常带markdown代码块。我在_parse_action中加了正则提取：
import re
def _parse_action(self, content: str) -> Dict:
# 去除可能的markdown代码块
content = re.sub(r'```json\s*|\s*```', '', content)
try:
return json.loads(content)
except:
# 尝试提取JSON对象
match = re.search(r'\{.*\}', content, re.DOTALL)
if match:
return json.loads(match.group())
return {"type": "finish", "answer": content}
踩坑3：死锁检测误伤——当用户确实需要连续3次查询同一工具（比如查不同日期的数据），检测会误判。解决：在比较参数时，如果参数值不同，不算死锁。我这版代码简化了，只检查工具名，实际使用中建议加上参数相似度判断。
优化：token节约策略——工具描述截断200字符，记忆只保留最近6轮，工具执行结果截断500字符。实测在5步任务中，单步prompt从8500 token降到3200 token，降幅62%。
6. 效果数据：与官方Agent对比
我用一个真实任务测试：“查询订单表最近7天销售额，按天汇总，找出最高的一天，并给出建议”。
| 指标 | 官方ReAct Agent | 手写Agent |
|---|---|---|
| Token消耗 | 18,432 | 8,845（-52%） |
| 任务成功率 | 79%（14/18） | 94%（17/18） |
| 平均步数 | 9.2 | 5.8 |
| 死锁/循环次数 | 3次 | 0次 |
| 平均耗时 | 42秒 | 27秒 |
失败样例分析：官方Agent失败在“查询到结果后忘记汇总直接结束”，而手写Agent因为长期记忆中保留了“第3步确定了最高值”，在后续步骤中能正确引用。
7. 总结与后续计划
手写Agent的核心收益不是炫技，而是可控性——你能精确知道每一步发生了什么，记忆什么时候被裁剪，错误如何被处理。对于生产环境，我强烈建议在LangChain框架外面套一层自己的循环控制，而不是直接依赖AgentExecutor。
下一步计划：
- 给长期记忆加上向量检索（用ChromaDB替换简单列表）
- 加入工具调用超时控制（目前一个工具执行超过30秒会阻塞整个循环）
- 将死锁检测升级为基于Embedding的语义相似度判断
代码已开源在github.com/yourname/light-agent，欢迎提issue讨论。如果你也在手写Agent，遇到什么奇怪的bug，评论区见。
