Пауза и человек в цикле | Курс LangGraph урок 6

"""Пример 1 урока 6: пауза внутри узла и возобновление.

Узел choose_file останавливает запуск вызовом interrupt() и отдаёт наружу
список файлов стенда. Человек отвечает именем файла, ответ приходит вызовом
с Command(resume=...). При возобновлении узел choose_file выполняется с первой
строки, interrupt() возвращает ответ, и узел run_file выполняет выбранный файл
отдельным процессом.
"""

import subprocess
import sys
from pathlib import Path

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
from typing_extensions import TypedDict

REPO = Path(__file__).parent / "mini_repo"
CONFIG = {"configurable": {"thread_id": "run-1"}}
CALLS: list[str] = []


class RunState(TypedDict):
    root: str
    files: list[str]
    chosen: str
    output: str


def collect_files(state: RunState) -> dict:
    """Собирает имена файлов стенда."""
    CALLS.append("collect_files")
    names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
    return {"files": names}


def choose_file(state: RunState) -> dict:
    """Ставит запуск на паузу с вопросом, какой файл выполнить."""
    CALLS.append("choose_file")
    chosen = interrupt({"вопрос": "Какой файл выполнить?", "файлы": state["files"]})
    return {"chosen": chosen}


def run_file(state: RunState) -> dict:
    """Выполняет выбранный файл стенда отдельным процессом."""
    CALLS.append("run_file")
    done = subprocess.run(
        [sys.executable, state["chosen"]],
        cwd=state["root"],
        capture_output=True,
        text=True,
        encoding="utf-8",
    )
    return {"output": f"код {done.returncode}, вывод: {done.stdout.strip() or '-'}"}


builder = StateGraph(RunState)
builder.add_node("collect_files", collect_files)
builder.add_node("choose_file", choose_file)
builder.add_node("run_file", run_file)
builder.add_edge(START, "collect_files")
builder.add_edge("collect_files", "choose_file")
builder.add_edge("choose_file", "run_file")
builder.add_edge("run_file", END)
graph = builder.compile(checkpointer=InMemorySaver())

START_STATE = {"root": str(REPO), "files": [], "chosen": "", "output": ""}

# 1. Первый вызов останавливается на interrupt() в узле choose_file.
result = graph.invoke(START_STATE, CONFIG)
pause = result["__interrupt__"][0]
print("1. ключи результата:", sorted(result))
print("   значение паузы:", pause.value)
print("   выполнены узлы:", CALLS)

# 2. Что сохранил чекпойнтер на паузе.
snapshot = graph.get_state(CONFIG)
print("2. next:", snapshot.next)
print("   задача на паузе:", snapshot.tasks[0].name)
print("   то же значение в get_state:", snapshot.interrupts[0].value == pause.value)

# 3. Ответ человека: возобновление тем же thread_id.
CALLS.clear()
result = graph.invoke(Command(resume="app.py"), CONFIG)
print("3. выполнены узлы:", CALLS)
print("   выбран:", result["chosen"])
print("   выполнение файла:", result["output"])
print("   пауза в результате:", "__interrupt__" in result)

# Вывод:
# 1. ключи результата: ['__interrupt__', 'chosen', 'files', 'output', 'root']
#    значение паузы: {'вопрос': 'Какой файл выполнить?', 'файлы': ['app.py', 'config.py', 'db.py', 'utils.py']}
#    выполнены узлы: ['collect_files', 'choose_file']
# 2. next: ('choose_file',)
#    задача на паузе: choose_file
#    то же значение в get_state: True
# 3. выполнены узлы: ['choose_file', 'run_file']
#    выбран: app.py
#    выполнение файла: код 0, вывод: [('select count(*) from orders', 3)]
#    пауза в результате: False

"""Пример 2 урока 6: что нужно возобновлению и как пауза ждёт ответа.

Граф из примера 1. Без чекпойнтера запуск останавливается, но возобновить
его нельзя. С SqliteSaver пауза лежит в файле: первый день заканчивается на
паузе и закрывает соединение. Второй день открывает файл новым соединением и
собирает граф заново. Вызов invoke(None, CONFIG) вместо ответа снова
останавливается на той же паузе, ответ даёт вызов с Command(resume=...).
Повтор от checkpoint с паузой ставит паузу заново, и новый ответ возобновляет
уже повтор.
"""

