一、为什么我要手写一个Agent，而不是直接用AutoGPT
先说结论：AutoGPT这类框架适合演示，不适合生产。我上一个项目用AutoGPT做内部工单处理，跑了三周，踩了一堆坑——它把"查数据库"和"发邮件"两个工具在同一个循环里反复调用了11次，最后因为上下文超长直接崩了。日志里根本看不出它为什么这么决策，因为整个prompt是框架拼的，你只能看到最终结果。
LangChain的AgentExecutor好一些，但它的默认实现里，max_iterations、early_stopping_method、handle_parsing_errors这些参数散落在各处，出问题时你得翻源码才知道行为。我需要的是一个每一行逻辑我都能解释清楚的Agent。
所以这次我决定手写。目标很明确：一个单文件、约300行、能跑通"查天气→决定是否带伞→写邮件"这类多步任务的Agent。技术选型：LangChain 0.2.16（只用它的ChatModel和Tool抽象，不用AgentExecutor）、OpenAI GPT-4o-mini（便宜，0.15美元/百万输入token）、Python 3.11。
二、环境与版本：别小看这些依赖
依赖清单我列一下，版本号都是实测能跑通的，别随便升：
langchain==0.2.16
langchain-openai==0.1.23
pydantic==2.8.2
openai==1.40.6
python-dotenv==1.0.1
环境变量就两个：OPENAI_API_KEY和OPENAI_BASE_URL（我用的是代理，国内直连经常超时）。temperature设成0，Agent场景不需要创造性，要的是稳定复现。
一个提醒：LangChain 0.2.x的ChatOpenAI默认会重试2次，但这个重试是HTTP层的，不处理"模型返回了非法JSON"这种逻辑错误。所以后面错误处理那节，我会自己再包一层。
三、方案设计：ReAct循环的四个支柱
整体架构就是一个while循环，每轮做四件事：
- 构造消息：系统prompt + 历史记忆 + 当前观察
- 调用模型：让LLM输出"思考+动作"
- 解析动作：如果是工具调用，执行工具，把结果作为Observation塞回消息；如果是Final Answer，退出循环
- 更新记忆：把这一轮压缩进记忆，防止上下文爆炸
四个支柱对应四个模块：工具定义、记忆管理、错误处理、循环控制。下面逐个拆。
四、核心实现：代码逐段讲
4.1 工具定义：用Pydantic做参数校验，别信LLM的JSON
LLM生成的工具参数经常缺字段、类型错。我第一版直接用json.loads，结果{"city": "北京", "days": "3"}里days是字符串，下游函数直接TypeError。所以工具必须带Schema校验。
from pydantic import BaseModel, Field, ValidationError
from langchain_core.tools import tool
import json
class WeatherInput(BaseModel):
city: str = Field(description="城市名，中文，如'北京'")
days: int = Field(default=1, ge=1, le=7, description="预报天数，1-7")
@tool("get_weather", args_schema=WeatherInput)
def get_weather(city: str, days: int = 1) -> str:
"""查询指定城市未来N天天气。返回温度和降水概率。"""
# 这里换成真实API，演示用mock
mock = {"北京": [(28, 0.1), (30, 0.6), (26, 0.3)],
"上海": [(31, 0.4), (29, 0.8), (30, 0.2)]}
data = mock.get(city, [(25, 0.5)] * 7)[:days]
return json.dumps([{"temp": t, "rain_prob": r} for t, r in data], ensure_ascii=False)
# 工具注册表：名字 -> 可调用对象
TOOLS = {"get_weather": get_weather}
关键点：args_schema让LangChain在tool.invoke()时自动做校验，非法参数抛ValidationError，我在错误处理里捕获它，把错误信息回传给LLM让它重试。这比让它自己猜"哪里错了"有效得多。
4.2 记忆管理：滑动窗口 + 摘要压缩，双保险
Agent跑长任务时，历史消息会撑爆上下文。我的策略是两层：
- 短期记忆：保留最近6轮完整对话（约1500 token）
- 长期摘要：超过6轮后，把最老的一轮用LLM压缩成一句话，存进
summary列表
class AgentMemory:
def __init__(self, window_size=6, max_summary=10):
self.window = [] # 最近N轮完整消息
self.summary = [] # 历史摘要
self.window_size = window_size
self.max_summary = max_summary
def add(self, role, content):
self.window.append({"role": role, "content": content})
if len(self.window) > self.window_size:
old = self.window.pop(0)
# 用LLM压缩，这里简化为截断；生产环境调一次小模型
snippet = old["content"][:80].replace("\n", " ")
self.summary.append(f"[{old['role']}] {snippet}...")
if len(self.summary) > self.max_summary:
self.summary.pop(0)
def build_context(self):
ctx = []
if self.summary:
ctx.append({"role": "system",
"content": "历史摘要：" + " | ".join(self.summary)})
return ctx + self.window
实测：窗口6 + 摘要10，单任务平均输入token从4200降到1800，降幅57%，而任务完成率没掉（因为关键信息都在最近几轮和摘要里）。
4.3 循环控制：最大步数 + 死循环检测
这是最容易出事的地方。我见过Agent连续5次调用同一个工具、参数一模一样，纯粹是模型卡住了。两个保护：
max_steps=8：硬上限，超了直接返回错误- 动作指纹：把
(tool_name, args_json)哈希，连续2次相同就强制打断，给模型注入一条"你重复了，换个思路"的提示
import hashlib
def action_fingerprint(tool_name, args):
raw = f"{tool_name}:{json.dumps(args, sort_keys=True)}"
return hashlib.md5(raw.encode()).hexdigest()
# 循环主体里
fingerprints = []
...
fp = action_fingerprint(tool_name, args)
if fingerprints[-2:] == [fp, fp]:
memory.add("system", "检测到重复动作，请换策略或直接给出最终答案。")
continue
fingerprints.append(fp)
4.4 错误处理：三级重试 + 降级
分三层：
- 参数校验失败（ValidationError）：把错误信息塞回Observation，让LLM重试，最多2次
- 工具执行异常（网络超时、API 500）：
tenacity重试3次，指数退避 - 模型输出解析失败（不是合法JSON）：降级为"请用纯文本回答"，跳过工具调用
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=8))
def safe_invoke(tool, args):
return tool.invoke(args)
def execute_action(action, memory):
try:
tool = TOOLS[action["tool"]]
return safe_invoke(tool, action["args"])
except ValidationError as e:
return f"参数错误：{e.errors()[0]['msg']}，请修正后重试。"
except Exception as e:
return f"工具执行失败：{type(e).__name__}，已重试3次。"
4.5 主循环：把上面拼起来
def run_agent(user_query, max_steps=8):
memory = AgentMemory()
memory.add("user", user_query)
for step in range(max_steps):
messages = [{"role": "system", "content": SYSTEM_PROMPT}] + memory.build_context()
resp = llm.invoke(messages) # ChatOpenAI, temperature=0
content = resp.content
# 解析：约定模型输出 {"thought":..., "action":..., "args":...} 或 {"final":...}
try:
parsed = json.loads(content)
except json.JSONDecodeError:
memory.add("system", "输出不是合法JSON，请只输出JSON。")
continue
if "final" in parsed:
return parsed["final"]
observation = execute_action(parsed, memory)
memory.add("assistant", content)
memory.add("system", f"Observation: {observation}")
return "达到最大步数，任务未完成。"
SYSTEM_PROMPT里我明确写了输出格式和"最多调3次工具"的软约束，配合硬上限8步，基本不会失控。
五、踩坑与优化：三个真实教训
坑1：temperature=0也会输出非JSON。 GPT-4o-mini在长上下文下偶尔加markdown代码块。解决：prompt里加"不要用``json包裹"，解析前先strip("json\n")。
坑2：记忆压缩用LLM反而更贵。 一开始每轮都调LLM做摘要，单任务多花0.001美元。改成截断+只在超过10条时压缩，成本降回可接受。
坑3：工具描述写太细反而误导。 我给get_weather写了200字描述，模型开始纠结"要不要先查城市坐标"。精简到一句话后，工具调用准确率从78%升到91%。
六、效果数据
在10类任务（天气查询、邮件草拟、数据计算、多步查询等）各跑10次，共100次：
| 指标 | 裸LLM | 本Agent |
|---|---|---|
| 任务完成率 | 42% | 87% |
| 平均工具调用轮次 | - | 2.8 |
| 平均输入token | 900 | 1800 |
| 单任务成本 | $0.0008 | $0.0029 |
| 死循环发生率 | - | 1% |
成本涨了3.6倍，但完成率翻倍，这个trade-off我认为值。死循环那1%是模型在"查不到城市"时反复重试，靠指纹检测兜住了。
七、总结
手写Agent最大的价值不是省依赖，是每个失败你都知道为什么。工具Schema校验、记忆双层结构、三级错误处理、指纹死循环检测，这四块加起来不到300行，但覆盖了生产环境80%的异常场景。下一步我打算把摘要压缩换成向量检索（用text-embedding-3-small），解决长任务里"早期关键信息被截断"的问题——目前窗口6轮在超过15步的任务上会丢上下文，这是已知缺陷。
代码我整理成了单文件，改个API key就能跑。如果你也在被AutoGPT的"黑盒决策"折磨，建议花半天自己写一遍，比读十篇源码分析都有用。