【LangGraph】Python SDK：连接 LangGraph API 的客户端实现

LangGraph Python SDK 提供异步（LangGraphClient）和同步（SyncLangGraphClient）客户端，用于管理 LangGraph API 的核心资源（助手、线程、运行、定时任务和存储），支持 Python 和 JavaScript 项目，适合开发、测试和生产部署。核心类：LangGraphClient / SyncLangGraphClient：顶级客户端，

正义的彬彬侠

1585人浏览 · 2025-05-22 10:12:16

正义的彬彬侠 · 2025-05-22 10:12:16 发布

LangGraph Python SDK 参考

LangGraph Python SDK 提供了用于连接 LangGraph API 的客户端实现，包含异步（get_client(url="http://localhost:2024") 或 LangGraphClient）和同步（get_sync_client(url="http://localhost:2024") 或 SyncLangGraphClient）客户端，用于与 LangGraph API 的核心资源（如助手、线程、运行和定时任务）以及持久化文档存储进行交互。

1. 类

1.1 类 `LangGraphClient`

描述：
LangGraph API 的顶级异步客户端。

1.1.1 属性

名称	类型	描述
`assistants`	`AssistantsClient`	管理图的版本化配置。
`threads`	`ThreadsClient`	处理（可能为）多轮交互，例如会话线程。
`runs`	`RunsClient`	控制图的单次调用。
`crons`	`CronClient`	管理定时操作。
`store`	`StoreClient`	与持久化、共享数据存储交互。

1.2 类 `HttpClient`

描述：
处理 LangGraph API 的异步请求，在提供的 httpx 客户端上添加额外的错误消息和内容处理。

1.2.1 属性

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

1.2.2 方法

名称	描述
`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 请求。

参数：

名称	类型	描述	默认值
`path`	`str`	请求路径。	必填
`params`	`Optional[QueryParamTypes]`	查询参数。	`None`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`Any`	请求响应内容。

方法 `post`

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

描述：
发送 POST 请求。

参数：

名称	类型	描述	默认值
`path`	`str`	请求路径。	必填
`json`	`Optional[dict]`	JSON 请求体。	`None`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`Any`	请求响应内容。

方法 `put`

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

描述：
发送 PUT 请求。

参数：

名称	类型	描述	默认值
`path`	`str`	请求路径。	必填
`json`	`dict`	JSON 请求体。	必填
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`Any`	请求响应内容。

方法 `patch`

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

描述：
发送 PATCH 请求。

参数：

名称	类型	描述	默认值
`path`	`str`	请求路径。	必填
`json`	`dict`	JSON 请求体。	必填
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`Any`	请求响应内容。

方法 `delete`

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

描述：
发送 DELETE 请求。

参数：

名称	类型	描述	默认值
`path`	`str`	请求路径。	必填
`json`	`Optional[Any]`	JSON 请求体。	`None`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`None`	无返回值。

方法 `stream`

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

描述：
使用服务器发送事件（SSE）流式传输结果。

参数：

名称	类型	描述	默认值
`path`	`str`	请求路径。	必填
`method`	`str`	HTTP 方法（如 `GET`、`POST`）。	必填
`json`	`Optional[dict]`	JSON 请求体。	`None`
`params`	`Optional[QueryParamTypes]`	查询参数。	`None`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

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

1.3 类 `AssistantsClient`

描述：
管理 LangGraph 中助手的异步客户端，助手是图的版本化配置。

1.3.1 示例

client = get_client(url="http://localhost:2024")
assistant = await client.assistants.get("assistant_id_123")

1.3.2 方法

名称	描述
`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]]`	可选的自定义请求头。	`None`

返回值：

名称	类型	描述
`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]]`	可选的自定义请求头。	`None`

返回值：

名称	类型	描述
`Graph`	`dict[str, list[dict[str, Any]]]`	助手的图信息（JSON 格式）。

示例：

client = get_client(url="http://localhost:2024")
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]]`	可选的自定义请求头。	`None`

