# Requests waiting for a GPU slot — the primary scaling signal
vllm:num_requests_waiting{namespace="inference"}
# Running sequences — are we at max-num-seqs capacity?
vllm:num_requests_running{namespace="inference"}

# KV cache fill rate — aim for 70-85% at peak, not 95%+
vllm:gpu_cache_usage_perc{namespace="inference"}

# Tokens generated per second (decode throughput)
rate(vllm:generation_tokens_total{namespace="inference"}[5m])
# Tokens processed in prefill per second
rate(vllm:prompt_tokens_total{namespace="inference"}[5m])

# p95 time to first token
histogram_quantile(0.95, rate(vllm:time_to_first_token_seconds_bucket{namespace="inference"}[10m]))
# p95 time per output token
histogram_quantile(0.95, rate(vllm:time_per_output_token_seconds_bucket{namespace="inference"}[10m]))
# p95 end-to-end latency
histogram_quantile(0.95, rate(vllm:e2e_request_latency_seconds_bucket{namespace="inference"}[10m]))

# GPU compute utilization — should be 70-90% under load
DCGM_FI_DEV_GPU_UTIL{namespace="inference"}
# GPU memory used vs total
DCGM_FI_DEV_FB_USED{namespace="inference"} / DCGM_FI_DEV_FB_TOTAL{namespace="inference"}
# GPU temperature — alert above 85°C
DCGM_FI_DEV_GPU_TEMP{namespace="inference"}

<outbound>
  <base />
  <llm-emit-token-metric namespace="InferenceTokens">
    <dimension name="consumer-id" value="@(context.Subscription.Id)" />
    <dimension name="model" value="@(context.Request.Body.As<JObject>(preserveContent: true)["model"]?.ToString() ?? "unknown")" />
    <dimension name="environment" value="@(context.Deployment.ServiceName.Contains("prod") ? "prod" : "dev")" />
  </llm-emit-token-metric>
</outbound>

customMetrics
| where name == "InferenceTokens"
| summarize total_tokens = sum(value) by tostring(customDimensions["consumer-id"]), bin(timestamp, 1d)
| order by total_tokens desc

helm repo add langfuse https://langfuse.github.io/langfuse-k8s
helm upgrade --install langfuse langfuse/langfuse \
  --namespace langfuse --create-namespace \
  --set langfuse.nextauth.secret="$(openssl rand -hex 32)" \
  --set langfuse.salt="$(openssl rand -hex 16)" \
  --set postgresql.enabled=true \
  --set postgresql.auth.database=langfuse \
  --set langfuse.resources.requests.memory="512Mi" \
  --set langfuse.resources.requests.cpu="250m"

from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context
langfuse = Langfuse(
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
    secret_key=os.environ["LANGFUSE_SECRET_KEY"],
    host=os.environ["LANGFUSE_HOST"],  # internal cluster URL
)
@observe()
def generate_response(user_query: str, session_id: str) -> str:
    langfuse_context.update_current_trace(
        user_id=session_id,
        tags=["production", "customer-support"],
    )
    # Retrieve context (if RAG)
    chunks = retrieve(user_query)
    langfuse_context.update_current_observation(
        input={"query": user_query, "chunks_retrieved": len(chunks)},
    )
    # Call vLLM (OpenAI-compatible endpoint)
    response = openai_client.chat.completions.create(
        model="phi-4-mini-instruct",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"Context:\n{chunks}\n\nQuestion: {user_query}"}
        ],
        max_tokens=512,
    )
    output = response.choices[0].message.content
    # Attach quality score if you have one
    langfuse_context.score_current_trace(
        name="groundedness",
        value=check_groundedness(output, chunks),
    )
    return output

<!-- APIM inbound: generate and forward trace ID -->
<set-header name="X-Trace-Id" exists-action="skip">
  <value>@(Guid.NewGuid().ToString())</value>
</set-header>
<set-variable name="traceId" value="@(context.Request.Headers.GetValueOrDefault("X-Trace-Id"))" />

# promptfooconfig.yaml
tests:
  - vars:
      query: "What port does vLLM listen on by default?"
    assert:
      - type: contains
        value: "8000"
      - type: not-contains
        value: "8080"        # common wrong answer
      - type: javascript
        value: |
          output.length < 200   # reject verbose responses

