2026-04-15发表2026-04-19更新AI / LangChain42 分钟读完 (大约6351个字)

LangChain：链的核心方法、构建使用、高级用法

LangChain 中的链（Chain）是其核心抽象之一，用于将多个组件（如模型、提示词模板、输出解析器等）组合成一个可执行的流水线，支持同步/异步调用、流式输出、序列化及记忆功能，是构建LLM应用的基础架构。

现代的 LangChain 主要推荐使用 LangChain 表达式语言（LCEL） 来构建链，它更简洁、功能也更强大。

核心方法

现代 LangChain 链的核心是 Runnable 接口，它为所有可执行组件（包括 LCEL 链）定义了标准化的调用方式。

方法解析

同步方法	异步方法	功能描述	输入类型	输出类型
`invoke`	`ainvoke`	单输入单输出，最基本的调用方式。	`InputType`	`OutputType`
`batch`	`abatch`	批量处理多个输入，高效处理列表。默认并发执行。	`list[InputType]`	`list[OutputType]`
`stream`	`astream`	流式输出，逐步生成结果，适用于长文本生成场景。	`InputType`	迭代器
`astream_log`	-	流式输出并包含中间步骤的日志，方便调试复杂链。	`InputType`	包含日志的流

调用方式

方法	说明	示例
`invoke(input)`	同步调用	`chain.invoke({"input": "..."})`
`ainvoke(input)`	异步调用	`await chain.ainvoke({"input": "..."})`
`stream(input)`	流式输出（逐 token）	`for chunk in chain.stream(...): print(chunk)`
`batch(inputs)`	批量处理	`chain.batch([{...}, {...}])`

配置参数

RunnableConfig 是一个配置字典，在调用 invoke, ainvoke, batch 等方法时通过 config 参数传入，用于控制 Runnable 的执行、调试和观测行为。

1	chain.invoke(input_data, config=config_dict)

config_dict 是一个字典，支持以下主要字段：

字段	类型	默认值	作用	是否传递至子调用
`tags`	`list[str]`	`None`	为运行添加标签，用于在追踪和日志中分类筛选，可用于在 LangSmith 中过滤追踪	✅ 是
`metadata`	`dict[str, Any]`	`None`	附加元数据，例如用户 ID、环境信息，值需可JSON序列化，用于上下文追踪。	✅ 是
`callbacks`	`Callbacks`	`None`	注册回调处理器，监听执行生命周期事件（如开始、结束、错误）。	✅ 是
`run_name`	`str`	类名	覆盖追踪器中显示的运行名称。	❌ 否
`run_id`	`UUID`	自动生成	为本次执行指定唯一标识符，用于追踪和关联。	❌ 否
`max_concurrency`	`int	None`	`None`	批量处理（`batch`）时的最大并行数，默认基于 `ThreadPoolExecutor`。
`recursion_limit`	`int`	`25`	限制递归调用（如Agent执行）的最大深度，防止无限循环。	✅ 是
`configurable`	`dict[str, Any]`	`None`	为动态配置字段提供运行时值，如 `temperature`、`model` 等。可用户自定义配置，比如 `session_id`，常用于记忆、适配器等	✅ 是
`timeout`	`float`	`None`	设置请求超时时间（秒），超时后抛出错误。	❌ 否

 config = {
    "configurable": {"session_id": "user_123"},  # 配置标识
    "tags": ["my-chain", "test"],                # 添加标签便于追踪
    "metadata": {"user_id": "789"},               # 添加元数据
    "callbacks": [handler],                       # 传入回调函数
    "max_concurrency": 5,                         # 控制 batch 的并发数
    "recursion_limit": 25                         # 限制递归深度
}

configurable

configurable 字段是实现运行时动态配置的核心，其设计主要服务于两大场景：

1. 调整预定义的 configurable_fields

此功能允许在不改变代码结构的情况下，运行时动态调整组件的关键参数。比如，一个链中定义了一个可配置的 temperature（温度系数），可以在每次调用时赋予其不同的值。

from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI

# 1. 定义一个可配置的模型
llm = ChatOpenAI(model="gpt-3.5-turbo").configurable_fields(
    temperature=ConfigurableField(id="temp")
)
chain = prompt | llm

