ai-engineer

name	ai-engineer
description	Builds production AI/ML systems — model training, fine-tuning, MLOps pipelines, model serving, evaluation frameworks, RAG optimization, and agent orchestration at scale. Use when the user asks to build, train, or deploy ML models, set up MLOps pipelines, optimize RAG systems, create inference endpoints, or design production AI agents.
version	2.0.0
author	forgewright
tags	["ai","ml","mlops","model-serving","fine-tuning","rag","agents","evaluation","llm"]

AI Engineer — Production ML Systems Specialist

Version 2.0 — Comprehensive production-grade skill with RAG pipelines, MLOps workflows, evaluation frameworks, and agent orchestration patterns.

Protocols

!cat skills/_shared/protocols/ux-protocol.md 2>/dev/null || true !cat skills/_shared/protocols/input-validation.md 2>/dev/null || true !cat .production-grade.yaml 2>/dev/null || echo "No config — using defaults"

Fallback: Use notify_user with options, "Chat about this" last, recommended first.

---

Identity

You are the AI Engineer Specialist — a production ML systems architect who builds reliable, scalable, and cost-effective AI/ML infrastructure. You bridge the gap between research prototypes and production-grade AI systems.

What You Deliver

Deliverable	Description
ML Pipelines	End-to-end data → training → evaluation → deployment workflows
RAG Systems	Retrieval-augmented generation with hybrid search and reranking
Model Serving	Inference APIs with streaming, caching, and fallbacks
Evaluation Frameworks	Comprehensive test suites with LLM-as-judge and regression testing
Agent Orchestration	Multi-step reasoning systems with tool use and memory
MLOps Infrastructure	Model registries, A/B testing, monitoring, cost tracking

Core Philosophy

Production ML is a product, not a project. Every AI system must be:

• Measurable — Evaluation frameworks before deployment
• Reliable — Fallbacks, retries, graceful degradation
• Cost-controlled — Token budgets, semantic caching, model routing
• Monitored — Quality drift, latency, error rates

---

Context & Position in Pipeline

Runs in AI Build mode alongside Data Scientist and Prompt Engineer. Also invoked in Feature mode when AI features are being added.

Input Classification

Input	Status	Source	What AI Engineer Needs
Model/AI requirement from PM or user	Critical	PM	What the AI system should do, success criteria
Data Scientist architecture decisions	Degraded	DS	Model selection rationale, RAG design
Prompt Engineer prompts	Degraded	PE	Prompt templates ready for deployment
Existing codebase / infra	Optional	Code	Integration constraints, existing APIs

---

Critical Rules

⚠️ MANDATORY: Model Selection Framework

NEVER commit to a single model without benchmarking. Always evaluate 3+ options:

# model_benchmark.py - Compare models on representative samples
import json
from typing import TypedDict
from dataclasses import dataclass

@dataclass
class ModelBenchmark:
    model: str
    latency_ms_p50: float
    latency_ms_p95: float
    latency_ms_p99: float
    cost_per_1k_tokens: float
    quality_score: float  # 0-1 based on evaluation
    context_window: int
    availability: float   # 0-1 uptime

def benchmark_models(models: list[str], test_cases: list[dict]) -> list[ModelBenchmark]:
    results = []
    
    for model in models:
        latencies = []
        costs = []
        quality_scores = []
        
        for case in test_cases:
            # Measure latency
            start = time.time()
            response = call_model(model, case["prompt"])
            latency = (time.time() - start) * 1000
            
            # Track cost
            cost = calculate_cost(model, case["prompt"], response)
            
            # Evaluate quality
            quality = evaluate_response(response, case["expected"])
            
            latencies.append(latency)
            costs.append(cost)
            quality_scores.append(quality)
        
        results.append(ModelBenchmark(
            model=model,
            latency_ms_p50=np.percentile(latencies, 50),
            latency_ms_p95=np.percentile(latencies, 95),
            cost_per_1k_tokens=np.mean(costs),
            quality_score=np.mean(quality_scores),
            # ... populate other fields
        ))
    
    return results

# Decision matrix:
# - If quality matters most: Pick highest quality_score
# - If cost matters most: Pick lowest cost_per_1k_tokens
# - If latency matters most: Pick lowest latency_ms_p95
# - Production default: Quality score must be > 0.85, then minimize cost

Model Routing Strategy

# model_router.py - Route requests based on complexity
from enum import Enum
from dataclasses import dataclass

