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组装你的第一个定制化Agentcreate_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_keyos.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” temperature0, openai_api_key“your-api-key”)最后使用create_react_agent这是create_agent的一种具体实现来创建Agent。from langchain.agents import create_react_agent # 创建Agent agent create_react_agent(llmllm, toolstools, promptagent_prompt)至此你得到了一个agent对象它是一个Runnable。但它还不能直接运行需要被一个AgentExecutor来驱动其循环。这就是create_agent与initialize_agent的一个区别它更清晰地分离了“策略定义”和“执行循环”。from langchain.agents import AgentExecutor agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) # 运行Agent result agent_executor.invoke({“input”: “北京现在的天气怎么样如果气温是25度那么华氏度是多少”}) print(result[“output”])实操心得handle_parsing_errorsTrue这个参数至关重要。LLM的输出可能偶尔不符合Agent期望的解析格式如缺少“行动”关键字设置这个参数能让执行器尝试修复或给出友好错误而不是直接崩溃。verboseTrue可以在控制台打印出详细的思考链是调试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(defaultNone, description“体感温度摄氏度”) humidity: Optional[int] Field(defaultNone, description“湿度百分比”) class AnalysisResult(BaseModel): 分析结果 task: str Field(description“分析的任务描述”) findings: List[str] Field(description“主要发现列表”) conclusion: str Field(description“总结性结论”) confidence: float Field(description“结论置信度0到1之间” ge0, le1)然后我们可以通过两种主要方式让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_objectWeatherInfo) # 2. 构建提示词将格式指令融入 prompt_template ChatPromptTemplate.from_messages([ (“system” “你是一个天气信息提取助手。请从用户的问题中提取天气信息。\n{format_instructions}”), (“user” “{query}”) ]) # 将解析器的格式说明注入提示词 prompt prompt_template.partial(format_instructionsparser.get_format_instructions()) # 3. 创建链 llm ChatOpenAI(model“gpt-3.5-turbo” temperature0) 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_objectAnalysisResult) analysis_prompt ChatPromptTemplate.from_messages([ (“system” “你是一个资深分析师。请仔细分析以下问题并给出结构化的分析报告。\n{format_instructions}”), (“user” “{question}”) ]) analysis_chain analysis_prompt.partial(format_instructionsanalysis_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” streamingTrue) # 注意 streamingTrue print(“开始流式输出”) for chunk in simple_chain.stream({“topic”: “人工智能”}): if hasattr(chunk, ‘content’): print(chunk.content, end“” flushTrue) # 逐块打印 print(“\n输出结束。”)对于包含Agent的复杂链流式输出需要处理更多的中间事件。AgentExecutor本身也支持流式。# 使用AgentExecutor的stream方法 agent_executor AgentExecutor(agentagent, toolstools, verboseFalse, handle_parsing_errorsTrue) async def run_agent_stream(query: str): 异步流式执行Agent full_output “” print(f“问 {query}”) print(“答 ” end“” flushTrue) 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“” flushTrue) full_output content # 你也可以监听工具调用开始/结束等事件实现更丰富的流式提示 # elif kind “on_tool_start”: # tool_name event[“name”] # print(f“\n[正在使用工具{tool_name}]...” flushTrue) # elif kind “on_tool_end”: # print(“[工具使用完成]” flushTrue) 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” temperature0.1, streamingTrue)步骤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(llmllm, toolstools, promptprompt) executor AgentExecutor(agentbase_agent, toolstools, verboseFalse, handle_parsing_errorsTrue, max_iterations5) # 包装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(indent2))步骤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系统时我踩过不少坑。这里记录下最常见的问题和解决方法。问题1Agent陷入循环或达到最大迭代次数max_iterations现象Agent不停地调用工具始终不输出最终答案直到达到max_iterations限制后报错。原因提示词不清晰没有明确告诉LLM何时应该停止思考给出最终答案。工具描述不准确LLM无法正确理解工具的功能导致错误调用。工具返回结果质量差工具返回的内容无法让LLM推理出答案迫使它继续尝试。排查与解决开启verboseTrue这是第一步查看完整的思考链看它卡在哪一步。优化提示词在system提示中强化“最终答案”的指令。例如“当你拥有足够的信息来直接、完整地回答用户问题时你必须输出以‘最终答案’开头的行并停止使用工具。”精简和精确化工具描述工具的描述description要简短、准确明确输入输出。避免模糊词汇。改进工具确保工具返回的信息是干净、相关、格式良好的。如果工具返回错误或无关信息LLM很容易困惑。设置合理的max_iterations和early_stopping_method根据任务复杂度调整。对于简单问答3-5次迭代足够复杂任务可能需要10次。early_stopping_method可以设为“generate”让LLM在觉得可以时提前结束。问题2结构化输出解析失败PydanticValidationError现象使用PydanticOutputParser时LLM返回的文本无法被解析成定义的Pydantic模型。原因LLM的输出格式不符合parser.get_format_instructions()生成的指令。排查与解决打印中间输出在解析前先打印出LLM返回的原始文本看看它到底输出了什么。强化格式指令在提示词中用更醒目的方式如json ...包裹格式示例。甚至可以提供一两个完整的示例Few-shot。使用OutputFixingParserLangChain提供了自动修复的解析器。from langchain.output_parsers import OutputFixingParser from langchain_openai import ChatOpenAI parser PydanticOutputParser(pydantic_objectWeatherInfo) fixing_parser OutputFixingParser.from_llm(parserparser, llmChatOpenAI()) # 使用fixing_parser代替原来的parser降级模型或提高温度有时更复杂的模型如GPT-4或稍微提高temperature如0.2能更好地遵循指令。但要注意平衡一致性和创造性。问题3流式输出中断或不流畅现象前端接收到的SSE流突然中断或者内容是一大块一起送达没有逐字效果。原因网络或服务器超时长时间运行的Agent可能导致网关或负载均衡器超时。缓冲区问题服务器或客户端的输出缓冲区未被及时刷新。异步处理错误在async for循环中发生未处理的异常。排查与解决增加超时设置在Web服务器如Nginx、云服务商LB和客户端增加SSE连接的超时时间例如设置为300秒。确保及时刷新在服务器端确保每个yield后立即刷新。在Python的print中使用flushTrue。完善的错误处理用try...except包裹async for循环内的代码确保即使某个工具调用或LLM生成出错也能yield一个错误事件给前端而不是让整个流崩溃。分块大小LLM的流式响应本身是分块的。如果你发现块太大可以考虑在服务端进行二次缓冲按字符或句子分割后再yield以获得更平滑的“打字机”效果。问题4工具调用权限或安全性问题现象Agent尝试调用一些它不应该调用或具有潜在危险的工具如执行任意系统命令。原因提示词或工具描述可能无意中暗示了LLM可以去调用危险操作。解决最小权限原则只给Agent提供完成当前任务所必需的最少工具。输入验证在每个工具的_run方法内部对输入参数进行严格的验证和清洗如检查文件路径、SQL语句、命令白名单。用户确认中间件对于高风险操作如发送邮件、修改数据可以在中间件中暂停执行并通过回调如向管理后台发送通知、等待用户在前端确认来获得明确授权后再继续。沙盒环境对于执行代码类的工具务必在安全的沙盒环境如Docker容器、专用虚拟机中运行。构建一个成熟可用的LangChain Agent系统远不止是调用一个API。它涉及到提示词工程、工具设计、流程控制、状态管理、错误处理和用户体验等多个层面的考量。从create_agent出发理解其底层机制再通过中间件、结构化输出和流式输出这些“增强插件”来完善它你才能真正驾驭这个强大的框架打造出既智能又可靠的AI应用。