Part A: The Project and the Toolkit

cd supportdesk
python3.11 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
export PYTHONPATH=.
python -m pytest -q tests/test_m01_foundations.py

  ........                                                                 [100%]
  8 passed in 13.28s

{
  "id": "T-1001",
  "subject": "Charged twice this month",
  "body": "Hi, my card was charged 288 USD twice on 3 September for the Team plan (invoice INV-2026-004512). Please refund the duplicate.",
  "customer_tier": "team",
  "language": "en",
  "split": "dev",
  "gold": {"category": "billing", "priority": "high", "kb_article": "billing-refunds", "answerable": true}
}

"""Module 1: a first look at the Brightlane dataset the whole course uses."""
from collections import Counter

from supportdesk.data import CATEGORIES, PRIORITIES, get_article, load_articles, load_tickets

tickets = load_tickets()
dev, test = load_tickets("dev"), load_tickets("test")
articles = load_articles()

print(f"tickets: {len(tickets)} (dev {len(dev)}, test {len(test)})")
print("languages:", dict(Counter(t.language for t in tickets).most_common()))
print("categories:", {c: sum(t.gold["category"] == c for t in tickets) for c in CATEGORIES})
print("priorities:", {p: sum(t.gold["priority"] == p for t in tickets) for p in PRIORITIES})
print("answerable from the help center:", sum(t.gold["answerable"] for t in tickets))
print(f"articles: {len(articles)}:", ", ".join(a.id for a in articles))

first = tickets[0]
print("\n--- one ticket, as an agent reads it ---")
print(first.id, first.customer_tier, first.language)
print(first.text)
print("gold:", first.gold)

article = get_article(first.gold["kb_article"])
print("\n--- the article that answers it ---")
print(article.title, article.tags)
print(article.body[:160] + "...")

  tickets: 72 (dev 48, test 24)
  languages: {'en': 64, 'de': 3, 'es': 2, 'ja': 2, 'hi': 1}
  categories: {'billing': 16, 'cancellation': 6, 'account_access': 17, 'bug': 9, 'how_to': 18, 'feature_request': 6}
  priorities: {'low': 24, 'normal': 31, 'high': 13, 'urgent': 4}
  answerable from the help center: 62
  articles: 12: account-login, account-sso, billing-invoices, billing-plans, billing-refunds, boards-automations, data-privacy, exports-data, feature-requests, integrations-slack, mobile-app, status-incidents

  --- one ticket, as an agent reads it ---
  T-1001 team en
  Subject: Charged twice this month

  Hi, my card was charged 288 USD twice on 3 September for the Team plan (invoice INV-2026-004512). Please refund the duplicate.
  gold: {'category': 'billing', 'priority': 'high', 'kb_article': 'billing-refunds', 'answerable': True}

  --- the article that answers it ---
  Refunds and cancellations ('billing', 'cancellation')
  You can cancel at any time from Settings > Billing > Cancel plan. Monthly plans stay active until the end of the current billing period and are not refunded pro...

"""Load the Brightlane support dataset: tickets with gold labels and help-center articles."""
from __future__ import annotations

import json
from dataclasses import dataclass
from pathlib import Path

DATA_DIR = Path(__file__).resolve().parents[1] / "data"
CATEGORIES = ("billing", "cancellation", "account_access", "bug", "how_to", "feature_request")
PRIORITIES = ("low", "normal", "high", "urgent")


@dataclass(frozen=True)
class Ticket:
    id: str
    subject: str
    body: str
    customer_tier: str
    language: str
    split: str
    gold: dict

    @property
    def text(self) -> str:
        """Subject and body as one string, the way an agent reads a ticket."""
        return f"Subject: {self.subject}\n\n{self.body}"


@dataclass(frozen=True)
class Article:
    id: str
    title: str
    tags: tuple[str, ...]
    body: str


def load_tickets(split: str | None = None) -> list[Ticket]:
    """All tickets, or only the 'dev' or 'test' split. Order is stable."""
    tickets = []
    with (DATA_DIR / "tickets.jsonl").open(encoding="utf-8") as f:
        for line in f:
            t = Ticket(**json.loads(line))
            if split is None or t.split == split:
                tickets.append(t)
    return tickets


