企业官网建设流程全解析

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( name="Calculator", func=calculate_power, description="用于计算一个数的幂。输入应该是包含底数和指数的字符串，用逗号分隔，例如 '2,3' 表示计算2的3次方。" ) search_tool = DuckDuckGoSearchRun() # 2. 准备大脑：选择LLM llm = ChatOpenAI(model="gpt-4o", temperature=0) # 3. 选择智能体预设（Prompt模板） from langchain import hub prompt = hub.pull("hwchase17/react") # 一个经典的ReAct格式提示词 # 4. 创建智能体 agent = create_react_agent(llm, tools=[math_tool, search_tool], prompt=prompt) # 5. 创建执行器并运行 agent_executor = AgentExecutor(agent=agent, tools=[math_tool, search_tool], verbose=True, handle_parsing_errors=True) 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_errors=True至关重要，它能优雅地处理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( func=structured_calculate_power, name="Structured_Calculator", description="计算幂运算。", args_schema=PowerInput # 使用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。中间件本质上是一个函数或可调用对象，它在执行链的特定位置被调用，可以访问和修改运行时的数据。

主要可以介入的生命周期点包括：

1. on_chain_start/end：整个Agent执行开始/结束时。
2. on_llm_start/end：调用LLM前/后。
3. on_tool_start/end：调用某个工具前/后。
4. on_agent_action：Agent决定采取某个行动（选择工具）时。

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_factory=list, 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(f"Tool: {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）**实时展示出来。这让用户感知到智能体正在“工作”，而不是卡住了。

流式的内容通常包括：

• 思考（Thought）：LLM在决定下一步行动前的推理。
• 行动（Action）：智能体决定调用的工具名称和输入。
• 观察（Observation）：工具返回的结果。
• 最终答案（Final Answer）。

5.2 实现支持流式输出的智能体

在LangChain中，实现流式的关键在于使用支持流式的LLM（如ChatOpenAI的streaming=True模式），并使用AgentExecutor的astream或astream_events方法。

from langchain.agents import AgentExecutor, create_react_agent from langchain_openai import ChatOpenAI import asyncio # 1. 创建支持流式的LLM streaming_llm = ChatOpenAI(model="gpt-4o", temperature=0, streaming=True) # 2. 创建智能体和执行器（同上） agent = create_react_agent(streaming_llm, tools=[structured_math_tool], prompt=prompt) streaming_agent_executor = AgentExecutor(agent=agent, tools=[structured_math_tool], verbose=False, handle_parsing_errors=True) # 3. 使用 astream_events 进行流式调用 async def run_agent_with_stream(): print("开始流式输出：") async for event in streaming_agent_executor.astream_events({"input": "请计算125的立方根是多少？"}, version="v1"): kind = event["event"] if kind == "on_chat_model_stream": # 流式输出LLM生成的内容（思考过程和最终答案） content = event["data"]["chunk"].content if content: print(content, end="", flush=True) # 逐词打印 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(f"Token: {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_errors=True是救命参数，但还不够。有时LLM会陷入“输出无效JSON -> 解析错误 -> 重试 -> 再次输出无效JSON”的循环。

解决方案：

1. 设置最大迭代次数：AgentExecutor(max_iterations=10, early_stopping_method="generate")。max_iterations限制循环次数，early_stopping_method="generate"会在达到限制时强制LLM生成最终答案。
2. 使用更鲁棒的输出解析器：对于StructuredTool，确保你的Pydantic模型定义足够简单明确。复杂嵌套结构更容易导致解析失败。
3. 在中间件中实现熔断：通过中间件记录连续失败次数，达到阈值时中断流程并返回友好错误。

from langchain_core.exceptions import OutputParserException class CircuitBreakerMiddleware: def __init__(self, max_failures=3): 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的测试比传统软件更复杂，因为其行为具有非确定性。

方法：

1. 单元测试工具：确保每个工具函数在各种边界情况下都能正确工作。
2. 集成测试场景：构建一个涵盖常见、边界和异常情况的测试用例集。例如：
   • 简单直接的问题（“2的10次方”）。
   • 需要多步推理的问题（“先算2的8次方，再除以4”）。
   • 工具描述模糊可能导致混淆的问题。
   • 故意包含错误输入的问题。
3. 评估指标：除了最终答案的正确性，还应评估工具调用效率（是否调用了不必要的工具？）和步骤合理性。
4. 使用LangSmith：这是LangChain官方提供的追踪和评估平台。它能自动记录每次Agent运行的完整链式调用、输入输出、耗时和Token使用量，并提供可视化界面进行分析和对比测试，是进行Agent调试和优化的神器。

我个人在部署关键业务的Agent之前，会建立一个包含50-100个测试用例的评估集，每次对模型或提示词做重大修改后都跑一遍，监控正确率和成本指标的变化。没有这种持续的评估，Agent的迭代就像蒙着眼睛走路。