# LangChain Runnable 动态替换运行组件：configurable_alternatives 与生产级路由

上一篇讲的是 configurable_fields()：同一个组件不变，只是在运行时覆盖组件里的某些字段，例如 temperature、max_tokens、template。

这一篇换一个更大的粒度：组件本身也可能要在运行时替换。

在生产系统里，这类需求非常常见：

普通用户走便宜模型，企业用户走更强模型。
简单问题走快模型，复杂问题走推理能力更强的模型。
主供应商故障时切到备用供应商。
A/B 实验中切换不同 prompt 版本。
不同知识库场景切换不同 retriever。
某些租户开启 reranker，某些租户关闭 reranker。

这些场景的核心不是“调整参数”，而是“替换链中的某个 Runnable”。LangChain 里对应的能力就是 configurable_alternatives()。

# 它解决的是什么问题

假设你有一条链：

chain = prompt | llm | parser

如果只是调整 llm.temperature，用 configurable_fields() 就够了。

但如果你要把 llm 从一个 OpenAI 模型替换成另一个模型，或者替换成另一个供应商的模型，那么字段覆盖就不够了。你真正要做的是：

chain = prompt | selected_llm | parser

其中 selected_llm 由本次请求的运行时配置决定。

configurable_alternatives() 允许你在构建链时先声明一组候选 Runnable，并给这组候选配置一个运行时选择键。调用链时，只需要通过 config["configurable"] 传入选择结果，LangChain 就会在链内部找到对应组件并继续执行。

# 基本写法

下面用模型切换举例：

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


prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个严谨、简洁的业务分析助手。"),
        ("human", "{query}"),
    ]
)

llm = ChatOpenAI(
    model="gpt-4.1-mini",
    temperature=0.2,
).configurable_alternatives(
    ConfigurableField(id="llm_profile"),
    default_key="fast",
    reasoning=ChatOpenAI(model="gpt-4.1", temperature=0),
)

chain = prompt | llm | StrOutputParser()

answer = chain.invoke(
    {"query": "分析这个投诉工单的核心诉求和处理优先级"},
    config={
        "configurable": {
            "llm_profile": "reasoning",
        }
    },
)

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

这段代码里有三个关键点：

ConfigurableField(id="llm_profile") 定义运行时选择键。
default_key="fast" 定义没有传配置时使用哪个分支。
原始模型就是 fast 默认分支，reasoning=... 定义额外候选项。

当 configurable.llm_profile 等于 reasoning 时，本次调用使用 reasoning 对应的模型。没有传值时，使用 default_key 指向的默认分支。

# 链内部运行流程

configurable_alternatives() 的底层思路很直接：它把候选组件保存成一张 alternatives 映射表，运行时从 config["configurable"] 里读取当前选择 key，再从映射表中取出对应 Runnable 执行。

流程可以理解为：

Runnable configurable_alternatives 运行流程

核心动作如下：

原始 Runnable 声明一个可替换选择键，例如 llm_profile。
声明默认分支，例如 fast。
注册候选 Runnable，例如 fast、reasoning、cheap、backup。
调用链时读取 config["configurable"]["llm_profile"]。
找到对应候选 Runnable。
用被选中的 Runnable 继续执行后续链路。

它不是在运行时拼接一堆 if/else，而是把“可替换点”变成了 Runnable 协议的一部分。

# 和 configurable_fields 的区别

这两个 API 容易混在一起，但它们的边界很清楚：

API	改变什么	适合场景
`configurable_fields()`	改组件字段	动态调整温度、输出长度、提示模板字段、检索 top_k
`configurable_alternatives()`	换整个 Runnable	动态切换模型、Prompt、Retriever、Reranker、Tool Runnable

例如：

llm = ChatOpenAI(model="gpt-4.1-mini").configurable_fields(
    temperature=ConfigurableField(id="llm_temperature"),
)

1
2
3

这是“同一个模型组件，运行时改温度”。

llm = ChatOpenAI(model="gpt-4.1-mini").configurable_alternatives(
    ConfigurableField(id="llm_profile"),
    default_key="fast",
    reasoning=ChatOpenAI(model="gpt-4.1"),
)

1
2
3
4
5

