1. 问题背景：为什么我不用现成的 AgentExecutor？
最近在做一个内部知识库问答系统，需要 Agent 能自主决定调用「数据库查询」「文档检索」「计算器」三个工具。一开始我直接用了 langchain.agents.create_react_agent + AgentExecutor，结果遇到两个问题：
- 记忆混乱：
AgentExecutor默认只保留chat_history，但工具调用的中间过程（如查询 SQL 的结果）会被塞进 prompt，导致上下文越来越长，GPT-4o-mini 的 128k 窗口也扛不住，第 5 轮对话后延迟从 1.2s 涨到 3.5s。 - 循环失控：当工具返回异常（比如数据库连接失败），Agent 会反复重试同一个工具，最多我见过连续调用 14 次，直接打爆 API 账单。
所以这次我决定用手写循环控制 + langgraph 状态机来替代 AgentExecutor。核心思路来自 AutoGPT 的任务拆解：每个循环 = 思考 + 工具调用 + 结果观察，但加上硬性约束。
2. 环境与版本
python==3.11.8
langchain==0.3.1
langchain-openai==0.2.5
langgraph==0.2.33
openai==1.51.0
注意：不要用 langchain 0.2.x，因为 langgraph 的 StateGraph 在 0.3 里才支持 add_sequence 和 checkpointer 的简单配置。
3. 方案设计：三个核心模块
我的 Agent 架构分三块，对应三个文件：
agent/
├── tools.py # 工具定义（@tool 装饰器）
├── memory.py # 双记忆：短期（对话）+ 长期（向量检索）
└── controller.py # 主循环：状态机 + 错误处理 + 循环上限
核心设计决策：不用 AgentExecutor，改用 langgraph 的 StateGraph 手动构建循环。原因就一句话——AgentExecutor 把「记忆」和「循环」耦合死了，我想自己控制每次迭代时 prompt 里放什么。
4. 核心实现：手写循环控制
4.1 工具定义（tools.py）
from langchain.tools import tool
from langchain_core.tools import ToolException
import random
@tool("weather_query")
def weather_query(city: str) -> str:
"""查询指定城市的实时天气。输入格式：城市名，如 '北京'。"""
# 模拟真实 API 调用，实际用 requests 调 wttr.in
if city == "故障城市":
raise ToolException("天气服务暂时不可用，请稍后重试")
temp = random.randint(15, 30)
return f"{city} 当前 {temp}°C，多云"
@tool("calculator")
def calculator(expression: str) -> str:
"""计算数学表达式，如 '2**10' 或 'sin(30)'。"""
try:
return str(eval(expression, {"__builtins__": {}}, {"sin": lambda x: __import__("math").sin(x)}))
except Exception as e:
raise ToolException(f"表达式错误: {e}")
# 工具列表注入到 prompt
TOOLS = [weather_query, calculator]
这里有个坑：@tool 装饰器默认会把函数的 docstring 作为工具描述，但如果函数名和参数名有歧义，LLM 会乱传参。我试过 weather_query(location="北京") 这种参数名，GPT-4o-mini 经常忽略 location 直接用位置参数，所以后来统一改成简单名字。
4.2 记忆管理（memory.py）
我实现了一个双记忆系统：
from langchain.memory import ConversationBufferWindowMemory
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
class AgentMemory:
def __init__(self, k=3):
# 短期记忆：最近 3 轮对话，避免 prompt 膨胀
self.short_term = ConversationBufferWindowMemory(k=k, return_messages=True)
# 长期记忆：向量存储，存关键事实
self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
self.long_term = FAISS.from_texts(["初始知识库"], self.embeddings)
def add_observation(self, observation: str):
"""把工具结果异步写入长期记忆（用关键词过滤）"""
if len(observation) > 50: # 只存有意义的观察
self.long_term.add_texts([observation])
def get_relevant_memory(self, query: str, k=2) -> str:
"""从长期记忆中检索相关事实"""
docs = self.long_term.similarity_search(query, k=k)
return "\n".join([d.page_content for d in docs])
关键参数：短期记忆窗口 k=3。我测试过 k=1 时，Agent 会忘记上一步工具结果，导致重复调用；k=5 时，第 10 轮对话后 prompt 超过 4k tokens，延迟明显。k=3 是性价比最高的。
4.3 主循环控制器（controller.py）
这是最核心的部分，用 langgraph 实现状态机：
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
class AgentState(TypedDict):
messages: list
remaining_steps: int # 循环上限
last_error: str | None
# 1. 定义节点：思考 + 行动
def call_llm(state: AgentState) -> AgentState:
"""LLM 决定下一步动作（工具调用 or 最终回答）"""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.1)
# 构造 prompt（简略版，实际用 few-shot 示例）
prompt = f"""你有以下工具可用：{TOOL_DESCRIPTIONS}
当前对话历史：{state['messages'][-4:]} # 只取最近4条，配合记忆管理
请输出你的思考步骤，然后决定调用工具或直接回答。
格式：THOUGHT: \n ACTION: (参数) 或 FINAL_ANSWER:
"""
response = llm.invoke([SystemMessage(content=prompt)] + state["messages"])
# 解析响应，这里用正则提取 ACTION 部分
action = parse_action(response.content)
if action.startswith("FINAL_ANSWER"):
state["messages"].append(HumanMessage(content=response.content))
return {**state, "messages": state["messages"], "remaining_steps": 0} # 结束
else:
state["messages"].append(HumanMessage(content=response.content))
return state
# 2. 定义节点：执行工具
def execute_tool(state: AgentState) -> AgentState:
"""执行工具，捕获错误，更新状态"""
last_msg = state["messages"][-1].content
tool_name, arg = parse_action(last_msg) # 解析出工具名和参数
try:
if tool_name == "weather_query":
result = weather_query(arg)
elif tool_name == "calculator":
result = calculator(arg)
else:
result = f"未知工具: {tool_name}"
except ToolException as e:
result = f"工具错误: {str(e)}"
state["last_error"] = str(e)
state["messages"].append(SystemMessage(content=f"工具结果: {result}"))
state["remaining_steps"] -= 1
return state
# 3. 构建图
def build_agent():
graph = StateGraph(AgentState)
graph.add_node("llm", call_llm)
graph.add_node("tool", execute_tool)
graph.add_edge("llm", "tool", condition=lambda s: s["remaining_steps"] > 0 and not s["messages"][-1].content.startswith("FINAL_ANSWER"))
graph.add_edge("llm", END, condition=lambda s: s["remaining_steps"] 50 的观察才存入长期记忆，短期记忆只保留对话轮次。
**优化：用 `asyncio` 并发调用工具**
实测当同时需要天气和计算器时，串行调用需要 2.5s，并发后降到 1.2s。但注意 `langgraph` 默认是同步的，需要把 `execute_tool` 节点改成 async 函数，并在 `graph.compile()` 后调用 `await graph.ainvoke(state)`。
### 6. 效果数据：12 个真实任务测试
我在以下场景测试了 Agent（每个场景跑 3 次取平均）：
| 任务 | 循环次数 | 工具调用准确率 | 延迟(s) |
|------|---------|---------------|---------|
| 查询北京天气 | 1 | 100% | 1.2 |
| 计算 2**10 + 5 | 1 | 100% | 1.1 |
| 天气+计算混合问题 | 3 | 83% | 2.8 |
| 故障城市（模拟错误） | 4 | 100% (正确重试) | 3.5 |
| 连续 5 轮多轮对话 | 8 | 80% | 5.2 |
**关键结论**：错误处理逻辑让故障场景的准确率从 40% 提升到 100%（因为 Agent 会正确重试）。但多轮对话的准确率下降明显，原因是短期记忆窗口 k=3 导致早期信息被丢弃——这是后续优化方向：用 `langchain` 的 `EntityMemory` 提取实体摘要。
### 7. 总结与完整代码
手写 Agent 的核心价值在于**可控性**。你可以精确控制循环上限、错误处理策略、记忆策略。相比 AutoGPT 的复杂任务拆解，这个轻量实现适合 90% 的业务场景。
**完整代码**已上传至 [GitHub Gist](https://gist.github.com/ 此处省略链接)，共 280 行。运行前需要设置 `OPENAI_API_KEY` 环境变量。
最后说一句：不要迷信框架，`AgentExecutor` 适合快速 demo，生产环境要么用 `langgraph` 手写，要么直接用 `autogen`。我这里展示的实现，性能和灵活性都更好。
**依赖安装**：
```bash
pip install langchain==0.3.1 langchain-openai==0.2.5 langgraph==0.2.33 faiss-cpu
下一步计划：加入 langchain 的 RouterChain 做工具预选择，预计可以把混合任务的循环次数从 3 降到 2。有进展会更新博客。
