跳到内容

Runtime

Runtime dataclass

基类:Generic[ContextT]

一个便利类,捆绑了运行范围的上下文和其他运行时工具。

在 v0.6.0 版本中添加

示例

from typing import TypedDict
from langgraph.graph import StateGraph
from dataclasses import dataclass
from langgraph.runtime import Runtime
from langgraph.store.memory import InMemoryStore


@dataclass
class Context:  # (1)!
    user_id: str


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


store = InMemoryStore()  # (2)!
store.put(("users",), "user_123", {"name": "Alice"})


def personalized_greeting(state: State, runtime: Runtime[Context]) -> State:
    '''Generate personalized greeting using runtime context and store.'''
    user_id = runtime.context.user_id  # (3)!
    name = "unknown_user"
    if runtime.store:
        if memory := runtime.store.get(("users",), user_id):
            name = memory.value["name"]

    response = f"Hello {name}! Nice to see you again."
    return {"response": response}


graph = (
    StateGraph(state_schema=State, context_schema=Context)
    .add_node("personalized_greeting", personalized_greeting)
    .set_entry_point("personalized_greeting")
    .set_finish_point("personalized_greeting")
    .compile(store=store)
)

result = graph.invoke({}, context=Context(user_id="user_123"))
print(result)
# > {'response': 'Hello Alice! Nice to see you again.'}
  1. 为运行时上下文定义一个模式。
  2. 创建一个存储库以持久化记忆和其他信息。
  3. 使用运行时上下文访问 user_id。

属性

名称 类型 描述
context ContextT

图运行的静态上下文,如 user_id、db_conn 等。

store BaseStore | None

图运行的存储库,用于实现持久化和记忆。

stream_writer StreamWriter

写入自定义流的函数。

previous Any

给定线程的上一个返回值。

context class-attribute instance-attribute

context: ContextT = field(default=None)

图运行的静态上下文,如 user_id、db_conn 等。

也可以被认为是“运行依赖项”。

store class-attribute instance-attribute

store: BaseStore | None = field(default=None)

图运行的存储库,用于实现持久化和记忆。

stream_writer class-attribute instance-attribute

stream_writer: StreamWriter = field(
    default=_no_op_stream_writer
)

写入自定义流的函数。

previous class-attribute instance-attribute

previous: Any = field(default=None)

给定线程的上一个返回值。

仅在提供了检查点(checkpointer)的情况下,与函数式 API 一起使用时可用。

函数

名称 描述
get_runtime

获取当前图运行的运行时。

get_runtime

get_runtime(
    context_schema: type[ContextT] | None = None,
) -> Runtime[ContextT]

获取当前图运行的运行时。

参数

名称 类型 描述 默认值
context_schema type[ContextT] | None

可选模式,用于类型提示运行时的返回类型。

None

返回

类型 描述
Runtime[ContextT]

当前图运行的运行时。