add file

main.py

@@ -1,66 +1,67 @@
-import sys
+import questionary
-from typing import TypedDict, List, Dict, Any
-
 from langgraph.graph import StateGraph, START
 from langgraph.constants import interrupt
 from langgraph.types import Command
 from langgraph.checkpoint.memory import InMemorySaver
-import questionary
+from typing import TypedDict, List, Dict
 
-# Define state
+class GraphState(TypedDict):
-class State(TypedDict):
     human_value: str | None
     foo: str | None
 
 # Node that triggers interrupt
-def interrupt_node(state: State) -> State:
+
-    # Trigger interrupt with question and options
+def interrupt_node(state: GraphState) -> GraphState:
-    interrupt_payload = {
+    # Trigger interrupt with structured payload
+    payload = {
         "type": "confirm",
         "question": "Уверены, что хотите продолжить?",
         "allow_responds": ["approve", "reject"],
     }
-    # Raise interrupt; graph will pause until resumed
+    # interrupt returns None, execution pauses until resume
-    interrupt(interrupt_payload)
+    interrupt(payload)
-    # After resume, the payload will be merged into state via resume
+    # After resume, the same payload will be passed back via state
-    # We expect state to contain 'human_value' set by resume
+    # We expect the resume payload to contain 'answer'
+    answer = state.get("answer")
+    state["human_value"] = answer
     return state
 
 # Build graph
-graph = StateGraph(State)
+builder = StateGraph(GraphState)
-graph.add_node("interrupt", interrupt_node)
+builder.add_node("interrupt_node", interrupt_node)
-graph.set_entry_point(START)
+builder.set_entry_point("interrupt_node")
-graph.add_edge(START, "interrupt")
+builder.set_finish_point("interrupt_node")
-# No further nodes; graph ends after interrupt node
+# Use InMemorySaver for checkpointing
-graph.set_finish_point("interrupt")
-
-# Compile with checkpoint
 checkpoint = InMemorySaver()
-compiled = graph.compile(checkpointer=checkpoint)
+graph = builder.compile(checkpointer=checkpoint)
 
 # Run graph with interrupt handling
-if __name__ == "__main__":
+config = {"configurable": {"thread_id": "thread-1"}}
-    thread_id = "demo_thread"
+# Initial state
-    config = {"configurable": {"thread_id": thread_id}}
+state: GraphState = {"human_value": None, "foo": None}
-    # Start stream
+
-    stream = compiled.stream({}, config)
+# Start streaming
+stream = graph.stream(state, config)
 for chunk in stream:
     if "__interrupt__" in chunk:
-            # Extract interrupt payload
+        # Extract payload
-            interrupt_obj = chunk["__interrupt__"][0]
+        payload = chunk["__interrupt__"][0].value
-            payload = interrupt_obj.value
         # Show question to user
         answer = questionary.select(
             payload["question"],
             choices=payload["allow_responds"],
         ).ask()
-            # Resume with answer
+        # Add answer to payload and resume
-            resume_payload = {"human_value": answer}
+        payload["answer"] = answer
-            stream = compiled.stream(Command(resume=resume_payload), config)
+        # Resume graph
+        stream = graph.stream(Command(resume=payload), config)
+        # Continue processing resumed stream
         for subchunk in stream:
-                if "human_value" in subchunk:
+            if "__interrupt__" in subchunk:
-                    print("\nFinal state:", subchunk)
+                # Should not happen in this simple example
-                    sys.exit(0)
+                continue
+            print(subchunk)
     else:
-            # Normal output (none expected here)
+        print(chunk)
-            pass
+
+print("Final state:", state)