返回值：

名称	类型	描述
`GraphSchema`	`GraphSchema`	助手的图模式。

示例：

client = get_client(url="http://localhost:2024")
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]`	可选的命名空间过滤器。	`None`
`recurse`	`bool`	是否递归获取子图。	`False`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

名称	类型	描述
`Subgraphs`	`Subgraphs`	助手的子图模式。

方法 `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。通常在 `langgraph.json` 配置中设置。	必填
`config`	`Optional[Config]`	图的配置。	`None`
`metadata`	`Json`	添加到助手的元数据。	`None`
`assistant_id`	`Optional[str]`	助手 ID，如果未提供，将默认生成随机 UUID。	`None`
`if_exists`	`Optional[OnConflictBehavior]`	处理重复创建的方式。默认在底层为 `'raise'`，可为 `'raise'`（如果重复则抛出错误）或 `'do_nothing'`（返回现有助手）。	`None`
`name`	`Optional[str]`	助手名称，默认在底层为 `'Untitled'`。	`None`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`
`description`	`Optional[str]`	助手的可选描述。需 `langgraph-api` 服务器版本 >= 0.0.45。	`None`

返回值：

名称	类型	描述
`Assistant`	`Assistant`	创建的助手。

示例：

client = get_client(url="http://localhost:2024")
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`	要更新的助手 ID。	必填
`graph_id`	`Optional[str]`	助手使用的图 ID。通常在 `langgraph.json` 配置中设置。如果为 `None`，助手保持指向原图。	`None`
`config`	`Optional[Config]`	图的配置。	`None`
`metadata`	`Json`	与现有助手元数据合并的元数据。	`None`
`name`	`Optional[str]`	助手的新名称。	`None`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`
`description`	`Optional[str]`	助手的可选描述。需 `langgraph-api` 服务器版本 >= 0.0.45。	`None`

返回值：

名称	类型	描述
`Assistant`	`Assistant`	更新后的助手。

示例：

client = get_client(url="http://localhost:2024")
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]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`None`	无返回值。

示例：

client = get_client(url="http://localhost:2024")
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`	用于过滤的元数据，每个键值对进行精确匹配。	`None`
`graph_id`	`Optional[str]`	用于过滤的图 ID。通常在 `langgraph.json` 配置中设置。	`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]`	匹配搜索参数的助手列表。

示例：

client = get_client(url="http://localhost:2024")
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`	用于过滤版本的元数据，每个键值对进行精确匹配。	`None`
`limit`	`int`	返回的最大版本数。	`10`
`offset`	`int`	跳过的版本数。	`0`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`list[AssistantVersion]`	助手版本列表。

示例：

client = get_client(url="http://localhost:2024")
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]]`	可选的自定义请求头。	`None`

返回值：

名称	类型	描述
`Assistant`	`Assistant`	助手对象。

示例：

client = get_client(url="http://localhost:2024")
new_version_assistant = await client.assistants.set_latest(
    assistant_id="my_assistant_id",
    version=3
)

1.4 类 `ThreadsClient`

描述：
管理 LangGraph 中线程的异步客户端。
线程维护图在多次交互/调用（即运行）之间的状态，累积并持久化图的状态，允许在图的单独调用之间保持连续性。

1.4.1 示例

client = get_client(url="http://localhost:2024")
new_thread = await client.threads.create(metadata={"user_id": "123"})

1.4.2 方法

名称	描述
`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]]`	可选的自定义请求头。	`None`

返回值：

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

示例：

client = get_client(url="http://localhost:2024")
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`	添加到线程的元数据。	`None`
`thread_id`	`Optional[str]`	线程 ID。如果为 `None`，将生成随机 UUID。	`None`
`if_exists`	`Optional[OnConflictBehavior]`	处理重复创建的方式。默认在底层为 `'raise'`，可为 `'raise'`（如果重复则抛出错误）或 `'do_nothing'`（返回现有线程）。	`None`
`supersteps`	`Optional[Sequence[dict[str, Sequence[dict[str, Any]]]]]`	创建线程时应用的超级步骤列表，每个包含更新序列。每个更新包含 `values` 或 `command` 和 `as_node`。用于在部署之间复制线程。	`None`
`graph_id`	`Optional[str]`	与线程关联的可选图 ID。	`None`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

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

示例：

client = get_client(url="http://localhost:2024")
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]]`	可选的自定义请求头。	`None`

