Human-in-the-loop (interrupt / resume): interrupt_graph.py

2026-05-27 10:03:34 +00:00
parent f85e5ab80c
commit 01735b12cc
1 changed files with 88 additions and 0 deletions
@@ -0,0 +1,88 @@
 <|channel|>final code<|message|>import sys
 from typing import TypedDict, List
 import questionary
 from langgraph.checkpoint.memory import InMemorySaver
 from langgraph.constants import START
 from langgraph.graph import StateGraph
 from langgraph.types import interrupt, Command
 # ---------- 1. Состояние графа ----------
 class GraphState(TypedDict):
    human_value: str | None
    foo: str | None
 # ---------- 2. Узел с прерыванием ----------
 def node_with_interrupt(state: GraphState) -> GraphState:
    # Если пользователь уже ответил, просто возвращаем состояние
    if state.get("human_value") is not None:
        return state
    # Создаём объект прерывания
    interrupt_payload = {
        "type": "confirm",
        "question": "Уверены, что хотите продолжить?",
        "allow_responds": ["approve", "reject"],
    }
    # Вызываем прерывание – выполнение остановится до возобновления
    return interrupt(interrupt_payload)
 # ---------- 3. Сборка графа ----------
 builder = StateGraph(GraphState)
 builder.add_node("interrupt_node", node_with_interrupt)
 builder.set_entry_point("interrupt_node")
 graph = builder.compile(checkpointer=InMemorySaver())
 # ---------- 4. Цикл запуска с обработкой прерывания ----------
 def run_graph() -> None:
    # Инициализируем состояние
    init_state: GraphState = {"human_value": None, "foo": None}
    thread_id = "demo_thread"
    # Первый запуск – поток до первого прерывания
    stream = graph.stream(
        init_state,
        configurable={"thread_id": thread_id},
    )
    for chunk in stream:
        if "__interrupt__" in chunk:
            # Получаем объект прерывания
            interrupt_obj = chunk["__interrupt__"][0].value  # type: ignore[index]
            print("\n=== Появилось прерывание ===")
            print(f"Тип: {interrupt_obj['type']}")
            print(f"Вопрос: {interrupt_obj['question']}")
            print(f"Варианты: {', '.join(interrupt_obj['allow_responds'])}")
            # Запрашиваем ответ пользователя
            answer = questionary.select(
                interrupt_obj["question"],
                choices=interrupt_obj["allow_responds"],
            ).ask()
            if answer is None:
                sys.exit("Отмена пользователем")
            # Добавляем ответ в объект прерывания и возобновляем граф
            interrupt_obj["answer"] = answer
            resume_cmd = Command(resume=interrupt_obj)
            stream = graph.stream(
                resume_cmd,
                configurable={"thread_id": thread_id},
            )
        else:
            # Выводим обычный результат (можно логировать)
            print(chunk)
    # После завершения выводим финальное состояние
    final_state: GraphState = stream.final_state()
    print("\n=== Финальное состояние ===")
    print(final_state)
 if __name__ == "__main__":
    run_graph()