类定义在 runtime.py:124，它是个带泛型的 dataclass：

# runtime.py:124
@dataclass(**_DC_KWARGS)
class Runtime(Generic[ContextT]):
    """Convenience class that bundles run-scoped context and other runtime utilities.
    This class is injected into graph nodes and middleware."""

    context: ContextT = field(default=None)         # 你的运行依赖：user_id/db_conn...
    store: BaseStore | None = field(default=None)   # 长期记忆库(D40)
    stream_writer: StreamWriter = field(default=_no_op_stream_writer)  # 自定义流(D57)
    heartbeat: Callable[[], None] = field(default=_no_op_heartbeat)    # 心跳(防 idle 超时)
    previous: Any = field(default=None)             # 函数式 API 上一次返回值(D47)
    execution_info: ExecutionInfo | None = field(default=None)        # 只读元信息
    server_info: ServerInfo | None = field(default=None)              # Platform 注入
    control: RunControl | None = field(default=None)                  # 停机控制面

逐个看清楚每个字段是干嘛的（字段文档就写在源码里，runtime.py:198-238）：

其中 ExecutionInfo 是个只读快照（runtime.py:26）——注意它用 frozen=True, slots=True：

# runtime.py:26
@dataclass(frozen=True, slots=True)
class ExecutionInfo:
    """Read-only execution info/metadata for the execution of current thread/run/node."""
    checkpoint_id: str
    checkpoint_ns: str
    task_id: str
    thread_id: str | None = None    # None 表示没配 checkpointer(无持久化)
    run_id: str | None = None
    node_attempt: int = 1           # 当前是第几次重试(1 起)
    node_first_attempt_time: float | None = None
    def patch(self, **overrides: Any) -> ExecutionInfo:
        return replace(self, **overrides)   # 只读→改要生成新对象

上面是「节点声明 runtime 参数」的写法。但有时你在一个被节点调用的普通函数深处想拿 store，总不能一层层往下传参。LangGraph 提供了「随地取用」的入口 runtime.py:296：

# runtime.py:296
def get_runtime(context_schema: type[ContextT] | None = None) -> Runtime[ContextT]:
    """Get the runtime for the current graph run."""
    runtime = cast(Runtime[ContextT], get_config()[CONF].get(CONFIG_KEY_RUNTIME))
    return runtime

它靠的是 get_config()（config.py:17），这是理解「随地取用」魔法的关键：

# config.py:17
def get_config() -> RunnableConfig:
    if sys.version_info < (3, 11):
        try:
            if asyncio.current_task():
                raise RuntimeError("Python 3.11 or later required to use this in an async context")
        except RuntimeError:
            pass
    if var_config := var_child_runnable_config.get():   # ← 从 contextvar 里取当前 config
        return var_config
    else:
        raise RuntimeError("Called get_config outside of a runnable context")

同一套机制还派生出两个便捷函数：get_store()（config.py:32）和 get_stream_writer()（config.py:126），本质都是 get_config()[CONF][CONFIG_KEY_RUNTIME].store / .stream_writer。

那 Runtime 到底是什么时候「填好塞进 config」的？答案在任务准备阶段（D21 讲过的 prepare_next_tasks），pregel/_algo.py:688：

# pregel/_algo.py:688
runtime = cast(Runtime, configurable.get(CONFIG_KEY_RUNTIME, DEFAULT_RUNTIME))
runtime = runtime.override(                          # ① 在基础 runtime 上叠加本任务的信息
    previous=checkpoint["channel_values"].get(PREVIOUS, None),
    store=store,
    execution_info=ExecutionInfo(                    # ② 现造一份只读元信息
        checkpoint_id=checkpoint["id"],
        checkpoint_ns=task_checkpoint_ns,
        task_id=task_id,
        thread_id=configurable.get(CONFIG_KEY_THREAD_ID),
        run_id=str(rid) if (rid := config.get("run_id")) else None,
    ),
)
# ... 随后 patch_config(..., configurable={ ..., CONFIG_KEY_RUNTIME: runtime })  # ③ 写回 config

还有个方法值得一提——多个 Runtime 怎么合并？merge（runtime.py:240）遵循「对方有就用对方、没有就保留自己」的规则，且对 _no_op 占位有专门判断：

# runtime.py:240
def merge(self, other: Runtime[ContextT]) -> Runtime[ContextT]:
    return Runtime(
        context=other.context or self.context,
        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,
        ... )

这就是 RunControl 的活儿（runtime.py:79）——一个极简的「停机广播喇叭」：