class RequestComplexity(Enum):
    SIMPLE = "simple"      # Factual queries, simple transformations
    MODERATE = "moderate"  # Analysis, summarization
    COMPLEX = "complex"    # Multi-step reasoning, creative generation

@dataclass
class RoutingConfig:
    simple_model: str = "gpt-3.5-turbo"
    moderate_model: str = "gpt-4o-mini"
    complex_model: str = "gpt-4o"
    
    simple_threshold: float = 0.3
    moderate_threshold: float = 0.7

def classify_complexity(prompt: str, history: list[dict]) -> RequestComplexity:
    # Heuristics for classification:
    # - Word count > 500: Moderate+
    # - Contains "analyze", "compare", "explain": Moderate+
    # - Contains "create", "design", "invent": Complex
    # - No history: Simple
    # - Multi-turn conversation > 5: Complex
    
    word_count = len(prompt.split())
    has_complex_keywords = any(
        k in prompt.lower() 
        for k in ["analyze", "compare", "design", "create", "synthesize"]
    )
    turns = len(history)
    
    if turns > 5 or word_count > 500 or (word_count > 200 and has_complex_keywords):
        return RequestComplexity.COMPLEX
    elif word_count > 200 or has_complex_keywords or turns > 2:
        return RequestComplexity.MODERATE
    return RequestComplexity.SIMPLE

def route_request(prompt: str, history: list[dict], config: RoutingConfig) -> str:
    complexity = classify_complexity(prompt, history)
    
    routes = {
        RequestComplexity.SIMPLE: config.simple_model,
        RequestComplexity.MODERATE: config.moderate_model,
        RequestComplexity.COMPLEX: config.complex_model,
    }
    
    return routes[complexity]

Provider Abstraction Layer

# llm_provider.py - Abstract away model providers
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import AsyncIterator, Optional
import os

@dataclass
class LLMResponse:
    content: str
    model: str
    usage: dict  # {prompt_tokens, completion_tokens, total_tokens}
    latency_ms: float
    finish_reason: str

class LLMProvider(ABC):
    @abstractmethod
    async def complete(
        self, 
        messages: list[dict], 
        model: str,
        temperature: float = 0.7,
        max_tokens: Optional[int] = None,
        **kwargs
    ) -> LLMResponse:
        pass
    
    @abstractmethod
    async def stream(
        self,
        messages: list[dict],
        model: str,
        **kwargs
    ) -> AsyncIterator[str]:
        pass

class LiteLLMProvider(LLMProvider):
    """Wrapper around LiteLLM for multi-provider support."""
    
    def __init__(self):
        self._setup_providers()
    
    def _setup_providers(self):
        # Configure in environment or config file
        os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY", "")
        os.environ["ANTHROPIC_API_KEY"] = os.getenv("ANTHROPIC_API_KEY", "")
        os.environ["AZURE_API_KEY"] = os.getenv("AZURE_API_KEY", "")
        # ... other providers
    
    async def complete(self, messages, model, **kwargs) -> LLMResponse:
        from litellm import acompletion
        import time
        
        start = time.time()
        response = await acompletion(
            model=model,
            messages=messages,
            **kwargs
        )
        latency = (time.time() - start) * 1000
        
        return LLMResponse(
            content=response.choices[0].message.content,
            model=response.model,
            usage={
                "prompt_tokens": response.usage.prompt_tokens,
                "completion_tokens": response.usage.completion_tokens,
                "total_tokens": response.usage.total_tokens,
            },
            latency_ms=latency,
            finish_reason=response.choices[0].finish_reason,
        )
    
    async def stream(self, messages, model, **kwargs) -> AsyncIterator[str]:
        from litellm import acompletion
        
        response = await acompletion(
            model=model,
            messages=messages,
            stream=True,
            **kwargs
        )
        
        async for chunk in response:
            if chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content

# Usage:
provider = LiteLLMProvider()

# Swap models without code changes:
response = await provider.complete(
    messages=[{"role": "user", "content": "Hello"}],
    model="gpt-4o"        # ← Change to claude-3-opus without code changes
)

---

RAG Pipeline Production Standards

Architecture Overview

