LangChain三大核心:链、记忆、代理实战解析
2026/9/8 21:12:27 网站建设 项目流程

你第一次真正用 LangChain 写业务,十有八九会觉得“这玩意不就是封装了一下提示词吗”。等到你被对话上下文搞得焦头烂额、被 Agent 的工具调用绕进死循环,再回头看框架的设计,才会明白链、记忆、代理这三个核心抽象到底是用来解决什么问题的。

这篇文章我从实际项目视角出发,把 LangChain 生态里开发者最该吃透的三大核心拆开讲:链(Chain)、记忆(Memory)、代理(Agent)。重点放在“为什么需要”、“怎么落地”,以及从链过渡到代理时那几步关键的思维转变。适合刚入门 LangChain、或者已经写过简单 Demo 但想往生产级 Agent 方向靠的开发者。全文按实战代码和踩坑记录来写,争取你看完能直接拿去用。

1. 三大核心到底是什么,为什么偏偏是它们

LangChain 早期最出名的概念就是链,那时候官方文档首页一句话就能概括:把大模型的调用和各种外部能力串成一个稳定的管道。后来 Agent 火了,LangChain 又顺着趋势把工具调用、自主决策做成了核心能力。但不管生态怎么扩张,底层的土地永远是那几块——链、记忆、代理,再加上一个贯穿一切的 Runnable 接口。

我习惯用一个类比来理解这三者的关系。链就是一家工厂的流水线:每个工位固定干什么,零件进来按顺序走,产出稳定可预测。记忆是这条流水线的“临时仓库”:上一批零件放在哪、状态是什么,下次还能接着用。代理则像流水线前的智能调度员:接到一个模糊订单,先拆解需求,再决定调用哪条产线、哪个工具,做完一步还知道回头检查一下结果对不对。

为什么必须掌握这三个?因为 LangChain 几乎所有上层功能,比如 RAG、多轮对话、Agent 执行器、LangGraph 的状态图,归根结底都是在组合这三样东西。你要是只写prompt | llm | parser这种最简链,那确实没必要用框架;可一旦要处理工具选择、历史上下文、多步任务编排,没有清晰的概念模型,代码必然失控。

另外要注意一点:链、记忆、代理不是三个并列的功能模块,而是层层递进的关系。先有链,你才能把 LLM 调用看成一个“可组合的单元”;有了单元,再用记忆给它加上状态;最后把决策权交给模型,让它自己决定下一步执行哪个单元,这就是代理。理解了这条演进线,就不会在写项目时把三层概念搅成一锅粥。

2. 第一大核心:链,把 LLM 调用变成可控的生产线

2.1 从最简链开始,理解 LCEL 的管道思想

很多教程一上来就给你一个LLMChain这样的老 API,我用的时候已经慢慢被官方标记为 legacy,新手照着写容易踩错版本。这里我更推荐直接对比 LCEL(LangChain Expression Language)的写法。看一个最基础的例子:

from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个精通{domain}的资深顾问。"), ("human", "请回答我的问题:{question}") ]) llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3) parser = StrOutputParser() chain = prompt | llm | parser result = chain.invoke({"domain": "Linux 运维", "question": "服务器 CPU load 突然飙高,怎么排查?"}) print(result)

这段代码的核心不是“调了一次 OpenAI”,而是那个竖线管道符。在 LCEL 里,每个组件都实现了Runnable接口,统一支持invoke()batch()stream()ainvoke()等方法。竖线的作用就是把上一个组件的输出,自动当成下一个组件的输入。Prompt 接收字典,格式化成消息列表送给 LLM;LLM 返回 AIMessage,StrOutputParser 再把它剥成纯文本。

这个设计的优势在于:每个环节都可以单独替换、单独测试。今天想从裸 LLM 换成带记忆的 Runnable,或者想在 LLM 后面挂一个 JSON 校验器,都只需要在管道中间加一个节点就行,不用大改业务代码。实际项目里,链的价值就是把那些“每次都要写一遍”的胶水代码收敛成一个对象。

