1. 问题背景：为什么还要手写Agent？
用LangChain的AgentExecutor或直接上AutoGPT不香吗？说实话，香。但当你需要精细控制每一步的代价、需要自定义错误恢复逻辑、或者想搞懂Agent内部到底发生了什么时，现成框架的“黑盒”特性会让人抓狂。AutoGPT那种无限循环的自主模式，在真实业务场景里就是成本炸弹——我见过它在一个简单任务上跑了27轮对话，烧了2美元API费用。
所以，我决定基于LangChain的底层组件（只取BaseTool和LLMChain），手写一个极简Agent内核。目标：单次任务成本控制在0.05美元以内，工具调用失败能自动重试，且循环有硬性上限。
2. 环境与版本
- Python 3.11.5
- LangChain 0.1.0（注意：0.2.x的
BaseTool接口有变动） - openai 1.12.0
- 模型：
gpt-4-turbo-preview（temperature=0.2，max_tokens=512）
pip install langchain==0.1.0 openai==1.12.0
3. 方案设计：五个核心模块
我的设计参照了AutoGPT的“思考-行动-观察”循环，但做了工程化裁剪：
- 工具定义：继承LangChain的
BaseTool，自定义输入schema - 记忆管理：滑动窗口，保留最近5轮对话，超出则摘要压缩
- 循环控制：最多执行8步，每一步都检查是否达到最终答案
- 错误处理：工具调用抛异常时，将错误信息反馈给LLM，让它修正
- 成本熔断：累计token超过阈值（如5万）立即终止
整体流程图：LLM生成结构化JSON → 解析工具名和参数 → 执行工具 → 记录结果到记忆 → 循环。
4. 核心实现(含代码)
4.1 工具定义协议
每个工具必须声明name、description和args_schema。description写得好不好，直接决定LLM调用准确率。我踩过坑：描述太短（ str:
try:
conn = sqlite3.connect(db_path)
cursor = conn.execute(sql)
rows = cursor.fetchall()[:10]
conn.close()
return str(rows)
except Exception as e:
return f"QUERY_ERROR: {str(e)}"
**4.2 记忆管理与循环控制**
这里最关键的是把“工具返回的错误”也塞进记忆，让LLM在下一轮能自我纠正。我维护一个`messages`列表，格式与OpenAI Chat API一致。
```python
from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage, AIMessage
class SimpleAgent:
def __init__(self, tools, max_steps=8, max_tokens_usage=50000):
self.llm = ChatOpenAI(model="gpt-4-turbo-preview", temperature=0.2)
self.tools = {t.name: t for t in tools}
self.tool_descriptions = "\n".join(
[f"{t.name}: {t.description}" for t in tools])
self.max_steps = max_steps
self.max_tokens_usage = max_tokens_usage
self.messages = [SystemMessage(content=
f"你是一个AI助手，你可以使用以下工具：\n{self.tool_descriptions}\n"
"请严格按JSON格式输出：{\"tool\": \"工具名\", \"args\": {...}} "
"或 {\"answer\": \"最终答案\"}")]
self.total_tokens = 0
def run(self, task: str) -> str:
self.messages.append(HumanMessage(content=task))
for step in range(self.max_steps):
# 1. 调用LLM获取下一步动作
response = self.llm.invoke(self.messages)
self.total_tokens += response.response_metadata.get("token_usage", {}).get("total_tokens", 0)
self.messages.append(AIMessage(content=response.content))
# 2. 解析JSON动作
import json
try:
action = json.loads(response.content)
except json.JSONDecodeError:
# 模型输出不合法，要求它重新输出
self.messages.append(HumanMessage(content="你的输出不是合法JSON，请重新输出。"))
continue
# 3. 执行工具或返回答案
if "answer" in action:
return action["answer"]
if "tool" in action:
tool_name = action["tool"]
tool_args = action.get("args", {})
if tool_name not in self.tools:
self.messages.append(HumanMessage(content=f"未知工具{tool_name}，可选：{list(self.tools.keys())}"))
continue
try:
result = self.tools[tool_name].run(tool_args)
self.messages.append(HumanMessage(content=f"工具返回：{result}"))
except Exception as e:
self.messages.append(HumanMessage(content=f"工具执行失败：{str(e)}，请修正参数重试。"))
# 4. 成本熔断
if self.total_tokens > self.max_tokens_usage:
return "ERROR: Token使用超限，任务终止。"
return "ERROR: 达到最大步数，任务未完成。"
5. 踩坑与优化：三个真实教训
坑1：工具描述里的“SQL”一词让模型放飞自我。它居然尝试DROP TABLE，尽管我schema里写了“只允许SELECT”。优化方案：在_run方法里加白名单校验，检测到非SELECT开头直接返回错误。
坑2：记忆窗口无限增长。进行到第5步时，上下文已经塞了3000多token，费用飙升。优化：只保留最近6条消息，更早的用LLMChain压缩成100字摘要。这里有个细节——摘要必须包含“已完成的操作”和“当前目标”，否则模型会迷失。
坑3：循环内的JSON解析脆弱。模型偶尔输出``json代码块标记，导致json.loads`失败。优化：用正则先提取花括号部分，再解析。
import re
def extract_json(text: str) -> dict:
match = re.search(r'\{.*\}', text, re.DOTALL)
if not match:
raise ValueError("No JSON found")
return json.loads(match.group())
6. 效果数据与对比
在“查询数据库中2024年Q1销售额Top3的客户，并给他们的邮箱发送问候邮件”任务上：
- 基线方案：直接让GPT-4生成SQL并执行，无循环。错误率（执行失败或结果错误）：34%
- 本方案：6步完成（查客户表→查订单表→聚合→查邮箱→发邮件→总结），工具调用共5次，成功4次，1次因SQL语法错误触发重试后成功。错误率：12%
- 成本：单次运行约0.042美元（输入token 12,300，输出token 1,780）
对比AutoGPT（配置了memory和web-search插件），同样任务跑了11步，成本0.18美元，而且有2步是在“思考人生”——它尝试搜索“如何发邮件”，完全跑偏。这说明任务导向的约束循环比自由发挥更可靠。
7. 总结与后续打算
手写Agent并不复杂，核心就是约束：工具描述的约束、记忆长度的约束、循环次数的约束、token成本的约束。AutoGPT给了我们灵感，但LangChain给了我们乐高积木。目前这个内核已经应用到内部的报表自动化工具上，下一步打算加入工具并行调用（类似OpenAI的function calling并行模式），以及基于向量数据库的长期记忆——毕竟，每次任务都从零开始太傻了。
最后说句实话：别迷信框架自带Agent，自己写一遍，你才能对“智能”二字祛魅——它不过是精心设计的循环和错误处理罢了。但正是这些细节，决定了你的Agent是“智能助手”还是“人工智障”。