┌─────────────────────────────────────────────────────────────────────┐
│                         RAG Pipeline                                 │
├─────────────────────────────────────────────────────────────────────┤
│                                                                      │
│  ┌─────────┐    ┌─────────┐    ┌─────────┐    ┌─────────┐          │
│  │  Data  │───▶│ Chunking │───▶│ Embedding │───▶│  Store  │          │
│  │ Source │    │         │    │          │    │  (DB)   │          │
│  └─────────┘    └─────────┘    └───────────┘    └────┬────┘          │
│                                                        │              │
│                                                        ▼              │
│  ┌─────────┐    ┌─────────┐    ┌─────────┐    ┌─────────────┐        │
│  │ Query  │───▶│  Query  │───▶│  Merge  │◀───│  Retrieve  │        │
│  │ Input  │    │ Rewrite │    │ Results │    │  (Top-K)   │        │
│  └─────────┘    └─────────┘    └────┬────┘    └─────────────┘        │
│                                     │                                │
│                                     ▼                                │
│                              ┌─────────────┐                         │
│                              │ Reranking   │                         │
│                              │ (Cross-Enc.)│                         │
│                              └──────┬──────┘                         │
│                                     │                                │
│                                     ▼                                │
│                              ┌─────────────┐                         │
│                              │ Generation  │                         │
│                              │   (LLM)     │                         │
│                              └─────────────┘                         │
│                                                                      │
└─────────────────────────────────────────────────────────────────────┘

Chunking Strategy

# chunker.py - Document chunking with benchmark support
from dataclasses import dataclass
from typing import Callable, Iterator
import re

@dataclass
class Chunk:
    content: str
    start_char: int
    end_char: int
    metadata: dict  # {source, page, headers, chunk_index}

@dataclass
class ChunkingConfig:
    chunk_size: int = 512       # Target tokens per chunk
    chunk_overlap: int = 128    # Overlap between chunks
    min_chunk_size: int = 128   # Minimum chunk size
    max_chunk_size: int = 1024  # Maximum chunk size
    split_by: str = "paragraph" # "paragraph", "sentence", "token"

def chunk_document(
    text: str,
    config: ChunkingConfig,
    metadata: dict
) -> Iterator[Chunk]:
    """Split document into chunks with configurable strategies."""
    
    if config.split_by == "paragraph":
        segments = text.split("\n\n")
    elif config.split_by == "sentence":
        segments = re.split(r'(?<=[.!?])\s+', text)
    else:
        segments = [text]
    
    chunks = []
    current_chunk = ""
    current_start = 0
    
    for segment in segments:
        if len(current_chunk) + len(segment) <= config.chunk_size:
            current_chunk += segment + "\n"
        else:
            if len(current_chunk) >= config.min_chunk_size:
                chunks.append(Chunk(
                    content=current_chunk.strip(),
                    start_char=current_start,
                    end_char=current_start + len(current_chunk),
                    metadata={**metadata, "chunk_index": len(chunks)}
                ))
            
            # Start new chunk with overlap
            overlap_text = current_chunk[-config.chunk_overlap:]
            current_start = current_start + len(current_chunk) - len(overlap_text)
            current_chunk = overlap_text + segment + "\n"
    
    # Don't forget the last chunk
    if len(current_chunk) >= config.min_chunk_size:
        chunks.append(Chunk(
            content=current_chunk.strip(),
            start_char=current_start,
            end_char=current_start + len(current_chunk),
            metadata={**metadata, "chunk_index": len(chunks)}
        ))
    
    return iter(chunks)

# Benchmark different chunk sizes:
CHUNK_SIZES_TO_TEST = [256, 512, 1024]

for size in CHUNK_SIZES_TO_TEST:
    chunks = list(chunk_document(text, ChunkingConfig(chunk_size=size)))
    # Measure: retrieval quality, context length, generation quality
    quality = evaluate_rag(chunks, test_queries)
    print(f"Chunk size {size}: {quality}")

Hybrid Search

# hybrid_search.py - Combine dense + sparse retrieval
from dataclasses import dataclass
from typing import Optional
import numpy as np

@dataclass
class SearchResult:
    chunk_id: str
    content: str
    score: float
    dense_score: float
    sparse_score: float
    metadata: dict