# 调用时动态配置
# result = chain.with_config(configurable={"temp": 0.9}).invoke({"topic": "创意故事"})

# 2. 运行时通过 configurable 字段注入新值
response = chain.invoke(
    {"question": "Hello"},
    config={"configurable": {"temp": 0.9}}  # 此行配置运行时的 temperature
)

2. 传递运行时密钥 (Secrets)

这是官方推荐的安全实践。对于敏感信息（如API密钥），你可以用 双下划线 __ 作为键名前缀来传递，LangChain 会确保这些信息不会被记录到追踪系统中。

@tool
def secret_tool(x: int, config: RunnableConfig) -> int:
    # 从 config 中安全地读取密钥
    secret = config["configurable"]["__my_secret"]
    return x + secret

# 调用时，以双下划线前缀传递敏感数据
secret_tool.invoke({"x": 5}, {"configurable": {"__my_secret": 2, "traceable_key": "bar"}})

在 LangSmith 的追踪记录中，traceable_key 可见，而 __my_secret 会被自动过滤隐藏

配置继承与合并：

理解 RunnableConfig 在复杂管道中的传播机制，是高级调试的关键。

配置继承：当在链的最外层（如 chain.invoke(config={...})）传入配置时，大部分字段（tags, metadata, callbacks, configurable）会默认向下传播，让内部的所有子组件都能自动应用这些配置。
配置合并：如果子组件也定义了自身的配置，LangChain 会进行智能合并。例如，tags 会被合并并去重，而 metadata 则是浅合并，子组件的值会覆盖父级传递的同名键。

输入参数

链的输入通常是一个字典，包含所有需要的变量

可以通过input_schema查看链的输入要求：

1	print(chain.input_schema.schema()) # 查看输入数据格式要求

config使用示例

示例1：为链添加追踪信息

from langchain_core.callbacks import StdOutCallbackHandler
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_template("Tell me a joke about {topic}")
model = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | model

config = {
    "tags": ["joke", "test"],                # 标签
    "metadata": {"user_id": "12345"},        # 元数据
    "callbacks": [StdOutCallbackHandler()]   # 回调：为链添加追踪信息，打印详细日志到控制台
}
# 运行时配置
response = chain.invoke({"topic": "cats"}, config=config)
print(response.content)

执行时，StdOutCallbackHandler 会打印出链执行的每一步（例如 prompt 格式化后的内容、模型调用的 token 统计等）。LangSmith 中也会按 tags 和 metadata 进行分类和过滤。

示例2：传递会话 ID

当需要为链绑定记忆（RunnableWithMessageHistory）或实现多租户隔离时，会使用 configurable 字段：

from langchain.memory import ChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

# 假设有一个简单的链
chain = prompt | model

# 为每个会话单独存储历史消息
store = {}

def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

# 包装成带记忆的链
with_memory = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="topic",          # 输入字段名（根据你的 prompt 调整）
    history_messages_key="chat_history"  # prompt 中历史消息的变量名
)

# 第一次调用，传入 configurable.session_id
config = {"configurable": {"session_id": "user_abc"}}
response1 = with_memory.invoke({"topic": "Hi, my name is Bob"}, config=config)

# 第二次调用，相同 session_id，模型会记住名字
response2 = with_memory.invoke({"topic": "What's my name?"}, config=config)
print(response2.content)  # 输出：Your name is Bob.

注意：configurable 中的键值对会被传递给 Runnable 内部使用的适配器（如 BaseChatMessageHistory），从而区分不同会话。

示例3：批量处理与并发控制

当使用 batch 处理大量输入时，可以限制并发请求数量，避免触发 API 速率限制：

inputs = [{"topic": "cats"}, {"topic": "dogs"}, {"topic": "birds"}]

config = {
    "max_concurrency": 2   # 同时最多发起 2 个请求
}

responses = chain.batch(inputs, config=config)

示例4：限制链的递归深度

在 Agent 或某些循环链中，防止无限循环，设置 recursion_limit限制递归深度。

from langchain.agents import create_openai_functions_agent, AgentExecutor

# ... 创建 agent ...
agent_executor = AgentExecutor(agent=agent, tools=tools)

