ML System Design for LLM-Powered Customer Support

Building an LLM-powered customer support system is a common first production LLM project. The demo is easy - connect an LLM to your knowledge base and it answers questions. The production system requires careful design around latency, reliability, cost, safety, and escalation.

System Overview

User message
      │
      ▼
┌─────────────────────────────────────────────────────────────────┐
│  Input Layer                                                    │
│  - PII detection + redaction                                    │
│  - Intent classification (support/other)                        │
│  - Language detection                                           │
│  - Guardrail check (harmful content)                            │
└─────────────────────────────────────────────────────────────────┘
      │
      ▼
┌─────────────────────────────────────────────────────────────────┐
│  Retrieval (RAG)                                                │
│  - Embed query                                                  │
│  - Search knowledge base (help articles, FAQs, policies)        │
│  - Retrieve conversation history for this session              │
└─────────────────────────────────────────────────────────────────┘
      │ retrieved context
      ▼
┌─────────────────────────────────────────────────────────────────┐
│  LLM Response Generation                                        │
│  - System prompt (persona, constraints, tone)                   │
│  - Retrieved context + conversation history + user message      │
│  - Tool calling: check order status, reset password, etc.       │
│  - Streaming SSE response to UI                                 │
└─────────────────────────────────────────────────────────────────┘
      │
      ▼
┌─────────────────────────────────────────────────────────────────┐
│  Output Layer                                                   │
│  - Confidence scoring / hallucination detection                 │
│  - Escalation trigger check                                     │
│  - Response logging for evaluation                              │
└─────────────────────────────────────────────────────────────────┘

The Knowledge Base and RAG Layer

Your support knowledge lives in help articles, FAQs, product documentation, and historical resolved tickets. Structure the knowledge base for retrieval:

pythonCopy
import anthropic
from typing import Optional

client = anthropic.Anthropic()

class SupportKnowledgeBase:
    def __init__(self, vector_db):
        self.vector_db = vector_db

    def ingest_article(self, article: dict):
        """Index a help article with chunking."""
        chunks = self._chunk_article(article['content'], max_tokens=300)
        for i, chunk in enumerate(chunks):
            embedding = self._embed(chunk)
            self.vector_db.upsert({
                'id': f"{article['id']}_chunk_{i}",
                'embedding': embedding,
                'metadata': {
                    'article_id': article['id'],
                    'title': article['title'],
                    'category': article['category'],
                    'content': chunk,
                    'last_updated': article['updated_at']
                }
            })

    def retrieve(self, query: str, top_k: int = 5) -> list[dict]:
        query_embedding = self._embed(query)
        results = self.vector_db.search(query_embedding, top_k=top_k)
        return [r['metadata'] for r in results]

    def _embed(self, text: str) -> list[float]:
        response = client.embeddings.create(
            model="voyage-3",
            input=[text]
        )
        return response.embeddings[0].embedding

    def _chunk_article(self, content: str, max_tokens: int = 300) -> list[str]:
        # Split on section headers first, then by token count
        sections = content.split('\n## ')
        chunks = []
        for section in sections:
            if len(section.split()) > max_tokens:
                # Further split long sections
                words = section.split()
                for i in range(0, len(words), max_tokens):
                    chunks.append(' '.join(words[i:i + max_tokens]))
            else:
                chunks.append(section)
        return chunks

The Response Generation Layer

pythonCopy
import anthropic
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()
client = anthropic.Anthropic()

SYSTEM_PROMPT = """You are a helpful customer support agent for Acme Inc.

Guidelines:
- Answer questions based on the provided knowledge base articles only
- If you don't know something, say so clearly and offer to escalate
- Never make up order numbers, dates, or policies
- Be concise  -  support users want quick answers
- If the user seems frustrated, acknowledge it before providing information
- For account-specific actions (refunds, cancellations), use the available tools

You have access to tools to look up order information and initiate account actions."""

@app.post("/support/chat")
async def chat(request: dict):
    session_id = request['session_id']
    user_message = request['message']

    # Retrieve relevant knowledge
    kb = SupportKnowledgeBase(vector_db)
    relevant_docs = kb.retrieve(user_message, top_k=5)

    context = "\n\n".join([
        f"--- {doc['title']} ---\n{doc['content']}"
        for doc in relevant_docs
    ])

    # Build messages with conversation history
    history = get_session_history(session_id)
    messages = history + [{"role": "user", "content": user_message}]

    # Tool definitions for account actions
    tools = [
        {
            "name": "lookup_order",
            "description": "Look up order status by order ID",
            "input_schema": {
                "type": "object",
                "properties": {"order_id": {"type": "string"}},
                "required": ["order_id"]
            }
        },
        {
            "name": "initiate_refund",
            "description": "Initiate a refund for a valid order",
            "input_schema": {
                "type": "object",
                "properties": {
                    "order_id": {"type": "string"},
                    "reason": {"type": "string"}
                },
                "required": ["order_id", "reason"]
            }
        }
    ]

    async def generate():
        full_response = ""
        with client.messages.stream(
            model="claude-haiku-4-5-20251001",
            max_tokens=1024,
            system=SYSTEM_PROMPT + f"\n\nKnowledge base:\n{context}",
            messages=messages,
            tools=tools
        ) as stream:
            for event in stream:
                if hasattr(event, 'delta') and hasattr(event.delta, 'text'):
                    chunk = event.delta.text
                    full_response += chunk
                    yield f"data: {chunk}\n\n"

        # Log for evaluation
        log_response(session_id, user_message, full_response, relevant_docs)

    return StreamingResponse(generate(), media_type="text/event-stream")