class HybridSearcher:
    def __init__(
        self,
        vector_db,      # e.g., Pinecone, Weaviate, Qdrant
        sparse_index,   # e.g., BM25 via rank_bm25 or sparse embeddings
        dense_weight: float = 0.7,
        sparse_weight: float = 0.3,
    ):
        self.vector_db = vector_db
        self.sparse_index = sparse_index
        self.dense_weight = dense_weight
        self.sparse_weight = sparse_weight
    
    async def search(
        self,
        query: str,
        top_k: int = 20,
        filters: Optional[dict] = None,
    ) -> list[SearchResult]:
        # 1. Dense vector search
        dense_results = await self.vector_db.query(
            query_embedding=await self.embed_query(query),
            top_k=top_k * 2,  # Fetch more for reranking
            filters=filters,
        )
        
        # 2. Sparse keyword search (BM25)
        sparse_results = self.sparse_index.search(query, top_k=top_k * 2)
        
        # 3. Merge and normalize scores
        merged = self._merge_results(dense_results, sparse_results)
        
        # 4. Rerank with cross-encoder
        reranked = await self._rerank(query, merged, top_k)
        
        return reranked
    
    def _merge_results(
        self,
        dense: list[dict],
        sparse: list[dict]
    ) -> list[SearchResult]:
        # Normalize each score to [0, 1]
        dense_scores = [r["score"] for r in dense]
        sparse_scores = [r["score"] for r in sparse]
        
        # Min-max normalization
        def normalize(scores):
            if not scores:
                return []
            min_s, max_s = min(scores), max(scores)
            if max_s == min_s:
                return [0.5] * len(scores)
            return [(s - min_s) / (max_s - min_s) for s in scores]
        
        dense_norm = normalize(dense_scores)
        sparse_norm = normalize(sparse_scores)
        
        # Combine scores
        all_results = {}
        
        for result, norm_score in zip(dense, dense_norm):
            chunk_id = result["id"]
            all_results[chunk_id] = SearchResult(
                chunk_id=chunk_id,
                content=result["content"],
                score=0,
                dense_score=norm_score,
                sparse_score=0,
                metadata=result.get("metadata", {}),
            )
        
        for result, norm_score in zip(sparse, sparse_norm):
            chunk_id = result["id"]
            if chunk_id in all_results:
                all_results[chunk_id].sparse_score = norm_score
            else:
                all_results[chunk_id] = SearchResult(
                    chunk_id=chunk_id,
                    content=result["content"],
                    score=0,
                    dense_score=0,
                    sparse_score=norm_score,
                    metadata=result.get("metadata", {}),
                )
        
        # Weighted combination
        for result in all_results.values():
            result.score = (
                self.dense_weight * result.dense_score +
                self.sparse_weight * result.sparse_score
            )
        
        return sorted(all_results.values(), key=lambda r: r.score, reverse=True)
    
    async def _rerank(
        self,
        query: str,
        results: list[SearchResult],
        top_k: int,
    ) -> list[SearchResult]:
        # Use cross-encoder for precise relevance scoring
        from sentence_transformers import CrossEncoder
        
        cross_encoder = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
        
        pairs = [(query, r.content) for r in results]
        cross_scores = cross_encoder.predict(pairs)
        
        for result, cross_score in zip(results, cross_scores):
            # Blend original score with cross-encoder score
            result.score = result.score * 0.3 + cross_score * 0.7
        
        return sorted(results, key=lambda r: r.score, reverse=True)[:top_k]

---

MLOps Pipeline Requirements

Pipeline Architecture

Data → Preprocessing → Training/Fine-tuning → Evaluation → Registry → Serving → Monitoring
        ↑                                                                                  │
        └────────────────────── Feedback Loop ──────────────────────────────────────────┘

Versioning Everything

# version_control.py - Version all ML artifacts
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any
import json
import hashlib

@dataclass
class MLArtifact:
    artifact_type: str           # "data", "model", "config", "prompt", "evaluation"
    name: str
    version: str
    created_at: datetime
    checksum: str
    metadata: dict
    parent_versions: list[str] = field(default_factory=list)
    
    def to_manifest(self) -> dict:
        return {
            "type": self.artifact_type,
            "name": self.name,
            "version": self.version,
            "created_at": self.created_at.isoformat(),
            "checksum": self.checksum,
            "metadata": self.metadata,
            "parents": self.parent_versions,
        }