tests:
  - vars:
      context: "The document says X, Y, and Z."
      query: "Summarize the key points."
    assert:
      - type: llm-rubric
        value: |
          The response should:
          1. Mention X, Y, and Z from the provided context
          2. Not introduce information not present in the context
          3. Be written in 2-4 sentences
          4. Not start with "Certainly!" or similar filler

# In your test suite
def check_json_output(output: str, context: dict) -> dict:
    """Validate structured output is valid JSON matching expected schema."""
    import json
    from jsonschema import validate
    expected_schema = {
        "type": "object",
        "required": ["category", "confidence", "reason"],
        "properties": {
            "category": {"type": "string", "enum": ["billing", "technical", "account"]},
            "confidence": {"type": "number", "minimum": 0, "maximum": 1},
            "reason": {"type": "string"},
        }
    }
    try:
        parsed = json.loads(output)
        validate(parsed, expected_schema)
        return {"pass": True, "score": parsed["confidence"]}
    except Exception as e:
        return {"pass": False, "reason": str(e)}

JUDGE_PROMPT = """You are an expert evaluator for a {task_type} system.
Evaluate the following response on a scale of 1-5 for {dimension}:
1 = Completely wrong or harmful
2 = Mostly wrong with minor correct elements
3 = Partially correct but missing key information
4 = Mostly correct with minor issues
5 = Completely correct and well-formed
Task: {task_description}
User question: {user_query}
{context_block}
Response to evaluate:
{response}
Provide your evaluation in this exact JSON format:
{{
  "score": <1-5>,
  "reasoning": "<one sentence explaining the score>",
  "key_issues": ["<issue 1>", "<issue 2>"]
}}
Do not consider response length in your score. Evaluate only accuracy and completeness."""

from datasets import Dataset
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
# Build evaluation dataset
data = {
    "question": ["What is the maximum context length for Phi-4 Mini?"],
    "answer": ["Phi-4 Mini supports up to 128K context tokens."],          # model output
    "contexts": [["Phi-4 Mini has a 128K context window and 3.8B params"]], # retrieved chunks
    "ground_truth": ["Phi-4 Mini has a 128K token context window."],       # expected answer
}
dataset = Dataset.from_dict(data)
results = evaluate(
    dataset=dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
    llm=your_azure_openai_client,  # judge model — use a stronger model than the one you're evaluating
)

CREATE TABLE eval_runs (
    run_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    run_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    model VARCHAR(100),
    prompt_version VARCHAR(20),
    dataset_name VARCHAR(100),
    pass_rate FLOAT,
    avg_faithfulness FLOAT,
    avg_relevancy FLOAT,
    p95_latency_ms INT
);

[Role] You are a {specific role} for {specific company/context}.
[Scope] You help users with: {list of in-scope tasks}.
You do not: {list of out-of-scope tasks}.
[Format] Respond in {format description}.
{example if format is complex}
[Constraints]
- {constraint 1}
- {constraint 2}
[Fallback] If you cannot answer from the provided context, say exactly:
"{fallback phrase}" — do not fabricate information.

SYSTEM_PROMPT = """You are a technical support classifier.
Classify the user's issue into one of: [billing, technical, account, other].
Examples:
User: "My invoice shows a double charge for March."
Output: {"category": "billing", "confidence": 0.95, "reason": "Payment/invoice dispute"}
User: "The API keeps returning 429 errors."
Output: {"category": "technical", "confidence": 0.9, "reason": "Rate limiting error"}
User: "How do I reset my password?"
Output: {"category": "account", "confidence": 0.98, "reason": "Credential management"}
Always output valid JSON matching the schema above. Never add extra fields."""

Question: If a T4 GPU has 16GB VRAM and a model uses 14.6GB for weights,
how many concurrent sequences can it run at max-model-len 4096?
Think through this step by step, then give the final answer.

Question: Calculate GPU tier needed for Llama 3.3 70B at int8.
Reasoning:
- Parameters: 70.6B
- int8 bytes per param: 1.0
- Weights memory: 70.6B × 1.0 = 70.6 GB
- Apply 1.3× headroom: 70.6 × 1.3 = 91.8 GB needed
- T4 (16GB): too small
- A100 80GB: 80 × 0.90 = 72 GB usable < 91.8 GB: too small
- 2× A100 80GB: 160 × 0.90 = 144 GB usable > 91.8 GB: sufficient
Answer: 2× A100 80GB (NC48ads_A100_v4)
Question: Calculate GPU tier needed for Phi-4 Mini at fp16.
Reasoning:
<model completes the pattern>

