LangChain Agent踩坑记:从Demo到生产的血泪教训

LangChain的Agent在Demo里跑得很顺,但上线后问题一堆:无限循环、工具选错、输出解析失败、响应超时...这篇文章记录我把一个Agent从Demo打磨到生产可用的全过程。

项目背景

2023年3月,我们要做一个"智能数据分析助手",用户用自然语言提问,Agent自动选择工具(查数据库、画图表、做计算)来回答。听起来很美好,现实很骨感。

第一版:天真的Demo

from langchain.agents import Tool, initialize_agent, AgentType
from langchain.chat_models import ChatOpenAI

# 定义工具
tools = [
    Tool(
        name="SQL查询",
        func=run_sql,
        description="用于查询数据库"
    ),
    Tool(
        name="画图",
        func=plot_chart,
        description="用于生成图表"
    ),
    Tool(
        name="计算器",
        func=calculator,
        description="用于数学计算"
    )
]

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# 测试
agent.run("帮我查一下上个月的销售额是多少")

Demo跑起来了!然后我信心满满地上线了...

问题1:无限循环

现象

Agent陷入循环,反复调用同一个工具,直到token耗尽或超时。

# 用户问题: "上个月销售额比上上个月多多少?"

# Agent的思考过程:
# Thought: 我需要查询上个月销售额
# Action: SQL查询
# Observation: 上个月销售额是100万

# Thought: 我还需要查询上上个月销售额  
# Action: SQL查询
# Observation: 上上个月销售额是80万

# Thought: 我需要计算差值
# Action: SQL查询  <- 又选了SQL!
# ... 无限循环

原因

Tool的description写得太模糊,模型分不清什么时候该用SQL,什么时候该用计算器。

解决方案

# 优化后的Tool定义
tools = [
    Tool(
        name="SQL查询",
        func=run_sql,
        description="""用于从数据库查询原始数据。
输入应该是一个具体的数据查询需求,如"查询2023年3月的销售额"。
注意:这个工具只能查询数据,不能做计算。如果需要计算(如求和、比较、百分比),请使用计算器工具。"""
    ),
    Tool(
        name="计算器",
        func=calculator,
        description="""用于数学计算,如加减乘除、百分比、比较大小。
输入应该是一个数学表达式,如"100-80"或"(100-80)/80*100"。
注意:这个工具需要具体的数字,如果数字未知,请先用SQL查询工具获取。"""
    ),
]

# 加上最大迭代次数
agent = initialize_agent(
    tools, llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    max_iterations=5,  # 最多5次工具调用
    early_stopping_method="generate"  # 超过次数时生成最终答案
)

问题2:输出解析失败

现象

# 报错
OutputParserException: Could not parse LLM output: 
"我已经查询到数据了,上个月销售额是100万元。"

原因

ReAct Agent期望LLM输出固定格式(Thought/Action/Observation),但GPT有时候会"忘记"格式,直接输出答案。

解决方案

from langchain.agents import AgentOutputParser
from langchain.schema import AgentAction, AgentFinish
import re

class RobustOutputParser(AgentOutputParser):
    """容错性更强的输出解析器"""
    
    def parse(self, text: str):
        # 尝试标准解析
        action_match = re.search(
            r"Action:\s*(.+?)\nAction Input:\s*(.+)", 
            text, 
            re.DOTALL
        )
        
        if action_match:
            action = action_match.group(1).strip()
            action_input = action_match.group(2).strip()
            return AgentAction(action, action_input, text)
        
        # 检查是否是最终答案
        if "Final Answer:" in text:
            answer = text.split("Final Answer:")[-1].strip()
            return AgentFinish({"output": answer}, text)
        
        # 兜底:如果LLM直接输出答案,当作Final Answer处理
        if not any(keyword in text for keyword in ["Action:", "Thought:"]):
            return AgentFinish({"output": text.strip()}, text)
        
        # 实在解析不了,抛异常
        raise OutputParserException(f"Could not parse: {text}")

# 使用自定义解析器
from langchain.agents import LLMSingleActionAgent

agent = LLMSingleActionAgent(
    llm_chain=llm_chain,
    output_parser=RobustOutputParser(),
    stop=["\nObservation:"],
    allowed_tools=tool_names
)

问题3:工具调用失败处理

现象

SQL执行报错时,Agent不知道怎么办,要么卡死,要么瞎编。

解决方案

def safe_tool_wrapper(func, tool_name: str):
    """包装工具函数,处理异常"""
    def wrapped(input_str: str) -> str:
        try:
            result = func(input_str)
            return str(result)
        except Exception as e:
            error_msg = f"工具'{tool_name}'执行失败: {str(e)}"
            # 返回错误信息,让Agent知道发生了什么
            return f"[错误] {error_msg}。请检查输入或尝试其他方法。"
    return wrapped