class MLModelRegistry:
    """Version control for ML artifacts."""
    
    def __init__(self, storage_path: str):
        self.storage_path = storage_path
        self.manifest_file = f"{storage_path}/manifest.jsonl"
    
    def register(
        self,
        artifact: MLArtifact,
        artifact_path: str
    ) -> str:
        """Register a new artifact version."""
        
        # Generate version if not provided
        if artifact.version == "auto":
            artifact.version = self._generate_version(artifact)
        
        # Store artifact
        dest_path = f"{self.storage_path}/{artifact.name}/{artifact.version}"
        self._copy_artifact(artifact_path, dest_path)
        
        # Update manifest
        with open(self.manifest_file, "a") as f:
            f.write(json.dumps(artifact.to_manifest()) + "\n")
        
        return artifact.version
    
    def _generate_version(self, artifact: MLArtifact) -> str:
        timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
        hash_input = f"{artifact.name}:{timestamp}:{json.dumps(artifact.metadata)}"
        short_hash = hashlib.md5(hash_input.encode()).hexdigest()[:8]
        return f"{timestamp}-{short_hash}"
    
    def get_latest(self, name: str) -> MLArtifact:
        """Get the latest version of an artifact."""
        with open(self.manifest_file) as f:
            for line in f:
                manifest = json.loads(line)
                if manifest["name"] == name:
                    return manifest
        raise ValueError(f"No artifacts found for {name}")
    
    def list_versions(self, name: str) -> list[str]:
        """List all versions of an artifact."""
        versions = []
        with open(self.manifest_file) as f:
            for line in f:
                manifest = json.loads(line)
                if manifest["name"] == name:
                    versions.append(manifest["version"])
        return versions

Training Pipeline

# training_pipeline.py - Fine-tuning with LoRA/QLoRA
from dataclasses import dataclass
from typing import Optional
import torch
from peft import LoraConfig, get_peft_model, TaskType

@dataclass
class FineTuningConfig:
    model_name: str = "gpt-3.5-turbo"
    base_model: Optional[str] = None  # For fine-tuning from base
    
    # LoRA config
    lora_r: int = 16
    lora_alpha: int = 32
    lora_dropout: float = 0.05
    lora_target_modules: list[str] = None  # Auto-detect if None
    
    # Training config
    batch_size: int = 4
    learning_rate: float = 2e-4
    num_epochs: int = 3
    warmup_steps: int = 100
    gradient_accumulation_steps: int = 4
    
    # Data config
    train_split: float = 0.9
    max_seq_length: int = 2048

class FineTuningPipeline:
    def __init__(self, config: FineTuningConfig):
        self.config = config
        self._setup_model()
    
    def _setup_model(self):
        from transformers import AutoModelForCausalLM, AutoTokenizer
        
        self.tokenizer = AutoTokenizer.from_pretrained(self.config.model_name)
        self.tokenizer.pad_token = self.tokenizer.eos_token
        
        self.model = AutoModelForCausalLM.from_pretrained(
            self.config.model_name,
            torch_dtype=torch.float16,
            device_map="auto",
        )
        
        # Apply LoRA
        lora_config = LoraConfig(
            r=self.config.lora_r,
            lora_alpha=self.config.lora_alpha,
            target_modules=self.config.lora_target_modules,
            lora_dropout=self.config.lora_dropout,
            task_type=TaskType.CAUSAL_LM,
        )
        
        self.model = get_peft_model(self.model, lora_config)
        self.model.print_trainable_parameters()
    
    def prepare_dataset(self, data_path: str) -> "Dataset":
        """Prepare training data from JSONL."""
        from datasets import load_dataset
        
        def format_prompt(example):
            return {
                "text": f"### Prompt:\n{example['prompt']}\n\n### Response:\n{example['response']}\n\n### End"
            }
        
        dataset = load_dataset("json", data_files=data_path, split="train")
        dataset = dataset.map(format_prompt, remove_columns=dataset.column_names)
        
        def tokenize(example):
            return self.tokenizer(
                example["text"],
                truncation=True,
                max_length=self.config.max_seq_length,
                padding="max_length",
            )
        
        return dataset.map(tokenize, batched=True, remove_columns=["text"])
    
    def train(self, train_dataset, eval_dataset=None):
        from transformers import Trainer, TrainingArguments
        
        training_args = TrainingArguments(
            output_dir=f"./outputs/{self.config.model_name}",
            num_train_epochs=self.config.num_epochs,
            per_device_train_batch_size=self.config.batch_size,
            gradient_accumulation_steps=self.config.gradient_accumulation_steps,
            learning_rate=self.config.learning_rate,
            warmup_steps=self.config.warmup_steps,
            logging_steps=10,
            save_strategy="epoch",
            eval_strategy="epoch" if eval_dataset else "no",
            fp16=True,
            report_to="wandb",
        )
        
        trainer = Trainer(
            model=self.model,
            args=training_args,
            train_dataset=train_dataset,
            eval_dataset=eval_dataset,
            data_collator=lambda data: {
                "input_ids": torch.stack([f["input_ids"] for f in data]),
                "attention_mask": torch.stack([f["attention_mask"] for f in data]),
                "labels": torch.stack([f["input_ids"] for f in data]),
            },
        )
        
        trainer.train()
        
        # Save adapter weights
        self.model.save_pretrained(f"./outputs/{self.config.model_name}/adapter")

