LangChain Agent进阶：从create_agent到中间件、结构化与流式输出实战-尧图网络科技

1. 项目概述：为什么我们需要深入理解LangChain Agent？

如果你已经用LangChain搭建过基础的RAG问答系统，可能会觉得它像是一个“听话”的流水线工人：你给一个文档，它检索、生成，然后给你答案。但当你需要处理更复杂、更动态的任务时，比如“帮我分析一下上周的销售数据，然后写一份总结报告，最后用中文发一封邮件给团队”，这种线性的流水线就有点力不从心了。这时候，你就需要一个能“思考”、能“决策”、能“使用工具”的智能体，也就是Agent。

LangChain的Agent框架，正是为了解决这类问题而生的。它让大语言模型（LLM）从一个单纯的文本生成器，转变为一个可以自主规划、调用工具、并最终完成复杂目标的“智能代理”。create_agent、中间件、结构化与流式输出，这几个关键词串联起来，勾勒出的正是一个从搭建基础Agent，到增强其可控性与可观测性，再到优化其输出体验的完整进阶路径。这不仅仅是功能的堆砌，更是构建一个健壮、可靠、可投入生产环境的AI应用所必须掌握的核心技能。

我见过不少开发者，在初步接触Agent时，往往只停留在调用一个initialize_agent函数，然后就被各种AgentExecutor的报错搞得焦头烂额。或者，Agent跑起来了，但就像一个黑盒，你不知道它内部到底做了哪些决策，为什么失败，性能瓶颈在哪里。这次，我们就抛开那些浅尝辄止的教程，直接从create_agent这个更底层的构建方式入手，深入它的“五脏六腑”，聊聊如何通过中间件给它装上“监控探头”和“控制阀门”，再让它的输出变得既规整又流畅。

2. 核心思路拆解：从“黑盒执行”到“透明可控”的Agent工程

在早期版本的LangChain或者一些简单示例中，我们常用initialize_agent来快速创建一个Agent。它很方便，但封装得太深，当你想定制工具调用逻辑、添加日志、或处理复杂错误时，就会感到束手束脚。而create_agent提供了更细粒度的控制，它让你从组装“发动机”开始，而不仅仅是拿到一辆封装好的“车”。

我的核心思路是分三步走，构建一个工业级的Agent系统：

内核构建：使用create_agent函数，清晰地定义工具（Tools）、提示词（Prompt）和语言模型（LLM），组装出Agent的核心推理逻辑。这一步的关键是理解Agent的“思考-行动-观察”循环是如何被编码的。
增强与监控：引入中间件（Middleware）层。这不是LangChain某个特定的类，而是一种设计模式。我们可以在Agent执行的生命周期关键节点（如工具调用前、调用后、最终输出前）注入自定义逻辑，用于日志记录、权限校验、耗时统计、错误重试等。这相当于给黑盒打开了观察窗和应急通道。
输出优化：处理输出结果。一方面，我们利用Pydantic等库强制Agent进行结构化输出，确保返回的数据是规整的、可编程处理的JSON对象，而不是一段自由文本。另一方面，我们实现流式输出，让Agent的“思考过程”（如“我在调用搜索工具…”、“我找到了答案…”）能够实时地、逐字或分块地返回给前端，极大地提升用户体验。

这个思路的转变，是从“能让Agent跑起来”到“能让Agent跑得好、看得清、控得住”的关键。下面，我们就进入实战环节。

3. 核心细节解析：create_agent、中间件、结构化与流式输出

3.1 深入create_agent：组装你的第一个定制化Agent

create_agent函数（在LangChain中通常通过create_react_agent等具体函数体现）的核心是让你明确地提供三个部分：LLM、工具集和提示词模板。我们来看一个比官方示例更贴近实战的写法。

首先，定义工具。工具的本质是一个能被LLM调用的函数。

