fix: main.py — build_graph() для автопроверки

main.py

@@ -1,6 +1,7 @@
-"""Human-in-the-loop: кастомное прерывание (interrupt / resume) в LangGraph."""
+"""Human-in-the-loop: кастомное прерывание в LangGraph."""
 from __future__ import annotations
 
+import uuid
 from typing import TypedDict
 
 import questionary
@@ -10,83 +11,76 @@ from langgraph.graph import StateGraph
 from langgraph.types import Command, interrupt
 
 
-class GraphState(TypedDict):
+class State(TypedDict, total=False):
-    """Состояние графа: начальные данные и ответ пользователя."""
-
     foo: str
     human_value: str
 
 
-def human_node(state: GraphState) -> GraphState:
+def human_node(state: State) -> dict:
-    """Узел с кастомным прерыванием — ждёт ответ пользователя."""
+    payload = {
-    payload = interrupt(
+        "type": "alert",
-        {
+        "question": "Уверены, что хотите продолжить?",
-            "type": "alert",
+        "allow_responds": ["approve", "reject"],
-            "question": "Уверены что хотите продолжить?",
+    }
-            "allow_responds": ["approve", "reject"],
+    resumed = interrupt(payload)
-        }
+
-    )
+    if isinstance(resumed, dict):
+        answer = resumed.get("answer", "")
+    else:
+        answer = str(resumed)
 
-    answer = payload.get("answer", "")
-    print(f"!!! {payload.get('type', 'alert')} !!!")
     print(f"> Received an input from the interrupt: {answer}")
-
     return {
         "foo": state.get("foo", ""),
         "human_value": answer,
     }
 
 
-def _ask_user(payload: dict) -> dict:
+def build_graph():
-    """Показать вопрос и записать ответ в payload."""
+    builder = StateGraph(State)
+    builder.add_node("node", human_node)
+    builder.add_edge(START, "node")
+    return builder.compile(checkpointer=InMemorySaver())
+
+
+def handle_interrupt(interrupts: tuple) -> dict:
+    first = interrupts[0]
+    payload = dict(first.value if hasattr(first, "value") else first)
+
+    print("Произошла остановка")
     print(payload)
-    options = payload.get("allow_responds") or ["approve", "reject"]
+    print(f"!!! {payload.get('type', 'alert')} !!!")
-    question = payload.get("question", "Выберите вариант:")
+
-    choice = questionary.select(question, choices=options).ask()
+    choice = questionary.select(
-    if choice is None:
+        payload["question"],
-        choice = options[0]
+        choices=payload["allow_responds"],
-    updated = dict(payload)
+    ).ask()
-    updated["answer"] = choice
+
-    return updated
+    payload["answer"] = choice or payload["allow_responds"][0]
+    return payload
 
 
-def run_graph() -> GraphState:
+def run_hitl() -> None:
-    """Запуск графа с обработкой прерывания и возобновлением."""
+    graph = build_graph()
-    builder = StateGraph(GraphState)
+    config = {"configurable": {"thread_id": str(uuid.uuid4())}}
-    builder.add_node("human_node", human_node)
+    stream_input: dict | Command = {"foo": "abc"}
-    builder.add_edge(START, "human_node")
 
-    graph = builder.compile(checkpointer=InMemorySaver())
+    while True:
-    config = {"configurable": {"thread_id": "hitl-interrupt-demo"}}
+        interrupted = False
-    initial: GraphState = {"foo": "initialized", "human_value": ""}
+        for chunk in graph.stream(stream_input, config=config):
-
+            if "__interrupt__" in chunk:
-    final_state: GraphState = initial
+                payload = handle_interrupt(chunk["__interrupt__"])
-
+                stream_input = Command(resume=payload)
-    stream = graph.stream(initial, config)
+                interrupted = True
-    for chunk in stream:
+                break
-        if "__interrupt__" in chunk:
-            print("Произошла остановка")
-            interrupt_tuple = chunk["__interrupt__"]
-            payload = dict(interrupt_tuple[0].value)
-            resumed_payload = _ask_user(payload)
-
-            resume_stream = graph.stream(Command(resume=resumed_payload), config)
-            for resume_chunk in resume_stream:
-                if "human_node" in resume_chunk:
-                    final_state = resume_chunk["human_node"]
-                    print({"node": resume_chunk})
-        elif "human_node" in chunk:
-            final_state = chunk["human_node"]
             print(chunk)
 
-    return final_state
+        if not interrupted:
+            break
 
-
+        for chunk in graph.stream(stream_input, config=config):
-def main() -> None:
+            print(chunk)
-    result = run_graph()
-    print("\n=== Итоговое состояние ===")
-    print(result)
 
 
 if __name__ == "__main__":
-    main()
+    run_hitl()