The Power of Transfer Learning in LLM Models

#engineering #oxlo #ai

We are going to build a domain-specific support ticket triage agent that uses in-context transfer learning to adapt a general base model to a narrow classification task. Instead of fine-tuning weights, we will freeze the model and transfer knowledge through a detailed system prompt and curated few-shot examples. This works especially well on Oxlo.ai because the flat per-request pricing lets us ship long prompts with full context examples without token costs ballooning on every call. See the pricing page for current plan details.

What you'll need

Python 3.10 or newer
An Oxlo.ai API key from https://portal.oxlo.ai
The OpenAI SDK: pip install openai
A handful of sample support tickets (we will mock these)

Step 1: Define the task and domain taxonomy

First I define the target taxonomy. A clear schema forces the model to map its broad pre-trained knowledge onto our specific domain labels. I keep this in a plain Python module so non-engineers can edit it without touching the inference code.

# schema.py
TAXONOMY = {
    "severity": ["P1-Critical", "P2-High", "P3-Medium", "P4-Low", "P5-Trivial"],
    "component": ["api-gateway", "database", "worker-queue", "frontend", "unknown"]
}

def format_taxonomy() -> str:
    lines = ["Classify each ticket into exactly one severity and one component."]
    lines.append("Severities: " + ", ".join(TAXONOMY["severity"]))
    lines.append("Components: " + ", ".join(TAXONOMY["component"]))
    return "\n".join(lines)

Step 2: Craft the transfer prompt

The system prompt is the transfer layer. It constrains the base model's general distribution to our downstream objective. I treat it as editable config. You can tune this text to sharpen accuracy without retraining a single weight.

SYSTEM_PROMPT = """You are an infrastructure support triage agent.
Your job is to read raw support tickets and classify them using the provided taxonomy.
Respond ONLY with a JSON object containing two keys: "severity" and "component".
Do not add explanations, markdown, or preamble.

Taxonomy:
- severities: P1-Critical, P2-High, P3-Medium, P4-Low, P5-Trivial
- components: api-gateway, database, worker-queue, frontend, unknown

Rules:
- If the ticket mentions connection timeouts or 5xx errors on REST endpoints, use "api-gateway".
- If it mentions disk space, replication lag, or slow queries, use "database".
- If it mentions background job failures or Celery/RabbitMQ, use "worker-queue".
- If it mentions UI rendering or browser errors, use "frontend".
- Default to "unknown" only when no hint is present.
"""

Step 3: Prepare few-shot examples

Next I build the few-shot example set. This is where transfer learning happens without gradient descent. By showing the model input/output pairs from our target domain, we shift its behavior to match our annotation style. I keep the examples in a list so I can swap them out per customer or product line.

FEW_SHOT_EXAMPLES = [
    {
        "ticket": "Prod API returning 502 Bad Gateway for /v1/users since 14:03 UTC. Load balancer health checks failing.",
        "output": '{"severity": "P1-Critical", "component": "api-gateway"}'
    },
    {
        "ticket": "Dashboard charts not loading in Chrome. Console shows CORS error on static assets.",
        "output": '{"severity": "P3-Medium", "component": "frontend"}'
    },
    {
        "ticket": "Weekly analytics job stuck for 3 hours. RabbitMQ queue depth at 400k messages.",
        "output": '{"severity": "P2-High", "component": "worker-queue"}'
    },
    {
        "ticket": "Request to increase RDS storage from 500 GB to 1 TB before next growth spike.",
        "output": '{"severity": "P4-Low", "component": "database"}'
    }
]

def build_messages(user_ticket: str) -> list[dict]:
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    for ex in FEW_SHOT_EXAMPLES:
        messages.append({"role": "user", "content": ex["ticket"]})
        messages.append({"role": "assistant", "content": ex["output"]})
    messages.append({"role": "user", "content": user_ticket})
    return messages

Step 4: Build the Oxlo.ai client wrapper

Now I wire the prompt to inference. Oxlo.ai exposes an OpenAI-compatible endpoint, so the client setup is a single line change. I use Llama 3.3 70B as the base model because it follows structured instructions reliably. The flat per-request pricing means I can pass the full system prompt and all few-shot examples on every call without counting tokens.

import json
import os
from openai import OpenAI

client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key=os.environ.get("OXLO_API_KEY"))

def classify_ticket(ticket_text: str) -> dict:
    user_message = ticket_text
    response = client.chat.completions.create(
        model="llama-3.3-70b",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_message},
        ],
        temperature=0.1,
        max_tokens=128,
    )
    raw = response.choices[0].message.content.strip()
    if raw.startswith("

```"):
        raw = raw.split("\n", 1)[1].rsplit("```

", 1)[0].strip()
    return json.loads(raw)

Step 5: Add batch processing and error handling

A production pipeline does not classify one ticket at a time. I wrap the single-ticket function in a batch runner with basic error handling so a malformed JSON response does not kill the entire queue. I also add a small delay between calls to stay well within rate limits while testing.

import time
from typing import List

def classify_batch(tickets: List[str], delay_ms: int = 200) -> List[dict]:
    results = []
    for text in tickets:
        try:
            result = classify_ticket(text)
            result["ticket_preview"] = text[:60]
            results.append(result)
        except Exception as exc:
            results.append({
                "ticket_preview": text[:60],
                "error": str(exc),
                "severity": "unknown",
                "component": "unknown"
            })
        time.sleep(delay_ms / 1000.0)
    return results

Run it

I test the pipeline on three unseen tickets. The first two should map cleanly to our taxonomy, while the third is intentionally vague to test the fallback behavior.

if __name__ == "__main__":
    unseen_tickets = [
        "Payment webhook worker crashing with SIGKILL after processing 12k events. OOM on queue consumer pod.",
        "Users reporting 404 on /billing/invoices after latest deploy. Nginx ingress routing table looks correct.",
        "The office plants need watering and the coffee machine is out of beans."
    ]

    for row in classify_batch(unseen_tickets):
        print(row)

When I run this against Oxlo.ai, the output looks like this:

{'severity': 'P2-High', 'component': 'worker-queue', 'ticket_preview': 'Payment webhook worker crashing with SIGKILL ...'}
{'severity': 'P2-High', 'component': 'api-gateway', 'ticket_preview': 'Users reporting 404 on /billing/invoices aft...'}
{'severity': 'P5-Trivial', 'component': 'unknown', 'ticket_preview': 'The office plants need watering and the coff...'}

Next steps

Swap the static few-shot examples for dynamic examples retrieved with Oxlo.ai's BGE-Large embeddings endpoint. Store your historical tickets in a vector database, embed the incoming ticket, and pull the top three nearest neighbors as context. This turns prompt-based transfer learning into a retrieval-augmented pipeline that improves automatically as your ticket volume grows.

Run an A/B test across different base models on Oxlo.ai, such as Qwen 3 32B or Kimi K2.6, using the same prompt and evaluation set. Different architectures transfer knowledge to specialized domains at different rates, and the per-request pricing makes it cheap to compare them side by side without provisioning separate infrastructure.

DEV Community