RunnableConfig

发表于 | 更新于 | 分类于 ， | 阅读 次
RunnableConfig 运行时传参：两层键、ContextVar 合并传播、configurable_fields / alternatives、同链多租户切模型。

同一条链在开发环境要打 trace、在生产要换模型名、在多租户里要带 tenant_id。若把这些写进构造器，每次改环境都要重建对象。RunnableConfig 是 invoke 的第二参：业务数据走 input，租户 / 模型旋钮 / 观测钩子走 config，链的输入输出类型不变。

段末注释：RunnableConfig = 单次 Runnable 调用的运行时配置字典（tags、metadata、callbacks、configurable 等）。

社区方案即官方 langchain_core.runnables.config.RunnableConfig（TypedDict）。适用「一条链、多种运行时旋钮」；风险是 configurable 键拼错时静默不生效，以及把密钥、个人身份信息（Personally Identifiable Information，PII）写进会进 trace 的字段。

段末注释：PII = 能直接或间接识别自然人的数据，如用户 id、邮箱；不要放进 metadata / 可被抄进 trace 的 configurable。

1. 一句话定位

维度 内容
角色 能力层运行时配置：观测、可配置字段、并发与递归上限
输入 → 输出 不改变链 I/O 类型；改变内部行为与 tracing
典型调用入口 chain.invoke(x, config={...})、with_config、configurable_fields
与 LangGraph 图的 config["configurable"] 同源；thread_id 等检查点键在图侧消费

出现背景：旧 Chain 把 verbose、callbacks、模型名散落在构造器。Runnable 协议把「这次怎么跑」收成 invoke(input, config) 第二参；TypedDict 设 total=False，允许部分键、再按规则合并。

异步入口相同：await chain.ainvoke(x, config={...})。

2. 实现逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
1. 组装链（与 config 无关）：prompt | model | parser
2. 可选：model.configurable_fields(...) 或 configurable_alternatives(...)
3. chain.invoke(
       {"q": "..."},                          # 业务输入
       config={
           "configurable": {"llm": "full"},   # 行为：换实现 / 换字段
           "tags": ["prod"],                  # 观测：过滤
           "metadata": {"tenant": "t1"},      # 观测：维度
           "callbacks": [handler],            # 观测：钩子
           "max_concurrency": 8,              # 行为：batch 并行
       },
   )
4. ensure_config 补默认键 → 写入 ContextVar
5. 子步骤 ensure_config / merge_configs：tags 并集，configurable 浅合并
6. 输出类型仍由链决定（如 str）；trace 上多出 tags / metadata

字段级变形：

1
2
3
4
input  {"q": "发票错了"}     → 仍是这条 dict，链尾仍是 str
config {"tags":["prod"], "configurable":{"llm":"full"}, "metadata":{"tenant":"t1"}}
        ↓ 随管道下行，不进入 prompt 变量
LangSmith / callback 看到 tags、metadata；model 读到 configurable["llm"]

图 1　config 是并行信封：input 走管道变 str，config 不改变 I/O 类型（对应上表第 3–6 步）

3. 原理说明

3.1 两层键：观测 vs 行为

RunnableConfig（数据类，TypedDict, total=False）
功能：一次 invoke 的观测与行为参数。合法键由 CONFIG_KEYS 限定；键表外的项会被塞进 configurable（历史兼容），不要依赖这个旁路。
默认值 / 最小维度：

键 类型 默认值 最小维度
tags list[str] [] 0 条合法
metadata dict {} 值须 JSON 可序列化
callbacks list / None None 0=无本地 handler
run_name str / None None 非空才覆盖默认名
run_id UUID / None 每步新建 不要手填子步骤
configurable dict {} 键 = ConfigurableField.id
max_concurrency int / None None ≥1 才限制并行
recursion_limit int 25 ≥1

空 config {} 合法。观测层（tags/metadata/callbacks）不应改变模型输出；行为层（configurable 等）可以。把工单正文放进 configurable 是用错层——那是 input。

传播规则：tags 并集去重；metadata 浅合并；callbacks 列表拼接；configurable 浅合并且后写覆盖；run_id 不传给子步骤（每步新建）。

3.2 传播：ContextVar + ensure_config + merge_configs

ensure_config(config=None)（函数）
功能：把 None / 残缺 dict 补成完整 RunnableConfig。默认 tags=[]、metadata={}、configurable={}、recursion_limit=25。最小输入：None。

1
2
from langchain_core.runnables.config import ensure_config
ensure_config(None)["recursion_limit"]  # 25

merge_configs(*configs)（函数）
功能：合并父子 config。tags 并集去重；configurable 后写覆盖；run_id 不向子步骤继承。参数均可 None。

一次 invoke 里 config 不是每个子步骤重新传入的普通 kwargs，而是：

1
2
3
4
5
1. ensure_config(用户 config)
   缺省：tags=[]、metadata={}、configurable={}、recursion_limit=25