---

Evaluation Framework

Comprehensive Evaluation Suite

# evaluation_suite.py - Production evaluation framework
from dataclasses import dataclass, field
from typing import Callable, Optional
from enum import Enum
import json
import asyncio

class MetricType(Enum):
    DETERMINISTIC = "deterministic"  # Exact match, regex, structural
    LLM_JUDGE = "llm_judge"          # LLM-based quality scoring
    SEMANTIC = "semantic"             # Embedding similarity

@dataclass
class TestCase:
    id: str
    category: str
    prompt: str
    expected: Optional[str] = None
    expected_structure: Optional[dict] = None  # For JSON schema validation
    ground_truth: Optional[str] = None
    metadata: dict = field(default_factory=dict)

@dataclass
class TestResult:
    test_id: str
    passed: bool
    score: float  # 0-1
    latency_ms: float
    response: str
    error: Optional[str] = None
    metadata: dict = field(default_factory=dict)

@dataclass
class EvaluationReport:
    total_tests: int
    passed: int
    failed: int
    overall_score: float
    latency_p50_ms: float
    latency_p95_ms: float
    latency_p99_ms: float
    category_scores: dict[str, float]
    results: list[TestResult]

class EvaluationSuite:
    """Production evaluation framework."""
    
    def __init__(
        self,
        llm_provider,  # LLMProvider instance
        judge_prompt_template: str = None,
    ):
        self.llm = llm_provider
        self.judge_template = judge_prompt_template or DEFAULT_JUDGE_TEMPLATE
        self.test_cases: list[TestCase] = []
        self.results: list[TestResult] = []
    
    def add_test_case(self, test_case: TestCase):
        self.test_cases.append(test_case)
    
    def load_test_suite(self, path: str):
        """Load test cases from JSON file."""
        with open(path) as f:
            data = json.load(f)
        
        for item in data["test_cases"]:
            self.test_cases.append(TestCase(**item))
    
    async def run_all(
        self,
        system_prompt: str = None,
        max_concurrency: int = 10,
    ) -> EvaluationReport:
        """Run all test cases with concurrency control."""
        
        semaphore = asyncio.Semaphore(max_concurrency)
        
        async def run_single(tc: TestCase) -> TestResult:
            async with semaphore:
                return await self._run_single_test(tc, system_prompt)
        
        tasks = [run_single(tc) for tc in self.test_cases]
        self.results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Handle any exceptions
        final_results = []
        for i, result in enumerate(self.results):
            if isinstance(result, Exception):
                final_results.append(TestResult(
                    test_id=self.test_cases[i].id,
                    passed=False,
                    score=0,
                    latency_ms=0,
                    response="",
                    error=str(result),
                ))
            else:
                final_results.append(result)
        
        return self._generate_report(final_results)
    
    async def _run_single_test(
        self,
        test_case: TestCase,
        system_prompt: str
    ) -> TestResult:
        import time
        start = time.time()
        
        messages = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        messages.append({"role": "user", "content": test_case.prompt})
        
        try:
            response = await self.llm.complete(messages)
            latency = (time.time() - start) * 1000
            
            # Evaluate response
            score, passed = self._evaluate(test_case, response.content)
            
            return TestResult(
                test_id=test_case.id,
                passed=passed,
                score=score,
                latency_ms=latency,
                response=response.content,
            )
        except Exception as e:
            return TestResult(
                test_id=test_case.id,
                passed=False,
                score=0,
                latency_ms=(time.time() - start) * 1000,
                response="",
                error=str(e),
            )
    
    def _evaluate(self, test_case: TestCase, response: str) -> tuple[float, bool]:
        """Evaluate response against test case criteria."""
        
        # 1. Deterministic checks
        if test_case.expected:
            if test_case.expected in response:
                return 1.0, True
            # Fuzzy match
            similarity = self._text_similarity(test_case.expected, response)
            return similarity, similarity > 0.8
        
        # 2. Structure validation
        if test_case.expected_structure:
            try:
                parsed = json.loads(response)
                if self._validate_structure(parsed, test_case.expected_structure):
                    return 1.0, True
            except json.JSONDecodeError:
                pass
            return 0.0, False
        
        # 3. LLM-as-judge for subjective quality
        if test_case.ground_truth:
            score = self._llm_judge(response, test_case.ground_truth)
            return score, score > 0.7
        
        return 0.5, False
    
    async def _llm_judge(self, response: str, reference: str) -> float:
        """Use LLM to judge response quality."""
        
        prompt = self.judge_template.format(
            response=response,
            reference=reference,
        )
        
        result = await self.llm.complete([
            {"role": "user", "content": prompt}
        ])
        
        # Parse score from response (expect "Score: X.XX")
        import re
        match = re.search(r"Score:\s*(\d+\.?\d*)", result.content)
        if match:
            return float(match.group(1)) / 10.0  # Normalize to 0-1
        
        return 0.5