import subprocess
import sys
from pathlib import Path

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
from typing_extensions import TypedDict

HERE = Path(__file__).parent
REPO = HERE / "mini_repo"
DB_PATH = HERE / "pause.sqlite"
CONFIG = {"configurable": {"thread_id": "run-1"}}
START_STATE = {"root": str(REPO), "files": [], "chosen": "", "output": ""}
CALLS: list[str] = []


class RunState(TypedDict):
    root: str
    files: list[str]
    chosen: str
    output: str


def collect_files(state: RunState) -> dict:
    """Собирает имена файлов стенда."""
    CALLS.append("collect_files")
    names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
    return {"files": names}


def choose_file(state: RunState) -> dict:
    """Ставит запуск на паузу с вопросом, какой файл выполнить."""
    CALLS.append("choose_file")
    chosen = interrupt({"вопрос": "Какой файл выполнить?", "файлы": state["files"]})
    return {"chosen": chosen}


def run_file(state: RunState) -> dict:
    """Выполняет выбранный файл стенда отдельным процессом."""
    CALLS.append("run_file")
    done = subprocess.run(
        [sys.executable, state["chosen"]],
        cwd=state["root"],
        capture_output=True,
        text=True,
        encoding="utf-8",
    )
    return {"output": f"код {done.returncode}, вывод: {done.stdout.strip() or '-'}"}


def build(checkpointer=None):
    """Собирает граф примера 1 с переданным чекпойнтером или без него."""
    builder = StateGraph(RunState)
    builder.add_node("collect_files", collect_files)
    builder.add_node("choose_file", choose_file)
    builder.add_node("run_file", run_file)
    builder.add_edge(START, "collect_files")
    builder.add_edge("collect_files", "choose_file")
    builder.add_edge("choose_file", "run_file")
    builder.add_edge("run_file", END)
    return builder.compile(checkpointer=checkpointer)


def remove_db() -> None:
    """Удаляет файл базы вместе со служебными файлами SQLite."""
    for suffix in ("", "-wal", "-shm"):
        Path(f"{DB_PATH}{suffix}").unlink(missing_ok=True)


# 1. Без чекпойнтера пауза случается, а возобновить запуск нечем.
plain = build()
result = plain.invoke(START_STATE)
print("1. без чекпойнтера, пауза в результате:", "__interrupt__" in result)
try:
    plain.invoke(Command(resume="app.py"))
except RuntimeError as error:
    print("   возобновление:", error)

# 2. Первый день: пауза записана в файл, соединение закрыто.
remove_db()
with SqliteSaver.from_conn_string(str(DB_PATH)) as checkpointer:
    result = build(checkpointer).invoke(START_STATE, CONFIG)
    print("2. первый день, пауза:", result["__interrupt__"][0].value["вопрос"])

# 3. Второй день: новое соединение, граф собран заново.
with SqliteSaver.from_conn_string(str(DB_PATH)) as checkpointer:
    graph = build(checkpointer)
    snapshot = graph.get_state(CONFIG)
    print("3. второй день, next:", snapshot.next)
    print("   пауза из файла:", snapshot.interrupts[0].value["вопрос"])

    # 4. invoke(None, CONFIG) не отвечает на паузу, узел снова встаёт на неё.
    CALLS.clear()
    result = graph.invoke(None, CONFIG)
    print("4. invoke(None): выполнены узлы", CALLS, "пауза:", "__interrupt__" in result)

    # 5. Ответ человека.
    CALLS.clear()
    result = graph.invoke(Command(resume="app.py"), CONFIG)
    print("5. Command(resume='app.py'): выполнены узлы", CALLS)
    print("   выполнение файла:", result["output"])

    # 6. Второй ответ законченному запуску узлов не выполняет.
    CALLS.clear()
    result = graph.invoke(Command(resume="utils.py"), CONFIG)
    print("6. повторный ответ: выполнены узлы", CALLS, "выбран", result["chosen"])

    # 7. Повтор от checkpoint с паузой (урок 5): прежний ответ не используется.
    paused = next(
        item
        for item in graph.get_state_history(CONFIG)
        if any(task.interrupts for task in item.tasks)
    )
    CALLS.clear()
    result = graph.invoke(None, paused.config)
    print("7. повтор: выполнены узлы", CALLS, "пауза:", "__interrupt__" in result)
    result = graph.invoke(Command(resume="utils.py"), CONFIG)
    print("   новый ответ: выбран", result["chosen"], "|", result["output"])