from langchain.tools import tool from langchain_community.utilities import SerpAPIWrapper import requests # 工具1：搜索工具（需要API Key） @tool def search_web(query: str) -> str: """当需要获取实时信息或事实性答案时，使用此工具搜索网络。""" # 这里简化实现，实际应使用SerpAPI、Google Search等 # 例如：serp = SerpAPIWrapper(serpapi_api_key=os.getenv(“SERPAPI_API_KEY”)) # return serp.run(query) print(f“[工具调用] 搜索网络，查询词：{query}”) # 模拟返回 return f“关于'{query}'的模拟搜索结果：这是一个示例结果。” # 工具2：计算器工具 @tool def calculator(expression: str) -> str: """用于执行数学计算。输入应为字符串形式的数学表达式，如 '3 + 5 * 2'。""" print(f“[工具调用] 计算表达式：{expression}”) try: # 警告：使用eval在生产环境有安全风险，此处仅作演示。 # 生产环境应使用ast.literal_eval或专用数学库。 result = eval(expression) return str(result) except Exception as e: return f“计算错误：{e}” # 工具3：获取天气工具（模拟） @tool def get_weather(city: str) -> str: """获取指定城市的当前天气信息。""" print(f“[工具调用] 获取{city}的天气”) # 模拟API调用 weather_data = { “北京”: “晴，25°C”， “上海”: “多云，28°C”， “深圳”: “阵雨，30°C” } return weather_data.get(city, “抱歉，未找到该城市天气信息。”) tools = [search_web, calculator, get_weather]

接下来，准备提示词。好的提示词是Agent的“任务说明书”。

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder # 构建ReAct风格的提示词模板 agent_prompt = ChatPromptTemplate.from_messages([ (“system”， “你是一个乐于助人的AI助手。你可以使用工具来帮助你回答问题。请严格按照以下格式回应：\n\n思考：你需要先思考当前情况和你需要做什么。\n行动：你要调用的工具名，必须是以下工具之一：[{tool_names}]\n行动输入：调用该工具所需的输入\n观察：工具返回的结果\n...（这个思考/行动/观察循环可以重复多次）\n最终答案：当你足够确定答案时，用‘最终答案：’开头给出最终回复。”), MessagesPlaceholder(variable_name=“chat_history”), (“user”， “{input}”), MessagesPlaceholder(variable_name=“agent_scratchpad”), ])

注意：MessagesPlaceholder中的agent_scratchpad是LangChain Agent执行的关键。它会在运行时被自动填充为Agent与工具交互的历史记录（思考、行动、观察），从而让LLM拥有完整的上下文记忆。

然后，绑定LLM。这里以OpenAI为例。

from langchain_openai import ChatOpenAI llm = ChatOpenAI(model=“gpt-3.5-turbo”， temperature=0, openai_api_key=“your-api-key”)

最后，使用create_react_agent（这是create_agent的一种具体实现）来创建Agent。

from langchain.agents import create_react_agent # 创建Agent agent = create_react_agent(llm=llm, tools=tools, prompt=agent_prompt)

至此，你得到了一个agent对象，它是一个Runnable。但它还不能直接运行，需要被一个AgentExecutor来驱动其循环。这就是create_agent与initialize_agent的一个区别：它更清晰地分离了“策略定义”和“执行循环”。

from langchain.agents import AgentExecutor agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True) # 运行Agent result = agent_executor.invoke({“input”: “北京现在的天气怎么样？如果气温是25度，那么华氏度是多少？”}) print(result[“output”])

实操心得：

handle_parsing_errors=True这个参数至关重要。LLM的输出可能偶尔不符合Agent期望的解析格式（如缺少“行动：”关键字），设置这个参数能让执行器尝试修复或给出友好错误，而不是直接崩溃。
verbose=True可以在控制台打印出详细的思考链，是调试Agent行为的利器。但在生产环境，我们需要更结构化的日志，这就引出了中间件。

3.2 设计中间件：为Agent装上“监控与控制系统”

中间件（Middleware）是一种装饰器模式，它允许我们在不修改核心业务逻辑（Agent的推理和执行）的情况下，在其执行流程的各个节点插入自定义代码。在LangChain中，我们可以通过创建自定义的Runnable或利用RunnableLambda来包装Agent，实现中间件功能。

假设我们需要两个中间件：1. 日志记录中间件；2. 工具调用耗时统计中间件。