这是“运行时选择另一个模型组件”。

生产里经常两者一起用：先用 alternatives 选择模型档位，再用 fields 调整该档位允许开放的参数。

# Prompt 版本切换

configurable_alternatives() 不只适合模型。Prompt 也是 Runnable，因此也可以做版本切换。

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import ConfigurableField


default_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个客服助手，请直接回答用户问题。"),
        ("human", "{query}"),
    ]
)

json_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个客服助手，只输出 JSON，字段为 answer、risk、next_action。"),
        ("human", "{query}"),
    ]
)

prompt = default_prompt.configurable_alternatives(
    ConfigurableField(id="prompt_variant"),
    default_key="default",
    strict_json=json_prompt,
)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

调用时：

chain.invoke(
    {"query": "客户说订单一直没发货，应该怎么处理？"},
    config={
        "configurable": {
            "prompt_variant": "strict_json",
        }
    },
)

1
2
3
4
5
6
7
8

这比让外部直接传完整 prompt 更安全。外部只能选择后端允许的 prompt_variant，不能随意注入模板正文。

# Retriever 与 RAG 策略切换

RAG 系统里也经常需要替换检索策略：

默认向量检索。
混合检索。
高召回检索。
带权限过滤的租户检索。
灰度 reranker 检索。

可以把不同 retriever 包成候选 Runnable：

from langchain_core.runnables import ConfigurableField


retriever = vector_retriever.configurable_alternatives(
    ConfigurableField(id="retriever_profile"),
    default_key="vector",
    hybrid=hybrid_retriever,
    high_recall=high_recall_retriever,
)

rag_chain = {
    "context": retriever,
    "query": lambda x: x["query"],
} | prompt | llm | parser

1
2
3
4
5
6
7
8
9
10
11
12
13
14

调用时：

answer = rag_chain.invoke(
    {"query": "退款规则有哪些？"},
    config={
        "configurable": {
            "retriever_profile": "hybrid",
        }
    },
)

1
2
3
4
5
6
7
8

这里要注意，检索器切换通常会影响召回质量、延迟、成本和权限边界。生产系统里不应该让用户直接决定 retriever_profile，而应该由服务端根据业务场景、租户权限、实验分组和风控策略生成。

# prefix_keys 适合什么场景

configurable_alternatives() 有一个容易忽略的点：当替换后的候选组件内部还有自己的 configurable 字段时，可能出现配置键冲突。

例如两个候选模型都暴露了 temperature，两个候选 retriever 都暴露了 top_k。如果所有配置都挤在同一层 configurable 里，复杂链很容易冲突。

prefix_keys=True 可以让候选分支的内部配置按分支前缀区分。可以理解成：

config={
    "configurable": {
        "llm_profile": "reasoning",
        "llm_profile==reasoning/output_token_number": 1200,
    }
}

1
2
3
4
5
6

这种写法适合复杂平台型系统，但普通业务链不一定需要。我的建议是：

简单链：用清晰的全局 key，例如 llm_profile、output_token_number。
多候选、多嵌套链：考虑 prefix_keys=True，避免不同分支的内部配置撞名。
平台系统：定义统一配置 schema，不要让调用方自由拼 key。

# 生产级模型路由

真正的生产系统通常不会把 "reasoning"、"fast" 这种选择直接交给用户。更合理的是服务端构造配置：

from dataclasses import dataclass


@dataclass(frozen=True)
class RequestContext:
    tenant_id: str
    user_hash: str
    plan: str
    feature: str
    risk_level: str
    experiment: str | None = None


def select_llm_profile(ctx: RequestContext, query: str) -> str:
    if ctx.risk_level == "high":
        return "reasoning"

    if ctx.plan == "enterprise":
        return "reasoning"

    if len(query) > 800:
        return "reasoning"

    return "fast"


def build_config(ctx: RequestContext, query: str) -> dict:
    return {
        "run_name": f"{ctx.feature}_chain",
        "tags": ["ai", ctx.feature, f"plan:{ctx.plan}"],
        "metadata": {
            "tenant_id": ctx.tenant_id,
            "user_hash": ctx.user_hash,
            "feature": ctx.feature,
            "experiment": ctx.experiment,
        },
        "configurable": {
            "llm_profile": select_llm_profile(ctx, query),
        },
    }

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