返回值：

名称	类型	描述
`Thread`	`Thread`	更新后的线程。

示例：

client = get_client(url="http://localhost:2024")
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]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`None`	无返回值。

示例：

client = get_client(url="http://localhost:2024")
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`	用于过滤的线程元数据。	`None`
`values`	`Json`	用于过滤的状态值。	`None`
`status`	`Optional[ThreadStatus]`	过滤的线程状态，必须为 `'idle'`、`'busy'`、`'interrupted'` 或 `'error'`。	`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]`	匹配搜索参数的线程列表。

示例：

client = get_client(url="http://localhost:2024")
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]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`None`	无返回值。

示例：

client = get_client(url="http://localhost:2024")
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]`	要获取状态的检查点。	`None`
`checkpoint_id`	`Optional[str]`	（已弃用）要获取状态的检查点 ID。	`None`
`subgraphs`	`bool`	是否包含子图状态。	`False`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

名称	类型	描述
`ThreadState`	`ThreadState`	线程状态。

示例：

client = get_client(url="http://localhost:2024")
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]`	以此节点刚执行的方式更新状态。	`None`
`checkpoint`	`Optional[Checkpoint]`	要更新状态的检查点。	`None`
`checkpoint_id`	`Optional[str]`	（已弃用）要更新状态的检查点 ID。	`None`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

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

示例：

client = get_client(url="http://localhost:2024")
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]`	返回此子图的状态。如果为空，默认返回根状态。	`None`
`limit`	`int`	返回的最大状态数。	`10`
`before`	`Optional[str	Checkpoint]`	返回此检查点之前的状态。
`metadata`	`Optional[dict]`	按元数据键值对过滤状态。	`None`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

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

示例：

client = get_client(url="http://localhost:2024")
thread_state = await client.threads.get_history(
    thread_id="my_thread_id",
    limit=5
)

1.5 类 `RunsClient`

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

1.5.1 示例

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

1.5.2 方法

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

方法 `stream`

async 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]`	图的输入。	`None`
`command`	`Optional[Command]`	要执行的命令。不可与 `input` 结合使用。	`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]`	（已弃用）恢复的检查点 ID。	`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]`	断开连接模式，必须为 `'cancel'` 或 `'continue'`。	`None`
`on_completion`	`Optional[OnCompletionBehavior]`	无状态运行创建的线程是删除还是保留，必须为 `'delete'` 或 `'keep'`。	`None`
`webhook`	`Optional[str]`	LangGraph API 调用完成后调用的 webhook。	`None`
`multitask_strategy`	`Optional[MultitaskStrategy]`	多任务策略，必须为 `'reject'`、`'interrupt'`、`'rollback'` 或 `'enqueue'`。	`None`
`if_not_exists`	`Optional[IfNotExists]`	处理缺失线程的方式。默认 `'reject'`，可为 `'reject'`（如果缺失则抛出错误）或 `'create'`（创建新线程）。	`None`
`after_seconds`	`Optional[int]`	开始运行前等待的秒数。用于调度未来运行。	`None`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

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

示例：

client = get_client(url="http://localhost:2024")
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

描述：
创建后台运行。

参数：

与 stream 方法类似，省略 feedback_keys，添加以下参数：

名称	类型	描述	默认值
`run_id`	`Optional[str]`	运行 ID。如果为 `None`，将生成随机 UUID。	`None`

返回值：

名称	类型	描述
`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]

描述：
创建一批无状态后台运行。

参数：