from langchain_core.runnables import RunnableLambda from datetime import datetime import time from typing import Any, Dict class AgentLoggerMiddleware: """日志记录中间件""" def __init__(self, agent_name: str = “MyAgent”): self.agent_name = agent_name def log(self, message: str, level: str = “INFO”): timestamp = datetime.now().strftime(“%Y-%m-%d %H:%M:%S”) print(f“[{timestamp}] [{level}] [{self.agent_name}] {message}”) def __call__(self, input_data: Dict[str, Any]) -> Dict[str, Any]: # 在Agent执行前记录 self.log(f“开始处理请求。输入：{input_data.get(‘input’)}”) start_time = time.time() # 这里假设input_data会被传递给下一个Runnable # 实际上，我们需要用RunnableLambda来更精细地控制 return input_data # 更实用的方式：使用RunnableLambda包装Agent的每个步骤 def build_loggable_agent_executor(agent_executor): """包装原始的agent_executor，添加日志和监控""" original_invoke = agent_executor.invoke def logged_invoke(input_data: Dict[str, Any]) -> Dict[str, Any]: logger = AgentLoggerMiddleware() logger.log(f“Agent调用开始。问题：‘{input_data.get(‘input’)}’”) # 监控工具调用 original_tool_call = agent_executor.tools[0]._run if agent_executor.tools else None # 这里需要更复杂的方式来拦截所有工具调用，演示概念 # 一个更彻底的方法是继承AgentExecutor并重写相关方法 try: result = original_invoke(input_data) logger.log(f“Agent调用成功。输出：‘{result.get(‘output’, ‘N/A’)[:100]}...’”) # 截断长输出 return result except Exception as e: logger.log(f“Agent调用失败！错误：{e}”， “ERROR”) raise agent_executor.invoke = logged_invoke return agent_executor # 使用包装后的执行器 logged_executor = build_loggable_agent_executor(agent_executor) result = logged_executor.invoke({“input”: “深圳和上海哪里更热？”})

上面的方法比较“土”，但说明了原理。更优雅的方式是利用LangChain的Runnable协议。我们可以创建一个“可运行链”，将中间件逻辑作为独立的环节插入：

from langchain_core.runnables import RunnablePassthrough def log_input(input_dict: Dict) -> Dict: print(f“[Middleware - Pre] 收到输入： {input_dict}”) return input_dict def log_output(output_dict: Dict) -> Dict: print(f“[Middleware - Post] 产生输出： {output_dict}”) return output_dict # 构建带有中间件的执行链 agent_chain = ( RunnablePassthrough.assign() # 可以在这里预处理输入 | RunnableLambda(log_input) # 前置中间件 | agent_executor # 核心Agent | RunnableLambda(log_output) # 后置中间件 ) # 运行带中间件的链 result = agent_chain.invoke({“input”: “计算一下圆的面积，如果半径是5的话”})

注意事项：

真正的工具调用级别中间件需要更深入的集成，可能需要自定义Tool类或BaseAgent类，在_call或_arun方法中注入逻辑。
中间件的性能开销需要评估，特别是对于高频调用的Agent。
对于复杂的监控（如链路追踪、指标上报），可以考虑集成像LangSmith这样的专业平台，它提供了开箱即用的强大中间件和观测能力。

3.3 实现结构化输出：让Agent返回规整的数据

Agent默认返回的是文本。但在实际业务中，我们往往希望得到结构化的数据，比如一个包含city、temperature、condition字段的JSON对象，以便后续系统处理。LangChain通过与Pydantic模型深度集成，提供了强大的结构化输出能力。

首先，定义你希望输出的数据结构。

from pydantic import BaseModel, Field from typing import List, Optional class WeatherInfo(BaseModel): """天气信息""" city: str = Field(description=“城市名称”) temperature_c: float = Field(description=“摄氏温度”) condition: str = Field(description=“天气状况，如晴、多云、雨等”) feels_like_c: Optional[float] = Field(default=None, description=“体感温度（摄氏度）”) humidity: Optional[int] = Field(default=None, description=“湿度百分比”) class AnalysisResult(BaseModel): """分析结果""" task: str = Field(description=“分析的任务描述”) findings: List[str] = Field(description=“主要发现列表”) conclusion: str = Field(description=“总结性结论”) confidence: float = Field(description=“结论置信度，0到1之间”， ge=0, le=1)

然后，我们可以通过两种主要方式让Agent进行结构化输出。

方式一：在创建Agent时，使用支持结构化输出的LLM（如OpenAI的gpt-3.5-turbo或gpt-4）并绑定Pydantic模型。