response = client.chat.completions.create(
    model="phi-4-mini-instruct",
    messages=[...],
    response_format={"type": "json_object"},  # forces JSON output
)

from vllm import LLM, SamplingParams
schema = {
    "type": "object",
    "properties": {
        "category": {"type": "string", "enum": ["billing", "technical", "account"]},
        "confidence": {"type": "number"},
    },
    "required": ["category", "confidence"]
}
sampling_params = SamplingParams(
    guided_decoding={"json": json.dumps(schema)}  # tokens that violate schema are masked
)

from guardrails import Guard
from guardrails.hub import ValidJSON, ValidChoices
guard = Guard().use_many(
    ValidJSON(),
    ValidChoices(choices=["billing", "technical", "account"], on_fail="reask"),
)
response = guard(
    openai_client.chat.completions.create,
    prompt="Classify this support ticket: ...",
    model="phi-4-mini-instruct",
    max_tokens=200,
)

def manage_context(messages: list, model_max_tokens: int, reserve_tokens: int = 1000) -> list:
    """Summarize old messages when context approaches limit."""
    current_tokens = count_tokens(messages)
    limit = model_max_tokens - reserve_tokens  # reserve for completion
    if current_tokens < limit * 0.7:
        return messages  # no action needed
    # Keep system prompt + last 4 turns + summarize the rest
    system = [m for m in messages if m["role"] == "system"]
    recent = messages[-4:]
    to_summarize = [m for m in messages if m not in system and m not in recent]
    if not to_summarize:
        return messages
    summary = summarize_conversation(to_summarize)  # call LLM to summarize
    summary_msg = {"role": "assistant", "content": f"[Previous conversation summary: {summary}]"}
    return system + [summary_msg] + recent

from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,       # tokens, not characters
    chunk_overlap=50,     # ~10% overlap to avoid boundary splits
    length_function=lambda text: len(tokenizer.encode(text)),  # use target model's tokenizer
)

from langchain.text_splitter import SpacyTextSplitter
# Respects sentence boundaries — never splits mid-sentence
splitter = SpacyTextSplitter(chunk_size=512)

from langchain.text_splitter import Language, RecursiveCharacterTextSplitter
# Splits at function/class boundaries, not arbitrary character positions
splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=1000,
    chunk_overlap=100,
)

chunks_with_metadata = [
    {
        "content": chunk.page_content,
        "metadata": {
            "source": document_path,
            "section": extract_section_heading(chunk),
            "doc_type": "technical_guide",
            "last_updated": document_date,
            "language": "en",
        }
    }
    for chunk in chunks
]

from azure.search.documents import SearchClient
from azure.search.documents.models import VectorizedQuery
def hybrid_retrieve(query: str, top_k: int = 20) -> list:
    """Combine BM25 and vector search, return top-k by reciprocal rank fusion."""
    query_embedding = embed(query)
    results = search_client.search(
        search_text=query,                                    # BM25 path
        vector_queries=[
            VectorizedQuery(
                vector=query_embedding,
                k_nearest_neighbors=top_k,
                fields="content_vector"                       # dense path
            )
        ],
        query_type="semantic",                                # rerank with semantic model
        semantic_configuration_name="inference-config",
        top=top_k,
        select=["content", "source", "section", "metadata"],
    )
    return list(results)

results = search_client.search(
    search_text=query,
    filter="doc_type eq 'technical_guide' and last_updated ge 2026-01-01",
    top=10,
)

def hyde_retrieve(query: str, llm_client, top_k: int = 5) -> list:
    """Retrieve using a hypothetical answer embedding instead of the raw query."""
    # Generate a hypothetical ideal answer (doesn't need to be accurate)
    hyde_response = llm_client.chat.completions.create(
        model="phi-4-mini-instruct",
        messages=[{
            "role": "user",
            "content": f"Write a 3-sentence technical explanation that would answer: {query}"
        }],
        max_tokens=150,
    )
    hypothetical_answer = hyde_response.choices[0].message.content
    # Embed the hypothetical answer and retrieve
    embedding = embed(hypothetical_answer)
    return vector_search(embedding, top_k=top_k)

