In recent years, the integration of AI agents into various business processes has revolutionized how decisions are made. AI-driven systems can automate routine tasks, saving time and resources. However, when it comes to complex decision-making, such as handling expense approvals or customer service queries, human judgment remains essential. This is where the Human-in-the-Loop (HITL) approach becomes crucial, offering a hybrid model that combines the speed and accuracy of AI with the nuanced understanding of humans.
In this blog, we'll explore a real-world use case where we combine AI agents with Human-in-the-Loop decision-making in an expense approval process. By leveraging AI to automatically review expense requests and involving human intervention for complex decisions, organizations can create a streamlined yet flexible approval system.
The Problem: Automating Expense Approvals with Human Intervention
Managing expense reports is an essential but time-consuming task in most organizations. While small expenses can be easily approved by AI agents, larger and more complex expenses often require human judgment to evaluate whether the request is legitimate. AI can assist in speeding up the process by handling straightforward requests, but human approval is needed when there is ambiguity or a need for additional context.
Consider the following scenario: an employee submits an expense report, and the AI agent initially reviews the request. If the amount is small enough, the AI approves the expense automatically. If the amount is higher, the request is escalated for human review. This allows the AI agent to handle routine tasks and save time, while humans step in when necessary to provide critical insights and final approval.
In LangGraph, crafting human-in-the-loop workflows hinges on two core functions: interrupt and Command. The interrupt function strategically pauses graph execution, presenting key information to a human for input, enabling real-time approvals, data corrections, or gathering additional insights. Once the human provides input, the Command object steps in, not only updating the graph's state with this new information but also dynamically controlling the graph's flow, directing it to the next appropriate node. Together, interrupt and Command 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 collaborations and adaptive task execution. By leveraging these functions, LangGraph simplifies integrating human expertise into complex processes, ensuring greater accuracy and control.
The Solution: AI-Powered Workflow with Human-in-the-Loop
The solution involves creating a StateGraph to model the expense approval process using LangGraph, a powerful tool that enables the design of workflows with AI agents and human decision-making.
Here’s a breakdown of the process:
AI Review of Expense: The AI agent first reviews the expense request. If the expense is below a predefined threshold, the request is automatically approved. For higher expenses, the AI sends the request to a human for approval.
Human Approval Node: If the AI deems that human intervention is required, it directs the process to a Human Approval Node. Here, a user (the human in the loop) is asked to approve, reject, or provide comments on the expense request. This interaction ensures that humans are involved when necessary, but the process can still be automated.
State Management: The entire process is managed using State, a dictionary that stores the current status of the request, including the employee’s name, the expense amount, the reason for the request, and the approval status. The AI agent can update this state as it moves through the workflow.
*Expense Approval workflow *
Code Walkthrough
To illustrate this concept, here's a simplified version of the code that powers the workflow. The code utilizes LangGraph’s StateGraph to model the approval process and includes a Human-in-the-Loop mechanism that requires user feedback before final approval.
import os
from typing import TypedDict, Annotated
import uuid
from langchain_openai import AzureChatOpenAI
import openai # OpenAI API for GPT-4
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()
app = FastAPI(
title="Sreeni's LangGraph Expense Approval API",
version="1.0.0",
contact={
"name": "Sreeni",
"email": "sreeniusa@outlook.com",
}
)
# Initialize the Azure OpenAI LLM for conversation-based tasks
llm = AzureChatOpenAI(
azure_deployment="gpt-4o-mini",
api_version="2024-08-01-preview",
temperature=0,
max_tokens=None,
timeout=None,
max_retries=2
)
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
class State(TypedDict):
"""
Graph state 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]
def review_expense(state: State):
"""
Initial AI-based 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 the expense request...")
expense_amount = state["expense_amount"]
if expense_amount <= 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"
print(f"[review_expense] Approval Status: {approval_status}\n")
state["approval_status"].append(approval_status)
return {"approval_status": state["approval_status"]}
def human_approval_node(state: State):
"""
Human intervention node for approving or rejecting the expense request.
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] Awaiting 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"]:
return Command(update={"approval_status": state["approval_status"] + ["Final Approved"]}, goto="end_node")
elif user_feedback.lower() in ["reject", "rejected"]:
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")
def end_node(state: State):
"""
Final node in the approval process.
Args:
state (State): Current state of the expense request.
Returns:
dict: Final approval status.
"""
print("\n[end_node] Process finished.")
print("Final Approval Status:", state["approval_status"][-1])
return {"approval_status": state["approval_status"]}
# Building the Graph
graph_builder = StateGraph(State)
graph_builder.add_node("review_expense", review_expense)
graph_builder.add_node("human_approval_node", human_approval_node)
graph_builder.add_node("end_node", end_node)
# Define the Flow
graph_builder.add_edge(START, "review_expense")
graph_builder.add_edge("review_expense", "human_approval_node")
graph_builder.add_edge("human_approval_node", "review_expense")
graph_builder.add_edge("human_approval_node", "end_node")
# Set the finish point
graph_builder.set_finish_point("end_node")
# Enable Interrupt Mechanism
checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)
# Save the graph as an image (optional)
image = graph.get_graph().draw_mermaid_png()
with open("expense_approval_graph.png", "wb") as file:
file.write(image)
@app.post("/submit-expense/")
@app.post("/submit-expense/")
def submit_expense(expense: ExpenseRequest):
"""
API endpoint to submit an expense request.
Args:
expense (ExpenseRequest): Expense request data.
Returns:
dict: Request ID and final approval status.
"""
# Print all values of the ExpenseRequest
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())
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()}}
final_state = initial_state
for chunk in graph.stream(initial_state, config=thread_config):
for node_id, value in chunk.items():
if node_id == "review_expense":
final_state.update(value)
if "Auto Approved" in final_state["approval_status"]:
print("[submit_expense] Auto-approved. Skipping human approval.")
return {"request_id": request_id, "approval_status": final_state["approval_status"]}
elif node_id == "__interrupt__":
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"]:
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)
Expense Submission API
Input I (Auto Approval)
Input II (Human Approval)
Waiting for Human Approval as shown below
I Rejected and Complete the Workflow
Key Takeaways
AI Autonomy: The AI handles routine tasks, such as approving low-value expenses, automatically. It saves time for organizations by streamlining decision-making in simple cases.
Human Intervention: When the AI cannot make a decision or the request requires nuanced understanding, it passes control to a human for approval, ensuring that critical decisions are handled appropriately.
Seamless Integration: The combination of AI agents and human feedback creates a system that is flexible and adaptive. Humans are brought into the loop when necessary, without hindering the overall efficiency of the process.
In conclusion, integrating AI with human decision-makers is a powerful way to automate repetitive tasks while ensuring that complex decisions benefit from human judgment. By using AI agents in tandem with human oversight, organizations can improve efficiency, reduce errors, and streamline workflows.
Thanks
Sreeni Ramadorai
Top comments (0)