2. _set_config_context → 写入 var_child_runnable_config（contextvars.ContextVar）
3. 子 Runnable.invoke 即使不写 config，ensure_config(None) 仍能从 ContextVar 读到父配置
4. 子步骤若自带 tags / configurable，走 merge_configs，规则见 3.1 表

段末注释：ContextVar = Python 按任务/协程隔离的上下文变量；LangChain 用它在调用栈上隐式传递 config，避免每层手写转发。

合并是合并不是覆盖：父 tags=["prod"]、子再加 retriever → ["prod","retriever"]（实现里会排序去重）。configurable 则是 {**父, **子}，同名键以靠近叶子的为准。

图 2　父 config 经 ContextVar 下行：tags 并集、configurable 后写覆盖（对应 §3.2）

因此：同线程、同一次 invoke 内，RunnableLambda 里再调 sub.invoke(x) 常常仍能读到父 config。新线程、进程、或显式传入 config={} 空字典会切断或冲掉继承——这是 trace 断树的主因。

3.3 四条改配置的路

with_config(...)（方法）
功能：返回新 Runnable，预置 tags / metadata / configurable。与单次 invoke(..., config=) 再 merge。至少 1 个非空键。

1
2
prod = chain.with_config(tags=["prod"])
prod.invoke(x, config={"tags": ["canary"]})  # tags 并集

ConfigurableField(id, name=None, description=None)（数据类）
id: str 必填、非空；config["configurable"] 必须用同一 id。

configurable_fields(**field_map)（方法）
功能：把构造参数暴露成运行时键。至少 1 个 ConfigurableField。

1
2
3
from langchain_core.runnables import ConfigurableField
llm.configurable_fields(max_tokens=ConfigurableField(id="max_tokens"))
llm.invoke(msg, config={"configurable": {"max_tokens": 64}})

configurable_alternatives(which, **alternatives)（方法）
功能：整段 Runnable 换实现。至少 1 个 alternative + 默认 self。

手段 生效时机 改什么 适用
构造器 / bind() 组装链时 模型 kwargs（如 temperature=0） 整条链永远同一组硬参数
with_config(...) 返回新 Runnable 绑一份默认 tags / metadata / configurable 预置「生产链」「租户 A 链」
configurable_fields 每次 invoke 的 configurable 同一类的字段值（如 max_tokens） 同厂商同接口，只拧数字或模型名
configurable_alternatives 每次 invoke 的 configurable 整段 Runnable 换实现 mini / full、不同厂商

with_config 与 invoke(..., config=) 会再走 merge_configs：预置 tags 不会被单次调用的 tags 清掉，而是并集。要「完全换一套」只能绑另一条 with_config，或不要预置、全部放进单次 config。

configurable 的键必须等于 ConfigurableField(id=...) 的 id。id 写成 "llm"、调用写成 "model"，会静默走默认实现（见 §5 案例的错误键）。

标量 configurable（str / int / float / bool，且非 __ 前缀）会被抄进 trace 的 metadata；api_key 在排除名单里，其它 token 字段不会自动排除。

3.4 callbacks 与 Lambda 签名

BaseCallbackHandler（类）
功能：实现 on_chain_start/end、on_llm_start/end 等钩子。无必填字段。最小实现：继承后覆盖 ≥1 个钩子。handler 内避免阻塞，勿再 invoke 同链。

子 Runnable 默认继承父 callbacks。RunnableLambda 第二参标注 RunnableConfig 时框架会传入；def f(x) 则纯函数读不到（下游 Runnable 仍可能经 ContextVar 继承）。

1
2
3
4
class LogHandler(BaseCallbackHandler):
    def on_chain_start(self, serialized, inputs, **kwargs):
        print("start", type(inputs).__name__)
chain.invoke(x, config={"callbacks": [LogHandler()]})

3.5 与环境变量

LANGCHAIN_TRACING_V2=true 加 API Key 是进程级打开 LangSmith；单次 invoke 仍用 config 覆盖 run_name / tags / metadata。没有 LangSmith 时，callbacks 照样触发本地 handler。

4. 最小可运行示例

1
pip install -U langchain-openai
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from langchain_openai import ChatOpenAI
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

class LogHandler(BaseCallbackHandler):
    def on_chain_start(self, serialized, inputs, **kwargs):
        print("chain_start", kwargs.get("tags"))

model = ChatOpenAI(
    model="qwen3.5:9b",
    api_key="ollama",
    base_url="http://localhost:11434/v1",
    temperature=0,
)
chain = (
    ChatPromptTemplate.from_template("只答一个词：{x} 的反义词")
    | model
    | StrOutputParser()
)

print(chain.invoke(
    {"x": "hot"},
    config={"tags": ["demo"], "metadata": {"tenant": "u1"}, "callbacks": [LogHandler()]},
))
# 预期：多次 chain_start（每步一次），tags 含 demo；答案形态为 cold 一类单词

5. 案例：同一条链，按租户切模型

教学场景：客服草稿只有一条链；租户 A 默认走 qwen3.5:9b，租户 B 在 configurable 里切到 llama3.1:8b；每次带上 tenant，便于过滤 trace。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import ConfigurableField, RunnableConfig, RunnableLambda