名称	类型	描述	默认值
`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]`	图的输入。	`None`
`command`	`Optional[Command]`	要执行的命令。不可与 `input` 结合使用。	`None`
`metadata`	`Optional[dict]`	分配给运行的元数据。	`None`
`config`	`Optional[Config]`	助手的配置。	`None`
`checkpoint`	`Optional[Checkpoint]`	恢复的检查点。	`None`
`checkpoint_id`	`Optional[str]`	（已弃用）恢复的检查点 ID。	`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]`	LangGraph API 调用完成后调用的 webhook。	`None`
`on_disconnect`	`Optional[DisconnectMode]`	断开连接模式，必须为 `'cancel'` 或 `'continue'`。	`None`
`on_completion`	`Optional[OnCompletionBehavior]`	无状态运行创建的线程是删除还是保留，必须为 `'delete'` 或 `'keep'`。	`None`
`multitask_strategy`	`Optional[MultitaskStrategy]`	多任务策略，必须为 `'reject'`、`'interrupt'`、`'rollback'` 或 `'enqueue'`。	`None`
`if_not_exists`	`Optional[IfNotExists]`	处理缺失线程的方式。默认 `'reject'`，可为 `'reject'`（如果缺失则抛出错误）或 `'create'`（创建新线程）。	`None`
`after_seconds`	`Optional[int]`	开始运行前等待的秒数。用于调度未来运行。	`None`
`raise_error`	`bool`	是否在运行失败时抛出错误。	`True`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

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

示例：

client = get_client(url="http://localhost:2024")
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]`	用于过滤的运行状态。	`None`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

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

示例：

client = get_client(url="http://localhost:2024")
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]]`	可选的自定义请求头。	`None`

返回值：

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

示例：

client = get_client(url="http://localhost:2024")
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]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`None`	无返回值。

示例：

client = get_client(url="http://localhost:2024")
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]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`dict`	线程的最终状态。

示例：

client = get_client(url="http://localhost:2024")
result = await client.runs.join(
    thread_id="thread_id_to_join",
    run_id="run_id_to_join"
)

方法 `join_stream`

async 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]]]`	使用的流模式，必须是创建运行时传递的流模式的子集。后台运行默认包含所有流模式的并集。	`None`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

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

示例：

client = get_client(url="http://localhost:2024")
async for part in client.runs.join_stream(
    thread_id="thread_id_to_join",
    run_id="run_id_to_join",
    stream_mode=["values", "debug"]
):
    print(part)

方法 `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]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`None`	无返回值。

示例：

client = get_client(url="http://localhost:2024")
await client.runs.delete(
    thread_id="thread_id_to_delete",
    run_id="run_id_to_delete"
)

1.6 类 `CronClient`

描述：
管理 LangGraph 中定时任务（cron 作业）的异步客户端。
运行是助手的单次调用，可包含可选输入和配置。此客户端允许自动调度定期运行。

1.6.1 示例

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

1.6.2 功能可用性

定时任务客户端功能并非所有许可证都支持。请检查相关许可证文档以获取最新的功能可用性详情。

1.6.3 方法

名称	描述
`create_for_thread`	为线程创建定时任务。
`create`	创建定时运行。
`delete`	删除定时任务。
`search`	获取定时任务列表。

方法 `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

描述：
为线程创建定时任务。

参数：

名称	类型	描述	默认值
`thread_id`	`str`	运行定时任务的线程 ID。	必填
`assistant_id`	`str`	用于定时任务的助手 ID 或图名称。如果使用图名称，将默认使用从该图创建的第一个助手。	必填
`schedule`	`str`	执行此任务的 cron 调度表达式。	必填
`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]`	LangGraph API 调用完成后调用的 webhook。	`None`
`multitask_strategy`	`Optional[str]`	多任务策略，必须为 `'reject'`、`'interrupt'`、`'rollback'` 或 `'enqueue'`。	`None`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

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

示例：

client = get_client(url="http://localhost:2024")
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

描述：
创建定时运行。

参数：

名称	类型	描述	默认值
`assistant_id`	`str`	用于定时任务的助手 ID 或图名称。如果使用图名称，将默认使用从该图创建的第一个助手。	必填
`schedule`	`str`	执行此任务的 cron 调度表达式。	必填
`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]`	LangGraph API 调用完成后调用的 webhook。	`None`
`multitask_strategy`	`Optional[str]`	多任务策略，必须为 `'reject'`、`'interrupt'`、`'rollback'` 或 `'enqueue'`。	`None`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

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

示例：

client = get_client(url="http://localhost:2024")
cron_run = await 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_id`	`str`	要删除的定时任务 ID。	必填
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`None`	无返回值。

