Python SDK 参考¶

LangGraph 客户端实现连接到 LangGraph API。

本模块提供了异步 (LangGraphClient) 和同步 (SyncLangGraphClient) 客户端，用于与 LangGraph API 的核心资源（如 Assistant、Thread、Run 和 Cron 作业）以及其持久性文档存储库 Store 进行交互。

类

名称	描述
`LangGraphClient`	LangGraph API 的顶级客户端。
`HttpClient`	处理对 LangGraph API 的异步请求。
`AssistantsClient`	用于管理 LangGraph 中的智能体的客户端。
`ThreadsClient`	用于管理 LangGraph 中的线程的客户端。
`RunsClient`	用于管理 LangGraph 中的运行的客户端。
`CronClient`	用于管理 LangGraph 中的周期性运行（cron 作业）的客户端。
`StoreClient`	用于与图的共享存储进行交互的客户端。
`SyncLangGraphClient`	用于与 LangGraph API 进行交互的同步客户端。
`SyncHttpClient`
`SyncAssistantsClient`	用于同步管理 LangGraph 中的智能体的客户端。
`SyncThreadsClient`	用于同步管理 LangGraph 中的线程的客户端。
`SyncRunsClient`	用于同步管理 LangGraph 中的运行的客户端。
`SyncCronClient`	用于同步管理 LangGraph 中的 cron 作业的客户端。
`SyncStoreClient`	用于对键值存储执行同步操作的客户端。

函数

名称	描述
`get_headers`	合并 api_key 和用户提供的自定义头。
`get_client`	获取 LangGraphClient 实例。
`get_sync_client`	获取同步的 LangGraphClient 实例。

LangGraphClient ¶

LangGraph API 的顶级客户端。

属性

名称	类型	描述
`assistants`		管理您的图的版本化配置。
`threads`		处理（可能）多轮交互，例如对话线程。
`runs`		控制图的单个调用。
`crons`		管理计划的操作。
`store`		与持久性共享数据存储进行交互。

HttpClient ¶

处理对 LangGraph API 的异步请求。

在提供的 httpx 客户端之上添加额外的错误消息和内容处理。

属性

名称	类型	描述
`client`	`AsyncClient`	底层的 HTTPX 异步客户端。

方法

名称	描述
`get`	发送 GET 请求。
`post`	发送 POST 请求。
`put`	发送 PUT 请求。
`patch`	发送 PATCH 请求。
`delete`	发送 DELETE 请求。
`stream`	使用 SSE 流式传输结果。

get `async` ¶

get(
    path: str,
    *,
    params: Optional[QueryParamTypes] = None,
    headers: Optional[dict[str, str]] = None
) -> Any

发送 GET 请求。

post `async` ¶

post(
    path: str,
    *,
    json: Optional[dict],
    headers: Optional[dict[str, str]] = None
) -> Any

发送 POST 请求。

put `async` ¶

put(
    path: str,
    *,
    json: dict,
    headers: Optional[dict[str, str]] = None
) -> Any

发送 PUT 请求。

patch `async` ¶

patch(
    path: str,
    *,
    json: dict,
    headers: Optional[dict[str, str]] = None
) -> Any

发送 PATCH 请求。

delete `async` ¶

delete(
    path: str,
    *,
    json: Optional[Any] = None,
    headers: Optional[dict[str, str]] = None
) -> None

发送 DELETE 请求。

stream `async` ¶

stream(
    path: str,
    method: str,
    *,
    json: Optional[dict] = None,
    params: Optional[QueryParamTypes] = None,
    headers: Optional[dict[str, str]] = None
) -> AsyncIterator[StreamPart]

使用 SSE 流式传输结果。

AssistantsClient ¶

用于管理 LangGraph 中的智能体的客户端。

此类提供了与智能体交互的方法，智能体是您的图的版本化配置。

示例

client = get_client()
assistant = await client.assistants.get("assistant_id_123")

方法

名称	描述
`get`	按 ID 获取智能体。
`get_graph`	按 ID 获取智能体的图。
`get_schemas`	按 ID 获取智能体的模式。
`get_subgraphs`	按 ID 获取智能体的模式。
`create`	创建一个新的智能体。
`update`	更新智能体。
`delete`	删除智能体。
`search`	搜索智能体。
`get_versions`	列出智能体的所有版本。
`set_latest`	更改智能体的版本。

get `async` ¶

get(
    assistant_id: str,
    *,
    headers: Optional[dict[str, str]] = None
) -> Assistant

按 ID 获取智能体。

参数

名称	类型	描述	默认值
`assistant_id`	`str`	要获取的智能体的 ID。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`Assistant`	`Assistant`	智能体对象。

使用示例

assistant = await client.assistants.get(
    assistant_id="my_assistant_id"
)
print(assistant)

----------------------------------------------------

{
    'assistant_id': 'my_assistant_id',
    'graph_id': 'agent',
    'created_at': '2024-06-25T17:10:33.109781+00:00',
    'updated_at': '2024-06-25T17:10:33.109781+00:00',
    'config': {},
    'metadata': {'created_by': 'system'},
    'version': 1,
    'name': 'my_assistant'
}

get_graph `async` ¶

get_graph(
    assistant_id: str,
    *,
    xray: Union[int, bool] = False,
    headers: Optional[dict[str, str]] = None
) -> dict[str, list[dict[str, Any]]]

按 ID 获取智能体的图。

参数

名称	类型	描述	默认值
`assistant_id`	`str`	要获取其图的智能体的 ID。	必需
`xray`	`Union[int, bool]`	包含子图的图表示。如果提供了整数值，则仅包含深度小于或等于该值的子图。	`False`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`图`	`dict[str, list[dict[str, Any]]]`	智能体的图信息，采用 JSON 格式。

使用示例

graph_info = await client.assistants.get_graph(
    assistant_id="my_assistant_id"
)
print(graph_info)

--------------------------------------------------------------------------------------------------------------------------

{
    'nodes':
        [
            {'id': '__start__', 'type': 'schema', 'data': '__start__'},
            {'id': '__end__', 'type': 'schema', 'data': '__end__'},
            {'id': 'agent','type': 'runnable','data': {'id': ['langgraph', 'utils', 'RunnableCallable'],'name': 'agent'}},
        ],
    'edges':
        [
            {'source': '__start__', 'target': 'agent'},
            {'source': 'agent','target': '__end__'}
        ]
}

get_schemas `async` ¶

get_schemas(
    assistant_id: str,
    *,
    headers: Optional[dict[str, str]] = None
) -> GraphSchema

按 ID 获取智能体的模式。

参数

名称	类型	描述	默认值
`assistant_id`	`str`	要获取其模式的智能体的 ID。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`GraphSchema`	`GraphSchema`	智能体的图模式。

使用示例

schema = await client.assistants.get_schemas(
    assistant_id="my_assistant_id"
)
print(schema)

----------------------------------------------------------------------------------------------------------------------------

