AI Agent 从原型到生产：我踩过的 5 个坑和最终方案

用 LangChain + FastAPI 搭了一个 AI Agent，从能跑到能用，中间差了整整 10 倍的距离。

问题是什么

年初接了个需求：做一个能理解自然语言、调用内部 API 完成操作的 Agent。Demo 阶段用 LangChain + OpenAI 两周就搓出来了，问"帮我查一下上个月销售额"这种问题能正常回答。结果一上生产就崩了——上下文爆炸、API 调用幻觉、Token 费用失控、响应延迟 10s 往上。

从"能跑"到"能用"，我重构了三回。这篇就是最终沉淀下来的方案，外加每个坑的解法。

解决思路

我对比过三条路：

最后选了 LangGraph。它把 Agent 的每一步决策都变成图里的一个节点，我能精确控制什么时候调 LLM、什么时候调工具、什么时候返回。

操作步骤

步骤1：从 AgentExecutor 迁移到 LangGraph

原来的代码：

from langchain.agents import AgentExecutor, create_openai_functions_agent

agent = AgentExecutor(agent=agent, tools=tools, max_iterations=5)
result = agent.invoke({"input": query})

重构为 LangGraph：

from langgraph.graph import StateGraph, END
from typing import TypedDict, List

class AgentState(TypedDict):
    messages: List
    next_action: str

graph = StateGraph(AgentState)

# 定义三个核心节点
graph.add_node("call_llm", call_llm_node)
graph.add_node("call_tool", call_tool_node)
graph.add_node("respond", respond_node)

# 条件边：让 LLM 决定下一步
graph.add_conditional_edges(
    "call_llm",
    decide_next,  # 返回 "call_tool" 或 "respond"
    {"call_tool": "call_tool", "respond": "respond"}
)

graph.set_entry_point("call_llm")
graph.add_edge("call_tool", "call_llm")  # 工具调用完回到 LLM
graph.add_edge("respond", END)

LangGraph 的核心思路是把 Agent 的循环变成一个显式的有向图。每个节点干一件事，边决定下一步去哪。这样你就能在任意节点插入拦截逻辑。

步骤2：给每轮对话加 Token 预算

原版 Agent 每轮对话都可能无限调下去。我加了个 Token 计数器，每次 LLM 调用后累加消耗，超过阈值直接截断：

def call_llm_node(state: AgentState):
    response = llm.invoke(state["messages"])
    
    # 累加 token 消耗
    usage = response.response_metadata["token_usage"]
    state["total_tokens"] = state.get("total_tokens", 0) + usage["total_tokens"]
    
    # 超过预算直接返回
    if state["total_tokens"] > TOKEN_BUDGET:
        return {**state, "next_action": "budget_exceeded"}
    
    return {**state, "messages": state["messages"] + [response]}

在 decide_next 里处理预算超限的分支，返回一条友好提示而不是报错。

实测效果：单次会话 Token 消耗下降 60%。之前有个用户连续问了 8 个"再帮我查一下"，Agent 反复调工具加 LLM 分析，一次会话烧掉 50K Token。加了预算后，到阈值直接给总结，用户也没投诉。

步骤3：给工具调用加校验层

Agent 最让我头疼的问题是幻觉调用——LLM 会自己编造参数去调 API。比如用户问"查一下去年销售额"，它传了个不存在的年份格式，API 返回 404 后它还能接着编。

解决方案：每个工具调用都先过一层 Pydantic 校验：

from pydantic import BaseModel, validator

class SalesQuery(BaseModel):
    start_date: str
    end_date: str
    
    @validator("start_date")
    def date_format(cls, v):
        # 只接受 YYYY-MM-DD 格式
        if not re.match(r"^\d{4}-\d{2}-\d{2}$", v):
            raise ValueError("日期格式必须为 YYYY-MM-DD")
        return v

# 在工具节点加校验
def call_tool_node(state: AgentState):
    tool_call = parse_tool_call(state["messages"][-1])
    
    try:
        validated = SalesQuery(**tool_call["args"])
    except ValidationError as e:
        return {
            **state,
            "messages": state["messages"] + [
                ToolMessage(
                    content=f"参数校验失败: {e}",
                    tool_call_id=tool_call["id"]
                )
            ]
        }
    
    # 校验通过才执行
    result = actual_api_call(validated)
    ...

别信 LLM 输出的参数格式。它输出的 JSON 不保证符合你的 API 契约。调用外部系统前一定加一层校验。

步骤4：用记忆管理解决上下文爆炸

默认 Agent 会把整段对话历史塞进 Prompt，几轮下来 Token 就飞了。我搞了个分层记忆：

class LayeredMemory:
    def __init__(self, max_recent=5, max_window_tokens=2000):
        self.max_recent = max_recent  # 保留最近 N 轮完整对话
        self.max_window_tokens = max_window_tokens  # 压缩窗口阈值
    
    def build_prompt_messages(self, history: List, new_query: str):
        recent = history[-self.max_recent:]  # 最近几轮完整保留
        older = history[:-self.max_recent]   # 更早的做摘要
        
        if older:
            summary = self.summarize(older)  # 用 LLM 压缩成一段话
            system_msg = SystemMessage(
                content=f"以下为历史对话摘要：
{summary}"
            )
            return [system_msg] + recent
        
        return recent

逻辑很简单：最近 N 轮保留完整对话保证上下文连续性，更早的压缩成摘要省 Token，总长度超阈值时触发压缩。

步骤5：加可观测性——不观测就没法优化

最后一步也是最容易被忽略的一步。没日志你根本不知道 Agent 在哪一步出了问题：

import structlog
logger = structlog.get_logger()

def call_llm_node(state: AgentState):
    logger.info("llm.call", 
        round=state.get("round", 0),
        token_used=state.get("total_tokens", 0),
        message_count=len(state["messages"])
    )
    
    try:
        response = llm.invoke(state["messages"])
    except Exception as e:
        logger.error("llm.error", error=str(e))
        raise
    
    logger.info("llm.response",
        finish_reason=response.response_metadata.get("finish_reason"),
        tokens=dict(response.response_metadata["token_usage"])
    )
    
    return {**state, "messages": state["messages"] + [response]}

我还会把每次 Agent 运行的完整轨迹——LLM 调用、工具调用、决策路径——写入数据库，方便回放调试。

结果与总结

这套架构上线跑了 3 个月，跟最初的 AgentExecutor 方案比：

• Token 消耗：下降 65%（主要靠预算控制和记忆管理）
• 工具调用成功率：从 72% 升到 96%（校验层拦截了大部分幻觉参数）
• 平均响应时间：从 8s 降到 2.3s（更少的 LLM 调用轮次）

几个容易忽略的点：

1. 别在系统 Prompt 里写太多规则——LLM 记不住，反而干扰核心指令。规则放校验层和逻辑层。
2. Agent 的错误恢复比预判更重要——与其让 LLM "尽量避免错误"，不如设计好出错后的恢复路径。
3. 用户的耐心 ≈ LLM 的响应速度——超过 5s 用户就会觉得"卡住了"。考虑加流式输出或中间状态提示。

延伸思考

• 流式输出：现在是等全部处理完再返回，下一步改成 Server-Sent Events 逐字输出 LLM 思考过程，让用户感知到"它正在工作"
• 多 Agent 协作：当前单 Agent 处理所有请求，工具超过 15 个时，可以考虑 Router Agent + Specialist Agent 的分层架构
• 缓存策略：对于"查销售额"这类频繁查询，可以考虑对语义相似的 Query 做结果缓存，减少重复计算