在深入讲解 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)

渲染错误: Mermaid 渲染失败: Parse error on line 3: ...voke[调用 graph.invoke(initial_state, conf -----------------------^ Expecting 'SQE', 'DOUBLECIRCLEEND', 'PE', '-)', 'STADIUMEND', 'SUBROUTINEEND', 'PIPE', 'CYLINDEREND', 'DIAMOND_STOP', 'TAGEND', 'TRAPEND', 'INVTRAPEND', 'UNICODE_TEXT', 'TEXT', 'TAGSTART', got 'PS'

三、详细流程文字说明

步骤 1:定义静态配置结构

  • 使用 TypedDict 定义了 CustomerConfig,包含:
    • tenant_id:租户ID(企业标识)
    • user_id:用户ID
    • model_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
  • 添加条件边:从 processshould_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_supportresolved
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__" 中,演示了两个场景:

  1. 公司A(专业语气,指定模型):问题“API密钥过期如何续期”
  2. 公司B(热情语气,默认模型):问题“想购买企业版”

每个场景都会:

  • 打印客服回复内容
  • 显示是否需要转人工客服

你可以通过修改 configurable 中的 personalitymodel_name 来观察 Agent 行为的变化。


通过以上流程图和详细说明,你应该能清晰理解 LangGraph 中静态配置(configurable)如何与图状态分离工作,以及如何在节点函数中显式获取并使用这些配置。

🎯 面试问答准备

Q1: LangGraph 中静态配置和状态有什么区别?
A: 静态配置(Config)适用于在单次执行中不会变化的数据,如 user_idtenant_id 或 API 密钥,通过 RunnableConfig 传递。而状态(State)会随着执行过程动态变化,如对话历史或中间结果,通过 StateGraph 管理。这种分离使系统职责更清晰。

Q2: 为什么要使用显式的 RunnableConfig 而不是旧版的 get_config()
A: 现代做法是在函数签名中显式声明 config: RunnableConfig 参数。这样做的优势在于:函数依赖关系一目了然,类型提示完善,测试更方便(可直接传入 mock 配置进行单元测试)。而 get_config() 依赖 Python 的 contextvar 机制,在复杂异步调用中存在信息丢失风险。

Q3: thread_iduser_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)的兼容性。

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