def ollama_chat(model: str) -> ChatOpenAI:
    """指向本地 Ollama 的 ChatOpenAI。输入为模型名，输出为客户端。"""
    return ChatOpenAI(
        model=model, api_key="ollama",
        base_url="http://localhost:11434/v1", temperature=0,
    )

qwen = ollama_chat("qwen3.5:9b")
llama = ollama_chat("llama3.1:8b")
model = qwen.configurable_alternatives(
    ConfigurableField(id="llm"),
    default_key="qwen",
    llama=llama,
)

def stamp(text: str, config: RunnableConfig) -> str:
    """把 tenant 戳到回复前。输入为模型文本与当前 config，输出为带前缀的字符串。"""
    tenant = (config.get("metadata") or {}).get("tenant", "?")
    return f"{tenant}|{text}"

chain = (
    ChatPromptTemplate.from_template("用一句话回复工单，不要标题。工单：{x}")
    | model
    | StrOutputParser()
    | RunnableLambda(stamp)
)

print(chain.invoke(
    {"x": "发票抬头开错了"},
    config={"metadata": {"tenant": "a"}, "tags": ["tenant:a"]},
))
# 预期形态：a|…（默认 qwen）

print(chain.invoke(
    {"x": "发票抬头开错了"},
    config={
        "configurable": {"llm": "llama"},
        "metadata": {"tenant": "b"},
        "tags": ["tenant:b", "env:prod"],
    },
))
# 预期形态：b|…（llama 实现；措辞与上一行通常不同）

print(chain.invoke(
    {"x": "发票抬头开错了"},
    config={"configurable": {"model": "llama"}, "metadata": {"tenant": "c"}},
))
# 预期：仍走 qwen（id 是 llm 不是 model，静默默认）

with_config 预置生产标签时，单次 tags 会并入而不是替换：

1
2
3
prod = chain.with_config(tags=["prod"])
print(prod.invoke({"x": "无法登录"}, config={"tags": ["call"], "metadata": {"tenant": "c"}}))
# tags 同时含 prod 与 call（并集）；stamp 读到 tenant=c

同一套调用点可把 qwen / llama 换成两个云厂商客户端。

重要配置参数

参数（API 名） 类型 / 默认值 功能说明 作用与影响 参考起点 配置指导
configurable dict，默认 {}，invoke config 覆盖 configurable_fields / alternatives 已声明的运行时字段 键拼错静默走默认，无报错 {"llm": "llama"} 键必须等于 ConfigurableField.id
tags list[str]，默认 [] 给本次 run 打标签，供 LangSmith/callback 过滤 与子步骤 并集，不会被单次调用清掉 ["env:prod"] 环境、版本分开打
metadata dict，默认 {} 附加可序列化维度（租户、场景） 会进 trace；勿放密钥/PII tenant_id 只放检索过滤需要的键
callbacks list[Handler]，默认空 挂生命周期钩子（on_llm_end 等） handler 阻塞会拖垮整条链 本地日志 勿在 handler 里再 invoke 本链
run_name str，可选 本步在 trace UI 上的显示名 不替代 tags，只改可读性 业务场景名 调试用短名
max_concurrency int / None 限制 batch 与 Parallel 的同时执行数 过大 429；None 由实现自定 5～20 视 API 限额
recursion_limit int，默认 25 限制链/图递归深度 Agent 环触及上限会停；线性链很少要拧 Agent 环 按工具轮次 ×2 估算
with_config 方法，返回新 Runnable 预置一份默认 tags/metadata/configurable 与单次 config 合并不是覆盖 生产链 预置「租户 A 链」时用

6. 易踩坑

  1. configurable 键与 ConfigurableField.id 不一致：覆盖不生效且无报错，案例里 "model" 仍走 qwen。
  2. metadata 过大或含 PII / 密钥：会进 LangSmith；configurable 里的标量也会被抄进 metadata。
  3. Lambda / 新线程里 sub.invoke(x) 不传 config：ContextVar 跨线程不走，callbacks 断树。签名加 config: RunnableConfig 并向下传。
  4. 以为 invoke(config=) 会清空 with_config 的 tags：实际是并集。
  5. 把业务字段塞进 configurable：那是 input；configurable 只放旋钮。

适用：同链多环境、多租户、运行时换模型或 max_tokens、给 trace 打标签。
不适用：按输入内容做业务分支（那是 RunnableBranch / 图）；持久会话 id 的检查点语义在图编排侧。

小结

  • RunnableConfig 是 invoke 第二参：input 管业务，config 管观测与旋钮，I/O 类型不变。
  • 传播靠 ContextVar + ensure_config + merge_configs：tags 并集，configurable 后写覆盖，run_id 每步新建。
  • configurable_fields 拧同一实现的字段；configurable_alternatives 换整段实现；键必须等于 Field id。
  • 子调用显式传 config（或靠同线程 ContextVar）；新线程必须手传，否则 trace 断树。

参考链接

-------------本文结束感谢您的阅读-------------