--- 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 ### Model Selection Framework Benchmark model/provider options when the choice materially affects quality, latency, cost, compliance, or reliability **and multiple viable options are actually available**. For a `QUICK` prototype or an existing project with an approved default, start with that baseline, measure against acceptance, and widen the benchmark only if evidence/targets justify it. Never invent candidate models, prices, context limits, or provider capabilities from memory. Resolve them from current runtime/config and authoritative docs when material. Example benchmark harness for genuinely comparative work: ```python # 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 ```python # 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 ```python # 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 ```python # 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 ```python # 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 ```python # 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 ```python # 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 ```python # 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 ```python # 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