16.1深入讲解 LangGraph 的静态配置(Static Context)
在深入讲解 LangGraph 的静态配置(Static Context)前,我们可以回顾一下核心概念。静态配置是一种在 Agent 执行期间保持不变的上下文,常用于用户元数据、API 密钥、模型选择等场景。由于它具备只读性,非常适合放那些与本次调用绑定但不在执行过程中变更的信息。
而且,官方文档在推荐使用 config 而非旧版的 get_config 方法,因为这能让代码依赖关系更清晰、更易于维护。接下来,我会用一个“多租户智能客服系统”案例,让你直观掌握它的用法。
📝 完整实操案例:多租户智能客服系统
这个案例会创建一个可同时服务多个企业的智能客服系统,并通过 configurable 静态配置来区分客户信息与业务模型。
# ========== 1. 环境准备与模型初始化 ==========
import os
from typing import TypedDict, Optional, Literal
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langchain_core.runnables import RunnableConfig
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
load_dotenv()
# 初始化模型(使用阿里云百炼 Qwen)
llm = ChatOpenAI(
model="qwen-plus",
temperature=0.7,
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url=os.getenv("DASHSCOPE_BASE_URL", "https://dashscope.aliyuncs.com/compatible-mode/v1"),
model_kwargs={"extra_body": {"enable_thinking": False}}
)
# ========== 2. 定义配置结构(使用 TypedDict 确保类型安全)==========
class CustomerConfig(TypedDict):
"""客户配置结构 - 静态上下文"""
tenant_id: str # 租户ID(企业标识)
user_id: str # 用户ID
model_name: Optional[str] # 可选:为该客户指定的模型
personality: Literal["professional", "friendly", "enthusiastic"] # 客服语气
# ========== 3. 定义图状态 ==========
class AgentState(TypedDict):
"""Agent 状态 - 动态上下文"""
messages: list # 对话历史
need_human_support: bool # 是否需要转人工客服
resolved: bool # 问题是否已解决
# ========== 4. 节点函数:访问配置的现代方式(显式传递 RunnableConfig)==========
def process_customer_query(state: AgentState, config: RunnableConfig) -> AgentState:
"""
处理客户查询的主节点
- 通过 config 参数获取静态配置(租户信息、模型选择、客服语气等)
"""
# 从 config 中提取客户配置
customer_config = config.get("configurable", {})
tenant_id = customer_config.get("tenant_id", "unknown")
user_id = customer_config.get("user_id", "anonymous")
personality = customer_config.get("personality", "friendly")
model_name = customer_config.get("model_name")
print(f"\n🏢 处理客户查询 [Tenant: {tenant_id}, User: {user_id}, 语气: {personality}]")
# 根据配置动态调整系统提示词
personality_prompts = {
"professional": "你是一个专业的客服代表,请用正式、准确的语言回答。",
"friendly": "你是一个友好的客服代表,请用亲切、热情的语气回答。",
"enthusiastic": "你是一个充满热情的客服代表,请用积极、鼓励的语气回答。"
}
# 构造消息列表
messages = state.get("messages", [])
system_prompt = f"""{personality_prompts.get(personality, personality_prompts["friendly"])}
你正在为租户 {tenant_id} 的客户提供客服服务,客户ID是 {user_id}。
请提供专业、有帮助的回答。如果客户的问题需要人工客服介入,请在回答末尾标注 [NEED_HUMAN]。
"""
# 调用模型(支持可选的模型配置)
if model_name:
# 如果配置了特定模型,可以动态切换
temp_llm = ChatOpenAI(
model=model_name,
temperature=0.7,
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url=os.getenv("DASHSCOPE_BASE_URL"),
model_kwargs={"extra_body": {"enable_thinking": False}}
)
response = temp_llm.invoke([SystemMessage(content=system_prompt)] + messages)
else:
response = llm.invoke([SystemMessage(content=system_prompt)] + messages)
# 判断是否需要转人工客服
need_human = "[NEED_HUMAN]" in response.content
return {
"messages": [AIMessage(content=response.content.replace("[NEED_HUMAN]", ""))],
"need_human_support": need_human,
"resolved": not need_human
}
def check_resolution(state: AgentState) -> AgentState:
"""检查问题是否解决"""
if state.get("resolved", False):
print("✅ 问题已解决,结束会话")
else:
print("⚠️ 问题未解决,记录工单")
return state
def should_continue(state: AgentState) -> Literal["check", "__end__"]:
"""判断是否继续处理"""
return "check" if not state.get("resolved", True) else "__end__"
# ========== 5. 构建图 ==========
builder = StateGraph(AgentState)
# 添加节点
builder.add_node("process", process_customer_query)
builder.add_node("check", check_resolution)
# 定义流程
builder.set_entry_point("process")
builder.add_conditional_edges("process", should_continue, {
"check": "check",
"__end__": END
})
builder.add_edge("check", END)
# 编译图
graph = builder.compile()
# ========== 6. 测试不同租户的配置 ==========
if __name__ == "__main__":
print("🚀 启动多租户智能客服系统")
print("=" * 60)
# 测试1:专业语气客户(A公司)
print("\n📞 【场景1】A公司 - 专业语气")
config_a = {
"configurable": {
"tenant_id": "company_a",
"user_id": "user_001",
"personality": "professional",
"model_name": "qwen-plus" # 使用指定模型
}
}
result_a = graph.invoke(
{"messages": [HumanMessage(content="我们的API密钥过期了,怎么续期?")]},
config=config_a
)
print(f"客服回复: {result_a['messages'][-1].content}")
print(f"需要转人工: {result_a['need_human_support']}")
# 测试2:热情语气客户(B公司)
print("\n📞 【场景2】B公司 - 热情语气")
config_b = {
"configurable": {
"tenant_id": "company_b",
"user_id": "user_002",
"personality": "enthusiastic"
# 未指定模型,将使用默认模型
}
}
result_b = graph.invoke(
{"messages": [HumanMessage(content="你们的产品太棒了!我想购买企业版,能介绍一下吗?")]},
config=config_b
)
print(f"客服回复: {result_b['messages'][-1].content}")
print(f"需要转人工: {result_b['need_human_support']}")
# 测试3:展示图结构(可选)
# print("\n📊 图结构:")
# print(graph.get_graph().draw_ascii())
💡 核心知识点总结
| 概念 | 说明 | 本例中的应用 |
|---|---|---|
| Static Context | 在单次执行期间不变的只读数据 | tenant_id, user_id, personality, model_name |
configurable 键 |
RunnableConfig 中专门用于存放静态配置的保留字段 |
config.get("configurable", {}) |
RunnableConfig 显式传递 |
在节点函数签名中声明 config: RunnableConfig 参数 |
def process_customer_query(state, config: RunnableConfig) |
| 类型安全配置 | 使用 TypedDict 定义配置结构 |
class CustomerConfig(TypedDict) |
| 动态模型选择 | 根据配置动态切换 LLM 模型 | model_name 配置影响使用的模型 |
| 配置与状态分离 | 配置(静态)与状态(动态)各司其职,不互相污染 | 配置存 tenant_id,状态存 need_human_support |
LangGraph 静态配置(Configurable)案例代码流程详解
一、整体功能概述
这个“多租户智能客服系统”演示了如何使用 LangGraph 的 静态配置(RunnableConfig + configurable) 来为不同租户(企业)提供个性化的客服服务。每个租户可以指定:
- 客服语气(专业/友好/热情)
- 使用的模型(可选)
- 租户ID、用户ID等元数据
系统将静态配置与动态状态(对话历史、是否转人工等)分离,通过 LangGraph 的图结构执行。
二、代码执行流程图(Mermaid)
三、详细流程文字说明
步骤 1:定义静态配置结构
- 使用
TypedDict定义了CustomerConfig,包含:tenant_id:租户ID(企业标识)user_id:用户IDmodel_name:可选的模型名称personality:客服语气(professional,friendly,enthusiastic)
步骤 2:定义图状态(动态部分)
AgentState包含:messages:对话历史(HumanMessage,AIMessage)need_human_support:是否需要转人工客服resolved:问题是否已解决
步骤 3:构建 LangGraph 图
- 创建
StateGraph(AgentState) - 添加两个节点:
process:调用process_customer_query函数(核心处理)check:调用check_resolution函数(检查解决状态)
- 设置入口点:
process - 添加条件边:从
process到should_continue判断- 如果
resolved == True→ 跳转到check节点 - 否则 → 结束(END)
- 如果
- 添加普通边:
check→ END
步骤 4:执行流程详解
4.1 调用 graph.invoke
- 传入初始状态(
messages包含用户问题) - 传入配置:
config={"configurable": {...}}
4.2 进入 process_customer_query 节点
- 参数接收:函数签名
def process_customer_query(state, config: RunnableConfig)显式接收配置。 - 提取配置:
customer_config = config.get("configurable", {})获取租户信息、语气、模型等。 - 构造系统提示词:根据
personality选择对应的语气模板。 - 调用 LLM:
- 如果配置中指定了
model_name,则临时创建对应模型实例(演示动态模型切换) - 否则使用默认的
llm实例(阿里云百炼 Qwen) - 调用
llm.invoke([SystemMessage(...)] + messages)
- 如果配置中指定了
- 解析响应:
- 检测响应中是否包含
[NEED_HUMAN]标记 - 去除标记后存储
AIMessage - 更新状态:
need_human_support和resolved
- 检测响应中是否包含
4.3 条件路由 should_continue
- 检查
state["resolved"] - 返回
"check"或"__end__"
4.4 进入 check_resolution 节点
- 打印日志(问题是否已解决),不做状态修改
- 然后通过
add_edge结束
步骤 5:返回最终状态
- 调用方获得更新后的
AgentState,包含客服回复和转人工标志
四、关键代码片段对应流程
| 流程环节 | 对应代码行 |
|---|---|
| 配置结构定义 | class CustomerConfig(TypedDict): |
| 图状态定义 | class AgentState(TypedDict): |
| 节点函数提取配置 | customer_config = config.get("configurable", {}) |
| 动态模型切换 | if model_name: temp_llm = ChatOpenAI(model=model_name, ...) |
| 条件边判断 | def should_continue(state) -> Literal["check", "__end__"] |
| 编译图 | graph.compile() |
| 调用时传入配置 | graph.invoke(initial_state, config=config_a) |
五、ASCII 简化流程图(适用于纯文本环境)
用户输入 + configurable配置
│
▼
┌─────────────────────────────┐
│ process_customer_query │
│ - 提取 tenant_id, persona │
│ - 构造系统提示词 │
│ - 调用 LLM(可动态切换模型)│
│ - 检测 [NEED_HUMAN] │
│ - 更新 state.resolved │
└─────────────────────────────┘
│
▼ (条件边)
resolved == True ?
│ │
是 否
│ │
▼ ▼
┌──────────┐ END
│ check │
│ 打印日志 │
└──────────┘
│
▼
END
六、运行示例说明
在 if __name__ == "__main__" 中,演示了两个场景:
- 公司A(专业语气,指定模型):问题“API密钥过期如何续期”
- 公司B(热情语气,默认模型):问题“想购买企业版”
每个场景都会:
- 打印客服回复内容
- 显示是否需要转人工客服
你可以通过修改 configurable 中的 personality 和 model_name 来观察 Agent 行为的变化。
通过以上流程图和详细说明,你应该能清晰理解 LangGraph 中静态配置(configurable)如何与图状态分离工作,以及如何在节点函数中显式获取并使用这些配置。
🎯 面试问答准备
Q1: LangGraph 中静态配置和状态有什么区别?
A: 静态配置(Config)适用于在单次执行中不会变化的数据,如 user_id、tenant_id 或 API 密钥,通过 RunnableConfig 传递。而状态(State)会随着执行过程动态变化,如对话历史或中间结果,通过 StateGraph 管理。这种分离使系统职责更清晰。
Q2: 为什么要使用显式的 RunnableConfig 而不是旧版的 get_config()?
A: 现代做法是在函数签名中显式声明 config: RunnableConfig 参数。这样做的优势在于:函数依赖关系一目了然,类型提示完善,测试更方便(可直接传入 mock 配置进行单元测试)。而 get_config() 依赖 Python 的 contextvar 机制,在复杂异步调用中存在信息丢失风险。
Q3: thread_id 和 user_id 在配置中有什么区别?
A: thread_id 是 LangGraph checkpointer 的存储密钥,用于标识对话线程,只有编译图时建立了 checkpoint 才会生效。而 user_id 是业务元数据,框架不抓取,只在节点里由业务逻辑自行取用。
Q4: 如何根据用户配置动态切换 LLM 模型?
A: 支持静态和动态两种模型选择方式。静态模型在创建代理时一次配置并保持不变。如果需要在运行中切换,可以通过配置传递模型名称,然后在节点中根据该配置重新初始化对应的 LLM 实例。
Q5: 配置如何传递给工具(Tool)?
A: 工具可以通过特殊的参数注解 RunnableConfig 访问配置。这些参数对 LLM 是隐藏的,不会被模型尝试填充。这种方式可以安全地在工具内部获取用户 ID、API 凭据等敏感信息。
Q6: LangGraph v0.6 的 Context API 与 configurable 是什么关系?
A: LangGraph v0.6 引入了 Context API,意图用 Context 统一管理用户定义的数据,逐步替代 config["configurable"]。但当前实践中推荐混合使用:用 Context 管理应用程序逻辑相关的不可变状态,用 config["configurable"] 管理系统级配置(如 thread_id)和与生态包(如 LangMem)的兼容性。
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。
更多推荐


所有评论(0)