使用服务器 API 的人机交互 - 以编程方式使用文档

要在代理或工作流中审查、编辑和批准工具调用，请使用 LangGraph 的 human-in-the-loop features.

动态中断

Python

    from langgraph_sdk import get_client
    from langgraph_sdk.schema import Command
    client = get_client(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'}

1. 使用一些初始状态调用图。 2. 当图遇到中断时，它返回一个包含负载和元数据的中断对象。 3. 使用以下方式恢复图 Command(resume=...)，注入人工输入并继续执行。

JavaScript

    const client = new Client({ apiUrl:  });

    // Using the graph deployed with the name "agent"
    const assistantID = "agent";

    // create a thread
    const thread = await client.threads.create();
    const threadID = thread["thread_id"];

    // Run the graph until the interrupt is hit.
    const result = await client.runs.wait(
      threadID,
      assistantID,
      { input: { "some_text": "original text" } }   # (1)!
    );

    console.log(result['__interrupt__']); # (2)!
    // > [
    # >     {
    # >         'value': {'text_to_revise': 'original text'},
    # >         'resumable': True,
    # >         'ns': ['human_node:fc722478-2f21-0578-c572-d9fc4dd07c3b'],
    # >         'when': 'during'
    # >     }
    # > ]

    // Resume the graph
    console.log(await client.runs.wait(
        threadID,
        assistantID,
        { command: { resume: "Edited text" }}   # (3)!
    ));
    # > {'some_text': 'Edited text'}

1. 使用一些初始状态调用图。 2. 当图遇到中断时，它返回一个包含负载和元数据的中断对象。 3. 使用以下方式恢复图 { resume: ... } 命令对象，注入人工输入并继续执行。

cURL

创建线程：

    curl --request POST \
    --url /threads \
    --header 'Content-Type: application/json' \
    --data '{}'

运行图直到遇到中断：

    curl --request POST \
    --url /threads//runs/wait \
    --header 'Content-Type: application/json' \
    --data "{
      \"assistant_id\": \"agent\",
      \"input\": {\"some_text\": \"original text\"}
    }"

恢复图：

    curl --request POST \
     --url /threads//runs/wait \
     --header 'Content-Type: application/json' \
     --data "{
       \"assistant_id\": \"agent\",
       \"command\": {
         \"resume\": \"Edited text\"
       }
     }"

Extended example: using `interrupt`

这是您可以在代理服务器中运行的示例图。请参阅 LangSmith 快速入门了解更多详情。

  from typing import TypedDict


  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 SDK

Python

      from langgraph_sdk import get_client
      from langgraph_sdk.schema import Command
      client = get_client(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'}

JavaScript

      const client = new Client({ apiUrl:  });

      // Using the graph deployed with the name "agent"
      const assistantID = "agent";

      // create a thread
      const thread = await client.threads.create();
      const threadID = thread["thread_id"];

      // Run the graph until the interrupt is hit.
      const result = await client.runs.wait(
        threadID,
        assistantID,
        { input: { "some_text": "original text" } }   # (1)!
      );

      console.log(result['__interrupt__']); # (2)!
      # > [
      # >     {
      # >         'value': {'text_to_revise': 'original text'},
      # >         'resumable': True,
      # >         'ns': ['human_node:fc722478-2f21-0578-c572-d9fc4dd07c3b'],
      # >         'when': 'during'
      # >     }
      # > ]

      // Resume the graph
      console.log(await client.runs.wait(
          threadID,
          assistantID,
          { command: { resume: "Edited text" }}   # (3)!
      ));
      # > {'some_text': 'Edited text'}

cURL

创建线程：

      curl --request POST \
      --url /threads \
      --header 'Content-Type: application/json' \
      --data '{}'

运行图直到遇到中断：

      curl --request POST \
      --url /threads//runs/wait \
      --header 'Content-Type: application/json' \
      --data "{
        \"assistant_id\": \"agent\",
        \"input\": {\"some_text\": \"original text\"}
      }"