示例：

client = get_client(url="http://localhost:2024")
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]

描述：
获取定时任务列表。

参数：

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

返回值：

类型	描述
`list[Cron]`	搜索返回的定时任务列表。

示例：

client = get_client(url="http://localhost:2024")
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'
    }
]

1.7 类 `StoreClient`

描述：
与 LangGraph 中图的共享存储交互的异步客户端。
存储提供键值存储系统，用于在图执行之间持久化数据，支持有状态操作和线程间数据共享。

1.7.1 示例

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

1.7.2 方法

名称	描述
`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`（禁用）或要索引的字段路径列表。	`None`
`ttl`	`Optional[int]`	项目的生存时间（分钟），若为 `None` 则不过期。	`None`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`None`	无返回值。

示例：

client = get_client(url="http://localhost:2024")
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

描述：
检索单个项目。

参数：

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

返回值：

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

示例：

client = get_client(url="http://localhost:2024")
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

描述：
删除项目。

参数：

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

返回值：

类型	描述
`None`	无返回值。

示例：

client = get_client(url="http://localhost:2024")
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]]`	用于过滤结果的键值对字典。	`None`
`limit`	`int`	返回的最大项目数（默认 10）。	`10`
`offset`	`int`	返回结果前跳过的项目数（默认 0）。	`0`
`query`	`Optional[str]`	用于自然语言搜索的可选查询。	`None`
`refresh_ttl`	`Optional[bool]`	是否刷新搜索返回项目的 TTL。若为 `None`，使用存储的默认行为。	`None`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`SearchItemsResponse`	匹配搜索条件的项目列表。

示例：

client = get_client(url="http://localhost:2024")
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"
        }
    ]
}

方法 `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]]`	用于过滤命名空间的前缀字符串列表。	`None`
`suffix`	`Optional[list[str]]`	用于过滤命名空间的后缀字符串列表。	`None`
`max_depth`	`Optional[int]`	返回命名空间的最大深度。	`None`
`limit`	`int`	返回的最大命名空间数（默认 100）。	`100`
`offset`	`int`	返回结果前跳过的命名空间数（默认 0）。	`0`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`ListNamespaceResponse`	匹配条件的命名空间列表。

示例：

client = get_client(url="http://localhost:2024")
namespaces = await client.store.list_namespaces(
    prefix=["documents"],
    max_depth=3,
    limit=10,
    offset=0
)
print(namespaces)
[
    ["documents", "user123", "reports"],
    ["documents", "user456", "invoices"]
]

1.8 类 `SyncLangGraphClient`

描述：
与 LangGraph API 交互的同步客户端。
此类为管理助手、线程、运行、定时任务和数据存储提供同步访问 LangGraph API 端点的方法。

1.8.1 示例

client = get_sync_client(url="http://localhost:2024")
assistant = client.assistants.get("asst_123")

1.9 类 `SyncHttpClient`

描述：
处理 LangGraph API 的同步请求。
在底层的 httpx 客户端上提供错误消息和内容处理增强，接口与 HttpClient 类似但用于同步使用。

1.9.1 属性

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

1.9.2 方法

名称	描述
`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 请求。

参数：

名称	类型	描述	默认值
`path`	`str`	请求路径。	必填
`params`	`Optional[QueryParamTypes]`	查询参数。	`None`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`Any`	请求响应内容。