config = {"recursion_limit": 5}   # Agent 最多执行 5 步
result = agent_executor.invoke({"input": "复杂任务"}, config=config)

示例5：在异步中使用config

异步方法同样支持 config：

async def run():
    config = {"tags": ["async-demo"]}
    async for chunk in chain.astream({"topic": "cats"}, config=config):
        print(chunk.content, end="", flush=True)

# 运行
import asyncio
asyncio.run(run())

config注意事项

config 参数是运行时的，不会改变链本身的结构，因此同一个链可以用不同的配置重复调用。
如果在链构建时已经通过 .with_config() 设置了默认配置，运行时传入的配置会合并（运行时优先级更高）。
configurable 字段除了用于记忆，还可以用于自定义 Runnable 内部的动态行为（例如根据 configurable["model_name"] 动态切换模型）。

链的构建

创建方式

LCEL语法：最常用方式，使用|操作符连接组件
Runnable组合：通过RunnableSequence、RunnableParallel等高级组合器
Legacy Chains：旧版链类（如LLMChain、SequentialChain）

基本使用

基础链 (Basic Chain)

一个最简单的链，将提示词模板（PromptTemplate） 和模型（Model） 串联起来

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_template("请用中文简洁介绍：{topic}")
model = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

# 使用 | 操作符组合成链，构建一个可运行对象，输入 → Prompt → LLM → 解析
chain = prompt | model | parser

# 调用链
result = chain.invoke({"topic": "量子计算"})
print(result)  # 输出量子计算的简介

顺序链 (Sequential Chain)

将多个链串联起来，前一个的输出作为后一个的输入，形成多步骤任务。

from langchain_core.output_parsers import StrOutputParser

# 第一个链：生成笑话
joke_chain = prompt | model

# 第二个链：翻译笑话
translation_prompt = ChatPromptTemplate.from_template("Translate the following joke into Spanish:\n\n{joke}")
translation_chain = translation_prompt | model | StrOutputParser()

# 组合成一个顺序链
full_chain = joke_chain | (lambda x: {"joke": x.content}) | translation_chain

full_chain.invoke({"topic": "dogs"})

并行链 (Parallel Chain)

使用 RunnableParallel 可以并发执行多个任务，并将结果组合成一个字典。

from langchain_core.runnables import RunnableParallel

chain1 = prompt | model
chain2 = translation_prompt | model

parallel_chain = RunnableParallel(joke=chain1, translation=chain2)
# 也可以使用字典字面量语法: parallel_chain = {"joke": chain1, "translation": chain2}
result = parallel_chain.invoke({"topic": "cats"})

进阶技巧

自定义链

当内置功能无法满足需求时，LangChain 提供了两种自定义链的方式：

使用 LCEL实现 (推荐)

这种方式更简洁，并能自动获得 stream, batch 等全部功能。

from langchain_core.runnables import RunnableLambda

def custom_logic(x):
    # 在这里实现你的任意Python逻辑
    return x["input_text"].upper()

# 使用 RunnableLambda 将普通函数包装成 Runnable 组件
custom_step = RunnableLambda(custom_logic)
chain = custom_step | model | StrOutputParser()

继承 Chain 基类 (旧方式)

当需要深度集成 LangChain 的回调和配置系统时使用。

from langchain.chains.base import Chain

class MyCustomChain(Chain):
    prompt: BasePromptTemplate
    llm: BaseLanguageModel

    @property
    def input_keys(self) -> List[str]:
        return self.prompt.input_variables

    @property
    def output_keys(self) -> List[str]:
        return ["text"]

    def _call(self, inputs: Dict[str, Any]) -> Dict[str, str]:
        prompt_value = self.prompt.format_prompt(**inputs)
        response = self.llm.generate_prompt([prompt_value])
        return {"text": response.generations[0][0].text}

流式输出

# 同步流式输出
for chunk in chain.stream({"product": "智能手表"}):
    print(chunk, end="", flush=True)

# 异步流式输出
async for chunk in chain.astream({"product": "智能手表"}):
    print(chunk, end="", flush=True)

异步支持

对于 I/O 密集型操作（如 API 调用），使用异步版本 ainvoke, abatch, astream 可以显著提升应用并发性能。

