第 13 章 配置系统与运行时控制
13.1 RunnableConfig 结构
RunnableConfig 是 LangChain/LangGraph 运行时在每一次 invoke/stream/batch 调用中沿着 Runnable 链(图→节点→工具→LLM)传递的配置对象,类型为 TypedDict,定义于 langchain_core.runnables。它既承载框架内置的控制字段(递归深度、标签、元数据、回调、运行名等),也承载用户自定义的业务参数(通过 configurable 子字典)。config 对象在调用链中通过 ContextVar 隐式传播,节点、工具、子图、条件路由函数都可以读取。
RunnableConfig 核心字段:
| 字段 | 类型 | 说明 |
|---|---|---|
| configurable | dict[str, Any] | 用户自定义字段,是最常用的扩展点。thread_id、user_id、model 名、业务参数、系统提示词覆盖等都放入此处 |
| tags | list[str] | 标签字符串列表,用于 LangSmith 追踪过滤与分类 |
| metadata | dict[str, Any] | 任意可 JSON 序列化元数据键值对,用于关联 trace 与业务对象 |
| recursion_limit | int | 图执行最大超步数(Superstep 数),默认 25,防止死循环 |
| run_name | str / None | 本次运行名称,在 LangSmith UI 中展示为 trace 标题 |
| callbacks | Callbacks / None | 回调处理器列表(BaseCallbackHandler 实例),覆盖 tracing、日志、流式事件 |
| run_id | UUID / None | 本次运行的唯一 ID(自动生成,一般不手动设置) |
| max_concurrency | int / None | batch/abatch 调用时的最大并发数,控制对下游服务的压力 |
| configurable/configurable_type | 内部字段 | 配合 pydantic config_schema 做类型校验 |
调用方在 invoke/stream 的第二个参数中传入 config,每一次调用可传入不同的 configurable 字段值,无需重新编译图——这是 config 与 compile 参数的核心区别:compile 参数决定图的静态能力(哪些持久化后端、哪些断点),config 决定单次调用的动态行为(哪个用户、什么参数、哪个模型)。
result = graph.invoke(
input_data,
config={
"configurable": {
"thread_id": "conv-1",
"user_id": "u-123",
"tenant_id": "t-xyz",
"model": "gpt-4o-mini",
"temperature": 0.2,
"system_prompt": "你是一个专业客服。",
},
"tags": ["production", "customer-support", "v2"],
"metadata": {"env": "prod", "trace_id": "abc-def", "ticket_id": "TK-2024-001"},
"recursion_limit": 50,
"run_name": "resolve_ticket",
},
)13.2 thread_id 与 config 的关系
config["configurable"]["thread_id"] 是 Checkpointer 识别执行线程的关键字段,具有以下严格语义:
- 当 compile 时传入了 checkpointer,每次 invoke/stream 必须在 configurable 中提供 thread_id,否则运行时抛
ValueError: thread_id is required when checkpointer is provided; - 相同 thread_id 的多次 invoke 会沿同一条 checkpoint 链表继续执行(多轮对话、断点恢复);
- 不同 thread_id 的执行状态完全隔离,即使由同一张编译后的图处理,State、messages、断点、版本链都互不干扰;
- thread_id 是字符串,无格式强制,推荐使用业务主键组合(见第 11 章);
- thread_id 不会被自动创建或复用,完全由调用方控制——传入新 thread_id 即启动新执行(source="input"),传入已有 thread_id 即续跑。
Store 不依赖 thread_id 做隔离,它使用 namespace tuple 独立管理隔离边界,这也是两个持久化机制的关键区别之一。
13.3 在节点与工具中读取配置
节点与工具读取 config 有两种方式,适用于不同场景。
方式一:通过函数签名第二个参数(推荐用于节点函数)。节点函数签名声明 config: RunnableConfig 参数,Pregel 运行时在调用节点时自动注入当前 config 对象。
from langchain_core.runnables import RunnableConfig
def my_node(state: State, config: RunnableConfig) -> dict:
user_id = config["configurable"].get("user_id")
model_name = config["configurable"].get("model", "gpt-4o-mini")
temperature = float(config["configurable"].get("temperature", 0.0))
recursion_limit = config.get("recursion_limit", 25)
tags = config.get("tags", [])
metadata = config.get("metadata", {})
# 使用这些参数执行业务逻辑
return {"user_id": user_id}方式二:通过 get_config() 函数(推荐用于工具函数与辅助函数)。在节点内部调用的任意深度辅助函数、工具函数中,都可以通过 from langgraph.config import get_config 拿到当前 config,无需在每层函数签名中显式传递 config 参数。
from langgraph.config import get_config
def some_deep_helper():
"""被节点间接调用的辅助函数,也能读取 config。"""
config = get_config()
user_id = config["configurable"].get("user_id")
return user_id工具函数中读取 config 可使用 InjectedToolArg 注解(类似 InjectedState/InjectedStore):
from typing import Annotated
from langchain_core.tools import tool
from langgraph.prebuilt import InjectedToolArg
from langchain_core.runnables import RunnableConfig
@tool
def my_tool(
query: str,
config: Annotated[RunnableConfig, InjectedToolArg], # 由运行时注入,不出现在 schema
) -> str:
"""一个能访问 config 的工具。"""
user_id = config["configurable"].get("user_id")
return f"用户 {user_id} 查询:{query}"两种方式的选择原则:节点主函数使用签名参数(类型安全、编辑器补全);深层辅助函数与工具函数使用 get_config()/InjectedToolArg(避免参数穿透)。
13.4 recursion_limit、tags、metadata、run_name
recursion_limit:图执行的最大超步数(Superstep 数),默认值 25。当执行超步数达到 recursion_limit 仍未走到 END,运行时抛 GraphRecursionError。该参数是防死循环的硬性安全阀,其计数方式是"每个超步 +1"而非"节点调用次数"——一次超步中并发执行的多个节点只算 1。设置经验:
| 图类型 | 推荐 recursion_limit |
|---|---|
| 简单单轮对话(无工具或 1–2 轮工具调用) | 10–15 |
| 标准 ReAct Agent | 25–50 |
| 多步研究/多智能体协作 | 50–100 |
| Map-Reduce 动态扇出 | 按分支数 × 每分支步数 + 余量 |
无论任何场景,recursion_limit 必须保留有限值,禁止设置为无限或极大值(如 10000)——死循环一旦发生会快速消耗 token 与时间。
tags 与 metadata:两者都不影响运行时行为,纯粹服务于可观测性与运维。tags 是扁平字符串列表,适合做粗粒度分类(如 "prod"、"beta"、"customer-support"、"v2"),可在 LangSmith UI 中按 tag 做筛选聚合;metadata 是键值对字典,适合做精确关联(如 user_id、tenant_id、ticket_id、request_id),客诉排查时通过 metadata 字段值在 LangSmith 中反查 trace。生产环境强制每次调用带 metadata 至少包含 user_id 与请求 ID;tags 至少包含环境标识与业务类型。
run_name:为单次运行命名,在 LangSmith 中展示为 trace 标题。建议使用业务动作名称(如 "resolve_ticket"、"draft_email"、"research_topic"),便于按业务场景做延迟/成本/失败率聚合,而不是全部显示为默认的 "LangGraph"。
13.5 configurable 的典型用途
configurable 是运行时动态参数的主要载体,典型场景:
- 动态切换模型:configurable 中传入
model与temperature字段,LLM 节点中根据该字段构造或选择模型实例(如 gpt-4o-mini 用于日常对话、gpt-4o 用于复杂推理、Qwen2.5 用于内部测试)。这种方式避免为每个模型变体编译多张图。 - 传入用户/租户信息:
user_id、role、tenant_id、权限级别等通过 configurable 传入,节点/工具通过 get_config() 读取用于鉴权、Store 命名空间隔离、审计日志。 - 传入业务上下文 ID:工单 ID、订单号、会话 ID、业务链路 trace_id 等不需要在节点间流转(不会被节点更新)但需要在各节点/工具中访问的业务标识。
- 覆盖系统提示词:不同场景传入不同 system_prompt(如客服场景、创作场景、代码审查场景),节点在构造 ChatPromptTemplate 时读取 configurable.system_prompt 填入。
- 传入运行时开关:是否启用流式、是否启用某个可选工具、是否开启详细调试输出、温度参数、top_p 等。
13.6 动态配置 vs State 字段
configurable 与 State 字段在职责上有明确分工,正确区分两者是避免架构混乱的关键。
| 维度 | configurable | State 字段 |
|---|---|---|
| 生命周期 | 单次 invoke/stream 调用级 | 跨超步持久化(进入 checkpoint) |
| 写入/修改 | 调用方 invoke 时传入,节点内只读 | 节点通过返回值/Command.update 写入,自动按 reducer 合并 |
| 节点间流转 | 同一调用链内所有节点/子图/工具自动可见 | 按 BSP 规则在超步间流转,跨 thread 隔离 |
| 是否持久化到 checkpoint | 默认不进入(除非节点主动复制到 State) | 每个超步自动快照到 checkpointer |
| 是否影响路由决策 | 可以(节点读取后做判断),但变更不会被记录 | 核心用途——节点返回值驱动下一超步的条件路由 |
| 适用数据 | 每次调用可变、不需要跨超步记忆的参数(模型名、温度、user_id、业务 ID、system prompt) | 需要在节点间流转、需要恢复、影响下一超步决策的数据(messages、中间结果、轮次计数、路由标志) |
| 节点可否修改 | 不应修改(只读语义) | 通过返回值更新 |
经验法则:需要"记住"并影响后续超步决策的数据放 State;本次调用上下文级别、执行过程中不变的参数放 configurable。典型区分:
- 模型名/温度 → configurable(同一轮调用内固定,不需要跨超步持久化);
- user_id/tenant_id → configurable(调用级上下文标识);
- messages/轮次计数/重试次数/路由标志 → State(需要跨超步流转和持久化);
- 是否完成回答/当前任务状态 → State(驱动条件路由)。
错误使用的表现:把"是否完成"放在 configurable 中,节点修改它不生效,下一轮还是初始值;把模型名放在 State 中,每次 checkpoint 都冗余保存且多轮切换逻辑复杂。
13.7 运行时资源获取工具函数
langgraph.config 模块提供一组基于 Python ContextVar 的工具函数,用于在节点执行上下文内隐式获取运行时资源,避免在函数签名中层层传递:
| 函数 | 返回类型 | 说明 |
|---|---|---|
| get_config() | RunnableConfig | 获取当前 RunnableConfig |
| get_store() | BaseStore / None | 获取 compile 时传入的 Store 实例;未配置返回 None |
| get_writer() / get_stream_writer() | Callable | 获取自定义流式写入器,用于 custom stream mode 向客户端推送自定义事件(进度、中间产物、状态更新) |
| get_context() | PregelTaskContext | 获取完整运行时上下文对象(含 task_id、state、config、writer、store、runtime、checkpoint_ns、subgraphs 等) |
这些函数依赖 Python ContextVar 机制,具有以下特性:在同一执行任务(Task)内自动传播;跨异步任务(asyncio.create_task)在 Python 3.11+ 中通过 TaskGroup 自动复制 context;同步线程池(invoke 路径下包装同步节点)需手动复制 context。Python 3.10 及以下版本在复杂嵌套任务场景中 ContextVar 传播偶发问题,生产建议锁定 Python 3.11+。
自定义流式写入(custom stream mode)示例:
from langgraph.config import get_stream_writer
import time
def long_research_node(state: State) -> dict:
"""长流程节点,通过 custom stream 向客户端推送进度。"""
writer = get_stream_writer()
results = []
sources = ["source_a", "source_b", "source_c"]
for i, src in enumerate(sources):
time.sleep(1) # 模拟 IO
chunk = f"已完成 {i+1}/{len(sources)}:处理 {src}"
writer({"event": "progress", "step": i+1, "total": len(sources), "message": chunk})
results.append(chunk)
return {"research": "\n".join(results)}
# 客户端通过 stream_mode="custom" 或 "messages"|"custom" 列表接收
for event in graph.stream(input, config, stream_mode=["custom", "updates"]):
print(event)自定义事件结构完全由业务定义(必须可 JSON 序列化),典型用途:长时间运行节点的进度条、中间产物预览(如正在读取的文档片段)、审批请求推送、多智能体协作中的 Agent 切换通知。
13.8 子图中的配置传递与 config_schema
子图(Subgraph)是 LangGraph 中构建模块化复杂工作流的核心机制(详见第 16 章)。在配置传递上,默认行为是:父图的 configurable 会自动传递到子图,子图节点调用 get_config() 可以拿到父图传入的全部 configurable 字段。这让 user_id、tenant_id、model 等通用参数无需在父图节点中显式转发。
子图可以定义自己的 config_schema(Pydantic BaseModel),对传入的 configurable 做运行时校验,并为子图专属字段提供默认值。config_schema 同时起到"白名单"作用——若父图传入的 configurable 字段未在子图 config_schema 中声明,则该字段不会自动传入子图(避免未预期字段污染子图)。
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph
class SubgraphConfig(BaseModel):
"""子图专属配置 schema。"""
model: str = Field(default="gpt-4o-mini", description="子图内部使用的模型")
temperature: float = Field(default=0.0, ge=0.0, le=2.0)
search_depth: int = Field(default=3, ge=1, le=10)
# 注意:user_id 等通用字段如需保留,必须显式声明
user_id: str = Field(default="")
class SubState(TypedDict):
query: str
result: str
def search_node(state: SubState, config: RunnableConfig) -> dict:
# config["configurable"] 已按 SubgraphConfig 校验与填充默认值
depth = config["configurable"]["search_depth"]
return {"result": f"searched with depth={depth}"}
sub_builder = StateGraph(SubState, config_schema=SubgraphConfig)
sub_builder.add_node("search", search_node)
sub_builder.add_edge(START, "search")
sub_builder.add_edge("search", END)
sub_graph = sub_builder.compile()子图定义 config_schema 后,父图在 add_node 时传入的 configurable 字段必须符合 schema 定义,否则运行时抛 ValidationError。这为大型项目中子图边界提供了类型安全保障。
13.9 完整示例:使用 configurable 动态切换模型与温度的 Agent
from typing import Annotated, TypedDict, Any
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.runnables import RunnableConfig
from langchain_core.messages import SystemMessage
from langchain_openai import ChatOpenAI
class State(TypedDict):
messages: Annotated[list, add_messages]
def chatbot(state: State, config: RunnableConfig) -> dict:
"""对话节点:从 configurable 读取模型、温度、系统提示词。"""
cfg = config["configurable"]
model_name = cfg.get("model", "gpt-4o-mini")
temperature = float(cfg.get("temperature", 0.0))
system_prompt = cfg.get(
"system_prompt",
"你是一个乐于助人的 AI 助手,回答简洁准确。",
)
# 每次调用根据 configurable 动态构造 LLM
llm = ChatOpenAI(model=model_name, temperature=temperature)
messages = [SystemMessage(content=system_prompt)] + state["messages"]
response = llm.invoke(messages)
return {"messages": [response]}
builder = StateGraph(State)
builder.add_node("chatbot", chatbot)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)
graph = builder.compile(checkpointer=MemorySaver())
# 调用一:默认模型与温度(低成本日常对话)
r1 = graph.invoke(
{"messages": [("user", "Python 中如何反转字符串?")]},
config={
"configurable": {"thread_id": "c-1"},
"tags": ["default"],
"run_name": "simple_qa",
"recursion_limit": 10,
"metadata": {"user_id": "u-1", "channel": "web"},
},
)
print("[default] ", r1["messages"][-1].content[:80])
# 调用二:同 thread 多轮,切换到 gpt-4o 并调高温度 + 创意系统提示
r2 = graph.invoke(
{"messages": [("user", "给这个问题讲一个冷笑话类比:为什么程序员喜欢黑夜模式?")]},
config={
"configurable": {
"thread_id": "c-1",
"model": "gpt-4o",
"temperature": 0.9,
"system_prompt": "你是一个冷面笑匠,回答充满黑色幽默,不要超过三句话。",
},
"tags": ["creative", "gpt4o"],
"run_name": "creative_joke",
"recursion_limit": 10,
"metadata": {"user_id": "u-1", "channel": "web", "experiment": "creative_mode"},
},
)
print("[creative] ", r2["messages"][-1].content)
# 调用三:调用三:客服场景,用专业 prompt + 低温
r3 = graph.invoke(
{"messages": [("user", "我的订单为什么还没发货?订单号 12345")]},
config={
"configurable": {
"thread_id": "c-2", # 新会话
"model": "gpt-4o-mini",
"temperature": 0.1,
"system_prompt": "你是专业电商客服,回答礼貌、简洁,不编造物流信息,不清楚时请用户联系人工。",
},
"tags": ["customer-service"],
"run_name": "cs_response",
"recursion_limit": 10,
"metadata": {"user_id": "u-2", "channel": "app", "ticket_id": "TK-12345"},
},
)
print("[CS] ", r3["messages"][-1].content[:100])该示例展示了如何用同一张编译后的图在不同调用中通过 configurable 切换模型、温度、系统提示词,配合 tags/metadata/run_name 让 LangSmith 中不同业务场景的 trace 清晰可辨。
本章小结
RunnableConfig 是每次 invoke/stream 调用沿调用链传递的运行时配置载体,核心字段包括 configurable(用户自定义动态参数)、tags(标签)、metadata(元数据)、recursion_limit(递归深度安全阀)、run_name(运行名称)、callbacks(回调)。config["configurable"]["thread_id"] 是 Checkpointer 识别执行线程的关键字段,每次调用必须传入。节点中可通过函数签名 config 参数或 get_config() 函数读取配置;工具函数通过 InjectedToolArg 注解或 get_config() 读取。recursion_limit 是防死循环安全阀(默认 25),必须保留有限值;tags 与 metadata 服务于 LangSmith 可观测性;run_name 为运行命名便于场景聚合。configurable 适合承载每次调用可变、不需要跨超步持久化的参数(模型名、温度、user_id、业务 ID、system prompt),State 承载需要在节点间流转并持久化的数据,两者职责不可混淆。get_store、get_stream_writer、get_context 等工具函数通过 ContextVar 提供隐式运行时资源访问,get_stream_writer 用于 custom stream mode 推送自定义进度事件。子图默认继承父图 configurable,可通过 config_schema(Pydantic BaseModel)做字段校验与默认值填充。
常见错误
- configurable 字段拼写或大小写错误导致读到 None。现象:节点读取 config["configurable"]["user_id"] 返回 None,后续鉴权或 Store 查询逻辑报错(KeyError、数据库查空、Store 命名空间错误)。原因:调用方传入
"userId"、"uid"等不同键名,与读取侧"user_id"不一致;或键名大小写混用。修复:团队内维护 configurable 字段名常量表(Python 模块),键名统一 snake_case;CI 中对关键字段做静态检查;节点读取时对必填字段做显式校验(缺字段抛 ValueError 并给出清晰提示)。 - 把应该放在 State 的数据误放入 configurable 导致节点间无法共享。现象:节点 A 试图"修改"configurable 中的值(通过局部变量赋值),节点 B 读到的仍是原始值;或条件路由依据 configurable 判断永远不变。原因:configurable 是调用方传入的只读上下文,不会跨超步持久化,节点对它的修改不影响下一超步。修复:需要在节点间流转并影响后续路由的数据(如是否完成、当前任务阶段、轮次计数)必须放入 State,通过节点返回值更新。
- config 未正确传递到子图。现象:子图节点 get_config() 拿不到父图传入的 user_id、tenant_id 等字段;或子图 config_schema 校验失败抛 ValidationError。原因:使用自定义 config_schema 后白名单过滤掉了未声明字段;或子图作为普通函数节点调用时手动构造新 RunnableConfig 覆盖了原始 config。修复:子图 config_schema 中显式声明需要从父图接收的字段;子图通过
builder.add_node("sub", sub_graph)方式加入父图,LangGraph 自动处理 config 传递,避免手动包装。 - recursion_limit 与真实循环不匹配。现象两种:一是复杂多 Agent 场景频繁 GraphRecursionError 导致任务中断;二是简单循环(bug 导致死循环)设置了过大的 limit,消耗大量 token 与时间后才报错。原因:没有根据业务最坏路径估算超步数。修复:对每类图显式设置 recursion_limit(简单对话 10–15、ReAct 25–50、多 Agent 50–100),在 State 中维护显式轮次/重试计数字段,条件路由基于该字段提供兜底终止分支。
- 在生产中依赖默认 metadata 导致 trace 无法排查。现象:LangSmith 中所有 trace 都叫 "LangGraph",tags 为空,metadata 为空,出现客诉时无法按 user_id/ticket_id 反查对应调用。原因:调用图时没有传 run_name/tags/metadata。修复:封装统一的调用入口(SDK 层),自动注入 run_name、tags、metadata 至少包含 service 名、env、user_id、request_id;应用层按需补充业务字段。
生产建议
- 团队维护一份 configurable 字段规范文档(Confluence/Wiki),列出字段名、类型、是否必填、默认值、使用节点,所有新字段通过 PR 评审加入;
- 业务 ID(user_id、tenant_id、ticket_id、request_id)每次调用都必须在 metadata 中携带,LangSmith 排查、审计日志、告警聚合都依赖这些字段;
- recursion_limit 在构图时通过 config 或 create_react_agent 参数显式设置,不依赖全局默认值;跨版本升级时默认值变化不会影响业务行为;
- 动态切换模型/温度/系统提示词优先走 configurable,不要为每个变体编译多张图;需要固化的 prompt 模板放入 ChatPromptTemplate 集中管理;
- Python 版本锁定 3.11 及以上,保证 get_store/get_stream_writer/get_config 的 ContextVar 跨 asyncio 任务传播稳定;
- custom stream mode 用于长任务进度推送(审批等待、研究进度、Map-Reduce 子任务完成),但不要用它承载应该持久化的数据;业务状态必须通过 State 更新;
- 子图一律声明 config_schema(Pydantic BaseModel),明确入参与默认值,防止父图 config 随意透传造成隐性耦合;
- 在封装的 SDK/中间件层对 config 做统一校验(必填 configurable 字段、recursion_limit 上限、metadata 必填字段),错误配置在入口处快速失败,避免执行到业务节点才报错。