from sentence_transformers import CrossEncoder
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
def rerank(query: str, candidates: list[str], top_n: int = 5) -> list[str]:
    """Score query-document pairs and return top_n."""
    pairs = [(query, doc) for doc in candidates]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(scores, candidates), reverse=True)
    return [doc for _, doc in ranked[:top_n]]

def multi_hop_retrieve(question: str, max_hops: int = 3) -> list:
    all_contexts = []
    current_query = question
    for hop in range(max_hops):
        chunks = retrieve(current_query, top_k=3)
        all_contexts.extend(chunks)
        # Ask the model: do we have enough information? If not, what do we still need?
        reflection = llm_client.chat.completions.create(
            model="phi-4-mini-instruct",
            messages=[{
                "role": "user",
                "content": f"""Question: {question}
Retrieved so far:
{format_chunks(all_contexts)}
Can you fully answer the question with the above context?
If yes, respond: "COMPLETE"
If no, respond with the specific follow-up question needed to find the missing information."""
            }],
            max_tokens=100,
        ).choices[0].message.content
        if reflection.strip() == "COMPLETE" or hop == max_hops - 1:
            break
        current_query = reflection  # next hop uses the model's follow-up question
    return all_contexts

{"messages": [
    {"role": "system", "content": "You are a customer support agent for AKS AI Lab."},
    {"role": "user", "content": "My GPU node didn't provision. What do I check?"},
    {"role": "assistant", "content": "Check these in order:\n1. Run `kubectl get nodeclaim -n karpenter` — look for a Failed NodeClaim\n2. Check AKS quota: `az vm list-usage -l eastus | grep NC`\n3. Check the Karpenter controller logs: `kubectl logs -n karpenter deployment/karpenter | grep ERROR`"}
]}

apiVersion: kaito.sh/v1alpha1
kind: Workspace
metadata:
  name: finetune-mistral-7b
  namespace: inference
spec:
  resource:
    instanceType: "Standard_NC24ads_A100_v4"   # A100 80GB for training
    labelSelector:
      matchLabels:
        apps: mistral-7b-finetune
  tuning:
    preset:
      name: mistral-7b-v0.3
    method: qlora
    input:
      urls:
        - "https://<storage>.blob.core.windows.net/training/dataset.jsonl"  # Workload Identity auth
    output:
      image: "<your-acr>.azurecr.io/mistral-7b-support:v1"
      imagePushSecret: acr-push-secret
    config:
      # LoRA hyperparameters
      lora_rank: 16
      lora_alpha: 32
      lora_dropout: 0.05
      target_modules: ["q_proj", "v_proj", "k_proj", "o_proj"]
      # Training config
      num_epochs: 3
      per_device_train_batch_size: 4
      gradient_accumulation_steps: 4   # effective batch = 16
      learning_rate: 2.0e-4
      warmup_ratio: 0.03
      lr_scheduler_type: "cosine"
      # Memory optimization
      gradient_checkpointing: true
      bf16: true                        # A100 supports bfloat16
      max_seq_length: 2048

def evaluate_fine_tuned_model(
    base_model_client,
    finetuned_model_client,
    eval_dataset: list[dict],
) -> dict:
    """Run eval on both models, compare on quality and format compliance."""
    results = {"base": [], "finetuned": []}
    for example in eval_dataset:
        for name, client in [("base", base_model_client), ("finetuned", finetuned_model_client)]:
            response = client.chat.completions.create(
                messages=example["messages"][:-1],  # exclude gold response
                max_tokens=512,
                temperature=0,
            )
            output = response.choices[0].message.content
            results[name].append({
                "output": output,
                "latency_ms": response.usage.completion_tokens * 30,  # rough estimate
                "format_valid": check_format(output, example["expected_format"]),
                "judge_score": llm_judge(example["messages"][-2]["content"], output),
            })
    return {
        "base_pass_rate": mean(r["format_valid"] for r in results["base"]),
        "finetuned_pass_rate": mean(r["format_valid"] for r in results["finetuned"]),
        "base_quality": mean(r["judge_score"] for r in results["base"]),
        "finetuned_quality": mean(r["judge_score"] for r in results["finetuned"]),
        "quality_delta": mean(r["judge_score"] for r in results["finetuned"]) - mean(r["judge_score"] for r in results["base"]),
    }

