93 lines
3.2 KiB
Python
93 lines
3.2 KiB
Python
"""
|
||
Human-in-the-loop (interrupt / resume) — LangGraph
|
||
|
||
Граф с кастомным прерыванием: узел запрашивает подтверждение у пользователя
|
||
через структурированный объект (тип, вопрос, варианты ответа).
|
||
Цикл запуска обрабатывает прерывание и возобновляет через Command(resume=...).
|
||
"""
|
||
|
||
from typing import TypedDict, Optional, List
|
||
from langgraph.graph import StateGraph, START, END
|
||
from langgraph.checkpoint.memory import MemorySaver
|
||
from langgraph.types import Command, interrupt
|
||
import questionary
|
||
|
||
|
||
# 1. Состояние графа
|
||
class GraphState(TypedDict):
|
||
foo: str
|
||
human_value: Optional[str]
|
||
|
||
|
||
# 2. Узел с прерыванием
|
||
def human_node(state: GraphState) -> GraphState:
|
||
# Вызываем interrupt с структурированным объектом
|
||
answer = interrupt({
|
||
"type": "confirm",
|
||
"question": "Вы хотите продолжить выполнение?",
|
||
"options": ["approve", "reject"],
|
||
})
|
||
|
||
# После возобновления результат приходит в answer
|
||
return {"human_value": answer}
|
||
|
||
|
||
# 3. Сборка графа
|
||
builder = StateGraph(GraphState)
|
||
builder.add_node("human_node", human_node)
|
||
builder.add_edge(START, "human_node")
|
||
builder.add_edge("human_node", END)
|
||
|
||
# Компиляция с InMemorySaver
|
||
checkpointer = MemorySaver()
|
||
graph = builder.compile(checkpointer=checkpointer)
|
||
|
||
|
||
# 4. Цикл запуска
|
||
def run_graph():
|
||
config = {"configurable": {"thread_id": "hitl-demo"}}
|
||
initial_state = {"foo": "bar", "human_value": None}
|
||
|
||
print("=" * 60)
|
||
print(" Human-in-the-loop: interrupt / resume")
|
||
print("=" * 60)
|
||
|
||
# Запускаем через stream
|
||
for chunk in graph.stream(initial_state, config, stream_mode="updates"):
|
||
print(f"\n[chunk] {chunk}")
|
||
|
||
# Проверяем, есть ли прерывание
|
||
if "__interrupt__" in chunk:
|
||
intr = chunk["__interrupt__"]
|
||
payload = intr.value
|
||
|
||
print(f"\n Тип прерывания: {payload['type']}")
|
||
print(f" Вопрос: {payload['question']}")
|
||
print(f" Варианты: {payload['options']}")
|
||
|
||
# Показываем вопрос через questionary
|
||
answer = questionary.select(
|
||
payload["question"],
|
||
choices=payload["options"],
|
||
).ask()
|
||
|
||
print(f"\n Выбран ответ: {answer}")
|
||
|
||
# Возобновляем через Command(resume=...)
|
||
for resume_chunk in graph.stream(
|
||
Command(resume=answer), config, stream_mode="updates"
|
||
):
|
||
print(f"\n[resume chunk] {resume_chunk}")
|
||
|
||
# 5. Итоговое состояние
|
||
final_state = graph.get_state(config)
|
||
print("\n" + "=" * 60)
|
||
print(" Итоговое состояние:")
|
||
print(f" foo: {final_state.values['foo']}")
|
||
print(f" human_value: {final_state.values['human_value']}")
|
||
print("=" * 60)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
run_graph()
|