恢复图：

      curl --request POST \
      --url /threads//runs/wait \
      --header 'Content-Type: application/json' \
      --data "{
        \"assistant_id\": \"agent\",
        \"command\": {
          \"resume\": \"Edited text\"
        }
      }"

静态中断

静态中断（也称为静态断点）在节点执行之前或之后触发。

您可以通过在编译时指定 interrupt_before 和 interrupt_after 来设置静态中断：

graph = graph_builder.compile( # (1)!
    interrupt_before=["node_a"], # (2)!
    interrupt_after=["node_b", "node_c"], # (3)!
)

断点是在以下期间设置的 compile time.
interrupt_before 指定应在节点执行之前暂停执行的位置。
interrupt_after 指定执行应在节点执行后暂停的节点。

或者，您可以在运行时设置静态中断：

Python

    await client.runs.wait( # (1)!
        thread_id,
        assistant_id,
        inputs=inputs,
        interrupt_before=["node_a"], # (2)!
        interrupt_after=["node_b", "node_c"] # (3)!
    )

1. client.runs.wait 使用以下参数调用 interrupt_before 和 interrupt_after 参数。这是运行时配置，可以在每次调用时更改。 2. interrupt_before 指定执行应在节点执行前暂停的节点。 3. interrupt_after 指定执行应在节点执行后暂停的节点。

JavaScript

    await client.runs.wait( // (1)!
        threadID,
        assistantID,
        {
        input: input,
        interruptBefore: ["node_a"], // (2)!
        interruptAfter: ["node_b", "node_c"] // (3)!
        }
    )

1. client.runs.wait 使用以下参数调用 interruptBefore 和 interruptAfter 参数。这是运行时配置，可以在每次调用时更改。 2. interruptBefore 指定执行应在节点执行前暂停的节点。 3. interruptAfter 指定执行应在节点执行后暂停的节点。

cURL

    curl --request POST \
    --url /threads//runs/wait \
    --header 'Content-Type: application/json' \
    --data "{
        \"assistant_id\": \"agent\",
        \"interrupt_before\": [\"node_a\"],
        \"interrupt_after\": [\"node_b\", \"node_c\"],
        \"input\":
    }"

以下示例展示如何添加静态中断：

Python

    from langgraph_sdk import get_client
    client = get_client(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)!
    )

1. 图运行直到第一个断点被触发。 2. 通过传入 None 作为输入来恢复图。这将运行图直到下一个断点被触发。

JavaScript

    const client = new Client({ apiUrl:  });

    // Using the graph deployed with the name "agent"
    const assistantID = "agent";

    // create a thread
    const thread = await client.threads.create();
    const threadID = thread["thread_id"];

    // Run the graph until the breakpoint
    const result = await client.runs.wait(
      threadID,
      assistantID,
      { input: input }   # (1)!
    );

    // Resume the graph
    await client.runs.wait(
      threadID,
      assistantID,
      { input: null }   # (2)!
    );

1. 图运行直到第一个断点被触发。 2. 通过传入 null 作为输入来恢复图。这将运行图直到下一个断点被触发。

cURL

创建线程：

    curl --request POST \
    --url /threads \
    --header 'Content-Type: application/json' \
    --data '{}'

运行图直到断点：

    curl --request POST \
    --url /threads//runs/wait \
    --header 'Content-Type: application/json' \
    --data "{
      \"assistant_id\": \"agent\",
      \"input\":
    }"

恢复图：

    curl --request POST \
    --url /threads//runs/wait \
    --header 'Content-Type: application/json' \
    --data "{
      \"assistant_id\": \"agent\"
    }"

了解更多

* 人工介入概念指南：了解更多关于 LangGraph 人工介入功能的信息。
* 常见模式: learn how to implement patterns like approving/rejecting actions, requesting user input, tool call review, and validating human input.