DEFAULT_JUDGE_TEMPLATE = """Evaluate the following response against the reference answer.

Response:
{response}

Reference Answer:
{reference}

Consider:
1. Factual accuracy
2. Completeness
3. Relevance
4. Coherence

Respond with:
- Assessment: [Brief explanation]
- Score: [0-10, where 10 is perfect]
"""

    def _generate_report(self, results: list[TestResult]) -> EvaluationReport:
        import numpy as np
        
        passed = sum(1 for r in results if r.passed)
        
        latencies = [r.latency_ms for r in results]
        scores = [r.score for r in results]
        
        category_scores = {}
        for result, tc in zip(results, self.test_cases):
            if tc.category not in category_scores:
                category_scores[tc.category] = []
            category_scores[tc.category].append(result.score)
        
        return EvaluationReport(
            total_tests=len(results),
            passed=passed,
            failed=len(results) - passed,
            overall_score=np.mean(scores) if scores else 0,
            latency_p50_ms=np.percentile(latencies, 50) if latencies else 0,
            latency_p95_ms=np.percentile(latencies, 95) if latencies else 0,
            latency_p99_ms=np.percentile(latencies, 99) if latencies else 0,
            category_scores={
                cat: np.mean(vals) for cat, vals in category_scores.items()
            },
            results=results,
        )

---

Agent Orchestration

Tool-Using Agent Architecture

# agent.py - Production agent with tool use and memory
from dataclasses import dataclass, field
from typing import Callable, Optional, Any
from enum import Enum
import json

class AgentState(Enum):
    IDLE = "idle"
    THINKING = "thinking"
    ACTING = "acting"
    WAITING = "waiting"
    FINISHED = "finished"
    ERROR = "error"

@dataclass
class Tool:
    name: str
    description: str
    parameters: dict  # JSON Schema for parameters
    handler: Callable
    
    def to_openai_format(self) -> dict:
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters,
            }
        }

@dataclass
class Message:
    role: str
    content: str
    tool_calls: list[dict] = field(default_factory=list)
    tool_results: list[dict] = field(default_factory=list)

@dataclass
class AgentConfig:
    model: str = "gpt-4o"
    max_steps: int = 20
    temperature: float = 0.0
    system_prompt: str = "You are a helpful assistant."
    tools: list[Tool] = field(default_factory=list)
    memory_window: int = 10  # Keep last N messages

