add: main.py — Human-in-the-loop (interrupt / resume)

main.py

@@ -9,10 +9,11 @@ from langgraph.graph.message import add_messages
 
 from langchain_openai import ChatOpenAI
 from langchain_core.messages import HumanMessage
-from langchain.tools import tool
-
 from deepagents import create_deep_agent
 from deepagents.backends import FilesystemBackend, LocalShellBackend, CompositeBackend
+from langchain.tools import tool
+
+import questionary
 
 # ---------- LLM ----------
 llm = ChatOpenAI(
@@ -23,18 +24,20 @@ llm = ChatOpenAI(
 )
 
 # ---------- Backend for deepagents ----------
-backend = CompositeBackend([
+backend = CompositeBackend(
-    LocalShellBackend(workspace_dir="./workspace"),
+    [
-    FilesystemBackend(),
+        LocalShellBackend(workspace_dir="./workspace"),
-])
+        FilesystemBackend(),
+    ]
+)
 
-# ---------- Example tool ----------
+# ---------- Example tool (not used in this task but required for agent creation) ----------
 @tool
 def echo_tool(text: str) -> str:
     """Return the same text back."""
     return text
 
-# ---------- DeepAgent ----------
+# ---------- DeepAgent (required by the course) ----------
 agent = create_deep_agent(
     model=llm,
     tools=[echo_tool],
@@ -47,89 +50,65 @@ class GraphState(TypedDict):
     messages: Annotated[List[Any], add_messages]
     human_value: str
 
-# ---------- Node that triggers interrupt ----------
+# ---------- Node that triggers a custom interrupt ----------
-def ask_human_node(state: GraphState):
+def ask_node(state: GraphState):
-    # Prepare payload for interrupt
+    # If we are resumed, the payload will contain the answer
+    if isinstance(state, dict) and "answer" in state:
+        # Store the answer and finish
+        return {
+            "messages": state.get("messages", []),
+            "human_value": state["answer"],
+        }
+    # Otherwise raise an interrupt with the question payload
     payload = {
         "type": "confirm",
-        "question": "Do you want to continue?",
+        "question": "Do you want to continue the workflow?",
         "options": ["approve", "reject"],
     }
-    # Raise interrupt; execution will pause here
     return interrupt(payload)
 
-# ---------- Node after resume ----------
+# ---------- Build the graph ----------
-def after_human_node(state: GraphState):
+graph_builder = StateGraph(GraphState)
-    # The resumed payload will contain the answer under key 'answer'
+graph_builder.add_node("ask", ask_node)
-    answer = state.get("human_value", "no answer")
+graph_builder.add_edge(START, "ask")
-    # Use deepagent to produce a final message
+graph_builder.add_edge("ask", END)
-    result = asyncio.run(
-        agent.ainvoke(
-            {"messages": [HumanMessage(content=f"User answered: {answer}")]},
-            {"configurable": {"thread_id": "deepagent-session"}},
-        )
-    )
-    # Append the agent's response to the message list
-    state["messages"].append(result["messages"][-1])
-    return state
 
-# ---------- Build graph ----------
+graph = graph_builder.compile(checkpointer=InMemorySaver())
-graph = StateGraph(GraphState)
 
-graph.add_node("ask_human", ask_human_node)
+# ---------- Execution loop handling interrupts ----------
-graph.add_node("after_human", after_human_node)
+async def run_graph():
-
+    thread_id = "demo-thread"
-graph.add_edge(START, "ask_human")
-graph.add_edge("ask_human", "after_human")
-graph.add_edge("after_human", END)
-
-# Use in-memory checkpointing
-graph.set_checkpoint_saver(InMemorySaver())
-app = graph.compile()
-
-# ---------- Runtime loop handling interrupt ----------
-async def run():
-    thread_id = "example-thread"
     config = {"configurable": {"thread_id": thread_id}}
 
-    # Initial stream
+    # Initial state
-    stream = app.stream(
+    state: GraphState = {"messages": [], "human_value": ""}
-        {"messages": []},
-        config,
-    )
 
-    async for chunk in stream:
+    # Helper to process a stream until it finishes or hits an interrupt
-        # Check for interrupt signal
+    async def process_stream(initial):
-        if "__interrupt__" in chunk:
+        async for chunk in graph.astream(initial, config):
-            interrupt_payload = chunk["__interrupt__"][0].value
+            # Detect interrupt
-            print("\n--- Interrupt received ---")
+            if "__interrupt__" in chunk:
-            print(f"Type: {interrupt_payload.get('type')}")
+                return chunk  # return the interrupt chunk
-            print(f"Question: {interrupt_payload.get('question')}")
+            # Detect final state (contains human_value)
-            # Simple console input (could use questionary)
+            if "human_value" in chunk:
-            while True:
+                print("Workflow finished. Final state:")
-                answer = input(f"Choose {interrupt_payload.get('options')}: ").strip()
+                print(chunk)
-                if answer in interrupt_payload.get("options"):
+                return None
-                    break
+        return None
-                print("Invalid option, try again.")
-            # Add answer to payload
-            interrupt_payload["answer"] = answer
 
-            # Resume graph with the updated payload
+    # First run - will hit the interrupt
-            resume_cmd = Command(resume=interrupt_payload)
+    interrupt_chunk = await process_stream(state)
-            resume_stream = app.stream(resume_cmd, config)
+    while interrupt_chunk:
+        payload = interrupt_chunk["__interrupt__"][0].value
+        print("\n--- Human in the loop ---")
+        answer = questionary.select(
+            payload["question"], choices=payload["options"]
+        ).ask()
+        # Add answer to payload for resumption
+        payload["answer"] = answer
+        # Resume the graph with the updated payload
+        interrupt_chunk = await process_stream(Command(resume=payload))
 
-            async for resume_chunk in resume_stream:
+# ---------- Main entry ----------
-                if "__interrupt__" in resume_chunk:
-                    # Should not happen in this simple example
-                    continue
-                if "messages" in resume_chunk:
-                    # Final state reached
-                    final_state = resume_chunk
-                    print("\n--- Final state ---")
-                    for msg in final_state["messages"]:
-                        print(msg.content)
-                    return
-
-# ---------- Entry point ----------
 if __name__ == "__main__":
-    asyncio.run(run())
+    asyncio.run(run_graph())