Interrupt and Human-in-the-Loop Workflows

pip install -U langgraph langchain-ollama langgraph-checkpoint-sqlite python-dotenv
ollama pull qwen3

from typing_extensions import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver

from langgraph.types import Command, interrupt

from langchain_ollama import ChatOllama
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode
import os
import re

# Configuration
BASE_URL = "http://localhost:11434"
MODEL_NAME = "qwen3"
llm = ChatOllama(model=MODEL_NAME, base_url=BASE_URL)

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]

@tool
def transfer_money(amount:int, recipient:str):
    """
    Transfer money. Large transfers require approval.

    Args:
        amount: Amount in dollars
        recipient: Recipient name
    """

    # let's ask user approval if money is more than 1000
    if amount > 1000:
        approval = interrupt(
            {
                "type": "approval_required",
                "amount": amount,
                "recipient": recipient
            }
        )

        if approval.get("decision") != "approve":
            return "Transfer is cancelled!"

    return f"Transferred {amount} to {recipient}"

patterns = {
        "SSN": r'\b\d{3}-\d{2}-\d{4}\b',  # SSN: 123-45-6789
        "Credit Card": r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b',  # Credit Card: 1234-5678-9012-3456
        "Mobile Number": r'\b(\+?\d{1,3}[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b',  # Mobile: +1-234-567-8900
        "Email": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',  # Email: user@example.com
        "URL/Link": r'https?://[^\s]+|www\.[^\s]+'  # URL: http://example.com or www.example.com
    }


def guardrail_node(state:AgentState):
    last_message = state['messages'][-1].content

    for pii_type, pattern in patterns.items():
        if re.search(pattern, last_message):
            return {
                "messages": [SystemMessage(f"Request Blocked: Contains {pii_type}. Please don't share sensitive personal information")]
            }
        
    return {"messages": []}

def agent_node(state:AgentState):

    tools = [transfer_money]

    llm_with_tools = llm.bind_tools(tools)

    system_message = SystemMessage("You are a financial assistant. use transfer_money tool to transfer the money.")

    messages = [system_message] + state['messages']

    response = llm_with_tools.invoke(messages)

    if hasattr(response, 'tool_calls') and response.tool_calls:
        for tc in response.tool_calls:
            print(f"[AGENT] called Tool {tc.get('name', '?')} with args {tc.get('args', '?')}")
    else:
        print(f"[AGENT] Responding...")

    
    return {"messages": [response]}

def should_continue(state: AgentState):
    last = state['messages'][-1]
    
    if hasattr(last, 'tool_calls') and last.tool_calls:
        return "tools"
    else:
        return END
    
def guardrail_router(state:AgentState):
    last = state['messages'][-1]

    if isinstance(last, SystemMessage):
        last.pretty_print()
        return  END
    
    return 'agent'

import sqlite3

db_name = "db/checkpoints.db"

os.makedirs('db', exist_ok=True)

def chat(query, thread_id):
    config = {"configurable": {'thread_id': thread_id}}
    result = agent.invoke({
        "messages": [HumanMessage(query)]
    }, config)
    

    if '__interrupt__' in result:
        # use this data in front end design for approval workflow
        interrupt_data = result['__interrupt__'][0]

        result = agent.invoke(
            Command(resume={'decision': input("do you approve or reject?")}),
            config
        )

        
    result['messages'][-1].pretty_print()