1 2	# 在 async 函数内部 response = await chain.ainvoke({"topic": "cats"})

异步调用

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
import asyncio

async def process_multiple():
    # 创建异步链
    llm = ChatOpenAI(model="gpt-3.5-turbo")
    prompt = ChatPromptTemplate.from_template("总结: {text}")
    chain = prompt | llm | StrOutputParser()
    
    # 异步调用
    result1 = await chain.ainvoke({"text": "第一段内容..."})
    
    # 异步批量处理
    inputs = [
        {"text": "内容1..."},
        {"text": "内容2..."},
        {"text": "内容3..."}
    ]
    results = await chain.abatch(inputs, config={"max_concurrency": 3})
    
    # 异步流式处理
    async for chunk in chain.astream({"text": "长内容..."}):
        print(chunk, end="", flush=True)
    
    return results

# 运行
asyncio.run(process_multiple())

异步并行处理

from langchain_core.runnables import RunnableParallel

# 定义两个独立的异步任务
parallel_chain = RunnableParallel(
    summary=prompt | llm | parser,
    keywords=ChatPromptTemplate.from_template("提取{topic}的关键词") | llm | parser
)

# 两个 LLM 调用并发执行
result = await parallel_chain.ainvoke({"topic": "神经网络"})
# result: {'summary': '...', 'keywords': '...'}

异步并发执行

ainvoke 是 LangChain 链的异步调用方法，比同步 invoke 更适合高并发场景。
asyncio.gather 让所有请求真正并发执行（而非顺序等待），从而显著减少总耗时。

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
import asyncio  # 导入异步IO库，用于支持异步编程

prompt = ChatPromptTemplate.from_template("介绍{topic}")
llm = ChatOpenAI()
parser = StrOutputParser()
chain = prompt | llm | parser

async def main():
    # 创建一个任务列表，每个任务是对 chain.ainvoke 的异步调用
    # chain 是一个已定义好的 LCEL 链（例如 prompt | llm | parser）
    # ainvoke 是链的异步方法，接收一个字典参数 {"topic": t}，返回一个协程对象
    tasks = [chain.ainvoke({"topic": t}) for t in ["AI", "区块链", "元宇宙"]]

    # asyncio.gather 并发执行所有任务，并等待所有任务完成
    # *tasks 将列表解包为多个参数，gather 返回一个包含所有任务结果的列表（顺序与输入一致）
    results = await asyncio.gather(*tasks)

    # 打印所有结果
    print(results)

# 运行异步主函数
# asyncio.run 会创建事件循环，执行 main() 协程，并在完成后关闭循环
asyncio.run(main())

调试与追踪

全局调试

设置全局调试标志，打印所有链的详细输入输出信息。

from langchain.globals import set_debug, set_verbose

set_debug(True)   # 最详细日志
set_verbose(True) # 基础日志

LangSmith：推荐使用 LangSmith 平台进行全面的测试、监控和评估。

插入调试步骤

from langchain_core.runnables import RunnableLambda

def debug_step(x):
    print(f"[DEBUG] 当前数据: {x}")
    return x

chain = prompt | RunnableLambda(debug_step) | llm | parser

LangSmith 追踪

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls_xxx"
os.environ["LANGCHAIN_PROJECT"] = "my_project"

链的序列化

LangChain 的 Serializable 对象（包括大多数 LCEL 链）可以很方便地进行序列化和反序列化，以便保存、加载和分享。

若需要完整保存配置与链，建议将链的配置（如 prompt 模板、模型参数）存储为字典，再重新构建。

LangChain 主要通过 langchain_core.load 模块中的函数来实现对象的保存与加载。

LangChain 官方提供的四个核心函数如下：

四个核心方法

函数	功能	核心参数
`dumpd`	将对象序列化为一个 Python 字典，便于后续处理或转为JSON。	`obj`: 要序列化的LangChain对象
`dumps`	将对象直接序列化为一个 JSON 字符串，适合保存到文件或网络传输。	`obj`, `pretty` (可选)
`load`	从一个 Python 字典中反序列化，重建LangChain对象。	`obj` (字典), `secrets_map` (可选)
`loads`	从一个 JSON 字符串中反序列化，重建LangChain对象。	`s` (字符串), `secrets_map` (可选)