2.2 并行与组合:用 RunnableParallel 解决“多个数据源拼答案”

真实业务很少只查一个东西。最常见的是:用户问“这个订单现在什么状态,包里都有啥”,系统得同时查订单接口和商品接口,拿到两份结果再让模型汇总。

一开始大家会习惯写成串行,先查订单,再查商品,最后组装 Prompt。但串行的问题有两个:一是慢,两个接口各自耗时,累加起来体验很差;二是代码碎,查询逻辑全散落在业务函数里,根本看不出还有一条“链”。用 RunnableParallel 可以直接在管道里做并行派发:

from langchain_core.runnables import RunnableParallel, RunnablePassthrough def query_order(order_id: str) -> dict: # 模拟订单服务 return {"order_id": order_id, "status": "shipped", "items": [{"sku": "A100", "name": "键盘"}]} def query_product(sku: str) -> dict: # 模拟商品服务 return {"sku": sku, "stock": 120, "price": 299} parallel_fetch = RunnableParallel( order=lambda x: query_order(x["order_id"]), product=lambda x: query_product(x["sku"]) ) answer_prompt = ChatPromptTemplate.from_messages([ ("human", "订单状态是:{order};第一件商品是:{product}。请用一句自然语言告诉用户结果。") ]) final_chain = parallel_fetch | answer_prompt | llm | parser result = final_chain.invoke({"order_id": "AB123", "sku": "A100"})

RunnableParallel 的输出是一个字典,键是你在构造时起的名字,值是各个分支执行的结果。这个能力尤其适合做 RAG 应用:一边用向量检索召回相关片段,一边实时拉取业务数据,最后两个结果一起进 Prompt。我在生产里用这套逻辑,把原本 3 秒多的多源查询压到了 1.2 秒左右,主要节省的就是并行化带来的那部分时间。

RunnablePassthrough 也经常和它搭配。它什么都不做,只是把输入原封不动传递下去,适合“既要把用户原始输入传给 Prompt,又要把中间结果传下去”的场景。比如{"question": RunnablePassthrough()} | ...,这样最终 Prompt 里既能拿到用户问题,也能拿到中间的检索结果。

2.3 别只会 invoke,batch 和 stream 才是性能优化切入点

写到链之后,我发现很多新手的性能问题不是模型慢,而是不会用链的执行模式。链对象自带了四种调用方式,每种都有明确的使用场景:

方法适用场景注意事项
invoke()单次请求,返回完整结果最简单,但流式场景体验差
batch()批量处理固定输入列表框架内部并发执行,替代手写 for 循环
stream()流式打字机效果需要 LLM 支持流式输出
ainvoke()异步接口,比如 FastAPI 路由配合 async/await 使用

batch 特别容易被忽略。举个例子,离线跑一批历史工单分类任务,100 条数据,你如果在 Python 里 for 循环 invoke,时间几乎是串行累加的;换成chain.batch([{"question": q} for q in queries]),框架会按内部策略并发调度,总体耗时能省一半以上。stream 则直接影响产品体感,用 ChatOpenAI 默认就支持流式,链的单元测试里可以直接断言首个 chunk 是否非空。

还需要提醒一点:LCEL 里prompt | llm | parser只是一个 Runnable,它本身可以继续被赋值、被组合、被嵌套。所以你可以先把某个子流程封装成一个函数,再在更大的链里调用它。这种“链套链”的思路,是后面做 Agent 时把工具逻辑复用的基础。

3. 第二大核心:记忆,让应用从“失忆”到“有状态”

3.1 三种主流记忆实现,怎么选

聊到 LangChain 多轮对话,绕不开记忆。LLM 本身是无状态的,你每次问它,它都不记得刚才聊过什么。记忆的作用就是把历史消息以某种形式重新塞进 Prompt。LangChain 传统上提供了很多 Memory 类,我建议大家以题型和成本为导向来选择:

