LangChain从链到代理:LCEL与LangGraph构建可控AI应用
2026/9/8 20:18:56 网站建设 项目流程

1. 从链到代理,LangChain 的核心演进逻辑

在聊 LangChain 之前,我先说个真实感受:这个框架的迭代速度,快到很多教程刚发出来就过时了。你如果翻到两年前的 LangChain 入门文章,看到的还是LLMChainSimpleSequentialChain那一套;再过半年去看,又变成了 LCEL(LangChain 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 到 LCEL,LangChain 做对了什么

如果你接触过 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(temperature=0) chain = LLMChain(llm=llm, prompt=prompt) 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(model="gpt-4o-mini") chain = prompt | llm | StrOutputParser() result = chain.invoke({"topic": "量子计算"})

注意|这个符号,它定义了一个明确的输入输出契约:左边组件的输出,一定是右边组件的输入。你可以把它想象成水管——每一段管子都有一个标准的接口,接上就能流水。这不是语法糖,而是一种统一的抽象:任何 Runnable 对象都有invokestreambatchainvoke等方法,这就意味着你可以在不改变上层代码的情况下,替换任意一个环节。

我记得有人问过一个很尖锐的问题:“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( complexity=analyze_complexity, bugs=analyze_bugs, readability=analyze_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 Calling,Agent 到底怎么落地

3.1 先弄清楚 ReAct 的原理:推理和行动交错的循环

要理解 Agent,绕不开 ReAct(Reasoning 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_schema=WeatherInput) def get_weather(city: str, date: str) -> str: """查询指定城市在指定日期的天气情况,返回温度和降水信息。""" # 实际项目里这里会调用天气 API return f"{city}在{date}的天气:晴,22°C"

然后把工具绑定给模型:

from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-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(model="gpt-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(content=str(result), tool_call_id=tool_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 产品),它能自动追踪链和代理的执行过程。如果你不想引入外部服务,也可以自己写个 CallbackHandler:

from langchain_core.callbacks import BaseCallbackHandler import logging logger = logging.getLogger("langchain") class LoggingCallbackHandler(BaseCallbackHandler): def on_llm_start(self, serialized, prompts, **kwargs): logger.info(f"LLM 开始调用, prompts: {prompts}") def on_llm_end(self, response, **kwargs): logger.info(f"LLM 调用结束, output: {response.generations[0][0].text}") def on_chain_start(self, serialized, inputs, **kwargs): logger.info(f"Chain 启动, 输入: {inputs}") def on_chain_end(self, outputs, **kwargs): logger.info(f"Chain 结束, 输出: {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(model="gpt-4o-mini")

这样同样的请求(同样的模型、同样的消息列表)会直接命中缓存,不会重复调用模型。对测试环境特别合适。生产环境建议用 Redis 缓存,跨进程共享,命中率更高。

限流也值得做。模型 API 有 QPS(每秒请求数)限制,Agent 循环里一旦模型同时发起多个工具调用,很容易突然飙高请求量导致限流报错。LangChain 里可以用rate_limiter或者自己包一层令牌桶限流逻辑。表面上这只是“控制请求频率”,实际上它避免了因为偶发限流错误而导致 Agent 流程整个崩溃的最坏情况。

5. 从链到代理,下一步该做什么

写到这里,我不太想搞一个“全文总结”。我更想分享的是,当我真正把一个基于 LangChain 的对话系统从链的架构迁移到 LangGraph 代理架构之后,最大的感受是:不要把 Agent 当成银弹,也不要觉得 Chain 已经过时。我现在的项目里,大量简单的“意图识别→定向回答”场景仍然用链加分支实现,只有那些需要多步推理、动态选择工具的任务才交给 Agent。两者之间不是替代关系,而是一个连续的光谱,你要做的是准确判断每个场景在这个光谱上的位置。

另外还有一句实在话:LangChain 生态变化很快,但核心思想没有变——无论链、代理还是状态图,本质上都是在回答同一个问题:如何把大模型组织成一个可靠、可控、可观测的应用。你只要抓住这条主线,官方文档再怎么改版,都不会迷失方向。

如果你正准备在自己的项目里引入 LangChain,我建议你先别急着写代码,先花半天时间把你要解决的任务拆清楚:哪些步骤是确定的?哪些步骤需要模型自己判断?工具的边界在哪?这份拆解文档,比任何框架技巧都重要。框架只是工具,工程思维才是你真正要掌握的“核心”。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询