From 37ac3955607bafe16127989f02215db978ca71e5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=90=D1=80=D1=82=D0=B5=D0=BC=20=D0=92=D0=BB=D0=B0=D0=B4?= =?UTF-8?q?=D0=B8=D0=BC=D0=B8=D1=80=D0=BE=D0=B2=D0=B8=D1=87=20=D0=91=D0=B0?= =?UTF-8?q?=D0=B1=D0=B0=D0=B9=D0=BA=D0=B8=D0=BD?= Date: Thu, 28 May 2026 14:11:17 +0000 Subject: [PATCH] feat: solution for unknown --- solutions/unknown/solution.py | 67 +++++++++++++++++++++++++++++++++++ 1 file changed, 67 insertions(+) create mode 100644 solutions/unknown/solution.py diff --git a/solutions/unknown/solution.py b/solutions/unknown/solution.py new file mode 100644 index 0000000..99c5cb5 --- /dev/null +++ b/solutions/unknown/solution.py @@ -0,0 +1,67 @@ +from typing import TypedDict, Dict, Any + +from langgraph.graph import StateGraph, START, END +from langgraph.checkpoint.memory import InMemorySaver + + +class GraphState(TypedDict): + human_value: str | None + __interrupt__: dict[str, Any] | None # for interrupt payload + __resume__: str | None # for resume payload + + +def interrupt_node(state: GraphState) -> Dict[str, Any]: + """Trigger an interrupt asking the user to choose a value.""" + return { + "__interrupt__": { + "type": "question", + "question": "Choose a value:", + "options": ["Option A", "Option B", "Option C"], + } + } + + +def resume_node(state: GraphState) -> Dict[str, Any]: + """Store the user's choice and finish.""" + # The chosen option is passed via the __resume__ key + if state.get("__resume__") is not None: + state["human_value"] = state["__resume__"] + return {"__end__": True} + + +builder = StateGraph(GraphState) +builder.add_node("interrupt", interrupt_node) +builder.add_node("resume", resume_node) + +# The graph starts with the interrupt node +builder.set_entry_point("interrupt") + +# After a successful resume, we end the graph +builder.add_edge(START, "interrupt") +builder.add_conditional_edges( + "interrupt", + lambda x: "__interrupt__" in x, + {"__interrupt__": "resume"}, +) +builder.add_edge("resume", END) + +graph = builder.compile(checkpointer=InMemorySaver()) + +# Run the graph and handle interrupts +state: GraphState = {"human_value": None, "__interrupt__": None, "__resume__": None} +while True: + result = graph.invoke(state) + if "__interrupt__" in result and result["__interrupt__"] is not None: + interrupt_info = result["__interrupt__"] + print(interrupt_info["question"]) + for idx, opt in enumerate(interrupt_info["options"], 1): + print(f"{idx}. {opt}") + choice_idx = int(input("Enter number: ")) - 1 + chosen = interrupt_info["options"][choice_idx] + # Resume the graph with the chosen value + state = graph.invoke(state, {"__resume__": chosen}) + else: + # Graph finished + break + +print("Final state:", state) \ No newline at end of file