一、为什么我还要手写Agent？LangChain的Agent框架不够用吗？
先说结论：LangChain的AgentExecutor确实能跑，但当你需要精细控制“何时停止、何时重试、记忆怎么存”时，它的黑盒特性会让人抓狂。
我最近在做一个内部运维机器人，需要Agent自主调用查日志、执行命令、读配置三个工具，并基于结果决定下一步操作。用LangChain自带的create_react_agent跑了一周，发现三个痛点：
- 循环失控：Agent在某个工具返回错误后，会反复调用同一工具，最多一次卡了9轮才退出。
- 记忆混乱：默认的
ConversationBufferMemory会把工具返回的冗长日志也塞进上下文，导致token爆掉，单次任务平均消耗12k token，成本高得离谱。 - 错误处理缺失：工具抛异常时，Agent会直接崩溃，而不是尝试换个方式。
所以，我决定基于LangChain的底层组件，手写一个精简版的AutoGPT风格Agent——自己控制循环、记忆和错误恢复。
二、环境与版本：锁死版本，避免玄学问题
我的开发环境如下，建议保持一致：
- Python 3.10.13
- LangChain 0.1.0（注意，0.1.x和0.2.x的API差异巨大，尤其是
BaseTool的接口） - langchain-openai 0.0.5
- OpenAI模型：
gpt-4-turbo-2024-04-09（temperature=0.2） - 依赖安装：
pip install langchain==0.1.0 langchain-openai==0.0.5
为什么不用LangChain 0.2+？因为create_react_agent在0.2里改成了create_agent，且工具装饰器@tool的行为有变化，我懒得重新适配。
三、方案设计：AutoGPT的“思考-行动-观察”循环，手动实现
AutoGPT的核心思路是让LLM在循环中做三步：思考（Thought） → 行动（Action） → 观察（Observation），直到满足终止条件。
我的设计如下：
- 工具定义：用
BaseTool子类，而不是@tool装饰器（因为需要更精细控制错误返回）。 - 记忆管理：只保留最近3轮（思考+行动+观察）作为短期记忆，更早的丢弃，控制token预算。
- 循环控制：最大迭代5次，如果第5次还没得到最终答案，强制终止并返回当前最佳结果。
- 错误处理：工具内部捕获异常，返回结构化错误信息，让LLM能“看到”错误并自我修正。
整体流程图：
while step str:
try:
# 这里替换为真实的日志查询逻辑
if service == "auth-service":
return "2024-06-01 10:00:01 ERROR timeout connecting to db"
elif service == "order-service":
return "2024-06-01 10:00:02 INFO order created"
else:
return f"未找到服务 {service} 的日志"
except Exception as e:
# 关键：返回错误字符串，而不是抛出
return f"[工具内部错误] {str(e)}，请检查service参数是否拼写正确"
4.2 记忆管理：用deque实现滑动窗口记忆
我用了collections.deque(maxlen=6)，这里的6表示最多保留6条消息（3轮思考+3轮观察）。每次循环后，把新的(思考, 观察)追加进去，最老的自动弹出。
from collections import deque
from typing import List, Dict
class SlidingMemory:
def __init__(self, max_rounds: int = 3):
# maxlen = max_rounds * 2 (思考+观察)
self.storage = deque(maxlen=max_rounds * 2)
def add(self, thought: str, observation: str):
self.storage.append(("thought", thought))
self.storage.append(("observation", observation))
def build_context(self) -> str:
context = ""
for role, content in self.storage:
if role == "thought":
context += f"AI思考: {content}\n"
else:
context += f"观察结果: {content}\n"
return context
实测效果：加入滑动窗口后，单次任务token消耗从12k降到了4.5k（GPT-4-turbo输入价格$0.01/1k，单次任务成本从$0.12降到$0.045）。
4.3 循环控制与错误重试：最大5轮，失败提前终止
这里有个关键设计：如果LLM连续两次选择同一个工具且参数完全相同，我直接判定为死循环，强制跳出。这个逻辑救了我很多次。
def run_agent(user_query: str, tools: List[BaseTool], llm, max_steps: int = 5):
memory = SlidingMemory(max_rounds=3)
tool_map = {t.name: t for t in tools}
last_action = None
last_input = None
repeat_count = 0
for step in range(max_steps):
prompt = build_prompt(user_query, tools, memory.build_context())
response = llm.invoke(prompt).content
if "Final Answer:" in response:
return response.split("Final Answer:")[-1].strip()
action, action_input = parse_response(response) # 正则解析
if action == "无法决策":
return "Agent判定无法处理，请人工介入。"
# 死循环检测
if action == last_action and action_input == last_input:
repeat_count += 1
else:
repeat_count = 0
if repeat_count >= 2:
return f"检测到重复操作({action}: {action_input})，已强制终止。"
last_action, last_input = action, action_input
if action not in tool_map:
observation = f"工具 {action} 不存在，可用工具: {list(tool_map.keys())}"
else:
observation = tool_map[action].run(action_input)
memory.add(f"调用工具 {action} 参数 {action_input}", observation)
print(f"[Step {step+1}] action={action}, observation={observation[:50]}...")
return "达到最大步数，返回当前记忆中的最后观察结果。" + memory.build_context()
4.4 构建Prompt：工具描述动态拼接
这一步很关键，我把工具描述、记忆、用户问题拼在一起，并强调“你必须输出JSON格式的Action，或者‘Final Answer’”。
def build_prompt(user_query: str, tools: List[BaseTool], memory_context: str) -> str:
tools_desc = "\n".join([f"- {t.name}: {t.description}" for t in tools])
prompt = f"""你是一个运维机器人，请根据工具返回结果决定下一步。你必须严格按以下格式输出：
1. 如果需要调用工具，输出：Action: 工具名\nAction Input: JSON字符串
2. 如果得到最终答案，输出：Final Answer: 你的答案
可用工具：
{tools_desc}
历史记忆：
{memory_context or "无"}
用户问题：{user_query}
现在开始："""
return prompt
五、踩坑与优化：三个真实教训
5.1 工具返回类型必须是字符串，否则LangChain会报错
第一次用BaseTool时，我的_run返回了dict，结果LangChain内部会尝试用str()转换，但嵌套的中文引号会被转义，导致LLM误读。解决方案：_run一律返回格式化好的字符串。
5.2 OpenAI的JSON输出不稳定，用正则兜底
我最初让LLM输出严格JSON，但实测在temperature=0.2下，大约有7%的概率输出Action: query_log\nAction Input: {"service": "auth"}这种非JSON格式。后来我写了正则re.search(r'Action Input: (.+)', response)，直接抓取后半部分，再用json.loads解析，失败就返回“无法决策”。
5.3 死循环检测比“最大步数”更有效
之前只设了max_steps=5，但LLM可能会在5步内重复调用同一个工具3次，浪费了步数。加了repeat_count >= 2的检测后，平均提前1.8步终止，单次任务耗时从21秒降到了14秒。
六、效果数据：对比LangChain原生Agent
我用了20个测试用例（包含正常查询、工具不存在、参数错误、死循环诱导）做对比：
| 指标 | LangChain AgentExecutor | 手写Agent |
|---|---|---|
| 工具调用成功率 | 67% | 94% |
| 平均完成时间 | 23s | 14s |
| 平均token消耗 | 11.8k | 4.6k |
| 死循环卡死率 | 25% | 0% |
| 错误恢复率（工具报错后自动换参） | 30% | 85% |
提升最明显的是“错误恢复率”——手写Agent可以在工具返回错误信息后，从错误字符串中提取关键词（比如“未找到服务”），然后修正参数重试。而LangChain原生Agent往往会把错误信息当作最终答案直接返回。
七、总结：什么时候该手写？
如果只是Demo级别，直接用AgentExecutor没问题。但如果是生产环境，涉及成本控制、死循环兜底、工具错误恢复，我强烈建议基于LangChain底层组件手写循环。我的代码总共约200行，比想象中少，但控制力强了一个量级。
下一步我会尝试把SlidingMemory换成向量数据库存储（比如Chroma），支持长对话的持久化记忆，等有数据了再来分享。有问题欢迎评论区交流，特别是那些被LangChain黑盒坑过的朋友，你们懂的。