remove_db()

# Вывод:
# 1. без чекпойнтера, пауза в результате: True
#    возобновление: Cannot use Command(resume=...) without checkpointer
# 2. первый день, пауза: Какой файл выполнить?
# 3. второй день, next: ('choose_file',)
#    пауза из файла: Какой файл выполнить?
# 4. invoke(None): выполнены узлы ['choose_file'] пауза: True
# 5. Command(resume='app.py'): выполнены узлы ['choose_file', 'run_file']
#    выполнение файла: код 0, вывод: [('select count(*) from orders', 3)]
# 6. повторный ответ: выполнены узлы [] выбран app.py
# 7. повтор: выполнены узлы ['choose_file'] пауза: True
#    новый ответ: выбран utils.py | код 0, вывод: -

"""Пример 3 урока 6: одобрить, исправить или отклонить выполнение файла.

Узел plan_run предлагает выполнить db.py. Узел review_run показывает
предложение человеку, LangGraph проверяет ответ по схеме Decision, и узел по
решению возвращает Command: approve ведёт в run_file как есть, edit меняет
файл и ведёт в run_file, reject записывает причину и ведёт в skip_run. Три
решения идут в трёх разных thread. В четвёртом thread первый ответ не
проходит проверку по схеме, и человек отвечает на ту же паузу ещё раз.
"""

import subprocess
import sys
from pathlib import Path
from typing import Literal

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
from pydantic import BaseModel, ValidationError
from typing_extensions import TypedDict

REPO = Path(__file__).parent / "mini_repo"


class RunState(TypedDict):
    root: str
    chosen: str
    note: str
    output: str


class Decision(BaseModel):
    """Ответ человека: решение и поля, которые оно использует."""

    type: Literal["approve", "edit", "reject"]
    file: str | None = None
    reason: str = "без причины"


def plan_run(state: RunState) -> dict:
    """Предлагает файл для выполнения. В проекте это решение модели."""
    return {"chosen": "db.py"}


def review_run(state: RunState) -> Command[Literal["run_file", "skip_run"]]:
    """Ставит паузу с предложением и переходит по решению человека."""
    decision = interrupt(
        {"действие": "выполнение файла", "файл": state["chosen"]},
        response_schema=Decision,
    )
    if decision.type == "approve":
        return Command(goto="run_file")
    if decision.type == "edit":
        return Command(update={"chosen": decision.file}, goto="run_file")
    return Command(update={"note": decision.reason}, goto="skip_run")


def run_file(state: RunState) -> dict:
    """Выполняет файл стенда отдельным процессом."""
    done = subprocess.run(
        [sys.executable, state["chosen"]],
        cwd=state["root"],
        capture_output=True,
        text=True,
        encoding="utf-8",
    )
    return {"output": f"код {done.returncode}, вывод: {done.stdout.strip() or '-'}"}


def skip_run(state: RunState) -> dict:
    """Записывает отказ вместо выполнения файла."""
    return {"output": f"не выполнен: {state['note']}"}


builder = StateGraph(RunState)
builder.add_node("plan_run", plan_run)
builder.add_node("review_run", review_run)
builder.add_node("run_file", run_file)
builder.add_node("skip_run", skip_run)
builder.add_edge(START, "plan_run")
builder.add_edge("plan_run", "review_run")
builder.add_edge("run_file", END)
builder.add_edge("skip_run", END)
graph = builder.compile(checkpointer=InMemorySaver())

START_STATE = {"root": str(REPO), "chosen": "", "note": "", "output": ""}
DECISIONS = {
    "review-approve": {"type": "approve"},
    "review-edit": {"type": "edit", "file": "app.py"},
    "review-reject": {"type": "reject", "reason": "db.py ходит в базу"},
}

for number, (thread_id, decision) in enumerate(DECISIONS.items(), start=1):
    config = {"configurable": {"thread_id": thread_id}}
    paused = graph.invoke(START_STATE, config)
    result = graph.invoke(Command(resume=decision), config)
    print(f"{number}. {decision['type']}")
    print("   человеку показано:", paused["__interrupt__"][0].value)
    print("   файл:", result["chosen"], "|", result["output"])