# runtime.py:79
class RunControl:
    """Run-scoped control surface for cooperative draining."""
    __slots__ = ("_drain_reason",)
    def __init__(self) -> None:
        self._drain_reason: str | None = None
    def request_drain(self, reason: str = "shutdown") -> None:
        self._drain_reason = reason          # ← 一次「原子」写：喊一声「下班」
    @property
    def drain_requested(self) -> bool:
        return self._drain_reason is not None
    @property
    def drain_reason(self) -> str | None:
        return self._drain_reason

节点里怎么感知？Runtime 上有两个转发属性（runtime.py:276）：

# runtime.py:276
@property
def drain_requested(self) -> bool:
    return self.control.drain_requested if self.control is not None else False

于是长节点可以在循环里 if runtime.drain_requested: 提前保存并返回。而框架层面，收尾会抛出一个专门的异常（errors.py:54）：

# errors.py:54
class GraphDrained(GraphBubbleUp):
    """Raised when a graph run exits early due to a drain request.
    This indicates the graph stopped cooperatively at a superstep boundary
    because `RunControl.request_drain()` was called (e.g., in response to SIGTERM).
    The checkpoint is saved and the run can be resumed later."""

errors.py（共 241 行）定义了图运行时的全部异常家族。先看一个巧妙的分类基类 errors.py:50：

# errors.py:50
class GraphBubbleUp(Exception):
    pass                          # “需要向上冒泡、由框架接住”的控制流异常基类

class GraphDrained(GraphBubbleUp): ...      # 停机(L05)
class GraphInterrupt(GraphBubbleUp): ...    # 中断(D41)，子图抛出被根图接住
class ParentCommand(GraphBubbleUp): ...     # Command(graph=PARENT) 跨图跳转(D15)

再看几个「真故障」类的错误，各有讲究：

还有个面向用户体验的小设计——错误码 + 文档直链（errors.py:34）：

# errors.py:34
class ErrorCode(Enum):
    GRAPH_RECURSION_LIMIT = "GRAPH_RECURSION_LIMIT"
    INVALID_CONCURRENT_GRAPH_UPDATE = "INVALID_CONCURRENT_GRAPH_UPDATE"
    ...
def create_error_message(*, message: str, error_code: ErrorCode) -> str:
    return (f"{message}\n"
        "For troubleshooting, visit: https://docs.langchain.com/oss/python/langgraph/"
        f"errors/{error_code.value}")

图写好了，怎么部署成一个带 REST API、持久化、任务队列的服务？LangGraph Platform 的做法是：你只写一份 langgraph.json，CLI 负责把它变成 Docker 镜像。配置 schema 定义在 libs/cli/langgraph_cli/schemas.py:615：

# schemas.py:615
class Config(TypedDict, total=False):
    python_version: str          # "3.11"
    dependencies: list[str]      # ["."] 或 ["langchain", "git+https://..."]
    graphs: dict[str, str | GraphDef]  # {"agent": "./agent.py:graph"}
    env: dict[str, str] | str    # {"OPENAI_API_KEY": "..."} 或 ".env"
    store: StoreConfig | None    # 长期记忆 + 向量检索配置
    auth: AuthConfig | None      # 鉴权
    http: HttpConfig | None      # CORS、自定义路由...
    dockerfile_lines: list[str]  # 追加到 Dockerfile 的额外指令

对应的 langgraph.json 长这样（最小可用版）：

{
  "dependencies": ["."],
  "graphs": { "agent": "./my_agent.py:graph" },
  "env": ".env"
}

CLI 命令一览（libs/cli/langgraph_cli/cli.py）：

✋ 10 分钟动手

# 1. 通读三个核心文件（都不长）
sed -n '124,160p' libs/langgraph/langgraph/runtime.py    # Runtime 类
sed -n '50,66p'   libs/langgraph/langgraph/errors.py     # GraphBubbleUp 家族
sed -n '615,660p' libs/cli/langgraph_cli/schemas.py      # 部署 Config

# 2. 亲手用 Runtime.context 注入依赖
python - <<'PY'
from dataclasses import dataclass
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START
from langgraph.runtime import Runtime
@dataclass
class Ctx: user_id: str
class S(TypedDict, total=False): out: str
def n(state: S, runtime: Runtime[Ctx]):
    return {"out": f"hi {runtime.context.user_id}, attempt={runtime.execution_info}"}
g=StateGraph(S, context_schema=Ctx).add_node(n).add_edge(START,"n").compile()
print(g.invoke({}, context=Ctx(user_id="u_123")))
PY

# 3. 校验一份 langgraph.json（先随便写一个）
# langgraph validate -c langgraph.json