Escalation Logic

The most critical part of any LLM support system: knowing when to hand off to a human.

pythonCopy
def should_escalate(
    user_message: str,
    llm_response: str,
    session_history: list,
    confidence_score: float
) -> tuple[bool, str]:
    """
    Returns (should_escalate, reason)
    """
    # Hard rules first
    escalation_keywords = [
        "legal action", "lawsuit", "attorney", "discrimination",
        "data breach", "security", "urgent", "emergency"
    ]
    if any(kw in user_message.lower() for kw in escalation_keywords):
        return True, "sensitive_topic"

    # Repeated contact (frustration signal)
    same_topic_count = sum(1 for h in session_history
                           if classify_intent(h['user_message']) == classify_intent(user_message))
    if same_topic_count >= 3:
        return True, "repeated_contact"

    # Low confidence in response
    if confidence_score < 0.6:
        return True, "low_confidence"

    # LLM expresses uncertainty
    uncertainty_phrases = ["I'm not sure", "I don't know", "I cannot", "please contact"]
    if any(phrase.lower() in llm_response.lower() for phrase in uncertainty_phrases):
        return True, "llm_uncertain"

    # Session too long (user not getting help)
    if len(session_history) > 8:
        return True, "long_session"

    return False, ""

Cost Control

LLM API costs can scale unexpectedly. The main levers:

Model selection by intent:

pythonCopy
def select_model(intent: str, complexity: str) -> str:
    """Use cheaper models for simple intents."""
    if intent in ['order_status', 'faq_simple', 'hours_location']:
        return "claude-haiku-4-5-20251001"  # fast, cheap
    elif intent in ['returns_policy', 'account_issue']:
        return "claude-sonnet-4-6"           # balanced
    else:
        return "claude-sonnet-4-6"           # complex, rare

Prompt caching: Your system prompt + knowledge base context is the same across thousands of requests. Use Anthropic's prompt caching to avoid re-processing it:

pythonCopy
messages_with_cache = [
    {
        "role": "user",
        "content": [
            {
                "type": "text",
                "text": SYSTEM_PROMPT + context,
                "cache_control": {"type": "ephemeral"}  # cache for 5 minutes
            },
            {
                "type": "text",
                "text": user_message
            }
        ]
    }
]

Context window management: Do not pass the full conversation history. Summarize old turns:

pythonCopy
def get_compressed_history(full_history: list, max_turns: int = 6) -> list:
    if len(full_history) <= max_turns:
        return full_history

    # Summarize older turns
    old_turns = full_history[:-max_turns]
    summary = summarize_conversation(old_turns)  # LLM call
    recent_turns = full_history[-max_turns:]

    return [{"role": "assistant", "content": f"[Earlier conversation summary: {summary}}"}] + recent_turns

Evaluation

pythonCopy
def evaluate_support_quality(sample_conversations: list) -> dict:
    """
    LLM-as-judge evaluation for support quality.
    """
    scores = []
    for conv in sample_conversations:
        prompt = f"""Evaluate this customer support exchange:

User: {conv['user_message']}
Agent: {conv['agent_response']}

Rate on each dimension (1-5):
1. Accuracy: Does the response correctly answer the question?
2. Helpfulness: Does it actually resolve the user's issue?
3. Tone: Is the tone appropriate and empathetic?
4. Conciseness: Is it appropriately concise?

Respond in JSON: {{"accuracy": N, "helpfulness": N, "tone": N, "conciseness": N, "reasoning": "..."}}"""

        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=256,
            messages=[{"role": "user", "content": prompt}]
        )
        score = json.loads(response.content[0].text)
        scores.append(score)

    return {
        metric: np.mean([s[metric] for s in scores])
        for metric in ['accuracy', 'helpfulness', 'tone', 'conciseness']
    }

Run this evaluation weekly on a sampled set of conversations. Track trends over time and after every knowledge base update.

Common Mistakes

Not building an escalation path when the LLM has low confidence. An LLM that gives a wrong answer confidently is worse than no answer at all in a customer support context because the customer acts on incorrect information. Every production LLM support system needs a confidence signal (refusal detection, entailment scoring, or a calibrated classifier) and a human handoff path for cases below the threshold.

Sending raw user data to a third-party API without PII stripping. Customer support conversations contain names, account numbers, email addresses, and sensitive context. Transmitting this data to an external LLM API without scrubbing PII may violate GDPR, CCPA, or your contracts with customers. Implement a PII detection and redaction step before any user text leaves your infrastructure.

Skipping an eval harness because "it seems to work." Informal testing on a few example conversations is not a substitute for a structured evaluation set with known-correct answers. Without an eval harness you cannot detect regressions when you change the prompt, the retrieval system, or the underlying model. Build at least a 50-case eval set before your first production deploy.

What to Practice Next

• Define the confidence threshold and escalation logic for your LLM support system before writing any model code: at what signal level does it hand off to a human, what information does it pass, and how is that handoff measured?
• Implement a PII redaction pass (regex or NER-based) that masks names, email addresses, and account numbers before they leave your service boundary; test it against 20 realistic support messages.
• Build a 30-case eval set covering common support topics, ambiguous queries, and edge cases; run it after every prompt or model change and track the pass rate over time.