登陆 注册
AI创想 › 综合版块 › LangGraph智能体 › 【LangGraph】Python SDK：连接 LangGraph API 的客户端 ...
25位大厂高管转战生成式AI创业！吸金猛，不

[hr] 智东西 作者 李水青 编辑 心缘 智东西7月2日报道，今年3月智东西曾报道，超14位互联网大厂离职高管 详细

返回列表 发新帖
开启左侧

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

[复制链接]
创想小编 发表于 11 小时前 | 显示全部楼层 |阅读模式 打印 上一主题 下一主题
作者：彬彬侠
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 属性

名称类型描述
assistantsAssistantsClient管理图的版本化配置。
threadsThreadsClient处理（可能为）多轮交互，例如会话线程。
runsRunsClient控制图的单次调用。
cronsCronClient管理定时操作。
storeStoreClient与持久化、共享数据存储交互。
1.2 类 HttpClient

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

名称类型描述
clientAsyncClient底层的 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 请求。
参数：
名称类型描述默认值
pathstr请求路径。必填
paramsOptional[QueryParamTypes]查询参数。None
headersOptional[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 请求。
参数：
名称类型描述默认值
pathstr请求路径。必填
jsonOptional[dict]JSON 请求体。None
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
类型描述
Any请求响应内容。
方法 put
  1. async put(
  2.     path:str,*,
  3.     json:dict,
  4.     headers: Optional[dict[str,str]]=None)-> Any
复制代码
描述：
发送 PUT 请求。
参数：
名称类型描述默认值
pathstr请求路径。必填
jsondictJSON 请求体。必填
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
类型描述
Any请求响应内容。
方法 patch
  1. async patch(
  2.     path:str,*,
  3.     json:dict,
  4.     headers: Optional[dict[str,str]]=None)-> Any
复制代码
描述：
发送 PATCH 请求。
参数：
名称类型描述默认值
pathstr请求路径。必填
jsondictJSON 请求体。必填
headersOptional[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 请求。
参数：
名称类型描述默认值
pathstr请求路径。必填
jsonOptional[Any]JSON 请求体。None
headersOptional[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）流式传输结果。
参数：
名称类型描述默认值
pathstr请求路径。必填
methodstrHTTP 方法（如 GET、POST）。必填
jsonOptional[dict]JSON 请求体。None
paramsOptional[QueryParamTypes]查询参数。None
headersOptional[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_idstr要获取的助手 ID 或图名称（使用默认助手）。必填
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
名称类型描述
AssistantAssistant助手对象。
示例：
  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_idstr要获取图的助手 ID。必填
xrayUnion[int, bool]是否包含子图表示。如果提供整数，仅包含深度小于或等于该值的子图。False
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
名称类型描述
Graphdict[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_idstr要获取模式的助手 ID。必填
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
名称类型描述
GraphSchemaGraphSchema助手的图模式。
示例：
  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_idstr要获取子图模式的助手 ID。必填
namespaceOptional[str]可选的命名空间过滤器。None
recursebool是否递归获取子图。False
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
名称类型描述
SubgraphsSubgraphs助手的子图模式。
方法 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_idOptional[str]助手使用的图 ID。通常在 langgraph.json 配置中设置。必填
configOptional[Config]图的配置。None
metadataJson添加到助手的元数据。None
assistant_idOptional[str]助手 ID，如果未提供，将默认生成随机 UUID。None
if_existsOptional[OnConflictBehavior]处理重复创建的方式。默认在底层为 'raise'，可为 'raise'（如果重复则抛出错误）或 'do_nothing'（返回现有助手）。None
nameOptional[str]助手名称，默认在底层为 'Untitled'。None
headersOptional[dict[str, str]]可选的自定义请求头。None
descriptionOptional[str]助手的可选描述。需 langgraph-api 服务器版本 >= 0.0.45。None
返回值：
名称类型描述
AssistantAssistant创建的助手。
示例：
  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_idstr要更新的助手 ID。必填
graph_idOptional[str]助手使用的图 ID。通常在 langgraph.json 配置中设置。如果为 None，助手保持指向原图。None
configOptional[Config]图的配置。None
metadataJson与现有助手元数据合并的元数据。None
nameOptional[str]助手的新名称。None
headersOptional[dict[str, str]]可选的自定义请求头。None
descriptionOptional[str]助手的可选描述。需 langgraph-api 服务器版本 >= 0.0.45。None
返回值：
名称类型描述
AssistantAssistant更新后的助手。
示例：
  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_idstr要删除的助手 ID。必填
headersOptional[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]
复制代码
描述：
搜索助手。
参数：
名称类型描述默认值
metadataJson用于过滤的元数据，每个键值对进行精确匹配。None
graph_idOptional[str]用于过滤的图 ID。通常在 langgraph.json 配置中设置。None
limitint返回的最大结果数。10
offsetint跳过的结果数。0
sort_byOptional[AssistantSortBy]排序字段。None
sort_orderOptional[SortOrder]排序顺序。None
headersOptional[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_idstr要获取版本的助手 ID。必填
metadataJson用于过滤版本的元数据，每个键值对进行精确匹配。None
limitint返回的最大版本数。10
offsetint跳过的版本数。0
headersOptional[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_idstr要更改版本的助手 ID。必填
versionint要更改到的版本。必填
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
名称类型描述
AssistantAssistant助手对象。
示例：
  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_idstr要获取的线程 ID。必填
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
名称类型描述
ThreadThread线程对象。
示例：
  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
复制代码
描述：
创建新线程。
参数：
名称类型描述默认值
metadataJson添加到线程的元数据。None
thread_idOptional[str]线程 ID。如果为 None，将生成随机 UUID。None
if_existsOptional[OnConflictBehavior]处理重复创建的方式。默认在底层为 'raise'，可为 'raise'（如果重复则抛出错误）或 'do_nothing'（返回现有线程）。None
superstepsOptional[Sequence[dict[str, Sequence[dict[str, Any]]]]]创建线程时应用的超级步骤列表，每个包含更新序列。每个更新包含 values 或 command 和 as_node。用于在部署之间复制线程。None
graph_idOptional[str]与线程关联的可选图 ID。None
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
名称类型描述
ThreadThread创建的线程。
示例：
  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_idstr要更新的线程 ID。必填
metadatadict[str, Any]与现有线程元数据合并的元数据。必填
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
名称类型描述
ThreadThread更新后的线程。
示例：
  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_idstr要删除的线程 ID。必填
headersOptional[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]
复制代码
描述：
搜索线程。
参数：
名称类型描述默认值
metadataJson用于过滤的线程元数据。None
valuesJson用于过滤的状态值。None
statusOptional[ThreadStatus]过滤的线程状态，必须为 'idle'、'busy'、'interrupted' 或 'error'。None
limitint返回的最大线程数。10
offsetint开始搜索的线程表偏移量。0
sort_byOptional[ThreadSortBy]排序字段。None
sort_orderOptional[SortOrder]排序顺序。None
headersOptional[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_idstr要复制的线程 ID。必填
headersOptional[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_idstr要获取状态的线程 ID。必填
checkpointOptional[Checkpoint]要获取状态的检查点。None
checkpoint_idOptional[str]（已弃用）要获取状态的检查点 ID。None
subgraphsbool是否包含子图状态。False
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
名称类型描述
ThreadStateThreadState线程状态。
示例：
  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_idstr要更新的线程 ID。必填
valuesOptional[Union[dict, Sequence[dict]]]更新状态的值。必填
as_nodeOptional[str]以此节点刚执行的方式更新状态。None
checkpointOptional[Checkpoint]要更新状态的检查点。None
checkpoint_idOptional[str]（已弃用）要更新状态的检查点 ID。None
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
名称类型描述
ThreadUpdateStateResponseThreadUpdateStateResponse更新线程状态后的响应。
示例：
  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_idstr要获取状态历史的线程 ID。必填
checkpointOptional[Checkpoint]返回此子图的状态。如果为空，默认返回根状态。None
limitint返回的最大状态数。10
before`Optional[strCheckpoint]`返回此检查点之前的状态。
metadataOptional[dict]按元数据键值对过滤状态。None
headersOptional[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_idOptional[str]分配给线程的线程 ID。如果为 None，将创建无状态运行。必填
assistant_idstr要流式传输的助手 ID 或图名称。如果使用图名称，将默认使用从该图创建的第一个助手。必填
inputOptional[dict]图的输入。None
commandOptional[Command]要执行的命令。不可与 input 结合使用。None
stream_modeUnion[StreamMode, Sequence[StreamMode]]使用的流模式。"values"
stream_subgraphsbool是否从子图流式传输输出。False
metadataOptional[dict]分配给运行的元数据。None
configOptional[Config]助手的配置。None
checkpointOptional[Checkpoint]恢复的检查点。None
checkpoint_idOptional[str]（已弃用）恢复的检查点 ID。None
checkpoint_duringOptional[bool]是否在运行期间进行检查点（或仅在结束/中断时）。None
interrupt_beforeOptional[Union[All, Sequence[str]]]在执行前立即中断的节点。None
interrupt_afterOptional[Union[All, Sequence[str]]]在执行后立即中断的节点。None
feedback_keysOptional[Sequence[str]]分配给运行的反馈键。None
on_disconnectOptional[DisconnectMode]断开连接模式，必须为 'cancel' 或 'continue'。None
on_completionOptional[OnCompletionBehavior]无状态运行创建的线程是删除还是保留，必须为 'delete' 或 'keep'。None
webhookOptional[str]LangGraph API 调用完成后调用的 webhook。None
multitask_strategyOptional[MultitaskStrategy]多任务策略，必须为 'reject'、'interrupt'、'rollback' 或 'enqueue'。None
if_not_existsOptional[IfNotExists]处理缺失线程的方式。默认 'reject'，可为 'reject'（如果缺失则抛出错误）或 'create'（创建新线程）。None
after_secondsOptional[int]开始运行前等待的秒数。用于调度未来运行。None
headersOptional[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_idOptional[str]运行 ID。如果为 None，将生成随机 UUID。None
返回值：
名称类型描述
RunRun创建的运行对象。
示例：
  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]
复制代码
描述：
创建一批无状态后台运行。
参数：
名称类型描述默认值
payloadslist[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_idOptional[str]创建运行的线程 ID。如果为 None，将创建无状态运行。必填
assistant_idstr要运行的助手 ID 或图名称。如果使用图名称，将默认使用从该图创建的第一个助手。必填
inputOptional[dict]图的输入。None
commandOptional[Command]要执行的命令。不可与 input 结合使用。None
metadataOptional[dict]分配给运行的元数据。None
configOptional[Config]助手的配置。None
checkpointOptional[Checkpoint]恢复的检查点。None
checkpoint_idOptional[str]（已弃用）恢复的检查点 ID。None
checkpoint_duringOptional[bool]是否在运行期间进行检查点（或仅在结束/中断时）。None
interrupt_beforeOptional[Union[All, Sequence[str]]]在执行前立即中断的节点。None
interrupt_afterOptional[Union[All, Sequence[str]]]在执行后立即中断的节点。None
webhookOptional[str]LangGraph API 调用完成后调用的 webhook。None
on_disconnectOptional[DisconnectMode]断开连接模式，必须为 'cancel' 或 'continue'。None
on_completionOptional[OnCompletionBehavior]无状态运行创建的线程是删除还是保留，必须为 'delete' 或 'keep'。None
multitask_strategyOptional[MultitaskStrategy]多任务策略，必须为 'reject'、'interrupt'、'rollback' 或 'enqueue'。None
if_not_existsOptional[IfNotExists]处理缺失线程的方式。默认 'reject'，可为 'reject'（如果缺失则抛出错误）或 'create'（创建新线程）。None
after_secondsOptional[int]开始运行前等待的秒数。用于调度未来运行。None
raise_errorbool是否在运行失败时抛出错误。True
headersOptional[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_idstr要列出运行的线程 ID。必填
limitint返回的最大结果数。10
offsetint跳过的结果数。0
statusOptional[RunStatus]用于过滤的运行状态。None
headersOptional[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_idstr要获取的线程 ID。必填
run_idstr要获取的运行 ID。必填
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
名称类型描述
RunRun运行对象。
示例：
  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_idstr要取消的线程 ID。必填
run_idstr要取消的运行 ID。必填
waitbool是否等待运行完成。False
actionCancelAction取消运行时采取的动作，可为 'interrupt' 或 'rollback'。默认 'interrupt'。'interrupt'
headersOptional[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_idstr要加入的线程 ID。必填
run_idstr要加入的运行 ID。必填
headersOptional[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_idstr要加入的线程 ID。必填
run_idstr要加入的运行 ID。必填
cancel_on_disconnectbool流断开时是否取消运行。False
stream_modeOptional[Union[StreamMode, Sequence[StreamMode]]]使用的流模式，必须是创建运行时传递的流模式的子集。后台运行默认包含所有流模式的并集。None
headersOptional[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_idstr要删除的线程 ID。必填
run_idstr要删除的运行 ID。必填
headersOptional[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_idstr运行定时任务的线程 ID。必填
assistant_idstr用于定时任务的助手 ID 或图名称。如果使用图名称，将默认使用从该图创建的第一个助手。必填
schedulestr执行此任务的 cron 调度表达式。必填
inputOptional[dict]图的输入。None
metadataOptional[dict]分配给定时任务运行的元数据。None
configOptional[Config]助手的配置。None
checkpoint_duringOptional[bool]是否在运行期间进行检查点（或仅在结束/中断时）。None
interrupt_beforeOptional[Union[All, list[str]]]在执行前立即中断的节点。None
interrupt_afterOptional[Union[All, list[str]]]在执行后立即中断的节点。None
webhookOptional[str]LangGraph API 调用完成后调用的 webhook。None
multitask_strategyOptional[str]多任务策略，必须为 'reject'、'interrupt'、'rollback' 或 'enqueue'。None
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
名称类型描述
RunRun定时运行对象。
示例：
  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_idstr用于定时任务的助手 ID 或图名称。如果使用图名称，将默认使用从该图创建的第一个助手。必填
schedulestr执行此任务的 cron 调度表达式。必填
inputOptional[dict]图的输入。None
metadataOptional[dict]分配给定时任务运行的元数据。None
configOptional[Config]助手的配置。None
checkpoint_duringOptional[bool]是否在运行期间进行检查点（或仅在结束/中断时）。None
interrupt_beforeOptional[Union[All, list[str]]]在执行前立即中断的节点。None
interrupt_afterOptional[Union[All, list[str]]]在执行后立即中断的节点。None
webhookOptional[str]LangGraph API 调用完成后调用的 webhook。None
multitask_strategyOptional[str]多任务策略，必须为 'reject'、'interrupt'、'rollback' 或 'enqueue'。None
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
名称类型描述
RunRun定时运行对象。
示例：
  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_idstr要删除的定时任务 ID。必填
headersOptional[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_idOptional[str]用于搜索的助手 ID 或图名称。None
thread_idOptional[str]用于搜索的线程 ID。None
limitint返回的最大结果数。10
offsetint跳过的结果数。0
headersOptional[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
复制代码
描述：
存储或更新项目。
参数：
名称类型描述默认值
namespaceSequence[str]表示命名空间路径的字符串列表。必填
keystr命名空间内项目的唯一标识符。必填
valuedict[str, Any]包含项目数据的字典。必填
indexOptional[Union[Literal[False], list[str]]]控制搜索索引：None（使用默认值）、False（禁用）或要索引的字段路径列表。None
ttlOptional[int]项目的生存时间（分钟），若为 None 则不过期。None
headersOptional[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
复制代码
描述：
检索单个项目。
参数：
名称类型描述默认值
namespaceSequence[str]表示命名空间路径的字符串列表。必填
keystr项目的唯一标识符。必填
refresh_ttlOptional[bool]是否在此读取操作中刷新 TTL。若为 None，使用存储的默认行为。None
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
名称类型描述
ItemItem检索到的项目。
示例：
  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
复制代码
描述：
删除项目。
参数：
名称类型描述默认值
namespaceSequence[str]表示命名空间路径的字符串列表。必填
keystr项目的唯一标识符。必填
headersOptional[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_prefixSequence[str]表示命名空间前缀的字符串列表。必填
filterOptional[dict[str, Any]]用于过滤结果的键值对字典。None
limitint返回的最大项目数（默认 10）。10
offsetint返回结果前跳过的项目数（默认 0）。0
queryOptional[str]用于自然语言搜索的可选查询。None
refresh_ttlOptional[bool]是否刷新搜索返回项目的 TTL。若为 None，使用存储的默认行为。None
headersOptional[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
复制代码
描述：
列出具有可选匹配条件的命名空间。
参数：
名称类型描述默认值
prefixOptional[list[str]]用于过滤命名空间的前缀字符串列表。None
suffixOptional[list[str]]用于过滤命名空间的后缀字符串列表。None
max_depthOptional[int]返回命名空间的最大深度。None
limitint返回的最大命名空间数（默认 100）。100
offsetint返回结果前跳过的命名空间数（默认 0）。0
headersOptional[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 属性

名称类型描述
clientClient底层的 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 请求。
参数：
名称类型描述默认值
pathstr请求路径。必填
paramsOptional[QueryParamTypes]查询参数。None
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
类型描述
Any请求响应内容。
方法 post
  1. post(
  2.     path:str,*,
  3.     json: Optional[dict],
  4.     headers: Optional[dict[str,str]]=None)-> Any
复制代码
描述：
发送 POST 请求。
参数：
名称类型描述默认值
pathstr请求路径。必填
jsonOptional[dict]JSON 请求体。None
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
类型描述
Any请求响应内容。
方法 put
  1. put(
  2.     path:str,*,
  3.     json:dict,
  4.     headers: Optional[dict[str,str]]=None)-> Any
复制代码
描述：
发送 PUT 请求。
参数：
名称类型描述默认值
pathstr请求路径。必填
jsondictJSON 请求体。必填
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
类型描述
Any请求响应内容。
方法 patch
  1. patch(
  2.     path:str,*,
  3.     json:dict,
  4.     headers: Optional[dict[str,str]]=None)-> Any
复制代码
描述：
发送 PATCH 请求。
参数：
名称类型描述默认值
pathstr请求路径。必填
jsondictJSON 请求体。必填
headersOptional[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 请求。
参数：
名称类型描述默认值
pathstr请求路径。必填
jsonOptional[Any]JSON 请求体。None
headersOptional[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 流式传输请求结果。
参数：
名称类型描述默认值
pathstr请求路径。必填
methodstrHTTP 方法（如 GET、POST）。必填
jsonOptional[dict]JSON 请求体。None
paramsOptional[QueryParamTypes]查询参数。None
headersOptional[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_idstr要获取的助手 ID 或图名称（使用默认助手）。必填
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
名称类型描述
AssistantAssistant助手对象。
示例：
  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_idstr要获取图的助手 ID。必填
xrayUnion[int, bool]是否包含子图表示。如果提供整数，仅包含深度小于或等于该值的子图。False
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
名称类型描述
Graphdict[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_idstr要获取模式的助手 ID。必填
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
名称类型描述
GraphSchemaGraphSchema助手的图模式。
示例：
  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_idstr要获取子图模式的助手 ID。必填
namespaceOptional[str]可选的命名空间过滤器。None
recursebool是否递归获取子图。False
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
名称类型描述
SubgraphsSubgraphs助手的子图模式。
方法 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_idOptional[str]助手使用的图 ID。通常在 langgraph.json 配置中设置。必填
configOptional[Config]图的配置。None
metadataJson添加到助手的元数据。None
assistant_idOptional[str]助手 ID，如果未提供，将默认生成随机 UUID。None
if_existsOptional[OnConflictBehavior]处理重复创建的方式。默认在底层为 'raise'，可为 'raise'（如果重复则抛出错误）或 'do_nothing'（返回现有助手）。None
nameOptional[str]助手名称，默认在底层为 'Untitled'。None
headersOptional[dict[str, str]]可选的自定义请求头。None
descriptionOptional[str]助手的可选描述。需 langgraph-api 服务器版本 >= 0.0.45。None
返回值：
名称类型描述
AssistantAssistant创建的助手。
示例：
  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_idstr要更新的助手 ID。必填
graph_idOptional[str]助手使用的图 ID。通常在 langgraph.json 配置中设置。如果为 None，助手保持指向原图。None
configOptional[Config]图的配置。None
metadataJson与现有助手元数据合并的元数据。None
nameOptional[str]助手的新名称。None
headersOptional[dict[str, str]]可选的自定义请求头。None
descriptionOptional[str]助手的可选描述。需 langgraph-api 服务器版本 >= 0.0.45。None
返回值：
名称类型描述
AssistantAssistant更新后的助手。
示例：
  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_idstr要删除的助手 ID。必填
headersOptional[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]
复制代码
描述：
搜索助手。
参数：
名称类型描述默认值
metadataJson用于过滤的元数据，每个键值对进行精确匹配。None
graph_idOptional[str]用于过滤的图 ID。通常在 langgraph.json 配置中设置。None
limitint返回的最大结果数。10
offsetint跳过的结果数。0
headersOptional[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_idstr要获取版本的助手 ID。必填
metadataJson用于过滤版本的元数据，每个键值对进行精确匹配。None
limitint返回的最大版本数。10
offsetint跳过的版本数。0
headersOptional[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_idstr要更改版本的助手 ID。必填
versionint要更改到的版本。必填
headersOptional[dict[str, str]]可选的自定义请求头。None
返回值：
名称类型描述
AssistantAssistant助手对象。
示例：
  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
回复

使用道具 举报

返回列表 发表新帖
高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

创想小编高级会员

关注0

粉丝0

帖子212

发布主题
推荐阅读更多+
阅读排行更多+
用心服务创业者
www.llms-ai.com
AI创想

Powered by Discuz! X3.4© 2001-2013 Discuz Team.( 京ICP备17022993号-3 )