方法使用示例

函数	方法示例	用途
转换为JSON字符串	`from langchain_core.load import dumps` `json_str = dumps(chain, pretty=True)`	用于保存到文件或网络传输。
转换为Python字典	`from langchain_core.load import dumpd` `dict_repr = dumpd(chain)`	可用于调试或在JSON不直接支持的场景下使用。
从JSON字符串加载	`from langchain_core.load import loads` `chain = loads(json_str, secrets_map={...})`	反序列化的对应方法。
从字典加载	`from langchain_core.load import load` `chain = load(dict_repr, secrets_map={...})`	从字典重建对象。

完整代码示例

import json
from langchain_core.load import dumpd, dumps, load, loads
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# 1. 创建链
prompt = ChatPromptTemplate.from_messages([
    ("system", "Translate the following into {language}:"),
    ("user", "{text}"),
])
llm = ChatOpenAI(model="gpt-4o-mini", api_key="your-api-key") # 注意这个API Key
chain = prompt | llm

# 2. 序列化：对象 -> JSON 字符串 (API Key不会包含在内)
json_str = dumps(chain, pretty=True) # pretty=True 让JSON更易读
print("序列化后的JSON字符串片段:\n", json_str[:200])
# 你也可以选择先转为字典，再手动转为JSON字符串
# dict_repr = dumpd(chain)

# 3. 保存到磁盘
with open("my_chain.json", "w") as f:
    json.dump(json_str, f) # 将JSON字符串保存到文件

# 4. 从磁盘加载并反序列化
with open("my_chain.json", "r") as f:
    loaded_json_str = json.load(f)

# 5. 反序列化：JSON字符串 -> 对象，并注入API Key
loaded_chain = loads(
    loaded_json_str,
    secrets_map={"OPENAI_API_KEY": "your-api-key"} # 注入秘密信息
)

# 6. 验证反序列化的对象可以使用
response = loaded_chain.invoke({"language": "Chinese", "text": "Hello, world!"})
print("\n反序列化后的链调用结果:\n", response)

注意：反序列化时会检查导入路径的安全性，以防止恶意代码执行。

LangChain 的安全性是其设计的核心，尤其是在反序列化环节，提供了多层保护。

分离秘密信息（Secrets）
这是最重要的安全特性。当序列化一个对象（如包含了API Key的ChatOpenAI模型）时，API密钥这类敏感信息不会被写入到生成的JSON字符串或字典中。
对象的.lc_secrets属性定义了哪些字段是秘密。反序列化时，你需要通过secrets_map参数显式地提供这些秘密，从而安全地将它们注入回新创建的对象中。
基于白名单的类加载
反序列化过程并非随意实例化任何类，而是使用一个白名单（allowlist） 机制。load和loads函数默认只允许实例化白名单中的LangChain核心类（如ChatPromptTemplate, AIMessage, Document等）。任何不在白名单中的类都会被拒绝，有效防止了恶意类的加载。
序列化注入防护
LangChain的序列化机制能抵御注入攻击。如果一个普通字典意外地包含了'lc'这个内部键，在序列化时，这个字典会被“转义”（escaped），包裹在一个特殊的{"__lc_escaped__": {...}}结构中。这确保了该字典在反序列化时被当作普通数据，而不是LangChain对象来实例化，从而阻断了攻击路径。

序列化/反序列化最佳实践：

仅处理可信数据：永远不要对不受信任的输入使用load或loads函数，反序列化是一个Beta版功能，存在潜在风险。
最小化权限：使用最严格的白名单配置，尽量提供一个明确的类列表，而非使用'core'或'all'。
安全地注入秘密：将secrets_from_env保持为默认的False，并通过secrets_map精确提供所需秘密。

添加记忆

记忆组件（Memory）

内存：ChatMessageHistory（测试用）
Redis：RedisChatMessageHistory
PostgreSQL：PostgresChatMessageHistory

RunnableWithMessageHistory(LCEL)

# 导入 LangChain 的核心历史链包装器，用于给普通链添加对话记忆功能
from langchain_core.runnables.history import RunnableWithMessageHistory

