1. 问题背景：为什么我不直接用LangChain的AgentExecutor

上个月在给一个量化团队做资讯聚合系统，需求很朴素：每天抓50篇研报，自动提取关键数据点。我一开始图省事直接用create_openai_tools_agent+AgentExecutor，结果生产环境跑了一周，三个问题炸得我头皮发麻：

  • 工具参数幻觉：模型在调用search_news(keyword="美联储 加息")时，凭空给我塞了个date_filter="2024Q3"参数，LangChain的pydantic解析直接抛ValidationError，整个链子崩掉。
  • 记忆污染：Agent每轮思考的历史消息全部塞进Prompt，到第4轮时上下文里全是Thought: I need to search...这种中间推理词，把真正有用的工具结果给挤没了。
  • 死循环：有一次模型连续7次调用get_stock_price，每次都返回同样的数据，它还在那“分析”，token烧了1.8万。

查了一圈，LangChain 0.3的AgentExecutor虽然有max_iterations参数，但它是硬性截断，用户体验极差。AutoGPT那种文件系统记忆又太重，我们只是做个轻量分析。最后决定自己写循环控制，只借LangChain的工具抽象和模型封装。

2. 环境与版本：坑在依赖锁死

先交代环境，这步省了后面全是泪：

python 3.11.8
langchain 0.3.1
langchain-openai 0.2.5
openai 1.52.2
pydantic 2.9.0

注意：LangChain 0.3把langchain.agents里的工具装饰器拆到了langchain.tools，如果你看老教程用from langchain.agents import tool，直接报ModuleNotFoundError。另外langchain-openai必须单独装，它和openai库版本有强绑定，我们试过openai 1.60配langchain-openai 0.2.5会静默丢stream参数。

3. 方案设计：一个裸的ReAct循环

我没用LangChain的AgentExecutor，自己写了80行循环控制。核心结构如下：

while steps  str:
        # 模拟搜索，实际替换为API调用
        results = [
            {"title": f"{keyword}相关新闻{i}", "score": 0.9 - i*0.1}
            for i in range(min(3, days))
        ]
        return json.dumps(results, ensure_ascii=False)

    async def _arun(self, **kwargs):
        return self._run(**kwargs)

这里有个坑：如果你用@tool装饰器，LangChain会自动把函数docstring塞进description，但docstring里换行符会被模型误解。我全部改用BaseTool子类，description纯字符串，省心。

5. 核心实现：循环控制与记忆滑窗

记忆管理是重点。我不用LangChain的ConversationBufferMemory，因为它不区分工具调用历史。自己维护一个deque：

```python
from collections import deque
from typing import List, Dict, Any
import json
from pydantic import BaseModel, ValidationError

class AgentStep(BaseModel):
thought: str
action: str
action_input: Dict[str, Any]
observation: str
success: bool

class SimpleAgent:
def init(self, llm, tools: List[BaseTool], max_steps: int = 6, memory_window: int = 4):
self.llm = llm
self.tools = {t.name: t for t in tools}
self.max_steps = max_steps
self.memory_window = memory_window
self.memory: deque[AgentStep] = deque(maxlen=memory_window)
self.step_count = 0

def _build_messages(self, user_query: str) -> List[Dict[str, str]]:
    # system指令里明确要求JSON输出
    system_prompt = """你是一个研究助手。你有以下工具: {tool_names}.
    每次思考后必须输出JSON，格式为:
    {{"thought": "你的推理", "action": "工具名或FinalAnswer", "action_input": {{}}}}
    如果得到答案，输出FinalAnswer。
    注意：action_input必须严格匹配工具参数，不要添加额外字段。""".format(
        tool_names=", ".join(self.tools.keys())
    )

    messages = [{"role": "system", "content": system_prompt}]
    # 记忆窗口只取最近N轮
    for step in self.memory:
        messages.append({"role": "assistant", "content": json.dumps({
            "thought": step.thought, "action": step.action, "action_input": step.action_input
        }, ensure_ascii=False)})
        messages.append({"role": "user", "content": f"Observation: {step.observation}"})

    messages.append({"role": "user", "content": user_query})
    return messages

def run(self, query: str) -> str:
    while self.step_count 8步） | 12次 | 0次 |

手写循环最大的收益不是性能，而是可控性。你可以随意插入审计日志、中断逻辑、人工确认环节。LangChain的AgentExecutor像一个黑盒，出了问题很难定位是Prompt问题还是工具问题。

最后建议：如果你项目刚起步，直接上LangChain的AgentExecutor没毛病。但一旦出现奇怪的参数幻觉或死循环，别调Prompt了，像我一样花半天时间写个40行循环，你会打开新世界的大门。代码在github.com/xxx/simple-agent，有问题评论区见。