Approval workflows
для рискованных операций

RISKY_TOOLS = {"delete_files", "send_money", "send_email", "deploy"}

def needs_approval(state: State) -> str:
    """Решает, требует ли запланированный вызов подтверждения."""
    last = state["messages"][-1]
    if not last.tool_calls:
        return "end"                       # агент закончил
    name = last.tool_calls[0]["name"]
    return "approval" if name in RISKY_TOOLS else "execute"   # риск → человек

builder.add_conditional_edges(
    "agent", needs_approval,
    {"approval": "approval", "execute": "execute", "end": END},
)

from langgraph.types import interrupt

def approval(state: State) -> dict:
    call = state["messages"][-1].tool_calls[0]

    # Останавливаемся и показываем человеку, ЧТО собираемся сделать
    decision = interrupt({
        "action": call["name"],
        "args": call["args"],
        "prompt": "Подтвердить выполнение?",
    })
    # decision приходит из Command(resume=...): {"type": "approve|reject|edit", ...}

    audit = {"action": call["name"], "decision": decision["type"]}

    if decision["type"] == "approve":
        return {"approved": True, "audit_log": [audit]}
    if decision["type"] == "edit":
        # человек поправил аргументы — подменяем их в запланированном вызове
        call["args"] = decision["args"]
        return {"approved": True, "audit_log": [audit]}
    # reject
    return {"approved": False, "audit_log": [audit]}

from typing import Annotated, TypedDict
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
from langchain_openai import ChatOpenAI

RISKY = {"delete_files"}

def delete_files(pattern: str) -> str: return f"Удалены: {pattern}"
def search(query: str) -> str:        return f"Результаты: {query}"

class State(TypedDict):
    messages: Annotated[list, add_messages]
    audit_log: Annotated[list, operator.add]   # аудит копится

llm = ChatOpenAI(model="gpt-4o-mini").bind_tools([delete_files, search])

def agent(state): return {"messages": [llm.invoke(state["messages"])]}

def approval(state):
    call = state["messages"][-1].tool_calls[0]
    decision = interrupt({"action": call["name"], "args": call["args"]})
    audit = [{"action": call["name"], "decision": decision["type"]}]
    return {"approved": decision["type"] == "approve", "audit_log": audit}

def execute(state):
    call = state["messages"][-1].tool_calls[0]
    fn = {"delete_files": delete_files, "search": search}[call["name"]]
    out = fn(**call["args"])
    return {"messages": [{"role": "tool", "content": out,
                          "tool_call_id": call["id"]}]}

def after_agent(state):
    last = state["messages"][-1]
    if not last.tool_calls: return "end"
    return "approval" if last.tool_calls[0]["name"] in RISKY else "execute"

def after_approval(state):
    return "execute" if state.get("approved") else "agent"

builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_node("approval", approval)
builder.add_node("execute", execute)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", after_agent,
    {"approval": "approval", "execute": "execute", "end": END})
builder.add_conditional_edges("approval", after_approval,
    {"execute": "execute", "agent": "agent"})
builder.add_edge("execute", "agent")

graph = builder.compile(checkpointer=MemorySaver())   # checkpointer обязателен
cfg = {"configurable": {"thread_id": "ops-1"}}

# Безопасный search выполнится сам; delete_files встанет на approval
graph.invoke({"messages": [{"role": "user", "content": "Удали *.tmp"}]}, cfg)
print(graph.get_state(cfg).next)                      # ('approval',)

# Человек одобряет
graph.invoke(Command(resume={"type": "approve"}), cfg)
print(graph.get_state(cfg).values["audit_log"])       # [{'action': 'delete_files', ...}]

from datetime import datetime, timezone

class State(TypedDict):
    messages: Annotated[list, add_messages]
    audit_log: Annotated[list, operator.add]    # неуничтожимый журнал

def approval(state) -> dict:
    call = state["messages"][-1].tool_calls[0]
    decision = interrupt({"action": call["name"], "args": call["args"]})
    record = {
        "ts": datetime.now(timezone.utc).isoformat(),
        "action": call["name"],
        "args": call["args"],
        "decision": decision["type"],
        "by": decision.get("user", "unknown"),     # кто решил
    }
    return {"approved": decision["type"] == "approve", "audit_log": [record]}

# Пять стадий воркфлоу подтверждения:

# 1. КЛАССИФИКАЦИЯ риска — условное ребро (только опасное → на человека)
def after_agent(state):
    name = state["messages"][-1].tool_calls[0]["name"]
    return "approval" if name in RISKY else "execute"

# 2. ПАУЗА — узел approval через interrupt() с payload
def approval(state):
    call = state["messages"][-1].tool_calls[0]
    d = interrupt({"action": call["name"], "args": call["args"]})
    return {"approved": d["type"] == "approve", "audit_log": [record(call, d)]}

# 3. РЕШЕНИЕ — ещё одно условное ребро по исходу
def after_approval(state):
    return "execute" if state["approved"] else "agent"   # approve/edit vs reject

# 4. АУДИТ — поле-журнал с operator.add (кто/что/когда/решение)
audit_log: Annotated[list, operator.add]

# 5. checkpointer ОБЯЗАТЕЛЕН; resume:
graph.invoke(Command(resume={"type": "approve"}), cfg)   # interrupt()

# Правила:
#  • маршрутизируй по риску, не подтверждай всё
#  • обрабатывай ВСЕ исходы (approve/reject/edit)
#  • аудит — в состоянии (operator.add)
#  • право одобрять проверяй на сервере, не из resume