def load_articles() -> list[Article]:
    """Help-center articles from data/kb, sorted by id."""
    articles = []
    for path in sorted((DATA_DIR / "kb").glob("*.md")):
        text = path.read_text(encoding="utf-8")
        _, header, body = text.split("---\n", 2)
        meta = dict(line.split(": ", 1) for line in header.strip().splitlines())
        tags = tuple(t.strip() for t in meta.get("tags", "").strip("[]").split(",") if t.strip())
        articles.append(Article(path.stem, meta["title"], tags, body.strip()))
    return articles


def get_article(article_id: str) -> Article:
    for article in load_articles():
        if article.id == article_id:
            return article
    raise KeyError(f"No article {article_id!r}")

"""One helper for three LLM providers: Groq, Gemini, and Ollama.

All three expose an OpenAI-compatible Chat Completions endpoint, so a single
client library (`openai`) talks to each of them. Choose a provider with the
LLM_PROVIDER environment variable (groq, gemini, ollama) and override the
model with LLM_MODEL. Keys come from GROQ_API_KEY and GEMINI_API_KEY; Ollama
runs locally and needs no key.

Every call returns a ChatResult with the text, any tool calls, token usage,
and latency, because the course measures cost and speed on every example.
"""
from __future__ import annotations

import json
import os
import time
from collections.abc import Iterator
from dataclasses import dataclass, field
from typing import Any

from openai import OpenAI

PROVIDERS: dict[str, dict[str, str]] = {
    "groq": {
        "base_url": "https://api.groq.com/openai/v1",
        "key_env": "GROQ_API_KEY",
        "default_model": "openai/gpt-oss-120b",
    },
    "gemini": {
        "base_url": "https://generativelanguage.googleapis.com/v1beta/openai/",
        "key_env": "GEMINI_API_KEY",
        "default_model": "gemini-3.5-flash",
    },
    "ollama": {
        "base_url": "http://localhost:11434/v1",
        "key_env": "",
        "default_model": "qwen3:8b",
    },
}


@dataclass
class ToolCall:
    id: str
    name: str
    arguments: dict[str, Any]
    raw_arguments: str = ""


@dataclass
class Usage:
    input_tokens: int = 0
    output_tokens: int = 0
    cached_tokens: int = 0
    reasoning_tokens: int = 0


@dataclass
class ChatResult:
    text: str
    tool_calls: list[ToolCall] = field(default_factory=list)
    usage: Usage = field(default_factory=Usage)
    latency_ms: float = 0.0
    finish_reason: str = ""
    model: str = ""
    provider: str = ""

    def as_message(self) -> dict[str, Any]:
        """The assistant message to append to the conversation history."""
        message: dict[str, Any] = {"role": "assistant", "content": self.text or ""}
        if self.tool_calls:
            message["tool_calls"] = [
                {
                    "id": call.id,
                    "type": "function",
                    "function": {"name": call.name, "arguments": call.raw_arguments or json.dumps(call.arguments)},
                }
                for call in self.tool_calls
            ]
        return message


def resolve(provider: str | None = None, model: str | None = None) -> tuple[str, str]:
    """Pick the provider and model from arguments first, then the environment, then defaults."""
    name = (provider or os.environ.get("LLM_PROVIDER", "groq")).lower()
    if name not in PROVIDERS:
        raise ValueError(f"Unknown LLM_PROVIDER {name!r}. Choose one of {sorted(PROVIDERS)}.")
    return name, model or os.environ.get("LLM_MODEL") or PROVIDERS[name]["default_model"]


def make_client(provider: str) -> OpenAI:
    """Build an OpenAI-compatible client for one provider, reading its key from the environment."""
    settings = PROVIDERS[provider]
    if settings["key_env"]:
        api_key = os.environ.get(settings["key_env"], "")
        if not api_key:
            raise RuntimeError(f"Set {settings['key_env']} in your environment to use {provider}.")
        base_url = settings["base_url"]
    else:
        api_key = "ollama"
        base_url = os.environ.get("OLLAMA_BASE_URL", settings["base_url"])
    return OpenAI(api_key=api_key, base_url=base_url, max_retries=2, timeout=120.0)