方法 `post`

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

描述：
发送 POST 请求。

参数：

名称	类型	描述	默认值
`path`	`str`	请求路径。	必填
`json`	`Optional[dict]`	JSON 请求体。	`None`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`Any`	请求响应内容。

方法 `put`

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

描述：
发送 PUT 请求。

参数：

名称	类型	描述	默认值
`path`	`str`	请求路径。	必填
`json`	`dict`	JSON 请求体。	必填
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`Any`	请求响应内容。

方法 `patch`

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

描述：
发送 PATCH 请求。

参数：

名称	类型	描述	默认值
`path`	`str`	请求路径。	必填
`json`	`dict`	JSON 请求体。	必填
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`Any`	请求响应内容。

方法 `delete`

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

描述：
发送 DELETE 请求。

参数：

名称	类型	描述	默认值
`path`	`str`	请求路径。	必填
`json`	`Optional[Any]`	JSON 请求体。	`None`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`None`	无返回值。

方法 `stream`

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

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

参数：

名称	类型	描述	默认值
`path`	`str`	请求路径。	必填
`method`	`str`	HTTP 方法（如 `GET`、`POST`）。	必填
`json`	`Optional[dict]`	JSON 请求体。	`None`
`params`	`Optional[QueryParamTypes]`	查询参数。	`None`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

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

1.10 类 `SyncAssistantsClient`

描述：
同步管理 LangGraph 中助手的客户端，助手是图的版本化配置。

1.10.1 示例

client = get_sync_client(url="http://localhost:2024")
assistant = client.assistants.get("assistant_id_123")

1.10.2 方法

名称	描述
`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]]`	可选的自定义请求头。	`None`

返回值：

名称	类型	描述
`Assistant`	`Assistant`	助手对象。

示例：

client = get_sync_client(url="http://localhost:2024")
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]]`	可选的自定义请求头。	`None`

返回值：

名称	类型	描述
`Graph`	`dict[str, list[dict[str, Any]]]`	助手的图信息（JSON 格式）。

示例：

client = get_sync_client(url="http://localhost:2024")
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]]`	可选的自定义请求头。	`None`

返回值：

名称	类型	描述
`GraphSchema`	`GraphSchema`	助手的图模式。

示例：

client = get_sync_client(url="http://localhost:2024")
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。	必填
`namespace`	`Optional[str]`	可选的命名空间过滤器。	`None`
`recurse`	`bool`	是否递归获取子图。	`False`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

名称	类型	描述
`Subgraphs`	`Subgraphs`	助手的子图模式。

方法 `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。通常在 `langgraph.json` 配置中设置。	必填
`config`	`Optional[Config]`	图的配置。	`None`
`metadata`	`Json`	添加到助手的元数据。	`None`
`assistant_id`	`Optional[str]`	助手 ID，如果未提供，将默认生成随机 UUID。	`None`
`if_exists`	`Optional[OnConflictBehavior]`	处理重复创建的方式。默认在底层为 `'raise'`，可为 `'raise'`（如果重复则抛出错误）或 `'do_nothing'`（返回现有助手）。	`None`
`name`	`Optional[str]`	助手名称，默认在底层为 `'Untitled'`。	`None`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`
`description`	`Optional[str]`	助手的可选描述。需 `langgraph-api` 服务器版本 >= 0.0.45。	`None`

返回值：

名称	类型	描述
`Assistant`	`Assistant`	创建的助手。

示例：

client = get_sync_client(url="http://localhost:2024")
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`	要更新的助手 ID。	必填
`graph_id`	`Optional[str]`	助手使用的图 ID。通常在 `langgraph.json` 配置中设置。如果为 `None`，助手保持指向原图。	`None`
`config`	`Optional[Config]`	图的配置。	`None`
`metadata`	`Json`	与现有助手元数据合并的元数据。	`None`
`name`	`Optional[str]`	助手的新名称。	`None`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`
`description`	`Optional[str]`	助手的可选描述。需 `langgraph-api` 服务器版本 >= 0.0.45。	`None`

