【LangChain系列九】Runnable 接口与可配置模型：LangChain 的底层协议

前一篇聊了链式编程，用 | 把模型、提示词、解析器串起来。但你有没有想过，为什么这些不同类型的组件能用 | 连接？

答案就是 Runnable 接口。

1. Runnable 是什么

Runnable 是 LangChain 定义的一套统一接口协议。凡是实现了这个接口的组件（模型、提示词、解析器、检索器……），都能用相同的方式调用。

这意味着你不需要记住每个组件各自的调用方式，一套方法走天下。

2. 核心方法一览

Runnable 接口提供了四个主要方法：

方法	说明	场景
invoke	同步调用，处理单个输入	最常用，日常开发
stream	同步流式输出	需要逐 token 输出时
batch	批量处理多个输入	需要并行处理时
astream	异步流式输出	异步场景

下面逐个看。

invoke：同步调用

from langchain_openai import ChatOpenAI

model = ChatOpenAI()
result = model.invoke("你好")
print(result.content)

invoke 接收一个输入，同步返回结果。这是最基础的调用方式。

它的完整签名是：

invoke(input, config=None, **kwargs)

• input — 输入数据，类型取决于组件（字符串、消息列表、字典等）
• config — 配置参数，比如运行时覆盖模型参数
• **kwargs — 其他参数

stream：流式输出

for chunk in model.stream("写一首关于春天的诗"):
    print(chunk.content, end="", flush=True)

stream 返回一个迭代器，每个元素是一个 token chunk。适合需要实时显示输出的场景，比如聊天界面。

batch：批量处理

inputs = ["你好", "今天天气怎么样？", "什么是机器学习？"]
results = model.batch(inputs)
for r in results:
    print(r.content)

batch 接收一个输入列表，并行处理。比循环调用 invoke 快得多，因为 LangChain 会自动并发。

astream：异步流式

import asyncio

async def main():
    async for chunk in model.astream("写一首诗"):
        print(chunk.content, end="", flush=True)

asyncio.run(main())

异步版本的 stream，适合在异步框架（FastAPI、aiohttp）中使用。

3. 可配置模型：config_model

这是 LangChain 里一个很巧妙的设计。你可以定义一个"可配置"的模型实例，运行时再决定用哪个具体的模型。

from langchain_openai import ChatOpenAI

# 不指定具体模型 → 这是一个"模拟器"
config_model = ChatOpenAI()

注意，不给 model 参数的话，它不会真的连接任何模型。它的作用是作为一个可配置的占位符。

调用时通过 invoke 的 config 参数指定具体模型：

from langchain_core.runnables import ConfigurableField

config_model = ChatOpenAI().configurable_fields(
    model_name=ConfigurableField(
        id="model_name",
        name="模型名称",
        description="要使用的模型名称"
    )
)

# 运行时指定模型
result = config_model.invoke(
    "你好",
    config={"configurable": {"model_name": "gpt-4"}}
)

这种模式在以下场景特别有用：

• A/B 测试：同一套代码，运行时切换不同模型对比效果
• 环境隔离：开发用便宜模型，生产用高质量模型
• 用户选择：让终端用户自己选模型

4. configurable_fields 详解

configurable_fields 方法用来标记哪些参数是可以在运行时修改的：

from langchain_openai import ChatOpenAI
from langchain_core.runnables import ConfigurableField

model = ChatOpenAI(temperature=0.7).configurable_fields(
    temperature=ConfigurableField(
        id="temperature",
        name="温度",
        description="控制输出的随机性，0-2之间"
    ),
    max_tokens=ConfigurableField(
        id="max_tokens",
        name="最大token数",
        description="控制输出的最大长度"
    )
)

# 运行时覆盖参数
result = model.invoke(
    "你好",
    config={"configurable": {"temperature": 0.1, "max_tokens": 100}}
)

5. config_prefix：避免命名冲突

当你的链里有多个可配置组件时，可能会出现参数名冲突。比如两个模型都有 temperature 参数。

config_prefix 就是解决这个问题的：

model1 = ChatOpenAI().configurable_fields(
    model_name=ConfigurableField(id="model_name", name="模型")
).with_config(config_prefix="model1")

model2 = ChatOpenAI().configurable_fields(
    model_name=ConfigurableField(id="model_name", name="模型")
).with_config(config_prefix="model2")

# 调用时用前缀区分
result = model1.invoke(
    "你好",
    config={"configurable": {"model1_model_name": "gpt-4"}}
)

通过前缀，同一个参数名在不同组件之间就不会打架了。

6. 完整实战：可切换模型的翻译链

来个综合例子，做一个可以在运行时切换模型的翻译链：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import ConfigurableField

# 可配置的模型
model = ChatOpenAI().configurable_fields(
    model_name=ConfigurableField(
        id="llm_model",
        name="LLM Model",
        description="选择使用的模型"
    )
)

# 提示词
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业翻译，把{source}翻译成{target}"),
    ("user", "{text}")
])

# 链
chain = prompt | model | StrOutputParser()

# 使用默认模型
result1 = chain.invoke({
    "source": "中文",
    "target": "英文",
    "text": "你好世界"
})

# 运行时切换模型
result2 = chain.invoke(
    {"source": "中文", "target": "英文", "text": "你好世界"},
    config={"configurable": {"llm_model": "gpt-4-turbo"}}
)

同一套链，通过 config 参数就能灵活切换模型，代码结构完全不用改。

7. 常用配置参数速查

参数	类型	说明
model	str	模型名称，如 gpt-4、gpt-3.5-turbo
model_provider	str	模型提供商，如 openai、anthropic
temperature	float	温度参数，控制随机性
max_tokens	int	最大输出 token 数
configurable_fields	method	标记可配置参数
config_prefix	str	参数前缀，避免冲突

8. Runnable 的组合能力

Runnable 真正强大的地方在于组合。通过 | 运算符，任意 Runnable 组件都能串联：

# prompt → model → parser，三个 Runnable 串联
chain = prompt | model | parser

# 也能并联（用 RunnableParallel）
from langchain_core.runnables import RunnableParallel

parallel = RunnableParallel(
    original=prompt | model | StrOutputParser(),
    length=prompt | model | StrOutputParser() | (lambda x: len(x))
)

所有组件共享同一套接口，所以组合起来毫无障碍。这就是面向接口编程的好处。

小结

• Runnable 是 LangChain 的底层协议，统一了所有组件的调用方式
• 四大方法：invoke（同步）、stream（流式）、batch（批量）、astream（异步流式）
• 可配置模型：定义时不确定模型，运行时通过 config 参数指定
• configurable_fields 和 config_prefix：灵活控制运行时参数

理解了 Runnable，你就理解了 LangChain 的骨架。后面不管加什么功能，都是在这个骨架上挂肉。

觉得有帮助的话，点个赞👍收藏⭐支持一下吧！