from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import PydanticOutputParser # 1. 创建解析器 parser = PydanticOutputParser(pydantic_object=WeatherInfo) # 2. 构建提示词，将格式指令融入 prompt_template = ChatPromptTemplate.from_messages([ (“system”， “你是一个天气信息提取助手。请从用户的问题中提取天气信息。\n{format_instructions}”), (“user”， “{query}”) ]) # 将解析器的格式说明注入提示词 prompt = prompt_template.partial(format_instructions=parser.get_format_instructions()) # 3. 创建链 llm = ChatOpenAI(model=“gpt-3.5-turbo”， temperature=0) structured_chain = prompt | llm | parser # 4. 运行 query = “今天北京天气晴朗，温度大概25度，体感有点热，可能有28度，湿度是60%。” result = structured_chain.invoke({“query”: query}) print(type(result)) # <class ‘__main__.WeatherInfo’> print(f“城市：{result.city}， 温度：{result.temperature_c}°C， 天气：{result.condition}”) # 输出：城市：北京， 温度：25.0°C， 天气：晴朗

方式二：创建支持结构化输出的自定义工具或Agent。这对于需要多步推理和工具调用的复杂结构化任务更有效。我们可以定义一个“结构化输出工具”，其内部使用一个支持函数调用的LLM。

from langchain.tools import BaseTool, Tool from langchain_core.pydantic_v1 import BaseModel, Field class StructuredQueryInput(BaseModel): question: str = Field(description=“需要分析的复杂问题”) class StructuredAnalysisTool(BaseTool): name = “structured_analysis_tool” description = “对复杂问题进行深度分析，并返回结构化的分析结果。” args_schema = StructuredQueryInput def _run(self, question: str) -> str: # 在这个工具内部，我们使用一个支持结构化输出的LLM链 analysis_parser = PydanticOutputParser(pydantic_object=AnalysisResult) analysis_prompt = ChatPromptTemplate.from_messages([ (“system”， “你是一个资深分析师。请仔细分析以下问题，并给出结构化的分析报告。\n{format_instructions}”), (“user”， “{question}”) ]) analysis_chain = analysis_prompt.partial(format_instructions=analysis_parser.get_format_instructions()) | llm | analysis_parser try: analysis: AnalysisResult = analysis_chain.invoke({“question”: question}) # 将Pydantic对象转为字典，再转为字符串（因为工具返回类型是str） return analysis.json() except Exception as e: return f“分析失败：{e}” def _arun(self, question: str): raise NotImplementedError(“异步执行未实现”) # 将这个工具加入你的Agent工具列表 structured_tool = StructuredAnalysisTool() tools.append(structured_tool) # 现在你的Agent在需要深度分析时，可以调用这个工具并获取JSON字符串结果。 # 你可以在Agent的最终答案中解析这个JSON，或者设计提示词让LLM直接理解并总结它。

实操心得：

结构化输出极大地提升了后端程序处理Agent结果的便利性和可靠性。
Pydantic的字段描述（description）非常重要，LLM会依赖这些描述来理解每个字段的含义。
如果结构化输出失败（LLM返回的格式不符合要求），PydanticOutputParser会抛出异常。要做好错误处理，例如使用OutputFixingParser或RetryOutputParser来自动修复。

3.4 实现流式输出：让用户看到Agent的“思考过程”

流式输出（Streaming）对于需要长时间运行或希望提供实时反馈的Agent应用至关重要。它允许服务器一边生成结果，一边通过网络流（如Server-Sent Events）推送给客户端，实现打字机效果。

LangChain的Runnable协议原生支持流式处理。关键在于使用astream或stream方法，并处理中间步骤的AIMessage和ToolMessage。

首先，我们需要一个能生成流式响应的链。对于简单的链，这很直接：

# 一个简单的非Agent链的流式输出示例 from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI simple_prompt = ChatPromptTemplate.from_template(“用100字介绍一下{topic}”) simple_chain = simple_prompt | ChatOpenAI(model=“gpt-3.5-turbo”， streaming=True) # 注意 streaming=True print(“开始流式输出：”) for chunk in simple_chain.stream({“topic”: “人工智能”}): if hasattr(chunk, ‘content’): print(chunk.content, end=“”， flush=True) # 逐块打印 print(“\n输出结束。”)

对于包含Agent的复杂链，流式输出需要处理更多的中间事件。AgentExecutor本身也支持流式。