返回值：

名称	类型	描述
`Assistant`	`Assistant`	更新后的助手。

示例：

client = get_sync_client(url="http://localhost:2024")
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]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`None`	无返回值。

示例：

client = get_sync_client(url="http://localhost:2024")
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`	用于过滤的元数据，每个键值对进行精确匹配。	`None`
`graph_id`	`Optional[str]`	用于过滤的图 ID。通常在 `langgraph.json` 配置中设置。	`None`
`limit`	`int`	返回的最大结果数。	`10`
`offset`	`int`	跳过的结果数。	`0`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`list[Assistant]`	匹配搜索参数的助手列表。

示例：

client = get_sync_client(url="http://localhost:2024")
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`	用于过滤版本的元数据，每个键值对进行精确匹配。	`None`
`limit`	`int`	返回的最大版本数。	`10`
`offset`	`int`	跳过的版本数。	`0`
`headers`	`Optional[dict[str, str]]`	可选的自定义请求头。	`None`

返回值：

类型	描述
`list[AssistantVersion]`	助手版本列表。

示例：

client = get_sync_client(url="http://localhost:2024")
assistant_versions = 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]]`	可选的自定义请求头。	`None`

返回值：

名称	类型	描述
`Assistant`	`Assistant`	助手对象。

示例：

client = get_sync_client(url="http://localhost:2024")
new_version_assistant = client.assistants.set_latest(
    assistant_id="my_assistant_id",
    version=3
)

2. 函数

2.1 函数 `get_client`

描述：
获取 LangGraphClient 实例。

用法：

client = get_client(url="http://localhost:2024")

2.2 函数 `get_sync_client`

描述：
获取同步 SyncLangGraphClient 实例。

用法：

client = get_sync_client(url="http://localhost:2024")

3. 总结

SDK 机制：LangGraph Python SDK 提供异步（LangGraphClient）和同步（SyncLangGraphClient）客户端，用于管理 LangGraph API 的核心资源（助手、线程、运行、定时任务和存储），支持 Python 和 JavaScript 项目，适合开发、测试和生产部署。
核心类：
- LangGraphClient / SyncLangGraphClient：顶级客户端，管理助手、线程、运行、定时任务和存储。
- HttpClient / SyncHttpClient：处理 HTTP 请求，支持 GET、POST、PUT、PATCH、DELETE 和 SSE 流。
- AssistantsClient / SyncAssistantsClient：管理图的版本化配置（助手）。
- ThreadsClient / SyncThreadsClient：管理多轮交互的状态（线程）。
- RunsClient / SyncRunsClient：管理图的单次调用（运行），支持有状态和无状态运行。
- CronClient / SyncCronClient：管理定时任务（需检查许可证支持）。
- StoreClient / SyncStoreClient：管理键值存储，支持持久化数据和语义搜索。
使用建议：
- 使用异步客户端（LangGraphClient）以获得更好的性能，特别是在高并发场景。
- 使用同步客户端（SyncLangGraphClient）简化开发，适合简单的脚本或低并发场景。
- 配置 langgraph.json 以定义图 ID 和默认助手，简化 assistant_id 的使用。
- 为存储配置 TTL 和语义搜索以优化数据管理和查询。
- 使用 stream 和 join_stream 方法实现实时交互，适合聊天或其他动态应用。
- 确保检查定时任务功能的许可证要求。
注意事项：
- 定时任务功能可能受许可证限制，需参考最新文档。
- 已弃用参数（如 checkpoint_id）应替换为推荐参数（如 checkpoint）。
- 确保 API 服务器运行在指定 URL（如 http://localhost:2024）并可访问。

LangGraph Python SDK 为开发者提供了强大的工具来构建和管理复杂的图驱动应用，适合从快速原型设计到生产环境的各种场景。

参考资料：

https://langchain-ai.github.io/langgraph/cloud/reference/sdk/python_sdk_ref/

火山引擎开发者社区

更多推荐

Dify 知识库构建实战指南

火山引擎开发者社区

论文AIGC痕迹太重？试试2个免费降AI率工具，还有免费ai查重！

火山引擎开发者社区

脉脉独家【AI创作者xAMA】｜当豆包手机遭遇“全网封杀”：AI学会操作手机，我们的饭碗还保得住吗？

火山引擎开发者社区

所有评论(0)

查看更多评论

正义的彬彬侠

@u013172930

已为社区贡献128条内容

【LangGraph】Python SDK：连接 LangGraph API 的客户端实现

正义的彬彬侠

LangGraph Python SDK 参考

1. 类

1.1 类 LangGraphClient

1.1.1 属性

1.2 类 HttpClient

1.2.1 属性

1.2.2 方法

方法 get

方法 post

方法 put

方法 patch

方法 delete

方法 stream

1.3 类 AssistantsClient

1.3.1 示例

1.3.2 方法

方法 get

方法 get_graph

方法 get_schemas

方法 get_subgraphs

方法 create

方法 update

方法 delete

方法 search

方法 get_versions

方法 set_latest

1.4 类 ThreadsClient

1.4.1 示例

1.4.2 方法

方法 get

方法 create

方法 update

方法 delete

方法 search

方法 copy

方法 get_state

方法 update_state

方法 get_history

1.5 类 RunsClient

1.5.1 示例

1.5.2 方法

方法 stream

方法 create

方法 create_batch

方法 wait

方法 list

方法 get

方法 cancel

方法 join

方法 join_stream

方法 delete

1.6 类 CronClient

1.6.1 示例

1.6.2 功能可用性

1.6.3 方法

方法 create_for_thread

方法 create

方法 delete

方法 search

1.7 类 StoreClient

1.7.1 示例

1.7.2 方法

方法 put_item

方法 get_item

方法 delete_item

方法 search_items

方法 list_namespaces

1.8 类 SyncLangGraphClient

1.8.1 示例

1.9 类 SyncHttpClient

1.9.1 属性

1.9.2 方法

方法 get

方法 post

方法 put

方法 patch

方法 delete

方法 stream

1.1 类 `LangGraphClient`

1.2 类 `HttpClient`

方法 `get`

方法 `post`

方法 `put`

方法 `patch`

方法 `delete`

方法 `stream`

1.3 类 `AssistantsClient`

方法 `get`

方法 `get_graph`

方法 `get_schemas`

方法 `get_subgraphs`

方法 `create`

方法 `update`

方法 `delete`

方法 `search`

方法 `get_versions`

方法 `set_latest`

1.4 类 `ThreadsClient`

方法 `get`

方法 `create`

方法 `update`

方法 `delete`

方法 `search`

方法 `copy`

方法 `get_state`

方法 `update_state`

方法 `get_history`

1.5 类 `RunsClient`

方法 `stream`

方法 `create`

方法 `create_batch`

方法 `wait`

方法 `list`

方法 `get`

方法 `cancel`

方法 `join`

方法 `join_stream`

方法 `delete`

1.6 类 `CronClient`

方法 `create_for_thread`

方法 `create`

方法 `delete`

方法 `search`

1.7 类 `StoreClient`

方法 `put_item`

方法 `get_item`

方法 `delete_item`

方法 `search_items`

方法 `list_namespaces`

1.8 类 `SyncLangGraphClient`

1.9 类 `SyncHttpClient`

方法 `get`

方法 `post`

方法 `put`

方法 `patch`

方法 `delete`

方法 `stream`

1.10 类 `SyncAssistantsClient`

方法 `get`

方法 `get_graph`

方法 `get_schemas`

方法 `get_subgraphs`

方法 `create`

方法 `update`

方法 `delete`

方法 `search`

方法 `get_versions`

方法 `set_latest`

2.1 函数 `get_client`

2.2 函数 `get_sync_client`