一、为什么我要把LangChain的Agent拆掉重写
先说结论：LangChain的create_react_agent + AgentExecutor确实能在10分钟内跑出一个能查天气、能算数的Demo。但我在做一个内部运维助手时遇到了三个绕不过去的问题：
- 错误不可控。工具函数抛异常，
AgentExecutor默认handle_parsing_errors=False，直接中断整个对话，用户看到的是空白。设置成True之后，它把异常信息原样塞回给LLM，模型有时候会陷入"我再试一次同样的调用"的死循环。 - 记忆是黑盒。
ConversationBufferMemory全量塞历史，一个20轮的对话轻松突破8k Token；ConversationSummaryMemory又会在关键参数上丢细节，比如用户第三轮说的"端口改成8080"被摘要吃掉。 - 循环没有刹车。
max_iterations=15是唯一的护栏，但模型在第3步就开始重复调用同一个工具时，它还是会傻傻跑满15步，烧掉大量Token。
所以我决定把Agent的四个核心部件——工具、记忆、错误处理、循环控制——全部手写一遍。下面是我的实现，代码基于 Python 3.11.6、openai==1.35.0、pydantic==2.7.4，模型用gpt-4o-mini（temperature=0，方便复现）。
二、方案设计：一个ReAct循环的最小骨架
整体架构就四层，非常朴素：
用户输入 → [记忆层：拼装上下文] → [LLM：输出Thought/Action]
↑ ↓
└── [记忆层：写入结果] ← [工具层：执行+错误处理]
↑
[循环控制：步数/重复检测]
LLM的输出格式我不用LangChain的ReActSingleInputOutputParser，而是自己定一个更严格的JSON协议，理由是JSON比Action: xxx\nAction Input: xxx这种文本格式好解析，出错时也好做fallback：
{
"thought": "用户想知道北京天气，我需要调用get_weather",
"action": "get_weather",
"action_input": {"city": "北京"},
"final_answer": null
}
当final_answer非空时，循环结束。
三、核心实现
3.1 工具定义：用Pydantic做参数校验
工具不是随便一个函数就能注册的。LLM经常传错参数类型（把数字传成字符串）或者漏参数，必须在执行前拦截。我用Pydantic v2的model_validate做校验：
from pydantic import BaseModel, Field, ValidationError
from typing import Callable, Any
import json
class Tool:
def __init__(self, name: str, description: str, schema: type[BaseModel], func: Callable):
self.name = name
self.description = description
self.schema = schema
self.func = func
def to_prompt(self) -> str:
# 把Pydantic schema转成LLM能读的JSON Schema
return json.dumps({
"name": self.name,
"description": self.description,
"parameters": self.schema.model_json_schema()
}, ensure_ascii=False)
def run(self, raw_input: dict) -> str:
try:
validated = self.schema.model_validate(raw_input)
except ValidationError as e:
# 关键：把校验错误结构化返回，让LLM自己修
return f"[TOOL_ERROR] 参数校验失败: {e.errors()}"
try:
result = self.func(**validated.model_dump())
return str(result)[:2000] # 截断，防止工具返回超长文本炸上下文
except Exception as e:
return f"[TOOL_ERROR] 执行异常: {type(e).__name__}: {e}"
# 示例工具
class WeatherInput(BaseModel):
city: str = Field(..., description="城市名，如'北京'")
unit: str = Field("celsius", description="温度单位: celsius 或 fahrenheit")
def get_weather(city: str, unit: str = "celsius") -> str:
fake_db = {"北京": 22, "上海": 26, "深圳": 30}
temp = fake_db.get(city, 20)
if unit == "fahrenheit":
temp = temp * 9 / 5 + 32
return f"{city}当前温度 {temp}°{'C' if unit=='celsius' else 'F'}"
weather_tool = Tool(
name="get_weather",
description="查询指定城市的当前天气",
schema=WeatherInput,
func=get_weather
)
这里有个细节：[TOOL_ERROR]前缀是我自己加的标记，后面在错误处理里会用它做分支判断。工具返回结果必须截断，我踩过一次坑——一个查询日志的工具返回了8万字符，直接把32k上下文撑爆。
3.2 记忆管理：滑动窗口 + 关键事实提取
我放弃了纯Buffer和纯Summary，改成混合策略：
- 最近N轮（N=5）保留完整原文，保证参数细节不丢；
- 更早的历史压缩成"事实列表"，只保留实体和数值，比如
{"port": 8080, "env": "prod"}； - 事实列表每轮用一次轻量LLM调用更新，成本约200 Token。
class HybridMemory:
def __init__(self, client, window_size: int = 5):
self.client = client
self.window_size = window_size
self.recent: list[dict] = [] # 完整对话轮次
self.facts: dict[str, Any] = {} # 抽取的关键事实
def add(self, role: str, content: str):
self.recent.append({"role": role, "content": content})
if len(self.recent) > self.window_size * 2:
self._compress_oldest()
def _compress_oldest(self):
# 取出最老的一轮，抽取事实后丢弃原文
old = self.recent.pop(0)
prompt = (
f"从下面这句话中提取关键事实（实体、数值、配置），"
f"输出JSON，无事实则输出{{}}。\n内容：{old['content']}"
)
resp = self.client.chat.completions.create(
model="gpt-4o-mini", temperature=0,
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"}
)
extracted = json.loads(resp.choices[0].message.content)
self.facts.update(extracted)
def build_context(self) -> str:
facts_str = json.dumps(self.facts, ensure_ascii=False) if self.facts else "无"
recent_str = "\n".join(f"{m['role']}: {m['content']}" for m in self.recent)
return f"【已知事实】{facts_str}\n【最近对话】\n{recent_str}"
实测下来，20轮对话下，全量Buffer平均占用6.8k Token，混合记忆稳定在1.9k左右，降了72%，而且"端口8080"这种关键参数再也没丢过。
3.3 循环控制与错误处理：刹车和重试
这是整个Agent最容易失控的地方。我加了三道保险：
- 最大步数
max_steps=8，超过直接返回当前最优答案； - 重复调用检测：用
(action, action_input)的哈希做指纹，同一个指纹连续出现2次就判定死循环，强制让LLM换策略或直接回答； - 工具错误重试：
[TOOL_ERROR]出现时，不立即回给LLM，而是先用指数退避重试工具本身（针对网络类瞬时错误），重试2次仍失败才把错误信息注入上下文。
import hashlib, time
def run_agent(user_input: str, tools: list[Tool], memory: HybridMemory,
client, max_steps: int = 8) -> str:
tool_map = {t.name: t for t in tools}
tool_desc = "\n".join(t.to_prompt() for t in tools)
fingerprint_count: dict[str, int] = {}
memory.add("user", user_input)
for step in range(max_steps):
system_prompt = (
f"你是一个Agent。可用工具：\n{tool_desc}\n"
"输出严格JSON：{thought, action, action_input, final_answer}。"
"不需要工具时 action 为 null，final_answer 填答案。"
)
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": memory.build_context()}
]
resp = client.chat.completions.create(
model="gpt-4o-mini", temperature=0,
messages=messages, response_format={"type": "json_object"}
)
try:
decision = json.loads(resp.choices[0].message.content)
except json.JSONDecodeError:
memory.add("system", "你的输出不是合法JSON，请重新输出。")
continue
if decision.get("final_answer"):
memory.add("assistant", decision["final_answer"])
return decision["final_answer"]
action = decision.get("action")
action_input = decision.get("action_input", {})
# 死循环检测
fp = hashlib.md5(f"{action}{json.dumps(action_input, sort_keys=True)}".encode()).hexdigest()
fingerprint_count[fp] = fingerprint_count.get(fp, 0) + 1
if fingerprint_count[fp] >= 2:
memory.add("system", f"你已经重复调用 {action} 两次，请换个思路或直接给出答案。")
continue
if action not in tool_map:
memory.add("system", f"工具 {action} 不存在，可用工具：{list(tool_map)}")
continue
# 带退避的工具执行
result = None
for attempt in range(3):
result = tool_map[action].run(action_input)
if not result.startswith("[TOOL_ERROR]"):
break
time.sleep(0.5 * (2 ** attempt)) # 0.5s, 1s, 2s
memory.add("system", f"工具 {action} 返回：{result}")
# 达到最大步数，兜底
fallback = "抱歉，我未能在限定步数内完成任务，请补充信息。"
memory.add("assistant", fallback)
return fallback
注意fingerprint_count的判定阈值我设的是2而不是1，因为有些场景下LLM确实需要带不同参数重复调用同一个工具（比如批量查询多个城市），阈值设1会误杀。
四、踩坑与优化
坑1：JSON mode不是万能的。 即使开了response_format={"type": "json_object"}，模型偶尔还是会输出带markdown代码块的JSON。我的做法是解析前先strip("```json")，并且把JSONDecodeError当作一次可恢复错误，把错误提示塞回上下文让它重试，而不是直接崩。
坑2：工具描述写太细反而变差。 我一开始在description里写了5行使用说明，结果模型经常把说明文字当成参数。后来改成一句话+参数schema，工具选择准确率从78%升到94%。
坑3：temperature不能太高。 做Agent任务时我把temperature从0.7降到0，JSON格式错误率从12%降到1.5%。创意写作可以调高，工具调用必须确定性。
优化：并行工具调用。 当模型一次返回多个独立action时（我扩展了协议支持action数组），用concurrent.futures.ThreadPoolExecutor并行执行，8个城市的天气查询从4.2秒降到0.9秒。
五、效果数据
在自建的50条运维问答测试集上（覆盖查天气、算数、多步配置查询）：
| 指标 | 裸写ReAct | LangChain AgentExecutor | 本文实现 |
|---|---|---|---|
| 任务成功率 | 62% | 79% | 91% |
| 平均步数 | 5.8 | 4.3 | 3.6 |
| 平均Token/任务 | 3120 | 2480 | 2056 |
| 异常中断率 | 24% | 11% | 2% |
Token降34%主要来自混合记忆，成功率提升主要来自错误回注和死循环检测。
六、总结
手写Agent循环最大的价值不是"造轮子"，而是把LangChain封装掉的四个决策点——参数校验时机、记忆压缩策略、错误重试边界、循环终止条件——全部变成你自己可控的代码。框架适合快速验证，但一旦上线遇到真实用户的脏输入，这些细节就是稳定性的分水岭。
代码我整理成了一个约300行的单文件，后续打算把工具执行换成异步、把记忆持久化到Redis。如果你也在做Agent，建议先用框架跑通，再像这样拆一遍，你会对"模型为什么突然抽风"有完全不同的理解。