记忆类型核心方式适合场景主要代价
BufferMemory保存完整消息列表轮次少、上下文短历史一长,Token 暴涨
BufferWindowMemory只留最近 K 轮轻度闲聊、客服太久远的信息会丢
SummaryMemory定期对历史生成摘要长时间深度对话摘要本身有误差和延迟
VectorStoreRetrieverMemory检索最相关的历史片段知识库问答、个性化推荐需要额外维护向量库

我踩过一个比较典型的坑:在长对话里直接上 BufferMemory,结果十几轮之后 Prompt 长度直接突破窗口限制,服务报 400。后来换成 SummaryMemory 才好一些,但也不是无脑能用,因为摘要每轮都要额外触发一次摘要模型,成本翻了一倍。所以做生产前一定要想清楚:你的业务到底需要多长的历史?大部分客服场景最近 10 轮足够,完全不需要长期记忆。

3.2 在 LCEL 链里优雅地挂载记忆

旧版 LangChain 的ConversationBufferMemory用法方便,但到了 LCEL 时代,官方更推荐用RunnableWithMessageHistory。因为 LCEL 本身不管理状态,它把状态交给外部存储,这样更容易做多用户隔离和持久化。

from langchain_core.runnables.history import RunnableWithMessageHistory from langchain_core.chat_history import InMemoryChatMessageHistory # 一个简单的内存存储,生产环境可以换成 Redis 实现 store = {} def get_session_history(session_id: str): if session_id not in store: store[session_id] = InMemoryChatMessageHistory() return store[session_id] chain = prompt | llm # 这里 prompt 已经包含 {history} 占位 chain_with_history = RunnableWithMessageHistory( chain, get_session_history, input_messages_key="question", history_messages_key="history" ) result = chain_with_history.invoke( {"question": "我刚才问你什么来着?"}, config={"configurable": {"session_id": "user-001"}} )

核心在于:get_session_history负责按 session_id 去拿历史消息,RunnableWithMessageHistory会在每次调用前自动把历史拼进 Prompt,调用后再把新消息写回存储。这样每个用户的数据天然隔离开,不会出现我把 A 用户聊的内容带到 B 用户面前的尴尬。

如果你追求极致控制,也可以不依赖这个封装,自己维护一个messages列表,然后手动拼接:

messages = [] messages.append(HumanMessage(content="你好")) messages.append(AIMessage(content="你好,有什么可以帮你?")) # 下一次调用 messages.append(HumanMessage(content="介绍一下你自己")) response = llm.invoke(messages) messages.append(response)

这种写法看着原始,但正好绕过了 Memory 类在各个版本里的 API 变动。我这里没有贬低框架的意思,只是如果你被乱七八糟的兼容性折磨过,就会明白直接操作消息列表是最稳的兜底方案。

3.3 记忆使用里绝对不能踩的三个坑

第一,多用户共用内存对象。我见过一个线上事故:history被定义成了模块级全局变量,所有用户共用同一个列表,几分钟后对话内容互相串。解决方式就是上面说的按 session_id 隔离,或者直接引入 Redis 存历史。

第二,上下文爆炸。历史消息无脑全塞,模型输入越来越长,响应越来越慢,成本越来越高。实测下来,一个 64k 上下文模型,每次塞入 5 万 token 历史之后,单次响应延迟能翻三倍。合理的方式是截断窗口 + 摘要降维,再配合 LangSmith 这类工具去观测实际 token 消耗。

第三,历史消息格式不统一。有些聊天场景你会存中间插件的输出、工具调用记录,这些非纯文本消息一旦拼错格式,LLM 理解就会漂。最好在进入 Prompt 前统一做一次消息清洗,把工具输出转成规范的角色消息。这一条在老版本 Memory 迁移时尤其常见。

4. 第三大核心:代理,从固定流程到自主决策

4.1 代理是怎么工作的:reAct 与 Function Calling

链是固定管道,代理则是“模型自己决定下一步”。这个机制的核心循环是“思考—行动—观察”:模型先根据用户问题判断需要什么信息,然后选择一个工具执行,把工具结果拿回来当成新的观察,再决定下一步是继续调用工具还是直接生成最终答案。