# 使用
tools = [
    Tool(
        name="SQL查询",
        func=safe_tool_wrapper(run_sql, "SQL查询"),
        description="..."
    )
]

问题4:响应太慢

现象

一个简单问题,Agent要思考10秒+,用户等不了。

分析

优化方案

# 方案1:用更快的模型做规划
from langchain.chat_models import ChatOpenAI

fast_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
smart_llm = ChatOpenAI(model="gpt-4", temperature=0)

# 用3.5做工具选择(快),用4做最终回答(准)
planning_agent = initialize_agent(tools, fast_llm, ...)

# 方案2:简单问题直接回答,不走Agent
def route_query(query: str):
    """判断是否需要Agent"""
    simple_patterns = ["你好", "谢谢", "什么是"]
    if any(p in query for p in simple_patterns):
        return llm.predict(query)  # 直接回答
    return agent.run(query)  # 走Agent

# 方案3:流式输出,边思考边展示
async def stream_agent_response(query: str):
    """流式输出Agent的思考过程"""
    async for event in agent.astream_events(query, version="v1"):
        kind = event["event"]
        if kind == "on_llm_stream":
            yield event["data"]["chunk"].content
        elif kind == "on_tool_end":
            yield f"\n[工具结果: {event['data']['output'][:100]}...]\n"

问题5:上下文丢失

现象

多轮对话中,Agent会忘记之前说过的话。

# 用户: 查一下北京的销售额
# Agent: 北京销售额是500万

# 用户: 那上海呢?
# Agent: 请问您想查询什么?  <- 忘了上下文!

解决方案

from langchain.memory import ConversationBufferWindowMemory

# 使用带记忆的Agent
memory = ConversationBufferWindowMemory(
    memory_key="chat_history",
    k=5,  # 保留最近5轮对话
    return_messages=True
)

agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION,
    memory=memory,
    verbose=True
)

# 或者手动管理上下文
class ConversationAgent:
    def __init__(self):
        self.history = []
        self.agent = create_agent()
    
    def chat(self, user_input: str) -> str:
        # 构建带上下文的输入
        context = "\n".join([
            f"用户: {h['user']}\n助手: {h['assistant']}" 
            for h in self.history[-3:]
        ])
        
        full_input = f"对话历史:\n{context}\n\n当前问题: {user_input}"
        
        response = self.agent.run(full_input)
        
        self.history.append({
            "user": user_input,
            "assistant": response
        })
        
        return response

最终架构

from typing import Optional
import asyncio

class ProductionAgent:
    """生产级Agent"""
    
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
        self.tools = self._create_tools()
        self.agent = self._create_agent()
        self.history = []
    
    def _create_tools(self):
        return [
            Tool(
                name="数据查询",
                func=safe_tool_wrapper(self.query_data, "数据查询"),
                description="""查询业务数据。
输入格式: {"metric": "销售额", "dimension": "城市", "time": "上个月"}
只能查询,不能计算。"""
            ),
            Tool(
                name="数据计算",
                func=safe_tool_wrapper(self.calculate, "数据计算"),
                description="""对数字进行计算。
输入格式: 数学表达式,如 "100-80" 或 "(100-80)/80*100"
需要先用数据查询获取数字。"""
            ),
        ]
    
    def _create_agent(self):
        return initialize_agent(
            self.tools,
            self.llm,
            agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION,
            max_iterations=5,
            early_stopping_method="generate",
            handle_parsing_errors=True,  # 自动处理解析错误
            verbose=True
        )
    
    async def chat(self, user_input: str, timeout: int = 30) -> str:
        """带超时的对话"""
        try:
            response = await asyncio.wait_for(
                self._run_agent(user_input),
                timeout=timeout
            )
            return response
        except asyncio.TimeoutError:
            return "抱歉,处理超时,请稍后重试。"
        except Exception as e:
            return f"抱歉,出现错误: {str(e)}"
    
    async def _run_agent(self, user_input: str) -> str:
        # 简单问题不走Agent
        if self._is_simple_query(user_input):
            return await self.llm.apredict(user_input)
        
        # 构建带历史的输入
        full_input = self._build_input_with_history(user_input)
        
        # 执行Agent
        response = await self.agent.arun(full_input)
        
        # 更新历史
        self.history.append({"user": user_input, "assistant": response})
        
        return response

性能对比

经验总结

LangChain的问题

说实话,LangChain在生产环境用起来挺痛苦的:

• 版本更新太快,API经常变
• 抽象层太多,debug困难
• 默认配置不适合生产

后来我们逐渐把核心逻辑迁移到了自己的实现,只保留LangChain的一些工具类。

更新记录:
2023-03-28: 初版发布
2023-08-15: 更新LangChain 0.0.300兼容性
2024-01-10: 补充自定义Agent实现