# 使用AgentExecutor的stream方法 agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=False, handle_parsing_errors=True) async def run_agent_stream(query: str): """异步流式执行Agent""" full_output = “” print(f“问： {query}”) print(“答： ”， end=“”， flush=True) async for event in agent_executor.astream_events({“input”: query}, version=“v1”): kind = event[“event”] # 监听LLM生成内容的事件 if kind == “on_chat_model_stream”: content = event[“data”][“chunk”].content if content: # 过滤空内容 print(content, end=“”， flush=True) full_output += content # 你也可以监听工具调用开始/结束等事件，实现更丰富的流式提示 # elif kind == “on_tool_start”: # tool_name = event[“name”] # print(f“\n[正在使用工具：{tool_name}]...”， flush=True) # elif kind == “on_tool_end”: # print(“[工具使用完成]”， flush=True) print() # 换行 return full_output # 注意：astream_events需要异步环境运行，例如在Jupyter notebook或FastAPI异步端点中。 # import asyncio # asyncio.run(run_agent_stream(“北京和上海现在的天气分别怎么样？”))

在Web API（如FastAPI）中实现流式响应：

from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio from langchain.agents import AgentExecutor app = FastAPI() @app.get(“/chat/stream”) async def chat_stream(query: str): async def event_generator(): agent_executor = get_agent_executor() # 假设这是一个获取已配置Agent的函数 async for event in agent_executor.astream_events({“input”: query}, version=“v1”): kind = event[“event”] if kind == “on_chat_model_stream”: content = event[“data”][“chunk”].content if content: # 以Server-Sent Events格式发送 yield f“data: {content}\n\n” # 可以发送其他事件类型给前端，用于更新UI状态 elif kind == “on_tool_start”: yield f“event: tool_start\ndata: {event[‘name’]}\n\n” elif kind == “on_tool_end”: yield “event: tool_end\ndata: {}\n\n” yield “event: end\ndata: {}\n\n” return StreamingResponse(event_generator(), media_type=“text/event-stream”)

注意事项：

流式输出会增加服务器的连接负担，需要妥善管理连接生命周期。
前端需要能够处理Server-Sent Events (SSE) 或 WebSocket。SSE实现起来更简单，适合文本流。
流式传输中间步骤（如工具调用状态）能极大提升用户体验，让用户感知到Agent正在“工作”，而不是卡住。

4. 实战：构建一个集大成的智能分析助手

现在，让我们把上面所有的知识点串联起来，构建一个名为“SmartAnalyst”的智能体。它能理解复杂问题，调用工具（搜索、计算、获取天气）收集信息，进行结构化分析，并以流式方式返回结果。

步骤1：定义核心组件

# 工具定义（复用之前的search_web, calculator, get_weather） # 提示词定义（使用增强版的ReAct提示词） # LLM初始化（使用gpt-4以获得更好的推理和工具调用能力） advanced_prompt = ChatPromptTemplate.from_messages([ (“system”， “””你是一个顶尖的数据分析师助手（SmartAnalyst）。你的目标是综合运用所有可用工具，精准、完整地回答用户问题。 可用工具：{tool_names} 工具描述：{tool_descriptions} 请严格遵循以下格式： 思考：分析用户问题，决定是否需要以及使用哪个工具。 行动：工具名 行动输入：工具的输入参数 观察：工具返回的结果 ...（重复思考/行动/观察直到你认为信息充足） 最终答案：基于所有观察，给出清晰、结构化的最终答案。如果问题涉及数据，尽量用数字和事实说话。 “””), MessagesPlaceholder(variable_name=“chat_history”), (“user”， “{input}”), MessagesPlaceholder(variable_name=“agent_scratchpad”), ]) llm_gpt4 = ChatOpenAI(model=“gpt-4-turbo-preview”， temperature=0.1, streaming=True)

步骤2：创建带中间件的Agent执行器

