customer-support-ai/
├── app.py                 # Main application
├── models.py             # Data models
├── services/
│   ├── __init__.py
│   ├── support_agent.py  # AI support agent
│   ├── knowledge_base.py # Knowledge base integration
│   └── escalation.py     # Human escalation logic
├── utils/
│   ├── __init__.py
│   ├── auth.py          # Authentication
│   └── analytics.py     # Analytics tracking
├── requirements.txt
└── .env

from dataclasses import dataclass
from typing import Optional, List, Dict
from datetime import datetime
from enum import Enum
 
class TicketStatus(Enum):
    OPEN = "open"
    IN_PROGRESS = "in_progress"
    RESOLVED = "resolved"
    ESCALATED = "escalated"
 
class TicketPriority(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    URGENT = "urgent"
 
@dataclass
class Company:
    id: str
    name: str
    support_email: str
    knowledge_base_url: Optional[str] = None
    escalation_rules: Optional[Dict] = None
 
@dataclass
class Customer:
    id: str
    company_id: str
    email: str
    name: str
    preferred_language: str = "en"
    history: List[str] = None
 
@dataclass
class SupportTicket:
    id: str
    company_id: str
    customer_id: str
    thread_id: str
    subject: str
    status: TicketStatus
    priority: TicketPriority
    created_at: datetime
    resolved_at: Optional[datetime] = None
    assigned_agent: Optional[str] = None
    tags: List[str] = None
    satisfaction_score: Optional[int] = None

import lumnisai
from typing import Dict, List, Optional, Tuple
import asyncio
from datetime import datetime
import json
 
class AISupportAgent:
    def __init__(self):
        self.client = lumnisai.AsyncClient()
        self.response_cache = {}
        
    async def analyze_ticket(
        self,
        company: Company,
        customer: Customer,
        message: str
    ) -> Tuple[TicketPriority, List[str]]:
        """Analyze incoming message to determine priority and tags."""
        
        prompt = f"""
        Analyze this customer support message and determine:
        1. Priority level (urgent/high/medium/low)
        2. Relevant tags/categories
        3. Sentiment (positive/neutral/negative)
        
        Company: {company.name}
        Customer Language: {customer.preferred_language}
        Message: {message}
        
        Return JSON format:
        {{
            "priority": "medium",
            "tags": ["billing", "subscription"],
            "sentiment": "neutral"
        }}
        """
        
        response = await self.client.invoke(
            prompt,
            user_id=f"{company.id}_{customer.id}"
        )
        
        try:
            analysis = json.loads(response.output_text)
            priority = TicketPriority[analysis["priority"].upper()]
            tags = analysis["tags"]
            return priority, tags
        except:
            return TicketPriority.MEDIUM, []
    
    async def generate_response(
        self,
        company: Company,
        customer: Customer,
        ticket: SupportTicket,
        message: str,
        knowledge_base: Optional[str] = None,
        previous_messages: List[Dict] = None
    ) -> Dict:
        """Generate contextual support response."""
        
        # Build context
        context = f"""
        You are a helpful customer support agent for {company.name}.
        Customer: {customer.name} (Language: {customer.preferred_language})
        Ticket Subject: {ticket.subject}
        Priority: {ticket.priority.value}
        Tags: {', '.join(ticket.tags or [])}
        """
        
        if knowledge_base:
            context += f"\n\nRelevant Documentation:\n{knowledge_base}"
        
        if previous_messages:
            context += "\n\nPrevious Conversation:"
            for msg in previous_messages[-5:]:  # Last 5 messages
                context += f"\n{msg['role']}: {msg['content']}"
        
        # Create the prompt
        prompt = f"""{context}
        
        Customer Message: {message}
        
        Instructions:
        1. Provide a helpful, empathetic response
        2. Use the customer's preferred language: {customer.preferred_language}
        3. Reference documentation when applicable
        4. If you cannot fully resolve the issue, acknowledge this
        5. Maintain a professional and friendly tone
        
        Response:"""
        
        # Generate response with progress tracking
        response_parts = []
        confidence_score = 0.0
        
        async for update in await self.client.invoke(
            prompt,
            user_id=f"{company.id}_{customer.id}",
            thread_id=ticket.thread_id,
            stream=True
        ):
            if update.state == "processing":
                print(f"🤖 {update.message}")
            elif update.state == "completed":
                response_parts.append(update.output_text)
                
                # Analyze confidence
                confidence_prompt = f"""
                Rate your confidence in this response from 0-1:
                Response: {update.output_text}
                Context: Solving customer support issue
                Return only a number between 0 and 1.
                """
                
                confidence_resp = await self.client.invoke(
                    confidence_prompt,
                    user_id=f"{company.id}_system"
                )
                
                try:
                    confidence_score = float(confidence_resp.output_text.strip())
                except:
                    confidence_score = 0.7
        
        return {
            "response": "\n".join(response_parts),
            "confidence": confidence_score,
            "needs_escalation": confidence_score < 0.6,
            "timestamp": datetime.utcnow().isoformat()
        }
    
    async def suggest_solutions(
        self,
        company: Company,
        ticket: SupportTicket,
        knowledge_base: List[Dict]
    ) -> List[Dict]:
        """Suggest potential solutions based on similar resolved tickets."""
        
        prompt = f"""
        Based on these previously resolved similar tickets, suggest solutions:
        
        Current Issue: {ticket.subject}
        Tags: {', '.join(ticket.tags or [])}
        
        Similar Resolved Tickets:
        {json.dumps(knowledge_base, indent=2)}
        
        Provide 3-5 potential solutions with confidence scores.
        Format as JSON array with 'solution' and 'confidence' fields.
        """
        
        response = await self.client.invoke(
            prompt,
            user_id=f"{company.id}_system"
        )
        
        try:
            return json.loads(response.output_text)
        except:
            return []

import lumnisai
from typing import List, Dict, Optional
import asyncio
from datetime import datetime
 
class KnowledgeBaseService:
    def __init__(self):
        self.client = lumnisai.AsyncClient()
        self.embeddings_cache = {}
        
    async def search_documentation(
        self,
        company: Company,
        query: str,
        limit: int = 5
    ) -> List[Dict]:
        """Search company's knowledge base for relevant articles."""
        
        # In production, this would search vector database
        # For demo, we'll use AI to simulate search
        
        prompt = f"""
        Simulate searching {company.name}'s documentation for: "{query}"
        
        Return top {limit} relevant documentation snippets.
        Format as JSON array with fields:
        - title: Document title
        - content: Relevant excerpt (2-3 sentences)
        - url: Simulated URL
        - relevance_score: 0-1
        
        Make the content realistic for a customer support knowledge base.
        """
        
        response = await self.client.invoke(
            prompt,
            user_id=f"{company.id}_kb_search"
        )
        
        try:
            return json.loads(response.output_text)
        except:
            return []
    
    async def generate_kb_article(
        self,
        company: Company,
        ticket: SupportTicket,
        resolution: str
    ) -> Dict:
        """Generate a knowledge base article from resolved ticket."""
        
        prompt = f"""
        Create a knowledge base article based on this resolved support ticket:
        
        Company: {company.name}
        Subject: {ticket.subject}
        Tags: {', '.join(ticket.tags or [])}
        Resolution: {resolution}
        
        Generate:
        1. Article title (clear and searchable)
        2. Problem description
        3. Step-by-step solution
        4. Related articles (2-3 suggestions)
        
        Format as professional documentation.
        """
        
        response = await self.client.invoke(
            prompt,
            user_id=f"{company.id}_kb_generator"
        )
        
        return {
            "content": response.output_text,
            "generated_at": datetime.utcnow().isoformat(),
            "source_ticket": ticket.id
        }

from typing import Dict, Optional
import asyncio
 
class EscalationService:
    def __init__(self):
        self.escalation_rules = {}
        
    async def check_escalation_needed(
        self,
        company: Company,
        ticket: SupportTicket,
        ai_confidence: float,
        customer_sentiment: str,
        response_time: float
    ) -> Dict:
        """Determine if ticket needs human escalation."""
        
        reasons = []
        
        # Check confidence threshold
        if ai_confidence < 0.6:
            reasons.append("Low AI confidence")
        
        # Check priority
        if ticket.priority in [TicketPriority.URGENT, TicketPriority.HIGH]:
            reasons.append(f"High priority: {ticket.priority.value}")
        
        # Check sentiment
        if customer_sentiment == "negative":
            reasons.append("Negative customer sentiment")
        
        # Check response time
        if response_time > 300:  # 5 minutes
            reasons.append("Slow response time")
        
        # Check custom company rules
        if company.escalation_rules:
            for rule_name, rule_check in company.escalation_rules.items():
                if rule_check(ticket):
                    reasons.append(f"Company rule: {rule_name}")
        
        return {
            "needs_escalation": len(reasons) > 0,
            "reasons": reasons,
            "suggested_agent": self.find_best_agent(company, ticket)
        }
    
    def find_best_agent(
        self,
        company: Company,
        ticket: SupportTicket
    ) -> Optional[str]:
        """Find the best human agent for escalation."""
        
        # In production, this would check agent availability,
        # expertise, workload, etc.
        
        if "billing" in (ticket.tags or []):
            return "billing_specialist"
        elif ticket.priority == TicketPriority.URGENT:
            return "senior_agent"
        else:
            return "next_available"

import asyncio
import lumnisai
from datetime import datetime
from typing import Dict, List
import json
import os
from dotenv import load_dotenv
 
from models import Company, Customer, SupportTicket, TicketStatus, TicketPriority
from services.support_agent import AISupportAgent
from services.knowledge_base import KnowledgeBaseService
from services.escalation import EscalationService
 
load_dotenv()
 
class CustomerSupportSystem:
    def __init__(self):
        self.lumnis_client = lumnisai.Client()
        self.ai_agent = AISupportAgent()
        self.kb_service = KnowledgeBaseService()
        self.escalation_service = EscalationService()
        
        # In production, these would be in a database
        self.companies = {}
        self.customers = {}
        self.tickets = {}
        
    async def setup_company(self, company_name: str) -> Company:
        """Set up a new company in the system."""
        
        company = Company(
            id=f"company_{len(self.companies)}",
            name=company_name,
            support_email=f"support@{company_name.lower().replace(' ', '')}.com",
            escalation_rules={
                "vip_customer": lambda t: "vip" in (t.tags or []),
                "multiple_contacts": lambda t: len(t.tags or []) > 3
            }
        )
        
        self.companies[company.id] = company
        
        # Create company user in Lumnis
        self.lumnis_client.create_user(
            email=company.support_email,
            first_name=company.name,
            last_name="Support"
        )
        
        print(f"✅ Set up company: {company.name}")
        return company
    
    async def register_customer(
        self,
        company: Company,
        email: str,
        name: str,
        language: str = "en"
    ) -> Customer:
        """Register a new customer."""
        
        customer = Customer(
            id=f"customer_{len(self.customers)}",
            company_id=company.id,
            email=email,
            name=name,
            preferred_language=language,
            history=[]
        )
        
        self.customers[customer.id] = customer
        
        # Create customer user in Lumnis
        user_id = f"{company.id}_{customer.id}"
        self.lumnis_client.create_user(
            email=f"{user_id}@support.lumnis.ai",
            first_name=name.split()[0],
            last_name=name.split()[-1] if len(name.split()) > 1 else ""
        )
        
        return customer
    
    async def handle_support_request(
        self,
        company: Company,
        customer: Customer,
        subject: str,
        message: str
    ) -> Dict:
        """Handle incoming support request."""
        
        start_time = datetime.utcnow()
        
        # Analyze the ticket
        priority, tags = await self.ai_agent.analyze_ticket(
            company, customer, message
        )
        
        # Create thread for conversation
        thread = self.lumnis_client.create_thread(
            user_id=f"{company.id}_{customer.id}",
            title=subject
        )
        
        # Create ticket
        ticket = SupportTicket(
            id=f"ticket_{len(self.tickets)}",
            company_id=company.id,
            customer_id=customer.id,
            thread_id=thread.thread_id,
            subject=subject,
            status=TicketStatus.OPEN,
            priority=priority,
            created_at=start_time,
            tags=tags
        )
        
        self.tickets[ticket.id] = ticket
        
        # Search knowledge base
        kb_results = await self.kb_service.search_documentation(
            company, subject + " " + message
        )
        
        kb_content = "\n\n".join([
            f"**{doc['title']}**\n{doc['content']}"
            for doc in kb_results[:3]
        ])
        
        # Generate AI response
        ai_response = await self.ai_agent.generate_response(
            company=company,
            customer=customer,
            ticket=ticket,
            message=message,
            knowledge_base=kb_content,
            previous_messages=[]
        )
        
        # Check if escalation needed
        response_time = (datetime.utcnow() - start_time).total_seconds()
        escalation_check = await self.escalation_service.check_escalation_needed(
            company=company,
            ticket=ticket,
            ai_confidence=ai_response["confidence"],
            customer_sentiment="neutral",  # Would be from sentiment analysis
            response_time=response_time
        )
        
        # Update ticket status
        if escalation_check["needs_escalation"]:
            ticket.status = TicketStatus.ESCALATED
            ticket.assigned_agent = escalation_check["suggested_agent"]
        else:
            ticket.status = TicketStatus.IN_PROGRESS
        
        # Prepare response
        result = {
            "ticket_id": ticket.id,
            "status": ticket.status.value,
            "priority": ticket.priority.value,
            "response": ai_response["response"],
            "confidence": ai_response["confidence"],
            "escalated": escalation_check["needs_escalation"],
            "escalation_reasons": escalation_check.get("reasons", []),
            "kb_articles": kb_results[:3],
            "response_time": response_time,
            "tags": tags
        }
        
        # Update customer history
        customer.history.append(ticket.id)
        
        return result
    
    async def resolve_ticket(
        self,
        ticket_id: str,
        resolution: str,
        satisfaction_score: int
    ) -> Dict:
        """Mark ticket as resolved and generate KB article."""
        
        ticket = self.tickets.get(ticket_id)
        if not ticket:
            return {"error": "Ticket not found"}
        
        ticket.status = TicketStatus.RESOLVED
        ticket.resolved_at = datetime.utcnow()
        ticket.satisfaction_score = satisfaction_score
        
        # Generate KB article if satisfaction is high
        kb_article = None
        if satisfaction_score >= 4:
            company = self.companies[ticket.company_id]
            kb_article = await self.kb_service.generate_kb_article(
                company, ticket, resolution
            )
        
        return {
            "ticket_id": ticket_id,
            "status": "resolved",
            "resolution_time": (ticket.resolved_at - ticket.created_at).total_seconds(),
            "kb_article_generated": kb_article is not None,
            "kb_article": kb_article
        }
 
# Example usage
async def main():
    system = CustomerSupportSystem()
    
    # Set up companies
    tech_corp = await system.setup_company("TechCorp Inc")
    retail_co = await system.setup_company("RetailCo")
    
    # Register customers
    alice = await system.register_customer(
        tech_corp, "alice@example.com", "Alice Johnson"
    )
    
    bob = await system.register_customer(
        tech_corp, "bob@example.com", "Bob Smith", "es"  # Spanish
    )
    
    carol = await system.register_customer(
        retail_co, "carol@example.com", "Carol Davis"
    )
    
    # Simulate support requests
    print("\n📧 Handling support requests...\n")
    
    # Alice's billing issue
    response1 = await system.handle_support_request(
        company=tech_corp,
        customer=alice,
        subject="Cannot access premium features",
        message="I upgraded to premium yesterday but still can't access the advanced analytics dashboard. My payment went through but the features are locked."
    )
    
    print(f"Ticket #{response1['ticket_id']} - {alice.name}")
    print(f"Priority: {response1['priority']}")
    print(f"Response: {response1['response'][:200]}...")
    print(f"Confidence: {response1['confidence']:.2f}")
    print(f"Escalated: {response1['escalated']}")
    print(f"Tags: {', '.join(response1['tags'])}")
    print("-" * 50)
    
    # Bob's technical issue (in Spanish)
    response2 = await system.handle_support_request(
        company=tech_corp,
        customer=bob,
        subject="Error de conexión",
        message="No puedo conectarme a la aplicación. Me da error 500 cada vez que intento iniciar sesión."
    )
    
    print(f"\nTicket #{response2['ticket_id']} - {bob.name}")
    print(f"Response: {response2['response'][:200]}...")
    print("-" * 50)
    
    # Carol's urgent issue
    response3 = await system.handle_support_request(
        company=retail_co,
        customer=carol,
        subject="URGENT: Order not delivered",
        message="My order #12345 was supposed to be delivered yesterday for an important event. It never arrived and tracking shows no updates. I need this resolved immediately!"
    )
    
    print(f"\nTicket #{response3['ticket_id']} - {carol.name}")
    print(f"Priority: {response3['priority']}")
    print(f"Escalated: {response3['escalated']}")
    if response3['escalated']:
        print(f"Escalation reasons: {', '.join(response3['escalation_reasons'])}")
    print("-" * 50)
    
    # Simulate ticket resolution
    print("\n✅ Resolving tickets...\n")
    
    resolution = await system.resolve_ticket(
        response1['ticket_id'],
        "Manually activated premium features. Issue was due to cache delay.",
        5  # satisfaction score
    )
    
    if resolution.get('kb_article'):
        print("Generated KB article:")
        print(resolution['kb_article']['content'][:300] + "...")
 
if __name__ == "__main__":
    asyncio.run(main())

import lumnisai
from typing import Dict, List
from datetime import datetime, timedelta
import json
 
class SupportAnalytics:
    def __init__(self):
        self.client = lumnisai.Client()
    
    async def generate_insights(
        self,
        company: Company,
        tickets: List[SupportTicket],
        timeframe: timedelta
    ) -> Dict:
        """Generate AI-powered insights from support tickets."""
        
        # Filter tickets by timeframe
        cutoff = datetime.utcnow() - timeframe
        recent_tickets = [
            t for t in tickets
            if t.created_at >= cutoff and t.company_id == company.id
        ]
        
        # Prepare data for analysis
        ticket_data = []
        for ticket in recent_tickets:
            ticket_data.append({
                "subject": ticket.subject,
                "tags": ticket.tags,
                "priority": ticket.priority.value,
                "status": ticket.status.value,
                "resolution_time": (
                    (ticket.resolved_at - ticket.created_at).total_seconds()
                    if ticket.resolved_at else None
                ),
                "satisfaction": ticket.satisfaction_score
            })
        
        prompt = f"""
        Analyze these customer support tickets from {company.name}:
        
        {json.dumps(ticket_data, indent=2)}
        
        Provide insights on:
        1. Most common issues (top 5)
        2. Average resolution time by category
        3. Customer satisfaction trends
        4. Recommendations for improvement
        5. Predicted ticket volume for next period
        
        Format as a comprehensive report with actionable insights.
        """
        
        response = await self.client.invoke(
            prompt,
            user_id=f"{company.id}_analytics"
        )
        
        return {
            "report": response.output_text,
            "generated_at": datetime.utcnow().isoformat(),
            "ticket_count": len(recent_tickets),
            "timeframe_days": timeframe.days
        }

async def translate_response(
    self,
    response: str,
    target_language: str
) -> str:
    """Translate response to customer's preferred language."""
    
    if target_language == "en":
        return response
    
    prompt = f"""
    Translate this customer support response to {target_language}.
    Maintain the professional tone and all technical details.
    
    Response: {response}
    """
    
    translated = await self.client.invoke(
        prompt,
        user_id="translation_service"
    )
    
    return translated.output_text

async def monitor_customer_sentiment(
    self,
    company: Company,
    customer: Customer,
    message_history: List[str]
) -> Dict:
    """Monitor sentiment and suggest proactive actions."""
    
    prompt = f"""
    Analyze this customer's support history for sentiment trends:
    
    Customer: {customer.name}
    Messages: {json.dumps(message_history[-10:])}
    
    Identify:
    1. Overall sentiment trend (improving/declining/stable)
    2. Risk of churn (low/medium/high)
    3. Suggested proactive actions
    
    Return as JSON with specific recommendations.
    """
    
    response = await self.client.invoke(
        prompt,
        user_id=f"{company.id}_sentiment"
    )
    
    return json.loads(response.output_text)

async def sync_with_crm(
    self,
    ticket: SupportTicket,
    crm_system: str = "salesforce"
) -> Dict:
    """Sync ticket data with external CRM."""
    
    # Add CRM integration via Composio
    self.lumnis_client.add_api_key(
        provider=lumnisai.ApiProvider.SALESFORCE_API_KEY,
        api_key=os.getenv("SALESFORCE_API_KEY"),
        user_id=f"{ticket.company_id}_crm"
    )
    
    prompt = f"""
    Create a Salesforce case for this support ticket:
    
    Ticket ID: {ticket.id}
    Subject: {ticket.subject}
    Priority: {ticket.priority.value}
    Customer: {ticket.customer_id}
    
    Use the Salesforce API to create the case and return the case ID.
    """
    
    response = await self.lumnis_client.invoke(
        prompt,
        user_id=f"{ticket.company_id}_crm"
    )
    
    return {"crm_case_id": response.output_text}

-- Companies table
CREATE TABLE companies (
    id UUID PRIMARY KEY,
    name VARCHAR(255) NOT NULL,
    support_email VARCHAR(255),
    settings JSONB,
    created_at TIMESTAMP DEFAULT NOW()
);
 
-- Customers table
CREATE TABLE customers (
    id UUID PRIMARY KEY,
    company_id UUID REFERENCES companies(id),
    email VARCHAR(255) NOT NULL,
    name VARCHAR(255),
    language VARCHAR(10) DEFAULT 'en',
    created_at TIMESTAMP DEFAULT NOW()
);
 
-- Tickets table
CREATE TABLE tickets (
    id UUID PRIMARY KEY,
    company_id UUID REFERENCES companies(id),
    customer_id UUID REFERENCES customers(id),
    thread_id VARCHAR(255),
    subject TEXT,
    status VARCHAR(50),
    priority VARCHAR(50),
    tags TEXT[],
    assigned_agent VARCHAR(255),
    created_at TIMESTAMP DEFAULT NOW(),
    resolved_at TIMESTAMP,
    satisfaction_score INTEGER
);
 
-- Create indexes
CREATE INDEX idx_tickets_company ON tickets(company_id);
CREATE INDEX idx_tickets_customer ON tickets(customer_id);
CREATE INDEX idx_tickets_status ON tickets(status);