ARTICLE DETAIL

建站实战干货

来自一线的建站与推广经验沉淀,每一条都经过真实交付验证。

LangChain从链到代理:LCEL与LangGraph构建可控AI应用

2026/9/8 20:19:08 拓冰建站 浏览量
LangChain从链到代理:LCEL与LangGraph构建可控AI应用 1. 从链到代理LangChain 的核心演进逻辑在聊 LangChain 之前我先说个真实感受这个框架的迭代速度快到很多教程刚发出来就过时了。你如果翻到两年前的 LangChain 入门文章看到的还是LLMChain、SimpleSequentialChain那一套再过半年去看又变成了 LCELLangChain Expression Language和 Runnable现在再打开官方文档首页已经全是 Agent、Tool calling、LangGraph 的状态图了。很多人卡在“我学过的 LangChain 怎么和社区里聊的不一样”这个困惑里本质原因就是没搞懂它背后的演进主线。这条主线其实特别清晰从“链”到“代理”本质是从“把大模型调用编排成固定流程”进化到“让大模型自己决定流程”。链Chain解决的是“我知道步骤是什么只是需要把步骤串起来”的问题代理Agent解决的是“我不确定步骤是什么需要模型根据输入动态决策”的问题。理解了这条线你再看 LangChain 生态里那些让人眼花缭乱的概念基本都能找到位置。这篇文章不打算做那种“复制粘贴官方文档”式的教程。我会从实际开发者的视角拆解 LangChain 生态的三大核心核心一链与 LCEL——理解它的设计哲学为什么 LCEL 成为所有链的底层表达方式。核心二代理与工具调用——从 ReAct 到 Tool calling再到 LangGraph 状态机搞清楚 Agent 的真正玩法。核心三工程化落地——内存管理、可观测性、测试评估这些才是上生产环境时真正要命的事。适合谁来读只要是打算用 LangChain 做正经项目而不是光跑通一个 demo 的开发者这篇文章都值得你花二十分钟认真看看。不管你是刚看完入门教程的新手还是已经在用 LangChain 但总觉得哪里别扭的老手我希望你看完之后能形成一套自己的判断框架什么场景该用 Chain什么场景该上 Agent什么场景干脆别用 LangChain。2. 链时代的关键拼图为什么 LCEL 是新的地基2.1 从 LLMChain 到 LCELLangChain 做对了什么如果你接触过 0.x 版本的 LangChain一定对LLMChain不陌生。它长这样from langchain.llms import OpenAI from langchain.chains import LLMChain from langchain.prompts import PromptTemplate prompt PromptTemplate.from_template(用一句话介绍{topic}) llm OpenAI(temperature0) chain LLMChain(llmllm, promptprompt) result chain.run(topic量子计算)这段代码看起来挺简洁但它在真实项目里有个很尴尬的问题只要你想把多个链串起来或者加个条件分支代码就变成了一堆chain1.run()、chain2.run()的嵌套调用调试起来非常痛苦。另外LLMChain这种类封装把很多东西藏在内部你想拿到中间结果、想流式输出、想做并行调用都得去翻源码找内部属性。LCEL 的出现就是为了解决这些问题。它的核心思路是把一切变成Runnable接口通过管道符|把组件串联起来。一段最简单的 LCEL 长这样from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser prompt ChatPromptTemplate.from_template(用一句话介绍{topic}) llm ChatOpenAI(modelgpt-4o-mini) chain prompt | llm | StrOutputParser() result chain.invoke({topic: 量子计算})注意|这个符号它定义了一个明确的输入输出契约左边组件的输出一定是右边组件的输入。你可以把它想象成水管——每一段管子都有一个标准的接口接上就能流水。这不是语法糖而是一种统一的抽象任何 Runnable 对象都有invoke、stream、batch、ainvoke等方法这就意味着你可以在不改变上层代码的情况下替换任意一个环节。我记得有人问过一个很尖锐的问题“LCEL 和直接用 Python 函数写流程有什么本质区别”我的看法是直接用函数写流程你拿到的是“最终输出”中间过程全丢了但 LCEL 的每个环节都是可观测的你可以在任意位置接上回调Callback拿到每一步的原始输入输出。这对调试和日志采集来说是质的提升。生产环境里你总要知道“用户问的这句话在检索步骤里到底召回哪些文档”LCEL 做到了。2.2 Runnable 家族的实用组合Parallel、Lambda、RunnableBranch搞清楚 LCEL 的基本管道还不够日常开发里你很快就会遇到更复杂的编排需求。我挑三个最常用的 Runnable 组合器配合场景说明一下。第一种RunnableParallel。它适合做“一次输入多路并行处理”的场景。举个例子用户问“帮我分析一下这份代码的复杂度并给出优化建议”你可能希望同时让模型分析复杂度、检查潜在 bug、评估可读性最后汇总。用RunnableParallel就可以让这三路分析同时进行from langchain_core.runnables import RunnableParallel def analyze_complexity(text): return analysis_chain.invoke({code: text}) def analyze_bugs(text): return bug_chain.invoke({code: text}) def analyze_readability(text): return readability_chain.invoke({code: text}) parallel_chain RunnableParallel( complexityanalyze_complexity, bugsanalyze_bugs, readabilityanalyze_readability ) result parallel_chain.invoke(code_content)这种并行不只是“看起来快”在真实场景里还能省 token——因为每路分析可以只用它关心的那部分上下文而不是把整段代码重复塞给同一个模型调用。我见过不少新手在写多步分析时习惯把所有历史结果拼在一起再调一次模型token 很快就爆了还会让模型抓不住重点。RunnableParallel天然就是干这个的。第二种RunnableLambda。它让你把普通 Python 函数包装成 Runnable插进 LCEL 管道里。比如你在管道中间需要对文本做过滤或者调用内部服务查询数据都可以用RunnableLambda包一层。这个操作也解决了“LCEL 和业务代码脱节”的顾虑它不是只能在模型和提示词之间流转而是可以和任意业务逻辑无缝衔接。第三种RunnableBranch也就是条件分支。它解决的问题是不同输入走不同路线。比如电商客服里用户问“退货政策”就走退换货知识库链问“订单状态”就调订单查询工具问“闲聊”就随机应变。用代码来说from langchain_core.runnables import RunnableBranch branch RunnableBranch( (lambda x: 退货 in x[question], return_chain), (lambda x: 订单 in x[question], order_status_chain), default_chain )这个分支逻辑如果写成手写 if-else一两个分支还好分支多了缩进层级和判断条件叠在一起代码可读性会变得非常差。RunnableBranch把这些分支变成了一等公民理解和维护都轻松很多。2.3 链的可靠性代价固定流程的得与失聊完了 LCEL 的好处我也想泼点冷水。链式编排最大的问题在于它要求你在设计阶段就预测好所有情况。一旦真实用户的输入超出了你预设的分支整个链的真值链就会断裂。比如你做了一个“先检索后回答”的链用户突然问一句“你能帮我写首诗吗”——检索步骤召回的文档完全没用后续步骤还是按部就班地生成答案最终的输出就可能牛头不对马嘴。那怎么补救这时就要引出代理Agent了。代理和链的根本区别是链把“决策逻辑”写在代码里代理把“决策逻辑”交给模型让模型根据当前上下文动态决定下一步调用哪个工具、以什么顺序调用、甚至要不要提前终止。这也是为什么近两年 Agent 会成为 LangChain 生态的绝对主角。有一条实用的经验是能用 Chain 解决就不要上 Agent。因为链是确定的好测试好排查代理是概率性的每一步都有不确定性测试和排查成本成倍上升。我见过不少团队一上来就用 Agent 做“万能客服”结果模型频繁选错工具最后又退回 Chain 加分支条件。链和代理不是先进与落后的关系是“确定性”和“灵活性”的权衡。3. 代理实战从 ReAct 到 Tool CallingAgent 到底怎么落地3.1 先弄清楚 ReAct 的原理推理和行动交错的循环要理解 Agent绕不开 ReActReasoning and Acting这个思路。它其实不复杂核心就是让模型在一个循环里反复做两件事推理Reasoning和行动Acting。推理时模型说出“我现在的想法”行动时模型决定“我要调用什么工具、传什么参数”。工具返回结果后模型再基于结果继续推理直到它能给出最终答案。举个例子如果用户问“今天北京和上海哪个城市更适合跑步”ReAct 风格的过程大概是模型思考我需要知道两个城市的天气和空气质量先查北京的天气。行动调用get_weather(city北京)。观察结果北京晴空气质量优。再思考还需要查上海的天气。行动调用get_weather(city上海)。观察结果上海有雨空气质量良。最终回答北京更适合跑步。这个循环看起来并不复杂但它的价值在于把“解题过程”显式地暴露出来。你可以看到模型每一步在做什么这就为调试和干预提供了条件。早期 LangChain 的AgentExecutor就是基于 ReAct 思路实现的但由于它的循环逻辑是封装死的你很难精细控制每一步的行为比如“某个工具返回异常时强制走另一条路”或者“连续三次推理没有进展就主动终止”。3.2 为什么 Tool calling 改变了代理开发的姿势讲到 Agent 的具体实现必须重点说 Tool calling。它不是 LangChain 的发明而是模型能力的一次升级——从 GPT-4 开始主流大模型都支持在 API 层面声明“有哪些工具可用每个工具的 input schema 长什么样”模型会返回一个结构化的工具调用指令比如get_weather(city北京)而不是在文本里自己写一段话描述要调用什么。这有什么区别以前模型说“我需要调用天气查询工具”你还得从自然语言里解析出“工具名”和“参数”现在模型直接返回一个 JSON{name: get_weather, arguments: {city: 北京}}你只要把这个 JSON 解析出来执行就行。这大大降低了代理开发的解析成本和出错概率。在 LangChain 里用工具的方式非常简单。先定义一个带类型注解和 docstring 的函数from pydantic import BaseModel, Field from langchain_core.tools import tool class WeatherInput(BaseModel): city: str Field(description城市名称) date: str Field(description日期格式为YYYY-MM-DD) tool(args_schemaWeatherInput) def get_weather(city: str, date: str) - str: 查询指定城市在指定日期的天气情况返回温度和降水信息。 # 实际项目里这里会调用天气 API return f{city}在{date}的天气晴22°C然后把工具绑定给模型from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini) llm_with_tools llm.bind_tools([get_weather]) response llm_with_tools.invoke(北京明天适合跑步吗)执行后response.tool_calls里就有结构化的工具调用信息。你要做的就是根据tool_calls去执行对应函数把结果返回给模型模型继续生成最终答复。这个“调用模型 → 检查是否有 tool_calls → 执行工具 → 把结果拼回去 → 再次调用模型”的循环就是 Agent 的核心骨架。我建议不要在项目里手动维护这个循环因为边界情况太多了工具报错了怎么办模型返回了不存在的工具名怎么办一次返回了多个工具调用怎么办这些在 LangChain 生态里都有现成方案比如AgentExecutor或者更推荐的方式是LangGraph手写状态机。3.3 LangGraph把代理从黑盒变成可控状态机前面提到AgentExecutor是黑盒循环这对复杂场景不够用。LangGraph 是 LangChain 团队推出的一个专门解决这个问题的框架核心概念是用图来定义代理的流程每个节点是一个处理步骤比如调用模型、执行工具、生成最终回复边是节点之间的跳转关系全局维护一个状态对象在节点间传递。这种方式为什么好因为它把“Agent 的循环逻辑”变成了一个你可以完全控制的图。你可以在任意节点之间加条件边可以加入人工审批步骤可以在执行工具之前打日志甚至可以回退到之前的节点重新执行。从“把控制权交给框架”变成了“把控制权握在自己手里”。我用 LangGraph 搭一个带天气工具的示例帮你直观感受一下。先定义状态在这个例子里状态就是一个 dict里面存着一组消息。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, AIMessage, ToolMessage class AgentState(TypedDict): messages: list然后定义两个节点一个调用模型一个执行工具。import json from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage model ChatOpenAI(modelgpt-4o-mini).bind_tools([get_weather]) def call_model(state: AgentState): response model.invoke(state[messages]) return {messages: [response]} def call_tool(state: AgentState): last_ai_message state[messages][-1] tool_messages [] for tool_call in last_ai_message.tool_calls: tool_name tool_call[name] tool_args tool_call[args] if tool_name get_weather: result get_weather.invoke(tool_args) tool_messages.append(ToolMessage(contentstr(result), tool_call_idtool_call[id])) return {messages: tool_messages}再把图建起来配上条件边def should_continue(state: AgentState): last_message state[messages][-1] if last_message.tool_calls: return call_tool return end graph_builder StateGraph(AgentState) graph_builder.add_node(call_model, call_model) graph_builder.add_node(call_tool, call_tool) graph_builder.set_entry_point(call_model) graph_builder.add_conditional_edges(call_model, should_continue, {call_tool: call_tool, end: END}) graph_builder.add_edge(call_tool, call_model) graph graph_builder.compile()运行这个图result graph.invoke({messages: [HumanMessage(content北京明天适合跑步吗)]}) for message in result[messages]: print(message.type, message.content)这个示例虽然短但已经体现了 LangGraph 的精髓你可以在should_continue里加更多判断条件在call_tool里加更多异常处理在节点之间插入人工确认逻辑。你画出的这张图就是代理真正的“思维链骨架”。3.4 Agent 的取舍经验什么时候用怎么避坑最后聊聊 Agent 的实战取舍。我的经验是Agent 的上限很高下限也很低。模型选错工具、参数传错、循环不终止这些都是常见问题。下面几条建议会帮你少踩很多坑工具定义要非常明确。工具的名字、描述、参数 schema决定了模型能不能正确选择它。描述里多写清“什么时候用这个工具什么时候不用”模型的表现会明显提升。比如你的天气工具只支持国内城市那描述里就要写清楚“仅支持中国主要城市海外城市请用另一个工具”。给工具加超时和异常兜底。工具挂起会导致整个 Agent 卡死工具抛异常会导致循环中断。我在生产环境里给每个工具都包了一层超时和异常捕获用 LangGraph 的节点内 catch 来兜底。限制最大迭代次数。模型有可能陷入“无限反思”的循环——一会儿想查资料一会儿又觉得信息不够反复调工具。LangGraph 里可以在状态里加一个iteration_count超过阈值直接走END避免 Token 白烧。保持状态精简。很多人把所有工具返回结果都塞进messages几轮交互后上下文越来越长模型越来越“糊涂”。建议在每次工具调用后对长文本做摘要或截断只保留关键信息进下一轮。4. 工程化落地记忆、可观测与测试缺一不可4.1 记忆不只是“历史消息”内存管理的三种方案聊完 Agent 的核心机制必须说说记忆。这个坑几乎每个做对话型应用的开发者都会踩到一开始只是把用户的所有历史消息一股脑塞进messages结果对话超过十几轮之后模型开始“忘记”最开始的需求回答质量下降Token 费用也肉眼可见地涨。LangChain 生态里记忆方案基本可以分三类窗口记忆只保留最近 N 轮对话。实现简单适合上下文不太长的场景。但缺点很直接用户如果问“我一开始说的那个产品需求你还记得吗”模型大概率已经忘了。摘要记忆每过几轮就让模型把之前的对话总结成摘要注入下一轮。它能在固定 Token 预算内保留大量历史信息——摘要可以写得很长比原始对话省很多空间。代价是会丢失细节比如用户明确说过“价格不要超过 500 块”摘要如果没记这条后面就麻烦了。向量数据库记忆把历史对话按语义切片写入向量库每次请求时检索相关片段加入上下文。这种方案适合“从 long-term memory 里捞细节”的场景比如客服系统里调取用户很久以前的订单问题。但它引入了检索质量的不确定性检索得好不好直接决定模型回得好不好。我的建议是不要一开始就奔着“最复杂”的方案去。先用窗口记忆跑通产品等用户反馈“模型不记得我说过什么”再升级到摘要或向量检索。很多时候摘要记忆对绝大多数场景已经够了向量库更多是锦上添花。4.2 可观测性别等线上出错了才后悔生产环境中可观测性比任何“炫酷功能”都重要。LangChain 的 Runnable 接口天然支持回调Callback你可以在回调里把每个步骤的输入输出、Token 用量、耗时全部记录到日志系统中。简单的方式是用 LangSmith官方 SaaS 产品它能自动追踪链和代理的执行过程。如果你不想引入外部服务也可以自己写个 CallbackHandlerfrom langchain_core.callbacks import BaseCallbackHandler import logging logger logging.getLogger(langchain) class LoggingCallbackHandler(BaseCallbackHandler): def on_llm_start(self, serialized, prompts, **kwargs): logger.info(fLLM 开始调用, prompts: {prompts}) def on_llm_end(self, response, **kwargs): logger.info(fLLM 调用结束, output: {response.generations[0][0].text}) def on_chain_start(self, serialized, inputs, **kwargs): logger.info(fChain 启动, 输入: {inputs}) def on_chain_end(self, outputs, **kwargs): logger.info(fChain 结束, 输出: {outputs})然后把回调传进chain.invoke(..., config{callbacks: [LoggingCallbackHandler()]})。这种方式的好处是全链路可见只要用户说了一句“你的回答不对”你就能立刻追踪到是检索没召回、还是模型理解偏了、还是工具返回了错误数据。4.3 测试与评估Agent 不能靠“感觉”上线链和代理的差异性带来了一个现实问题链可以用固定输入输出做回归测试代理因为每一步都是概率性的同样的输入你测十次可能得十种结果。那怎么测试我的经验是分三层单元层每个工具本身要测。工具返回格式是否稳定、是否有异常兜底、超时是否合理。流程层给每个“典型路径”写测试用例。比如“用户问天气 → 模型调用工具 → 工具返回 → 模型生成回答”你可以用 mock 工具的方式不走真实 API固定工具返回结果验证模型是否能正确利用这个结果。质量层用一份评估集比如包含 50 条典型问题和期望回答让 Agent 批量跑一遍再用模型或人工打分。这个分数不用追求一次就很高关键是上线后每次改动都要跑一遍确保没有“改一个 bug 引来另一个 bug”。我还发现一个很好用的思路质量层的评估可以用模型当裁判。把一个测试问题和 Agent 的回答连同“参考正确回答”一起发给一个更强大或更便宜的模型让它评分并写理由。这不算完美但能低成本解决大部分“Agent 是不是答非所问”的检测需求。4.4 缓存、限流与成本控制工程化的最后一环把 Agent 部署上线成本是一个绕不开的话题。一个简单的天气对话一次 Agent 循环可能调用好几次模型第一次推理、工具返回后再推理、最后生成回答可能还会因为上下文太长而多次消耗 Token。成本控制就成了实打实的需求。LangChain 提供缓存机制比如给ChatOpenAI配置内存缓存from langchain.cache import InMemoryCache from langchain.globals import set_llm_cache set_llm_cache(InMemoryCache()) llm ChatOpenAI(modelgpt-4o-mini)这样同样的请求同样的模型、同样的消息列表会直接命中缓存不会重复调用模型。对测试环境特别合适。生产环境建议用 Redis 缓存跨进程共享命中率更高。限流也值得做。模型 API 有 QPS每秒请求数限制Agent 循环里一旦模型同时发起多个工具调用很容易突然飙高请求量导致限流报错。LangChain 里可以用rate_limiter或者自己包一层令牌桶限流逻辑。表面上这只是“控制请求频率”实际上它避免了因为偶发限流错误而导致 Agent 流程整个崩溃的最坏情况。5. 从链到代理下一步该做什么写到这里我不太想搞一个“全文总结”。我更想分享的是当我真正把一个基于 LangChain 的对话系统从链的架构迁移到 LangGraph 代理架构之后最大的感受是不要把 Agent 当成银弹也不要觉得 Chain 已经过时。我现在的项目里大量简单的“意图识别→定向回答”场景仍然用链加分支实现只有那些需要多步推理、动态选择工具的任务才交给 Agent。两者之间不是替代关系而是一个连续的光谱你要做的是准确判断每个场景在这个光谱上的位置。另外还有一句实在话LangChain 生态变化很快但核心思想没有变——无论链、代理还是状态图本质上都是在回答同一个问题如何把大模型组织成一个可靠、可控、可观测的应用。你只要抓住这条主线官方文档再怎么改版都不会迷失方向。如果你正准备在自己的项目里引入 LangChain我建议你先别急着写代码先花半天时间把你要解决的任务拆清楚哪些步骤是确定的哪些步骤需要模型自己判断工具的边界在哪这份拆解文档比任何框架技巧都重要。框架只是工具工程思维才是你真正要掌握的“核心”。