LangChain 构建 Agent 的核心步骤

一、 Agent 的核心机制与 LangChain 的代际演进要掌握使用 LangChain 构建 Agent 的步骤首先需要明晰 Agent 的本质拓扑结构以及框架层面的技术选型演进。┌────────────────────────────────────────────────────────────────────────┐ │ Chain vs Agent 架构拓扑对比 │ ├────────────────────────────────────────────────────────────────────────┤ │ 1. 传统 Chain (确定性流水线) │ │ [输入] ──► [步骤 A: 检索] ──► [步骤 B: 摘要] ──► [步骤 C: 翻译] ──► [输出] │ │ - 控制流硬编码无条件分支或仅含简单分支执行路径预先完全确定 │ ├────────────────────────────────────────────────────────────────────────┤ │ 2. Agent 智能体 (自适应动态循环) │ │ ┌────────────────────────────────────────┐ │ │ ▼ │ │ │ [输入] ──► [LLM 推理决策] ──(判断是否需要工具?) │ │ │ │ │ │ │ ├── 是 ──► [执行指定工具 Action] ─┘ (返回环境反馈) │ │ │ │ │ └── 否 ──► [生成最终结果 Answer] ──► [输出终止] │ └────────────────────────────────────────────────────────────────────────┘1.1 从 Chain 到 Agent 的范式转变Chain静态工作流将预定义的组件如 PromptTemplate、LLM、OutputParser按照线性或有向无环图DAG的顺序连接。执行顺序在代码编译期就已经确定大模型仅充当流水线中的“文本转换节点”。Agent动态决策智能体将大模型置于核心控制回路Control Loop中。系统不预设具体的执行步骤而是向模型暴露一组可用的工具Tools由模型根据当前环境反馈自主规划下一步动作并在“思考-执行-观察”ReAct 循环中持续迭代直到满足终止条件。1.2 LangChain 的架构重构为什么全面转向 LangGraph在 LangChain 0.1.x 时代Agent 的构建高度依赖AgentExecutor传统模式缺陷旧版 Agent 主要基于正则匹配解析字符串如强制模型输出Action: xxx与Action Input: yyy。模型一旦少输出一个换行或冒号解析器即崩溃同时循环控制流被封装在黑盒内部开发者难以在中间步骤插入自定义的分支判断、人工审批或异步并发控制。现代架构升级LangChain 在 0.2.x 及 0.3.x 版本中正式废弃了旧版AgentExecutor全面采用原生 Function Calling / Tool Calling配合LangGraph。Agent 被形式化建模为一个显式的有限状态图StateGraph所有节点Node、边Edge与上下文状态State均完全透明且可自由拦截。二、 使用 LangChain 构建 Agent 的五大核心步骤在现代 LangChain 体系下构建一个生产就绪的 Agent 必须遵循以下标准化工程步骤。┌────────────────────────────────────────────────────────────────────────┐ │ LangChain Agent 工业级构建五步法 │ ├────────────────────────────────────────────────────────────────────────┤ │ 步骤一工具设计与 Schema 规范定义 (Tools Definition) │ │ - 基于 Pydantic 严格校验输入参数编写清晰的无歧义 Docstring │ ├────────────────────────────────────────────────────────────────────────┤ │ 步骤二模型选型与原生 Tool Calling 绑定 (Model Tool Binding) │ │ - 使用 .bind_tools() 将工具元数据转换为标准的 JSON Schema │ ├────────────────────────────────────────────────────────────────────────┤ │ 步骤三上下文编排与提示词设计 (Prompt Messages Orchestration) │ │ - 设计固定前缀 System Prompt 并挂载 MessagesPlaceholder │ ├────────────────────────────────────────────────────────────────────────┤ │ 步骤四状态图编排与运行时构建 (Runtime Orchestration / LangGraph) │ │ - 定义 AgentState、模型节点、工具执行节点以及条件分支边 │ ├────────────────────────────────────────────────────────────────────────┤ │ 步骤五状态持久化与人机协同集成 (Memory Human-in-the-Loop) │ │ - 接入 Checkpointer 实现多轮对话记忆对高危工具配置审批断点 │ └────────────────────────────────────────────────────────────────────────┘步骤一工具设计与 Schema 规范定义Tools Definition工具是 Agent 与外部环境交互的唯一触手。大模型本身并不执行工具代码它依靠工具的名称Name、自然语言描述Description和参数模式Input Schema来判定“何时调用”以及“如何传参”。在 LangChain 中定义工具的核心标准是使用tool装饰器并显式绑定 Pydantic 模型进行类型约束名称与描述必须清晰明确描述中需明确阐述该工具的职责边界、输入限制以及返回格式强制入参类型校验使用pydantic.BaseModel定义每个参数的类型Type、默认值Default与字段描述Field Description防止模型传入格式非法的参数区分只读工具与破坏性工具为后续的权限拦截预留元数据标记。from pydantic import BaseModel, Field from langchain_core.tools import tool class DatabaseQueryInput(BaseModel): sql_query: str Field( description用于查询生产指标的只读 SELECT SQL 语句禁止包含 DROP、DELETE、UPDATE 等修改操作 ) limit: int Field( default10, description返回的最大记录行数防止大结果集撑爆上下文窗口 ) tool(args_schemaDatabaseQueryInput) def execute_sql_query(sql_query: str, limit: int 10) - str: 在生产只读从库中执行 SQL 查询并返回 JSON 格式结果。 # 模拟真实执行逻辑 return fExecuted: {sql_query} with limit {limit}步骤二模型选型与原生 Tool Calling 绑定Model Tool Binding传统基于自然语言解析工具调用的方式已被淘汰。现代开发必须选用原生支持 Tool Calling函数调用的基座模型如 GPT-4o、Claude 3.5 Sonnet、DeepSeek 等。在 LangChain 中通过.bind_tools()方法将工具集合无缝挂载至模型实例上from langchain_openai import ChatOpenAI # 1. 实例化支持 Tool Calling 的模型对象 llm ChatOpenAI( modelgpt-4o-mini, temperature0.0 # 生产环境执行任务推荐设为 0.0确保输出确定性 ) # 2. 将定义好的工具列表绑定到模型 tools [execute_sql_query] model_with_tools llm.bind_tools(tools)底层机制bind_tools()会在模型发起网络请求时将 Python 函数的 Docstring 和 Pydantic Schema 自动转化为 OpenAI / Anthropic 规范的标准 JSON Schema 注入 HTTP 请求体中。当模型判定需要调用工具时其返回的响应对象中将包含tool_calls结构体明确列出目标函数名与解析完毕的参数字典。步骤三提示词编排与消息流设计Prompt Messages OrchestrationAgent 的核心记忆是基于消息时序列表Messages List流转的。提示词工程需要解决两个核心要素静态规则约束与动态上下文挂载。System Prompt静态系统指令必须置于消息列表首部。明确指定 Agent 的角色身份、可用工具的调用准则、遇到报错时的自我修正策略以及拒绝回答违规请求的防御规则。MessagesPlaceholder动态消息占位符在提示词中预留一个承载完整对话流的插槽按时序动态追加HumanMessage用户输入、AIMessage模型回复或工具调用意图和ToolMessage工具执行后的物理返回值。from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的企业运维数据分析智能体。你必须优先使用工具查询客观数据严禁凭空捏造指标。), MessagesPlaceholder(variable_namemessages), ])步骤四状态图编排与运行时构建Runtime Orchestration / LangGraph这是构建现代 Agent 的关键步骤。我们需要构建一个具备回环逻辑的计算图Computation Graph。一个典型的 ReAct 状态图包含以下核心组件State全局状态包含一个消息列表messages并在每次节点运行后通过追加Append而非覆盖Overwrite更新数据。Agent Node决策节点接收当前状态下的全部消息调用绑定了工具的模型产出包含文本或tool_calls的回复。Tools Node工具执行节点接收模型输出的tool_calls并行/串行调用对应的 Python 本地函数将执行结果包装为ToolMessage追加回状态中。Conditional Edge条件路由边检查 Agent Node 的最新输出如果消息包含tool_calls路由至 Tools Node如果消息不包含tool_calls模型决定直接回答路由至结束节点END。┌───────────────────────┐ │ Start (用户输入) │ └──────────┬────────────┘ │ ▼ ┌───────────────────────┐ │ Agent Node (LLM) │◄─────────────────┐ └──────────┬────────────┘ │ │ │ [是否触发 tool_calls?] │ │ │ ┌─────────────┴─────────────┐ │ ▼ ▼ │ 【是】 【否】 │ │ │ │ ▼ ▼ │ ┌───────────────────┐ ┌───────────────────┐ │ │ Tools Node (执行) │ │ End (返回结果) │ │ └─────────┬─────────┘ └───────────────────┘ │ │ │ └─────────────────────────────────────────────┘步骤五记忆持久化与人机协同集成Memory Human-in-the-Loop为了让 Agent 能够投入真实生产业务必须跳脱单次运行的限制具备持久化会话记忆与高危操作防线Checkpointer检查点持久化将每一步执行后的状态自动序列化并存储至 SQLite、PostgreSQL 或 Redis 中。Agent 在接收到带有thread_id的请求时能无缝恢复上一次中断的会话现场。Human-in-the-Loop人工干预与审批在状态图中针对特定敏感节点如涉及资金转账、数据库修改、发送公网邮件的工具节点设置断点interrupt_before。计算图执行至该节点时会自动挂起待外部管理员通过 API 提交审批确认后方可唤醒并沿断点继续推进。三、 工业级端到端代码实战基于 LangGraph 现代架构本节提供一份完整的、开箱即用的企业级 Agent 实现代码集成 Pydantic 参数校验、LangGraph 显式状态编排、持久化 Checkpointer 以及高危动作拦截能力。3.1 环境依赖安装pip install langchain-core langchain-openai langgraph pydantic3.2 完整工程源码import os import json from typing import Annotated, Sequence, TypedDict from pydantic import BaseModel, Field from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.checkpoint.memory import MemorySaver from langgraph.prebuilt import ToolNode # 1. 定义强类型工具集 class ServerQueryInput(BaseModel): server_id: str Field(description目标服务器 ID例如 srv-node-01) class RebootServerInput(BaseModel): server_id: str Field(description需要重启的目标服务器 ID) reason: str Field(description执行服务器重启的业务原因描述) tool(args_schemaServerQueryInput) def query_server_status(server_id: str) - str: 查询指定服务器的运行状态与 CPU 负载情况。 # 模拟查询逻辑 mock_data { srv-node-01: {status: RUNNING, cpu_load: 94.2%, memory_load: 88.1%}, srv-node-02: {status: RUNNING, cpu_load: 22.5%, memory_load: 35.0%} } data mock_data.get(server_id, {status: NOT_FOUND}) return json.dumps(data, ensure_asciiFalse) tool(args_schemaRebootServerInput) def reboot_server(server_id: str, reason: str) - str: 【高危动作】重启指定的物理或虚拟服务器。 return json.dumps({ server_id: server_id, action: REBOOT, result: SUCCESS, detail: f服务器已成功触发软重启流程原因: {reason} }, ensure_asciiFalse) tools [query_server_status, reboot_server] # 2. 定义状态结构与模型绑定 class AgentState(TypedDict): # 使用 add_messages 规则新消息会自动单调追加至 messages 列表中 messages: Annotated[Sequence[BaseMessage], add_messages] # 初始化模型并绑定工具 llm ChatOpenAI( modelgpt-4o-mini, temperature0.0, api_keyos.getenv(OPENAI_API_KEY, your-api-key), base_urlos.getenv(OPENAI_BASE_URL, https://api.openai.com/v1) ) model_with_tools llm.bind_tools(tools) # 3. 构建核心节点逻辑 def call_model_node(state: AgentState) - dict: Agent 决策节点调用模型生成思考与动作 system_instruction BaseMessage( typesystem, content你是一个机房运维智能体。遇到问题先查询状态。执行重启操作前必须向用户阐述重启原因。 ) # 拼接固定 System Prompt 与动态上下文 full_messages [system_instruction] list(state[messages]) response model_with_tools.invoke(full_messages) return {messages: [response]} # 工具执行节点自动处理 tool_calls 并执行对应的 Python 函数 tool_node ToolNode(tools) # 4. 条件路由分支 def should_continue_router(state: AgentState) - str: 路由决策判断是否需要执行工具或者直接结束 last_message state[messages][-1] # 若模型给出了工具调用指令转移至工具执行节点 if hasattr(last_message, tool_calls) and len(last_message.tool_calls) 0: return tools # 无工具调用任务完成进入结束状态 return END # 5. 编译状态图与持久化 # 初始化状态图 workflow StateGraph(AgentState) # 添加计算节点 workflow.add_node(agent, call_model_node) workflow.add_node(tools, tool_node) # 编排执行流边 workflow.add_edge(START, agent) workflow.add_conditional_edges( agent, should_continue_router, {tools: tools, END: END} ) # 工具执行完成后必须重新回到 agent 节点进行结果反思与再决策 workflow.add_edge(tools, agent) # 配置内存级检查点存储器生产环境可替换为 PostgresSaver checkpointer MemorySaver() # 编译应用并对 tools 节点注入安全拦截模拟高危动作人工审计 app workflow.compile( checkpointercheckpointer, interrupt_before[tools] # 遇到任何工具调用均自动挂起等待外部审核 ) # 6. 多轮交互与人机协同测试运行 def run_agent_demonstration(): config {configurable: {thread_id: session-ops-001}} user_query 请帮我检查服务器 srv-node-01 的负载。如果 CPU 使用率高于 90%请执行重启。 print(f\n[用户下发指令]: {user_query}) # 阶段 1启动工作流直到命中断点 for event in app.stream({messages: [HumanMessage(contentuser_query)]}, configconfig): print(f节点输出跟踪: {list(event.keys())}) # 检查当前系统挂起状态 current_state app.get_state(config) last_msg current_state.values[messages][-1] if hasattr(last_msg, tool_calls) and len(last_msg.tool_calls) 0: pending_tool last_msg.tool_calls[0] print(f\n 触发安全拦截机制) print(f拟执行工具: {pending_tool[name]}) print(f拟传入参数: {pending_tool[args]}) # 模拟管理员确认同意查询执行 print(外部管理员审计决策: [同意执行]\n) # 阶段 2通过传入 None 恢复执行挂起的工作流 for event in app.stream(None, configconfig): print(f恢复后执行追踪: {list(event.keys())}) # 获取当前完整的最新对话状态 final_state app.get_state(config) final_reply final_state.values[messages][-1].content print(f\n[Agent 最终汇报结果]:\n{final_reply}) if __name__ __main__: run_agent_demonstration()四、 生产级工程避坑指南与性能调优在实验室 Demo 中运行流畅的 Agent部署到高并发生产环境中往往会遭遇严重的可用性挑战。以下为生产级架构的核心防御方案4.1 死循环震荡检测与最大步数硬截断故障现象Agent 陷入“调用工具 ➔ 工具报错 ➔ 尝试修复 ➔ 再次传入错误参数 ➔ 再次报错”的震荡死循环中持续消耗 Token 直至网关超时。防护准则最大步数限制Recursion Limit在 LangGraph 运行时必须硬性设置recursion_limit例如设定为 15 步。一旦超过步数阈值立即抛出异常并降级动作重复检测Action Hash Deduplication记录最近 3 次工具调用的(tool_name, arguments)哈希值。若发现同一个工具以完全相同的参数被连续调用 2 次以上强行向上下文中注入系统警告信息System Override阻断无意义的重试。┌─────────────────────────────────────────────────────────────┐ │ 死循环与状态震荡拦截机制 │ ├─────────────────────────────────────────────────────────────┤ │ Step 1: Agent 发起 Action(query_db, {id: 100}) │ │ Step 2: Tool 返回 Record not found │ │ Step 3: Agent 再次发起 Action(query_db, {id: 100}) │ │ ──► 命中 Action Hash 重复检测拦截器! │ │ ──► 强制插入 System 消息: 已检测到重复查询禁止 │ │ 再调用 query_db请根据现状生成最终结论。 │ └─────────────────────────────────────────────────────────────┘4.2 工具异常捕获机制与自我修复反馈绝大多数开发者在工具函数发生异常时直接让 Python 抛出未捕获的 Exception这会导致整个 Agent 运行时崩溃断开。正确做法工具层必须通过try...except拦截一切潜在错误并将错误转化为结构化的字符串或 JSON 返回给模型。自我修复闭环当模型在ToolMessage中看到明确的错误提示如Error: column user_name does not exist in table users具备高阶推理能力的基座模型会自发根据报错修正 SQL 语法形成自愈闭环。tool def safe_api_caller(param: str) - str: 包含严格异常包裹的工具模板 try: # 业务逻辑 result perform_network_call(param) return json.dumps({status: SUCCESS, data: result}) except Exception as e: # 将报错信息友好暴露给模型引导模型修正参数而非崩溃 return json.dumps({ status: ERROR, error_type: type(e).__name__, message: str(e), suggestion: 请检查输入参数格式并重新构造请求 }, ensure_asciiFalse)4.3 上下文窗口膨胀与消息历史剪枝Context Pruning随着 ReAct 循环轮数的增加消息列表中的ToolMessage往往包含大量的原始数据导致单次请求的 Token 消耗呈二次方爆炸并可能瞬间击穿模型的最大上下文窗口Context Window。方案一返回值精简Payload Reduction在工具内部对返回数据进行预脱敏和压缩严禁直接将未经处理的几十万字符原始 HTML 或庞大 JSON 塞入状态方案二滑动窗口与摘要归并Message Trimming在调用模型之前使用 LangChain 提供的trim_messages函数仅保留最近 N 条工具交互历史对早期的历史信息调用小模型进行文本摘要折叠。4.4 针对 Prompt Caching 的前缀对齐优化现代大模型服务商OpenAI、Anthropic、DeepSeek 等普遍提供了基于前缀匹配的Prompt Caching提示词缓存命中缓存通常能够节省 50% 到 90% 的输入 Token 费用并极大削减首字延迟。缓存命中核心法则前缀保持绝对静态将不变的 System Prompt、工具定义Tool Schemas以及少样本示例Few-Shot严格固定在消息序列的最前端严禁在 System Prompt 中插入动态变量切勿在系统提示词中拼接当前时间: 2026-09-22 10:32:05或动态变化的会话 ID这会导致每次请求的前缀字节发生偏移使全局 Prompt Cache 完全失效。若需引入当前时间应通过只读工具如get_current_time由模型按需主动获取。五、 总结与架构选型指南使用 LangChain 构建工业级 Agent 的过程本质上是将大模型的概率推理能力与确定性的软件状态机进行系统解耦与受控集成的过程。[企业级 Agent 技术选型决策树] │ ┌─────────────────────┴─────────────────────┐ ▼ ▼ 【极简单轮或无状态工具调用】 【复杂业务流程 / 长链路状态协同】 │ │ ▼ ▼ 直接使用厂商原生 API 使用 LangChain LangGraph (OpenAI Tool Calling 原生封装) (显式状态图 Checkpointer) │ │ ├─ 优势: 零框架依赖、调试直观 ├─ 优势: 内置记忆持久化、容灾续跑 └─ 劣势: 缺乏多轮状态持久化规范 └─ 劣势: 存在一定框架学习成本在落地实际业务时不要再使用过时的AgentExecutor基于显式状态图编排的LangGraph是当下的唯一标准规范工具是 Agent 的地基高质量、强类型的 Pydantic Schema 比长篇累牍的 Prompt 指令更能有效约束模型的调用行为安全与可控性高于自主性在任何涉及物理写入与资金交割的关键节点必须引入 Checkpointer 机制并挂载人工审批Human-in-the-Loop防线。掌握这套以状态图为骨架、以原生 Tool Calling 为肌肉、以防御性工程为护城河的构建范式开发者才能跨越 Demo 与生产之间的鸿沟构建出高鲁棒性、可维护、可审计的工业级智能体应用。