1 Introduction
In recent years, AI agents have been widely integrated into various business processes, fundamentally changing the way decisions are made. AI-driven systems can automatically perform routine tasks, saving time and resources. However, when dealing with complex decisions, such as expense approvals or customer service inquiries, human judgment remains indispensable. This is where the Human-in-the-Loop (HITL) approach plays a crucial role, providing a hybrid model that combines the speed and accuracy of AI with human nuanced understanding.
In this article, we will explore a real-world use case that demonstrates how to combine AI agents with human decision-making in the expense approval process. By leveraging AI to automatically review expense requests and introducing human intervention when necessary for complex decisions, organizations can create an approval system that is both streamlined and flexible.
2 Problem: Automating Expense Approvals with Human Intervention
In most organizations, managing expense reports is an essential yet time-consuming task. While AI agents can easily approve small expenses, larger and more complex expenses often require human judgment to assess the legitimacy of requests. AI can expedite the process by handling simple requests, but human approval is still needed in cases of ambiguity or additional context.
Consider the following scenario: an employee submits an expense report, and the AI agent first reviews the request. If the amount is small enough, the AI will automatically approve the expense; if the amount is higher, it will escalate the request for human review. This way, the AI agent can handle routine tasks and save time, while humans intervene when necessary, providing crucial insights and final approval.
In LangGraph, building a Human-in-the-Loop workflow relies on two core features: Interrupt and Command. The interrupt feature strategically pauses the execution of the graph to present key information to humans for input, enabling real-time approvals, data corrections, or gathering additional insights. Once humans provide input, the Command object intervenes, not only updating the graph’s state with this new information but also dynamically controlling the flow of the graph to guide it to the next appropriate node. Interrupt and Command together empower developers to design AI workflows that are not only automated but also highly interactive and responsive to human guidance, unlocking new possibilities for dynamic, multi-agent collaboration and adaptive task execution. By leveraging these features, LangGraph simplifies the process of integrating human expertise into complex workflows, ensuring greater accuracy and control.
3 Solution: AI-Driven Workflows with Human-in-the-Loop
This solution involves using LangGraph to create a StateGraph to model the expense approval process. LangGraph is a powerful tool that supports the design of workflows using both AI agents and human decisions.
Here’s a breakdown of the process:
- AI Review of Expenses
: The AI agent first reviews the expense request. If the expense is below a predefined threshold, it automatically approves the request; for higher expenses, the AI sends the request for human approval. - Human Approval Node
: If the AI determines that human intervention is needed, it directs the flow to the human approval node. Here, the user (the human in the loop) is asked to approve, reject, or comment on the expense request. This interaction ensures human involvement when necessary while keeping the overall process automated. - Status Management
: The entire process is managed through State, a dictionary that stores the current status of the request, including employee name, expense amount, reason for the request, and approval status. The AI agent can update this status as it moves through the workflow.
4 Code Exercise
To better illustrate this concept, below is a simplified version of the code that demonstrates how to model the approval process using LangGraph’s StateGraph and introduce the Human-in-the-Loop mechanism for user feedback before final approval.
import os
from typing import TypedDict, Annotated
import uuid
from langchain_openai import AzureChatOpenAI
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langgraph.checkpoint.memory import MemorySaver
from langgraph.constants import START
from langgraph.graph import StateGraph, add_messages
from langgraph.types import interrupt, Command
# Load environment variables from .env file
load_dotenv()
# Initialize FastAPI application
app = FastAPI(
title="LangGraph Expense Approval API",
version="1.0.0",
contact={
"name": "Sreeni",
"email": "[email protected]",
}
)
# Initialize Azure OpenAI LLM for conversational tasks
llm = AzureChatOpenAI(
azure_deployment="gpt-4o-mini",
api_version="2024-08-01-preview",
temperature=0,
max_tokens=None,
timeout=None,
max_retries=2
)
# Define Pydantic model for expense requests
class ExpenseRequest(BaseModel):
"""
Pydantic model for receiving expense data.
Attributes:
employee_name (str): Name of the employee submitting the expense.
expense_amount (float): Amount of the expense.
expense_reason (str): Reason for the expense.
"""
employee_name: str
expense_amount: float
expense_reason: str
# Define state type for the state graph
class State(TypedDict):
"""
State graph containing expense details and approval status.
Attributes:
request_id (str): Unique identifier for the expense request.
employee_name (str): Name of the employee submitting the expense.
expense_amount (float): Amount of the expense.
expense_reason (str): Reason for the expense.
approval_status (list[str]): List of approval statuses and comments.
"""
request_id: str
employee_name: str
expense_amount: float
expense_reason: str
approval_status: Annotated[list[str], add_messages]
# Function to review expense requests with AI
def review_expense(state: State):
"""
Preliminary AI review of the expense request.
Args:
state (State): Current state of the expense request.
Returns:
dict or Command: Updated state or command to move to the next node.
"""
print("\n[review_expense] Reviewing expense request...")
expense_amount = state["expense_amount"]
if expense_amount <= 50: # Automatically approve if expense is less than or equal to $50
approval_status = "Auto Approved"
print(f"[review_expense] Approval status: {approval_status}\n")
state["approval_status"].append(approval_status)
return Command(update={"approval_status": state["approval_status"]}, goto="end_node")
approval_status = "Needs Human Review" # Requires human review
print(f"[review_expense] Approval status: {approval_status}\n")
state["approval_status"].append(approval_status)
return {"approval_status": state["approval_status"]}
# Function for human approval node
def human_approval_node(state: State):
"""
Human intervention node for approving or rejecting expense requests.
Args:
state (State): Current state of the expense request.
Returns:
Command: Command to update the state and move to the next node.
"""
print("\n[human_approval_node] Waiting for human approval...")
approval_status = state["approval_status"]
user_feedback = interrupt(
{"approval_status": approval_status, "message": "Approve, reject, or provide comments."})
print(f"[human_approval_node] Received human feedback: {user_feedback}")
if user_feedback.lower() in ["approve", "approved"]: # If human approves
return Command(update={"approval_status": state["approval_status"] + ["Final Approved"]}, goto="end_node")
elif user_feedback.lower() in ["reject", "rejected"]: # If human rejects
return Command(update={"approval_status": state["approval_status"] + ["Final Rejected"]}, goto="end_node")
return Command(update={"approval_status": state["approval_status"] + [user_feedback]}, goto="review_expense")
# Function for the end node of the approval process
def end_node(state: State):
"""
Final node of the approval process.
Args:
state (State): Current state of the expense request.
Returns:
dict: Final approval status.
"""
print("\n[end_node] Process completed.")
print("Final approval status:", state["approval_status"][-1])
return {"approval_status": state["approval_status"]}
# Build state graph
graph_builder = StateGraph(State)
graph_builder.add_node("review_expense", review_expense) # Add AI review node
graph_builder.add_node("human_approval_node", human_approval_node) # Add human approval node
graph_builder.add_node("end_node", end_node) # Add end node
# Define the workflow
graph_builder.add_edge(START, "review_expense") # Start with AI review
graph_builder.add_edge("review_expense", "human_approval_node") # AI review leads to human approval
graph_builder.add_edge("human_approval_node", "review_expense") # Human approval may return to AI review
graph_builder.add_edge("human_approval_node", "end_node") # Human approval ends the process
# Set end point
graph_builder.set_finish_point("end_node")
# Enable interrupt mechanism
checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)
# Optional: Save the state graph as an image
image = graph.get_graph().draw_mermaid_png()
with open("expense_approval_graph.png", "wb") as file:
file.write(image)
# Define API endpoint for submitting expense requests
@app.post("/submit-expense/")
def submit_expense(expense: ExpenseRequest):
"""
API endpoint for submitting expense requests.
Args:
expense (ExpenseRequest): Expense request data.
Returns:
dict: Request ID and final approval status.
"""
# Print details of the expense request
print("\n--- Expense Request Details ---")
print(f"Employee Name: {expense.employee_name}")
print(f"Expense Amount: ${expense.expense_amount:.2f}")
print(f"Expense Reason: {expense.expense_reason}")
print("-------------------------------\n")
request_id = str(uuid.uuid4()) # Generate unique request ID
initial_state = {
"request_id": request_id,
"employee_name": expense.employee_name,
"expense_amount": expense.expense_amount,
"expense_reason": expense.expense_reason,
"approval_status": []
}
thread_config = {"configurable": {"thread_id": uuid.uuid4()}} # Configure thread ID
final_state = initial_state
for chunk in graph.stream(initial_state, config=thread_config): # Stream the state graph
for node_id, value in chunk.items():
if node_id == "review_expense": # If current node is AI review
final_state.update(value)
if "Auto Approved" in final_state["approval_status"]: # If auto approved
print("[submit_expense] Automatically approved, skipping human approval.")
return {"request_id": request_id, "approval_status": final_state["approval_status"]}
elif node_id == "__interrupt__": # If interrupt mechanism is triggered
while True:
user_feedback = input("Approve, reject, or provide comments:")
final_state["approval_status"] = final_state.get("approval_status", []) + [user_feedback]
graph.invoke(Command(resume=user_feedback), config=thread_config)
if user_feedback.lower() in ["approve", "approved", "reject", "rejected"]: # If human makes a final decision
break
else:
final_state.update(value)
return {"request_id": request_id, "approval_status": final_state["approval_status"]}
# Run FastAPI application
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)
5 Process Interaction Demonstration
5.1 Expense Submission API
5.2 Input I (Auto Approval)
5.3 Input II (Human Approval)
5.4 Waiting for Human Approval, as Shown Below
5.5 I Reject and Complete the Workflow
6 Key Points
- AI Autonomy
: AI can autonomously handle routine tasks, such as approving low-value expenses. By streamlining decision-making in simple cases, it saves valuable time for organizations. - Human Intervention
: When AI cannot make a decision or a request requires more nuanced understanding, it hands control over to human approval, ensuring that critical decisions are handled appropriately. - Seamless Integration
: The combination of AI agents and human feedback creates a flexible and adaptive system. Humans are brought into the loop when necessary without hindering the overall efficiency of the process.
7 Conclusion
In summary, integrating AI with human decision-makers is an effective way to automate repetitive tasks while ensuring that complex decisions benefit from human judgment. By using AI agents in conjunction with human oversight, organizations can enhance efficiency, reduce errors, and streamline workflows.