Kundensupport

version: "1.0"

name: support-pipeline
type: sequential
description: "Full support flow: triage the ticket, then draft a response"

agents:
  - support-triager
  - support-responder

version: "1.0"

name: support-responder-test-flash
type: llm
model: connic/gemini-3.7-flash
description: "A/B variant of support-responder on a cheaper Flash model for cost comparison"

system_prompt: |
  You are a customer support agent drafting responses to tickets that have
  already been triaged. The triage data is included in your input.

  Response workflow:
  1. ALWAYS search Retrieval first using search_solutions with relevant
     terms from the ticket. Search multiple times with different queries if the
     first search doesn't return useful results.
  2. If you find a matching article or FAQ, base your response on it and adapt
     the language to match the customer's tone and technical level.
  3. If no relevant knowledge exists but you can solve it, draft the response
     AND store the solution using save_solution so future tickets benefit.
  4. If you cannot solve it, acknowledge the issue clearly and explain that
     you're escalating to a specialist. Never make up answers.

  Tone matching (based on triage sentiment):
  - frustrated/angry: lead with empathy, acknowledge the pain, then solve
  - neutral: be helpful and efficient, no fluff
  - positive: match the energy, be warm

  Use the customer's name from context if the middleware provided it.
  Keep responses under 200 words unless the solution requires detailed steps.

temperature: 0.5
max_iterations: 15

guardrails:
  output:
    - type: system_prompt_leakage
      mode: block
    - type: pii_leakage
      mode: block
      config:
        entities: [credit_card, ssn]

tools:
  - support_tools.search_solutions
  - support_tools.save_solution
  - support_tools.format_escalation

version: "1.0"

name: support-responder
type: llm
model: connic/gpt-5.6-terra
description: "Drafts empathetic support responses using Retrieval for accurate answers"

system_prompt: |
  You are a customer support agent drafting responses to tickets that have
  already been triaged. The triage data is included in your input.

  Response workflow:
  1. ALWAYS search Retrieval first using search_solutions with relevant
     terms from the ticket. Search multiple times with different queries if the
     first search doesn't return useful results.
  2. If you find a matching article or FAQ, base your response on it and adapt
     the language to match the customer's tone and technical level.
  3. If no relevant knowledge exists but you can solve it, draft the response
     AND store the solution using save_solution so future tickets benefit.
  4. If you cannot solve it, acknowledge the issue clearly and explain that
     you're escalating to a specialist. Never make up answers.

  Tone matching (based on triage sentiment):
  - frustrated/angry: lead with empathy, acknowledge the pain, then solve
  - neutral: be helpful and efficient, no fluff
  - positive: match the energy, be warm

  Use the customer's name from context if the middleware provided it.
  Keep responses under 200 words unless the solution requires detailed steps.

temperature: 0.5
max_iterations: 15

guardrails:
  output:
    - type: system_prompt_leakage
      mode: block
    - type: pii_leakage
      mode: block
      config:
        entities: [credit_card, ssn]

tools:
  - support_tools.search_solutions
  - support_tools.save_solution
  - support_tools.format_escalation

version: "1.0"

name: support-triager
type: llm
model: connic/gpt-5.6-luna
description: "Classifies incoming support tickets by priority, category, and sentiment"

system_prompt: |
  You are a support ticket triage specialist. Analyze the incoming message and
  classify it precisely.

  Priority rules (strictly follow):
  - critical: service outage, data loss, security breach, complete inability to use product
  - high: major feature broken, blocking issue for paying customer, billing error
  - medium: feature request, non-blocking bug, how-to question from paying customer
  - low: general inquiry, feedback, feature suggestion, free-tier question

  Category rules:
  - billing: payments, invoices, subscription changes, refunds
  - technical: bugs, errors, API issues, integration problems
  - account: login, permissions, team management, SSO
  - onboarding: setup help, getting started, documentation questions
  - general: everything else

  Also assess customer sentiment (positive, neutral, frustrated, angry) to help
  the responder match tone. Extract any product/version mentions for routing.

  If the customer mentions they are on an enterprise plan, always bump priority
  by one level (e.g. medium -> high).

temperature: 0.2
output_schema: ticket-classification

guardrails:
  input:
    - type: prompt_injection
      mode: block
    - type: pii
      mode: redact
      config:
        entities: [email, phone]
    - type: topic_restriction
      mode: block
      config:
        allowed_topics: [billing, technical, account, onboarding, general support]
        off_topic_message: "I can only help with product support and billing."

"""Support tools for escalation and managed semantic retrieval."""

from typing import Any
from connic.tools import retrieval_query, retrieval_store


async def search_solutions(query: str) -> list[dict]:
    """Search support Retrieval for solutions matching the query.

    Args:
        query: Describe the customer's issue (e.g. "password reset not working").

    Returns:
        List of matching solution entries with content and relevance score.
    """
    result = await retrieval_query(query, namespace="solutions", max_results=3)
    return result.get("results", [])


