一、为什么我要手写一个Agent
先说结论：LangChain的AgentExecutor和AutoGPT的规划循环都很好用，但当你需要排查"为什么Agent在第4轮突然不调用工具了""为什么记忆把关键上下文挤掉了"这类问题时，黑盒封装会让你抓狂。
上个月我做一个内部运维助手，用LangChain 0.3.x搭了一版，行为不稳定：同样的输入，有时2轮结束，有时陷入工具调用死循环。翻源码发现是max_iterations默认值和early_stopping_method的组合在作怪，加上记忆模块默认只保留最近对话，工具返回的长文本把历史任务目标挤没了。
于是我决定手写一遍，把工具定义、记忆管理、错误处理、循环控制这四件事的每一行都握在自己手里。最终产物约300行Python，跑在GPT-4o-mini上，反而比封装版更可控。
本文代码基于：Python 3.11.7、openai==1.51.0、pydantic==2.9.2、tiktoken==0.7.0。不依赖LangChain。
二、方案设计：Agent的最小闭环
一个Agent的本质是一个循环：
while 未完成 and 步数 dict:
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
"strict": True,
},
}
def run(self, args: dict) -> str:
# 第一层：参数校验交给 pydantic 或手写检查
required = self.parameters.get("required", [])
missing = [k for k in required if k not in args]
if missing:
return f"[ERROR] 缺少必填参数: {missing}"
try:
result = self.func(**args)
except Exception as e: # 第二层：执行异常
return f"[ERROR] 工具执行失败: {type(e).__name__}: {e}"
text = str(result)
if len(text) > self.max_result_chars: # 第三层：结果截断
text = text[: self.max_result_chars] + "...[truncated]"
return text
注册三个工具：计算器（用ast.literal_eval白名单，别用eval）、当前时间、文件读取（限制在白名单目录）。
import ast, operator, datetime, os
ALLOWED_OPS = {ast.Add: operator.add, ast.Sub: operator.sub,
ast.Mult: operator.mul, ast.Div: operator.truediv}
def safe_eval(node):
if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
return node.value
if isinstance(node, ast.BinOp) and type(node.op) in ALLOWED_OPS:
return ALLOWED_OPS[type(node.op)](safe_eval(node.left), safe_eval(node.right))
raise ValueError("不支持的表达式")
def calculator(expression: str) -> float:
return safe_eval(ast.parse(expression, mode="eval").body)
def now() -> str:
return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def read_file(path: str) -> str:
base = os.path.realpath("./workspace")
real = os.path.realpath(os.path.join(base, path))
if not real.startswith(base):
raise PermissionError("路径越界")
with open(real, encoding="utf-8") as f:
return f.read()
TOOLS = {
"calculator": Tool("calculator", "计算数学表达式，支持+-*/",
{"type": "object", "properties": {"expression": {"type": "string"}},
"required": ["expression"], "additionalProperties": False}, calculator),
"get_time": Tool("get_time", "获取当前系统时间",
{"type": "object", "properties": {}, "additionalProperties": False}, now),
"read_file": Tool("read_file", "读取workspace目录下的文件",
{"type": "object", "properties": {"path": {"type": "string"}},
"required": ["path"], "additionalProperties": False}, read_file),
}
3.2 记忆管理：滑动窗口 + 滚动摘要
纯滑动窗口的问题是任务目标会被挤掉。我的做法：保留system消息 + 最近6条消息 + 一条summary。当历史超过6条时，把最老的2条交给LLM压缩成摘要并合并。
import tiktoken
enc = tiktoken.encoding_for_model("gpt-4o-mini")
class Memory:
def __init__(self, window: int = 6, client=None):
self.window = window
self.summary = ""
self.recent: list[dict] = []
self.client = client
def add(self, msg: dict):
self.recent.append(msg)
if len(self.recent) > self.window:
self._compress()
def _compress(self):
to_fold = self.recent[:2]
self.recent = self.recent[2:]
text = "\n".join(f"{m['role']}: {str(m.get('content'))[:300]}" for m in to_fold)
prompt = f"已有摘要：{self.summary or '无'}\n新增对话：\n{text}\n请合并为不超过150字的摘要，保留任务目标、关键数据和结论。"
resp = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0,
)
self.summary = resp.choices[0].message.content.strip()
def build(self, system_prompt: str) -> list[dict]:
msgs = [{"role": "system", "content": system_prompt}]
if self.summary:
msgs.append({"role": "system", "content": f"[历史摘要] {self.summary}"})
msgs.extend(self.recent)
return msgs
实测：摘要压缩平均消耗约380 tokens/次，相比全量历史（任务后期常达3000+ tokens），单轮成本下降约60%。
3.3 循环控制与错误处理
循环体的核心是三个终止条件：模型返回finish、达到max_steps、检测到重复调用。
def run_agent(task: str, max_steps: int = 8) -> str:
mem = Memory(window=6, client=client)
mem.add({"role": "user", "content": task})
last_sig, repeat = None, 0
for step in range(1, max_steps + 1):
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=mem.build(SYSTEM_PROMPT),
tools=[t.to_openai_schema() for t in TOOLS.values()],
tool_choice="auto",
temperature=0,
parallel_tool_calls=False, # 单工具执行，简化状态管理
)
msg = resp.choices[0].message
if not msg.tool_calls: # 终止条件1：自然结束
mem.add({"role": "assistant", "content": msg.content})
return msg.content
mem.add({"role": "assistant", "content": msg.content,
"tool_calls": [tc.model_dump() for tc in msg.tool_calls]})
for tc in msg.tool_calls:
name = tc.function.name
try:
args = json.loads(tc.function.arguments)
except json.JSONDecodeError as e: # 模型偶尔产出坏 JSON
result = f"[ERROR] 参数不是合法JSON: {e}，请重新生成"
else:
sig = (name, json.dumps(args, sort_keys=True))
repeat = repeat + 1 if sig == last_sig else 0
last_sig = sig
if repeat >= 1: # 终止条件2：重复调用
result = "[ERROR] 检测到重复调用，请换一种方式或直接给出答案"
else:
result = TOOLS[name].run(args) if name in TOOLS else f"[ERROR] 未知工具 {name}"
mem.add({"role": "tool", "tool_call_id": tc.id, "content": result})
return "[终止] 达到最大步数，最后摘要：" + (mem.summary or "无") # 终止条件3
SYSTEM_PROMPT里明确写了"每轮最多一个工具调用""工具报错时先分析原因，最多重试一次"。
四、踩坑与优化
坑1：strict: True下模型不传可选参数。 我最初给read_file的path设了默认值并标为非必填，结果strict模式要求所有字段必须在required里，API直接400。解决办法是全部字段必填，可选逻辑交给工具函数内部处理。
坑2：工具结果污染上下文。 一次read_file读了个12KB的日志，直接把后续轮次挤爆。加了max_result_chars=2000后解决，同时提示模型"结果被截断时可分段读取"。
坑3：重复调用死循环。 模型遇到[ERROR] 路径越界时会原样重试。加了签名比对后，第二次相同调用直接返回"请换方式"，模型随即改用相对路径或放弃，实测死循环率从17%降到2%。
坑4：摘要丢任务目标。 早期摘要prompt没强调"保留任务目标"，压缩后模型忘了原始问题。在prompt里显式要求保留"任务目标、关键数据、结论"三项后修复。
五、效果数据
在50条测试任务（含计算、时间查询、文件读取及组合任务）上对比：
| 指标 | LangChain 0.3 封装版 | 手写版 |
|---|---|---|
| 任务完成率 | 84% | 88% |
| 平均迭代轮数 | 3.4 | 2.7 |
| 单任务平均token | 4200 | 2650 |
| 死循环发生率 | 9% | 2% |
| 错误自恢复成功率 | — | 78% |
token下降主要来自记忆压缩和结果截断。注意样本量小，数据仅供方向参考。
六、总结
手写Agent的收益不在性能，而在可观测性。四个模块里，工具定义决定能力边界，记忆管理决定长任务稳定性，错误处理决定鲁棒性，循环控制决定成本上限。LangChain/AutoGPT适合快速验证，但一旦要调优，理解这套循环的每一行是绕不开的。
下一步我打算把工具执行改成异步并发（parallel_tool_calls=True），并给记忆加一层向量检索做长期记忆。代码已放在本地workspace，约300行，无外部框架依赖。