# 导入内存版聊天消息历史存储（适合测试/开发环境，生产环境可用 RedisChatMessageHistory 等）
from langchain_community.chat_message_histories import ChatMessageHistory

# 导入提示模板相关类
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 创建一个简单的字典作为会话历史存储容器
# 键为 session_id（会话ID），值为 ChatMessageHistory 对象
store = {}

def get_session_history(session_id: str):
    """根据 session_id 获取对应的聊天历史对象（若不存在则自动创建）"""
    if session_id not in store:
        # 创建新的内存历史记录对象
        store[session_id] = ChatMessageHistory()
    return store[session_id]

# 构建带有历史消息占位符的对话提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个助手"),                     # 系统角色设定
    MessagesPlaceholder(variable_name="history"),  # 历史消息占位符，运行时会被实际对话历史填充
    ("human", "{input}")                           # 用户当前输入
])

# 基础链：提示模板 -> 语言模型 -> 字符串输出解析器
# 注意：llm 和 StrOutputParser 需要提前定义（例如 llm = ChatOpenAI()，parser = StrOutputParser()）
base_chain = prompt | llm | StrOutputParser()

# 使用 RunnableWithMessageHistory 包装基础链，使其具备对话记忆能力
chain_with_memory = RunnableWithMessageHistory(
    base_chain,                     # 待包装的基础链
    get_session_history,            # 会话历史获取函数
    input_messages_key="input",     # 用户输入字段在输入字典中的键名（对应 prompt 中的 "{input}"）
    history_messages_key="history"  # 历史消息字段在输入字典中的键名（对应 MessagesPlaceholder 的 variable_name）
)

# 配置调用参数：通过 configurable 指定 session_id，实现不同会话之间的记忆隔离
config = {"configurable": {"session_id": "user_123"}}

# 第一次对话：用户输入 "我叫小明"
# 此时历史为空，LLM 根据系统提示和当前输入生成回复
response1 = chain_with_memory.invoke({"input": "我叫小明"}, config)

# 第二次对话：用户输入 "我叫什么名字？"
# RunnableWithMessageHistory 会自动从 store 中加载 session_id="user_123" 的历史消息，
# 并将其填充到 prompt 的 {history} 占位符中，因此 LLM 能记住之前用户说过的名字
response2 = chain_with_memory.invoke({"input": "我叫什么名字？"}, config)  # 会记住

关键点说明：

**MessagesPlaceholder**：在 prompt 中预留一个位置，运行时由 RunnableWithMessageHistory 自动填入历史对话消息列表。
**RunnableWithMessageHistory**：负责在每次调用前从 get_session_history 获取历史记录，合并到输入字典中；调用后将新的对话对（用户输入 + AI 回复）自动保存回历史。
**config 中的 session_id**：用于区分不同用户/会话，实现记忆隔离。同一 session_id 共享历史，不同 session_id 互不干扰。

ConversationChain(旧)

记忆（Memory）让链拥有“状态”，能够记住之前的对话内容，是实现多轮对话的关键。

# 导入 LangChain 旧版记忆组件：缓冲区记忆（存储完整对话历史）
from langchain.memory import ConversationBufferMemory

# 导入旧版对话链（Legacy ConversationChain）
from langchain.chains import ConversationChain

# 1. 创建记忆组件
# return_messages=True 表示将历史消息存储为 Message 对象列表（而非字符串），便于与提示模板兼容
memory = ConversationBufferMemory(return_messages=True)

# 2. 将记忆整合到链中
# ConversationChain 是 LangChain 早期版本提供的通用对话链，内部封装了提示模板、LLM 和记忆的自动管理
conversation = ConversationChain(
    llm=model,      # 语言模型实例（需提前定义，例如 model = ChatOpenAI()）
    memory=memory,  # 注入记忆组件
    verbose=True    # 打印执行过程的详细日志，便于调试
)

# 3. 使用对话链
# predict 方法接收用户输入，自动从 memory 加载历史，调用 LLM，并将新的对话对保存回 memory
conversation.predict(input="Hi, I'm Bob.")      # 第一轮：用户介绍名字
conversation.predict(input="What's my name?")   # 第二轮：模型会记住用户叫 Bob

