LangChain v1.0 核心解析：create_agent、LCEL 与 LangGraph 实战指南

1. 从零理解 LangChain v1.0 的定位与核心变化LangChain 这个项目从 2022 年底开始进入大众视野到 v1.0 版本落地中间经历了至少三次比较大的架构调整。我最早接触它的时候整个库还处于一种“什么都想做”的状态链、代理、记忆、检索、工具调用全部混在一起API 变动频繁今天写的代码下周可能就跑不通了。v1.0 最大的意义在于它终于把职责边界划清楚了LangChain 负责模型抽象、工具定义、消息协议和 Agent 循环LangGraph 负责有状态编排和复杂控制流LCEL 负责声明式的链式组合。这三者各司其职不再互相打架。如果你之前用过旧版本的initialize_agent或者AgentExecutorv1.0 会给你一种“换了个框架”的错觉。实际上核心思路没变只是把原来藏在黑盒里的东西暴露出来了。最典型的就是create_agent这个新入口它取代了以前一堆零散的 agent 构造函数把模型、工具、提示词、中间件这几个关键要素统一到一个函数签名里。这个设计的好处是你不需要再去记create_react_agent、create_openai_tools_agent、create_structured_chat_agent这些名字一个create_agent走天下底层根据你传入的模型能力自动选择最合适的执行策略。那 v1.0 到底解决了什么问题我总结下来是三个痛点。第一旧版 Agent 的调试体验极差出错之后你根本不知道是模型选错了工具还是工具执行抛了异常还是提示词格式不对。v1.0 引入了中间件机制你可以在 Agent 循环的每个关键节点插入钩子把中间状态打出来。第二LCEL 和 Agent 之间的割裂以前用 LCEL 写的链很难直接塞进 Agent 里当工具用现在通过统一的Runnable接口链和工具可以互相转换。第三状态管理混乱旧版靠memory对象多轮对话一长就丢上下文LangGraph 的引入让状态变成了显式的、可持久化的图结构。适合谁来学这套东西我的判断是如果你只是想让模型调用一两个 API 做个 demo其实用原生 SDK 就够了LangChain 反而增加理解成本。但如果你要做的场景涉及多步骤推理、多工具协同、需要人工介入审批、需要断点续跑、需要把整个执行过程可视化追踪那 LangChain v1.0 加 LangGraph 这套组合是目前 Python 生态里比较成熟的选择。特别是做工业智能体、复杂工作流自动化的团队这套东西能帮你省掉大量自己造轮子的时间。1.1 LangChain、LangGraph、LCEL 三者的分工关系很多人搞不清楚这三个东西到底谁管什么我用一个类比来说明。假设你要开一家餐厅LCEL是菜谱它定义了“先切菜、再炒、最后装盘”这种线性流程每一步的输出是下一步的输入语法上用管道符|串起来非常直观。LangChain是厨房里的设备和食材供应商它提供统一的接口去调用不同品牌的灶台各种大模型、冰箱向量库、刀具工具函数。LangGraph是餐厅的运营管理系统它管的是“如果客人不满意就退回重做”“同时开三个灶台做不同菜”“某个菜需要等另一个菜做完才能上”这种带分支、带循环、带并行、带状态的复杂调度。在 v1.0 里这三者的边界非常清晰。你写一个简单的问答链用 LCEL 就够了不需要碰 LangGraph。你要做一个能自己决定调用哪个工具、调用几次、失败了怎么重试的 Agent那就用create_agent它内部已经帮你把 LangGraph 的图搭好了。你要做的是一个多 Agent 协作系统每个 Agent 有自己的职责之间还要传递消息和共享状态那就直接上手 LangGraph用它的StateGraph自己画流程图。注意v1.0 之后官方文档里AgentExecutor已经被标记为 legacy新项目不要再用了。如果你在网上搜到旧教程还在讲AgentExecutor那套代码在 v1.0 里虽然还能跑但很多新特性用不上建议直接看新版文档。1.2 为什么 v1.0 要引入中间件机制中间件这个词在 Java 微服务里很常见比如 Spring Boot 的 Filter、Nacos 的配置监听本质都是在主流程的某个切面上插入自定义逻辑。LangChain v1.0 把这个概念搬过来解决的是 Agent 执行过程中的可观测性和可干预性问题。在没有中间件之前Agent 的一次完整执行是这样的你给它一个任务它内部循环“思考→选工具→执行工具→看结果→再思考”直到得出最终答案。整个过程对你来说是个黑盒你只能看到最后的输出。如果中间某一步工具调用失败了你只能从最终的错误信息里猜是哪一步出了问题。有了中间件之后你可以在这些节点插入钩子before_model模型调用之前你可以修改提示词、注入额外上下文、做输入校验。after_model模型返回之后你可以检查它选的工具是否合法、是否需要强制修正。before_tool工具执行之前你可以做权限检查、参数脱敏、记录审计日志。after_tool工具执行之后你可以对结果做后处理、缓存、异常兜底。这套机制在实际项目里非常有用。比如你做的是一个工业设备控制 Agent工具调用会真实地操作物理设备那你在before_tool里加一道人工审批中间件所有危险操作必须等人工确认后才能执行。再比如你做的是客服 Agent在after_model里加一个敏感词过滤中间件模型生成的回复先过一遍过滤器再发给用户。2. create_agent 的核心参数与实操拆解create_agent是 v1.0 里最核心的入口函数它的签名看起来简单但每个参数背后都有讲究。我先把一个最小可运行的例子摆出来然后逐个拆解。from langchain.agents import create_agent from langchain_openai import ChatOpenAI from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的天气 return f{city}今天晴气温25度 model ChatOpenAI(modelgpt-4o, temperature0) agent create_agent( modelmodel, tools[get_weather], system_prompt你是一个天气助手用中文回答用户问题。, ) result agent.invoke({messages: [{role: user, content: 北京天气怎么样}]}) print(result[messages][-1].content)这段代码跑起来之后Agent 会自动判断需要调用get_weather工具传入city北京拿到结果后再组织语言回复用户。整个过程不需要你手动写循环create_agent内部已经用 LangGraph 搭好了一个标准的 ReAct 图。2.1 model 参数不只是传一个模型对象model参数接受的是BaseChatModel的实例但不同模型的能力差异会直接影响 Agent 的行为。比如 GPT-4o 支持原生 tool callingAgent 会优先用模型自带的工具调用能力而一些开源模型如果不支持 tool callingAgent 会退化成用提示词解析的方式来决定调哪个工具。这里有个坑我踩过temperature 设成 0 并不保证每次选同一个工具。因为工具选择本质上是一个分类任务模型在概率接近的时候仍然会有随机性。如果你需要严格可复现的工具调用除了 temperature0还要在中间件里加一层校验或者在 system_prompt 里把工具选择的规则写死。另外v1.0 支持传入一个模型列表做 fallback。比如主模型用 GPT-4o备用模型用 Claude当主模型调用失败或者超时自动切到备用模型。这个配置在create_agent里通过model参数传一个列表实现底层会按顺序尝试。2.2 tools 参数工具定义的质量决定 Agent 的上限工具是 Agent 的手和脚工具定义写得好不好直接决定 Agent 能不能正确使用它们。我见过太多项目Agent 表现差不是因为模型不行而是因为工具描述写得太烂。一个高质量的工具定义包含三个要素函数名要语义明确get_weather比func1好一万倍模型靠名字做第一轮筛选。docstring 要写清楚用途和参数含义模型会读 docstring 来决定什么时候调用这个工具。如果你的 docstring 只写“查询天气”模型可能不知道要不要传城市参数。参数类型标注要准确用str、int、Literal这些类型模型会据此生成合法的参数。如果参数是枚举值用Literal[北京, 上海, 广州]比str好能大幅降低模型传错参数的概率。from typing import Literal from langchain_core.tools import tool tool def query_order( order_id: str, action: Literal[查询状态, 申请退款, 修改地址] ) - str: 根据订单号执行指定操作。 Args: order_id: 订单编号格式为 ORD 开头的12位字符串 action: 要执行的操作类型只能是查询状态、申请退款、修改地址三者之一 # 实际业务逻辑 return f订单{order_id}的{action}操作已完成这个工具定义里Literal类型让模型只能在三个选项里选避免了它自由发挥传一个“取消订单”这种不存在的操作。docstring 里明确写了订单号的格式模型在生成参数时会尽量遵守。实操心得工具数量超过 10 个之后模型的工具选择准确率会明显下降。这时候有两个优化方向一是用中间件做工具的动态筛选根据用户输入先召回最相关的几个工具再传给模型二是用 LangGraph 做分层路由先让一个轻量模型判断任务类型再路由到对应的工具子集。2.3 system_prompt 参数Agent 的行为准则system_prompt不是随便写一句“你是一个助手”就完事了。它是 Agent 的宪法决定了 Agent 的角色、边界、输出格式和异常处理策略。我通常会把 system_prompt 分成四个部分来写角色定义你是谁你的职责范围是什么。工具使用规则什么情况下必须调用工具什么情况下不能调用工具调用失败后怎么处理。输出格式要求最终回复用什么格式是否需要结构化输出。安全边界哪些操作绝对不能做遇到不确定的情况怎么处理。举个例子一个电商客服 Agent 的 system_prompt 可能是这样的你是一个电商平台的客服助手负责处理用户的订单咨询和售后请求。 工具使用规则 - 用户询问订单状态时必须先调用 query_order 工具查询不能凭记忆回答。 - 用户申请退款时先查询订单状态只有状态为已发货或已完成的订单才能申请退款。 - 如果工具返回错误向用户说明情况并建议稍后重试不要编造结果。 输出格式 - 用简洁的中文回复不超过三句话。 - 涉及金额时保留两位小数。 安全边界 - 不讨论与订单无关的话题。 - 不透露其他用户的订单信息。这种颗粒度的 system_prompt 能显著降低 Agent 的幻觉和越界行为。实测下来同样的模型和工具system_prompt 写得好任务成功率能从 60% 提到 85% 以上。2.4 中间件参数v1.0 最值得花时间研究的部分中间件在create_agent里通过middleware参数传入是一个列表按顺序执行。v1.0 内置了几种常用的中间件也支持自定义。内置中间件里我用得最多的是这几个HumanInTheLoopMiddleware在指定工具调用前暂停等待人工审批。做危险操作时必加。SummarizationMiddleware当对话历史太长时自动摘要压缩避免超出模型上下文窗口。ToolRetryMiddleware工具调用失败时自动重试可以配置重试次数和退避策略。自定义中间件的写法是继承AgentMiddleware基类实现对应的钩子方法。下面是一个记录工具调用日志的中间件示例from langchain.agents.middleware import AgentMiddleware class AuditLogMiddleware(AgentMiddleware): def before_tool(self, state, tool_name, tool_input): print(f[审计] 准备调用工具: {tool_name}, 参数: {tool_input}) return state def after_tool(self, state, tool_name, tool_output): print(f[审计] 工具 {tool_name} 返回: {tool_output}) return state这个中间件看起来简单但在生产环境里非常关键。所有工具调用都有日志留痕出了问题可以回溯。我在一个金融项目里就是靠这个中间件定位到模型在某次调用中传错了参数导致查询了错误的账户。3. LCEL 链式组合的实战用法LCEL 是 LangChain Expression Language 的缩写核心思想是用管道符把多个Runnable串起来前一个的输出作为后一个的输入。它的语法极简但能表达的能力很丰富。3.1 从最简单的链开始理解 LCELfrom langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser prompt ChatPromptTemplate.from_template(用一句话解释{concept}) model ChatOpenAI(modelgpt-4o) parser StrOutputParser() chain prompt | model | parser result chain.invoke({concept: 量子纠缠}) print(result)这个链的执行流程是prompt 接收字典输入渲染成消息列表model 接收消息列表返回 AIMessageparser 接收 AIMessage提取文本内容。每一步都是独立的Runnable可以单独测试也可以自由组合。LCEL 最大的好处是自动支持流式输出、批量处理和异步调用。你不需要改任何代码chain.invoke换成chain.stream就变成流式换成chain.batch就变成批量换成chain.ainvoke就变成异步。这是旧版 Chain 类做不到的。3.2 用 RunnableParallel 做并行分支有时候你需要同一个输入同时走多条链最后把结果合并。比如用户输入一段文本你既要提取关键词又要做情感分析还要生成摘要。这三件事互不依赖可以并行跑。from langchain_core.runnables import RunnableParallel keyword_chain ChatPromptTemplate.from_template(提取关键词{text}) | model | parser sentiment_chain ChatPromptTemplate.from_template(判断情感倾向{text}) | model | parser summary_chain ChatPromptTemplate.from_template(生成摘要{text}) | model | parser parallel_chain RunnableParallel( keywordskeyword_chain, sentimentsentiment_chain, summarysummary_chain, ) result parallel_chain.invoke({text: ...}) # result 是一个字典包含 keywords、sentiment、summary 三个键RunnableParallel会并发执行三条链总耗时取决于最慢的那条而不是三条之和。这在处理长文本分析时能省不少时间。3.3 用 RunnableBranch 做条件路由条件路由的场景很常见根据用户输入的类型走不同的处理链。比如用户输入如果是问题就走问答链如果是投诉就走投诉处理链如果是闲聊就走闲聊链。from langchain_core.runnables import RunnableBranch qa_chain ChatPromptTemplate.from_template(回答问题{input}) | model | parser complaint_chain ChatPromptTemplate.from_template(处理投诉{input}) | model | parser chat_chain ChatPromptTemplate.from_template(闲聊回复{input}) | model | parser router RunnableBranch( (lambda x: 投诉 in x[input], complaint_chain), (lambda x: ? in x[input] or in x[input], qa_chain), chat_chain, # 默认分支 ) result router.invoke({input: 我要投诉订单延迟})RunnableBranch按顺序检查条件第一个为真的分支被执行都不满足就走默认分支。这个机制比在 prompt 里让模型自己判断要可靠得多因为路由逻辑是确定性的代码不受模型随机性影响。注意事项RunnableBranch的条件函数接收的是整个输入字典不是上一步的输出。如果你需要根据上一步的输出做路由要先用RunnablePassthrough.assign把中间结果存到字典里。3.4 LCEL 和 Agent 的互相转换v1.0 里LCEL 链可以直接当工具塞给 Agent 用。比如你有一个用 LCEL 写的翻译链想让它成为 Agent 的一个工具from langchain_core.tools import tool translate_chain ChatPromptTemplate.from_template(把以下内容翻译成英文{text}) | model | parser tool def translate(text: str) - str: 把中文翻译成英文 return translate_chain.invoke({text: text}) agent create_agent(modelmodel, tools[translate])反过来Agent 本身也是一个Runnable可以塞进 LCEL 链里。比如你先用一条链做输入预处理再把处理后的结果交给 Agentpreprocess_chain ChatPromptTemplate.from_template(把用户输入改写成标准查询{input}) | model | parser full_chain preprocess_chain | agent result full_chain.invoke({input: 那个啥帮我看看明天天气})这种组合能力让 LCEL 和 Agent 不再是两个独立的世界而是可以互相嵌套的积木。4. LangGraph 状态编排与复杂控制流LangGraph 是 v1.0 体系里负责“重活”的部分。当你的流程不是简单的线性链而是有分支、有循环、有并行、有状态持久化需求时LCEL 就不够用了得上 LangGraph。4.1 StateGraph 的基本结构LangGraph 的核心是StateGraph它由三部分组成状态定义、节点、边。from langgraph.graph import StateGraph, START, END from typing import TypedDict class AgentState(TypedDict): messages: list next_step: str def node_a(state: AgentState): return {messages: state[messages] [经过节点A]} def node_b(state: AgentState): return {messages: state[messages] [经过节点B]} graph StateGraph(AgentState) graph.add_node(a, node_a) graph.add_node(b, node_b) graph.add_edge(START, a) graph.add_edge(a, b) graph.add_edge(b, END) app graph.compile() result app.invoke({messages: [], next_step: })状态是一个TypedDict每个节点接收当前状态返回一个字典表示状态的更新。LangGraph 会自动把更新合并到全局状态里。默认的合并策略是覆盖但你可以用Annotated指定自定义的 reducer比如operator.add表示追加而不是覆盖。4.2 条件边实现动态路由条件边是 LangGraph 最强大的特性之一。它允许你根据当前状态动态决定下一步走哪个节点。def should_continue(state: AgentState) - str: last_message state[messages][-1] if 需要工具 in last_message: return tool_node return end graph.add_conditional_edges( agent_node, should_continue, { tool_node: tool_node, end: END, } )这个机制就是 Agent 循环的本质模型节点判断是否需要调工具需要就去工具节点不需要就结束。create_agent内部搭的图就是这个结构只不过它帮你封装好了你不需要自己写。但当你需要更复杂的控制流时比如“工具调用失败后先重试两次还失败就走人工审批节点审批通过后继续不通过就结束”这种逻辑用create_agent表达不了必须自己用 LangGraph 画。4.3 状态持久化与断点续跑LangGraph 支持 checkpoint可以把每一步的状态存到数据库里。这意味着一个长时间运行的任务可以在中途暂停过几天再继续跑状态不会丢。from langgraph.checkpoint.sqlite import SqliteSaver memory SqliteSaver.from_conn_string(checkpoints.db) app graph.compile(checkpointermemory) config {configurable: {thread_id: user-123}} result app.invoke({messages: [开始任务]}, config) # 中断后重新调用会从上次的状态继续 result app.invoke({messages: [继续]}, config)这个特性在做需要人工介入的审批流时特别有用。Agent 执行到需要审批的节点把状态存下来等人工在后台系统里点了通过再用同一个thread_id恢复执行。实操心得checkpoint 的存储后端选择要根据场景来。开发阶段用 SQLite 就够了生产环境建议用 PostgreSQL因为并发写入更稳定。如果状态数据量大还要考虑定期清理旧的 checkpoint不然数据库会膨胀得很快。4.4 LangGraph 和 LangChain 的边界怎么划我自己的判断标准是如果流程可以用“输入→处理→输出”的线性方式描述用 LCEL如果需要“根据中间结果决定下一步做什么”用 LangGraph。具体来说以下场景必须用 LangGraph需要循环执行直到满足某个条件比如 Agent 的多轮工具调用。需要并行执行多个分支再合并结果。需要人工审批节点流程要暂停等待。需要状态持久化支持断点续跑。多个 Agent 之间需要协作和消息传递。而以下场景用 LCEL 就够了简单的提示词模板加模型调用。固定的多步骤处理流程步骤之间没有条件分支。需要流式输出或批量处理的场景。5. 常见问题排查与避坑指南这一部分是我在实际项目里踩过的坑和总结出来的排查思路网上大部分教程不会讲这些。5.1 工具调用失败的高频原因问题现象可能原因排查方法解决方案模型不调用工具直接编答案工具描述不清晰或 system_prompt 没要求必须调用打印模型返回的 tool_calls 字段在 system_prompt 里明确“必须先调用工具查询”工具参数格式错误参数类型标注不准确检查工具的 args_schema用 Pydantic 模型定义参数加 Literal 约束工具调用后 Agent 不继续工具返回值格式不对检查工具返回的是否为字符串或可序列化对象工具统一返回字符串复杂结构先 json.dumps多个工具之间选择混乱工具功能重叠描述相似打印每次调用的工具名合并功能相似的工具或加中间件做动态筛选5.2 上下文窗口超限的处理策略多轮对话一长消息列表就会超出模型的上下文窗口。v1.0 提供了几种处理方式SummarizationMiddleware自动把旧消息摘要成一段话保留最近几条原始消息。配置简单效果稳定。手动裁剪在中间件里根据 token 数裁剪消息列表只保留最近 N 条。向量检索把历史消息存到向量库每次根据当前问题检索最相关的几条塞回去。我一般用 SummarizationMiddleware 打底再配合手动裁剪做兜底。摘要中间件的触发阈值要设得比模型窗口小一些留出空间给工具返回结果。5.3 流式输出在 Agent 场景下的特殊处理Agent 的流式输出比普通链复杂因为中间夹杂着工具调用。你看到的流可能先是模型的一段思考文本然后是一个工具调用请求然后工具执行结果然后又是模型文本。v1.0 的astream_events可以让你精确地区分这些事件类型async for event in agent.astream_events(input, versionv2): kind event[event] if kind on_chat_model_stream: print(event[data][chunk].content, end) elif kind on_tool_start: print(f\n[调用工具: {event[name]}]) elif kind on_tool_end: print(f[工具返回: {event[data][output]}])这样你可以在前端把工具调用的过程也展示出来用户能看到 Agent 在做什么体验比只显示最终答案好很多。5.4 模型选择与成本控制Agent 场景下 token 消耗比普通对话大得多因为每次工具调用都要把完整的消息历史重新发给模型。一个五轮工具调用的任务token 消耗可能是普通对话的十倍。控制成本的几个手段用便宜模型做路由先用一个小模型判断任务类型简单任务直接处理复杂任务才转给大模型。工具返回结果精简工具返回的内容越短后续模型调用的 token 越少。不要把整个数据库查询结果塞回去只返回关键字段。设置 max_iterations限制 Agent 的最大循环次数防止它陷入死循环烧 token。缓存重复调用相同的工具调用参数结果可以缓存避免重复执行。注意max_iterations设得太小会导致复杂任务做不完设得太大又浪费 token。我的经验值是 10 到 15 之间具体看任务复杂度调整。5.5 调试 Agent 的实用技巧Agent 出问题的时候最有效的调试手段是把完整的消息历史打出来。v1.0 的返回结果里messages字段包含了每一轮的 HumanMessage、AIMessage、ToolMessage按顺序看一遍就能定位问题。result agent.invoke({messages: [{role: user, content: ...}]}) for msg in result[messages]: print(f[{msg.__class__.__name__}] {msg.content[:200]}) if hasattr(msg, tool_calls) and msg.tool_calls: print(f 工具调用: {msg.tool_calls})另一个技巧是用 LangSmith 做追踪。它能把整个执行过程可视化每个节点的输入输出、耗时、token 消耗都看得到。虽然要额外配置但在排查复杂问题时能省大量时间。6. 从学习到落地的路径建议学 LangChain v1.0 最容易走的弯路是一上来就啃文档把每个 API 都看一遍结果看完还是不知道怎么写项目。我的建议是以项目驱动学习先定一个你想做的场景然后缺什么补什么。如果你是完全新手按这个顺序走先用 LCEL 写一个最简单的问答链理解prompt | model | parser这个模式。加一个工具用create_agent做一个能调用工具的 Agent。给 Agent 加一个中间件观察中间件的执行时机。用 LangGraph 手写一个带条件分支的图理解状态和边的概念。把前面写的东西组合起来做一个完整的小项目。如果你已经有旧版 LangChain 的经验重点看三个地方create_agent的新参数、中间件机制、LangGraph 的状态管理。旧版的AgentExecutor、initialize_agent、ConversationBufferMemory这些都可以忘掉了。最后分享一个我在实际项目里的体会不要为了用框架而用框架。LangChain v1.0 的能力很强但它的抽象层也厚。如果你的场景很简单直接用模型 SDK 可能更直接。框架的价值在于处理复杂性当你的流程复杂到需要状态管理、需要中间件、需要可视化追踪的时候它才真正发挥作用。判断标准很简单如果你发现自己开始手写循环、手写状态机、手写重试逻辑那就是该上 LangGraph 的时候了。