Agentic Workflow Automation: Moving Beyond Single-Task AI to End-to-End Orchestration

import asyncio
import json
from dataclasses import dataclass, field
from typing import Optional
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
import redis


@dataclass
class SupportTicket:
    ticket_id: str
    subject: str
    body: str
    customer_id: str
    channel: str  # email | chat | api
    priority: Optional[str] = None
    category: Optional[str] = None
    assigned_queue: Optional[str] = None


# --- Tools ---

@tool
def get_customer_context(customer_id: str) -> str:
    """Retrieve customer tier, open tickets, and recent interactions from CRM."""
    # Production: call your CRM API here
    mock_data = {
        "tier": "enterprise",
        "open_tickets": 2,
        "last_interaction_days_ago": 3,
        "csm": "sarah.chen@example.com"
    }
    return json.dumps(mock_data)


@tool
def search_knowledge_base(query: str) -> str:
    """Search internal KB for relevant resolution steps."""
    # Production: call your vector DB search here
    return "KB Article #4421: For authentication errors, check SSO config and token expiry."


@tool
def route_ticket(ticket_id: str, queue: str, priority: str, summary: str) -> str:
    """Route ticket to the specified queue with priority and an AI-generated summary."""
    # Production: write to your ticketing system (Zendesk, Jira Service Mgmt, etc.)
    print(f"[ROUTED] Ticket {ticket_id} → queue={queue}, priority={priority}")
    return f"Ticket {ticket_id} routed to {queue} with priority {priority}."


@tool
def escalate_to_csm(ticket_id: str, csm_email: str, reason: str) -> str:
    """Send escalation alert to the assigned Customer Success Manager."""
    print(f"[ESCALATED] Ticket {ticket_id} → {csm_email}: {reason}")
    return f"Escalation sent to {csm_email}."


# --- Agent ---

SYSTEM_PROMPT = """You are a support workflow automation agent.
Given a support ticket, you must:
1. Retrieve customer context to understand their tier and history.
2. Search the knowledge base for relevant resolution information.
3. Classify the ticket category (billing, technical, access, feature-request, other).
4. Set priority: critical (enterprise + production down), high (enterprise other),
   normal (pro tier), low (free tier or feature requests).
5. Route to the correct queue: billing-queue, technical-queue, csm-queue, or general-queue.
6. If enterprise tier and open_tickets > 3, also escalate to the CSM.
Always route the ticket before finishing."""


class TicketAutomationAgent:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
        self.tools = [get_customer_context, search_knowledge_base,
                      route_ticket, escalate_to_csm]
        prompt = ChatPromptTemplate.from_messages([
            ("system", SYSTEM_PROMPT),
            ("human", "Ticket ID: {ticket_id}\nCustomer: {customer_id}\n"
                       "Subject: {subject}\nBody: {body}"),
            MessagesPlaceholder("agent_scratchpad"),
        ])
        agent = create_openai_tools_agent(self.llm, self.tools, prompt)
        self.executor = AgentExecutor(agent=agent, tools=self.tools,
                                      verbose=True, max_iterations=8)

    async def process(self, ticket: SupportTicket) -> dict:
        result = await asyncio.to_thread(
            self.executor.invoke,
            {
                "ticket_id": ticket.ticket_id,
                "customer_id": ticket.customer_id,
                "subject": ticket.subject,
                "body": ticket.body,
            }
        )
        return {"ticket_id": ticket.ticket_id, "output": result["output"]}


# --- Event loop (simulate queue consumption) ---

async def consume_ticket_queue():
    agent = TicketAutomationAgent()
    # Production: replace with Redis BLPOP, SQS, or Pub/Sub consumption
    sample_tickets = [
        SupportTicket(
            ticket_id="TKT-9921",
            subject="SSO login broken since this morning",
            body="All users in our org are getting 401 errors when authenticating via SSO. "
                 "This is blocking our entire team. Please treat as urgent.",
            customer_id="cust_enterprise_447",
            channel="email"
        )
    ]
    results = await asyncio.gather(*[agent.process(t) for t in sample_tickets])
    for r in results:
        print(f"Done: {r}")


if __name__ == "__main__":
    asyncio.run(consume_ticket_queue())