Capstone: Building a Complete Customer Support Platform with Multi-Agent AI

Project Overview and Architecture

This capstone project brings together every skill from the Learn Agentic AI series into a single, deployable customer support platform. The system handles inbound customer messages, routes them to specialized agents, resolves issues using tools, and escalates to humans when confidence is low. By the end, you will have a working platform with a React frontend, a FastAPI backend, a PostgreSQL database, and a multi-agent orchestration layer.

The high-level architecture consists of five layers. The presentation layer is a Next.js chat widget embeddable on any website. The API layer is a FastAPI application exposing REST endpoints for conversations, tickets, and analytics. The orchestration layer is a triage agent that classifies incoming messages and delegates to specialist agents. The tool layer connects agents to your knowledge base, order system, and ticketing database. The monitoring layer tracks agent performance, response times, and escalation rates.

Data Model Design

Start with the database schema. Every conversation gets a record, every message within it gets a record, and every ticket generated from a conversation gets a record.

flowchart LR
    CLIENT(["Client SDK"])
    GW["API Gateway<br/>auth plus rate limit"]
    APP["FastAPI app<br/>handlers and DI"]
    VAL["Pydantic validation"]
    SVC["Service layer<br/>business logic"]
    DB[(Database)]
    QUEUE[(Background queue)]
    OBS[(Tracing)]
    CLIENT --> GW --> APP --> VAL --> SVC
    SVC --> DB
    SVC --> QUEUE
    SVC --> OBS
    SVC --> CLIENT
    style GW fill:#4f46e5,stroke:#4338ca,color:#fff
    style APP fill:#f59e0b,stroke:#d97706,color:#1f2937
    style DB fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b

# models.py
from sqlalchemy import Column, String, Text, DateTime, ForeignKey, Enum, Float
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
import uuid, datetime, enum

class TicketStatus(enum.Enum):
    OPEN = "open"
    IN_PROGRESS = "in_progress"
    RESOLVED = "resolved"
    ESCALATED = "escalated"

class Conversation(Base):
    __tablename__ = "conversations"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    customer_email = Column(String(255), nullable=False, index=True)
    channel = Column(String(50), default="web")
    started_at = Column(DateTime, default=datetime.datetime.utcnow)
    messages = relationship("Message", back_populates="conversation")

class Message(Base):
    __tablename__ = "messages"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    conversation_id = Column(UUID(as_uuid=True), ForeignKey("conversations.id"))
    role = Column(String(20))  # "user", "assistant", "system"
    content = Column(Text, nullable=False)
    agent_name = Column(String(100), nullable=True)
    confidence = Column(Float, nullable=True)
    created_at = Column(DateTime, default=datetime.datetime.utcnow)
    conversation = relationship("Conversation", back_populates="messages")

class Ticket(Base):
    __tablename__ = "tickets"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    conversation_id = Column(UUID(as_uuid=True), ForeignKey("conversations.id"))
    status = Column(Enum(TicketStatus), default=TicketStatus.OPEN)
    category = Column(String(100))
    summary = Column(Text)
    assigned_to = Column(String(255), nullable=True)
    created_at = Column(DateTime, default=datetime.datetime.utcnow)

Multi-Agent Orchestration Layer

The orchestration layer uses a triage agent that classifies the customer intent and hands off to the appropriate specialist. Each specialist agent has access to domain-specific tools.

# agents/orchestrator.py
from agents import Agent, Runner, handoff, function_tool

@function_tool
def search_knowledge_base(query: str) -> str:
    """Search the FAQ and documentation knowledge base."""
    results = kb_client.search(query, top_k=3)
    return "\n".join([r["content"] for r in results])

@function_tool
def lookup_order(order_id: str) -> str:
    """Look up order status by order ID."""
    order = db.query(Order).filter(Order.id == order_id).first()
    if not order:
        return "Order not found."
    return f"Order {order.id}: status={order.status}, shipped={order.shipped_at}"