调用层保持简单：

answer = chain.invoke(
    {"query": query},
    config=build_config(ctx, query),
)

1
2
3
4

这样做的好处是，业务策略被集中管理，链定义层只关心“有哪些可替换点”，调用层只关心“传业务输入”。

# 容灾切换

configurable_alternatives() 也可以用于供应商容灾：

llm = primary_llm.configurable_alternatives(
    ConfigurableField(id="provider_profile"),
    default_key="primary",
    backup=backup_llm,
)

1
2
3
4
5

但要注意，容灾不是简单换一个模型就结束。生产环境至少还要处理：

不同供应商的消息格式差异。
工具调用能力是否一致。
JSON 输出稳定性是否一致。
token 计费口径是否一致。
内容安全策略是否一致。
trace 和错误码是否统一。
降级后的用户体验是否可接受。

因此，容灾分支最好放在统一模型适配层之后，让上层链看到的是一致的 Runnable 协议。

# 失败行为与默认值

如果运行时传入了未知 key，configurable_alternatives() 会报错。生产系统里不要把这个错误直接暴露给用户。

更稳妥的做法是：

ALLOWED_LLM_PROFILES = {"fast", "reasoning", "backup"}


def normalize_llm_profile(value: str | None) -> str:
    if value in ALLOWED_LLM_PROFILES:
        return value
    return "fast"

1
2
3
4
5
6
7

然后只把归一化后的值放进 configurable。

同时要给默认分支一个明确语义。不要随便叫 default，最好叫：

fast
balanced
reasoning
cheap
primary

名字本身就是配置协议的一部分，后续观测、实验、报表都会用到。

# 可观测性

动态替换组件后，必须能看清“这次到底选了哪个分支”。否则线上排障会很痛苦。

建议把选择结果同步写入 metadata：

