Interrupts

from langgraph.graph import END, START, StateGraph
from langgraph.types import Interrupt, Command
from langchain_openai import ChatOpenAI
from typing_extensions import TypedDict, Annotated
from operator import add
 
llm = ChatOpenAI(model="gpt-5-mini")
 
class State(TypedDict):
    messages: Annotated[list, add]
    proposed_action: dict
    approval_result: dict
 
def plan_action(state: State) -> dict:
    """Agent analyzes the request and proposes an action."""
    response = llm.invoke([
        {"role": "system", "content": (
            "Analyze the user's request. If it requires sending "
            "an email, modifying data, or any irreversible action, "
            "return a JSON action plan with keys: action, target, "
            "description, risk_level."
        )},
        *state["messages"]
    ])
    action = parse_json(response.content)
    return {
        "proposed_action": action,
        "messages": [response],
    }
 
def request_approval(state: State) -> dict:
    """Pause the graph and ask the human for approval."""
    action = state["proposed_action"]
    raise Interrupt(value={
        "action": action["action"],
        "target": action["target"],
        "description": action["description"],
        "risk_level": action.get("risk_level", "medium"),
    })
 
def execute_action(state: State) -> dict:
    """Run the approved action or explain the rejection."""
    result = state.get("approval_result", {})
    if result.get("approved"):
        # Execute the real action
        outcome = perform_action(state["proposed_action"])
        return {
            "messages": [{"role": "assistant", "content": (
                f"Done. {outcome}"
            )}]
        }
    else:
        reason = result.get("reason", "No reason given")
        return {
            "messages": [{"role": "assistant", "content": (
                f"Action cancelled. Reason: {reason}"
            )}]
        }
 
# Build the graph: plan → approve → execute
builder = StateGraph(State)
builder.add_node("plan", plan_action)
builder.add_node("approve", request_approval)
builder.add_node("execute", execute_action)
builder.add_edge(START, "plan")
builder.add_edge("plan", "approve")
builder.add_edge("approve", "execute")
builder.add_edge("execute", END)
 
graph = builder.compile()

import {
  Component,
  computed,
  signal,
  ChangeDetectionStrategy,
} from '@angular/core';
import { agent } from '@ngaf/langgraph';
import type { BaseMessage } from '@langchain/core/messages';
 
interface ApprovalPayload {
  action: string;
  target: string;
  description: string;
  risk_level: 'low' | 'medium' | 'high';
}
 
interface AgentState {
  messages: BaseMessage[];
  proposed_action: ApprovalPayload;
  approval_result: { approved: boolean; reason?: string };
}
 
@Component({
  selector: 'app-approval',
  templateUrl: './approval.component.html',
  changeDetection: ChangeDetectionStrategy.OnPush,
})
export class ApprovalComponent {
  agent = agent<AgentState>({
    assistantId: 'approval_agent',
  });
 
  messages = computed(() => this.agent.messages());
  pendingApproval = computed(() => this.agent.interrupt());
  isLoading = computed(() => this.agent.isLoading());
 
  rejectionReason = signal('');
 
  riskClass = computed(() => {
    const interrupt = this.pendingApproval();
    if (!interrupt) return '';
    const level = interrupt.value?.risk_level ?? 'medium';
    return `risk-${level}`;
  });
 
  send(input: string) {
    this.agent.submit({ message: input });
  }
 
  approve() {
    this.agent.submit({
      resume: { approved: true },
    });
  }
 
  reject() {
    this.agent.submit({
      resume: {
        approved: false,
        reason: this.rejectionReason() || 'User rejected',
      },
    });
    this.rejectionReason.set('');
  }
}

from langgraph.graph import END, START, StateGraph
from langgraph.types import Interrupt
from typing_extensions import TypedDict, Annotated
from operator import add
 
class DeployState(TypedDict):
    messages: Annotated[list, add]
    plan: list[dict]
    current_step: int
    completed_steps: list[str]
 
def create_plan(state: DeployState) -> dict:
    """Generate a multi-step deployment plan."""
    plan = [
        {"step": "backup", "description": "Back up current database"},
        {"step": "migrate", "description": "Run schema migrations"},
        {"step": "deploy", "description": "Deploy new application version"},
    ]
    return {"plan": plan, "current_step": 0}
 
def approve_step(state: DeployState) -> dict:
    """Interrupt for each step that needs approval."""
    step_index = state["current_step"]
    step = state["plan"][step_index]
    raise Interrupt(value={
        "step_number": step_index + 1,
        "total_steps": len(state["plan"]),
        "step": step["step"],
        "description": step["description"],
        "completed": state.get("completed_steps", []),
    })
 
def execute_step(state: DeployState) -> dict:
    """Execute the approved step and advance."""
    step = state["plan"][state["current_step"]]
    # ... perform the actual deployment step ...
    return {
        "completed_steps": [step["step"]],
        "current_step": state["current_step"] + 1,
        "messages": [{"role": "assistant", "content": (
            f"Completed: {step['description']}"
        )}],
    }
 
def should_continue(state: DeployState) -> str:
    if state["current_step"] < len(state["plan"]):
        return "approve_step"
    return END
 
builder = StateGraph(DeployState)
builder.add_node("create_plan", create_plan)
builder.add_node("approve_step", approve_step)
builder.add_node("execute_step", execute_step)
builder.add_edge(START, "create_plan")
builder.add_edge("create_plan", "approve_step")
builder.add_edge("approve_step", "execute_step")
builder.add_conditional_edges("execute_step", should_continue)
 
graph = builder.compile()