Cost per request = (prompt_tokens + completion_tokens) × $/token
                 = prompt_tokens × (GPU $/hr) / (prompt_throughput tok/s × 3600)
                 + completion_tokens × (GPU $/hr) / (generation_throughput tok/s × 3600)

Prompt cost:  500 / 15,000 × $0.53/hr / 3,600 = $0.0000049
Generation:   300 / 3,000  × $0.53/hr / 3,600 = $0.0000147
Total:        ~$0.000020 per request

# In manifests/vllm/vllm-standalone.yaml
args:
  - --enable-prefix-caching

# Cache hit rate — should be > 50% for chatbot use cases
vllm:cache_config_info{namespace="inference"}
rate(vllm:request_success_total{finished_reason="length", namespace="inference"}[5m])

import hashlib
import redis
cache = redis.Redis(host="redis.inference.svc.cluster.local", port=6379)
CACHE_TTL = 3600  # 1 hour
def cached_inference(messages: list, model: str, **kwargs) -> str:
    cache_key = hashlib.sha256(
        json.dumps({"messages": messages, "model": model}).encode()
    ).hexdigest()
    if cached := cache.get(cache_key):
        return json.loads(cached)
    response = llm_client.chat.completions.create(
        messages=messages, model=model, **kwargs
    )
    result = response.choices[0].message.content
    cache.setex(cache_key, CACHE_TTL, json.dumps(result))
    return result

from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
class SemanticCache:
    def __init__(self, similarity_threshold: float = 0.95):
        self.threshold = similarity_threshold
        self.cache: list[tuple[np.ndarray, str, str]] = []  # (embedding, query, response)
    def get(self, query: str) -> str | None:
        query_emb = np.array(embed(query))
        for stored_emb, stored_query, stored_response in self.cache:
            sim = cosine_similarity([query_emb], [stored_emb])[0][0]
            if sim >= self.threshold:
                return stored_response
        return None
    def set(self, query: str, response: str):
        self.cache.append((np.array(embed(query)), query, response))

ROUTING_PROMPT = """Classify this request's complexity:
- "simple": factual lookup, yes/no, short answer, format conversion
- "complex": multi-step reasoning, code generation, analysis, comparison
Request: {query}
Respond with only "simple" or "complex"."""
def cascade_route(query: str, context: str) -> str:
    # Use a tiny fast model to classify request complexity
    complexity = phi4_mini_client.chat.completions.create(
        messages=[{"role": "user", "content": ROUTING_PROMPT.format(query=query)}],
        max_tokens=5,
        temperature=0,
    ).choices[0].message.content.strip().lower()
    if complexity == "simple":
        client = phi4_mini_client   # T4, ~$0.53/hr
    else:
        client = llama70b_client    # 2× A100, ~$7.34/hr
    return client.chat.completions.create(
        messages=[{"role": "system", "content": context},
                  {"role": "user", "content": query}],
        max_tokens=512,
    ).choices[0].message.content

from llmlingua import PromptCompressor
compressor = PromptCompressor(
    model_name="microsoft/llmlingua-2-bert-base-multilingual-cased-meetingbank",
    use_llmlingua2=True,
)
def compress_context(question: str, retrieved_chunks: list[str]) -> str:
    context = "\n\n".join(retrieved_chunks)
    result = compressor.compress_prompt(
        context,
        question=question,
        target_token=512,          # compress to 512 tokens regardless of input size
        condition_in_question="after_condition",
        rank_method="llmlingua",
    )
    return result["compressed_prompt"]

# .github/workflows/prompt-eval.yml
name: Prompt Eval
on:
  pull_request:
    paths:
      - "prompts/**"