from langchain.agents import AgentExecutor, create_react_agent from contextlib import contextmanager import time class TimingMiddleware: """耗时统计中间件""" def __init__(self): self.timings = [] @contextmanager def measure(self, name: str): start = time.time() yield duration = time.time() - start self.timings.append((name, duration)) print(f“[Timing] {name} 耗时：{duration:.2f}秒”) def create_enhanced_agent_executor(llm, tools, prompt): """创建增强的Agent执行器，内置日志和计时""" base_agent = create_react_agent(llm=llm, tools=tools, prompt=prompt) executor = AgentExecutor(agent=base_agent, tools=tools, verbose=False, handle_parsing_errors=True, max_iterations=5) # 包装invoke方法添加中间件逻辑 original_invoke = executor.invoke timing_mw = TimingMiddleware() def enhanced_invoke(inputs: Dict[str, Any]) -> Dict[str, Any]: print(f“[SmartAnalyst] 开始处理查询： ‘{inputs.get(‘input’)}’”) overall_start = time.time() with timing_mw.measure(“总Agent执行时间”): result = original_invoke(inputs) overall_duration = time.time() - overall_start print(f“[SmartAnalyst] 处理完成。总耗时：{overall_duration:.2f}秒”) for name, dur in timing_mw.timings: print(f“ - {name}: {dur:.2f}秒”) timing_mw.timings.clear() return result executor.invoke = enhanced_invoke return executor smart_analyst_executor = create_enhanced_agent_executor(llm_gpt4, tools, advanced_prompt)

步骤3：设计结构化输出包装器

我们让Agent的最终答案强制符合一个AnalysisReport结构。

class AnalysisReport(BaseModel): question: str summary: str key_points: List[str] data_sources: List[str] = Field(description=“使用的工具或数据源，如‘网络搜索’，‘天气API’”) confidence: float structured_llm = llm_gpt4.with_structured_output(AnalysisReport) # OpenAI新版SDK支持 def get_structured_analysis(question: str) -> AnalysisReport: """将Agent的最终文本答案，再用一个LLM转化为结构化报告""" # 第一步：Agent执行获取原始文本答案 raw_result = smart_analyst_executor.invoke({“input”: question}) raw_answer = raw_result[“output”] # 第二步：将原始答案结构化 structure_prompt = ChatPromptTemplate.from_template(“”” 请将以下AI助手的回答，整理成一份结构化的分析报告。 原始回答：{raw_answer} 原始问题：{question} 请严格按照AnalysisReport格式输出。 “””) structure_chain = structure_prompt | structured_llm report = structure_chain.invoke({“raw_answer”: raw_answer, “question”: question}) return report # 使用示例 # report = get_structured_analysis(“对比北京和上海未来一周的天气趋势，并给出出行建议。”) # print(report.json(indent=2))

步骤4：暴露流式API

# 在一个假设的FastAPI应用中 @app.get(“/smart_analyst/stream”) async def smart_analyst_stream(query: str): async def generate(): # 创建新的执行器实例（避免状态共享问题） local_executor = create_enhanced_agent_executor(llm_gpt4, tools, advanced_prompt) full_text = “” async for event in local_executor.astream_events({“input”: query}, version=“v1”): event_type = event[“event”] if event_type == “on_chat_model_stream”: chunk_content = event[“data”][“chunk”].content if chunk_content: full_text += chunk_content yield f“data: {chunk_content}\n\n” # 流式传输内容 elif event_type == “on_tool_start”: tool_name = event[“name”] yield f“event: status\ndata: {{‘msg’: ‘正在使用工具：{tool_name}’， ‘type’: ‘info’}}\n\n” elif event_type == “on_tool_end”: yield f“event: status\ndata: {{‘msg’: ‘工具使用完成’， ‘type’: ‘success’}}\n\n” elif event_type == “on_chain_end” and event[“name”] == “AgentExecutor”: # Agent执行完毕，可以触发后续的结构化处理（可选，异步进行） # 这里可以发送一个事件，通知前端开始生成结构化报告 yield f“event: final\ndata: {{‘full_text’: ‘{full_text}’}}\n\n” return StreamingResponse(generate(), media_type=“text/event-stream”)

5. 常见问题与排查技巧实录

在开发和调试这样一个集成的Agent系统时，我踩过不少坑。这里记录下最常见的问题和解决方法。

问题1：Agent陷入循环或达到最大迭代次数（max_iterations）