async def save_solution(
    content: str,
    entry_id: str | None = None,
) -> dict:
    """Store a reusable solution in Retrieval for future tickets.

    Args:
        content: The solution text. Include the problem description for better retrieval.
        entry_id: Optional stable ID for future updates (e.g. "password-reset-steps").

    Returns:
        Store result with entry_id.
    """
    return await retrieval_store(content, namespace="solutions", entry_id=entry_id)


def format_escalation(
    ticket_summary: str,
    priority: str,
    category: str,
    reason: str,
    context: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Format an escalation payload for human handoff.

    Args:
        ticket_summary: Brief summary of the customer issue
        priority: Ticket priority (low/medium/high/critical)
        category: Ticket category (billing/technical/account/onboarding/general)
        reason: Why the agent is escalating (no knowledge found, complex issue, etc.)
        context: Auto-injected run context

    Returns:
        Structured escalation payload ready for webhook delivery
    """
    return {
        "escalation": True,
        "summary": ticket_summary,
        "priority": priority,
        "category": category,
        "reason": reason,
        "run_id": context.get("run_id") if context else None,
        "connector_id": context.get("connector_id") if context else None,
    }

"""Middleware for support-triager.

Enforces rate limiting per customer and enriches context with customer metadata.
Demonstrates StopProcessing for graceful early exits.
"""

from typing import Any

from connic import StopProcessing

BLOCKED_DOMAINS = {"spam.example.com", "throwaway.mail"}


async def before(content: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    """Filter spam and enrich with customer metadata before triage.

    Raises StopProcessing for blocked senders so the LLM never runs,
    saving tokens and latency.
    """
    parts = content.get("parts", [])
    text = ""
    for part in parts:
        if isinstance(part, dict) and "text" in part:
            text += part["text"]

    text_lower = text.lower()
    for domain in BLOCKED_DOMAINS:
        if domain in text_lower:
            raise StopProcessing(
                "This message has been automatically filtered. "
                "Please contact support through our official channels."
            )

    context["channel"] = "websocket" if (context.get("connector_id") or "").startswith("ws-") else "async"

    return content


async def after(response: str, context: dict[str, Any]) -> str:
    """Pass triage result through unchanged."""
    return response

{
  "type": "object",
  "description": "Support ticket triage classification",
  "properties": {
    "priority": {
      "type": "string",
      "enum": ["low", "medium", "high", "critical"],
      "description": "Ticket urgency level"
    },
    "category": {
      "type": "string",
      "enum": ["billing", "technical", "account", "onboarding", "general"],
      "description": "Ticket category for routing"
    },
    "sentiment": {
      "type": "string",
      "enum": ["positive", "neutral", "frustrated", "angry"],
      "description": "Customer emotional state"
    },
    "summary": {
      "type": "string",
      "description": "One-line ticket summary"
    },
    "product_mentions": {
      "type": "array",
      "items": {"type": "string"},
      "description": "Products or features mentioned"
    },
    "suggested_assignee": {
      "type": "string",
      "description": "Recommended team (billing-team, engineering, onboarding, general-support)"
    }
  },
  "required": ["priority", "category", "sentiment", "summary"]
}

version: "1.0"

# The responder's entire tool list is custom file tools, so strict_mocks holds:
# every tool call is one we stubbed, and nothing touches Retrieval.
# Output is raw text, so cases assert the run completed.
defaults:
  runs: 1
  timeout_s: 120
  mocks: support_mocks
  strict_mocks: true

tests:
  - name: answers_from_retrieval
    payload: '{"priority":"medium","category":"account","sentiment":"neutral","summary":"User cannot reset their password.","text":"I cannot reset my password. After you draft the reply, save the solution for future tickets."}'
    expected_result: status == "completed"
    expected_tool_calls:
      - support_tools.search_solutions
      - support_tools.save_solution

  - name: escalates_unsolvable_ticket
    payload: '{"priority":"critical","category":"technical","sentiment":"angry","summary":"Full production outage with data loss.","text":"Our entire production database is down and we are losing data. This needs a specialist immediately."}'
    expected_result: status == "completed"
    expected_tool_calls:
      - support_tools.format_escalation

version: "1.0"

# The triager has no custom tools, so a single file with these cases gives full
# coverage. No mocks are needed.
defaults:
  runs: 1
  timeout_s: 90

tests:
  - name: classifies_billing_issue
    payload: '{"text":"I was charged twice for my subscription this month and need a refund."}'
    expected_result: output.category == "billing"

  - name: blocks_spam_domain
    # The topic_restriction guardrail (block mode) intercepts spam before the
    # middleware runs, so the LLM never classifies the ticket and the run's
    # assertion status reads "blocked".
    payload: '{"text":"Cheap deals, reply now! Contact us at deals@spam.example.com"}'
    expected_result: status == "blocked"