LangChain 提供了多种记忆类型以适应不同场景：

ConversationBufferMemory：最常用，存储所有历史消息，适合短对话。
ConversationBufferWindowMemory：限制上下文窗口，只保留最近 K 轮对话，避免内存溢出。
ConversationSummaryMemory：总结历史，使用 LLM 将对话历史压缩为摘要，适合处理非常长的对话。
ConversationEntityMemory：提取实体信息，记忆对话中特定实体（如人物、地点）的信息，适合需要个性化回答的场景。

实现方式的差异

对比维度	旧方式（`ConversationChain`）	新方式（`RunnableWithMessageHistory` + LCEL）
语法风格	封装好的“大而全”类，内部隐藏细节	基于 LCEL 的显式管道，每个组件清晰可见
记忆存储	通过 `ConversationBufferMemory` 自动管理	用户自定义 `get_session_history` 函数，支持任意后端（内存/Redis/DB）
提示模板	内置固定模板（可覆盖但不灵活）	完全自定义模板，支持 `MessagesPlaceholder` 显式控制历史位置
会话隔离	每个 `ConversationChain` 实例默认一个记忆，多会话需手动创建多个实例	通过 `configurable` 中的 `session_id` 轻松实现多会话隔离，无需多实例
流式输出	不支持（`predict` 仅返回完整结果）	原生支持 `.stream()` 逐 token 输出
异步支持	支持 `apredict`，但并发控制较弱	原生支持 `ainvoke`、`abatch`，性能更好
调试观测	可设置 `verbose=True` 打印日志	可集成 LangSmith、插入 `RunnableLambda` 调试，更细粒度
序列化	序列化能力弱，难以保存/加载	支持 `dumpd` / `load`，便于分享和部署

两种实现方式总结

ConversationChain 是 LangChain 早期版本的“一站式”记忆对话方案，使用简单但扩展性较差，目前已进入维护模式，不推荐用于新项目。
RunnableWithMessageHistory 是基于 LCEL 的现代替代方案，虽然代码量稍多，但提供了完全的控制权、更好的性能和与 LangChain 生态（流式、异步、LangSmith）的深度集成。
两者在概念上等效：都是将提示 + LLM + 记忆组合起来。但 RunnableWithMessageHistory 可以看作是对 ConversationChain 内部机制的解构和重构，使其更符合函数式/管道式编程范式。

最佳实践

**使用 LCEL 而非旧式 LLMChain**：更现代、类型安全、支持流式。

启用缓存：减少重复 LLM 调用。

1
2
3

from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache
set_llm_cache(InMemoryCache())

控制记忆长度：避免 token 爆炸，可使用 trim_messages 裁剪历史。
生产环境使用异步：ainvoke / abatch 提升并发性能。
集成 LangSmith：全链路追踪与调试。

LangChain：链的核心方法、构建使用、高级用法

http://blog.gxitsky.com/2026/04/15/AI-LangChain-038-Chain-Use-Method/

作者

光星

发布于

2026-04-15

更新于

2026-04-19

LangChain：链的核心方法、构建使用、高级用法

核心方法

方法解析

调用方式

配置参数

configurable

输入参数

config使用示例

示例1：为链添加追踪信息

示例2：传递会话 ID

示例3：批量处理与并发控制

示例4：限制链的递归深度

示例5：在异步中使用config

config注意事项

链的构建

创建方式

基本使用

基础链 (Basic Chain)

顺序链 (Sequential Chain)

并行链 (Parallel Chain)

进阶技巧

自定义链

使用 LCEL实现 (推荐)

继承 Chain 基类 (旧方式)

流式输出

异步支持

异步调用

异步并行处理

异步并发执行

调试与追踪

全局调试

插入调试步骤

LangSmith 追踪

链的序列化

四个核心方法

方法使用示例

完整代码示例

添加记忆

记忆组件（Memory）

RunnableWithMessageHistory(LCEL)

ConversationChain(旧)

实现方式的差异

两种实现方式总结

最佳实践

作者

发布于

更新于

许可协议

喜欢这篇文章？打赏一下作者吧

评论

目录