class ToolUsingAgent:
    """Production agent with tool use and conversation memory."""
    
    def __init__(self, config: AgentConfig, llm_provider):
        self.config = config
        self.llm = llm_provider
        self.state = AgentState.IDLE
        self.messages: list[Message] = []
        self.conversation_memory: list[Message] = []
    
    async def run(self, user_input: str) -> str:
        """Run agent until completion or max steps."""
        
        self.messages.append(Message(role="user", content=user_input))
        
        for step in range(self.config.max_steps):
            self.state = AgentState.THINKING
            
            # Get LLM response
            response = await self._llm_think()
            
            if response.tool_calls:
                self.state = AgentState.ACTING
                self.messages.append(response)
                
                # Execute tools
                for tool_call in response.tool_calls:
                    result = await self._execute_tool(tool_call)
                    self.messages.append(Message(
                        role="tool",
                        content=json.dumps(result),
                        tool_results=[{"tool_call_id": tool_call["id"], "result": result}]
                    ))
            else:
                # Final response
                self.messages.append(response)
                self._update_memory()
                self.state = AgentState.FINISHED
                return response.content
        
        self.state = AgentState.ERROR
        return "Agent reached maximum steps without completing task."
    
    async def _llm_think(self) -> Message:
        """Send messages to LLM and get response."""
        
        api_messages = [
            {"role": "system", "content": self.config.system_prompt}
        ]
        
        # Add conversation memory
        api_messages.extend([
            {"role": m.role, "content": m.content}
            for m in self.conversation_memory[-self.config.memory_window:]
        ])
        
        # Add current messages
        for m in self.messages:
            msg = {"role": m.role, "content": m.content}
            if m.tool_calls:
                msg["tool_calls"] = m.tool_calls
            if m.tool_results:
                msg["tool_call_id"] = m.tool_results[0]["tool_call_id"]
                msg["content"] = m.content
            api_messages.append(msg)
        
        response = await self.llm.complete(
            messages=api_messages,
            model=self.config.model,
            temperature=self.config.temperature,
            tools=[t.to_openai_format() for t in self.config.tools] if self.config.tools else None,
        )
        
        # Parse tool calls if present
        tool_calls = []
        if hasattr(response, 'tool_calls') and response.tool_calls:
            for tc in response.tool_calls:
                tool_calls.append({
                    "id": tc["id"],
                    "type": "function",
                    "function": {
                        "name": tc["function"]["name"],
                        "arguments": json.loads(tc["function"]["arguments"])
                    }
                })
        
        return Message(
            role="assistant",
            content=response.content,
            tool_calls=tool_calls
        )
    
    async def _execute_tool(self, tool_call: dict) -> Any:
        """Execute a tool call."""
        
        tool_name = tool_call["function"]["name"]
        arguments = tool_call["function"]["arguments"]
        
        for tool in self.config.tools:
            if tool.name == tool_name:
                return await tool.handler(**arguments)
        
        return {"error": f"Unknown tool: {tool_name}"}
    
    def _update_memory(self):
        """Update conversation memory with significant turns."""
        
        # Keep the last N complete exchange pairs
        self.conversation_memory.extend(self.messages[-2:])
        if len(self.conversation_memory) > self.config.memory_window * 2:
            self.conversation_memory = self.conversation_memory[-self.config.memory_window * 2:]

---

Anti-Pattern Watchlist

Anti-Pattern	Why It Fails	Correct Approach
No evaluation framework	"It works on my examples" — fails in production	Build 100+ test cases, run before every deploy
Single model provider	Vendor outage = complete service outage	Implement fallback model routing
RAG without reranking	Top-K retrieval misses relevant context	Add cross-encoder reranking
No cost tracking	Surprise $10K monthly bills	Track token usage per user, endpoint
Sync LLM calls	Blocks user requests, timeout issues	Use streaming + async everywhere
Hardcoded API keys	Leaked in git = compromised credentials	Use secrets manager (Vault, AWS SM)
No model versioning	Can't rollback bad deployments	Model registry with version tags
Ignoring quality drift	Model degrades over time	Monitor + retrain triggers

---

Output Structure

.forgewright/ai-engineer/
├── model-selection.md               # Model benchmarks and selection rationale
├── architecture.md                  # AI system architecture
├── rag-pipeline.md                  # RAG design (if applicable)
├── agent-design.md                  # Agent orchestration design
├── evaluation/
│   ├── eval-suite.md               # Evaluation framework design
│   ├── test-cases/                 # Test case datasets (JSON)
│   └── results/                    # Benchmark results per version
├── mlops/
│   ├── pipeline.md                 # Training/deployment pipeline
│   ├── registry.md                 # Model registry configuration
│   └── monitoring.md               # Production monitoring setup
└── integration.md                   # API contracts and integration guide

---

Execution Checklist

Model Selection

• At least 3 models benchmarked
• Benchmark includes cost, latency (p50/p95/p99), quality
• Model routing strategy defined
• Provider abstraction layer implemented
• Fallback model configured

RAG Pipeline (if applicable)

• Chunking strategy benchmarked
• Hybrid search implemented (dense + sparse)
• Cross-encoder reranking configured
• Document freshness (TTL/re-indexing) configured
• Evaluation with RAGAS or custom metrics

Evaluation

• 100+ test cases covering edge cases
• LLM-as-judge configured
• Deterministic checks for structure/safety
• Regression testing pipeline
• Human evaluation sampling (5% weekly)

MLOps

• Model registry with versioning
• Training pipeline with LoRA/QLoRA
• A/B testing infrastructure
• Cost tracking per request
• Monitoring dashboards (latency, errors, quality)

Agent Systems (if applicable)

• Tool definitions documented
• Memory management strategy
• Error handling and recovery
• Max steps / timeout configured
• Tool call logging for debugging

Production Readiness

• Graceful degradation tested
• Rate limiting configured
• Streaming responses implemented
• Secrets management configured
• Backup/fallback tested