def _usage(raw: Any) -> Usage:
    if raw is None:
        return Usage()
    prompt_details = getattr(raw, "prompt_tokens_details", None)
    completion_details = getattr(raw, "completion_tokens_details", None)
    return Usage(
        input_tokens=raw.prompt_tokens or 0,
        output_tokens=raw.completion_tokens or 0,
        cached_tokens=(getattr(prompt_details, "cached_tokens", 0) or 0) if prompt_details else 0,
        reasoning_tokens=(getattr(completion_details, "reasoning_tokens", 0) or 0) if completion_details else 0,
    )


def _parse_arguments(raw: str) -> dict[str, Any]:
    try:
        value = json.loads(raw or "{}")
    except json.JSONDecodeError:
        return {"_unparseable": raw}
    return value if isinstance(value, dict) else {"_value": value}


def chat(
    messages: list[dict[str, Any]],
    *,
    provider: str | None = None,
    model: str | None = None,
    temperature: float | None = 0.0,
    top_p: float | None = None,
    max_tokens: int | None = None,
    stop: list[str] | None = None,
    seed: int | None = None,
    tools: list[dict[str, Any]] | None = None,
    tool_choice: str | dict[str, Any] | None = None,
    response_format: dict[str, Any] | None = None,
    reasoning_effort: str | None = None,
    extra: dict[str, Any] | None = None,
) -> ChatResult:
    """Send one Chat Completions request and return text, tool calls, usage, and latency.

    Parameters left as None are not sent, so each provider uses its own default.
    `extra` passes provider-specific fields through unchanged.
    """
    name, model_id = resolve(provider, model)
    kwargs: dict[str, Any] = {"model": model_id, "messages": messages}
    optional = {
        "temperature": temperature, "top_p": top_p, "max_tokens": max_tokens, "stop": stop,
        "seed": seed, "tools": tools, "tool_choice": tool_choice,
        "response_format": response_format, "reasoning_effort": reasoning_effort,
    }
    kwargs.update({k: v for k, v in optional.items() if v is not None})
    if extra:
        kwargs["extra_body"] = extra
    started = time.perf_counter()
    response = make_client(name).chat.completions.create(**kwargs)
    latency_ms = (time.perf_counter() - started) * 1000
    choice = response.choices[0]
    calls = [
        ToolCall(c.id, c.function.name, _parse_arguments(c.function.arguments), c.function.arguments or "")
        for c in (choice.message.tool_calls or [])
        if c.type == "function"
    ]
    return ChatResult(
        text=choice.message.content or "",
        tool_calls=calls,
        usage=_usage(response.usage),
        latency_ms=round(latency_ms, 1),
        finish_reason=choice.finish_reason or "",
        model=model_id,
        provider=name,
    )


@dataclass
class StreamStats:
    ttft_ms: float = 0.0
    total_ms: float = 0.0
    chunks: int = 0


def stream_chat(
    messages: list[dict[str, Any]],
    stats: StreamStats | None = None,
    *,
    provider: str | None = None,
    model: str | None = None,
    temperature: float | None = 0.0,
    max_tokens: int | None = None,
) -> Iterator[str]:
    """Yield text pieces as they arrive. Fills `stats` with time to first token and total time."""
    name, model_id = resolve(provider, model)
    kwargs: dict[str, Any] = {"model": model_id, "messages": messages, "stream": True}
    if temperature is not None:
        kwargs["temperature"] = temperature
    if max_tokens is not None:
        kwargs["max_tokens"] = max_tokens
    stats = stats if stats is not None else StreamStats()
    started = time.perf_counter()
    for chunk in make_client(name).chat.completions.create(**kwargs):
        if not chunk.choices:
            continue
        piece = chunk.choices[0].delta.content or ""
        if piece:
            if stats.chunks == 0:
                stats.ttft_ms = round((time.perf_counter() - started) * 1000, 1)
            stats.chunks += 1
            yield piece
    stats.total_ms = round((time.perf_counter() - started) * 1000, 1)