def build_config(ctx: RequestContext, query: str) -> dict:
    llm_profile = select_llm_profile(ctx, query)

    return {
        "run_name": f"{ctx.feature}_chain",
        "tags": ["ai", ctx.feature, f"llm:{llm_profile}"],
        "metadata": {
            "tenant_id": ctx.tenant_id,
            "feature": ctx.feature,
            "llm_profile": llm_profile,
            "experiment": ctx.experiment,
        },
        "configurable": {
            "llm_profile": llm_profile,
        },
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

这样在 LangSmith 或自定义 tracing 里就能按 llm_profile 聚合：

不同模型分支的成功率。
平均延迟。
平均成本。
用户满意度。
JSON 解析失败率。
工具调用失败率。

动态替换一旦进入生产，就不是单纯的代码技巧，而是实验平台、成本平台和稳定性治理的一部分。

# 测试策略

最少要覆盖三类测试。

第一类是策略测试：

def test_enterprise_uses_reasoning_profile():
    ctx = RequestContext(
        tenant_id="t1",
        user_hash="usr_test",
        plan="enterprise",
        feature="support_answer",
        risk_level="low",
    )

    assert select_llm_profile(ctx, "普通问题") == "reasoning"

1
2
3
4
5
6
7
8
9
10

第二类是配置测试：

def test_build_config_records_selected_profile():
    ctx = RequestContext(
        tenant_id="t1",
        user_hash="usr_test",
        plan="free",
        feature="support_answer",
        risk_level="low",
    )

    config = build_config(ctx, "普通问题")

    assert config["configurable"]["llm_profile"] == "fast"
    assert config["metadata"]["feature"] == "support_answer"

1
2
3
4
5
6
7
8
9
10
11
12
13

第三类是链路测试：用 fake model 或 stub Runnable 验证不同 key 确实走到不同分支。

from langchain_core.runnables import RunnableLambda


fast = RunnableLambda(lambda x: "fast")
reasoning = RunnableLambda(lambda x: "reasoning")

runnable = fast.configurable_alternatives(
    ConfigurableField(id="profile"),
    default_key="fast",
    reasoning=reasoning,
)

assert runnable.invoke("x") == "fast"
assert runnable.invoke("x", config={"configurable": {"profile": "reasoning"}}) == "reasoning"

1
2
3
4
5
6
7
8
9
10
11
12
13
14

不要只测“能跑通”。动态替换最重要的是选路正确、默认值正确、非法配置不越权。

# 问题

configurable_alternatives() 的主要风险来自“替换粒度太大”。

常见问题包括：

不同候选组件能力不一致，例如一个模型支持工具调用，另一个不支持。
不同模型输出格式差异太大，导致后续 parser 失败。
不同 retriever 权限过滤不一致，造成数据越权。
候选分支命名随意，后续报表和 trace 难以归类。
外部请求可以直接选择高成本模型，导致预算失控。
灰度分支没有观测指标，出问题后不知道哪些请求受影响。
fallback 分支没有经过同等测试，只在事故时第一次被真实流量打到。

所以它不是“动态想换啥就换啥”，而是“提前声明可替换边界，并由服务端策略选择”。

# 拓展

这个能力可以扩展到很多工程场景：

模型路由：fast、balanced、reasoning、backup。
Prompt 版本：default、strict_json、cot_hidden、short_answer。
检索策略：vector、hybrid、high_recall、permission_first。
Reranker：disabled、cheap_rerank、strong_rerank。
工具集：readonly_tools、write_tools、admin_tools。
输出策略：plain_text、json_schema、markdown_report。

它也适合做实验平台底座。实验系统只负责生成受控配置，Runnable 链负责按配置选择组件，观测系统负责记录效果。

# 实际生产是否使用

会使用，而且很适合生产，但通常出现在“内部配置层”，不是直接暴露给用户。

适合使用的生产场景：

多模型路由。
多供应商容灾。
Prompt 灰度发布。
RAG 检索策略灰度。
不同租户差异化配置。
成本和延迟分层。

不适合的场景：

让用户直接传模型名。
让用户直接选择任意 prompt。
没有统一适配层就混用能力差异很大的供应商。
没有 trace 和指标就做动态路由。

生产里真正稳定的写法，是把 configurable_alternatives() 放在链定义层，把选择逻辑放在后端策略层，把结果写进观测层。

# 现在是否抛弃

没有抛弃。

configurable_alternatives() 仍然是当前 LangChain Runnable 体系里用于“运行时替换 Runnable”的正式能力。它和 RunnableConfig.configurable、configurable_fields()、with_config() 属于同一套运行时配置机制。

需要区分的是：旧式 ConversationMemory 类组件在新项目里已经不再是首选；但 Runnable 的动态配置和动态替换能力仍然是 LCEL 工程化的重要部分。新项目可以继续使用它做模型路由、Prompt 版本切换和检索策略切换。

# 最新生产如何实现

最新版生产实现建议按这个结构落地：

第一层，定义可替换点：

llm = fast_llm.configurable_alternatives(
    ConfigurableField(id="llm_profile"),
    default_key="fast",
    reasoning=reasoning_llm,
    backup=backup_llm,
)

1
2
3
4
5
6

第二层，所有候选组件先经过统一适配，保证输入输出协议一致：

chain = prompt | llm | StrOutputParser()

第三层，服务端策略生成配置，而不是客户端直传：

llm_profile = select_llm_profile(ctx, query)

config = {
    "run_name": "support_answer_chain",
    "tags": ["support", f"llm:{llm_profile}"],
    "metadata": {
        "tenant_id": ctx.tenant_id,
        "feature": ctx.feature,
        "llm_profile": llm_profile,
    },
    "configurable": {
        "llm_profile": llm_profile,
    },
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14

第四层，调用链：

answer = chain.invoke(
    {"query": query},
    config=config,
)

1
2
3
4

第五层，用观测数据反向调整策略：

哪个分支成本过高。
哪个分支延迟过高。
哪个分支解析失败率高。
哪个实验组用户满意度更好。
哪个供应商在某些时间段更不稳定。

最终原则是：Runnable 负责可组合，configurable_alternatives() 负责可替换，服务端策略负责可治理，观测系统负责可验证。这样动态替换才不是炫技，而是能长期运行的生产能力。

← LangChain Runnable 配置运行时链内部：configurable_fields 与生产级参数治理 LangChain Runnable 重试与回退机制：withretry、withfallbacks 与生产级容错 →