jobs:
  eval:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Run Promptfoo evals
        run: npx promptfoo eval --config promptfooconfig.yaml --output results.json
        env:
          AZURE_OPENAI_API_KEY: ${{ secrets.AZURE_OPENAI_API_KEY }}
          VLLM_ENDPOINT: ${{ secrets.VLLM_ENDPOINT }}
      - name: Parse results and gate
        run: |
          PASS_RATE=$(jq '.results.stats.successes / .results.stats.total' results.json)
          AVG_SCORE=$(jq '.results.stats.assertPassCount / .results.stats.assertCount' results.json)
          echo "Pass rate: $PASS_RATE"
          echo "Avg score: $AVG_SCORE"
          # Gate: 95% pass rate, average score > 4.0 on 5-point scale
          if (( $(echo "$PASS_RATE < 0.95" | bc -l) )); then
            echo "::error::Pass rate $PASS_RATE below 0.95 threshold"
            exit 1
          fi
      - name: Comment results on PR
        uses: actions/github-script@v7
        with:
          script: |
            const results = require('./results.json');
            const body = `## Eval Results\n\n` +
              `Pass rate: ${(results.results.stats.successes / results.results.stats.total * 100).toFixed(1)}%\n` +
              `Failed cases: ${results.results.stats.failures}\n\n` +
              `[Full results artifact](${process.env.GITHUB_SERVER_URL}/${process.env.GITHUB_REPOSITORY}/actions/runs/${process.env.GITHUB_RUN_ID})`;
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body,
            });

<!-- APIM inbound: A/B split with version tracking -->
<set-variable name="promptVersion"
  value="@(new Random().Next(100) < 10 ? "v2" : "v1")" />
<choose>
  <when condition="@(context.Variables.GetValueOrDefault<string>("promptVersion") == "v2")">
    <!-- Route to backend that loads v2 prompt -->
    <set-header name="X-Prompt-Version" exists-action="override">
      <value>v2</value>
    </set-header>
  </when>
</choose>
<!-- Always emit version dimension for comparison in Langfuse / App Insights -->
<set-header name="X-Prompt-Version-Actual" exists-action="override">
  <value>@(context.Variables.GetValueOrDefault<string>("promptVersion"))</value>
</set-header>

1. Update model reference in staging deployment
2. Run full eval suite against staging (all golden datasets)
3. Run adversarial test suite (jailbreaks, injection attempts, refusal cases)
4. Run latency benchmark (TTFT, TPOT, throughput at target concurrency)
5. Human review of 20 randomly sampled outputs on complex cases
6. If all gates pass → blue-green deploy to production
7. Monitor for 48 hours at 5% traffic before full cutover
8. Rollback trigger: any SLO breach or quality drop > 5% in online eval

# Two deployments, one service
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: vllm-phi4-v1
  namespace: inference
spec:
  selector:
    matchLabels:
      app: vllm
      version: phi4-v1
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: vllm-phi4-v2
  namespace: inference
spec:
  selector:
    matchLabels:
      app: vllm
      version: phi4-v2
---
# HTTPRoute: route 5% to v2
apiVersion: gateway.networking.k8s.io/v1
kind: HTTPRoute
metadata:
  name: inference-route
spec:
  rules:
    - backendRefs:
        - name: vllm-phi4-v1-svc
          weight: 95
        - name: vllm-phi4-v2-svc
          weight: 5

INDEX_VERSION = "v3-chunk512-bge-m3"  # bump this on any pipeline change
def index_document(doc: str, source: str):
    chunks = splitter.split_text(doc)
    embeddings = embed_batch(chunks)
    search_client.upload_documents([
        {
            "id": f"{source}-{i}-{INDEX_VERSION}",
            "content": chunk,
            "content_vector": emb,
            "index_version": INDEX_VERSION,
            "source": source,
        }
        for i, (chunk, emb) in enumerate(zip(chunks, embeddings))
    ])

KEDA scale trigger fires
        ↓ ~10s
New pod scheduled, NAP provisions node
        ↓ ~2-4 min
Node joins cluster, GPU driver initializes
        ↓ ~1-2 min
Pod starts, model weights loaded into VRAM
        ↓ ~1-2 min (Phi-4 Mini) to ~8 min (Llama 70B)
First request served

# KEDA ScaledObject: keep 1 warm, allow scale-to-zero for large cost savings
# Use only for workloads where the cold-start is tolerable (batch jobs, async APIs)
minReplicaCount: 0
scaleTargetRef:
  apiVersion: apps/v1
  kind: Deployment
  name: vllm-standalone
triggers:
  - type: prometheus
    metadata:
      query: avg(vllm:num_requests_waiting{namespace="inference"})
      threshold: "1"
      activationThreshold: "1"

