第14章 Runtime 与 Context

14.1 引言

在构建 LLM 应用的图时，节点函数往往需要访问一些"运行时依赖"——当前用户的身份信息、数据库连接池、API 密钥、或者一个全局的向量存储。这些依赖既不属于图的状态（它们不随步骤变化），也不应该被硬编码在节点函数中（它们因调用而异）。传统做法是通过闭包或全局变量传递，但这在测试、多租户和类型安全方面都存在痛点。

LangGraph 1.1.6 引入了 Runtime 类和 ContextT 泛型来解决这个问题。Runtime 是一个不可变的数据容器，在图执行开始时由调用方创建，自动注入到每个节点函数中。它携带了 context（用户自定义的运行时上下文）、store（持久化存储）、stream_writer（流式写入器）、execution_info（执行元数据）等运行时信息。

本章将从 Runtime 的数据类定义出发，分析 ContextT 泛型的设计理念、ExecutionInfo 和 ServerInfo 的信息模型、context 与 state 的本质区别，以及 Runtime 在 Pregel 循环中的注入机制。

本章要点

1. Runtime 类的完整字段定义——context、store、stream_writer、previous、execution_info、server_info
2. ContextT 泛型的类型传播——从 StateGraph 到节点函数的端到端类型安全
3. ExecutionInfo 与 ServerInfo 的信息模型——执行元数据的结构化表达
4. Context vs State 的本质区别——不可变依赖 vs 可变状态
5. Runtime 注入机制——从编译到执行的完整链路

14.2 Runtime 类的设计

14.2.1 数据类定义

Runtime 定义在 langgraph/runtime.py 中，是一个泛型冻结数据类：

@dataclass(**_DC_KWARGS)  # kw_only=True, slots=True, frozen=True
class Runtime(Generic[ContextT]):
    """Convenience class that bundles run-scoped context and other runtime utilities."""

    context: ContextT = field(default=None)
    """Static context for the graph run, like user_id, db_conn, etc."""

    store: BaseStore | None = field(default=None)
    """Store for the graph run, enabling persistence and memory."""

    stream_writer: StreamWriter = field(default=_no_op_stream_writer)
    """Function that writes to the custom stream."""

    previous: Any = field(default=None)
    """The previous return value for the given thread (functional API only)."""

    execution_info: ExecutionInfo | None = field(default=None)
    """Read-only execution information/metadata for the current node run."""

    server_info: ServerInfo | None = field(default=None)
    """Metadata injected by LangGraph Server. None for open-source."""

_DC_KWARGS 展开为 kw_only=True, slots=True, frozen=True，这意味着：

• kw_only：所有字段必须通过关键字参数传递，避免位置参数的歧义
• slots：使用 __slots__ 优化内存和属性访问速度
• frozen：实例创建后不可修改，确保运行时安全

14.2.2 字段语义

六个字段覆盖了节点函数可能需要的所有运行时信息：

字段	类型	来源	可变性
context	ContextT（泛型）	调用方传入	整个执行期间不变
store	BaseStore	图编译时配置	引用不变，内容可变
stream_writer	StreamWriter	框架自动注入	每个任务独立
previous	Any	Checkpoint 读取	只读
execution_info	ExecutionInfo	框架生成	每个任务独立
server_info	ServerInfo	LangGraph Server	只读

14.2.3 不可变性与 override/merge

虽然 Runtime 是 frozen 的，但它提供了两个方法来创建修改后的副本：

def merge(self, other: Runtime[ContextT]) -> Runtime[ContextT]:
    """Merge two runtimes together. If a value is not provided in other,
    the value from self is used."""
    return Runtime(
        context=other.context or self.context,
        store=other.store or self.store,
        stream_writer=other.stream_writer
            if other.stream_writer is not _no_op_stream_writer
            else self.stream_writer,
        previous=self.previous if other.previous is None else other.previous,
        execution_info=other.execution_info or self.execution_info,
        server_info=other.server_info or self.server_info,
    )

def override(self, **overrides) -> Runtime[ContextT]:
    """Replace the runtime with a new runtime with the given overrides."""
    return replace(self, **overrides)

merge 用于子图继承父图的 Runtime 时，合并两个 Runtime 对象。override 用于框架在任务准备阶段注入特定字段（如 execution_info）。

14.3 ContextT 泛型

14.3.1 定义

# langgraph/typing.py
ContextT = TypeVar("ContextT", bound=StateLike | None, default=None)

ContextT 是一个带默认值的类型变量，约束为 StateLike | None。StateLike 包括 TypedDict、BaseModel、dataclass 等结构化类型。默认值为 None，这意味着如果不指定 context_schema，Runtime 的 context 字段类型就是 None。

14.3.2 类型传播链路

类型从 StateGraph 的 context_schema 参数开始，贯穿编译、调用、注入的全过程。IDE 和类型检查器可以在每一步提供准确的类型补全。

14.3.3 使用示例

from dataclasses import dataclass
from langgraph.graph import StateGraph
from langgraph.runtime import Runtime
from typing_extensions import TypedDict

@dataclass
class AppContext:
    user_id: str
    api_key: str
    is_admin: bool = False

class State(TypedDict, total=False):
    response: str

def my_node(state: State, runtime: Runtime[AppContext]) -> State:
    # IDE 知道 runtime.context 的类型是 AppContext
    user_id = runtime.context.user_id
    if runtime.context.is_admin:
        return {"response": f"Admin {user_id}: full access"}
    return {"response": f"User {user_id}: limited access"}

graph = (
    StateGraph(state_schema=State, context_schema=AppContext)
    .add_node("my_node", my_node)
    .set_entry_point("my_node")
    .set_finish_point("my_node")
    .compile()
)

result = graph.invoke({}, context=AppContext(user_id="alice", api_key="sk-..."))