# 4. Схема ответа приходит вместе с паузой, в поле response_schema.
schema = paused["__interrupt__"][0].response_schema
print("4. решения в схеме:", schema["properties"]["type"]["enum"])
print("   обязательные поля:", schema["required"])

# 5. Ответ не по схеме: ValidationError, пауза остаётся на месте.
config = {"configurable": {"thread_id": "review-invalid"}}
graph.invoke(START_STATE, config)
try:
    graph.invoke(Command(resume={"type": "approved"}), config)
except ValidationError as error:
    problem = error.errors()[0]
    print("5. ошибка в поле", problem["loc"], "|", problem["msg"])
snapshot = graph.get_state(config)
print("   next:", snapshot.next, "| пауз:", len(snapshot.interrupts))
try:
    graph.invoke(None, config)
except ValidationError as error:
    print("   invoke(None) после неверного ответа:", type(error).__name__)
result = graph.invoke(Command(resume={"type": "approve"}), config)
print("   исправленный ответ: файл", result["chosen"], "|", result["output"])

# Вывод:
# 1. approve
#    человеку показано: {'действие': 'выполнение файла', 'файл': 'db.py'}
#    файл: db.py | код 0, вывод: -
# 2. edit
#    человеку показано: {'действие': 'выполнение файла', 'файл': 'db.py'}
#    файл: app.py | код 0, вывод: [('select count(*) from orders', 3)]
# 3. reject
#    человеку показано: {'действие': 'выполнение файла', 'файл': 'db.py'}
#    файл: db.py | не выполнен: db.py ходит в базу
# 4. решения в схеме: ['approve', 'edit', 'reject']
#    обязательные поля: ['type']
# 5. ошибка в поле ('type',) | Input should be 'approve', 'edit' or 'reject'
#    next: ('review_run',) | пауз: 1
#    invoke(None) после неверного ответа: ValidationError
#    исправленный ответ: файл db.py | код 0, вывод: -

"""Пример 4 урока 6: несколько пауз в одном шаге.

Send запускает по копии узла ask_and_run на каждый файл стенда, все копии
выполняются одним шагом, и каждая останавливается на своей паузе. Одно
значение в Command(resume=...) на четыре паузы не подходит. Ответ даётся
словарём: id паузы и ответ на неё. Сначала человек отвечает на две паузы,
потом на оставшиеся.
"""

import operator
import subprocess
import sys
from pathlib import Path
from typing import Annotated

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, Send, interrupt
from typing_extensions import TypedDict

REPO = Path(__file__).parent / "mini_repo"
CONFIG = {"configurable": {"thread_id": "many-1"}}
CALLS: list[str] = []


class RunState(TypedDict):
    root: str
    files: list[str]
    runs: Annotated[list[str], operator.add]


class FileState(TypedDict):
    """Состояние одной копии узла ask_and_run: свой файл и корень стенда."""

    root: str
    name: str


def collect_files(state: RunState) -> dict:
    """Собирает имена файлов стенда."""
    names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
    return {"files": names}


def fan_out(state: RunState) -> list[Send]:
    """По одному Send на файл."""
    return [Send("ask_and_run", {"root": state["root"], "name": name}) for name in state["files"]]


def ask_and_run(state: FileState) -> dict:
    """Ставит паузу с вопросом о своём файле и выполняет файл при ответе "да"."""
    CALLS.append(state["name"])
    answer = interrupt({"вопрос": "Выполнить файл?", "файл": state["name"]})
    if answer != "да":
        return {"runs": [f"{state['name']}: пропущен"]}
    done = subprocess.run(
        [sys.executable, state["name"]],
        cwd=state["root"],
        capture_output=True,
        text=True,
        encoding="utf-8",
    )
    return {"runs": [f"{state['name']}: код {done.returncode}"]}


builder = StateGraph(RunState)
builder.add_node("collect_files", collect_files)
builder.add_node("ask_and_run", ask_and_run, input_schema=FileState)
builder.add_edge(START, "collect_files")
builder.add_conditional_edges("collect_files", fan_out, ["ask_and_run"])
builder.add_edge("ask_and_run", END)
graph = builder.compile(checkpointer=InMemorySaver())

