人工干预¶

from langgraph.types import interrupt

def human_node(state: State):
    value = interrupt(
        # 任何可序列化为 JSON 的值，供人类查看。
        # 例如，一个问题、一段文本或状态中的一组键
       {
          "text_to_revise": state["some_text"]
       }
    )
    # 使用人类的输入更新状态或根据输入调整图形。
    return {
        "some_text": value
    }

graph = graph_builder.compile(
    checkpointer=checkpointer # `interrupt` 工作所需
)

# 运行图形直到遇到中断
thread_config = {"configurable": {"thread_id": "some_id"}}
graph.invoke(some_input, config=thread_config)

# 用人类的输入恢复图形
graph.invoke(Command(resume=value_from_human), config=thread_config)

from typing import Literal
from langgraph.types import interrupt, Command

def human_approval(state: State) -> Command[Literal["some_node", "another_node"]]:
    is_approved = interrupt(
        {
            "question": "这是正确的吗？",
            # 展示应由人类审查和批准的输出。
            "llm_output": state["llm_output"]
        }
    )

    if is_approved:
        return Command(goto="some_node")
    else:
        return Command(goto="another_node")

# 将节点添加到图形中的适当位置并连接到相关节点。
graph_builder.add_node("human_approval", human_approval)
graph = graph_builder.compile(checkpointer=checkpointer)

# 在运行图形并触发中断后，图形将暂停。
# 用批准或拒绝恢复。
thread_config = {"configurable": {"thread_id": "some_id"}}
graph.invoke(Command(resume=True), config=thread_config)

from langgraph.types import interrupt

def human_editing(state: State):
    ...
    result = interrupt(
        # 中断信息以展示给客户端。
        # 可以是任何可序列化为 JSON 的值。
        {
            "task": "审查 LLM 的输出并进行必要的编辑。",
            "llm_generated_summary": state["llm_generated_summary"]
        }
    )

    # 使用编辑后的文本更新状态
    return {
        "llm_generated_summary": result["edited_text"] 
    }

# 将节点添加到图形中的适当位置并连接到相关节点。
graph_builder.add_node("human_editing", human_editing)
graph = graph_builder.compile(checkpointer=checkpointer)

...

# 在运行图形并触发中断后，图形将暂停。
# 用编辑后的文本恢复。
thread_config = {"configurable": {"thread_id": "some_id"}}
graph.invoke(
    Command(resume={"edited_text": "编辑后的文本"}), 
    config=thread_config
)

def human_review_node(state) -> Command[Literal["call_llm", "run_tool"]]:
    # 这是我们将通过 Command(resume=<human_review>) 提供的值
    human_review = interrupt(
        {
            "question": "这个正确吗？",
            # 提供工具调用以供审查
            "tool_call": tool_call
        }
    )

    review_action, review_data = human_review

    # 批准工具调用并继续
    if review_action == "continue":
        return Command(goto="run_tool")

    # 手动修改工具调用，然后继续
    elif review_action == "update":
        ...
        updated_msg = get_updated_msg(review_data)
        # 请记住，要修改现有消息，您需要
        # 传递具有匹配ID的消息。
        return Command(goto="run_tool", update={"messages": [updated_message]})

    # 提供自然语言反馈，然后将其传回代理
    elif review_action == "feedback":
        ...
        feedback_msg = get_feedback_msg(review_data)
        return Command(goto="call_llm", update={"messages": [feedback_msg]})

from langgraph.types import interrupt

def human_input(state: State):
    human_message = interrupt("human_input")
    return {
        "messages": [
            {
                "role": "human",
                "content": human_message
            }
        ]
    }

def agent(state: State):
    # 代理逻辑
    ...

graph_builder.add_node("human_input", human_input)
graph_builder.add_edge("human_input", "agent")
graph = graph_builder.compile(checkpointer=checkpointer)

# 运行图形并触发中断后，图形将暂停。
# 使用人类的输入恢复它。
graph.invoke(
    Command(resume="你好！"),
    config=thread_config
)

from langgraph.types import interrupt

def human_node(state: MessagesState) -> Command[Literal["agent_1", "agent_2", ...]]:
    """一个用于收集用户输入的节点。"""
    user_input = interrupt(value="准备好进行用户输入。")

    # 从状态中确定**活动代理**，因此
    # 收集输入后可以路由到正确的代理。
    # 例如，可以在状态中添加一个字段，或使用最后一个活动代理。
    # 或填写代理生成的AI消息的`name`属性。
    active_agent = ...

    return Command(
        update={
            "messages": [{
                "role": "human",
                "content": user_input,
            }]
        },
        goto=active_agent,
    )

from langgraph.types import interrupt

def human_node(state: State):
    """带验证的人类节点。"""
    question = "你多大年龄？"

    while True:
        answer = interrupt(question)

        # 验证答案，如果答案无效，则重新请求输入。
        if not isinstance(answer, int) or answer < 0:
            question = f"'{answer} 不是一个有效的年龄。你多大年龄？"
            answer = None
            continue
        else:
            # 如果答案有效，我们可以继续。
            break

    print(f"回路中的人类年龄为 {answer} 岁。")
    return {
        "age": answer
    }

# 运行图形直到中断
result = graph.invoke(inputs, thread_config)
# 获取图的状态以获取中断信息。
state = graph.get_state(thread_config)
# 打印状态值
print(state.values)
# 打印待处理任务
print(state.tasks)
# 使用用户的输入恢复图。
graph.invoke(Command(resume={"age": "25"}), thread_config)

counter = 0
def node(state: State):
    # 从节点开始到中断的所有代码将在图恢复时重新执行。
    global counter
    counter += 1
    print(f"> 进入节点: {counter} 次")
    # 暂停图并等待用户输入。
    answer = interrupt()
    print("计数器的值是:", counter)
    ...

def node_in_parent_graph(state: State):
    some_code()  # <-- 这将在子图恢复时再次执行。
    # 以函数方式调用子图。
    # 子图包含一个 `interrupt` 调用。
    subgraph_result = subgraph.invoke(some_input)
    ...