1. 项目概述从工具链到智能体的范式跃迁如果你已经用LangChain搭建过几个RAG应用可能会发现一个瓶颈整个流程是线性的。用户提问 - 检索文档 - 拼接上下文 - 大模型生成答案。这套流程很稳定但不够“聪明”。它无法根据复杂问题自主决定先查数据库还是先调用计算器也无法在发现答案不完整时主动发起新一轮搜索。这正是Agent要解决的问题。它不是另一个组件而是一种全新的架构思想——让大模型成为调度中心自主调用工具Tools来完成任务。最近社区里“AI Agent”的概念火得不行但很多讨论停留在概念层面。作为一个从LangChain早期版本就跟进的开发者我发现create_agent这个高阶API是理解LangChain Agent精髓的最佳切入点。它封装了智能体运行的核心循环Plan、Action、Observation让我们能更专注于定义“智能体本身该做什么”而不是去重复造轮子处理执行逻辑。这次我们就深入这个API并结合中间件、结构化输出等实战中绕不开的特性把一个基础智能体打磨成真正可靠、可观测、可交互的生产级应用。2. Agent核心架构与create_agent深度解析2.1 智能体的本质基于LLM的推理与执行引擎在LangChain的语境下一个Agent由几个核心部分构成一个大型语言模型LLM作为“大脑”一套工具Tools作为“手脚”以及一个驱动它们协同工作的“代理执行器”Agent Executor。大脑负责理解目标、规划步骤、决定下一步调用哪个工具手脚负责执行具体的、模型不擅长的任务如计算、搜索、查询执行器则负责安全、有序地运行这个“思考-行动-观察”的循环直到任务完成或达到限制。create_agent函数是LangChain 1.x中构建这种智能体的推荐方式。与早期版本中需要手动组装AgentExecutor相比它提供了一种更声明式、更集成的体验。它的核心价值在于你只需要关心两件事给智能体配备什么样的“大脑”LLM和“工具箱”Tools它就能返回一个可以直接运行的智能体对象。2.2create_agentAPI实战构建你的第一个智能体让我们从一个最简单的例子开始感受一下create_agent的便捷性。假设我们要构建一个能回答数学问题和实时信息的智能体。from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain.tools import Tool from langchain_community.tools import DuckDuckGoSearchRun import math # 1. 定义工具赋予智能体“手脚” def calculate_power(base: float, exponent: float) - float: 计算一个数的幂。 return math.pow(base, exponent) math_tool Tool( nameCalculator, funccalculate_power, description用于计算一个数的幂。输入应该是包含底数和指数的字符串用逗号分隔例如 2,3 表示计算2的3次方。 ) search_tool DuckDuckGoSearchRun() # 2. 准备大脑选择LLM llm ChatOpenAI(modelgpt-4o, temperature0) # 3. 选择智能体预设Prompt模板 from langchain import hub prompt hub.pull(hwchase17/react) # 一个经典的ReAct格式提示词 # 4. 创建智能体 agent create_react_agent(llm, tools[math_tool, search_tool], promptprompt) # 5. 创建执行器并运行 agent_executor AgentExecutor(agentagent, tools[math_tool, search_tool], verboseTrue, handle_parsing_errorsTrue) result agent_executor.invoke({input: 3的5次方是多少然后再搜索一下今天火星上的天气新闻。}) print(result[output])这段代码清晰地展示了工作流。create_react_agent内部使用了create_agent它根据react这个预设将LLM、工具列表和提示词模板组合起来生成一个符合ReAct推理框架的智能体。AgentExecutor则是“驾驶员”负责解析智能体的输出、调用工具、处理错误并控制循环。注意create_react_agent返回的是一个Runnable对象代表了智能体的决策逻辑而不是完整的执行器。必须将其放入AgentExecutor中才能运行。AgentExecutor的关键参数handle_parsing_errorsTrue至关重要它能优雅地处理LLM输出格式不符合预期的情况避免整个流程崩溃。2.3 工具Tools的定义与高级技巧工具是Agent能力的边界。定义好工具是成功的一半。基础定义如上例所示使用Tool类提供name、func和description。description必须清晰准确它是LLM选择工具的唯一依据。处理复杂输入工具函数通常只接收一个字符串参数。对于多参数需要在函数内部解析或使用StructuredTool。from langchain.tools import StructuredTool from pydantic import BaseModel, Field class PowerInput(BaseModel): base: float Field(description底数) exponent: float Field(description指数) def structured_calculate_power(base: float, exponent: float) - float: return math.pow(base, exponent) structured_math_tool StructuredTool.from_function( funcstructured_calculate_power, nameStructured_Calculator, description计算幂运算。, args_schemaPowerInput # 使用Pydantic模型定义输入结构 )使用StructuredTool后LangChain会帮助将LLM的输出解析成结构化的参数再传给工具函数更加鲁棒。工具检索Tool Retrieval当工具数量很多比如超过10个时让LLM从长列表中直接选择效果会变差。此时可以使用create_retriever_tool或结合Toolkit的概念先根据问题语义检索出最相关的几个工具再让LLM做选择这能显著提升复杂Agent的可靠性。3. 中间件Middleware为智能体注入可观测性与控制力如果把Agent执行器看作一个黑盒中间件就是安装在输入、输出以及每次工具调用环节的“探头”和“拦截器”。它是实现日志记录、成本监控、安全检查、流程修改等高级功能的基石。3.1 中间件的工作原理与类型LangChain的AgentExecutor是基于Runnable协议构建的。该协议提供了with_config方法允许我们注入callbacks或middleware。中间件本质上是一个函数或可调用对象它在执行链的特定位置被调用可以访问和修改运行时的数据。主要可以介入的生命周期点包括on_chain_start/end整个Agent执行开始/结束时。on_llm_start/end调用LLM前/后。on_tool_start/end调用某个工具前/后。on_agent_actionAgent决定采取某个行动选择工具时。3.2 实战构建一个日志与耗时监控中间件假设我们需要监控每次工具调用的耗时和输入输出用于性能分析和调试。import time from langchain_core.runnables import RunnableConfig from langchain.callbacks.base import BaseCallbackHandler class ToolMetricsCallbackHandler(BaseCallbackHandler): 自定义回调处理器用于记录工具指标。 def on_tool_start(self, serialized: dict, input_str: str, **kwargs): tool_name serialized.get(name, unknown_tool) self.start_time time.time() print(f[Tool Start] {tool_name} | Input: {input_str[:100]}...) # 打印前100字符 def on_tool_end(self, output: str, **kwargs): elapsed time.time() - self.start_time print(f[Tool End] | Output: {output[:100]}... | Time: {elapsed:.2f}s) # 在实际项目中这里可以将数据发送到监控系统如Prometheus, Datadog # 使用中间件 config RunnableConfig(callbacks[ToolMetricsCallbackHandler()]) result agent_executor.with_config(config).invoke({input: 计算2的16次方})通过继承BaseCallbackHandler并重写特定方法我们就能在关键节点插入自定义逻辑。这是最常用的中间件实现方式。3.3 高级应用利用中间件进行输入/输出过滤与安全审查中间件更强大的地方在于可以修改流程。例如我们可以添加一个安全检查中间件在问题传递给LLM之前过滤掉某些不恰当的词汇。from typing import Any, Dict from langchain_core.runnables import RunnableLambda def safety_filter(input_data: Dict[str, Any]) - Dict[str, Any]: 简单的安全过滤中间件。 user_input input_data.get(input, ) forbidden_words [恶意指令, 敏感词A, 敏感词B] # 示例列表 for word in forbidden_words: if word in user_input: input_data[input] [输入因包含不当内容已被过滤] break return input_data # 将安全过滤器包装成Runnable并组合到执行器之前 safe_agent_chain RunnableLambda(safety_filter) | agent_executor result safe_agent_chain.invoke({input: 这是一个包含恶意指令的测试}) print(result[output]) # LLM将收到被过滤后的输入这里我们创建了一个RunnableLambda来执行过滤函数然后用管道操作符|将其与原有的agent_executor连接起来形成一个新的执行链。这样所有输入都会先经过安全过滤。4. 结构化输出Structured Output让智能体的回答规整可控默认情况下Agent的最终输出是一段自由文本。但在集成到自动化系统如自动生成报告、填充数据库时我们需要机器可读的、结构化的数据。这就是结构化输出的用武之地。4.1 使用Pydantic强制定义输出格式LangChain通过与Pydantic模型深度集成使得让LLM输出结构化数据变得异常简单。核心是使用with_structured_output方法。from langchain_core.pydantic_v1 import BaseModel, Field from typing import List class AgentResponse(BaseModel): 定义智能体最终输出的结构。 final_answer: str Field(description给用户的直接答案) calculation_steps: List[str] Field(description展示推理或计算步骤) confidence: float Field(description对此答案的信心度0到1之间) sources: List[str] Field(default_factorylist, description引用到的信息来源) # 为LLM绑定结构化输出模式 structured_llm llm.with_structured_output(AgentResponse) # 注意我们需要一个能理解并输出此结构的智能体。 # 一种方法是在Prompt中明确要求并使用支持结构化输出的LLM如GPT-4o。 # 更简单的方式是将结构化输出应用于Agent的最终答案生成阶段。4.2 在Agent工作流中集成结构化输出让整个Agent流程输出结构化数据需要一些设计。一个常见模式是让Agent在完成所有工具调用后将其“工作记忆”中间步骤、观察结果整理成一个结构化的总结。from langchain_core.runnables import RunnablePassthrough def format_agent_steps(agent_output: dict) - dict: 将Agent执行器的原始输出格式化成我们需要的结构。 # agent_output 通常包含 input, output, intermediate_steps intermediate_steps agent_output.get(intermediate_steps, []) steps_log [] for action, observation in intermediate_steps: steps_log.append(fTool: {action.tool} | Input: {action.tool_input} | Obs: {observation[:50]}...) # 这里可以调用一个专用的“总结LLM”让它基于原始输出和步骤日志生成结构化答案 # 为简化我们直接构造一个示例 return { final_answer: agent_output[output], calculation_steps: steps_log, confidence: 0.95, # 实际中可根据逻辑判断 sources: [User Query, Calculator Tool] } # 构建一个输出结构化的Agent流程链 structured_agent_chain agent_executor | RunnableLambda(format_agent_steps) | structured_llm # 注意这里的structured_llm需要接收format_agent_steps输出的dict并返回AgentResponse实例。 # 更完整的实现可能需要调整prompt让最后一个LLM调用完成结构化总结。这个示例展示了思路先让Agent以传统方式自由执行然后将它的“行动轨迹”和原始输出作为素材交给一个被约束了输出格式的LLM进行整理和格式化最终得到规整的AgentResponse对象。4.3 结构化输出的优势与陷阱优势集成友好直接输出JSON或Pydantic对象方便被下游系统如API、数据库消费。验证可靠Pydantic会在输出时进行数据验证和类型转换确保数据质量。提示清晰明确的输出模式本身也是对LLM的一种清晰指引能提高回答的规整度。陷阱复杂性增加要求LLM严格遵循复杂模式可能增加其认知负荷有时会导致输出失败或需要更多重试。并非所有模型都支持虽然OpenAI的Chat模型支持良好但一些开源模型在复杂结构化输出上可能表现不稳定。错误处理当LLM无法生成有效结构时需要有备选方案如降级为文本输出并解析。5. 流式输出Streaming Output提升复杂任务的用户体验当Agent执行一个需要调用多次工具、耗时较长的任务时让用户盯着空白屏幕等待几十秒是灾难性的。流式输出允许我们将Agent的“思考过程”实时地、逐字逐句或分块地返回给前端极大地提升用户体验。5.1 理解Agent的流式输出内容Agent的流式输出不仅仅是最终答案的流式传输更重要的是将其**推理过程Reasoning和工具调用Actions**实时展示出来。这让用户感知到智能体正在“工作”而不是卡住了。流式的内容通常包括思考ThoughtLLM在决定下一步行动前的推理。行动Action智能体决定调用的工具名称和输入。观察Observation工具返回的结果。最终答案Final Answer。5.2 实现支持流式输出的智能体在LangChain中实现流式的关键在于使用支持流式的LLM如ChatOpenAI的streamingTrue模式并使用AgentExecutor的astream或astream_events方法。from langchain.agents import AgentExecutor, create_react_agent from langchain_openai import ChatOpenAI import asyncio # 1. 创建支持流式的LLM streaming_llm ChatOpenAI(modelgpt-4o, temperature0, streamingTrue) # 2. 创建智能体和执行器同上 agent create_react_agent(streaming_llm, tools[structured_math_tool], promptprompt) streaming_agent_executor AgentExecutor(agentagent, tools[structured_math_tool], verboseFalse, handle_parsing_errorsTrue) # 3. 使用 astream_events 进行流式调用 async def run_agent_with_stream(): print(开始流式输出) async for event in streaming_agent_executor.astream_events({input: 请计算125的立方根是多少}, versionv1): kind event[event] if kind on_chat_model_stream: # 流式输出LLM生成的内容思考过程和最终答案 content event[data][chunk].content if content: print(content, end, flushTrue) # 逐词打印 elif kind on_tool_start: # 工具开始调用 print(f\n\n[调用工具] {event[name]} 输入: {event[data].get(input)}) elif kind on_tool_end: # 工具调用结束输出结果 output event[data].get(output) print(f\n[工具结果] {output}) # 运行异步函数 await run_agent_with_stream()astream_events提供了极其精细的事件流你可以从中筛选出需要展示给用户的部分。对于前端集成通常你会将on_chat_model_stream事件中的内容块chunk通过WebSocket发送到浏览器。5.3 流式输出中的中间件与状态管理在流式场景下中间件同样可以工作。例如你可以创建一个中间件来计量每个工具调用在流式过程中的耗时或者实时将流式内容推送到消息队列。class StreamingLoggerMiddleware: def __init__(self): self.logs [] async def on_llm_new_token(self, token: str, **kwargs): # 可以在这里将token实时写入日志系统或数据库 self.logs.append(fToken: {token}) # 在实际应用中这里可能是 websocket.send_text(token) # 通过回调注入 config RunnableConfig(callbacks[StreamingLoggerMiddleware()]) async for event in streaming_agent_executor.with_config(config).astream_events(...): ...实操心得流式输出对调试非常有帮助。将astream_events的事件日志保存下来你能完整地复盘Agent的整个决策轨迹对于分析为什么智能体做出了错误决策比如选错了工具至关重要。在生产环境中建议将关键事件的流特别是工具调用输入输出持久化下来用于后续的故障排查和效果分析。6. 避坑指南与性能优化实战将基础Agent投入生产环境会遇到许多在教程中不会提及的问题。以下是我在多个项目中总结出的核心经验。6.1 工具描述Description是成败关键LLM完全依靠工具的description来选择工具。模糊、冗长或错误的描述是导致Agent行为异常的首要原因。错误示例description“一个用于计算的工具”太模糊LLM不知道它能算什么。优秀示例description“当需要计算幂运算一个数的多少次方时使用此工具。输入应为两个用逗号分隔的数字如‘2,3’表示计算2的3次方。”技巧使用关键词在描述中包含可能的问题表述如“平方”、“立方”、“次方”、“power”。明确输入格式像写API文档一样严格定义输入格式。说明使用场景和限制例如“仅适用于整数和浮点数的幂运算不支持复数。”6.2 处理解析错误与无限循环handle_parsing_errorsTrue是救命参数但还不够。有时LLM会陷入“输出无效JSON - 解析错误 - 重试 - 再次输出无效JSON”的循环。解决方案设置最大迭代次数AgentExecutor(max_iterations10, early_stopping_methodgenerate)。max_iterations限制循环次数early_stopping_methodgenerate会在达到限制时强制LLM生成最终答案。使用更鲁棒的输出解析器对于StructuredTool确保你的Pydantic模型定义足够简单明确。复杂嵌套结构更容易导致解析失败。在中间件中实现熔断通过中间件记录连续失败次数达到阈值时中断流程并返回友好错误。from langchain_core.exceptions import OutputParserException class CircuitBreakerMiddleware: def __init__(self, max_failures3): self.failure_count 0 self.max_failures max_failures def on_agent_action(self, action, **kwargs): # 每次成功行动后重置计数器 self.failure_count 0 def on_tool_error(self, error, **kwargs): if isinstance(error, OutputParserException): self.failure_count 1 if self.failure_count self.max_failures: # 触发熔断可以抛出一个特殊异常或返回预设结果 raise ValueError(解析失败次数过多已触发熔断。请检查工具描述或用户输入。)6.3 优化性能与成本Agent的每次“思考”LLM调用和“行动”工具调用都有开销。策略工具设计工具函数本身要高效。涉及网络请求的工具如搜索要设置超时并考虑缓存结果。提示词优化清晰、简洁的提示词能减少不必要的Token消耗。在create_react_agent中你可以拉取不同的提示词模板如hwchase17/react-chat进行对比测试。选择性流式如果前端不需要完整的思考过程可以只流式传输最终答案这能减少后端处理事件流的开销。使用更便宜的模型进行规划一种高级模式是使用快速、廉价的模型如GPT-3.5-turbo进行任务规划和工具选择只在需要生成最终答案时调用强大且昂贵的模型如GPT-4。这需要对Agent执行流程进行更底层的定制。6.4 测试与评估Agent的测试比传统软件更复杂因为其行为具有非确定性。方法单元测试工具确保每个工具函数在各种边界情况下都能正确工作。集成测试场景构建一个涵盖常见、边界和异常情况的测试用例集。例如简单直接的问题“2的10次方”。需要多步推理的问题“先算2的8次方再除以4”。工具描述模糊可能导致混淆的问题。故意包含错误输入的问题。评估指标除了最终答案的正确性还应评估工具调用效率是否调用了不必要的工具和步骤合理性。使用LangSmith这是LangChain官方提供的追踪和评估平台。它能自动记录每次Agent运行的完整链式调用、输入输出、耗时和Token使用量并提供可视化界面进行分析和对比测试是进行Agent调试和优化的神器。我个人在部署关键业务的Agent之前会建立一个包含50-100个测试用例的评估集每次对模型或提示词做重大修改后都跑一遍监控正确率和成本指标的变化。没有这种持续的评估Agent的迭代就像蒙着眼睛走路。