{
    'graph_id': 'agent',
    'state_schema':
        {
            'title': 'LangGraphInput',
            '$ref': '#/definitions/AgentState',
            'definitions':
                {
                    'BaseMessage':
                        {
                            'title': 'BaseMessage',
                            'description': 'Base abstract Message class. Messages are the inputs and outputs of ChatModels.',
                            'type': 'object',
                            'properties':
                                {
                                 'content':
                                    {
                                        'title': 'Content',
                                        'anyOf': [
                                            {'type': 'string'},
                                            {'type': 'array','items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]}}
                                        ]
                                    },
                                'additional_kwargs':
                                    {
                                        'title': 'Additional Kwargs',
                                        'type': 'object'
                                    },
                                'response_metadata':
                                    {
                                        'title': 'Response Metadata',
                                        'type': 'object'
                                    },
                                'type':
                                    {
                                        'title': 'Type',
                                        'type': 'string'
                                    },
                                'name':
                                    {
                                        'title': 'Name',
                                        'type': 'string'
                                    },
                                'id':
                                    {
                                        'title': 'Id',
                                        'type': 'string'
                                    }
                                },
                            'required': ['content', 'type']
                        },
                    'AgentState':
                        {
                            'title': 'AgentState',
                            'type': 'object',
                            'properties':
                                {
                                    'messages':
                                        {
                                            'title': 'Messages',
                                            'type': 'array',
                                            'items': {'$ref': '#/definitions/BaseMessage'}
                                        }
                                },
                            'required': ['messages']
                        }
                }
        },
    'config_schema':
        {
            'title': 'Configurable',
            'type': 'object',
            'properties':
                {
                    'model_name':
                        {
                            'title': 'Model Name',
                            'enum': ['anthropic', 'openai'],
                            'type': 'string'
                        }
                }
        }
}

get_subgraphs `async` ¶

get_subgraphs(
    assistant_id: str,
    namespace: Optional[str] = None,
    recurse: bool = False,
    *,
    headers: Optional[dict[str, str]] = None
) -> Subgraphs

按 ID 获取智能体的模式。

参数

名称	类型	描述	默认值
`assistant_id`	`str`	要获取其模式的智能体的 ID。	必需
`namespace`	`Optional[str]`	用于过滤的可选命名空间。	`无`
`recurse`	`bool`	是否递归获取子图。	`False`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`子图`	`子图`	智能体的图模式。

create `async` ¶

create(
    graph_id: Optional[str],
    config: Optional[Config] = None,
    *,
    metadata: Json = None,
    assistant_id: Optional[str] = None,
    if_exists: Optional[OnConflictBehavior] = None,
    name: Optional[str] = None,
    headers: Optional[dict[str, str]] = None,
    description: Optional[str] = None
) -> Assistant

创建一个新的智能体。

当图是可配置的并且您希望基于不同配置创建不同的智能体时很有用。

参数

名称	类型	描述	默认值
`graph_id`	`Optional[str]`	智能体应使用的图的 ID。图 ID 通常在您的 langgraph.json 配置中设置。	必需
`config`	`Optional[Config]`	用于图的配置。	`无`
`metadata`	`Json`	要添加到智能体的元数据。	`无`
`assistant_id`	`Optional[str]`	要使用的智能体 ID，如果未提供，将默认为随机 UUID。	`无`
`if_exists`	`Optional[OnConflictBehavior]`	如何处理重复创建。底层默认为 'raise'。必须是 'raise'（如果重复则抛出错误）或 'do_nothing'（返回现有智能体）。	`无`
`name`	`Optional[str]`	智能体的名称。底层默认为 'Untitled'。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`
`description`	`Optional[str]`	可选的智能体描述。description 字段适用于 langgraph-api 服务器版本 >= 0.0.45	`无`

返回值

名称	类型	描述
`Assistant`	`Assistant`	创建的智能体。

使用示例

assistant = await client.assistants.create(
    graph_id="agent",
    config={"configurable": {"model_name": "openai"}},
    metadata={"number":1},
    assistant_id="my-assistant-id",
    if_exists="do_nothing",
    name="my_name"
)

update `async` ¶

update(
    assistant_id: str,
    *,
    graph_id: Optional[str] = None,
    config: Optional[Config] = None,
    metadata: Json = None,
    name: Optional[str] = None,
    headers: Optional[dict[str, str]] = None,
    description: Optional[str] = None
) -> Assistant

更新智能体。

使用此方法指向不同的图、更新配置或更改智能体的元数据。

参数

名称	类型	描述	默认值
`assistant_id`	`str`	要更新的智能体。	必需
`graph_id`	`Optional[str]`	智能体应使用的图的 ID。图 ID 通常在您的 langgraph.json 配置中设置。如果为 None，智能体将继续指向同一个图。	`无`
`config`	`Optional[Config]`	用于图的配置。	`无`
`metadata`	`Json`	要与现有智能体元数据合并的元数据。	`无`
`name`	`Optional[str]`	智能体的新名称。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`
`description`	`Optional[str]`	可选的智能体描述。description 字段适用于 langgraph-api 服务器版本 >= 0.0.45	`无`

返回值

名称	类型	描述
`Assistant`	`Assistant`	更新后的智能体。

使用示例

assistant = await client.assistants.update(
    assistant_id='e280dad7-8618-443f-87f1-8e41841c180f',
    graph_id="other-graph",
    config={"configurable": {"model_name": "anthropic"}},
    metadata={"number":2}
)

delete `async` ¶

delete(
    assistant_id: str,
    *,
    headers: Optional[dict[str, str]] = None
) -> None

删除智能体。

参数

名称	类型	描述	默认值
`assistant_id`	`str`	要删除的智能体 ID。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`无`	无

使用示例

await client.assistants.delete(
    assistant_id="my_assistant_id"
)

search `async` ¶

search(
    *,
    metadata: Json = None,
    graph_id: Optional[str] = None,
    limit: int = 10,
    offset: int = 0,
    sort_by: Optional[AssistantSortBy] = None,
    sort_order: Optional[SortOrder] = None,
    headers: Optional[dict[str, str]] = None
) -> list[Assistant]

搜索智能体。

参数

名称	类型	描述	默认值
`metadata`	`Json`	用于过滤的元数据。每个 KV 对都进行精确匹配过滤。	`无`
`graph_id`	`Optional[str]`	用于过滤的图的 ID。图 ID 通常在您的 langgraph.json 配置中设置。	`无`
`limit`	`int`	要返回的最大结果数。	`10`
`offset`	`int`	要跳过的结果数。	`0`
`sort_by`	`Optional[AssistantSortBy]`	用于排序的字段。	`无`
`sort_order`	`Optional[SortOrder]`	排序顺序。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`list[Assistant]`	list[Assistant]：智能体列表。

使用示例

assistants = await client.assistants.search(
    metadata = {"name":"my_name"},
    graph_id="my_graph_id",
    limit=5,
    offset=5
)

get_versions `async` ¶

get_versions(
    assistant_id: str,
    metadata: Json = None,
    limit: int = 10,
    offset: int = 0,
    *,
    headers: Optional[dict[str, str]] = None
) -> list[AssistantVersion]

列出智能体的所有版本。

参数

名称	类型	描述	默认值
`assistant_id`	`str`	要获取其版本的智能体 ID。	必需
`metadata`	`Json`	用于过滤版本的元数据。每个 KV 对都进行精确匹配过滤。	`无`
`limit`	`int`	要返回的最大版本数。	`10`
`offset`	`int`	要跳过的版本数。	`0`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`list[AssistantVersion]`	list[AssistantVersion]：智能体版本列表。

使用示例

assistant_versions = await client.assistants.get_versions(
    assistant_id="my_assistant_id"
)

set_latest `async` ¶

set_latest(
    assistant_id: str,
    version: int,
    *,
    headers: Optional[dict[str, str]] = None
) -> Assistant

更改智能体的版本。

参数

名称	类型	描述	默认值
`assistant_id`	`str`	要删除的智能体 ID。	必需
`version`	`int`	要更改到的版本。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`Assistant`	`Assistant`	智能体对象。

使用示例

new_version_assistant = await client.assistants.set_latest(
    assistant_id="my_assistant_id",
    version=3
)

ThreadsClient ¶

用于管理 LangGraph 中的线程的客户端。

线程在多次交互/调用（即运行）中保持图的状态。它累积并持久化图的状态，从而允许图的独立调用之间保持连续性。

示例

client = get_client()
new_thread = await client.threads.create(metadata={"user_id": "123"})

方法

名称	描述
`get`	按 ID 获取线程。
`create`	创建一个新线程。
`update`	更新线程。
`delete`	删除线程。
`search`	搜索线程。
`copy`	复制线程。
`get_state`	获取线程的状态。
`update_state`	更新线程的状态。
`get_history`	获取线程的状态历史。

get `async` ¶

get(
    thread_id: str,
    *,
    headers: Optional[dict[str, str]] = None
) -> Thread

按 ID 获取线程。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要获取的线程 ID。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`Thread`	`Thread`	Thread 对象。

使用示例

thread = await client.threads.get(
    thread_id="my_thread_id"
)
print(thread)

-----------------------------------------------------

{
    'thread_id': 'my_thread_id',
    'created_at': '2024-07-18T18:35:15.540834+00:00',
    'updated_at': '2024-07-18T18:35:15.540834+00:00',
    'metadata': {'graph_id': 'agent'}
}

create `async` ¶

create(
    *,
    metadata: Json = None,
    thread_id: Optional[str] = None,
    if_exists: Optional[OnConflictBehavior] = None,
    supersteps: Optional[
        Sequence[dict[str, Sequence[dict[str, Any]]]]
    ] = None,
    graph_id: Optional[str] = None,
    headers: Optional[dict[str, str]] = None
) -> Thread

创建一个新线程。

参数

名称	类型	描述	默认值
`metadata`	`Json`	要添加到线程的元数据。	`无`
`thread_id`	`Optional[str]`	线程 ID。如果为 None，将生成随机 UUID 作为 ID。	`无`
`if_exists`	`Optional[OnConflictBehavior]`	如何处理重复创建。底层默认值为 'raise'。必须是 'raise'（如果重复则抛出错误）或 'do_nothing'（返回现有线程）。	`无`
`supersteps`	`Optional[Sequence[dict[str, Sequence[dict[str, Any]]]]]`	创建线程时应用 supersteps 列表，每个 superstep 包含一系列更新。每个更新都有 `values` 或 `command` 以及 `as_node`。用于在不同部署之间复制线程。	`无`
`graph_id`	`Optional[str]`	可选的图 ID，用于与线程关联。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`Thread`	`Thread`	创建的线程。

使用示例

thread = await client.threads.create(
    metadata={"number":1},
    thread_id="my-thread-id",
    if_exists="raise"
)

update `async` ¶

update(
    thread_id: str,
    *,
    metadata: dict[str, Any],
    headers: Optional[dict[str, str]] = None
) -> Thread

更新线程。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要更新的线程 ID。	必需
`metadata`	`dict[str, Any]`	要与现有线程元数据合并的元数据。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`Thread`	`Thread`	创建的线程。

使用示例

thread = await client.threads.update(
    thread_id="my-thread-id",
    metadata={"number":1},
)

delete `async` ¶

delete(
    thread_id: str,
    *,
    headers: Optional[dict[str, str]] = None
) -> None

删除线程。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要删除的线程 ID。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`无`	无

使用示例

await client.threads.delete(
    thread_id="my_thread_id"
)

search `async` ¶

search(
    *,
    metadata: Json = None,
    values: Json = None,
    status: Optional[ThreadStatus] = None,
    limit: int = 10,
    offset: int = 0,
    sort_by: Optional[ThreadSortBy] = None,
    sort_order: Optional[SortOrder] = None,
    headers: Optional[dict[str, str]] = None
) -> list[Thread]

搜索线程。

参数

名称	类型	描述	默认值
`metadata`	`Json`	用于过滤的线程元数据。	`无`
`values`	`Json`	用于过滤的状态值。	`无`
`status`	`Optional[ThreadStatus]`	用于过滤的线程状态。必须是 'idle'、'busy'、'interrupted' 或 'error' 中的一个。	`无`
`limit`	`int`	返回的线程数量限制。	`10`
`offset`	`int`	在线程表中开始搜索的偏移量。	`0`
`sort_by`	`Optional[ThreadSortBy]`	排序字段。	`无`
`sort_order`	`Optional[SortOrder]`	排序顺序。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`list[Thread]`	list[Thread]: 匹配搜索参数的线程列表。

使用示例

threads = await client.threads.search(
    metadata={"number":1},
    status="interrupted",
    limit=15,
    offset=5
)

copy `async` ¶

copy(
    thread_id: str,
    *,
    headers: Optional[dict[str, str]] = None
) -> None

复制线程。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要复制的线程 ID。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`无`	无

使用示例

await client.threads.copy(
    thread_id="my_thread_id"
)

get_state `async` ¶

get_state(
    thread_id: str,
    checkpoint: Optional[Checkpoint] = None,
    checkpoint_id: Optional[str] = None,
    *,
    subgraphs: bool = False,
    headers: Optional[dict[str, str]] = None
) -> ThreadState

获取线程的状态。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要获取状态的线程 ID。	必需
`checkpoint`	`Optional[Checkpoint]`	要获取状态的检查点。	`无`
`checkpoint_id`	`Optional[str]`	(已弃用) 要获取状态的检查点 ID。	`无`
`subgraphs`	`bool`	包含子图状态。	`False`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`ThreadState`	`ThreadState`	该状态所属的线程。

使用示例

thread_state = await client.threads.get_state(
    thread_id="my_thread_id",
    checkpoint_id="my_checkpoint_id"
)
print(thread_state)

----------------------------------------------------------------------------------------------------------------------------------------------------------------------

{
    'values': {
        'messages': [
            {
                'content': 'how are you?',
                'additional_kwargs': {},
                'response_metadata': {},
                'type': 'human',
                'name': None,
                'id': 'fe0a5778-cfe9-42ee-b807-0adaa1873c10',
                'example': False
            },
            {
                'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.",
                'additional_kwargs': {},
                'response_metadata': {},
                'type': 'ai',
                'name': None,
                'id': 'run-159b782c-b679-4830-83c6-cef87798fe8b',
                'example': False,
                'tool_calls': [],
                'invalid_tool_calls': [],
                'usage_metadata': None
            }
        ]
    },
    'next': [],
    'checkpoint':
        {
            'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2',
            'checkpoint_ns': '',
            'checkpoint_id': '1ef4a9b8-e6fb-67b1-8001-abd5184439d1'
        }
    'metadata':
        {
            'step': 1,
            'run_id': '1ef4a9b8-d7da-679a-a45a-872054341df2',
            'source': 'loop',
            'writes':
                {
                    'agent':
                        {
                            'messages': [
                                {
                                    'id': 'run-159b782c-b679-4830-83c6-cef87798fe8b',
                                    'name': None,
                                    'type': 'ai',
                                    'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.",
                                    'example': False,
                                    'tool_calls': [],
                                    'usage_metadata': None,
                                    'additional_kwargs': {},
                                    'response_metadata': {},
                                    'invalid_tool_calls': []
                                }
                            ]
                        }
                },
    'user_id': None,
    'graph_id': 'agent',
    'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2',
    'created_by': 'system',
    'assistant_id': 'fe096781-5601-53d2-b2f6-0d3403f7e9ca'},
    'created_at': '2024-07-25T15:35:44.184703+00:00',
    'parent_config':
        {
            'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2',
            'checkpoint_ns': '',
            'checkpoint_id': '1ef4a9b8-d80d-6fa7-8000-9300467fad0f'
        }
}

update_state `async` ¶

update_state(
    thread_id: str,
    values: Optional[Union[dict, Sequence[dict]]],
    *,
    as_node: Optional[str] = None,
    checkpoint: Optional[Checkpoint] = None,
    checkpoint_id: Optional[str] = None,
    headers: Optional[dict[str, str]] = None
) -> ThreadUpdateStateResponse

更新线程的状态。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要更新的线程 ID。	必需
`values`	`Optional[Union[dict, Sequence[dict]]]`	用于更新状态的值。	必需
`as_node`	`Optional[str]`	更新状态，如同此节点刚刚执行完毕。	`无`
`checkpoint`	`Optional[Checkpoint]`	要更新状态的检查点。	`无`
`checkpoint_id`	`Optional[str]`	(已弃用) 要更新状态的检查点 ID。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`ThreadUpdateStateResponse`	`ThreadUpdateStateResponse`	更新线程状态后的响应。

使用示例

response = await client.threads.update_state(
    thread_id="my_thread_id",
    values={"messages":[{"role": "user", "content": "hello!"}]},
    as_node="my_node",
)
print(response)

----------------------------------------------------------------------------------------------------------------------------------------------------------------------

{
    'checkpoint': {
        'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2',
        'checkpoint_ns': '',
        'checkpoint_id': '1ef4a9b8-e6fb-67b1-8001-abd5184439d1',
        'checkpoint_map': {}
    }
}

get_history `async` ¶

get_history(
    thread_id: str,
    *,
    limit: int = 10,
    before: Optional[str | Checkpoint] = None,
    metadata: Optional[dict] = None,
    checkpoint: Optional[Checkpoint] = None,
    headers: Optional[dict[str, str]] = None
) -> list[ThreadState]

获取线程的状态历史。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要获取状态历史的线程 ID。	必需
`checkpoint`	`Optional[Checkpoint]`	返回此子图的状态。如果为空，默认为根图。	`无`
`limit`	`int`	返回的最大状态数量。	`10`
`before`	`Optional[str \| Checkpoint]`	返回此检查点之前的状态。	`无`
`metadata`	`Optional[dict]`	按元数据键值对过滤状态。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`list[ThreadState]`	list[ThreadState]: 线程的状态历史。

使用示例

thread_state = await client.threads.get_history(
    thread_id="my_thread_id",
    limit=5,
)

RunsClient ¶

用于管理 LangGraph 中的运行的客户端。

运行（Run）是助手的单次调用，可包含可选的输入、配置和元数据。此客户端管理运行，运行可以是带状态的（在线程上）或无状态的。

示例

client = get_client()
run = await client.runs.create(assistant_id="asst_123", thread_id="thread_456", input={"query": "Hello"})

方法

名称	描述
`stream`	创建运行并流式传输结果。
`create`	创建后台运行。
`create_batch`	创建一批无状态的后台运行。
`wait`	创建运行，等待其完成并返回最终状态。
`list`	列出运行。
`get`	获取运行。
`cancel`	获取运行。
`join`	阻塞直到运行完成。返回线程的最终状态。
`join_stream`	实时流式传输运行的输出，直到运行完成。
`delete`	删除运行。

stream ¶

stream(
    thread_id: Optional[str],
    assistant_id: str,
    *,
    input: Optional[dict] = None,
    command: Optional[Command] = None,
    stream_mode: Union[
        StreamMode, Sequence[StreamMode]
    ] = "values",
    stream_subgraphs: bool = False,
    metadata: Optional[dict] = None,
    config: Optional[Config] = None,
    checkpoint: Optional[Checkpoint] = None,
    checkpoint_id: Optional[str] = None,
    checkpoint_during: Optional[bool] = None,
    interrupt_before: Optional[
        Union[All, Sequence[str]]
    ] = None,
    interrupt_after: Optional[
        Union[All, Sequence[str]]
    ] = None,
    feedback_keys: Optional[Sequence[str]] = None,
    on_disconnect: Optional[DisconnectMode] = None,
    on_completion: Optional[OnCompletionBehavior] = None,
    webhook: Optional[str] = None,
    multitask_strategy: Optional[MultitaskStrategy] = None,
    if_not_exists: Optional[IfNotExists] = None,
    after_seconds: Optional[int] = None,
    headers: Optional[dict[str, str]] = None
) -> AsyncIterator[StreamPart]

创建运行并流式传输结果。

参数

名称	类型	描述	默认值
`thread_id`	`Optional[str]`	分配给线程的线程 ID。如果为 None，将创建无状态运行。	必需
`assistant_id`	`str`	要从中流式传输的助手 ID 或图名称。如果使用图名称，默认为从该图创建的第一个助手。	必需
`input`	`Optional[dict]`	图的输入。	`无`
`command`	`Optional[Command]`	要执行的命令。不能与 input 结合使用。	`无`
`stream_mode`	`Union[StreamMode, Sequence[StreamMode]]`	要使用的流模式。	`'values'`
`stream_subgraphs`	`bool`	是否流式传输子图的输出。	`False`
`metadata`	`Optional[dict]`	分配给运行的元数据。	`无`
`config`	`Optional[Config]`	助手的配置。	`无`
`checkpoint`	`Optional[Checkpoint]`	要从中恢复的检查点。	`无`
`checkpoint_during`	`Optional[bool]`	是否在运行期间（或仅在结束/中断时）创建检查点。	`无`
`interrupt_before`	`Optional[Union[All, Sequence[str]]]`	在节点执行之前立即中断。	`无`
`interrupt_after`	`Optional[Union[All, Sequence[str]]]`	在节点执行之后立即中断。	`无`
`feedback_keys`	`Optional[Sequence[str]]`	分配给运行的反馈键。	`无`
`on_disconnect`	`Optional[DisconnectMode]`	要使用的断开连接模式。必须是 'cancel' 或 'continue' 中的一个。	`无`
`on_completion`	`Optional[OnCompletionBehavior]`	是否删除或保留为无状态运行创建的线程。必须是 'delete' 或 'keep' 中的一个。	`无`
`webhook`	`Optional[str]`	LangGraph API 调用完成后要调用的 Webhook。	`无`
`multitask_strategy`	`Optional[MultitaskStrategy]`	要使用的多任务策略。必须是 'reject'、'interrupt'、'rollback' 或 'enqueue' 中的一个。	`无`
`if_not_exists`	`Optional[IfNotExists]`	如何处理缺失的线程。默认为 'reject'。必须是 'reject'（如果缺失则抛出错误）或 'create'（创建新线程）。	`无`
`after_seconds`	`Optional[int]`	开始运行前等待的秒数。用于安排未来的运行。	`无`

返回值

类型	描述
`AsyncIterator[StreamPart]`	AsyncIterator[StreamPart]: 流结果的异步迭代器。

使用示例

async for chunk in client.runs.stream(
    thread_id=None,
    assistant_id="agent",
    input={"messages": [{"role": "user", "content": "how are you?"}]},
    stream_mode=["values","debug"],
    metadata={"name":"my_run"},
    config={"configurable": {"model_name": "anthropic"}},
    interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"],
    interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"],
    feedback_keys=["my_feedback_key_1","my_feedback_key_2"],
    webhook="https://my.fake.webhook.com",
    multitask_strategy="interrupt"
):
    print(chunk)

------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------

StreamPart(event='metadata', data={'run_id': '1ef4a9b8-d7da-679a-a45a-872054341df2'})
StreamPart(event='values', data={'messages': [{'content': 'how are you?', 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'human', 'name': None, 'id': 'fe0a5778-cfe9-42ee-b807-0adaa1873c10', 'example': False}]})
StreamPart(event='values', data={'messages': [{'content': 'how are you?', 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'human', 'name': None, 'id': 'fe0a5778-cfe9-42ee-b807-0adaa1873c10', 'example': False}, {'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.", 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'ai', 'name': None, 'id': 'run-159b782c-b679-4830-83c6-cef87798fe8b', 'example': False, 'tool_calls': [], 'invalid_tool_calls': [], 'usage_metadata': None}]})
StreamPart(event='end', data=None)

create `async` ¶

create(
    thread_id: Optional[str],
    assistant_id: str,
    *,
    input: Optional[dict] = None,
    command: Optional[Command] = None,
    stream_mode: Union[
        StreamMode, Sequence[StreamMode]
    ] = "values",
    stream_subgraphs: bool = False,
    metadata: Optional[dict] = None,
    config: Optional[Config] = None,
    checkpoint: Optional[Checkpoint] = None,
    checkpoint_id: Optional[str] = None,
    checkpoint_during: Optional[bool] = None,
    interrupt_before: Optional[
        Union[All, Sequence[str]]
    ] = None,
    interrupt_after: Optional[
        Union[All, Sequence[str]]
    ] = None,
    webhook: Optional[str] = None,
    multitask_strategy: Optional[MultitaskStrategy] = None,
    if_not_exists: Optional[IfNotExists] = None,
    on_completion: Optional[OnCompletionBehavior] = None,
    after_seconds: Optional[int] = None,
    headers: Optional[dict[str, str]] = None
) -> Run

创建后台运行。

参数

名称	类型	描述	默认值
`thread_id`	`Optional[str]`	分配给线程的线程 ID。如果为 None，将创建无状态运行。	必需
`assistant_id`	`str`	要从中流式传输的助手 ID 或图名称。如果使用图名称，默认为从该图创建的第一个助手。	必需
`input`	`Optional[dict]`	图的输入。	`无`
`command`	`Optional[Command]`	要执行的命令。不能与 input 结合使用。	`无`
`stream_mode`	`Union[StreamMode, Sequence[StreamMode]]`	要使用的流模式。	`'values'`
`stream_subgraphs`	`bool`	是否流式传输子图的输出。	`False`
`metadata`	`Optional[dict]`	分配给运行的元数据。	`无`
`config`	`Optional[Config]`	助手的配置。	`无`
`checkpoint`	`Optional[Checkpoint]`	要从中恢复的检查点。	`无`
`checkpoint_during`	`Optional[bool]`	是否在运行期间（或仅在结束/中断时）创建检查点。	`无`
`interrupt_before`	`Optional[Union[All, Sequence[str]]]`	在节点执行之前立即中断。	`无`
`interrupt_after`	`Optional[Union[All, Sequence[str]]]`	在节点执行之后立即中断。	`无`
`webhook`	`Optional[str]`	LangGraph API 调用完成后要调用的 Webhook。	`无`
`multitask_strategy`	`Optional[MultitaskStrategy]`	要使用的多任务策略。必须是 'reject'、'interrupt'、'rollback' 或 'enqueue' 中的一个。	`无`
`on_completion`	`Optional[OnCompletionBehavior]`	是否删除或保留为无状态运行创建的线程。必须是 'delete' 或 'keep' 中的一个。	`无`
`if_not_exists`	`Optional[IfNotExists]`	如何处理缺失的线程。默认为 'reject'。必须是 'reject'（如果缺失则抛出错误）或 'create'（创建新线程）。	`无`
`after_seconds`	`Optional[int]`	开始运行前等待的秒数。用于安排未来的运行。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`Run`	`Run`	创建的后台运行。

使用示例

background_run = await client.runs.create(
    thread_id="my_thread_id",
    assistant_id="my_assistant_id",
    input={"messages": [{"role": "user", "content": "hello!"}]},
    metadata={"name":"my_run"},
    config={"configurable": {"model_name": "openai"}},
    interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"],
    interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"],
    webhook="https://my.fake.webhook.com",
    multitask_strategy="interrupt"
)
print(background_run)

--------------------------------------------------------------------------------

{
    'run_id': 'my_run_id',
    'thread_id': 'my_thread_id',
    'assistant_id': 'my_assistant_id',
    'created_at': '2024-07-25T15:35:42.598503+00:00',
    'updated_at': '2024-07-25T15:35:42.598503+00:00',
    'metadata': {},
    'status': 'pending',
    'kwargs':
        {
            'input':
                {
                    'messages': [
                        {
                            'role': 'user',
                            'content': 'how are you?'
                        }
                    ]
                },
            'config':
                {
                    'metadata':
                        {
                            'created_by': 'system'
                        },
                    'configurable':
                        {
                            'run_id': 'my_run_id',
                            'user_id': None,
                            'graph_id': 'agent',
                            'thread_id': 'my_thread_id',
                            'checkpoint_id': None,
                            'model_name': "openai",
                            'assistant_id': 'my_assistant_id'
                        }
                },
            'webhook': "https://my.fake.webhook.com",
            'temporary': False,
            'stream_mode': ['values'],
            'feedback_keys': None,
            'interrupt_after': ["node_to_stop_after_1","node_to_stop_after_2"],
            'interrupt_before': ["node_to_stop_before_1","node_to_stop_before_2"]
        },
    'multitask_strategy': 'interrupt'
}

create_batch `async` ¶

create_batch(payloads: list[RunCreate]) -> list[Run]

创建一批无状态的后台运行。

wait `async` ¶

wait(
    thread_id: Optional[str],
    assistant_id: str,
    *,
    input: Optional[dict] = None,
    command: Optional[Command] = None,
    metadata: Optional[dict] = None,
    config: Optional[Config] = None,
    checkpoint: Optional[Checkpoint] = None,
    checkpoint_id: Optional[str] = None,
    checkpoint_during: Optional[bool] = None,
    interrupt_before: Optional[
        Union[All, Sequence[str]]
    ] = None,
    interrupt_after: Optional[
        Union[All, Sequence[str]]
    ] = None,
    webhook: Optional[str] = None,
    on_disconnect: Optional[DisconnectMode] = None,
    on_completion: Optional[OnCompletionBehavior] = None,
    multitask_strategy: Optional[MultitaskStrategy] = None,
    if_not_exists: Optional[IfNotExists] = None,
    after_seconds: Optional[int] = None,
    raise_error: bool = True,
    headers: Optional[dict[str, str]] = None
) -> Union[list[dict], dict[str, Any]]

创建运行，等待其完成并返回最终状态。

参数

名称	类型	描述	默认值
`thread_id`	`Optional[str]`	创建运行的线程 ID。如果为 None，将创建无状态运行。	必需
`assistant_id`	`str`	要运行的助手 ID 或图名称。如果使用图名称，默认为从该图创建的第一个助手。	必需
`input`	`Optional[dict]`	图的输入。	`无`
`command`	`Optional[Command]`	要执行的命令。不能与 input 结合使用。	`无`
`metadata`	`Optional[dict]`	分配给运行的元数据。	`无`
`config`	`Optional[Config]`	助手的配置。	`无`
`checkpoint`	`Optional[Checkpoint]`	要从中恢复的检查点。	`无`
`checkpoint_during`	`Optional[bool]`	是否在运行期间（或仅在结束/中断时）创建检查点。	`无`
`interrupt_before`	`Optional[Union[All, Sequence[str]]]`	在节点执行之前立即中断。	`无`
`interrupt_after`	`Optional[Union[All, Sequence[str]]]`	在节点执行之后立即中断。	`无`
`webhook`	`Optional[str]`	LangGraph API 调用完成后要调用的 Webhook。	`无`
`on_disconnect`	`Optional[DisconnectMode]`	要使用的断开连接模式。必须是 'cancel' 或 'continue' 中的一个。	`无`
`on_completion`	`Optional[OnCompletionBehavior]`	是否删除或保留为无状态运行创建的线程。必须是 'delete' 或 'keep' 中的一个。	`无`
`multitask_strategy`	`Optional[MultitaskStrategy]`	要使用的多任务策略。必须是 'reject'、'interrupt'、'rollback' 或 'enqueue' 中的一个。	`无`
`if_not_exists`	`Optional[IfNotExists]`	如何处理缺失的线程。默认为 'reject'。必须是 'reject'（如果缺失则抛出错误）或 'create'（创建新线程）。	`无`
`after_seconds`	`Optional[int]`	开始运行前等待的秒数。用于安排未来的运行。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`Union[list[dict], dict[str, Any]]`	Union[list[dict], dict[str, Any]]: 运行的输出。

使用示例

final_state_of_run = await client.runs.wait(
    thread_id=None,
    assistant_id="agent",
    input={"messages": [{"role": "user", "content": "how are you?"}]},
    metadata={"name":"my_run"},
    config={"configurable": {"model_name": "anthropic"}},
    interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"],
    interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"],
    webhook="https://my.fake.webhook.com",
    multitask_strategy="interrupt"
)
print(final_state_of_run)

-------------------------------------------------------------------------------------------------------------------------------------------

{
    'messages': [
        {
            'content': 'how are you?',
            'additional_kwargs': {},
            'response_metadata': {},
            'type': 'human',
            'name': None,
            'id': 'f51a862c-62fe-4866-863b-b0863e8ad78a',
            'example': False
        },
        {
            'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.",
            'additional_kwargs': {},
            'response_metadata': {},
            'type': 'ai',
            'name': None,
            'id': 'run-bf1cd3c6-768f-4c16-b62d-ba6f17ad8b36',
            'example': False,
            'tool_calls': [],
            'invalid_tool_calls': [],
            'usage_metadata': None
        }
    ]
}

list `async` ¶

list(
    thread_id: str,
    *,
    limit: int = 10,
    offset: int = 0,
    status: Optional[RunStatus] = None,
    headers: Optional[dict[str, str]] = None
) -> List[Run]

列出运行。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要列出运行的线程 ID。	必需
`limit`	`int`	要返回的最大结果数。	`10`
`offset`	`int`	要跳过的结果数。	`0`
`status`	`Optional[RunStatus]`	用于过滤的运行状态。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`List[Run]`	List[Run]: 线程的运行列表。

使用示例

await client.runs.list(
    thread_id="thread_id",
    limit=5,
    offset=5,
)

get `async` ¶

get(
    thread_id: str,
    run_id: str,
    *,
    headers: Optional[dict[str, str]] = None
) -> Run

获取运行。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要获取的线程 ID。	必需
`run_id`	`str`	要获取的运行 ID。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`Run`	`Run`	运行对象。

使用示例

run = await client.runs.get(
    thread_id="thread_id_to_delete",
    run_id="run_id_to_delete",
)

cancel `async` ¶

cancel(
    thread_id: str,
    run_id: str,
    *,
    wait: bool = False,
    action: CancelAction = "interrupt",
    headers: Optional[dict[str, str]] = None
) -> None

获取运行。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要取消的线程 ID。	必需
`run_id`	`str`	要取消的运行 ID。	必需
`wait`	`bool`	是否等待运行完成。	`False`
`action`	`CancelAction`	取消运行时采取的操作。可能的值为 `interrupt` 或 `rollback`。默认值为 `interrupt`。	`'interrupt'`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`无`	无

使用示例

await client.runs.cancel(
    thread_id="thread_id_to_cancel",
    run_id="run_id_to_cancel",
    wait=True,
    action="interrupt"
)

join `async` ¶

join(
    thread_id: str,
    run_id: str,
    *,
    headers: Optional[dict[str, str]] = None
) -> dict

阻塞直到运行完成。返回线程的最终状态。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要等待完成的线程 ID。	必需
`run_id`	`str`	要等待完成的运行 ID。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`dict`	无

使用示例

result =await client.runs.join(
    thread_id="thread_id_to_join",
    run_id="run_id_to_join"
)

join_stream ¶

join_stream(
    thread_id: str,
    run_id: str,
    *,
    cancel_on_disconnect: bool = False,
    stream_mode: Optional[
        Union[StreamMode, Sequence[StreamMode]]
    ] = None,
    headers: Optional[dict[str, str]] = None
) -> AsyncIterator[StreamPart]

实时流式传输运行的输出，直到运行完成。输出不带缓冲，因此在此调用之前产生的任何输出都不会在此处接收到。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要等待完成的线程 ID。	必需
`run_id`	`str`	要等待完成的运行 ID。	必需
`cancel_on_disconnect`	`bool`	流断开连接时是否取消运行。	`False`
`stream_mode`	`Optional[Union[StreamMode, Sequence[StreamMode]]]`	要使用的流模式。必须是创建运行时光传递的流模式的子集。后台运行默认为所有流模式的并集。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`AsyncIterator[StreamPart]`	无

使用示例

await client.runs.join_stream(
    thread_id="thread_id_to_join",
    run_id="run_id_to_join",
    stream_mode=["values", "debug"]
)

delete `async` ¶

delete(
    thread_id: str,
    run_id: str,
    *,
    headers: Optional[dict[str, str]] = None
) -> None

删除运行。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要删除的线程 ID。	必需
`run_id`	`str`	要删除的运行 ID。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`无`	无

使用示例

await client.runs.delete(
    thread_id="thread_id_to_delete",
    run_id="run_id_to_delete"
)

CronClient ¶

用于管理 LangGraph 中的周期性运行（cron 作业）的客户端。

运行（Run）是助手的单次调用，可包含可选的输入和配置。此客户端允许安排定期运行以自动执行。

示例

client = get_client()
cron_job = await client.crons.create_for_thread(
    thread_id="thread_123",
    assistant_id="asst_456",
    schedule="0 9 * * *",
    input={"message": "Daily update"}
)

方法

名称	描述
`create_for_thread`	为线程创建一个 cron 作业。
`create`	创建 cron 运行。
`delete`	删除 cron。
`search`	获取 cron 作业列表。

create_for_thread `async` ¶

create_for_thread(
    thread_id: str,
    assistant_id: str,
    *,
    schedule: str,
    input: Optional[dict] = None,
    metadata: Optional[dict] = None,
    config: Optional[Config] = None,
    checkpoint_during: Optional[bool] = None,
    interrupt_before: Optional[
        Union[All, list[str]]
    ] = None,
    interrupt_after: Optional[Union[All, list[str]]] = None,
    webhook: Optional[str] = None,
    multitask_strategy: Optional[str] = None,
    headers: Optional[dict[str, str]] = None
) -> Run

为线程创建一个 cron 作业。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要运行 cron 作业的线程 ID。	必需
`assistant_id`	`str`	用于 cron 作业的助手 ID 或图名称。如果使用图名称，默认为从该图创建的第一个助手。	必需
`schedule`	`str`	执行此作业的 cron 计划表。	必需
`input`	`Optional[dict]`	图的输入。	`无`
`metadata`	`Optional[dict]`	分配给 cron 作业运行的元数据。	`无`
`config`	`Optional[Config]`	助手的配置。	`无`
`checkpoint_during`	`Optional[bool]`	是否在运行期间（或仅在结束/中断时）创建检查点。	`无`
`interrupt_before`	`Optional[Union[All, list[str]]]`	在节点执行之前立即中断。	`无`
`interrupt_after`	`Optional[Union[All, list[str]]]`	在节点执行之后立即中断。	`无`
`webhook`	`Optional[str]`	LangGraph API 调用完成后要调用的 Webhook。	`无`
`multitask_strategy`	`Optional[str]`	要使用的多任务策略。必须是 'reject'、'interrupt'、'rollback' 或 'enqueue' 中的一个。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`Run`	`Run`	cron 运行。

使用示例

cron_run = await client.crons.create_for_thread(
    thread_id="my-thread-id",
    assistant_id="agent",
    schedule="27 15 * * *",
    input={"messages": [{"role": "user", "content": "hello!"}]},
    metadata={"name":"my_run"},
    config={"configurable": {"model_name": "openai"}},
    interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"],
    interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"],
    webhook="https://my.fake.webhook.com",
    multitask_strategy="interrupt"
)

create `async` ¶

create(
    assistant_id: str,
    *,
    schedule: str,
    input: Optional[dict] = None,
    metadata: Optional[dict] = None,
    config: Optional[Config] = None,
    checkpoint_during: Optional[bool] = None,
    interrupt_before: Optional[
        Union[All, list[str]]
    ] = None,
    interrupt_after: Optional[Union[All, list[str]]] = None,
    webhook: Optional[str] = None,
    multitask_strategy: Optional[str] = None,
    headers: Optional[dict[str, str]] = None
) -> Run

创建 cron 运行。

参数

名称	类型	描述	默认值
`assistant_id`	`str`	用于 cron 作业的助手 ID 或图名称。如果使用图名称，默认为从该图创建的第一个助手。	必需
`schedule`	`str`	执行此作业的 cron 计划表。	必需
`input`	`Optional[dict]`	图的输入。	`无`
`metadata`	`Optional[dict]`	分配给 cron 作业运行的元数据。	`无`
`config`	`Optional[Config]`	助手的配置。	`无`
`checkpoint_during`	`Optional[bool]`	是否在运行期间（或仅在结束/中断时）创建检查点。	`无`
`interrupt_before`	`Optional[Union[All, list[str]]]`	在节点执行之前立即中断。	`无`
`interrupt_after`	`Optional[Union[All, list[str]]]`	在节点执行之后立即中断。	`无`
`webhook`	`Optional[str]`	LangGraph API 调用完成后要调用的 Webhook。	`无`
`multitask_strategy`	`Optional[str]`	要使用的多任务策略。必须是 'reject'、'interrupt'、'rollback' 或 'enqueue' 中的一个。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`Run`	`Run`	cron 运行。

使用示例

cron_run = client.crons.create(
    assistant_id="agent",
    schedule="27 15 * * *",
    input={"messages": [{"role": "user", "content": "hello!"}]},
    metadata={"name":"my_run"},
    config={"configurable": {"model_name": "openai"}},
    interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"],
    interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"],
    webhook="https://my.fake.webhook.com",
    multitask_strategy="interrupt"
)

delete `async` ¶

delete(
    cron_id: str, headers: Optional[dict[str, str]] = None
) -> None

删除 cron。

参数

名称	类型	描述	默认值
`cron_id`	`str`	要删除的 cron ID。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`无`	无

使用示例

await client.crons.delete(
    cron_id="cron_to_delete"
)

search `async` ¶

search(
    *,
    assistant_id: Optional[str] = None,
    thread_id: Optional[str] = None,
    limit: int = 10,
    offset: int = 0,
    headers: Optional[dict[str, str]] = None
) -> list[Cron]

获取 cron 作业列表。

参数

名称	类型	描述	默认值
`assistant_id`	`Optional[str]`	要搜索的助手 ID 或图名称。	`无`
`thread_id`	`Optional[str]`	要搜索的线程 ID。	`无`
`limit`	`int`	要返回的最大结果数。	`10`
`offset`	`int`	要跳过的结果数。	`0`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`list[Cron]`	list[Cron]: 搜索返回的 cron 作业列表，

使用示例

cron_jobs = await client.crons.search(
    assistant_id="my_assistant_id",
    thread_id="my_thread_id",
    limit=5,
    offset=5,
)
print(cron_jobs)

----------------------------------------------------------

[
    {
        'cron_id': '1ef3cefa-4c09-6926-96d0-3dc97fd5e39b',
        'assistant_id': 'my_assistant_id',
        'thread_id': 'my_thread_id',
        'user_id': None,
        'payload':
            {
                'input': {'start_time': ''},
                'schedule': '4 * * * *',
                'assistant_id': 'my_assistant_id'
            },
        'schedule': '4 * * * *',
        'next_run_date': '2024-07-25T17:04:00+00:00',
        'end_time': None,
        'created_at': '2024-07-08T06:02:23.073257+00:00',
        'updated_at': '2024-07-08T06:02:23.073257+00:00'
    }
]

StoreClient ¶

用于与图的共享存储进行交互的客户端。

Store 提供了一个键值存储系统，用于在图执行之间持久化数据，从而实现有状态操作和跨线程数据共享。

示例

client = get_client()
await client.store.put_item(["users", "user123"], "mem-123451342", {"name": "Alice", "score": 100})

方法

名称	描述
`put_item`	存储或更新一个项。
`get_item`	检索单个项。
`delete_item`	删除一个项。
`search_items`	在命名空间前缀内搜索项。
`list_namespaces`	列出命名空间，可带可选匹配条件。

put_item `async` ¶

put_item(
    namespace: Sequence[str],
    /,
    key: str,
    value: dict[str, Any],
    index: Optional[
        Union[Literal[False], list[str]]
    ] = None,
    ttl: Optional[int] = None,
    headers: Optional[dict[str, str]] = None,
) -> None

存储或更新一个项。

参数

名称	类型	描述	默认值
`namespace`	`Sequence[str]`	表示命名空间路径的字符串列表。	必需
`key`	`str`	项在命名空间内的唯一标识符。	必需
`value`	`dict[str, Any]`	包含项数据的字典。	必需
`index`	`Optional[Union[Literal[False], list[str]]]`	控制搜索索引 - None（使用默认值），False（禁用），或要索引的字段路径列表。	`无`
`ttl`	`Optional[int]`	项的可选生存时间（分钟），或 None 表示不设过期。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`无`	无

使用示例

await client.store.put_item(
    ["documents", "user123"],
    key="item456",
    value={"title": "My Document", "content": "Hello World"}
)

get_item `async` ¶

get_item(
    namespace: Sequence[str],
    /,
    key: str,
    *,
    refresh_ttl: Optional[bool] = None,
    headers: Optional[dict[str, str]] = None,
) -> Item

检索单个项。

参数

名称	类型	描述	默认值
`key`	`str`	项的唯一标识符。	必需
`namespace`	`Sequence[str]`	表示命名空间路径的可选字符串列表。	必需
`refresh_ttl`	`Optional[bool]`	在此读取操作时是否刷新 TTL。如果为 None，则使用存储的默认行为。	`无`

返回值

名称	类型	描述
`Item`	`Item`	检索到的项。
`headers`	`Item`	可选的自定义头，包含在请求中。

使用示例

item = await client.store.get_item(
    ["documents", "user123"],
    key="item456",
)
print(item)

----------------------------------------------------------------

{
    'namespace': ['documents', 'user123'],
    'key': 'item456',
    'value': {'title': 'My Document', 'content': 'Hello World'},
    'created_at': '2024-07-30T12:00:00Z',
    'updated_at': '2024-07-30T12:00:00Z'
}

delete_item `async` ¶

delete_item(
    namespace: Sequence[str],
    /,
    key: str,
    headers: Optional[dict[str, str]] = None,
) -> None

删除一个项。

参数

名称	类型	描述	默认值
`key`	`str`	项的唯一标识符。	必需
`namespace`	`Sequence[str]`	表示命名空间路径的可选字符串列表。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`无`	无

使用示例

await client.store.delete_item(
    ["documents", "user123"],
    key="item456",
)

search_items `async` ¶

search_items(
    namespace_prefix: Sequence[str],
    /,
    filter: Optional[dict[str, Any]] = None,
    limit: int = 10,
    offset: int = 0,
    query: Optional[str] = None,
    refresh_ttl: Optional[bool] = None,
    headers: Optional[dict[str, str]] = None,
) -> SearchItemsResponse

在命名空间前缀内搜索项。

参数

名称	类型	描述	默认值
`namespace_prefix`	`Sequence[str]`	表示命名空间前缀的字符串列表。	必需
`filter`	`Optional[dict[str, Any]]`	用于过滤结果的可选键值对字典。	`无`
`limit`	`int`	返回的最大项数量（默认为 10）。	`10`
`offset`	`int`	返回结果前要跳过的项数量（默认为 0）。	`0`
`query`	`Optional[str]`	可选的自然语言搜索查询。	`无`
`refresh_ttl`	`Optional[bool]`	是否刷新此搜索返回的项的 TTL。如果为 None，则使用存储的默认行为。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`SearchItemsResponse`	List[Item]: 匹配搜索条件的项列表。

使用示例

items = await client.store.search_items(
    ["documents"],
    filter={"author": "John Doe"},
    limit=5,
    offset=0
)
print(items)

----------------------------------------------------------------

{
    "items": [
        {
            "namespace": ["documents", "user123"],
            "key": "item789",
            "value": {
                "title": "Another Document",
                "author": "John Doe"
            },
            "created_at": "2024-07-30T12:00:00Z",
            "updated_at": "2024-07-30T12:00:00Z"
        },
        # ... additional items ...
    ]
}

list_namespaces `async` ¶

list_namespaces(
    prefix: Optional[List[str]] = None,
    suffix: Optional[List[str]] = None,
    max_depth: Optional[int] = None,
    limit: int = 100,
    offset: int = 0,
    headers: Optional[dict[str, str]] = None,
) -> ListNamespaceResponse

列出命名空间，可带可选匹配条件。

参数

名称	类型	描述	默认值
`prefix`	`Optional[List[str]]`	表示用于过滤命名空间的前缀的可选字符串列表。	`无`
`suffix`	`Optional[List[str]]`	表示用于过滤命名空间的后缀的可选字符串列表。	`无`
`max_depth`	`Optional[int]`	可选整数，指定要返回的命名空间的最大深度。	`无`
`limit`	`int`	返回的最大命名空间数量（默认为 100）。	`100`
`offset`	`int`	返回结果前要跳过的命名空间数量（默认为 0）。	`0`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`ListNamespaceResponse`	List[List[str]]: 匹配条件的命名空间列表。

使用示例

namespaces = await client.store.list_namespaces(
    prefix=["documents"],
    max_depth=3,
    limit=10,
    offset=0
)
print(namespaces)

----------------------------------------------------------------

[
    ["documents", "user123", "reports"],
    ["documents", "user456", "invoices"],
    ...
]

SyncLangGraphClient ¶

用于与 LangGraph API 进行交互的同步客户端。

此类提供对 LangGraph API 端点的同步访问，用于管理助手、线程、运行、cron 作业和数据存储。

示例

client = get_sync_client()
assistant = client.assistants.get("asst_123")

SyncHttpClient ¶

方法

名称	描述
`get`	发送 GET 请求。
`post`	发送 POST 请求。
`put`	发送 PUT 请求。
`patch`	发送 PATCH 请求。
`delete`	发送 DELETE 请求。
`stream`	使用 SSE 流式传输请求结果。

get ¶

get(
    path: str,
    *,
    params: Optional[QueryParamTypes] = None,
    headers: Optional[dict[str, str]] = None
) -> Any

发送 GET 请求。

post ¶

post(
    path: str,
    *,
    json: Optional[dict],
    headers: Optional[dict[str, str]] = None
) -> Any

发送 POST 请求。

put ¶

put(
    path: str,
    *,
    json: dict,
    headers: Optional[dict[str, str]] = None
) -> Any

发送 PUT 请求。

patch ¶

patch(
    path: str,
    *,
    json: dict,
    headers: Optional[dict[str, str]] = None
) -> Any

发送 PATCH 请求。

delete ¶

delete(
    path: str,
    *,
    json: Optional[Any] = None,
    headers: Optional[dict[str, str]] = None
) -> None

发送 DELETE 请求。

stream ¶

stream(
    path: str,
    method: str,
    *,
    json: Optional[dict] = None,
    params: Optional[QueryParamTypes] = None,
    headers: Optional[dict[str, str]] = None
) -> Iterator[StreamPart]

使用 SSE 流式传输请求结果。

SyncAssistantsClient ¶

用于同步管理 LangGraph 中的智能体的客户端。

此类提供了与智能体交互的方法，智能体是您的图的版本化配置。

示例

client = get_client()
assistant = client.assistants.get("assistant_id_123")

方法

名称	描述
`get`	按 ID 获取智能体。
`get_graph`	按 ID 获取智能体的图。
`get_schemas`	按 ID 获取智能体的模式。
`get_subgraphs`	按 ID 获取智能体的模式。
`create`	创建一个新的智能体。
`update`	更新智能体。
`delete`	删除智能体。
`search`	搜索智能体。
`get_versions`	列出智能体的所有版本。
`set_latest`	更改智能体的版本。

get ¶

get(
    assistant_id: str,
    *,
    headers: Optional[dict[str, str]] = None
) -> Assistant

按 ID 获取智能体。

参数

名称	类型	描述	默认值
`assistant_id`	`str`	要获取的智能体的 ID。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`Assistant`	`Assistant`	智能体对象。

使用示例

assistant = client.assistants.get(
    assistant_id="my_assistant_id"
)
print(assistant)

----------------------------------------------------

{
    'assistant_id': 'my_assistant_id',
    'graph_id': 'agent',
    'created_at': '2024-06-25T17:10:33.109781+00:00',
    'updated_at': '2024-06-25T17:10:33.109781+00:00',
    'config': {},
    'metadata': {'created_by': 'system'}
}

get_graph ¶

get_graph(
    assistant_id: str,
    *,
    xray: Union[int, bool] = False,
    headers: Optional[dict[str, str]] = None
) -> dict[str, list[dict[str, Any]]]

按 ID 获取智能体的图。

参数

名称	类型	描述	默认值
`assistant_id`	`str`	要获取其图的智能体的 ID。	必需
`xray`	`Union[int, bool]`	包含子图的图表示。如果提供了整数值，则仅包含深度小于或等于该值的子图。	`False`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`图`	`dict[str, list[dict[str, Any]]]`	智能体的图信息，采用 JSON 格式。

使用示例

graph_info = client.assistants.get_graph(
    assistant_id="my_assistant_id"
)
print(graph_info)

--------------------------------------------------------------------------------------------------------------------------

{
    'nodes':
        [
            {'id': '__start__', 'type': 'schema', 'data': '__start__'},
            {'id': '__end__', 'type': 'schema', 'data': '__end__'},
            {'id': 'agent','type': 'runnable','data': {'id': ['langgraph', 'utils', 'RunnableCallable'],'name': 'agent'}},
        ],
    'edges':
        [
            {'source': '__start__', 'target': 'agent'},
            {'source': 'agent','target': '__end__'}
        ]
}

get_schemas ¶

get_schemas(
    assistant_id: str,
    *,
    headers: Optional[dict[str, str]] = None
) -> GraphSchema

按 ID 获取智能体的模式。

参数

名称	类型	描述	默认值
`assistant_id`	`str`	要获取其模式的智能体的 ID。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`GraphSchema`	`GraphSchema`	智能体的图模式。

使用示例

schema = client.assistants.get_schemas(
    assistant_id="my_assistant_id"
)
print(schema)

----------------------------------------------------------------------------------------------------------------------------

{
    'graph_id': 'agent',
    'state_schema':
        {
            'title': 'LangGraphInput',
            '$ref': '#/definitions/AgentState',
            'definitions':
                {
                    'BaseMessage':
                        {
                            'title': 'BaseMessage',
                            'description': 'Base abstract Message class. Messages are the inputs and outputs of ChatModels.',
                            'type': 'object',
                            'properties':
                                {
                                 'content':
                                    {
                                        'title': 'Content',
                                        'anyOf': [
                                            {'type': 'string'},
                                            {'type': 'array','items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]}}
                                        ]
                                    },
                                'additional_kwargs':
                                    {
                                        'title': 'Additional Kwargs',
                                        'type': 'object'
                                    },
                                'response_metadata':
                                    {
                                        'title': 'Response Metadata',
                                        'type': 'object'
                                    },
                                'type':
                                    {
                                        'title': 'Type',
                                        'type': 'string'
                                    },
                                'name':
                                    {
                                        'title': 'Name',
                                        'type': 'string'
                                    },
                                'id':
                                    {
                                        'title': 'Id',
                                        'type': 'string'
                                    }
                                },
                            'required': ['content', 'type']
                        },
                    'AgentState':
                        {
                            'title': 'AgentState',
                            'type': 'object',
                            'properties':
                                {
                                    'messages':
                                        {
                                            'title': 'Messages',
                                            'type': 'array',
                                            'items': {'$ref': '#/definitions/BaseMessage'}
                                        }
                                },
                            'required': ['messages']
                        }
                }
        },
    'config_schema':
        {
            'title': 'Configurable',
            'type': 'object',
            'properties':
                {
                    'model_name':
                        {
                            'title': 'Model Name',
                            'enum': ['anthropic', 'openai'],
                            'type': 'string'
                        }
                }
        }
}

get_subgraphs ¶

get_subgraphs(
    assistant_id: str,
    namespace: Optional[str] = None,
    recurse: bool = False,
    *,
    headers: Optional[dict[str, str]] = None
) -> Subgraphs

按 ID 获取智能体的模式。

参数

名称	类型	描述	默认值
`assistant_id`	`str`	要获取其模式的智能体的 ID。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`子图`	`子图`	智能体的图模式。

create ¶

create(
    graph_id: Optional[str],
    config: Optional[Config] = None,
    *,
    metadata: Json = None,
    assistant_id: Optional[str] = None,
    if_exists: Optional[OnConflictBehavior] = None,
    name: Optional[str] = None,
    headers: Optional[dict[str, str]] = None,
    description: Optional[str] = None
) -> Assistant

创建一个新的智能体。

当图是可配置的并且您希望基于不同配置创建不同的智能体时很有用。

参数

名称	类型	描述	默认值
`graph_id`	`Optional[str]`	智能体应使用的图的 ID。图 ID 通常在您的 langgraph.json 配置中设置。	必需
`config`	`Optional[Config]`	用于图的配置。	`无`
`metadata`	`Json`	要添加到智能体的元数据。	`无`
`assistant_id`	`Optional[str]`	要使用的智能体 ID，如果未提供，将默认为随机 UUID。	`无`
`if_exists`	`Optional[OnConflictBehavior]`	如何处理重复创建。底层默认为 'raise'。必须是 'raise'（如果重复则抛出错误）或 'do_nothing'（返回现有智能体）。	`无`
`name`	`Optional[str]`	智能体的名称。底层默认为 'Untitled'。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`
`description`	`Optional[str]`	可选的智能体描述。description 字段适用于 langgraph-api 服务器版本 >= 0.0.45	`无`

返回值

名称	类型	描述
`Assistant`	`Assistant`	创建的智能体。

使用示例

assistant = client.assistants.create(
    graph_id="agent",
    config={"configurable": {"model_name": "openai"}},
    metadata={"number":1},
    assistant_id="my-assistant-id",
    if_exists="do_nothing",
    name="my_name"
)

update ¶

update(
    assistant_id: str,
    *,
    graph_id: Optional[str] = None,
    config: Optional[Config] = None,
    metadata: Json = None,
    name: Optional[str] = None,
    headers: Optional[dict[str, str]] = None,
    description: Optional[str] = None
) -> Assistant

更新智能体。

使用此方法指向不同的图、更新配置或更改智能体的元数据。

参数

名称	类型	描述	默认值
`assistant_id`	`str`	要更新的智能体。	必需
`graph_id`	`Optional[str]`	智能体应使用的图的 ID。图 ID 通常在您的 langgraph.json 配置中设置。如果为 None，智能体将继续指向同一个图。	`无`
`config`	`Optional[Config]`	用于图的配置。	`无`
`metadata`	`Json`	要与现有智能体元数据合并的元数据。	`无`
`name`	`Optional[str]`	智能体的新名称。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`
`description`	`Optional[str]`	可选的智能体描述。description 字段适用于 langgraph-api 服务器版本 >= 0.0.45	`无`

返回值

名称	类型	描述
`Assistant`	`Assistant`	更新后的智能体。

使用示例

assistant = client.assistants.update(
    assistant_id='e280dad7-8618-443f-87f1-8e41841c180f',
    graph_id="other-graph",
    config={"configurable": {"model_name": "anthropic"}},
    metadata={"number":2}
)

delete ¶

delete(
    assistant_id: str,
    *,
    headers: Optional[dict[str, str]] = None
) -> None

删除智能体。

参数

名称	类型	描述	默认值
`assistant_id`	`str`	要删除的智能体 ID。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`无`	无

使用示例

client.assistants.delete(
    assistant_id="my_assistant_id"
)

search ¶

search(
    *,
    metadata: Json = None,
    graph_id: Optional[str] = None,
    limit: int = 10,
    offset: int = 0,
    headers: Optional[dict[str, str]] = None
) -> list[Assistant]

搜索智能体。

参数

名称	类型	描述	默认值
`metadata`	`Json`	用于过滤的元数据。每个 KV 对都进行精确匹配过滤。	`无`
`graph_id`	`Optional[str]`	用于过滤的图的 ID。图 ID 通常在您的 langgraph.json 配置中设置。	`无`
`limit`	`int`	要返回的最大结果数。	`10`
`offset`	`int`	要跳过的结果数。	`0`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`list[Assistant]`	list[Assistant]：智能体列表。

使用示例

assistants = client.assistants.search(
    metadata = {"name":"my_name"},
    graph_id="my_graph_id",
    limit=5,
    offset=5
)

get_versions ¶

get_versions(
    assistant_id: str,
    metadata: Json = None,
    limit: int = 10,
    offset: int = 0,
    *,
    headers: Optional[dict[str, str]] = None
) -> list[AssistantVersion]

列出智能体的所有版本。

参数

名称	类型	描述	默认值
`assistant_id`	`str`	要获取其版本的智能体 ID。	必需
`metadata`	`Json`	用于过滤版本的元数据。每个 KV 对都进行精确匹配过滤。	`无`
`limit`	`int`	要返回的最大版本数。	`10`
`offset`	`int`	要跳过的版本数。	`0`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`list[AssistantVersion]`	list[Assistant]：智能体列表。

使用示例

assistant_versions = await client.assistants.get_versions(
    assistant_id="my_assistant_id"
)

set_latest ¶

set_latest(
    assistant_id: str,
    version: int,
    *,
    headers: Optional[dict[str, str]] = None
) -> Assistant

更改智能体的版本。

参数

名称	类型	描述	默认值
`assistant_id`	`str`	要删除的智能体 ID。	必需
`version`	`int`	要更改到的版本。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`Assistant`	`Assistant`	智能体对象。

使用示例

new_version_assistant = await client.assistants.set_latest(
    assistant_id="my_assistant_id",
    version=3
)

SyncThreadsClient ¶

用于同步管理 LangGraph 中的线程的客户端。

此类提供用于创建、检索和管理线程的方法，线程代表对话或有状态交互。

示例

client = get_sync_client()
thread = client.threads.create(metadata={"user_id": "123"})

方法

名称	描述
`get`	按 ID 获取线程。
`create`	创建一个新线程。
`update`	更新线程。
`delete`	删除线程。
`search`	搜索线程。
`copy`	复制线程。
`get_state`	获取线程的状态。
`update_state`	更新线程的状态。
`get_history`	获取线程的状态历史。

get ¶

get(
    thread_id: str,
    *,
    headers: Optional[dict[str, str]] = None
) -> Thread

按 ID 获取线程。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要获取的线程 ID。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`Thread`	`Thread`	Thread 对象。

使用示例

thread = client.threads.get(
    thread_id="my_thread_id"
)
print(thread)

-----------------------------------------------------

{
    'thread_id': 'my_thread_id',
    'created_at': '2024-07-18T18:35:15.540834+00:00',
    'updated_at': '2024-07-18T18:35:15.540834+00:00',
    'metadata': {'graph_id': 'agent'}
}

create ¶

create(
    *,
    metadata: Json = None,
    thread_id: Optional[str] = None,
    if_exists: Optional[OnConflictBehavior] = None,
    supersteps: Optional[
        Sequence[dict[str, Sequence[dict[str, Any]]]]
    ] = None,
    graph_id: Optional[str] = None,
    headers: Optional[dict[str, str]] = None
) -> Thread

创建一个新线程。

参数

名称	类型	描述	默认值
`metadata`	`Json`	要添加到线程的元数据。	`无`
`thread_id`	`Optional[str]`	线程 ID。如果为 None，将生成随机 UUID 作为 ID。	`无`
`if_exists`	`Optional[OnConflictBehavior]`	如何处理重复创建。底层默认值为 'raise'。必须是 'raise'（如果重复则抛出错误）或 'do_nothing'（返回现有线程）。	`无`
`supersteps`	`Optional[Sequence[dict[str, Sequence[dict[str, Any]]]]]`	创建线程时应用 supersteps 列表，每个 superstep 包含一系列更新。每个更新都有 `values` 或 `command` 以及 `as_node`。用于在不同部署之间复制线程。	`无`
`graph_id`	`Optional[str]`	可选的图 ID，用于与线程关联。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`Thread`	`Thread`	创建的线程。

使用示例

thread = client.threads.create(
    metadata={"number":1},
    thread_id="my-thread-id",
    if_exists="raise"
)

update ¶

update(
    thread_id: str,
    *,
    metadata: dict[str, Any],
    headers: Optional[dict[str, str]] = None
) -> Thread

更新线程。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要更新的线程 ID。	必需
`metadata`	`dict[str, Any]`	要与现有线程元数据合并的元数据。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`Thread`	`Thread`	创建的线程。

使用示例

thread = client.threads.update(
    thread_id="my-thread-id",
    metadata={"number":1},
)

delete ¶

delete(
    thread_id: str,
    *,
    headers: Optional[dict[str, str]] = None
) -> None

删除线程。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要删除的线程 ID。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`无`	无

使用示例

client.threads.delete(
    thread_id="my_thread_id"
)

search ¶

search(
    *,
    metadata: Json = None,
    values: Json = None,
    status: Optional[ThreadStatus] = None,
    limit: int = 10,
    offset: int = 0,
    headers: Optional[dict[str, str]] = None
) -> list[Thread]

搜索线程。

参数

名称	类型	描述	默认值
`metadata`	`Json`	用于过滤的线程元数据。	`无`
`values`	`Json`	用于过滤的状态值。	`无`
`status`	`Optional[ThreadStatus]`	用于过滤的线程状态。必须是 'idle'、'busy'、'interrupted' 或 'error' 中的一个。	`无`
`limit`	`int`	返回的线程数量限制。	`10`
`offset`	`int`	在线程表中开始搜索的偏移量。	`0`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`list[Thread]`	list[Thread]: 匹配搜索参数的线程列表。

使用示例

threads = client.threads.search(
    metadata={"number":1},
    status="interrupted",
    limit=15,
    offset=5
)

copy ¶

copy(
    thread_id: str,
    *,
    headers: Optional[dict[str, str]] = None
) -> None

复制线程。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要复制的线程 ID。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`无`	无

使用示例

client.threads.copy(
    thread_id="my_thread_id"
)

get_state ¶

get_state(
    thread_id: str,
    checkpoint: Optional[Checkpoint] = None,
    checkpoint_id: Optional[str] = None,
    *,
    subgraphs: bool = False,
    headers: Optional[dict[str, str]] = None
) -> ThreadState

获取线程的状态。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要获取状态的线程 ID。	必需
`checkpoint`	`Optional[Checkpoint]`	要获取状态的检查点。	`无`
`subgraphs`	`bool`	包含子图状态。	`False`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`ThreadState`	`ThreadState`	该状态所属的线程。

使用示例

thread_state = client.threads.get_state(
    thread_id="my_thread_id",
    checkpoint_id="my_checkpoint_id"
)
print(thread_state)

----------------------------------------------------------------------------------------------------------------------------------------------------------------------

{
    'values': {
        'messages': [
            {
                'content': 'how are you?',
                'additional_kwargs': {},
                'response_metadata': {},
                'type': 'human',
                'name': None,
                'id': 'fe0a5778-cfe9-42ee-b807-0adaa1873c10',
                'example': False
            },
            {
                'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.",
                'additional_kwargs': {},
                'response_metadata': {},
                'type': 'ai',
                'name': None,
                'id': 'run-159b782c-b679-4830-83c6-cef87798fe8b',
                'example': False,
                'tool_calls': [],
                'invalid_tool_calls': [],
                'usage_metadata': None
            }
        ]
    },
    'next': [],
    'checkpoint':
        {
            'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2',
            'checkpoint_ns': '',
            'checkpoint_id': '1ef4a9b8-e6fb-67b1-8001-abd5184439d1'
        }
    'metadata':
        {
            'step': 1,
            'run_id': '1ef4a9b8-d7da-679a-a45a-872054341df2',
            'source': 'loop',
            'writes':
                {
                    'agent':
                        {
                            'messages': [
                                {
                                    'id': 'run-159b782c-b679-4830-83c6-cef87798fe8b',
                                    'name': None,
                                    'type': 'ai',
                                    'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.",
                                    'example': False,
                                    'tool_calls': [],
                                    'usage_metadata': None,
                                    'additional_kwargs': {},
                                    'response_metadata': {},
                                    'invalid_tool_calls': []
                                }
                            ]
                        }
                },
    'user_id': None,
    'graph_id': 'agent',
    'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2',
    'created_by': 'system',
    'assistant_id': 'fe096781-5601-53d2-b2f6-0d3403f7e9ca'},
    'created_at': '2024-07-25T15:35:44.184703+00:00',
    'parent_config':
        {
            'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2',
            'checkpoint_ns': '',
            'checkpoint_id': '1ef4a9b8-d80d-6fa7-8000-9300467fad0f'
        }
}

update_state ¶

update_state(
    thread_id: str,
    values: Optional[Union[dict, Sequence[dict]]],
    *,
    as_node: Optional[str] = None,
    checkpoint: Optional[Checkpoint] = None,
    checkpoint_id: Optional[str] = None,
    headers: Optional[dict[str, str]] = None
) -> ThreadUpdateStateResponse

更新线程的状态。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要更新的线程 ID。	必需
`values`	`Optional[Union[dict, Sequence[dict]]]`	用于更新状态的值。	必需
`as_node`	`Optional[str]`	更新状态，如同此节点刚刚执行完毕。	`无`
`checkpoint`	`Optional[Checkpoint]`	要更新状态的检查点。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`ThreadUpdateStateResponse`	`ThreadUpdateStateResponse`	更新线程状态后的响应。

使用示例

response = await client.threads.update_state(
    thread_id="my_thread_id",
    values={"messages":[{"role": "user", "content": "hello!"}]},
    as_node="my_node",
)
print(response)

----------------------------------------------------------------------------------------------------------------------------------------------------------------------

{
    'checkpoint': {
        'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2',
        'checkpoint_ns': '',
        'checkpoint_id': '1ef4a9b8-e6fb-67b1-8001-abd5184439d1',
        'checkpoint_map': {}
    }
}

get_history ¶

get_history(
    thread_id: str,
    *,
    limit: int = 10,
    before: Optional[str | Checkpoint] = None,
    metadata: Optional[dict] = None,
    checkpoint: Optional[Checkpoint] = None,
    headers: Optional[dict[str, str]] = None
) -> list[ThreadState]

获取线程的状态历史。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要获取状态历史的线程 ID。	必需
`checkpoint`	`Optional[Checkpoint]`	返回此子图的状态。如果为空，默认为根图。	`无`
`limit`	`int`	返回的最大状态数量。	`10`
`before`	`Optional[str \| Checkpoint]`	返回此检查点之前的状态。	`无`
`metadata`	`Optional[dict]`	按元数据键值对过滤状态。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`list[ThreadState]`	list[ThreadState]: 线程的状态历史。

使用示例

thread_state = client.threads.get_history(
    thread_id="my_thread_id",
    limit=5,
    before="my_timestamp",
    metadata={"name":"my_name"}
)

SyncRunsClient ¶

用于同步管理 LangGraph 中的运行的客户端。

此类提供用于创建、检索和管理运行的方法，运行代表图的个体执行。

示例

client = get_sync_client()
run = client.runs.create(thread_id="thread_123", assistant_id="asst_456")

方法

名称	描述
`stream`	创建运行并流式传输结果。
`create`	创建后台运行。
`create_batch`	创建一批无状态的后台运行。
`wait`	创建运行，等待其完成并返回最终状态。
`list`	列出运行。
`get`	获取运行。
`cancel`	获取运行。
`join`	阻塞直到运行完成。返回线程的最终状态。
`join_stream`	实时流式传输运行的输出，直到运行完成。
`delete`	删除运行。

stream ¶

stream(
    thread_id: Optional[str],
    assistant_id: str,
    *,
    input: Optional[dict] = None,
    command: Optional[Command] = None,
    stream_mode: Union[
        StreamMode, Sequence[StreamMode]
    ] = "values",
    stream_subgraphs: bool = False,
    metadata: Optional[dict] = None,
    config: Optional[Config] = None,
    checkpoint: Optional[Checkpoint] = None,
    checkpoint_id: Optional[str] = None,
    checkpoint_during: Optional[bool] = None,
    interrupt_before: Optional[
        Union[All, Sequence[str]]
    ] = None,
    interrupt_after: Optional[
        Union[All, Sequence[str]]
    ] = None,
    feedback_keys: Optional[Sequence[str]] = None,
    on_disconnect: Optional[DisconnectMode] = None,
    on_completion: Optional[OnCompletionBehavior] = None,
    webhook: Optional[str] = None,
    multitask_strategy: Optional[MultitaskStrategy] = None,
    if_not_exists: Optional[IfNotExists] = None,
    after_seconds: Optional[int] = None,
    headers: Optional[dict[str, str]] = None
) -> Iterator[StreamPart]

创建运行并流式传输结果。

参数

名称	类型	描述	默认值
`thread_id`	`Optional[str]`	分配给线程的线程 ID。如果为 None，将创建无状态运行。	必需
`assistant_id`	`str`	要从中流式传输的助手 ID 或图名称。如果使用图名称，默认为从该图创建的第一个助手。	必需
`input`	`Optional[dict]`	图的输入。	`无`
`command`	`Optional[Command]`	要执行的命令。	`无`
`stream_mode`	`Union[StreamMode, Sequence[StreamMode]]`	要使用的流模式。	`'values'`
`stream_subgraphs`	`bool`	是否流式传输子图的输出。	`False`
`metadata`	`Optional[dict]`	分配给运行的元数据。	`无`
`config`	`Optional[Config]`	助手的配置。	`无`
`checkpoint`	`Optional[Checkpoint]`	要从中恢复的检查点。	`无`
`checkpoint_during`	`Optional[bool]`	是否在运行期间（或仅在结束/中断时）创建检查点。	`无`
`interrupt_before`	`Optional[Union[All, Sequence[str]]]`	在节点执行之前立即中断。	`无`
`interrupt_after`	`Optional[Union[All, Sequence[str]]]`	在节点执行之后立即中断。	`无`
`feedback_keys`	`Optional[Sequence[str]]`	分配给运行的反馈键。	`无`
`on_disconnect`	`Optional[DisconnectMode]`	要使用的断开连接模式。必须是 'cancel' 或 'continue' 中的一个。	`无`
`on_completion`	`Optional[OnCompletionBehavior]`	是否删除或保留为无状态运行创建的线程。必须是 'delete' 或 'keep' 中的一个。	`无`
`webhook`	`Optional[str]`	LangGraph API 调用完成后要调用的 Webhook。	`无`
`multitask_strategy`	`Optional[MultitaskStrategy]`	要使用的多任务策略。必须是 'reject'、'interrupt'、'rollback' 或 'enqueue' 中的一个。	`无`
`if_not_exists`	`Optional[IfNotExists]`	如何处理缺失的线程。默认为 'reject'。必须是 'reject'（如果缺失则抛出错误）或 'create'（创建新线程）。	`无`
`after_seconds`	`Optional[int]`	开始运行前等待的秒数。用于安排未来的运行。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`Iterator[StreamPart]`	Iterator[StreamPart]：流结果的迭代器。

使用示例

async for chunk in client.runs.stream(
    thread_id=None,
    assistant_id="agent",
    input={"messages": [{"role": "user", "content": "how are you?"}]},
    stream_mode=["values","debug"],
    metadata={"name":"my_run"},
    config={"configurable": {"model_name": "anthropic"}},
    interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"],
    interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"],
    feedback_keys=["my_feedback_key_1","my_feedback_key_2"],
    webhook="https://my.fake.webhook.com",
    multitask_strategy="interrupt"
):
    print(chunk)

------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------

StreamPart(event='metadata', data={'run_id': '1ef4a9b8-d7da-679a-a45a-872054341df2'})
StreamPart(event='values', data={'messages': [{'content': 'how are you?', 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'human', 'name': None, 'id': 'fe0a5778-cfe9-42ee-b807-0adaa1873c10', 'example': False}]})
StreamPart(event='values', data={'messages': [{'content': 'how are you?', 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'human', 'name': None, 'id': 'fe0a5778-cfe9-42ee-b807-0adaa1873c10', 'example': False}, {'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.", 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'ai', 'name': None, 'id': 'run-159b782c-b679-4830-83c6-cef87798fe8b', 'example': False, 'tool_calls': [], 'invalid_tool_calls': [], 'usage_metadata': None}]})
StreamPart(event='end', data=None)

create ¶

create(
    thread_id: Optional[str],
    assistant_id: str,
    *,
    input: Optional[dict] = None,
    command: Optional[Command] = None,
    stream_mode: Union[
        StreamMode, Sequence[StreamMode]
    ] = "values",
    stream_subgraphs: bool = False,
    metadata: Optional[dict] = None,
    config: Optional[Config] = None,
    checkpoint: Optional[Checkpoint] = None,
    checkpoint_id: Optional[str] = None,
    checkpoint_during: Optional[bool] = None,
    interrupt_before: Optional[
        Union[All, Sequence[str]]
    ] = None,
    interrupt_after: Optional[
        Union[All, Sequence[str]]
    ] = None,
    webhook: Optional[str] = None,
    multitask_strategy: Optional[MultitaskStrategy] = None,
    on_completion: Optional[OnCompletionBehavior] = None,
    if_not_exists: Optional[IfNotExists] = None,
    after_seconds: Optional[int] = None,
    headers: Optional[dict[str, str]] = None
) -> Run

创建后台运行。

参数

名称	类型	描述	默认值
`thread_id`	`Optional[str]`	分配给线程的线程 ID。如果为 None，将创建无状态运行。	必需
`assistant_id`	`str`	要从中流式传输的助手 ID 或图名称。如果使用图名称，默认为从该图创建的第一个助手。	必需
`input`	`Optional[dict]`	图的输入。	`无`
`command`	`Optional[Command]`	要执行的命令。	`无`
`stream_mode`	`Union[StreamMode, Sequence[StreamMode]]`	要使用的流模式。	`'values'`
`stream_subgraphs`	`bool`	是否流式传输子图的输出。	`False`
`metadata`	`Optional[dict]`	分配给运行的元数据。	`无`
`config`	`Optional[Config]`	助手的配置。	`无`
`checkpoint`	`Optional[Checkpoint]`	要从中恢复的检查点。	`无`
`checkpoint_during`	`Optional[bool]`	是否在运行期间（或仅在结束/中断时）创建检查点。	`无`
`interrupt_before`	`Optional[Union[All, Sequence[str]]]`	在节点执行之前立即中断。	`无`
`interrupt_after`	`Optional[Union[All, Sequence[str]]]`	在节点执行之后立即中断。	`无`
`webhook`	`Optional[str]`	LangGraph API 调用完成后要调用的 Webhook。	`无`
`multitask_strategy`	`Optional[MultitaskStrategy]`	要使用的多任务策略。必须是 'reject'、'interrupt'、'rollback' 或 'enqueue' 中的一个。	`无`
`on_completion`	`Optional[OnCompletionBehavior]`	是否删除或保留为无状态运行创建的线程。必须是 'delete' 或 'keep' 中的一个。	`无`
`if_not_exists`	`Optional[IfNotExists]`	如何处理缺失的线程。默认为 'reject'。必须是 'reject'（如果缺失则抛出错误）或 'create'（创建新线程）。	`无`
`after_seconds`	`Optional[int]`	开始运行前等待的秒数。用于安排未来的运行。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`Run`	`Run`	创建的后台运行。

使用示例

background_run = client.runs.create(
    thread_id="my_thread_id",
    assistant_id="my_assistant_id",
    input={"messages": [{"role": "user", "content": "hello!"}]},
    metadata={"name":"my_run"},
    config={"configurable": {"model_name": "openai"}},
    interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"],
    interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"],
    webhook="https://my.fake.webhook.com",
    multitask_strategy="interrupt"
)
print(background_run)

--------------------------------------------------------------------------------

{
    'run_id': 'my_run_id',
    'thread_id': 'my_thread_id',
    'assistant_id': 'my_assistant_id',
    'created_at': '2024-07-25T15:35:42.598503+00:00',
    'updated_at': '2024-07-25T15:35:42.598503+00:00',
    'metadata': {},
    'status': 'pending',
    'kwargs':
        {
            'input':
                {
                    'messages': [
                        {
                            'role': 'user',
                            'content': 'how are you?'
                        }
                    ]
                },
            'config':
                {
                    'metadata':
                        {
                            'created_by': 'system'
                        },
                    'configurable':
                        {
                            'run_id': 'my_run_id',
                            'user_id': None,
                            'graph_id': 'agent',
                            'thread_id': 'my_thread_id',
                            'checkpoint_id': None,
                            'model_name': "openai",
                            'assistant_id': 'my_assistant_id'
                        }
                },
            'webhook': "https://my.fake.webhook.com",
            'temporary': False,
            'stream_mode': ['values'],
            'feedback_keys': None,
            'interrupt_after': ["node_to_stop_after_1","node_to_stop_after_2"],
            'interrupt_before': ["node_to_stop_before_1","node_to_stop_before_2"]
        },
    'multitask_strategy': 'interrupt'
}

create_batch ¶

create_batch(
    payloads: list[RunCreate],
    *,
    headers: Optional[dict[str, str]] = None
) -> list[Run]

创建一批无状态的后台运行。

wait ¶

wait(
    thread_id: Optional[str],
    assistant_id: str,
    *,
    input: Optional[dict] = None,
    command: Optional[Command] = None,
    metadata: Optional[dict] = None,
    config: Optional[Config] = None,
    checkpoint_during: Optional[bool] = None,
    checkpoint: Optional[Checkpoint] = None,
    checkpoint_id: Optional[str] = None,
    interrupt_before: Optional[
        Union[All, Sequence[str]]
    ] = None,
    interrupt_after: Optional[
        Union[All, Sequence[str]]
    ] = None,
    webhook: Optional[str] = None,
    on_disconnect: Optional[DisconnectMode] = None,
    on_completion: Optional[OnCompletionBehavior] = None,
    multitask_strategy: Optional[MultitaskStrategy] = None,
    if_not_exists: Optional[IfNotExists] = None,
    after_seconds: Optional[int] = None,
    headers: Optional[dict[str, str]] = None
) -> Union[list[dict], dict[str, Any]]

创建运行，等待其完成并返回最终状态。

参数

名称	类型	描述	默认值
`thread_id`	`Optional[str]`	创建运行的线程 ID。如果为 None，将创建无状态运行。	必需
`assistant_id`	`str`	要运行的助手 ID 或图名称。如果使用图名称，默认为从该图创建的第一个助手。	必需
`input`	`Optional[dict]`	图的输入。	`无`
`command`	`Optional[Command]`	要执行的命令。	`无`
`metadata`	`Optional[dict]`	分配给运行的元数据。	`无`
`config`	`Optional[Config]`	助手的配置。	`无`
`checkpoint`	`Optional[Checkpoint]`	要从中恢复的检查点。	`无`
`checkpoint_during`	`Optional[bool]`	是否在运行期间（或仅在结束/中断时）创建检查点。	`无`
`interrupt_before`	`Optional[Union[All, Sequence[str]]]`	在节点执行之前立即中断。	`无`
`interrupt_after`	`Optional[Union[All, Sequence[str]]]`	在节点执行之后立即中断。	`无`
`webhook`	`Optional[str]`	LangGraph API 调用完成后要调用的 Webhook。	`无`
`on_disconnect`	`Optional[DisconnectMode]`	要使用的断开连接模式。必须是 'cancel' 或 'continue' 中的一个。	`无`
`on_completion`	`Optional[OnCompletionBehavior]`	是否删除或保留为无状态运行创建的线程。必须是 'delete' 或 'keep' 中的一个。	`无`
`multitask_strategy`	`Optional[MultitaskStrategy]`	要使用的多任务策略。必须是 'reject'、'interrupt'、'rollback' 或 'enqueue' 中的一个。	`无`
`if_not_exists`	`Optional[IfNotExists]`	如何处理缺失的线程。默认为 'reject'。必须是 'reject'（如果缺失则抛出错误）或 'create'（创建新线程）。	`无`
`after_seconds`	`Optional[int]`	开始运行前等待的秒数。用于安排未来的运行。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`Union[list[dict], dict[str, Any]]`	Union[list[dict], dict[str, Any]]: 运行的输出。

使用示例

final_state_of_run = client.runs.wait(
    thread_id=None,
    assistant_id="agent",
    input={"messages": [{"role": "user", "content": "how are you?"}]},
    metadata={"name":"my_run"},
    config={"configurable": {"model_name": "anthropic"}},
    interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"],
    interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"],
    webhook="https://my.fake.webhook.com",
    multitask_strategy="interrupt"
)
print(final_state_of_run)

-------------------------------------------------------------------------------------------------------------------------------------------

{
    'messages': [
        {
            'content': 'how are you?',
            'additional_kwargs': {},
            'response_metadata': {},
            'type': 'human',
            'name': None,
            'id': 'f51a862c-62fe-4866-863b-b0863e8ad78a',
            'example': False
        },
        {
            'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.",
            'additional_kwargs': {},
            'response_metadata': {},
            'type': 'ai',
            'name': None,
            'id': 'run-bf1cd3c6-768f-4c16-b62d-ba6f17ad8b36',
            'example': False,
            'tool_calls': [],
            'invalid_tool_calls': [],
            'usage_metadata': None
        }
    ]
}

list ¶

list(
    thread_id: str,
    *,
    limit: int = 10,
    offset: int = 0,
    headers: Optional[dict[str, str]] = None
) -> List[Run]

列出运行。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要列出运行的线程 ID。	必需
`limit`	`int`	要返回的最大结果数。	`10`
`offset`	`int`	要跳过的结果数。	`0`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`List[Run]`	List[Run]: 线程的运行列表。

使用示例

client.runs.list(
    thread_id="thread_id",
    limit=5,
    offset=5,
)

get ¶

get(
    thread_id: str,
    run_id: str,
    *,
    headers: Optional[dict[str, str]] = None
) -> Run

获取运行。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要获取的线程 ID。	必需
`run_id`	`str`	要获取的运行 ID。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`Run`	`Run`	运行对象。

使用示例

run = client.runs.get(
    thread_id="thread_id_to_delete",
    run_id="run_id_to_delete",
)

cancel ¶

cancel(
    thread_id: str,
    run_id: str,
    *,
    wait: bool = False,
    action: CancelAction = "interrupt",
    headers: Optional[dict[str, str]] = None
) -> None

获取运行。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要取消的线程 ID。	必需
`run_id`	`str`	要取消的运行 ID。	必需
`wait`	`bool`	是否等待运行完成。	`False`
`action`	`CancelAction`	取消运行时采取的操作。可能的值为 `interrupt` 或 `rollback`。默认值为 `interrupt`。	`'interrupt'`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`无`	无

使用示例

client.runs.cancel(
    thread_id="thread_id_to_cancel",
    run_id="run_id_to_cancel",
    wait=True,
    action="interrupt"
)

join ¶

join(
    thread_id: str,
    run_id: str,
    *,
    headers: Optional[dict[str, str]] = None
) -> dict

阻塞直到运行完成。返回线程的最终状态。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要等待完成的线程 ID。	必需
`run_id`	`str`	要等待完成的运行 ID。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`dict`	无

使用示例

client.runs.join(
    thread_id="thread_id_to_join",
    run_id="run_id_to_join"
)

join_stream ¶

join_stream(
    thread_id: str,
    run_id: str,
    *,
    stream_mode: Optional[
        Union[StreamMode, Sequence[StreamMode]]
    ] = None,
    cancel_on_disconnect: bool = False,
    headers: Optional[dict[str, str]] = None
) -> Iterator[StreamPart]

实时流式传输运行的输出，直到运行完成。输出不带缓冲，因此在此调用之前产生的任何输出都不会在此处接收到。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要等待完成的线程 ID。	必需
`run_id`	`str`	要等待完成的运行 ID。	必需
`stream_mode`	`Optional[Union[StreamMode, Sequence[StreamMode]]]`	要使用的流模式。必须是创建运行时光传递的流模式的子集。后台运行默认为所有流模式的并集。	`无`
`cancel_on_disconnect`	`bool`	流断开连接时是否取消运行。	`False`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`Iterator[StreamPart]`	无

使用示例

client.runs.join_stream(
    thread_id="thread_id_to_join",
    run_id="run_id_to_join",
    stream_mode=["values", "debug"]
)

delete ¶

delete(
    thread_id: str,
    run_id: str,
    *,
    headers: Optional[dict[str, str]] = None
) -> None

删除运行。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要删除的线程 ID。	必需
`run_id`	`str`	要删除的运行 ID。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`无`	无

使用示例

client.runs.delete(
    thread_id="thread_id_to_delete",
    run_id="run_id_to_delete"
)

SyncCronClient ¶

用于同步管理 LangGraph 中的 cron 作业的客户端。

此类提供用于创建和管理自动化图执行的计划任务（cron 作业）的方法。

示例

client = get_sync_client()
cron_job = client.crons.create_for_thread(thread_id="thread_123", assistant_id="asst_456", schedule="0 * * * *")

方法

名称	描述
`create_for_thread`	为线程创建一个 cron 作业。
`create`	创建 cron 运行。
`delete`	删除 cron。
`search`	获取 cron 作业列表。

create_for_thread ¶

create_for_thread(
    thread_id: str,
    assistant_id: str,
    *,
    schedule: str,
    input: Optional[dict] = None,
    metadata: Optional[dict] = None,
    checkpoint_during: Optional[bool] = None,
    config: Optional[Config] = None,
    interrupt_before: Optional[
        Union[All, list[str]]
    ] = None,
    interrupt_after: Optional[Union[All, list[str]]] = None,
    webhook: Optional[str] = None,
    multitask_strategy: Optional[str] = None,
    headers: Optional[dict[str, str]] = None
) -> Run

为线程创建一个 cron 作业。

参数

名称	类型	描述	默认值
`thread_id`	`str`	要运行 cron 作业的线程 ID。	必需
`assistant_id`	`str`	用于 cron 作业的助手 ID 或图名称。如果使用图名称，默认为从该图创建的第一个助手。	必需
`schedule`	`str`	执行此作业的 cron 计划表。	必需
`input`	`Optional[dict]`	图的输入。	`无`
`metadata`	`Optional[dict]`	分配给 cron 作业运行的元数据。	`无`
`config`	`Optional[Config]`	助手的配置。	`无`
`checkpoint_during`	`Optional[bool]`	是否在运行期间（或仅在结束/中断时）创建检查点。	`无`
`interrupt_before`	`Optional[Union[All, list[str]]]`	在节点执行之前立即中断。	`无`
`interrupt_after`	`Optional[Union[All, list[str]]]`	在节点执行之后立即中断。	`无`
`webhook`	`Optional[str]`	LangGraph API 调用完成后要调用的 Webhook。	`无`
`multitask_strategy`	`Optional[str]`	要使用的多任务策略。必须是 'reject'、'interrupt'、'rollback' 或 'enqueue' 中的一个。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`Run`	`Run`	cron 运行。

使用示例

cron_run = client.crons.create_for_thread(
    thread_id="my-thread-id",
    assistant_id="agent",
    schedule="27 15 * * *",
    input={"messages": [{"role": "user", "content": "hello!"}]},
    metadata={"name":"my_run"},
    config={"configurable": {"model_name": "openai"}},
    interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"],
    interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"],
    webhook="https://my.fake.webhook.com",
    multitask_strategy="interrupt"
)

create ¶

create(
    assistant_id: str,
    *,
    schedule: str,
    input: Optional[dict] = None,
    metadata: Optional[dict] = None,
    config: Optional[Config] = None,
    checkpoint_during: Optional[bool] = None,
    interrupt_before: Optional[
        Union[All, list[str]]
    ] = None,
    interrupt_after: Optional[Union[All, list[str]]] = None,
    webhook: Optional[str] = None,
    multitask_strategy: Optional[str] = None,
    headers: Optional[dict[str, str]] = None
) -> Run

创建 cron 运行。

参数

名称	类型	描述	默认值
`assistant_id`	`str`	用于 cron 作业的助手 ID 或图名称。如果使用图名称，默认为从该图创建的第一个助手。	必需
`schedule`	`str`	执行此作业的 cron 计划表。	必需
`input`	`Optional[dict]`	图的输入。	`无`
`metadata`	`Optional[dict]`	分配给 cron 作业运行的元数据。	`无`
`config`	`Optional[Config]`	助手的配置。	`无`
`checkpoint_during`	`Optional[bool]`	是否在运行期间（或仅在结束/中断时）创建检查点。	`无`
`interrupt_before`	`Optional[Union[All, list[str]]]`	在节点执行之前立即中断。	`无`
`interrupt_after`	`Optional[Union[All, list[str]]]`	在节点执行之后立即中断。	`无`
`webhook`	`Optional[str]`	LangGraph API 调用完成后要调用的 Webhook。	`无`
`multitask_strategy`	`Optional[str]`	要使用的多任务策略。必须是 'reject'、'interrupt'、'rollback' 或 'enqueue' 中的一个。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`Run`	`Run`	cron 运行。

使用示例

cron_run = client.crons.create(
    assistant_id="agent",
    schedule="27 15 * * *",
    input={"messages": [{"role": "user", "content": "hello!"}]},
    metadata={"name":"my_run"},
    config={"configurable": {"model_name": "openai"}},
    checkpoint_during=True,
    interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"],
    interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"],
    webhook="https://my.fake.webhook.com",
    multitask_strategy="interrupt"
)

delete ¶

delete(
    cron_id: str,
    *,
    headers: Optional[dict[str, str]] = None
) -> None

删除 cron。

参数

名称	类型	描述	默认值
`cron_id`	`str`	要删除的 cron ID。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`无`	无

使用示例

client.crons.delete(
    cron_id="cron_to_delete"
)

search ¶

search(
    *,
    assistant_id: Optional[str] = None,
    thread_id: Optional[str] = None,
    limit: int = 10,
    offset: int = 0,
    headers: Optional[dict[str, str]] = None
) -> list[Cron]

获取 cron 作业列表。

参数

名称	类型	描述	默认值
`assistant_id`	`Optional[str]`	要搜索的助手 ID 或图名称。	`无`
`thread_id`	`Optional[str]`	要搜索的线程 ID。	`无`
`limit`	`int`	要返回的最大结果数。	`10`
`offset`	`int`	要跳过的结果数。	`0`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`list[Cron]`	list[Cron]: 搜索返回的 cron 作业列表，

使用示例

cron_jobs = client.crons.search(
    assistant_id="my_assistant_id",
    thread_id="my_thread_id",
    limit=5,
    offset=5,
)
print(cron_jobs)

----------------------------------------------------------

[
    {
        'cron_id': '1ef3cefa-4c09-6926-96d0-3dc97fd5e39b',
        'assistant_id': 'my_assistant_id',
        'thread_id': 'my_thread_id',
        'user_id': None,
        'payload':
            {
                'input': {'start_time': ''},
                'schedule': '4 * * * *',
                'assistant_id': 'my_assistant_id'
            },
        'schedule': '4 * * * *',
        'next_run_date': '2024-07-25T17:04:00+00:00',
        'end_time': None,
        'created_at': '2024-07-08T06:02:23.073257+00:00',
        'updated_at': '2024-07-08T06:02:23.073257+00:00'
    }
]

SyncStoreClient ¶

用于对键值存储执行同步操作的客户端。

提供与远程键值存储交互的方法，允许在命名空间层次结构中存储和检索项目。

示例

client = get_sync_client()
client.store.put_item(["users", "profiles"], "user123", {"name": "Alice", "age": 30})

方法

名称	描述
`put_item`	存储或更新一个项。
`get_item`	检索单个项。
`delete_item`	删除一个项。
`search_items`	在命名空间前缀内搜索项。
`list_namespaces`	列出命名空间，可带可选匹配条件。

put_item ¶

put_item(
    namespace: Sequence[str],
    /,
    key: str,
    value: dict[str, Any],
    index: Optional[
        Union[Literal[False], list[str]]
    ] = None,
    ttl: Optional[int] = None,
    headers: Optional[dict[str, str]] = None,
) -> None

存储或更新一个项。

参数

名称	类型	描述	默认值
`namespace`	`Sequence[str]`	表示命名空间路径的字符串列表。	必需
`key`	`str`	项在命名空间内的唯一标识符。	必需
`value`	`dict[str, Any]`	包含项数据的字典。	必需
`index`	`Optional[Union[Literal[False], list[str]]]`	控制搜索索引 - None（使用默认值），False（禁用），或要索引的字段路径列表。	`无`
`ttl`	`Optional[int]`	项的可选生存时间（分钟），或 None 表示不设过期。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`无`	无

使用示例

client.store.put_item(
    ["documents", "user123"],
    key="item456",
    value={"title": "My Document", "content": "Hello World"}
)

get_item ¶

get_item(
    namespace: Sequence[str],
    /,
    key: str,
    *,
    refresh_ttl: Optional[bool] = None,
    headers: Optional[dict[str, str]] = None,
) -> Item

检索单个项。

参数

名称	类型	描述	默认值
`key`	`str`	项的唯一标识符。	必需
`namespace`	`Sequence[str]`	表示命名空间路径的可选字符串列表。	必需
`refresh_ttl`	`Optional[bool]`	在此读取操作时是否刷新 TTL。如果为 None，则使用存储的默认行为。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

名称	类型	描述
`Item`	`Item`	检索到的项。

使用示例

item = client.store.get_item(
    ["documents", "user123"],
    key="item456",
)
print(item)

----------------------------------------------------------------

{
    'namespace': ['documents', 'user123'],
    'key': 'item456',
    'value': {'title': 'My Document', 'content': 'Hello World'},
    'created_at': '2024-07-30T12:00:00Z',
    'updated_at': '2024-07-30T12:00:00Z'
}

delete_item ¶

delete_item(
    namespace: Sequence[str],
    /,
    key: str,
    headers: Optional[dict[str, str]] = None,
) -> None

删除一个项。

参数

名称	类型	描述	默认值
`key`	`str`	项的唯一标识符。	必需
`namespace`	`Sequence[str]`	表示命名空间路径的可选字符串列表。	必需
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`无`	无

使用示例

client.store.delete_item(
    ["documents", "user123"],
    key="item456",
)

search_items ¶

search_items(
    namespace_prefix: Sequence[str],
    /,
    filter: Optional[dict[str, Any]] = None,
    limit: int = 10,
    offset: int = 0,
    query: Optional[str] = None,
    refresh_ttl: Optional[bool] = None,
    headers: Optional[dict[str, str]] = None,
) -> SearchItemsResponse

在命名空间前缀内搜索项。

参数

名称	类型	描述	默认值
`namespace_prefix`	`Sequence[str]`	表示命名空间前缀的字符串列表。	必需
`filter`	`Optional[dict[str, Any]]`	用于过滤结果的可选键值对字典。	`无`
`limit`	`int`	返回的最大项数量（默认为 10）。	`10`
`offset`	`int`	返回结果前要跳过的项数量（默认为 0）。	`0`
`query`	`Optional[str]`	可选的自然语言搜索查询。	`无`
`refresh_ttl`	`Optional[bool]`	是否刷新此搜索返回的项的 TTL。如果为 None，则使用存储的默认行为。	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`SearchItemsResponse`	List[Item]: 匹配搜索条件的项列表。

使用示例

items = client.store.search_items(
    ["documents"],
    filter={"author": "John Doe"},
    limit=5,
    offset=0
)
print(items)

----------------------------------------------------------------

{
    "items": [
        {
            "namespace": ["documents", "user123"],
            "key": "item789",
            "value": {
                "title": "Another Document",
                "author": "John Doe"
            },
            "created_at": "2024-07-30T12:00:00Z",
            "updated_at": "2024-07-30T12:00:00Z"
        },
        # ... additional items ...
    ]
}

list_namespaces ¶

list_namespaces(
    prefix: Optional[List[str]] = None,
    suffix: Optional[List[str]] = None,
    max_depth: Optional[int] = None,
    limit: int = 100,
    offset: int = 0,
    headers: Optional[dict[str, str]] = None,
) -> ListNamespaceResponse

列出命名空间，可带可选匹配条件。

参数

名称	类型	描述	默认值
`prefix`	`Optional[List[str]]`	表示用于过滤命名空间的前缀的可选字符串列表。	`无`
`suffix`	`Optional[List[str]]`	表示用于过滤命名空间的后缀的可选字符串列表。	`无`
`max_depth`	`Optional[int]`	可选整数，指定要返回的命名空间的最大深度。	`无`
`limit`	`int`	返回的最大命名空间数量（默认为 100）。	`100`
`offset`	`int`	返回结果前要跳过的命名空间数量（默认为 0）。	`0`
`headers`	`Optional[dict[str, str]]`	可选的自定义头，包含在请求中。	`无`

返回值

类型	描述
`ListNamespaceResponse`	List[List[str]]: 匹配条件的命名空间列表。

使用示例

namespaces = client.store.list_namespaces(
    prefix=["documents"],
    max_depth=3,
    limit=10,
    offset=0
)
print(namespaces)

----------------------------------------------------------------

[
    ["documents", "user123", "reports"],
    ["documents", "user456", "invoices"],
    ...
]

get_headers ¶

get_headers(
    api_key: Optional[str],
    custom_headers: Optional[dict[str, str]],
) -> dict[str, str]

合并 api_key 和用户提供的自定义头。

get_client ¶

get_client(
    *,
    url: Optional[str] = None,
    api_key: Optional[str] = None,
    headers: Optional[dict[str, str]] = None
) -> LangGraphClient

获取 LangGraphClient 实例。

参数

名称	类型	描述	默认值
`url`	`Optional[str]`	LangGraph API 的 URL。	`无`
`api_key`	`Optional[str]`	API 密钥。如果未提供，将从环境变量中读取。优先级：1. 显式参数 2. LANGGRAPH_API_KEY 3. LANGSMITH_API_KEY 4. LANGCHAIN_API_KEY	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头	`无`

返回值

名称	类型	描述
`LangGraphClient`	`LangGraphClient`	用于访问 AssistantsClient、
	`LangGraphClient`	ThreadsClient、RunsClient 和 CronClient 的顶级客户端。

示例

from langgraph_sdk import get_client

# get top-level LangGraphClient
client = get_client(url="http://localhost:8123")

# example usage: client.<model>.<method_name>()
assistants = await client.assistants.get(assistant_id="some_uuid")

get_sync_client ¶

get_sync_client(
    *,
    url: Optional[str] = None,
    api_key: Optional[str] = None,
    headers: Optional[dict[str, str]] = None
) -> SyncLangGraphClient

获取同步的 LangGraphClient 实例。

参数

名称	类型	描述	默认值
`url`	`Optional[str]`	LangGraph API 的 URL。	`无`
`api_key`	`Optional[str]`	API 密钥。如果未提供，将从环境变量中读取。优先级：1. 显式参数 2. LANGGRAPH_API_KEY 3. LANGSMITH_API_KEY 4. LANGCHAIN_API_KEY	`无`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头	`无`

返回：SyncLangGraphClient：用于访问 AssistantsClient、ThreadsClient、RunsClient 和 CronClient 的顶级同步客户端。

示例

from langgraph_sdk import get_sync_client

# get top-level synchronous LangGraphClient
client = get_sync_client(url="http://localhost:8123")

# example usage: client.<model>.<method_name>()
assistant = client.assistants.get(assistant_id="some_uuid")

用于与 LangGraph API 交互的数据模型。

类

名称	描述
`配置`	调用的配置选项。
`检查点`	代表执行过程中的一个检查点。
`GraphSchema`	定义图的结构和属性。
`AssistantBase`	助手的基本模型。
`AssistantVersion`	代表助手的特定版本。
`Assistant`	代表具有附加属性的助手。
`中断`	代表执行流中的中断。
`Thread`	代表一个对话线程。
`ThreadTask`	代表线程中的一个任务。
`ThreadState`	代表线程的状态。
`ThreadUpdateStateResponse`	代表更新线程状态的响应。
`Run`	代表单个执行运行。
`Cron`	代表一个计划任务。
`RunCreate`	定义启动后台运行的参数。
`Item`	代表图中 Store 中的单个文档或数据条目。
`ListNamespaceResponse`	列出命名空间的响应结构。
`SearchItem`	搜索操作中带有可选相关性得分的项目。
`SearchItemsResponse`	搜索项目的响应结构。
`StreamPart`	代表流响应的一部分。
`Send`	代表要发送到图中特定节点的消息。
`Command`	代表一个或多个用于控制图执行流和状态的命令。

属性

名称	类型	描述
`Json`		代表一种类似 JSON 的结构，可以是 None 或一个带有字符串键和任意值的字典。
`RunStatus`		代表运行的状态
`ThreadStatus`		代表线程的状态
`StreamMode`		定义流模式
`DisconnectMode`		指定断开连接时的行为
`MultitaskStrategy`		定义如何处理多个任务
`OnConflictBehavior`		指定冲突时的行为
`OnCompletionBehavior`		定义完成后的动作
`All`		代表通配符或“所有”选择器。
`IfNotExists`		指定线程不存在时的行为
`CancelAction`		取消运行时采取的动作。
`AssistantSortBy`		用于排序的字段。
`ThreadSortBy`		用于排序的字段。
`SortOrder`		排序顺序。

Json `模块属性` ¶

Json = Optional[dict[str, Any]]

代表一种类似 JSON 的结构，可以是 None 或一个带有字符串键和任意值的字典。

RunStatus `模块属性` ¶

RunStatus = Literal[
    "pending",
    "running",
    "error",
    "success",
    "timeout",
    "interrupted",
]

代表运行的状态：- "pending"：运行正在等待开始。- "running"：运行当前正在执行。- "error"：运行遇到错误并停止。- "success"：运行成功完成。- "timeout"：运行超出其时间限制。- "interrupted"：运行被手动停止或中断。

ThreadStatus `模块属性` ¶

ThreadStatus = Literal[
    "idle", "busy", "interrupted", "error"
]

代表线程的状态：- "idle"：线程当前未处理任何任务。- "busy"：线程正在积极处理任务。- "interrupted"：线程的执行被中断。- "error"：任务处理期间发生异常。

StreamMode `模块属性` ¶

StreamMode = Literal[
    "values",
    "messages",
    "updates",
    "events",
    "debug",
    "custom",
    "messages-tuple",
]

定义流模式：- "values"：仅流式传输值。- "messages"：流式传输完整消息。- "updates"：流式传输状态更新。- "events"：流式传输执行期间发生的事件。- "debug"：流式传输详细调试信息。- "custom"：流式传输自定义事件。

DisconnectMode `模块属性` ¶

DisconnectMode = Literal['cancel', 'continue']

指定断开连接时的行为：- "cancel"：断开连接时取消操作。- "continue"：即使断开连接也继续操作。

MultitaskStrategy `模块属性` ¶

MultitaskStrategy = Literal[
    "reject", "interrupt", "rollback", "enqueue"
]

定义如何处理多个任务：- "reject"：忙碌时拒绝新任务。- "interrupt"：中断当前任务以执行新任务。- "rollback"：回滚当前任务并开始新任务。- "enqueue"：将新任务排队以供稍后执行。

OnConflictBehavior `模块属性` ¶

OnConflictBehavior = Literal['raise', 'do_nothing']

指定冲突时的行为：- "raise"：冲突发生时抛出异常。- "do_nothing"：忽略冲突并继续。

OnCompletionBehavior `模块属性` ¶

OnCompletionBehavior = Literal['delete', 'keep']

定义完成后的动作：- "delete"：完成后删除资源。- "keep"：完成后保留资源。

All `模块属性` ¶

All = Literal['*']

代表通配符或“所有”选择器。

IfNotExists `模块属性` ¶

IfNotExists = Literal['create', 'reject']

指定线程不存在时的行为：- "create"：如果线程不存在，则创建新线程。- "reject"：如果线程不存在，则拒绝操作。

CancelAction `模块属性` ¶

CancelAction = Literal['interrupt', 'rollback']

取消运行时采取的动作。- "interrupt"：简单取消运行。- "rollback"：取消运行。然后删除运行和相关的检查点。

AssistantSortBy `模块属性` ¶

AssistantSortBy = Literal[
    "assistant_id",
    "graph_id",
    "name",
    "created_at",
    "updated_at",
]

用于排序的字段。

ThreadSortBy `模块属性` ¶

ThreadSortBy = Literal[
    "thread_id", "status", "created_at", "updated_at"
]

用于排序的字段。

SortOrder `模块属性` ¶

SortOrder = Literal['asc', 'desc']

排序顺序。

Config ¶

基类：TypedDict

调用的配置选项。

属性

名称	类型	描述
`tags`	`list[str]`	用于此调用及任何子调用（例如链调用 LLM）的标签。
`recursion_limit`	`int`	调用可以递归的最大次数。如果未提供，默认为 25。
`configurable`	`dict[str, Any]`	此 Runnable 上先前配置为可配置属性的运行时值，

tags `实例属性` ¶

tags: list[str]

用于此调用及任何子调用（例如链调用 LLM）的标签。您可以使用这些标签来过滤调用。

recursion_limit `实例属性` ¶

recursion_limit: int

调用可以递归的最大次数。如果未提供，默认为 25。

configurable `实例属性` ¶

configurable: dict[str, Any]

通过 .configurable_fields() 或 .configurable_alternatives() 在此 Runnable 或子 Runnable 上先前配置为可配置属性的运行时值。请查看 .output_schema() 以获取已配置为可配置属性的说明。

Checkpoint ¶

基类：TypedDict

代表执行过程中的一个检查点。

属性

名称	类型	描述
`thread_id`	`str`	与此检查点关联的线程的唯一标识符。
`checkpoint_ns`	`str`	检查点的命名空间；内部用于管理子图状态。
`checkpoint_id`	`Optional[str]`	检查点本身的唯一可选标识符。
`checkpoint_map`	`Optional[dict[str, Any]]`	包含检查点特定数据的可选字典。

thread_id `实例属性` ¶

thread_id: str

与此检查点关联的线程的唯一标识符。

checkpoint_ns `实例属性` ¶

checkpoint_ns: str

检查点的命名空间；内部用于管理子图状态。

checkpoint_id `实例属性` ¶

checkpoint_id: Optional[str]

检查点本身的唯一可选标识符。

checkpoint_map `实例属性` ¶

checkpoint_map: Optional[dict[str, Any]]

包含检查点特定数据的可选字典。

GraphSchema ¶

基类：TypedDict

定义图的结构和属性。

属性

名称	类型	描述
`graph_id`	`str`	图的 ID。
`input_schema`	`Optional[dict]`	图输入的模式。
`output_schema`	`Optional[dict]`	图输出的模式。
`state_schema`	`Optional[dict]`	图状态的模式。
`config_schema`	`Optional[dict]`	图配置的模式。

graph_id `实例属性` ¶

graph_id: str

图的 ID。

input_schema `实例属性` ¶

input_schema: Optional[dict]

图输入的模式。如果无法从图生成 JSON 模式，则缺失。

output_schema `实例属性` ¶

output_schema: Optional[dict]

图输出的模式。如果无法从图生成 JSON 模式，则缺失。

state_schema `实例属性` ¶

state_schema: Optional[dict]

图状态的模式。如果无法从图生成 JSON 模式，则缺失。

config_schema `实例属性` ¶

config_schema: Optional[dict]

图配置的模式。如果无法从图生成 JSON 模式，则缺失。

AssistantBase ¶

基类：TypedDict

助手的基本模型。

属性

名称	类型	描述
`assistant_id`	`str`	助手的 ID。
`graph_id`	`str`	图的 ID。
`config`	`配置`	助手的配置。
`created_at`	`datetime`	助手创建的时间。
`metadata`	`Json`	助手的元数据。
`version`	`int`	助手的版本
`name`	`str`	助手的名称
`description`	`Optional[str]`	助手的描述

assistant_id `实例属性` ¶

assistant_id: str

助手的 ID。

graph_id `实例属性` ¶

graph_id: str

图的 ID。

config `实例属性` ¶

config: Config

助手的配置。

created_at `实例属性` ¶

created_at: datetime

助手创建的时间。

metadata `实例属性` ¶

metadata: Json

助手的元数据。

version `实例属性` ¶

version: int

助手的版本

name `实例属性` ¶

name: str

助手的名称

description `实例属性` ¶

description: Optional[str]

助手的描述

AssistantVersion ¶

基类：AssistantBase

代表助手的特定版本。

属性

名称	类型	描述
`assistant_id`	`str`	助手的 ID。
`graph_id`	`str`	图的 ID。
`config`	`配置`	助手的配置。
`created_at`	`datetime`	助手创建的时间。
`metadata`	`Json`	助手的元数据。
`version`	`int`	助手的版本
`name`	`str`	助手的名称
`description`	`Optional[str]`	助手的描述

assistant_id `实例属性` ¶

assistant_id: str

助手的 ID。

graph_id `实例属性` ¶

graph_id: str

图的 ID。

config `实例属性` ¶

config: Config

助手的配置。

created_at `实例属性` ¶

created_at: datetime

助手创建的时间。

metadata `实例属性` ¶

metadata: Json

助手的元数据。

version `实例属性` ¶

version: int

助手的版本

name `实例属性` ¶

name: str

助手的名称

description `实例属性` ¶

description: Optional[str]

助手的描述

Assistant ¶

基类：AssistantBase

代表具有附加属性的助手。

属性

名称	类型	描述
`updated_at`	`datetime`	助手最后更新的时间。
`assistant_id`	`str`	助手的 ID。
`graph_id`	`str`	图的 ID。
`config`	`配置`	助手的配置。
`created_at`	`datetime`	助手创建的时间。
`metadata`	`Json`	助手的元数据。
`version`	`int`	助手的版本
`name`	`str`	助手的名称
`description`	`Optional[str]`	助手的描述

updated_at `实例属性` ¶

updated_at: datetime

助手最后更新的时间。

assistant_id `实例属性` ¶

assistant_id: str

助手的 ID。

graph_id `实例属性` ¶

graph_id: str

图的 ID。

config `实例属性` ¶

config: Config

助手的配置。

created_at `实例属性` ¶

created_at: datetime

助手创建的时间。

metadata `实例属性` ¶

metadata: Json

助手的元数据。

version `实例属性` ¶

version: int

助手的版本

name `实例属性` ¶

name: str

助手的名称

description `实例属性` ¶

description: Optional[str]

助手的描述

Interrupt ¶

基类：TypedDict

代表执行流中的中断。

属性

名称	类型	描述
`value`	`Any`	与中断关联的值。
`when`	`Literal['during']`	中断发生时。
`resumable`	`bool`	中断是否可以恢复。
`ns`	`Optional[list[str]]`	中断的可选命名空间。

value `实例属性` ¶

value: Any

与中断关联的值。

when `实例属性` ¶

when: Literal['during']

中断发生时。

resumable `实例属性` ¶

resumable: bool

中断是否可以恢复。

ns `实例属性` ¶

ns: Optional[list[str]]

中断的可选命名空间。

Thread ¶

基类：TypedDict

代表一个对话线程。

属性

名称	类型	描述
`thread_id`	`str`	线程的 ID。
`created_at`	`datetime`	线程创建的时间。
`updated_at`	`datetime`	线程最后更新的时间。
`metadata`	`Json`	线程元数据。
`status`	`ThreadStatus`	线程状态，取值为 'idle'、'busy'、'interrupted'之一。
`values`	`Json`	线程的当前状态。
`interrupts`	`Dict[str, list[Interrupt]]`	在此线程中抛出的中断。

thread_id `实例属性` ¶

thread_id: str

线程的 ID。

created_at `实例属性` ¶

created_at: datetime

线程创建的时间。

updated_at `实例属性` ¶

updated_at: datetime

线程最后更新的时间。

metadata `实例属性` ¶

metadata: Json

线程元数据。

status `实例属性` ¶

status: ThreadStatus

线程状态，取值为 'idle'、'busy'、'interrupted'之一。

values `实例属性` ¶

values: Json

线程的当前状态。

interrupts `实例属性` ¶

interrupts: Dict[str, list[Interrupt]]

在此线程中抛出的中断。

ThreadTask ¶

基类：TypedDict

代表线程中的一个任务。

ThreadState ¶

基类：TypedDict

代表线程的状态。

属性

名称	类型	描述
`values`	`Union[list[dict], dict[str, Any]]`	状态值。
`next`	`Sequence[str]`	要执行的下一个节点。如果为空，则线程处于完成状态，直到接收到新的输入。
`checkpoint`	`检查点`	检查点的 ID。
`metadata`	`Json`	此状态的元数据。
`created_at`	`Optional[str]`	状态创建的时间戳。
`parent_checkpoint`	`Optional[Checkpoint]`	父检查点的 ID。如果缺失，则这是根检查点。
`tasks`	`Sequence[ThreadTask]`	在此步骤中要执行的任务。如果已经尝试过，可能包含错误。

values `实例属性` ¶

values: Union[list[dict], dict[str, Any]]

状态值。

next `实例属性` ¶

next: Sequence[str]

要执行的下一个节点。如果为空，则线程处于完成状态，直到接收到新的输入。

checkpoint `实例属性` ¶

checkpoint: Checkpoint

检查点的 ID。

metadata `实例属性` ¶

metadata: Json

此状态的元数据。

created_at `实例属性` ¶

created_at: Optional[str]

状态创建的时间戳。

parent_checkpoint `实例属性` ¶

parent_checkpoint: Optional[Checkpoint]

父检查点的 ID。如果缺失，则这是根检查点。

tasks `实例属性` ¶

tasks: Sequence[ThreadTask]

在此步骤中要执行的任务。如果已经尝试过，可能包含错误。

ThreadUpdateStateResponse ¶

基类：TypedDict

代表更新线程状态的响应。

属性

名称	类型	描述
`checkpoint`	`检查点`	最新状态的检查点。

checkpoint `实例属性` ¶

checkpoint: Checkpoint

最新状态的检查点。

Run ¶

基类：TypedDict

代表单个执行运行。

属性

名称	类型	描述
`run_id`	`str`	运行的 ID。
`thread_id`	`str`	线程的 ID。
`assistant_id`	`str`	用于此运行的助手。
`created_at`	`datetime`	运行创建的时间。
`updated_at`	`datetime`	运行最后更新的时间。
`status`	`RunStatus`	运行的状态。取值为 'pending', 'running', "error", 'success', "timeout", "interrupted"之一。
`metadata`	`Json`	运行元数据。
`multitask_strategy`	`MultitaskStrategy`	处理同一线程上并发运行的策略。

run_id `实例属性` ¶

run_id: str

运行的 ID。

thread_id `实例属性` ¶

thread_id: str

线程的 ID。

assistant_id `实例属性` ¶

assistant_id: str

用于此运行的助手。

created_at `实例属性` ¶

created_at: datetime

运行创建的时间。

updated_at `实例属性` ¶

updated_at: datetime

运行最后更新的时间。

status `实例属性` ¶

status: RunStatus

运行的状态。取值为 'pending', 'running', "error", 'success', "timeout", "interrupted"之一。

metadata `实例属性` ¶

metadata: Json

运行元数据。

multitask_strategy `实例属性` ¶

multitask_strategy: MultitaskStrategy

处理同一线程上并发运行的策略。

Cron ¶

基类：TypedDict

代表一个计划任务。

属性

名称	类型	描述
`cron_id`	`str`	cron 的 ID。
`thread_id`	`Optional[str]`	线程的 ID。
`end_time`	`Optional[datetime]`	停止运行 cron 的结束日期。
`schedule`	`str`	运行计划，cron 格式。
`created_at`	`datetime`	cron 创建的时间。
`updated_at`	`datetime`	cron 最后更新的时间。
`payload`	`dict`	用于创建新运行的运行负载。

cron_id `实例属性` ¶

cron_id: str

cron 的 ID。

thread_id `实例属性` ¶

thread_id: Optional[str]

线程的 ID。

end_time `实例属性` ¶

end_time: Optional[datetime]

停止运行 cron 的结束日期。

schedule `实例属性` ¶

schedule: str

运行计划，cron 格式。

created_at `实例属性` ¶

created_at: datetime

cron 创建的时间。

updated_at `实例属性` ¶

updated_at: datetime

cron 最后更新的时间。

payload `实例属性` ¶

payload: dict

用于创建新运行的运行负载。

RunCreate ¶

基类：TypedDict

定义启动后台运行的参数。

属性

名称	类型	描述
`thread_id`	`Optional[str]`	要运行的线程标识符。如果未提供，则运行是无状态的。
`assistant_id`	`str`	用于此运行的助手标识符。
`input`	`Optional[dict]`	运行的初始输入数据。
`metadata`	`Optional[dict]`	与运行关联的附加元数据。
`config`	`Optional[Config]`	运行的配置选项。
`checkpoint_id`	`Optional[str]`	要从中恢复的检查点标识符。
`interrupt_before`	`Optional[list[str]]`	在执行前中断的节点名称列表。
`interrupt_after`	`Optional[list[str]]`	在执行后中断的节点名称列表。
`webhook`	`Optional[str]`	发送关于运行进度的 webhook 通知 URL。
`multitask_strategy`	`Optional[MultitaskStrategy]`	处理同一线程上并发运行的策略。

thread_id `实例属性` ¶

thread_id: Optional[str]

要运行的线程标识符。如果未提供，则运行是无状态的。

assistant_id `实例属性` ¶

assistant_id: str

用于此运行的助手标识符。

input `实例属性` ¶

input: Optional[dict]

运行的初始输入数据。

metadata `实例属性` ¶

metadata: Optional[dict]

与运行关联的附加元数据。

config `实例属性` ¶

config: Optional[Config]

运行的配置选项。

checkpoint_id `实例属性` ¶

checkpoint_id: Optional[str]

要从中恢复的检查点标识符。

interrupt_before `实例属性` ¶

interrupt_before: Optional[list[str]]

在执行前中断的节点名称列表。

interrupt_after `实例属性` ¶

interrupt_after: Optional[list[str]]

在执行后中断的节点名称列表。

webhook `实例属性` ¶

webhook: Optional[str]

发送关于运行进度的 webhook 通知 URL。

multitask_strategy `实例属性` ¶

multitask_strategy: Optional[MultitaskStrategy]

处理同一线程上并发运行的策略。

Item ¶

基类：TypedDict

代表图中 Store 中的单个文档或数据条目。

项目用于存储跨线程内存。

属性

名称	类型	描述
`namespace`	`list[str]`	项目的命名空间。命名空间类似于文档目录。
`key`	`str`	项目在其命名空间内的唯一标识符。
`value`	`dict[str, Any]`	存储在项目中的值。这即文档本身。
`created_at`	`datetime`	项目创建的时间戳。
`updated_at`	`datetime`	项目最后更新的时间戳。

namespace `实例属性` ¶

namespace: list[str]

项目的命名空间。命名空间类似于文档目录。

key `实例属性` ¶

key: str

项目在其命名空间内的唯一标识符。

通常，键无需全局唯一。

value `实例属性` ¶

value: dict[str, Any]

存储在项目中的值。这即文档本身。

created_at `实例属性` ¶

created_at: datetime

项目创建的时间戳。

updated_at `实例属性` ¶

updated_at: datetime

项目最后更新的时间戳。

ListNamespaceResponse ¶

基类：TypedDict

列出命名空间的响应结构。

属性

名称	类型	描述
`namespaces`	`list[list[str]]`	命名空间路径列表，其中每个路径是一个字符串列表。

namespaces `实例属性` ¶

namespaces: list[list[str]]

命名空间路径列表，其中每个路径是一个字符串列表。

SearchItem ¶

基类: Item

搜索操作中带有可选相关性得分的项目。

属性

名称	类型	描述
`score`	`Optional[float]`	相关性/相似性得分。在使用自然语言查询搜索兼容存储时包含。

namespace `实例属性` ¶

namespace: list[str]

项目的命名空间。命名空间类似于文档目录。

key `实例属性` ¶

key: str

项目在其命名空间内的唯一标识符。

通常，键无需全局唯一。

value `实例属性` ¶

value: dict[str, Any]

存储在项目中的值。这即文档本身。

created_at `实例属性` ¶

created_at: datetime

项目创建的时间戳。

updated_at `实例属性` ¶

updated_at: datetime

项目最后更新的时间戳。

SearchItemsResponse ¶

基类：TypedDict

搜索项目的响应结构。

属性

名称	类型	描述
`items`	`list[SearchItem]`	与搜索条件匹配的项目列表。

items `实例属性` ¶

items: list[SearchItem]

与搜索条件匹配的项目列表。

StreamPart ¶

基类: NamedTuple

代表流响应的一部分。

属性

名称	类型	描述
`event`	`str`	此流部分事件的类型。
`data`	`dict`	与事件关联的数据负载。

event `实例属性` ¶

event: str

此流部分事件的类型。

data `实例属性` ¶

data: dict

与事件关联的数据负载。

Send ¶

基类：TypedDict

代表要发送到图中特定节点的消息。

此类型用于显式发送消息到图中的节点，通常在 Command 对象中使用以控制图的执行流。

属性

名称	类型	描述
`node`	`str`	要发送消息的目标节点名称。
`input`	`Optional[dict[str, Any]]`	包含传递给节点的输入数据的可选字典。

node `实例属性` ¶

node: str

要发送消息的目标节点名称。

input `实例属性` ¶

input: Optional[dict[str, Any]]

包含传递给节点的输入数据的可选字典。

如果为 None，则调用节点时无输入。

Command ¶

基类：TypedDict

代表一个或多个用于控制图执行流和状态的命令。

此类型定义了节点可以返回的控制命令，用于影响图的执行。它允许您导航到其他节点、更新图状态以及从中断恢复。

属性

名称	类型	描述
`goto`	`Union[Send, str, Sequence[Union[Send, str]]]`	指定执行应在哪里继续。可以是：
`update`	`Union[dict[str, Any], Sequence[Tuple[str, Any]]]`	要应用于图状态的更新。可以是：
`resume`	`Any`	中断后用于恢复执行的值。

goto `实例属性` ¶

goto: Union[Send, str, Sequence[Union[Send, str]]]

指定执行应在哪里继续。可以是：

一个字符串节点名称，用于导航到该节点。
一个 Send 对象，用于使用特定输入执行节点。
节点名称或 Send 对象的序列，按顺序执行。

update `实例属性` ¶

update: Union[dict[str, Any], Sequence[Tuple[str, Any]]]

要应用于图状态的更新。可以是：

一个用于合并状态更新的字典。
一个 (键, 值) 元组序列，用于有序更新。

resume `实例属性` ¶

resume: Any

中断后用于恢复执行的值。与 interrupt() 结合使用以实现控制流。

模块

名称	描述
`exceptions`	身份验证系统使用的异常。
`types`	LangGraph 的身份验证和授权类型。

类

名称	描述
`Auth`	向您的 LangGraph 应用程序添加自定义身份验证和授权管理。

Auth ¶

向您的 LangGraph 应用程序添加自定义身份验证和授权管理。

Auth 类提供了一个统一的系统来处理 LangGraph 应用程序中的身份验证和授权。它支持自定义用户身份验证协议和针对不同资源和操作的细粒度授权规则。

要使用，创建一个单独的 Python 文件，并将该文件的路径添加到您的 LangGraph API 配置文件（langgraph.json）中。在该文件中，创建一个 Auth 类的实例，并根据需要注册身份验证和授权处理程序。

langgraph.json 示例文件

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

然后 LangGraph 服务器将在收到请求时加载您的 Auth 文件并在服务器端运行它。

基本用法

from langgraph_sdk import Auth

my_auth = Auth()

async def verify_token(token: str) -> str:
    # Verify token and return user_id
    # This would typically be a call to your auth server
    return "user_id"

@auth.authenticate
async def authenticate(authorization: str) -> str:
    # Verify token and return user_id
    result = await verify_token(authorization)
    if result != "user_id":
        raise Auth.exceptions.HTTPException(
            status_code=401, detail="Unauthorized"
        )
    return result

# Global fallback handler
@auth.on
async def authorize_default(params: Auth.on.value):
    return False # Reject all requests (default behavior)

@auth.on.threads.create
async def authorize_thread_create(params: Auth.on.threads.create.value):
    # Allow the allowed user to create a thread
    assert params.get("metadata", {}).get("owner") == "allowed_user"

@auth.on.store
async def authorize_store(ctx: Auth.types.AuthContext, value: Auth.types.on):
    assert ctx.user.identity in value["namespace"], "Not authorized"

请求处理流程

身份验证（您的 @auth.authenticate 处理程序）在**每个请求**上首先执行。
对于授权，调用最具体的匹配处理程序。
- 如果存在针对特定资源和操作的处理程序，则使用该处理程序（例如，@auth.on.threads.create）。
- 否则，如果存在针对任意操作的资源处理程序，则使用该处理程序（例如，@auth.on.threads）。
- 最后，如果没有特定处理程序匹配，则使用全局处理程序（例如，@auth.on）。
- 如果未设置全局处理程序，则接受请求。

这允许您使用全局处理器设置默认行为，并根据需要覆盖特定路由。

方法

名称	描述
`authenticate`	注册一个身份验证处理器函数。

属性

名称	类型	描述
`types`		引用身份验证类型定义。
`exceptions`		引用身份验证异常定义。
`on`		授权处理器的入口点，用于控制对特定资源的访问。

types `class-attribute` `instance-attribute` ¶

types = types

引用身份验证类型定义。

提供对身份验证系统中使用的所有类型定义的访问，例如 ThreadsCreate、AssistantsRead 等。

exceptions `class-attribute` `instance-attribute` ¶

exceptions = exceptions

引用身份验证异常定义。

提供对身份验证系统中使用的所有异常定义的访问，例如 HTTPException 等。

on `instance-attribute` ¶

on = _On(self)

授权处理器的入口点，用于控制对特定资源的访问。

on 类提供了一种灵活的方式来定义应用程序中不同资源和操作的授权规则。它支持三种主要的使用模式：

全局处理器，对所有资源和操作运行
资源特定的处理器，对特定资源上的所有操作运行
资源和操作特定的处理器，用于精细控制

每个处理器必须是一个接受两个参数的异步函数：

ctx (AuthContext)：包含请求上下文和已认证用户信息
value：正在授权的数据（类型因端点而异）

处理器应返回以下之一：

- None or True: Accept the request
- False: Reject with 403 error
- FilterType: Apply filtering rules to the response

示例

所有请求的全局处理器

@auth.on
async def reject_unhandled_requests(ctx: AuthContext, value: Any) -> None:
    print(f"Request to {ctx.path} by {ctx.user.identity}")
    return False

资源特定的处理器。对于 threads 资源上的所有操作，这将优先于全局处理器。

@auth.on.threads
async def check_thread_access(ctx: AuthContext, value: Any) -> bool:
    # Allow access only to threads created by the user
    return value.get("created_by") == ctx.user.identity

资源和操作特定的处理器

@auth.on.threads.delete
async def prevent_thread_deletion(ctx: AuthContext, value: Any) -> bool:
    # Only admins can delete threads
    return "admin" in ctx.user.permissions

多个资源或操作

@auth.on(resources=["threads", "runs"], actions=["create", "update"])
async def rate_limit_writes(ctx: AuthContext, value: Any) -> bool:
    # Implement rate limiting for write operations
    return await check_rate_limit(ctx.user.identity)

store 资源的身份验证有所不同，因为其结构由开发者定义。您通常希望在命名空间中强制执行用户凭据。

@auth.on.store
async def check_store_access(ctx: AuthContext, value: Auth.types.on) -> bool:
    # Assuming you structure your store like (store.aput((user_id, application_context), key, value))
    assert value["namespace"][0] == ctx.user.identity

authenticate ¶

authenticate(fn: AH) -> AH

注册一个身份验证处理器函数。

身份验证处理器负责验证凭据并返回用户作用域。它可以按名称接受以下任何参数：

- request (Request): The raw ASGI request object
- body (dict): The parsed request body
- path (str): The request path, e.g., "/threads/abcd-1234-abcd-1234/runs/abcd-1234-abcd-1234/stream"
- method (str): The HTTP method, e.g., "GET"
- path_params (dict[str, str]): URL path parameters, e.g., {"thread_id": "abcd-1234-abcd-1234", "run_id": "abcd-1234-abcd-1234"}
- query_params (dict[str, str]): URL query parameters, e.g., {"stream": "true"}
- headers (dict[bytes, bytes]): Request headers
- authorization (str | None): The Authorization header value (e.g., "Bearer <token>")

参数

名称	类型	描述	默认值
`fn`	`AH`	要注册的身份验证处理器函数。必须返回用户的表示。这可以是：- 字符串（用户 ID）- 包含 {"identity": str, "permissions": list[str]} 的字典 - 或具有 identity 和 permissions 属性的对象您的下游处理器可以选择使用 permissions。	必需

返回值

类型	描述
`AH`	已注册的处理器函数。

抛出

类型	描述
`ValueError`	如果身份验证处理器已注册。

示例

基本令牌身份验证

@auth.authenticate
async def authenticate(authorization: str) -> str:
    user_id = verify_token(authorization)
    return user_id

接受完整的请求上下文

@auth.authenticate
async def authenticate(
    method: str,
    path: str,
    headers: dict[str, bytes]
) -> str:
    user = await verify_request(method, path, headers)
    return user

返回用户名和权限

@auth.authenticate
async def authenticate(
    method: str,
    path: str,
    headers: dict[str, bytes]
) -> Auth.types.MinimalUserDict:
    permissions, user = await verify_request(method, path, headers)
    # Permissions could be things like ["runs:read", "runs:write", "threads:read", "threads:write"]
    return {
        "identity": user["id"],
        "permissions": permissions,
        "display_name": user["name"],
    }

LangGraph 的身份验证和授权类型。

本模块定义了 LangGraph 中用于身份验证、授权和请求处理的核心类型。它包括用户协议、身份验证上下文以及各种 API 操作的类型化字典。

注意

所有 typing.TypedDict 类都使用 total=False 使所有字段默认为 typing.Optional。

类

名称	描述
`ThreadsCreate`	创建新线程的参数。
`ThreadsRead`	读取线程状态或运行信息的参数。
`ThreadsUpdate`	更新线程或运行的参数。
`ThreadsDelete`	删除线程的参数。
`ThreadsSearch`	搜索线程的参数。
`RunsCreate`	创建运行的有效载荷。
`AssistantsCreate`	创建助手的有效载荷。
`AssistantsRead`	读取助手的有效载荷。
`AssistantsUpdate`	更新助手的有效载荷。
`AssistantsDelete`	删除助手的有效载荷。
`AssistantsSearch`	搜索助手的有效载荷。
`StoreGet`	根据命名空间和键检索特定项的操作。
`StoreSearch`	在指定命名空间层次结构中搜索项的操作。
`StoreListNamespaces`	列出和过滤 store 中的命名空间的操作。
`StorePut`	在 store 中存储、更新或删除项的操作。
`StoreDelete`	从 store 中删除项的操作。
`on`	不同 API 操作的类型定义的命名空间。

属性

名称	类型	描述
`MetadataInput`		附加到实体的任意元数据类型。

RunStatus `module-attribute` ¶

RunStatus = Literal[
    "pending", "error", "success", "timeout", "interrupted"
]

运行执行的状态。

值

pending：运行已排队或正在进行中
error：运行失败并出现错误
success：运行成功完成
timeout：运行超出时间限制
interrupted：运行被手动中断

MultitaskStrategy `module-attribute` ¶

MultitaskStrategy = Literal[
    "reject", "rollback", "interrupt", "enqueue"
]

处理多个并发任务的策略。

值

reject：当有一个任务正在进行时，拒绝新任务
rollback：取消当前任务并开始新任务
interrupt：中断当前任务并开始新任务
enqueue：将新任务排队，以便在当前任务之后运行

OnConflictBehavior `module-attribute` ¶

OnConflictBehavior = Literal['raise', 'do_nothing']

遇到冲突时的行为。

值

raise：在冲突时抛出异常
do_nothing：静默忽略冲突

IfNotExists `module-attribute` ¶

IfNotExists = Literal['create', 'reject']

实体不存在时的行为。

值

create：创建实体
reject：拒绝操作

FilterType `module-attribute` ¶

FilterType = Union[
    Dict[
        str,
        Union[str, Dict[Literal["$eq", "$contains"], str]],
    ],
    Dict[str, str],
]

授权处理器的响应类型。

支持精确匹配和运算符

精确匹配简写：{"field": "value"}
精确匹配：{"field": {"$eq": "value"}}
包含：{"field": {"$contains": "value"}}

示例

资源所有者的简单精确匹配过滤器

filter = {"owner": "user-abcd123"}

精确匹配过滤器的显式版本

filter = {"owner": {"$eq": "user-abcd123"}}

包含性

filter = {"participants": {"$contains": "user-abcd123"}}

组合过滤器（视为逻辑 AND）

filter = {"owner": "user-abcd123", "participants": {"$contains": "user-efgh456"}}

ThreadStatus `module-attribute` ¶

ThreadStatus = Literal[
    "idle", "busy", "interrupted", "error"
]

线程的状态。

值

idle：线程可供工作
busy：线程当前正在处理
interrupted：线程被中断
error：线程遇到错误

MetadataInput `module-attribute` ¶

MetadataInput = Dict[str, Any]

附加到实体的任意元数据类型。

允许将自定义键值对存储到任何实体。键必须是字符串，值可以是任何 JSON 可序列化类型。

示例

metadata = {
    "created_by": "user123",
    "priority": 1,
    "tags": ["important", "urgent"]
}

HandlerResult `module-attribute` ¶

HandlerResult = Union[None, bool, FilterType]

处理器结果可以是：* None | True：接受请求。* False：拒绝请求并返回 403 错误。* FilterType：要应用的过滤器。

Authenticator `module-attribute` ¶

Authenticator = Callable[
    ...,
    Awaitable[
        Union[
            MinimalUser,
            str,
            BaseUser,
            MinimalUserDict,
            Mapping[str, Any],
        ],
    ],
]

身份验证函数的类型。

身份验证器可以返回以下任一：1. 字符串 (user_id) 2. 包含 {"identity": str, "permissions": list[str]} 的字典 3. 具有 identity 和 permissions 属性的对象

您的授权逻辑可以在下游使用权限来确定对不同资源的访问权限。

authenticate 装饰器将自动按名称注入以下任何参数（如果包含在您的函数签名中）：

参数

名称	类型	描述	默认值
`request`	`Request`	原始 ASGI 请求对象	必需
`body`	`dict`	已解析的请求体	必需
`path`	`str`	请求路径	必需
`method`	`str`	HTTP 方法（GET、POST 等）	必需
`path_params`	`dict[str, str] \| None`	URL 路径参数	必需
`query_params`	`dict[str, str] \| None`	URL 查询参数	必需
`headers`	`dict[str, bytes] \| None`	请求头	必需
`authorization`	`str \| None`	Authorization 头的值（例如 "Bearer"）")	必需

示例

使用令牌的基本身份验证

from langgraph_sdk import Auth

auth = Auth()

@auth.authenticate
async def authenticate1(authorization: str) -> Auth.types.MinimalUserDict:
    return await get_user(authorization)

使用多个参数进行身份验证

@auth.authenticate
async def authenticate2(
    method: str,
    path: str,
    headers: dict[str, bytes]
) -> Auth.types.MinimalUserDict:
    # Custom auth logic using method, path and headers
    user = verify_request(method, path, headers)
    return user

接受原始 ASGI 请求

MY_SECRET = "my-secret-key"
@auth.authenticate
async def get_current_user(request: Request) -> Auth.types.MinimalUserDict:
    try:
        token = (request.headers.get("authorization") or "").split(" ", 1)[1]
        payload = jwt.decode(token, MY_SECRET, algorithms=["HS256"])
    except (IndexError, InvalidTokenError):
        raise HTTPException(
            status_code=401,
            detail="Invalid token",
            headers={"WWW-Authenticate": "Bearer"},
        )

    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://api.myauth-provider.com/auth/v1/user",
            headers={"Authorization": f"Bearer {MY_SECRET}"}
        )
        if response.status_code != 200:
            raise HTTPException(status_code=401, detail="User not found")

        user_data = response.json()
        return {
            "identity": user_data["id"],
            "display_name": user_data.get("name"),
            "permissions": user_data.get("permissions", []),
            "is_authenticated": True,
        }

MinimalUser ¶

Bases: Protocol

用户对象至少必须公开 identity 属性。

属性

名称	类型	描述
`identity`	`str`	用户的唯一标识符。

identity `property` ¶

identity: str

用户的唯一标识符。

这可以是用户名、电子邮件或用于区分系统中不同用户的任何其他唯一标识符。

MinimalUserDict ¶

基类：TypedDict

用户的字典表示。

属性

名称	类型	描述
`identity`	`Required[str]`	用户所需的唯一标识符。
`display_name`	`str`	用户的可选显示名称 (typing.Optional)。
`is_authenticated`	`bool`	用户是否已认证。默认为 True。
`permissions`	`Sequence[str]`	与用户关联的权限列表。

identity `instance-attribute` ¶

identity: Required[str]

用户所需的唯一标识符。

display_name `instance-attribute` ¶

display_name: str

用户的可选显示名称 (typing.Optional)。

is_authenticated `instance-attribute` ¶

is_authenticated: bool

用户是否已认证。默认为 True。

permissions `instance-attribute` ¶

permissions: Sequence[str]

与用户关联的权限列表。

您可以在您的 @auth.on 授权逻辑中使用这些权限来确定对不同资源的访问权限。

BaseUser ¶

Bases: Protocol

基本 ASGI 用户协议

方法

名称	描述
`__getitem__`	从您的最小用户字典中获取键。
`__contains__`	检查属性是否存在。
`__iter__`	迭代用户的键。

属性

名称	类型	描述
`is_authenticated`	`bool`	用户是否已认证。
`display_name`	`str`	用户的显示名称。
`identity`	`str`	用户的唯一标识符。
`permissions`	`Sequence[str]`	与用户关联的权限。

is_authenticated `property` ¶

is_authenticated: bool

用户是否已认证。

display_name `property` ¶

display_name: str

用户的显示名称。

identity `property` ¶

identity: str

用户的唯一标识符。

permissions `property` ¶

permissions: Sequence[str]

与用户关联的权限。

getitem ¶

__getitem__(key)

从您的最小用户字典中获取键。

contains ¶

__contains__(key)

检查属性是否存在。

iter ¶

__iter__()

迭代用户的键。

StudioUser ¶

一个用户对象，它由来自 LangGraph studio 的已认证请求填充。

注意：可以在您的 langgraph.json 配置中禁用 Studio 身份验证。

{
  "auth": {
    "disable_studio_auth": true
  }
}

您可以在授权处理器 (@auth.on) 中使用 isinstance 检查来专门控制从 LangGraph Studio UI 访问实例的开发人员的访问权限。

示例

@auth.on
async def allow_developers(ctx: Auth.types.AuthContext, value: Any) -> None:
    if isinstance(ctx.user, Auth.types.StudioUser):
        return None
    ...
    return False

BaseAuthContext ¶

身份验证上下文的基类。

提供授权决策所需的基本身份验证信息。

属性

名称	类型	描述
`permissions`	`Sequence[str]`	授予已认证用户的权限。
`user`	`BaseUser`	已认证用户。

permissions `instance-attribute` ¶

permissions: Sequence[str]

授予已认证用户的权限。

user `instance-attribute` ¶

user: BaseUser

已认证用户。

AuthContext ¶

Bases: BaseAuthContext

包含资源和操作信息的完整身份验证上下文。

扩展 BaseAuthContext，包含正在访问的特定资源和操作，允许进行精细的访问控制决策。

属性

名称	类型	描述
`resource`	`Literal['runs', 'threads', 'crons', 'assistants', 'store']`	正在访问的资源。
`action`	`Literal['create', 'read', 'update', 'delete', 'search', 'create_run', 'put', 'get', 'list_namespaces']`	对资源执行的操作。
`permissions`	`Sequence[str]`	授予已认证用户的权限。
`user`	`BaseUser`	已认证用户。

resource `instance-attribute` ¶

resource: Literal[
    "runs", "threads", "crons", "assistants", "store"
]

正在访问的资源。

action `instance-attribute` ¶

action: Literal[
    "create",
    "read",
    "update",
    "delete",
    "search",
    "create_run",
    "put",
    "get",
    "list_namespaces",
]

对资源执行的操作。

大多数资源支持以下操作：- create：创建新资源 - read：读取资源信息 - update：更新现有资源 - delete：删除资源 - search：搜索资源

store 支持以下操作：- put：在 store 中添加或更新文档 - get：从 store 获取文档 - list_namespaces：列出 store 中的命名空间

permissions `instance-attribute` ¶

permissions: Sequence[str]

授予已认证用户的权限。

user `instance-attribute` ¶

user: BaseUser

已认证用户。

ThreadsCreate ¶

基类：TypedDict

创建新线程的参数。

示例

create_params = {
    "thread_id": UUID("123e4567-e89b-12d3-a456-426614174000"),
    "metadata": {"owner": "user123"},
    "if_exists": "do_nothing"
}

属性

名称	类型	描述
`thread_id`	`UUID`	线程的唯一标识符。
`metadata`	`MetadataInput`	附加到线程的可选元数据 (typing.Optional)。
`if_exists`	`OnConflictBehavior`	当具有相同 ID 的线程已存在时的行为。

thread_id `instance-attribute` ¶

thread_id: UUID

线程的唯一标识符。

metadata `instance-attribute` ¶

metadata: MetadataInput

附加到线程的可选元数据 (typing.Optional)。

if_exists `instance-attribute` ¶

if_exists: OnConflictBehavior

当具有相同 ID 的线程已存在时的行为。

ThreadsRead ¶

基类：TypedDict

读取线程状态或运行信息的参数。

此类型用于三种上下文：1. 读取线程、线程版本或线程状态信息：仅提供 thread_id 2. 读取运行信息：同时提供 thread_id 和 run_id

属性

名称	类型	描述
`thread_id`	`UUID`	线程的唯一标识符。
`run_id`	`Optional[UUID]`	用于过滤的运行 ID。仅在读取线程内的运行信息时使用。

thread_id `instance-attribute` ¶

thread_id: UUID

线程的唯一标识符。

run_id `instance-attribute` ¶

run_id: Optional[UUID]

用于过滤的运行 ID。仅在读取线程内的运行信息时使用。

ThreadsUpdate ¶

基类：TypedDict

更新线程或运行的参数。

当更新线程、线程版本或取消运行时调用。

属性

名称	类型	描述
`thread_id`	`UUID`	线程的唯一标识符。
`metadata`	`MetadataInput`	要更新的可选元数据 (typing.Optional)。
`action`	`Optional[Literal['interrupt', 'rollback']]`	对线程执行的可选操作 (typing.Optional)。

thread_id `instance-attribute` ¶

thread_id: UUID

线程的唯一标识符。

metadata `instance-attribute` ¶

metadata: MetadataInput

要更新的可选元数据 (typing.Optional)。

action `instance-attribute` ¶

action: Optional[Literal['interrupt', 'rollback']]

对线程执行的可选操作 (typing.Optional)。

ThreadsDelete ¶

基类：TypedDict

删除线程的参数。

当删除线程、线程版本或运行时调用

属性

名称	类型	描述
`thread_id`	`UUID`	线程的唯一标识符。
`run_id`	`Optional[UUID]`	用于过滤的可选运行 ID (typing.Optional)。

thread_id `instance-attribute` ¶

thread_id: UUID

线程的唯一标识符。

run_id `instance-attribute` ¶

run_id: Optional[UUID]

用于过滤的可选运行 ID (typing.Optional)。

ThreadsSearch ¶

基类：TypedDict

搜索线程的参数。

当搜索线程或运行时调用。

属性

名称	类型	描述
`metadata`	`MetadataInput`	用于过滤的可选元数据 (typing.Optional)。
`values`	`MetadataInput`	用于过滤的可选值 (typing.Optional)。
`status`	`Optional[ThreadStatus]`	用于过滤的可选状态 (typing.Optional)。
`limit`	`int`	要返回的最大结果数。
`offset`	`int`	分页偏移量。
`thread_id`	`Optional[UUID]`	用于过滤的可选线程 ID (typing.Optional)。

metadata `instance-attribute` ¶

metadata: MetadataInput

用于过滤的可选元数据 (typing.Optional)。

values `instance-attribute` ¶

values: MetadataInput

用于过滤的可选值 (typing.Optional)。

status `instance-attribute` ¶

status: Optional[ThreadStatus]

用于过滤的可选状态 (typing.Optional)。

limit `instance-attribute` ¶

limit: int

要返回的最大结果数。

offset `instance-attribute` ¶

offset: int

分页偏移量。

thread_id `instance-attribute` ¶

thread_id: Optional[UUID]

用于过滤的可选线程 ID (typing.Optional)。

RunsCreate ¶

基类：TypedDict

创建运行的有效载荷。

示例

create_params = {
    "assistant_id": UUID("123e4567-e89b-12d3-a456-426614174000"),
    "thread_id": UUID("123e4567-e89b-12d3-a456-426614174001"),
    "run_id": UUID("123e4567-e89b-12d3-a456-426614174002"),
    "status": "pending",
    "metadata": {"owner": "user123"},
    "prevent_insert_if_inflight": True,
    "multitask_strategy": "reject",
    "if_not_exists": "create",
    "after_seconds": 10,
    "kwargs": {"key": "value"},
    "action": "interrupt"
}

属性

名称	类型	描述
`assistant_id`	`Optional[UUID]`	此运行要使用的可选助手 ID (typing.Optional)。
`thread_id`	`Optional[UUID]`	此运行要使用的可选线程 ID (typing.Optional)。
`run_id`	`Optional[UUID]`	此运行要使用的可选运行 ID (typing.Optional)。
`status`	`Optional[RunStatus]`	此运行的可选状态 (typing.Optional)。
`metadata`	`MetadataInput`	此运行的可选元数据 (typing.Optional)。
`prevent_insert_if_inflight`	`bool`	如果已有正在进行的运行，则阻止插入新运行。
`multitask_strategy`	`MultitaskStrategy`	此运行的多任务策略。
`if_not_exists`	`IfNotExists`	此运行的 IfNotExists。
`after_seconds`	`int`	创建运行前等待的秒数。
`kwargs`	`Dict[str, Any]`	传递给运行的关键字参数。
`action`	`Optional[Literal['interrupt', 'rollback']]`	如果更新现有运行要采取的操作。

assistant_id `instance-attribute` ¶

assistant_id: Optional[UUID]

此运行要使用的可选助手 ID (typing.Optional)。

thread_id `instance-attribute` ¶

thread_id: Optional[UUID]

此运行要使用的可选线程 ID (typing.Optional)。

run_id `instance-attribute` ¶

run_id: Optional[UUID]

此运行要使用的可选运行 ID (typing.Optional)。

status `instance-attribute` ¶

status: Optional[RunStatus]

此运行的可选状态 (typing.Optional)。

metadata `instance-attribute` ¶

metadata: MetadataInput

此运行的可选元数据 (typing.Optional)。

prevent_insert_if_inflight `instance-attribute` ¶

prevent_insert_if_inflight: bool

如果已有正在进行的运行，则阻止插入新运行。

multitask_strategy `instance-attribute` ¶

multitask_strategy: MultitaskStrategy

此运行的多任务策略。

if_not_exists `instance-attribute` ¶

if_not_exists: IfNotExists

此运行的 IfNotExists。

after_seconds `instance-attribute` ¶

after_seconds: int

创建运行前等待的秒数。

kwargs `instance-attribute` ¶

kwargs: Dict[str, Any]

传递给运行的关键字参数。

action `instance-attribute` ¶

action: Optional[Literal['interrupt', 'rollback']]

如果更新现有运行要采取的操作。

AssistantsCreate ¶

基类：TypedDict

创建助手的有效载荷。

示例

create_params = {
    "assistant_id": UUID("123e4567-e89b-12d3-a456-426614174000"),
    "graph_id": "graph123",
    "config": {"key": "value"},
    "metadata": {"owner": "user123"},
    "if_exists": "do_nothing",
    "name": "Assistant 1"
}

属性

名称	类型	描述
`assistant_id`	`UUID`	助手的唯一标识符。
`graph_id`	`str`	此助手要使用的图谱ID。
`config`	`Optional[Union[Dict[str, Any], Any]]`	助手的可选配置。
`metadata`	`MetadataInput`	要附加到助手的可选元数据。
`if_exists`	`OnConflictBehavior`	当具有相同ID的助手已存在时的行为。
`name`	`str`	助手的名称。

assistant_id `instance-attribute` ¶

assistant_id: UUID

助手的唯一标识符。

graph_id `instance-attribute` ¶

graph_id: str

此助手要使用的图谱ID。

config `instance-attribute` ¶

config: Optional[Union[Dict[str, Any], Any]]

助手的可选配置。

metadata `instance-attribute` ¶

metadata: MetadataInput

要附加到助手的可选元数据。

if_exists `instance-attribute` ¶

if_exists: OnConflictBehavior

当具有相同ID的助手已存在时的行为。

name `instance-attribute` ¶

name: str

助手的名称。

AssistantsRead ¶

基类：TypedDict

读取助手的有效载荷。

示例

read_params = {
    "assistant_id": UUID("123e4567-e89b-12d3-a456-426614174000"),
    "metadata": {"owner": "user123"}
}

属性

名称	类型	描述
`assistant_id`	`UUID`	助手的唯一标识符。
`metadata`	`MetadataInput`	用于过滤的可选元数据 (typing.Optional)。

assistant_id `instance-attribute` ¶

assistant_id: UUID

助手的唯一标识符。

metadata `instance-attribute` ¶

metadata: MetadataInput

用于过滤的可选元数据 (typing.Optional)。

AssistantsUpdate ¶

基类：TypedDict

更新助手的有效载荷。

示例

update_params = {
    "assistant_id": UUID("123e4567-e89b-12d3-a456-426614174000"),
    "graph_id": "graph123",
    "config": {"key": "value"},
    "metadata": {"owner": "user123"},
    "name": "Assistant 1",
    "version": 1
}

属性

名称	类型	描述
`assistant_id`	`UUID`	助手的唯一标识符。
`graph_id`	`Optional[str]`	要更新的可选图谱ID。
`config`	`Optional[Union[Dict[str, Any], Any]]`	要更新的可选配置。
`metadata`	`MetadataInput`	要更新的可选元数据 (typing.Optional)。
`name`	`Optional[str]`	要更新的可选名称。
`version`	`Optional[int]`	要更新的可选版本。

assistant_id `instance-attribute` ¶

assistant_id: UUID

助手的唯一标识符。

graph_id `instance-attribute` ¶

graph_id: Optional[str]

要更新的可选图谱ID。

config `instance-attribute` ¶

config: Optional[Union[Dict[str, Any], Any]]

要更新的可选配置。

metadata `instance-attribute` ¶

metadata: MetadataInput

要更新的可选元数据 (typing.Optional)。

name `instance-attribute` ¶

name: Optional[str]

要更新的可选名称。

version `instance-attribute` ¶

version: Optional[int]

要更新的可选版本。

AssistantsDelete ¶

基类：TypedDict

删除助手的有效载荷。

示例

delete_params = {
    "assistant_id": UUID("123e4567-e89b-12d3-a456-426614174000")
}

属性

名称	类型	描述
`assistant_id`	`UUID`	助手的唯一标识符。

assistant_id `instance-attribute` ¶

assistant_id: UUID

助手的唯一标识符。

AssistantsSearch ¶

基类：TypedDict

搜索助手的有效载荷。

示例

search_params = {
    "graph_id": "graph123",
    "metadata": {"owner": "user123"},
    "limit": 10,
    "offset": 0
}

属性

名称	类型	描述
`graph_id`	`Optional[str]`	用于筛选的可选图谱ID。
`metadata`	`MetadataInput`	用于过滤的可选元数据 (typing.Optional)。
`limit`	`int`	要返回的最大结果数。
`offset`	`int`	分页偏移量。

graph_id `instance-attribute` ¶

graph_id: Optional[str]

用于筛选的可选图谱ID。

metadata `instance-attribute` ¶

metadata: MetadataInput

用于过滤的可选元数据 (typing.Optional)。

limit `instance-attribute` ¶

limit: int

要返回的最大结果数。

offset `instance-attribute` ¶

offset: int

分页偏移量。

CronsCreate ¶

基类：TypedDict

创建定时任务的载荷。

示例

create_params = {
    "payload": {"key": "value"},
    "schedule": "0 0 * * *",
    "cron_id": UUID("123e4567-e89b-12d3-a456-426614174000"),
    "thread_id": UUID("123e4567-e89b-12d3-a456-426614174001"),
    "user_id": "user123",
    "end_time": datetime(2024, 3, 16, 10, 0, 0)
}

属性

名称	类型	描述
`payload`	`Dict[str, Any]`	定时任务的载荷。
`schedule`	`str`	定时任务的计划。
`cron_id`	`Optional[UUID]`	定时任务的可选唯一标识符。
`thread_id`	`Optional[UUID]`	此定时任务要使用的可选线程ID。
`user_id`	`Optional[str]`	此定时任务要使用的可选用户ID。
`end_time`	`Optional[datetime]`	定时任务的可选结束时间。

payload `instance-attribute` ¶

payload: Dict[str, Any]

定时任务的载荷。

schedule `instance-attribute` ¶

schedule: str

定时任务的计划。

cron_id `instance-attribute` ¶

cron_id: Optional[UUID]

定时任务的可选唯一标识符。

thread_id `instance-attribute` ¶

thread_id: Optional[UUID]

此定时任务要使用的可选线程ID。

user_id `instance-attribute` ¶

user_id: Optional[str]

此定时任务要使用的可选用户ID。

end_time `instance-attribute` ¶

end_time: Optional[datetime]

定时任务的可选结束时间。

CronsDelete ¶

基类：TypedDict

删除定时任务的载荷。

示例

delete_params = {
    "cron_id": UUID("123e4567-e89b-12d3-a456-426614174000")
}

属性

名称	类型	描述
`cron_id`	`UUID`	定时任务的唯一标识符。

cron_id `instance-attribute` ¶

cron_id: UUID

定时任务的唯一标识符。

CronsRead ¶

基类：TypedDict

读取定时任务的载荷。

示例

read_params = {
    "cron_id": UUID("123e4567-e89b-12d3-a456-426614174000")
}

属性

名称	类型	描述
`cron_id`	`UUID`	定时任务的唯一标识符。

cron_id `instance-attribute` ¶

cron_id: UUID

定时任务的唯一标识符。

CronsUpdate ¶

基类：TypedDict

更新定时任务的载荷。

示例

update_params = {
    "cron_id": UUID("123e4567-e89b-12d3-a456-426614174000"),
    "payload": {"key": "value"},
    "schedule": "0 0 * * *"
}

属性

名称	类型	描述
`cron_id`	`UUID`	定时任务的唯一标识符。
`payload`	`Optional[Dict[str, Any]]`	要更新的可选载荷。
`schedule`	`Optional[str]`	要更新的可选计划。

cron_id `instance-attribute` ¶

cron_id: UUID

定时任务的唯一标识符。

payload `instance-attribute` ¶

payload: Optional[Dict[str, Any]]

要更新的可选载荷。

schedule `instance-attribute` ¶

schedule: Optional[str]

要更新的可选计划。

CronsSearch ¶

基类：TypedDict

搜索定时任务的载荷。

示例

search_params = {
    "assistant_id": UUID("123e4567-e89b-12d3-a456-426614174000"),
    "thread_id": UUID("123e4567-e89b-12d3-a456-426614174001"),
    "limit": 10,
    "offset": 0
}

属性

名称	类型	描述
`assistant_id`	`Optional[UUID]`	用于筛选的可选助手ID。
`thread_id`	`Optional[UUID]`	用于过滤的可选线程 ID (typing.Optional)。
`limit`	`int`	要返回的最大结果数。
`offset`	`int`	分页偏移量。

assistant_id `instance-attribute` ¶

assistant_id: Optional[UUID]

用于筛选的可选助手ID。

thread_id `instance-attribute` ¶

thread_id: Optional[UUID]

用于过滤的可选线程 ID (typing.Optional)。

limit `instance-attribute` ¶

limit: int

要返回的最大结果数。

offset `instance-attribute` ¶

offset: int

分页偏移量。

StoreGet ¶

基类：TypedDict

根据命名空间和键检索特定项的操作。

属性

名称	类型	描述
`namespace`	`tuple[str, ...]`	唯一标识项目位置的分层路径。
`key`	`str`	项目在其特定命名空间内的唯一标识符。

namespace `instance-attribute` ¶

namespace: tuple[str, ...]

唯一标识项目位置的分层路径。

key `instance-attribute` ¶

key: str

项目在其特定命名空间内的唯一标识符。

StoreSearch ¶

基类：TypedDict

在指定命名空间层次结构中搜索项的操作。

属性

名称	类型	描述
`namespace`	`tuple[str, ...]`	用于定义搜索范围的前缀过滤器。
`filter`	`Optional[dict[str, Any]]`	用于基于精确匹配或比较运算符筛选结果的键值对。
`limit`	`int`	搜索结果中要返回的最大项目数。
`offset`	`int`	用于分页跳过的匹配项数。
`query`	`Optional[str]`	自然语言搜索查询，用于语义搜索功能。

namespace `instance-attribute` ¶

namespace: tuple[str, ...]

用于定义搜索范围的前缀过滤器。

filter `instance-attribute` ¶

filter: Optional[dict[str, Any]]

用于基于精确匹配或比较运算符筛选结果的键值对。

limit `instance-attribute` ¶

limit: int

搜索结果中要返回的最大项目数。

offset `instance-attribute` ¶

offset: int

用于分页跳过的匹配项数。

query `instance-attribute` ¶

query: Optional[str]

自然语言搜索查询，用于语义搜索功能。

StoreListNamespaces ¶

基类：TypedDict

列出和过滤 store 中的命名空间的操作。

属性

名称	类型	描述
`namespace`	`Optional[tuple[str, ...]]`	前缀筛选命名空间。
`suffix`	`Optional[tuple[str, ...]]`	筛选命名空间的可选条件。
`max_depth`	`Optional[int]`	要返回的命名空间层级的最大深度。
`limit`	`int`	要返回的最大命名空间数。
`offset`	`int`	用于分页跳过的命名空间数。

namespace `instance-attribute` ¶

namespace: Optional[tuple[str, ...]]

前缀筛选命名空间。

suffix `instance-attribute` ¶

suffix: Optional[tuple[str, ...]]

筛选命名空间的可选条件。

max_depth `instance-attribute` ¶

max_depth: Optional[int]

要返回的命名空间层级的最大深度。

注意

深度超过此级别的命名空间将被截断。

limit `instance-attribute` ¶

limit: int

要返回的最大命名空间数。

offset `instance-attribute` ¶

offset: int

用于分页跳过的命名空间数。

StorePut ¶

基类：TypedDict

在 store 中存储、更新或删除项的操作。

属性

名称	类型	描述
`namespace`	`tuple[str, ...]`	标识项目位置的分层路径。
`key`	`str`	项目在其命名空间内的唯一标识符。
`value`	`Optional[dict[str, Any]]`	要存储的数据，或None表示标记删除该项目。
`index`	`Optional[Union[Literal[False], list[str]]]`	全文搜索的可选索引配置。

namespace `instance-attribute` ¶

namespace: tuple[str, ...]

标识项目位置的分层路径。

key `instance-attribute` ¶

key: str

项目在其命名空间内的唯一标识符。

value `instance-attribute` ¶

value: Optional[dict[str, Any]]

要存储的数据，或None表示标记删除该项目。

index `instance-attribute` ¶

index: Optional[Union[Literal[False], list[str]]]

全文搜索的可选索引配置。

StoreDelete ¶

基类：TypedDict

从 store 中删除项的操作。

属性

名称	类型	描述
`namespace`	`tuple[str, ...]`	唯一标识项目位置的分层路径。
`key`	`str`	项目在其特定命名空间内的唯一标识符。

namespace `instance-attribute` ¶

namespace: tuple[str, ...]

唯一标识项目位置的分层路径。

key `instance-attribute` ¶

key: str

项目在其特定命名空间内的唯一标识符。

on ¶

不同 API 操作的类型定义的命名空间。

此类组织了跨不同资源（线程、助手、定时任务、存储）的创建、读取、更新、删除和搜索操作的类型定义。

用法

from langgraph_sdk import Auth

auth = Auth()

@auth.on
def handle_all(params: Auth.on.value):
    raise Exception("Not authorized")

@auth.on.threads.create
def handle_thread_create(params: Auth.on.threads.create.value):
    # Handle thread creation
    pass

@auth.on.assistants.search
def handle_assistant_search(params: Auth.on.assistants.search.value):
    # Handle assistant search
    pass

类

名称	描述
`threads`	与线程相关的操作类型。
`assistants`	与助手相关的操作类型。
`crons`	与定时任务相关的操作类型。
`store`	与存储相关的操作类型。

threads ¶

与线程相关的操作类型。

类

名称	描述
`create`	线程创建参数类型。
`create_run`	创建或流式传输运行的类型。
`read`	线程读取参数类型。
`update`	线程更新参数类型。
`delete`	线程删除参数类型。
`search`	线程搜索参数类型。

create ¶

线程创建参数类型。

create_run ¶

创建或流式传输运行的类型。

read ¶

线程读取参数类型。

update ¶

线程更新参数类型。

delete ¶

线程删除参数类型。

search ¶

线程搜索参数类型。

assistants ¶

与助手相关的操作类型。

类

名称	描述
`create`	助手创建参数类型。
`read`	助手读取参数类型。
`update`	助手更新参数类型。
`delete`	助手删除参数类型。
`search`	助手搜索参数类型。

create ¶

助手创建参数类型。

read ¶

助手读取参数类型。

update ¶

助手更新参数类型。

delete ¶

助手删除参数类型。

search ¶

助手搜索参数类型。

crons ¶

与定时任务相关的操作类型。

类

名称	描述
`create`	定时任务创建参数类型。
`read`	定时任务读取参数类型。
`update`	定时任务更新参数类型。
`delete`	定时任务删除参数类型。
`search`	定时任务搜索参数类型。

create ¶

定时任务创建参数类型。

read ¶

定时任务读取参数类型。

update ¶

定时任务更新参数类型。

delete ¶

定时任务删除参数类型。

search ¶

定时任务搜索参数类型。

store ¶

与存储相关的操作类型。

类

名称	描述
`put`	存储put参数类型。
`get`	存储get参数类型。
`search`	存储search参数类型。
`delete`	存储delete参数类型。
`list_namespaces`	存储list namespaces参数类型。

put ¶

存储put参数类型。

get ¶

存储get参数类型。

search ¶

存储search参数类型。

delete ¶

存储delete参数类型。

list_namespaces ¶

存储list namespaces参数类型。

身份验证系统使用的异常。

类

名称	描述
`HTTPException`	您可以抛出此HTTP异常来返回特定的HTTP错误响应。

HTTPException ¶

基类: Exception

您可以抛出此HTTP异常来返回特定的HTTP错误响应。

由于此异常定义在 auth 模块中，因此默认状态码为 401。

参数

名称	类型	描述	默认值
`status_code`	`int`	错误的HTTP状态码。默认为 401 "Unauthorized"。	`401`
`detail`	`Optional[str]`	详细错误消息。如果为 None，则根据状态码使用默认消息。	`无`
`headers`	`Optional[Mapping[str, str]]`	错误响应中包含的额外HTTP头部。	`无`

示例

默认值

raise HTTPException()
# HTTPException(status_code=401, detail='Unauthorized')

添加头部

raise HTTPException(headers={"X-Custom-Header": "Custom Value"})
# HTTPException(status_code=401, detail='Unauthorized', headers={"WWW-Authenticate": "Bearer"})

自定义错误

raise HTTPException(status_code=404, detail="Not found")

Python SDK 参考¶

LangGraphClient ¶

HttpClient ¶

get async ¶

post async ¶

put async ¶

patch async ¶

delete async ¶

stream async ¶

AssistantsClient ¶

get async ¶

get_graph async ¶

get_schemas async ¶

get_subgraphs async ¶

create async ¶

update async ¶

delete async ¶

search async ¶

get_versions async ¶

set_latest async ¶

ThreadsClient ¶

get async ¶

create async ¶

update async ¶

delete async ¶

search async ¶

copy async ¶

get_state async ¶

update_state async ¶

get_history async ¶

RunsClient ¶

stream ¶

create async ¶

create_batch async ¶

wait async ¶

list async ¶

get async ¶

cancel async ¶

join async ¶

join_stream ¶

delete async ¶

CronClient ¶

create_for_thread async ¶

create async ¶

delete async ¶

search async ¶

StoreClient ¶

put_item async ¶

get_item async ¶

delete_item async ¶

search_items async ¶

list_namespaces async ¶

SyncLangGraphClient ¶

SyncHttpClient ¶

get ¶

post ¶

put ¶

patch ¶

delete ¶

stream ¶

SyncAssistantsClient ¶

get ¶

get_graph ¶

get_schemas ¶

get_subgraphs ¶

create ¶

update ¶

delete ¶

search ¶

get_versions ¶

set_latest ¶

SyncThreadsClient ¶

get ¶

create ¶

update ¶

delete ¶

search ¶

copy ¶

get_state ¶

update_state ¶

get `async` ¶

post `async` ¶

put `async` ¶

patch `async` ¶

delete `async` ¶

stream `async` ¶

get `async` ¶

get_graph `async` ¶

get_schemas `async` ¶

get_subgraphs `async` ¶

create `async` ¶

update `async` ¶

delete `async` ¶

search `async` ¶

get_versions `async` ¶

set_latest `async` ¶

get `async` ¶

create `async` ¶

update `async` ¶

delete `async` ¶

search `async` ¶

copy `async` ¶

get_state `async` ¶

update_state `async` ¶

get_history `async` ¶

create `async` ¶

create_batch `async` ¶

wait `async` ¶

list `async` ¶

get `async` ¶

cancel `async` ¶

join `async` ¶

delete `async` ¶

create_for_thread `async` ¶

create `async` ¶

delete `async` ¶

search `async` ¶

put_item `async` ¶

get_item `async` ¶

delete_item `async` ¶

search_items `async` ¶

list_namespaces `async` ¶

Json `模块属性` ¶

RunStatus `模块属性` ¶

ThreadStatus `模块属性` ¶

StreamMode `模块属性` ¶

DisconnectMode `模块属性` ¶

MultitaskStrategy `模块属性` ¶

OnConflictBehavior `模块属性` ¶

OnCompletionBehavior `模块属性` ¶

All `模块属性` ¶

IfNotExists `模块属性` ¶

CancelAction `模块属性` ¶

AssistantSortBy `模块属性` ¶

ThreadSortBy `模块属性` ¶

SortOrder `模块属性` ¶

tags `实例属性` ¶

recursion_limit `实例属性` ¶

configurable `实例属性` ¶

thread_id `实例属性` ¶

checkpoint_ns `实例属性` ¶

checkpoint_id `实例属性` ¶

checkpoint_map `实例属性` ¶

graph_id `实例属性` ¶

input_schema `实例属性` ¶

output_schema `实例属性` ¶

state_schema `实例属性` ¶

config_schema `实例属性` ¶

assistant_id `实例属性` ¶

graph_id `实例属性` ¶

config `实例属性` ¶

created_at `实例属性` ¶

metadata `实例属性` ¶

version `实例属性` ¶

name `实例属性` ¶

description `实例属性` ¶

assistant_id `实例属性` ¶

graph_id `实例属性` ¶

config `实例属性` ¶

created_at `实例属性` ¶