要审查、编辑和批准代理或工作流中的工具调用，请使用 LangGraph 的人机协作 功能。
动态中断 from langgraph_sdk import get_client from langgraph_sdk.schema import Command client = get_client( url =< DEPLOYMENT_URL > ) # Using the graph deployed with the name "agent" assistant_id = "agent" # create a thread thread = await client.threads.create() thread_id = thread[ "thread_id" ] # Run the graph until the interrupt is hit. result = await client.runs.wait( thread_id, assistant_id, input = { "some_text" : "original text" } # (1)! ) print (result[ '__interrupt__' ]) # (2)! # > [ # > { # > 'value': {'text_to_revise': 'original text'}, # > 'resumable': True, # > 'ns': ['human_node:fc722478-2f21-0578-c572-d9fc4dd07c3b'], # > 'when': 'during' # > } # > ] # Resume the graph print ( await client.runs.wait( thread_id, assistant_id, command = Command( resume = "Edited text" ) # (3)! )) # > {'some_text': 'Edited text'}
图表以某些初始状态调用。
当图表达到中断时，它会返回一个带有有效负载和元数据的中断对象。3. 图表通过 `Command(resume=...)` 恢复，注入人类输入并继续执行。
这是一个可以在 LangGraph API 服务器中运行的图表示例。更多详细信息请参阅LangSmith 快速入门 。 from typing import TypedDict import uuid from langgraph.checkpoint.memory import InMemorySaver from langgraph.constants import START from langgraph.graph import StateGraph from langgraph.types import interrupt, Command class State ( TypedDict ): some_text: str def human_node ( state : State): value = interrupt( # (1)! { "text_to_revise" : state[ "some_text" ] # (2)! } ) return { "some_text" : value # (3)! } # Build the graph graph_builder = StateGraph(State) graph_builder.add_node( "human_node" , human_node) graph_builder.add_edge( START , "human_node" ) graph = graph_builder.compile()
`interrupt(...)` 在 `human_node` 处暂停执行，将给定有效负载呈现给人类。
任何可 JSON 序列化的值都可以传递给`interrupt` 函数。此处是一个包含要修订文本的字典。
恢复后，`interrupt(...)` 的返回值是人类提供的输入，用于更新状态。
一旦您有一个运行中的 LangGraph API 服务器，您就可以使用LangGraph SDK 与其交互。 from langgraph_sdk import get_client from langgraph_sdk.schema import Command client = get_client( url =< DEPLOYMENT_URL > ) # Using the graph deployed with the name "agent" assistant_id = "agent" # create a thread thread = await client.threads.create() thread_id = thread[ "thread_id" ] # Run the graph until the interrupt is hit. result = await client.runs.wait( thread_id, assistant_id, input = { "some_text" : "original text" } # (1)! ) print (result[ '__interrupt__' ]) # (2)! # > [ # > { # > 'value': {'text_to_revise': 'original text'}, # > 'resumable': True, # > 'ns': ['human_node:fc722478-2f21-0578-c572-d9fc4dd07c3b'], # > 'when': 'during' # > } # > ] # Resume the graph print ( await client.runs.wait( thread_id, assistant_id, command = Command( resume = "Edited text" ) # (3)! )) # > {'some_text': 'Edited text'}
图表以某些初始状态调用。
当图表达到中断时，它会返回一个带有有效负载和元数据的中断对象。3. 图表通过 `Command(resume=...)` 恢复，注入人类输入并继续执行。
静态中断 静态中断（也称为静态断点）在节点执行之前或之后触发。
不建议将静态中断用于人机协作工作流。它们最适合用于调试和测试。
您可以通过在编译时指定 `interrupt_before` 和 `interrupt_after` 来设置静态中断。
graph = graph_builder.compile( # (1)! interrupt_before = [ "node_a" ], # (2)! interrupt_after = [ "node_b" , "node_c" ], # (3)! )
断点在 compile 时设置。
`interrupt_before` 指定在节点执行之前应该暂停执行的节点。
`interrupt_after` 指定在节点执行之后应该暂停执行的节点。
或者，您可以在运行时设置静态中断。
await client.runs.wait( # (1)! thread_id, assistant_id, inputs = inputs, interrupt_before = [ "node_a" ], # (2)! interrupt_after = [ "node_b" , "node_c" ] # (3)! )
`client.runs.wait` 用 `interrupt_before` 和 `interrupt_after` 参数调用。这是一个运行时配置，可以为每次调用更改。
`interrupt_before` 指定在节点执行之前应该暂停执行的节点。
`interrupt_after` 指定在节点执行之后应该暂停执行的节点。
以下示例展示了如何添加静态中断。
from langgraph_sdk import get_client client = get_client( url =< DEPLOYMENT_URL > ) # Using the graph deployed with the name "agent" assistant_id = "agent" # create a thread thread = await client.threads.create() thread_id = thread[ "thread_id" ] # Run the graph until the breakpoint result = await client.runs.wait( thread_id, assistant_id, input = inputs # (1)! ) # Resume the graph await client.runs.wait( thread_id, assistant_id, input = None # (2)! )
图将运行直到遇到第一个断点。
通过为输入传入 `None` 来恢复图表。这将运行图表直到达到下一个断点。
了解更多
人机协作概念指南 ：了解有关 LangGraph 人机协作功能的更多信息。常见模式 ：了解如何实现批准/拒绝操作、请求用户输入、工具调用审查和验证人类输入等模式。