# 1. Четыре копии, четыре паузы.
result = graph.invoke({"root": str(REPO), "files": [], "runs": []}, CONFIG)
pauses = result["__interrupt__"]
print("1. пауз:", len(pauses), [pause.value["файл"] for pause in pauses])

# 2. Одно значение на несколько пауз не принимается.
try:
    graph.invoke(Command(resume="да"), CONFIG)
except RuntimeError as error:
    print("2.", str(error).split(".")[0])

# 3. Ответ на две паузы из четырёх: словарь id паузы -> ответ.
by_file = {pause.value["файл"]: pause.id for pause in pauses}
CALLS.clear()
result = graph.invoke(Command(resume={by_file["app.py"]: "да", by_file["config.py"]: "нет"}), CONFIG)
print("3. выполнены копии:", sorted(CALLS))
print("   готово:", result["runs"])
print("   пауз в результате:", len(result["__interrupt__"]))
print("   пауз в get_state:", len(graph.get_state(CONFIG).interrupts))

# 4. Ответ на оставшиеся паузы по id из последнего результата.
CALLS.clear()
rest = {pause.id: "нет" for pause in result["__interrupt__"]}
result = graph.invoke(Command(resume=rest), CONFIG)
print("4. выполнены копии:", sorted(CALLS))
print("   готово:", result["runs"])

# Вывод:
# 1. пауз: 4 ['app.py', 'config.py', 'db.py', 'utils.py']
# 2. When there are multiple pending interrupts, you must specify the interrupt id when resuming
# 3. выполнены копии: ['app.py', 'config.py', 'db.py', 'utils.py']
#    готово: ['app.py: код 0', 'config.py: пропущен']
#    пауз в результате: 2
#    пауз в get_state: 2
# 4. выполнены копии: ['db.py', 'utils.py']
#    готово: ['app.py: код 0', 'config.py: пропущен', 'db.py: пропущен', 'utils.py: пропущен']

"""Пример 5 урока 6: пауза внутри инструмента.

Инструмент run_file сам ставит паузу вызовом interrupt() и только после
ответа "да" выполняет файл. Узел run_tools выполняет заранее заданный
список вызовов инструментов по порядку. Два вызова в одном узле дают две
паузы подряд в одной задаче. Список EXECUTED показывает, сколько раз файл
выполнялся на самом деле.
"""

import subprocess
import sys
from pathlib import Path

from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
from typing_extensions import TypedDict

REPO = Path(__file__).parent / "mini_repo"
CONFIG = {"configurable": {"thread_id": "tools-1"}}
EXECUTED: list[str] = []


@tool
def run_file(name: str) -> str:
    """Выполняет файл стенда отдельным процессом после ответа "да" на паузе."""
    answer = interrupt({"инструмент": "run_file", "файл": name})
    if answer != "да":
        return f"{name}: отказ"
    EXECUTED.append(name)
    done = subprocess.run(
        [sys.executable, name],
        cwd=REPO,
        capture_output=True,
        text=True,
        encoding="utf-8",
    )
    return f"{name}: код {done.returncode}"


TOOLS = {"run_file": run_file}


class ToolState(TypedDict):
    tool_calls: list[dict]
    results: list[str]


def run_tools(state: ToolState) -> dict:
    """Выполняет вызовы инструментов по порядку."""
    results = [TOOLS[call["name"]].invoke(call["args"]) for call in state["tool_calls"]]
    return {"results": results}


builder = StateGraph(ToolState)
builder.add_node("run_tools", run_tools)
builder.add_edge(START, "run_tools")
builder.add_edge("run_tools", END)
graph = builder.compile(checkpointer=InMemorySaver())

TOOL_CALLS = [
    {"name": "run_file", "args": {"name": "app.py"}},
    {"name": "run_file", "args": {"name": "db.py"}},
]

# 1. Первый вызов инструмента останавливает узел.
result = graph.invoke({"tool_calls": TOOL_CALLS, "results": []}, CONFIG)
first = result["__interrupt__"][0]
print("1. пауза:", first.value, "| выполнены файлы:", EXECUTED)

# 2. Ответ на первую паузу: app.py выполняется, второй вызов встаёт на паузу.
result = graph.invoke(Command(resume="да"), CONFIG)
second = result["__interrupt__"][0]
print("2. пауза:", second.value, "| выполнены файлы:", EXECUTED)
print("   id пауз совпадают:", first.id == second.id)

