LangChain RunnableConfig 详解：运行时传参、并发控制与回调注入

LangChain RunnableConfig 详解：运行时传参、并发控制与回调注入

链写好了，但运行时想临时换个模型、换个用户、换个并发数怎么办？
每次都重建一条链太笨了。

RunnableConfig就是解决"同一条链，不同运行参数"这个问题的。
这篇讲它能干什么，以及configurable这个最实用的字段。

一、RunnableConfig 长什么样

config={# 1. 每次调用的唯一标识"run_id":None,# 一般不用手动设# 2. 打标签，便于在日志/追踪里筛选"tags":["rag","prod","v2"],# 3. 附加的任意元数据"metadata":{"user_id":"u-123","trace_id":"abc"},# 4. 限制最大并发（batch 时生效）"max_concurrency":5,# 5. 回调（埋点、日志、流式）"callbacks":[my_handler],# 6. 【最常用】运行时可变的自定义参数"configurable":{"model":"gpt-4o","user_level":"vip"},# 7. 递归深度上限（LangGraph）"recursion_limit":12,}

调用时传入：

chain.invoke(inputs,config=config)chain.batch(list_of_inputs,config=config)awaitchain.ainvoke(inputs,config=config)

二、tags 和 metadata：给调用打标

chain.invoke({"question":"..."},config={"tags":["customer-service","tenant-a"],"metadata":{"user_id":"u-123","session":"s-456"},},)

这两个字段不会影响链的行为，纯粹是为了可观测性：

• 在 LangSmith 里可以按 tag 筛选这批调用；
• 在自定义 callback 里能读到 metadata 做统计。

fromlangchain_core.callbacksimportBaseCallbackHandlerclassMyHandler(BaseCallbackHandler):defon_chain_start(self,serialized,inputs,*,run_id,tags=None,metadata=None,**kwargs):print(f"run={run_id}tags={tags}user={metadata.get('user_id')}")

生产环境强烈建议每次调用都带 metadata——
用户投诉"刚才答错了"时，你靠user_id和trace_id才能定位到那次调用。

三、max_concurrency：别把 API 打挂

# 100 个输入，但最多同时发 5 个请求results=chain.batch(inputs_list,config={"max_concurrency":5},)

不设的话，batch会一次性把所有请求发出去。
100 个并发请求大概率触发限流（429），然后重试，反而更慢。

怎么定值：

比如你的 key 限制 3000 RPM，单次请求约 2 秒，
理论并发上限 ≈ 3000/60 × 2 = 100，取 70% 就是 70。

# 异步批量更推荐results=awaitchain.abatch(inputs_list,config={"max_concurrency":10})

四、configurable：运行时换参数（最实用）

这是RunnableConfig里最有价值的字段。
它让你把链的一部分做成"可配置"，运行时再决定。

4.1 configurable_fields：替换链上的字段

fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnablesimportConfigurableField llm=ChatOpenAI(model="gpt-4o-mini",temperature=0).configurable_fields(model_name=ConfigurableField(id="model",name="模型名",description="运行时可以换成别的模型",),temperature=ConfigurableField(id="temp",description="运行时可以调温度",),)chain=ChatPromptTemplate.from_template("{q}")|llm|StrOutputParser()# 默认跑chain.invoke({"q":"你好"})# 运行时换成强模型 + 有创意chain.invoke({"q":"写首诗"},config={"configurable":{"model":"gpt-4o","temp":0.9}},)

同一条链，不用重建，运行时换模型、换温度。

4.2 configurable_alternatives：整段替换

不只是改参数，可以整段换成另一个实现：

cheap=ChatOpenAI(model="gpt-4o-mini")strong=ChatOpenAI(model="gpt-4o")llm=cheap.configurable_alternatives(ConfigurableField(id="tier"),default_key="cheap",strong=strong,)chain=prompt|llm|parser chain.invoke({"q":"..."})# 默认便宜模型chain.invoke({"q":"..."},config={"configurable":{"tier":"strong"}})# 换强模型

典型用法：简单问题走便宜模型，复杂问题走强模型，成本直降。

defroute_and_answer(question:str):tier="strong"ifis_complex(question)else"cheap"returnchain.invoke({"q":question},config={"configurable":{"tier":tier}})

4.3 在自定义 Runnable 里读 configurable

这是最灵活的用法。写节点时通过第二个参数拿到 config：

fromlangchain_core.runnablesimportRunnableLambdadefmy_step(inputs,config):# 从 configurable 里取运行时参数user_level=config["configurable"].get("user_level","normal")max_results=20ifuser_level=="vip"else5returnsearch(inputs["query"],top_k=max_results)step=RunnableLambda(my_step)

注意：函数必须接受两个参数（inputs, config），
LangChain 会自动把 config 传进来。只写一个参数就拿不到 config。

在 LangGraph 节点里同理：

defnode(state,config):tenant=config["configurable"].get("tenant_id")...

五、recursion_limit：LangGraph 的步数刹车

agent.invoke({"messages":[("user",q)]},config={"recursion_limit":12},)

生产必设，否则模型死循环会烧掉大量调用。
具体见前面 Agent 进阶那篇。

六、callbacks：在运行时注入埋点

fromlangchain_core.callbacksimportBaseCallbackHandlerclassCostTracker(BaseCallbackHandler):def__init__(self):self.total_tokens=0defon_llm_end(self,response,**kwargs):usage=response.llm_output.get("token_usage",{})self.total_tokens+=usage.get("total_tokens",0)tracker=CostTracker()chain.invoke({"q":"..."},config={"callbacks":[tracker]})print(tracker.total_tokens)

按调用注入比全局注入更灵活——你可以只给某些请求开追踪。

七、config 会向下传递

一个重要特性：config 会传给链里的每一个 Runnable，
包括嵌套的子链。

outer=({"a":inner_chain_1,"b":inner_chain_2}|...)outer.invoke(x,config={"tags":["outer"]})# inner_chain_1 和 inner_chain_2 都会收到这个 config

这意味着：

• 你在最外层设的max_concurrency，对里面所有并发都生效；
• 你在最外层设的metadata，里面每个节点都能读到；
• tags会累加——子链的 tag 会追加到父链后面。

八、一个生产级的 config 模板

importuuiddefbuild_config(user_id:str,session_id:str,tier:str="cheap")->dict:return{"tags":[f"tenant:{tenant_id(user_id)}",f"tier:{tier}"],"metadata":{"user_id":user_id,"session_id":session_id,"trace_id":str(uuid.uuid4()),},"max_concurrency":5,"configurable":{"tier":tier,"user_id":user_id,"thread_id":session_id,# LangGraph 用这个做记忆},"recursion_limit":12,"callbacks":[metrics_handler],}result=chain.invoke({"question":q},config=build_config("u-123","s-456"))

九、常见坑

十、小结

• RunnableConfig是"同一条链，不同运行参数"的解决方案；
• tags/metadata用于可观测性，生产每次调用都要带；
• max_concurrency必设，否则batch会一次性打爆 API；
• configurable最实用：configurable_fields改参数、
  configurable_alternatives整段替换（便宜/强模型切换）；
• 自定义函数要写def f(inputs, config)两个参数才能拿到 config；
• config会向下传递给所有子链，一次设置处处生效。

下一篇讲自定义 Runnable——把任意 Python 逻辑接进 LCEL 链。