# Watch KV cache in real time
kubectl exec -n inference <vllm-pod> -- curl -s http://localhost:8000/metrics \
  | grep "gpu_cache_usage"

# In Langfuse, tag traces with turn number
langfuse_context.update_current_trace(
    metadata={"conversation_turn": turn_number, "context_tokens": current_context_tokens}
)

response = client.chat.completions.create(
    messages=...,
    stream=True,
    timeout=httpx.Timeout(connect=5, read=120, write=10, pool=5),
)

# When user rates a response
langfuse_client.score(
    trace_id=trace_id,
    name="user_rating",
    value=rating,  # 1-5 or 0/1 thumbs
    comment=user_comment,
)

# Log implicit signals as Langfuse scores
def on_user_reprompt(trace_id: str, seconds_since_response: float):
    if seconds_since_response < 10:
        langfuse_client.score(
            trace_id=trace_id,
            name="implicit_quality",
            value=0,
            comment=f"Re-prompted after {seconds_since_response:.1f}s",
        )

helm repo add argilla https://argilla-io.github.io/argilla
helm upgrade --install argilla argilla/argilla \
  --namespace argilla --create-namespace \
  --set replicaCount=1 \
  --set resources.requests.memory="1Gi"

import argilla as rg
# Initialize Argilla connection
rg.init(api_url="http://argilla.argilla.svc.cluster.local:6900", api_key=...)
# Push low-rated traces to Argilla for review
def export_low_quality_traces(min_date: datetime, max_traces: int = 200):
    low_quality = langfuse_client.fetch_traces(
        tags=["production"],
        min_score={"name": "user_rating", "op": "lt", "value": 3},
        limit=max_traces,
    )
    records = [
        rg.TextClassificationRecord(
            text=trace.input["messages"][-1]["content"],
            prediction=[("bad_answer", 1.0)],
            annotation=None,  # labeler will fill this in
            metadata={
                "trace_id": trace.id,
                "model": trace.metadata.get("model"),
                "full_context": json.dumps(trace.input),
                "response": trace.output,
            },
        )
        for trace in low_quality.data
    ]
    rg.log(records, name="low_quality_traces", workspace="production-review")

Quality gap observed
        │
        ├─ Wrong format / style?
        │         → Fix the prompt (system prompt + examples)
        │         → If persistent after prompt fix → fine-tune on format
        │
        ├─ Factually wrong on domain knowledge?
        │         ├─ Knowledge available in documents?
        │         │         → RAG: add to index, improve chunking/retrieval
        │         └─ Knowledge not in documents?
        │                   → Fine-tune if static knowledge; accept limitation if dynamic
        │
        ├─ Wrong answer on complex reasoning?
        │         ├─ Passes on larger model (GPT-4o / Llama 70B)?
        │         │         → Either use larger model OR fine-tune smaller model on CoT examples
        │         └─ Fails on all models?
        │                   → Task may be inherently ambiguous; clarify requirements
        │
        ├─ Inconsistent behavior (same query, different answers)?
        │         → Lower temperature; use CoT; add few-shot examples; fine-tune
        │
        └─ Safety/refusal failures?
                  → Add to adversarial test suite; fix system prompt; fine-tune on refusals

# .github/workflows/weekly-drift-check.yml
on:
  schedule:
    - cron: '0 6 * * 1'  # every Monday at 6am UTC
jobs:
  drift-check:
    steps:
      - name: Run eval against production endpoint
        run: |
          npx promptfoo eval \
            --config promptfooconfig.yaml \
            --providers vllm:http://inference-prod.yourdomain.com/v1 \
            --output weekly-results.json
      - name: Compare to baseline
        run: |
          python scripts/compare_eval_results.py \
            --baseline eval-baselines/latest.json \
            --current weekly-results.json \
            --threshold 0.05  # alert if pass rate drops > 5%

<!-- APIM: PTU primary, consumption overflow on 429 -->
<retry condition="@(context.Response.StatusCode == 429)" count="1" interval="0">
  <set-backend-service base-url="{{aoai-consumption-endpoint}}" />
  <set-header name="api-key" exists-action="override">
    <value>{{aoai-consumption-key}}</value>
  </set-header>
</retry>