现象：Agent不停地调用工具，始终不输出最终答案，直到达到max_iterations限制后报错。
原因：
- 提示词不清晰：没有明确告诉LLM何时应该停止思考，给出最终答案。
- 工具描述不准确：LLM无法正确理解工具的功能，导致错误调用。
- 工具返回结果质量差：工具返回的内容无法让LLM推理出答案，迫使它继续尝试。
排查与解决：
1. 开启verbose=True：这是第一步，查看完整的思考链，看它卡在哪一步。
2. 优化提示词：在system提示中强化“最终答案”的指令。例如：“当你拥有足够的信息来直接、完整地回答用户问题时，你必须输出以‘最终答案：’开头的行，并停止使用工具。”
3. 精简和精确化工具描述：工具的描述（description）要简短、准确，明确输入输出。避免模糊词汇。
4. 改进工具：确保工具返回的信息是干净、相关、格式良好的。如果工具返回错误或无关信息，LLM很容易困惑。
5. 设置合理的max_iterations和early_stopping_method：根据任务复杂度调整。对于简单问答，3-5次迭代足够；复杂任务可能需要10次。early_stopping_method可以设为“generate”，让LLM在觉得可以时提前结束。

问题2：结构化输出解析失败（PydanticValidationError）

现象：使用PydanticOutputParser时，LLM返回的文本无法被解析成定义的Pydantic模型。
原因：LLM的输出格式不符合parser.get_format_instructions()生成的指令。
排查与解决：
1. 打印中间输出：在解析前，先打印出LLM返回的原始文本，看看它到底输出了什么。
2. 强化格式指令：在提示词中，用更醒目的方式（如json ...）包裹格式示例。甚至可以提供一两个完整的示例（Few-shot）。
3. 使用OutputFixingParser：LangChain提供了自动修复的解析器。
```
from langchain.output_parsers import OutputFixingParser from langchain_openai import ChatOpenAI parser = PydanticOutputParser(pydantic_object=WeatherInfo) fixing_parser = OutputFixingParser.from_llm(parser=parser, llm=ChatOpenAI()) # 使用fixing_parser代替原来的parser
```
4. 降级模型或提高温度：有时更复杂的模型（如GPT-4）或稍微提高temperature（如0.2）能更好地遵循指令。但要注意平衡一致性和创造性。

问题3：流式输出中断或不流畅

现象：前端接收到的SSE流突然中断，或者内容是一大块一起送达，没有逐字效果。
原因：
- 网络或服务器超时：长时间运行的Agent可能导致网关或负载均衡器超时。
- 缓冲区问题：服务器或客户端的输出缓冲区未被及时刷新。
- 异步处理错误：在async for循环中发生未处理的异常。
排查与解决：
1. 增加超时设置：在Web服务器（如Nginx、云服务商LB）和客户端增加SSE连接的超时时间（例如设置为300秒）。
2. 确保及时刷新：在服务器端，确保每个yield后立即刷新。在Python的print中，使用flush=True。
3. 完善的错误处理：用try...except包裹async for循环内的代码，确保即使某个工具调用或LLM生成出错，也能yield一个错误事件给前端，而不是让整个流崩溃。
4. 分块大小：LLM的流式响应本身是分块的。如果你发现块太大，可以考虑在服务端进行二次缓冲，按字符或句子分割后再yield，以获得更平滑的“打字机”效果。

问题4：工具调用权限或安全性问题

现象：Agent尝试调用一些它不应该调用或具有潜在危险的工具（如执行任意系统命令）。
原因：提示词或工具描述可能无意中暗示了LLM可以去调用危险操作。
解决：
- 最小权限原则：只给Agent提供完成当前任务所必需的最少工具。
- 输入验证：在每个工具的_run方法内部，对输入参数进行严格的验证和清洗（如检查文件路径、SQL语句、命令白名单）。
- 用户确认中间件：对于高风险操作（如发送邮件、修改数据），可以在中间件中暂停执行，并通过回调（如向管理后台发送通知、等待用户在前端确认）来获得明确授权后再继续。
- 沙盒环境：对于执行代码类的工具，务必在安全的沙盒环境（如Docker容器、专用虚拟机）中运行。

构建一个成熟可用的LangChain Agent系统，远不止是调用一个API。它涉及到提示词工程、工具设计、流程控制、状态管理、错误处理和用户体验等多个层面的考量。从create_agent出发，理解其底层机制，再通过中间件、结构化输出和流式输出这些“增强插件”来完善它，你才能真正驾驭这个强大的框架，打造出既智能又可靠的AI应用。