@function_tool
def create_ticket(category: str, summary: str) -> str:
    """Create a support ticket for human review."""
    ticket = Ticket(category=category, summary=summary)
    db.add(ticket)
    db.commit()
    return f"Ticket {ticket.id} created."

faq_agent = Agent(
    name="FAQ Agent",
    instructions="Answer customer questions using the knowledge base. Be concise.",
    tools=[search_knowledge_base],
)

order_agent = Agent(
    name="Order Agent",
    instructions="Help customers with order status, returns, and shipping.",
    tools=[lookup_order],
)

escalation_agent = Agent(
    name="Escalation Agent",
    instructions="Create a ticket for issues that need human review.",
    tools=[create_ticket],
)

triage_agent = Agent(
    name="Triage Agent",
    instructions="""Classify the customer message and route:
    - FAQ/general questions -> FAQ Agent
    - Order/shipping/returns -> Order Agent
    - Complaints, billing disputes, complex issues -> Escalation Agent""",
    handoffs=[handoff(faq_agent), handoff(order_agent), handoff(escalation_agent)],
)

API Layer with FastAPI

The API exposes a single chat endpoint that creates or continues a conversation.

# api/routes.py
from fastapi import APIRouter, Depends
from agents import Runner

router = APIRouter()

@router.post("/conversations/{conv_id}/messages")
async def send_message(conv_id: str, body: MessageRequest, db=Depends(get_db)):
    conversation = db.query(Conversation).get(conv_id)
    history = [{"role": m.role, "content": m.content} for m in conversation.messages]

    user_msg = Message(conversation_id=conv_id, role="user", content=body.content)
    db.add(user_msg)

    result = await Runner.run(triage_agent, body.content, context={"history": history})

    assistant_msg = Message(
        conversation_id=conv_id,
        role="assistant",
        content=result.final_output,
        agent_name=result.last_agent.name,
    )
    db.add(assistant_msg)
    db.commit()
    return {"reply": result.final_output, "agent": result.last_agent.name}

Monitoring and Deployment

For monitoring, track three key metrics: average response latency, escalation rate, and customer satisfaction. Store these in a metrics table and expose a /analytics endpoint for the admin dashboard.

Deploy with Docker Compose for development and Kubernetes for production. The FastAPI backend uses a Dockerfile with uvicorn, the frontend is a static Next.js build served by nginx, and PostgreSQL runs as a managed service or a StatefulSet.

# monitoring/metrics.py
import time
from functools import wraps

def track_latency(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        start = time.time()
        result = await func(*args, **kwargs)
        latency = time.time() - start
        await store_metric("response_latency", latency)
        return result
    return wrapper

The complete project demonstrates every pillar of production AI: data modeling, agent orchestration, tool integration, API design, error handling, monitoring, and deployment. Each component is independently testable, and the architecture supports horizontal scaling by running multiple API replicas behind a load balancer.

FAQ

How do I add a new specialist agent without modifying the triage agent?

Register the new agent as a handoff on the triage agent and update the triage instructions to include the new routing rule. Because agents are defined as data, you can dynamically load agent configurations from a database or config file and register handoffs at startup.

What happens when an agent response has low confidence?

Attach a confidence scorer that evaluates the agent output against the original query. If confidence falls below a threshold (for example 0.6), automatically route to the escalation agent. Store the confidence score on the message record for analytics and quality review.

How should I handle concurrent conversations at scale?

Use async database sessions with connection pooling (SQLAlchemy async + asyncpg). Each FastAPI request handler runs in its own coroutine, so hundreds of conversations can proceed in parallel. For the LLM calls, the OpenAI SDK is natively async, so agent runs do not block the event loop.

---

#CapstoneProject #MultiAgent #CustomerSupport #FullStackAI #FastAPI #Deployment #AgenticAI #LearnAI #AIEngineering