85 lines
3.2 KiB
Python
85 lines
3.2 KiB
Python
"""
|
||
Main entry point for the Human‑in‑the‑loop LangGraph example.
|
||
|
||
The program demonstrates a simple graph with one node that pauses execution and asks the user to confirm an action. The pause is implemented using ``langgraph``'s custom interrupt mechanism. After the user answers, the graph resumes and prints the final state.
|
||
|
||
Requirements:
|
||
pip install -r requirements.txt
|
||
|
||
Run:
|
||
python main.py
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from typing import Dict, Any
|
||
|
||
import questionary
|
||
from langgraph.constants import START
|
||
from langgraph.types import Command
|
||
from langgraph.checkpoint.memory import InMemorySaver
|
||
from graph import build_graph
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 1. Node that triggers an interrupt (defined in graph.py)
|
||
# ---------------------------------------------------------------------------
|
||
# The node is already defined in graph.py; we just use the compiled graph.
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Main loop that runs the graph and handles interrupts
|
||
# ---------------------------------------------------------------------------
|
||
async def run_graph() -> None:
|
||
"""Execute the graph, handle the custom interrupt, and resume execution."""
|
||
from asyncio import run # Imported lazily to keep top‑level imports minimal.
|
||
|
||
graph = build_graph()
|
||
config = {"configurable": {"thread_id": "demo-thread-1"}}
|
||
|
||
# Initial state – ``foo`` can be any value; it is not used in this demo.
|
||
init_state: Dict[str, Any] = {"human_value": None, "foo": 42}
|
||
|
||
async for chunk in graph.stream(init_state, config):
|
||
if "__interrupt__" in chunk:
|
||
interrupt_obj = chunk["__interrupt__"][0]
|
||
payload: Dict[str, Any] = interrupt_obj.value # type: ignore[assignment]
|
||
|
||
print("\n--- Human‑in‑the‑loop interrupt received ---")
|
||
print(f"Type: {payload.get('type')}")
|
||
print(f"Question: {payload.get('question')}\n")
|
||
|
||
# Ask the user for a choice.
|
||
answer = questionary.select(
|
||
"Choose an option:",
|
||
choices=payload.get("options", []),
|
||
).ask()
|
||
|
||
if answer is None:
|
||
raise RuntimeError("User cancelled the prompt")
|
||
|
||
# Attach the answer to the payload and resume.
|
||
payload["answer"] = answer
|
||
print(f"\n> User selected: {answer}\n")
|
||
await graph.stream(Command(resume=payload), config)
|
||
else:
|
||
# Normal output – just print it.
|
||
if "messages" in chunk:
|
||
for msg in chunk["messages"]:
|
||
print(msg.content, end="", flush=True)
|
||
|
||
# After the stream ends, fetch the final state from the checkpoint.
|
||
final_state = graph.checkpointer.get_state(config)
|
||
print("\n--- Final state ---")
|
||
print(final_state)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Entry point
|
||
# ---------------------------------------------------------------------------
|
||
if __name__ == "__main__":
|
||
import asyncio
|
||
|
||
try:
|
||
asyncio.run(run_graph())
|
||
except KeyboardInterrupt:
|
||
print("\nInterrupted by user.")
|