"""A scripted stand-in for `llm.chat`, for running examples and tests without an API key.

It has the same call signature as `supportdesk.llm.chat` and returns the same
ChatResult type. It is NOT a language model: it replays scripted replies or
applies a rule you give it. Use it to test plumbing (parsing, loops, retries,
budgets), never to measure model quality.
"""
from __future__ import annotations

import itertools
from collections.abc import Callable
from typing import Any

from supportdesk.llm import ChatResult, Usage
from supportdesk.tokens import count_messages, count_tokens

Responder = Callable[[list[dict[str, Any]], dict[str, Any]], ChatResult | str]


class ScriptedLLM:
    """Replays a list of replies in order, or calls `responder(messages, kwargs)`."""

    _ids = itertools.count(1)

    def __init__(self, replies: list[ChatResult | str] | None = None, responder: Responder | None = None,
                 model: str = "scripted-stand-in") -> None:
        if (replies is None) == (responder is None):
            raise ValueError("Pass exactly one of `replies` or `responder`.")
        self.replies = list(replies or [])
        self.responder = responder
        self.model = model
        self.calls: list[dict[str, Any]] = []

    def __call__(self, messages: list[dict[str, Any]], **kwargs: Any) -> ChatResult:
        self.calls.append({"messages": [dict(m) for m in messages], **kwargs})
        if self.responder is not None:
            reply = self.responder(messages, kwargs)
        elif self.replies:
            reply = self.replies.pop(0)
        else:
            raise RuntimeError("ScriptedLLM ran out of scripted replies.")
        result = ChatResult(text=reply) if isinstance(reply, str) else reply
        if result.usage.input_tokens == 0:
            result.usage = Usage(input_tokens=count_messages(messages), output_tokens=count_tokens(result.text or ""))
        result.model = result.model or self.model
        result.provider = result.provider or "stand-in"
        result.finish_reason = result.finish_reason or ("tool_calls" if result.tool_calls else "stop")
        return result

"""Module 1: the first call to a hosted model, and the same call through the stand-in.

With a key set (for example GROQ_API_KEY, or LLM_PROVIDER=ollama with Ollama running),
this sends a real request through supportdesk.llm.chat. Without one, it falls back to
ScriptedLLM, which is NOT a model: it replays a scripted reply so the plumbing runs.
"""
import os

from supportdesk import llm
from supportdesk.data import CATEGORIES, load_tickets
from supportdesk.stand_in import ScriptedLLM

ticket = next(t for t in load_tickets() if t.id == "T-1008")
messages = [
    {"role": "system", "content": "You triage support tickets for Brightlane, a project-management SaaS. "
                                  f"Reply with one category from: {', '.join(CATEGORIES)}. "
                                  "Then one short sentence explaining why."},
    {"role": "user", "content": ticket.text},
]

provider, model = llm.resolve()
key_env = llm.PROVIDERS[provider]["key_env"]
use_real = provider == "ollama" or bool(os.environ.get(key_env))

if use_real:
    chat = llm.chat
    print(f"calling {provider} / {model}")
else:
    chat = ScriptedLLM(replies=["account_access\nThe customer is locked out after failed password attempts."])
    print(f"no {key_env} set: using ScriptedLLM (scripted reply, not model output)")

result = chat(messages, temperature=0.0, max_tokens=200)
print("text:         ", result.text.replace("\n", " | "))
print("provider/model:", result.provider, "/", result.model)
print("finish_reason: ", result.finish_reason)
print("usage:         ", result.usage)
print("latency_ms:    ", result.latency_ms)
print("gold category: ", ticket.gold["category"])

  no GROQ_API_KEY set: using ScriptedLLM (scripted reply, not model output)
  text:          account_access | The customer is locked out after failed password attempts.
  provider/model: stand-in / scripted-stand-in
  finish_reason:  stop
  usage:          Usage(input_tokens=90, output_tokens=13, cached_tokens=0, reasoning_tokens=0)
  latency_ms:     0.0
  gold category:  account_access

calling groq / openai/gpt-oss-120b
text:          account_access | The customer is locked out after too many failed password attempts and needs access restored.
provider/model: groq / openai/gpt-oss-120b
finish_reason:  stop
usage:          Usage(input_tokens=131, output_tokens=74, cached_tokens=0, reasoning_tokens=48)
latency_ms:     640.2
gold category:  account_access