# 3. Ответ на вторую паузу: узел снова выполняется с начала.
result = graph.invoke(Command(resume="нет"), CONFIG)
print("3. результаты:", result["results"])
print("   выполнены файлы:", EXECUTED)

# Вывод:
# 1. пауза: {'инструмент': 'run_file', 'файл': 'app.py'} | выполнены файлы: []
# 2. пауза: {'инструмент': 'run_file', 'файл': 'db.py'} | выполнены файлы: ['app.py']
#    id пауз совпадают: True
# 3. результаты: ['app.py: код 0', 'db.py: отказ']
#    выполнены файлы: ['app.py', 'app.py']

"""Пример 6 урока 6: четыре нарушения правил паузы.

Каждое нарушение стоит в своём узле, узел собирается в граф из одного узла
и запускается в своём thread:
1) interrupt() внутри try/except Exception;
2) первая пауза задаётся по условию, которое меняется между вызовами;
3) в значении паузы лежит функция;
4) запись в журнал стоит до паузы.
"""

import operator
from typing import Annotated

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
from typing_extensions import TypedDict

ASK_FILE = {"value": True}
JOURNAL: list[str] = []


class AskState(TypedDict):
    answers: Annotated[list[str], operator.add]


def build(node):
    """Граф из одного узла с чекпойнтером."""
    builder = StateGraph(AskState)
    builder.add_node(node.__name__, node)
    builder.add_edge(START, node.__name__)
    builder.add_edge(node.__name__, END)
    return builder.compile(checkpointer=InMemorySaver())


def ask_inside_try(state: AskState) -> dict:
    """Нарушение 1: except Exception перехватывает остановку."""
    try:
        answer = interrupt("Выполнить app.py?")
    except Exception as error:
        answer = f"перехвачено {type(error).__name__}"
    return {"answers": [answer]}


def ask_by_condition(state: AskState) -> dict:
    """Нарушение 2: число пауз зависит от флага, который меняется между вызовами."""
    answers = []
    if ASK_FILE["value"]:
        answers.append("файл: " + interrupt("Какой файл выполнить?"))
    answers.append("аргументы: " + interrupt("С какими аргументами?"))
    return {"answers": answers}


def ask_with_function(state: AskState) -> dict:
    """Нарушение 3: в значении паузы функция."""
    answer = interrupt({"вопрос": "Выполнить app.py?", "проверка": str.isdigit})
    return {"answers": [answer]}


def journal_then_ask(state: AskState) -> dict:
    """Нарушение 4: запись в журнал до паузы."""
    JOURNAL.append("запрошено выполнение app.py")
    answer = interrupt("Выполнить app.py?")
    return {"answers": [answer]}


START_STATE = {"answers": []}

# 1. Остановка перехвачена: паузы нет, запуск закончился без ответа человека.
graph = build(ask_inside_try)
config = {"configurable": {"thread_id": "rule-1"}}
result = graph.invoke(START_STATE, config)
print("1. пауза:", "__interrupt__" in result, "| ответы:", result["answers"])

# 2. Между ответами флаг сменился, и ответ ушёл не в тот вопрос.
graph = build(ask_by_condition)
config = {"configurable": {"thread_id": "rule-2"}}
graph.invoke(START_STATE, config)
graph.invoke(Command(resume="app.py"), config)
ASK_FILE["value"] = False
result = graph.invoke(Command(resume="--verbose"), config)
print("2. ответы:", result["answers"])

# 3. Функцию в значении паузы чекпойнтер не сохраняет.
graph = build(ask_with_function)
config = {"configurable": {"thread_id": "rule-3"}}
try:
    graph.invoke(START_STATE, config)
except TypeError as error:
    print("3.", type(error).__name__, error)

# 4. Узел выполнился дважды, запись в журнале задвоилась.
graph = build(journal_then_ask)
config = {"configurable": {"thread_id": "rule-4"}}
graph.invoke(START_STATE, config)
graph.invoke(Command(resume="да"), config)
print("4. журнал:", JOURNAL)

# Вывод:
# 1. пауза: False | ответы: ['перехвачено GraphInterrupt']
# 2. ответы: ['аргументы: app.py']
# 3. TypeError Type is not msgpack serializable: Interrupt
# 4. журнал: ['запрошено выполнение app.py', 'запрошено выполнение app.py']