AI Agent for Business Process Automation

AI Agent Development for Business Process Automation

An AI agent for BPA (Business Process Automation) handles repetitive multi-step tasks: collecting data from multiple systems, making structured decisions by rules, filling documents, routing tasks, sending notifications. Unlike classical RPA, LLM agents understand unstructured content and adapt to input variations.

Typical Business Processes for AI Agent

Incoming request processing: agent parses email/form, classifies, validates data, routes to executor, creates task in tracker.

New employee onboarding: account creation in systems, welcome materials distribution, mentor assignment, initial task setup.

Invoice processing: extract details from PDF/photo, verify against contract, check budget, route for signature, mark in ERP.

Monthly reporting: collect data from multiple sources, calculate metrics, generate report, distribute to stakeholders.

Architecture for Request Processing Agent

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Optional

class ApplicationState(TypedDict):
    raw_input: str              # Incoming request text
    applicant_name: str
    application_type: str       # request type
    extracted_data: dict        # extracted data
    validation_result: dict     # validation result
    routing_decision: str       # where to route
    task_id: Optional[str]      # created task ID
    notification_sent: bool

llm = ChatOpenAI(model="gpt-4o", temperature=0)

def classify_and_extract(state: ApplicationState) -> ApplicationState:
    """Classify request and extract data"""
    response = llm.invoke(f"""Analyze incoming request and extract structured data.

Request:
{state['raw_input']}

Return JSON:
{{
  "application_type": "vacation|expense|equipment|access|other",
  "applicant_name": "...",
  "department": "...",
  "details": {{}},  // type-specific fields
  "urgency": "normal|urgent|critical",
  "missing_info": []  // missing fields
}}""")

    import json
    data = json.loads(response.content)
    return {
        **state,
        "application_type": data["application_type"],
        "applicant_name": data.get("applicant_name", ""),
        "extracted_data": data,
    }

def validate_application(state: ApplicationState) -> ApplicationState:
    """Check completeness and policy compliance"""
    app_type = state["application_type"]
    extracted = state["extracted_data"]

    validation = {"valid": True, "errors": [], "warnings": []}

    if app_type == "vacation":
        # Check vacation balance
        days = extracted["details"].get("days", 0)
        balance = hr_api.get_vacation_balance(state["applicant_name"])
        if days > balance:
            validation["valid"] = False
            validation["errors"].append(f"Insufficient vacation days: requested {days}, available {balance}")

    elif app_type == "expense":
        amount = extracted["details"].get("amount", 0)
        if amount > 50000:  # Self-approval limit
            validation["warnings"].append("Requires manager approval")

    return {**state, "validation_result": validation}

def route_application(state: ApplicationState) -> ApplicationState:
    """Determine processing route"""
    app_type = state["application_type"]
    validation = state["validation_result"]
    urgency = state["extracted_data"].get("urgency", "normal")

    if not validation["valid"]:
        routing = "reject_with_explanation"
    elif app_type == "vacation":
        routing = "hr_manager"
    elif app_type == "expense" and state["extracted_data"]["details"].get("amount", 0) > 50000:
        routing = "director_approval"
    elif app_type == "access":
        routing = "it_department"
    else:
        routing = "auto_approve"

    return {**state, "routing_decision": routing}

def execute_routing(state: ApplicationState) -> ApplicationState:
    """Execute routing"""
    routing = state["routing_decision"]

    if routing == "auto_approve":
        task_id = jira_api.create_task(
            title=f"Auto-approved: {state['application_type']} from {state['applicant_name']}",
            status="Done",
            assignee="system",
        )
    elif routing in ["hr_manager", "director_approval", "it_department"]:
        assignee_map = {
            "hr_manager": "[email protected]",
            "director_approval": "[email protected]",
            "it_department": "[email protected]",
        }
        task_id = jira_api.create_task(
            title=f"Request for {state['application_type']} from {state['applicant_name']}",
            assignee=assignee_map[routing],
            description=json.dumps(state["extracted_data"], ensure_ascii=False),
            priority="High" if state["extracted_data"].get("urgency") == "urgent" else "Normal",
        )
    else:
        task_id = None

    notification_service.send(
        to=state["applicant_name"],
        message=f"Your request accepted. Routing: {routing}. ID: {task_id}"
    )

    return {**state, "task_id": task_id, "notification_sent": True}

# Build process graph
graph = StateGraph(ApplicationState)
graph.add_node("classify_and_extract", classify_and_extract)
graph.add_node("validate", validate_application)
graph.add_node("route", route_application)
graph.add_node("execute", execute_routing)

graph.set_entry_point("classify_and_extract")
graph.add_edge("classify_and_extract", "validate")
graph.add_edge("validate", "route")
graph.add_edge("route", "execute")
graph.add_edge("execute", END)

application_agent = graph.compile()

Practical Case: Incoming Invoice Processing

Task: 180+ invoices per month. Before automation: 3.5 hours/day for chief accountant.

Agent pipeline:

1. Extract text from PDF (pdfplumber / LlamaParse)
2. LLM extracts: supplier, registration number, amount, tax, date, number, contract
3. Verify against contract registry (vector search)
4. Check in ERP: contract balance, budget line
5. If OK → create payment order in ERP
6. If mismatch → task for accountant with explanation

Metrics after 3 months:

• Automatically processed without intervention: 73%
• Information extraction accuracy: 96%
• Errors (wrong contract binding): 1.2%
• Time savings: 2.5 hours/day

Human-in-the-Loop: When Agent Requests Approval

def requires_human_approval(state: ApplicationState) -> bool:
    """Determine if human intervention needed"""
    return (
        not state["validation_result"]["valid"] or
        state["extracted_data"].get("amount", 0) > 100000 or
        state["application_type"] == "termination" or
        state["extracted_data"].get("urgency") == "critical"
    )

# In LangGraph: interrupt_before for HITL
agent = graph.compile(
    interrupt_before=["execute"],  # Interrupt before execution
    checkpointer=MemorySaver(),
)

Timeline

• Business process analysis and design: 1–2 weeks
• Agent development with integrations: 3–5 weeks
• Testing and HITL tuning: 1–2 weeks
• Total: 5–9 weeks