语言上,Agent 的世界最先火起来的是 ReAct 模式。它的思路是让模型输出类似“Thought: 我需要查一下天气;Action: call weather tool;Action Input: Beijing”这样的结构化文本,框架解析后执行工具,再把结果拼回 Prompt。LangChain 的create_react_agent就是这种思路。

后来 OpenAI 推出了 Function Calling,模型不再是“生成一段带格式的文本”,而是直接返回一个结构化的工具调用请求,里面包含工具名和参数 JSON。LangChain 里对应的是create_tool_calling_agent。实操下来,Function Calling 比 ReAct 稳定很多,因为工具名和参数不再依赖文本格式猜测,基本不会出现“解析失败后整体崩掉”的问题。

我给一个当前版本的参考代码,langchain 0.2+环境下可用:

from langchain_openai import ChatOpenAI from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate from langchain_core.tools import tool import datetime @tool def get_utc_time() -> str: """获取当前 UTC 时间。需要和时间相关的命令时使用。""" return datetime.datetime.utcnow().isoformat() @tool def add_numbers(a: float, b: float) -> float: """计算两个数字的和。参数 a、b 必须是数字。""" return a + b tools = [get_utc_time, add_numbers] llm = ChatOpenAI(model="gpt-4o", temperature=0) prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个能调用工具的助手。请根据用户问题决定是否调用工具,必要时可以多次调用。"), ("placeholder", "{agent_scratchpad}"), ("human", "{input}") ]) agent = create_tool_calling_agent(llm, tools, prompt) executor = AgentExecutor(agent=agent, tools=tools, verbose=True) result = executor.invoke({"input": "现在 UTC 时间是几点?顺便帮我算一下 123.45 加 67.8 的和。"}) print(result["output"])

注意两个细节。第一,prompt里必须有{agent_scratchpad}这个占位符,它专门用来存放模型之前的思考和工具返回结果。第二,工具函数的 docstring 极其重要,模型就是靠这个描述来决定“该不该用这个工具”,描述写得含糊,模型基本不会调用它。

4.2 从链改造成代理,差的不是代码量,而是决策点

很多团队不敢上代理,觉得不可控。我的建议是:先用链跑通流程,再逐步把决策点交给代理,不要一步到位。

什么意思呢?假设你要做一个“运维助手”。第一版用链,固定先执行df -h看磁盘,再把输出丢给模型生成建议。这个版本很稳,但用户一旦问“CPU 负载怎么样”,它就傻了,因为链里没有“判断用户意图”这一环。

第二版改成代理,给模型两个工具:一个查磁盘,一个查 CPU。模型接到任意问题,先决策该调用哪个工具。这就是核心差异:链的分支是提前写死的,代理的分支是运行时模型自己选的。

@tool def check_disk() -> str: """检查服务器磁盘使用率,返回 df -h 命令的输出。""" import subprocess return subprocess.check_output(["df", "-h"], text=True) @tool def check_cpu() -> str: """检查服务器 CPU 负载,返回 top 命令前 5 行。""" import subprocess return subprocess.check_output(["top", "-b", "-n", "1", "|", "head", "-5"], shell=True, text=True)

改造过程中你会发现,真正要花心思的是把工具定义清楚,引导模型在合适的时候选择正确的工具。尤其当工具超过 5 个,模型选择就可能开始不稳定。进阶一点的做法是再加一层“工具路由器”,用一个成本更低的模型先做意图分类,再决定把任务路由给哪几个工具空间,这比把所有工具一次性塞给大模型要稳定得多,也是官方工具选择器之外的实战技巧。

4.3 AgentExecutor 控制代理行为的关键参数

Agent 跑起来之后,第一件事是设置安全上限。AgentExecutormax_iterationsmax_execution_time是我建议必开的:

executor = AgentExecutor( agent=agent, tools=tools, verbose=True, max_iterations=3, # 最多思考-行动-观察几轮 max_execution_time=30, # 整个 Agent 最多执行 30 秒 early_stopping_method="force" # 到上限后强制返回 )

没有上限的 Agent 是真的可以在工具调用之间反复横跳的,尤其是模型某次工具结果不理想时,会被困在循环里出不来。刚开始调试时把verbose=True打开,你能眼睁睁看到它在想什么、调了什么工具、结果是什么,这是排查问题的第一现场。

5. 实战:一个“查询 + 计算”的混合 Agent,从链改到代理

5.1 先用链写死流程,把业务逻辑跑通

为了能更直观地感受“从链到代理”,我拿一个具体场景走一遍完整过程:用户会问两种问题——查订单状态、算发票金额(含税/不含税)。第一版用链,固定路由就是先判断问题里是否包含“订单”。包含就走查询逻辑,否则走计算逻辑。你可以写成一个带判断的 RunnableLambda,但分支仍然是代码写死的。

from langchain_core.runnables import RunnableLambda def route(inputs: dict) -> dict: question = inputs["question"] if "订单" in question: return {"type": "order", "order_id": inputs.get("order_id", "UNKNOWN")} return {"type": "calc", "expr": inputs.get("expr", "")} order_chain = order_prompt | llm | parser calc_chain = calc_prompt | llm | parser def smart_chain(inputs: dict): routed = route(inputs) if routed["type"] == "order": return order_chain.invoke(inputs) return calc_chain.invoke(inputs)

这版能跑,但问题很明显:只要问题稍微绕一点,比如“用户说帮我看看上次买的东西到哪了”,它匹配不到“订单”,就去走计算流程,结果完全跑偏。这就是链的局限性。

5.2 换成代理,让模型根据语义选工具

第二步,把同样的两个业务逻辑包成两个 tool,交给create_tool_calling_agent调度。这里代码反而更简洁一点:

@tool def query_order_status(order_id: str) -> str: """查询订单物流状态。当用户提到订单、快递、物流、到货时使用。""" # 这里调用真实订单服务 return f"订单 {order_id} 当前状态:已发货,预计 2 天后送达。" @tool def calculate_amount(principal: float, tax_rate: float = 0.13) -> str: """计算含税金额。当用户需要计算金额、税额、含税价时使用。principal 是不含税金额,tax_rate 默认 13%。""" total = principal * (1 + tax_rate) return f"不含税金额 {principal},税率 {tax_rate:.0%},含税金额为 {total:.2f}。" agent = create_tool_calling_agent(llm, [query_order_status, calculate_amount], prompt) executor = AgentExecutor(agent=agent, tools=[query_order_status, calculate_amount], max_iterations=3)

现在模型面对“我那个手机订单发货了没”这种说法,也能通过语义匹配到订单查询工具。这就是代理的核心收益:把“识别用户意图”这件事从你的代码里搬到了模型决策里。代价是你要多准备一些典型的测试问题,确保模型不会乱选工具。我现在每个 Agent 上线前都会准备一个二十条左右的回归用例集,专门验证工具路由的准确性。

5.3 给代理加上记忆,形成完整闭环

光有工具还不够,用户很可能上一句说下单号,这一句问“那它快递到哪了”,如果不带记忆,代理根本不知道“它”指哪个订单。把记忆和代理组合起来是最后一步:

from langchain_core.runnables.history import RunnableWithMessageHistory agent_with_history = RunnableWithMessageHistory( executor, get_session_history, input_messages_key="input", history_messages_key="chat_history" ) result = agent_with_history.invoke( {"input": "帮我查一下订单 AB123"}, config={"configurable": {"session_id": "user-001"}} ) # 下一轮直接问“它现在到哪了” result2 = agent_with_history.invoke( {"input": "它现在到哪了?"}, config={"configurable": {"session_id": "user-001"}} )

这里要注意,AgentExecutor 这个 Runnable 的输入键是input,不是question,所以配置input_messages_key="input"是必要的。踩过几次坑之后,我发现一个规律:凡是套了 RunnableWithMessageHistory 的链或代理,第一件事就应该去看它的输入输出键定义,别想当然。

6. 常见问题与排查技巧实录

6.1 代理和链使用中的高频问题速查表

现象可能原因解决方案
代理一直循环执行同一个工具模型无法从工具结果中推断下一步设置 max_iterations;检查工具描述是否清晰
“Agent stopped after X iterations”超过轮次上限调大 max_iterations,或简化任务,拆成多个 Agent
函数调用返回 JSON 解析失败模型生成的参数不符合函数 schema换用更强模型;给 function schema 加严格的类型约束
上下文超长,报 token 超限历史消息和工具中间结果都塞进 Prompt裁剪历史 + 工具输出只保留摘要
多用户对话串数据全局共享 history 对象用 session_id 隔离,外部化存储
工具明明返回了正确结果,模型还是瞎编结果格式不明确,或模型本身能力不足工具返回结构化文本;严重时补充 few-shot 示例

6.2 langchain 和 langgraph 到底怎么选

LangChain 和 LangGraph 是很多新人的疑惑点,其实分工很简单:LangChain 提供组件库和各类 Runnable;LangGraph 则关注有状态的图结构,允许你把 Agent 的每一步建模成节点和边,支持分支、循环、人工审批这种更复杂的控制流。

我的经验是:先学 LangChain,把链、工具、记忆这些基本元素搞熟;当你发现 AgentExecutor 那个“黑盒循环”已经满足不了需求——比如你想在工具调用前加人工审批、想在特定条件下强制结束、想让多个角色 Agent 协作——这时候再上 LangGraph。它本质上就是让你自己画出那张状态流转图,每个节点可以是 LangChain 的 Runnable,图的转移条件由代码或者模型决定。官方在 0.3+ 已经把 AgentExecutor 标记为 legacy,推荐用 LangGraph 构建生产级 Agent,但它并没有替代 LangChain,而是在其上做编排。

6.3 学习路径建议:不要一上来就奔着“智能体”去

我看到太多人直接跳到 Agent 实战,结果失败后归咎于框架。我的建议是,按这样的顺序稳扎稳打:

  1. 用 LCEL 写十个不同的最简链,把invokebatchstreamainvoke全部用一遍。
  2. 找一个小业务场景,用 RunnableParallel 做一次并行数据源组装。
  3. 给链挂上 RunnableWithMessageHistory,跑通多轮对话,并测试不同 session 之间的隔离。
  4. 封装两个自己的业务工具,用create_tool_calling_agent组装一个低风险的 Agent,比如查天气、算时间、做简单计算。
  5. 最后再去看 LangGraph,用状态图重构一个你写过的 Agent,体会节点和边的控制力。

至于“LangChain 有 Java 版吗”这种问题,顺带说一句:LangChain 官方主推是 Python 和 TypeScript 两个版本,Java 生态有社区衍生的 langchain4j。如果团队技术栈主要是 Java,可以直接了解 langchain4j,不过轮子成熟度还是没法跟 Python 版比,想长期深耕这个方向,Python 绕不过去。

一些写在最后的经验

我自己的项目里,纯链和纯代理从来不是非此即彼。比较成熟的做法是“外围用链,决策用代理”:固定的数据清洗、检索、格式化流程全部写成链,保证稳定可控;只有当任务需要模型自己做意图判断、多步骤执行时,才把这一段交给代理。这样既有了链的确定性,也有代理的灵活性,出了问题也容易定位。

回到 LangChain 本身,它确实不是个性能最强的框架,更新节奏也快得让人头疼。但它的价值在于把 LLM 应用开发里那些反复出现的模式——管道编排、状态管理、工具调用——进行了统一抽象。早期多学一点底层的 Runnable 和消息结构,比追着最新 API 抄代码要划算得多。踩过几次坑之后你就会发现,真正让你离不开的不是某个具体类,而是那一套把大模型变成工程化服务的设计思路。

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

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

立即咨询