AI创想

标题: 【LangGraph】Python SDK：连接 LangGraph API 的客户端实现 [打印本页]

---

作者: 创想小编    时间: 11 小时前
标题: 【LangGraph】Python SDK：连接 LangGraph API 的客户端实现
作者：彬彬侠
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

1. async get(
2.     path:str,*,
3.     params: Optional[QueryParamTypes]=None,
4.     headers: Optional[dict[str,str]]=None)-> Any

复制代码

描述：
发送 GET 请求。
参数：

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

返回值：

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

方法 post

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

复制代码

描述：
发送 POST 请求。
参数：

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

返回值：

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

方法 put

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

复制代码

描述：
发送 PUT 请求。
参数：

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

返回值：

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

方法 patch

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

复制代码

描述：
发送 PATCH 请求。
参数：

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

返回值：

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

方法 delete

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

复制代码

描述：
发送 DELETE 请求。
参数：

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

返回值：

类型	描述
None	无返回值。

方法 stream

1. async stream(
2.     path:str,
3.     method:str,*,
4.     json: Optional[dict]=None,
5.     params: Optional[QueryParamTypes]=None,
6.     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 示例

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

1. async get(
2.     assistant_id:str,*,
3.     headers: Optional[dict[str,str]]=None)-> Assistant

复制代码

描述：
按 ID 获取助手。
参数：

名称	类型	描述	默认值
assistant_id	str	要获取的助手 ID 或图名称（使用默认助手）。	必填
headers	Optional[dict[str, str]]	可选的自定义请求头。	None

返回值：

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

示例：

1. assistant =await client.assistants.get(
2.     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

1. async get_graph(
2.     assistant_id:str,*,
3.     xray: Union[int,bool]=False,
4.     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 格式）。

示例：

1. client = get_client(url="http://localhost:2024")
2. graph_info =await client.assistants.get_graph(
3.     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

1. async get_schemas(
2.     assistant_id:str,*,
3.     headers: Optional[dict[str,str]]=None)-> GraphSchema

复制代码

描述：
按 ID 获取助手的模式。
参数：

名称	类型	描述	默认值
assistant_id	str	要获取模式的助手 ID。	必填
headers	Optional[dict[str, str]]	可选的自定义请求头。	None

返回值：

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

示例：

1. client = get_client(url="http://localhost:2024")
2. schema =await client.assistants.get_schemas(
3.     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

1. async get_subgraphs(
2.     assistant_id:str,
3.     namespace: Optional[str]=None,
4.     recurse:bool=False,*,
5.     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

1. async create(
2.     graph_id: Optional[str],
3.     config: Optional[Config]=None,*,
4.     metadata: Json =None,
5.     assistant_id: Optional[str]=None,
6.     if_exists: Optional[OnConflictBehavior]=None,
7.     name: Optional[str]=None,
8.     headers: Optional[dict[str,str]]=None,
9.     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	创建的助手。

示例：

1. client = get_client(url="http://localhost:2024")
2. assistant =await client.assistants.create(
3.     graph_id="agent",
4.     config={"configurable":{"model_name":"openai"}},
5.     metadata={"number":1},
6.     assistant_id="my-assistant-id",
7.     if_exists="do_nothing",
8.     name="my_name")

复制代码

方法 update

1. async update(
2.     assistant_id:str,*,
3.     graph_id: Optional[str]=None,
4.     config: Optional[Config]=None,
5.     metadata: Json =None,
6.     name: Optional[str]=None,
7.     headers: Optional[dict[str,str]]=None,
8.     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	更新后的助手。

示例：

1. client = get_client(url="http://localhost:2024")
2. assistant =await client.assistants.update(
3.     assistant_id='e280dad7-8618-443f-87f1-8e41841c180f',
4.     graph_id="other-graph",
5.     config={"configurable":{"model_name":"anthropic"}},
6.     metadata={"number":2})

复制代码

方法 delete

1. async delete(
2.     assistant_id:str,*,
3.     headers: Optional[dict[str,str]]=None)->None

复制代码

描述：
删除助手。
参数：

名称	类型	描述	默认值
assistant_id	str	要删除的助手 ID。	必填
headers	Optional[dict[str, str]]	可选的自定义请求头。	None

返回值：

类型	描述
None	无返回值。

示例：

1. client = get_client(url="http://localhost:2024")await client.assistants.delete(
2.     assistant_id="my_assistant_id")

复制代码

方法 search

1. async search(*,
2.     metadata: Json =None,
3.     graph_id: Optional[str]=None,
4.     limit:int=10,
5.     offset:int=0,
6.     sort_by: Optional[AssistantSortBy]=None,
7.     sort_order: Optional[SortOrder]=None,
8.     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]	匹配搜索参数的助手列表。

示例：

1. client = get_client(url="http://localhost:2024")
2. assistants =await client.assistants.search(
3.     metadata={"name":"my_name"},
4.     graph_id="my_graph_id",
5.     limit=5,
6.     offset=5)

复制代码

方法 get_versions

1. async get_versions(
2.     assistant_id:str,
3.     metadata: Json =None,
4.     limit:int=10,
5.     offset:int=0,*,
6.     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]	助手版本列表。

示例：

1. client = get_client(url="http://localhost:2024")
2. assistant_versions =await client.assistants.get_versions(
3.     assistant_id="my_assistant_id")

复制代码

方法 set_latest

1. async set_latest(
2.     assistant_id:str,
3.     version:int,*,
4.     headers: Optional[dict[str,str]]=None)-> Assistant

复制代码

描述：
更改助手的版本。
参数：

名称	类型	描述	默认值
assistant_id	str	要更改版本的助手 ID。	必填
version	int	要更改到的版本。	必填
headers	Optional[dict[str, str]]	可选的自定义请求头。	None

返回值：

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

示例：

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

复制代码

1.4 类 ThreadsClient

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

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

1. async get(
2.     thread_id:str,*,
3.     headers: Optional[dict[str,str]]=None)-> Thread

复制代码

描述：
按 ID 获取线程。
参数：

名称	类型	描述	默认值
thread_id	str	要获取的线程 ID。	必填
headers	Optional[dict[str, str]]	可选的自定义请求头。	None

返回值：

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

示例：

1. client = get_client(url="http://localhost:2024")
2. thread =await client.threads.get(
3.     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

1. async create(*,
2.     metadata: Json =None,
3.     thread_id: Optional[str]=None,
4.     if_exists: Optional[OnConflictBehavior]=None,
5.     supersteps: Optional[Sequence[dict[str, Sequence[dict[str, Any]]]]]=None,
6.     graph_id: Optional[str]=None,
7.     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	创建的线程。

示例：

1. client = get_client(url="http://localhost:2024")
2. thread =await client.threads.create(
3.     metadata={"number":1},
4.     thread_id="my-thread-id",
5.     if_exists="raise")

复制代码

方法 update

1. async update(
2.     thread_id:str,*,
3.     metadata:dict[str, Any],
4.     headers: Optional[dict[str,str]]=None)-> Thread

复制代码

描述：
更新线程。
参数：

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

返回值：

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

示例：

1. client = get_client(url="http://localhost:2024")
2. thread =await client.threads.update(
3.     thread_id="my-thread-id",
4.     metadata={"number":1})

复制代码

方法 delete

1. async delete(
2.     thread_id:str,*,
3.     headers: Optional[dict[str,str]]=None)->None

复制代码

描述：
删除线程。
参数：

名称	类型	描述	默认值
thread_id	str	要删除的线程 ID。	必填
headers	Optional[dict[str, str]]	可选的自定义请求头。	None

返回值：

类型	描述
None	无返回值。

示例：

1. client = get_client(url="http://localhost:2024")await client.threads.delete(
2.     thread_id="my_thread_id")

复制代码

方法 search

1. async search(*,
2.     metadata: Json =None,
3.     values: Json =None,
4.     status: Optional[ThreadStatus]=None,
5.     limit:int=10,
6.     offset:int=0,
7.     sort_by: Optional[ThreadSortBy]=None,
8.     sort_order: Optional[SortOrder]=None,
9.     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]	匹配搜索参数的线程列表。

示例：

1. client = get_client(url="http://localhost:2024")
2. threads =await client.threads.search(
3.     metadata={"number":1},
4.     status="interrupted",
5.     limit=15,
6.     offset=5)

复制代码

方法 copy

1. async copy(
2.     thread_id:str,*,
3.     headers: Optional[dict[str,str]]=None)->None

复制代码

描述：
复制线程。
参数：

名称	类型	描述	默认值
thread_id	str	要复制的线程 ID。	必填
headers	Optional[dict[str, str]]	可选的自定义请求头。	None

返回值：

类型	描述
None	无返回值。

示例：

1. client = get_client(url="http://localhost:2024")await client.threads.copy(
2.     thread_id="my_thread_id")

复制代码

方法 get_state

1. async get_state(
2.     thread_id:str,
3.     checkpoint: Optional[Checkpoint]=None,
4.     checkpoint_id: Optional[str]=None,*,
5.     subgraphs:bool=False,
6.     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	线程状态。

示例：

1. client = get_client(url="http://localhost:2024")
2. thread_state =await client.threads.get_state(
3.     thread_id="my_thread_id",
4.     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

1. async update_state(
2.     thread_id:str,
3.     values: Optional[Union[dict, Sequence[dict]]],*,
4.     as_node: Optional[str]=None,
5.     checkpoint: Optional[Checkpoint]=None,
6.     checkpoint_id: Optional[str]=None,
7.     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	更新线程状态后的响应。

示例：

1. client = get_client(url="http://localhost:2024")
2. response =await client.threads.update_state(
3.     thread_id="my_thread_id",
4.     values={"messages":[{"role":"user","content":"hello!"}]},
5.     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

1. async get_history(
2.     thread_id:str,*,
3.     limit:int=10,
4.     before: Optional[str| Checkpoint]=None,
5.     metadata: Optional[dict]=None,
6.     checkpoint: Optional[Checkpoint]=None,
7.     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]	线程的状态历史列表。

示例：

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

复制代码

1.5 类 RunsClient

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

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

1. async stream(
2.     thread_id: Optional[str],
3.     assistant_id:str,*,input: Optional[dict]=None,
4.     command: Optional[Command]=None,
5.     stream_mode: Union[StreamMode, Sequence[StreamMode]]="values",
6.     stream_subgraphs:bool=False,
7.     metadata: Optional[dict]=None,
8.     config: Optional[Config]=None,
9.     checkpoint: Optional[Checkpoint]=None,
10.     checkpoint_id: Optional[str]=None,
11.     checkpoint_during: Optional[bool]=None,
12.     interrupt_before: Optional[Union[All, Sequence[str]]]=None,
13.     interrupt_after: Optional[Union[All, Sequence[str]]]=None,
14.     feedback_keys: Optional[Sequence[str]]=None,
15.     on_disconnect: Optional[DisconnectMode]=None,
16.     on_completion: Optional[OnCompletionBehavior]=None,
17.     webhook: Optional[str]=None,
18.     multitask_strategy: Optional[MultitaskStrategy]=None,
19.     if_not_exists: Optional[IfNotExists]=None,
20.     after_seconds: Optional[int]=None,
21.     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]	流式传输结果的异步迭代器。

示例：

1. client = get_client(url="http://localhost:2024")asyncfor chunk in client.runs.stream(
2.     thread_id=None,
3.     assistant_id="agent",input={"messages":[{"role":"user","content":"how are you?"}]},
4.     stream_mode=["values","debug"],
5.     metadata={"name":"my_run"},
6.     config={"configurable":{"model_name":"anthropic"}},
7.     interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"],
8.     interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"],
9.     feedback_keys=["my_feedback_key_1","my_feedback_key_2"],
10.     webhook="https://my.fake.webhook.com",
11.     multitask_strategy="interrupt"):print(chunk)# 输出：
12. StreamPart(event='metadata', data={'run_id':'1ef4a9b8-d7da-679a-a45a-872054341df2'})
13. 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}]})
14. 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}]})
15. StreamPart(event='end', data=None)

复制代码

方法 create

1. async create(
2.     thread_id: Optional[str],
3.     assistant_id:str,*,input: Optional[dict]=None,
4.     command: Optional[Command]=None,
5.     stream_mode: Union[StreamMode, Sequence[StreamMode]]="values",
6.     stream_subgraphs:bool=False,
7.     metadata: Optional[dict]=None,
8.     config: Optional[Config]=None,
9.     checkpoint: Optional[Checkpoint]=None,
10.     checkpoint_id: Optional[str]=None,
11.     checkpoint_during: Optional[bool]=None,
12.     interrupt_before: Optional[Union[All, Sequence[str]]]=None,
13.     interrupt_after: Optional[Union[All, Sequence[str]]]=None,
14.     webhook: Optional[str]=None,
15.     multitask_strategy: Optional[MultitaskStrategy]=None,
16.     if_not_exists: Optional[IfNotExists]=None,
17.     on_completion: Optional[OnCompletionBehavior]=None,
18.     after_seconds: Optional[int]=None,
19.     headers: Optional[dict[str,str]]=None)-> Run

复制代码

描述：
创建后台运行。
参数：
与 stream 方法类似，省略 feedback_keys，添加以下参数：

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

返回值：

名称	类型	描述
Run	Run	创建的运行对象。

示例：

1. background_run =await client.runs.create(
2.     thread_id="my_thread_id",
3.     assistant_id="my_assistant_id",input={"messages":[{"role":"user","content":"hello!"}]},
4.     metadata={"name":"my_run"},
5.     config={"configurable":{"model_name":"openai"}},
6.     interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"],
7.     interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"],
8.     webhook="https://my.fake.webhook.com",
9.     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

1. async create_batch(
2.     payloads:list[RunCreate])->list[Run]

复制代码

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

名称	类型	描述	默认值
payloads	list[RunCreate]	运行创建参数的列表。	必填

返回值：

类型	描述
list[Run]	创建的运行对象列表。

方法 wait

1. async wait(
2.     thread_id: Optional[str],
3.     assistant_id:str,*,input: Optional[dict]=None,
4.     command: Optional[Command]=None,
5.     metadata: Optional[dict]=None,
6.     config: Optional[Config]=None,
7.     checkpoint: Optional[Checkpoint]=None,
8.     checkpoint_id: Optional[str]=None,
9.     checkpoint_during: Optional[bool]=None,
10.     interrupt_before: Optional[Union[All, Sequence[str]]]=None,
11.     interrupt_after: Optional[Union[All, Sequence[str]]]=None,
12.     webhook: Optional[str]=None,
13.     on_disconnect: Optional[DisconnectMode]=None,
14.     on_completion: Optional[OnCompletionBehavior]=None,
15.     multitask_strategy: Optional[MultitaskStrategy]=None,
16.     if_not_exists: Optional[IfNotExists]=None,
17.     after_seconds: Optional[int]=None,
18.     raise_error:bool=True,
19.     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]]	运行的输出。

示例：

1. client = get_client(url="http://localhost:2024")
2. final_state_of_run =await client.runs.wait(
3.     thread_id=None,
4.     assistant_id="agent",input={"messages":[{"role":"user","content":"how are you?"}]},
5.     metadata={"name":"my_run"},
6.     config={"configurable":{"model_name":"anthropic"}},
7.     interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"],
8.     interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"],
9.     webhook="https://my.fake.webhook.com",
10.     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

1. asynclist(
2.     thread_id:str,*,
3.     limit:int=10,
4.     offset:int=0,
5.     status: Optional[RunStatus]=None,
6.     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]	线程的运行列表。

示例：

1. client = get_client(url="http://localhost:2024")await client.runs.list(
2.     thread_id="thread_id",
3.     limit=5,
4.     offset=5)

复制代码

方法 get

1. async get(
2.     thread_id:str,
3.     run_id:str,*,
4.     headers: Optional[dict[str,str]]=None)-> Run

复制代码

描述：
获取运行。
参数：

名称	类型	描述	默认值
thread_id	str	要获取的线程 ID。	必填
run_id	str	要获取的运行 ID。	必填
headers	Optional[dict[str, str]]	可选的自定义请求头。	None

返回值：

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

示例：

1. client = get_client(url="http://localhost:2024")
2. run =await client.runs.get(
3.     thread_id="thread_id_to_delete",
4.     run_id="run_id_to_delete")

复制代码

方法 cancel

1. async cancel(
2.     thread_id:str,
3.     run_id:str,*,
4.     wait:bool=False,
5.     action: CancelAction ="interrupt",
6.     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	无返回值。

示例：

1. client = get_client(url="http://localhost:2024")await client.runs.cancel(
2.     thread_id="thread_id_to_cancel",
3.     run_id="run_id_to_cancel",
4.     wait=True,
5.     action="interrupt")

复制代码

方法 join

1. async join(
2.     thread_id:str,
3.     run_id:str,*,
4.     headers: Optional[dict[str,str]]=None)->dict

复制代码

描述：
阻塞直到运行完成，返回线程的最终状态。
参数：

名称	类型	描述	默认值
thread_id	str	要加入的线程 ID。	必填
run_id	str	要加入的运行 ID。	必填
headers	Optional[dict[str, str]]	可选的自定义请求头。	None

返回值：

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

示例：

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

复制代码

方法 join_stream

1. async join_stream(
2.     thread_id:str,
3.     run_id:str,*,
4.     cancel_on_disconnect:bool=False,
5.     stream_mode: Optional[Union[StreamMode, Sequence[StreamMode]]]=None,
6.     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]	流式传输结果的异步迭代器。

示例：

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

复制代码

方法 delete

1. async delete(
2.     thread_id:str,
3.     run_id:str,*,
4.     headers: Optional[dict[str,str]]=None)->None

复制代码

描述：
删除运行。
参数：

名称	类型	描述	默认值
thread_id	str	要删除的线程 ID。	必填
run_id	str	要删除的运行 ID。	必填
headers	Optional[dict[str, str]]	可选的自定义请求头。	None

返回值：

类型	描述
None	无返回值。

示例：

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

复制代码

1.6 类 CronClient

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

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

复制代码

1.6.2 功能可用性

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

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

方法 create_for_thread

1. async create_for_thread(
2.     thread_id:str,
3.     assistant_id:str,*,
4.     schedule:str,input: Optional[dict]=None,
5.     metadata: Optional[dict]=None,
6.     config: Optional[Config]=None,
7.     checkpoint_during: Optional[bool]=None,
8.     interrupt_before: Optional[Union[All,list[str]]]=None,
9.     interrupt_after: Optional[Union[All,list[str]]]=None,
10.     webhook: Optional[str]=None,
11.     multitask_strategy: Optional[str]=None,
12.     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	定时运行对象。

示例：

1. client = get_client(url="http://localhost:2024")
2. cron_run =await client.crons.create_for_thread(
3.     thread_id="my-thread-id",
4.     assistant_id="agent",
5.     schedule="27 15 * * *",input={"messages":[{"role":"user","content":"hello!"}]},
6.     metadata={"name":"my_run"},
7.     config={"configurable":{"model_name":"openai"}},
8.     interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"],
9.     interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"],
10.     webhook="https://my.fake.webhook.com",
11.     multitask_strategy="interrupt")

复制代码

方法 create

1. async create(
2.     assistant_id:str,*,
3.     schedule:str,input: Optional[dict]=None,
4.     metadata: Optional[dict]=None,
5.     config: Optional[Config]=None,
6.     checkpoint_during: Optional[bool]=None,
7.     interrupt_before: Optional[Union[All,list[str]]]=None,
8.     interrupt_after: Optional[Union[All,list[str]]]=None,
9.     webhook: Optional[str]=None,
10.     multitask_strategy: Optional[str]=None,
11.     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	定时运行对象。

示例：

1. client = get_client(url="http://localhost:2024")
2. cron_run =await client.crons.create(
3.     assistant_id="agent",
4.     schedule="27 15 * * *",input={"messages":[{"role":"user","content":"hello!"}]},
5.     metadata={"name":"my_run"},
6.     config={"configurable":{"model_name":"openai"}},
7.     interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"],
8.     interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"],
9.     webhook="https://my.fake.webhook.com",
10.     multitask_strategy="interrupt")

复制代码

方法 delete

1. async delete(
2.     cron_id:str,
3.     headers: Optional[dict[str,str]]=None)->None

复制代码

描述：
删除定时任务。
参数：

名称	类型	描述	默认值
cron_id	str	要删除的定时任务 ID。	必填
headers	Optional[dict[str, str]]	可选的自定义请求头。	None

返回值：

类型	描述
None	无返回值。

示例：

1. client = get_client(url="http://localhost:2024")await client.crons.delete(
2.     cron_id="cron_to_delete")

复制代码

方法 search

1. async search(*,
2.     assistant_id: Optional[str]=None,
3.     thread_id: Optional[str]=None,
4.     limit:int=10,
5.     offset:int=0,
6.     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]	搜索返回的定时任务列表。

示例：

1. client = get_client(url="http://localhost:2024")
2. cron_jobs =await client.crons.search(
3.     assistant_id="my_assistant_id",
4.     thread_id="my_thread_id",
5.     limit=5,
6.     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 示例

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

1. async put_item(
2.     namespace: Sequence[str],/,
3.     key:str,
4.     value:dict[str, Any],
5.     index: Optional[Union[Literal[False],list[str]]]=None,
6.     ttl: Optional[int]=None,
7.     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	无返回值。

示例：

1. client = get_client(url="http://localhost:2024")await client.store.put_item(["documents","user123"],
2.     key="item456",
3.     value={"title":"My Document","content":"Hello World"})

复制代码

方法 get_item

1. async get_item(
2.     namespace: Sequence[str],/,
3.     key:str,*,
4.     refresh_ttl: Optional[bool]=None,
5.     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	检索到的项目。

示例：

1. client = get_client(url="http://localhost:2024")
2. item =await client.store.get_item(["documents","user123"],
3.     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

1. async delete_item(
2.     namespace: Sequence[str],/,
3.     key:str,
4.     headers: Optional[dict[str,str]]=None)->None

复制代码

描述：
删除项目。
参数：

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

返回值：

类型	描述
None	无返回值。

示例：

1. client = get_client(url="http://localhost:2024")await client.store.delete_item(["documents","user123"],
2.     key="item456")

复制代码

方法 search_items

1. async search_items(
2.     namespace_prefix: Sequence[str],/,filter: Optional[dict[str, Any]]=None,
3.     limit:int=10,
4.     offset:int=0,
5.     query: Optional[str]=None,
6.     refresh_ttl: Optional[bool]=None,
7.     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	匹配搜索条件的项目列表。

示例：

1. client = get_client(url="http://localhost:2024")
2. items =await client.store.search_items(["documents"],filter={"author":"John Doe"},
3.     limit=5,
4.     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

1. async list_namespaces(
2.     prefix: Optional[list[str]]=None,
3.     suffix: Optional[list[str]]=None,
4.     max_depth: Optional[int]=None,
5.     limit:int=100,
6.     offset:int=0,
7.     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	匹配条件的命名空间列表。

示例：

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

复制代码

1.8 类 SyncLangGraphClient

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

1. client = get_sync_client(url="http://localhost:2024")
2. 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

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

复制代码

描述：
发送 GET 请求。
参数：

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

返回值：

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

方法 post

1. post(
2.     path:str,*,
3.     json: Optional[dict],
4.     headers: Optional[dict[str,str]]=None)-> Any

复制代码

描述：
发送 POST 请求。
参数：

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

返回值：

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

方法 put

1. put(
2.     path:str,*,
3.     json:dict,
4.     headers: Optional[dict[str,str]]=None)-> Any

复制代码

描述：
发送 PUT 请求。
参数：

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

返回值：

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

方法 patch

1. patch(
2.     path:str,*,
3.     json:dict,
4.     headers: Optional[dict[str,str]]=None)-> Any

复制代码

描述：
发送 PATCH 请求。
参数：

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

返回值：

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

方法 delete

1. delete(
2.     path:str,*,
3.     json: Optional[Any]=None,
4.     headers: Optional[dict[str,str]]=None)->None

复制代码

描述：
发送 DELETE 请求。
参数：

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

返回值：

类型	描述
None	无返回值。

方法 stream

1. stream(
2.     path:str,
3.     method:str,*,
4.     json: Optional[dict]=None,
5.     params: Optional[QueryParamTypes]=None,
6.     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 示例

1. client = get_sync_client(url="http://localhost:2024")
2. 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

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

复制代码

描述：
按 ID 获取助手。
参数：

名称	类型	描述	默认值
assistant_id	str	要获取的助手 ID 或图名称（使用默认助手）。	必填
headers	Optional[dict[str, str]]	可选的自定义请求头。	None

返回值：

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

示例：

1. client = get_sync_client(url="http://localhost:2024")
2. assistant = client.assistants.get(
3.     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

1. get_graph(
2.     assistant_id:str,*,
3.     xray: Union[int,bool]=False,
4.     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 格式）。

示例：

1. client = get_sync_client(url="http://localhost:2024")
2. graph_info = client.assistants.get_graph(
3.     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

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

复制代码

描述：
按 ID 获取助手的模式。
参数：

名称	类型	描述	默认值
assistant_id	str	要获取模式的助手 ID。	必填
headers	Optional[dict[str, str]]	可选的自定义请求头。	None

返回值：

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

示例：

1. client = get_sync_client(url="http://localhost:2024")
2. schema = client.assistants.get_schemas(
3.     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

1. get_subgraphs(
2.     assistant_id:str,
3.     namespace: Optional[str]=None,
4.     recurse:bool=False,*,
5.     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

1. create(
2.     graph_id: Optional[str],
3.     config: Optional[Config]=None,*,
4.     metadata: Json =None,
5.     assistant_id: Optional[str]=None,
6.     if_exists: Optional[OnConflictBehavior]=None,
7.     name: Optional[str]=None,
8.     headers: Optional[dict[str,str]]=None,
9.     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	创建的助手。

示例：

1. client = get_sync_client(url="http://localhost:2024")
2. assistant = client.assistants.create(
3.     graph_id="agent",
4.     config={"configurable":{"model_name":"openai"}},
5.     metadata={"number":1},
6.     assistant_id="my-assistant-id",
7.     if_exists="do_nothing",
8.     name="my_name")

复制代码

方法 update

1. update(
2.     assistant_id:str,*,
3.     graph_id: Optional[str]=None,
4.     config: Optional[Config]=None,
5.     metadata: Json =None,
6.     name: Optional[str]=None,
7.     headers: Optional[dict[str,str]]=None,
8.     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	更新后的助手。

示例：

1. client = get_sync_client(url="http://localhost:2024")
2. assistant = client.assistants.update(
3.     assistant_id='e280dad7-8618-443f-87f1-8e41841c180f',
4.     graph_id="other-graph",
5.     config={"configurable":{"model_name":"anthropic"}},
6.     metadata={"number":2})

复制代码

方法 delete

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

复制代码

描述：
删除助手。
参数：

名称	类型	描述	默认值
assistant_id	str	要删除的助手 ID。	必填
headers	Optional[dict[str, str]]	可选的自定义请求头。	None

返回值：

类型	描述
None	无返回值。

示例：

1. client = get_sync_client(url="http://localhost:2024")
2. client.assistants.delete(
3.     assistant_id="my_assistant_id")

复制代码

方法 search

1. search(*,
2.     metadata: Json =None,
3.     graph_id: Optional[str]=None,
4.     limit:int=10,
5.     offset:int=0,
6.     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]	匹配搜索参数的助手列表。

示例：

1. client = get_sync_client(url="http://localhost:2024")
2. assistants = client.assistants.search(
3.     metadata={"name":"my_name"},
4.     graph_id="my_graph_id",
5.     limit=5,
6.     offset=5)

复制代码

方法 get_versions

1. get_versions(
2.     assistant_id:str,
3.     metadata: Json =None,
4.     limit:int=10,
5.     offset:int=0,*,
6.     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]	助手版本列表。

示例：

1. client = get_sync_client(url="http://localhost:2024")
2. assistant_versions = client.assistants.get_versions(
3.     assistant_id="my_assistant_id")

复制代码

方法 set_latest

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

复制代码

描述：
更改助手的版本。
参数：

名称	类型	描述	默认值
assistant_id	str	要更改版本的助手 ID。	必填
version	int	要更改到的版本。	必填
headers	Optional[dict[str, str]]	可选的自定义请求头。	None

返回值：

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

示例：

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

复制代码

2. 函数

2.1 函数 get_client

描述：
获取 LangGraphClient 实例。
用法：

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

复制代码

2.2 函数 get_sync_client

描述：
获取同步 SyncLangGraphClient 实例。
用法：

1. 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/

原文地址：https://blog.csdn.net/u013172930/article/details/148124458

---

欢迎光临 AI创想 (https://llms-ai.com/)

Powered by Discuz! X3.4