# AI Engineer

> Expert AI engineer for LLM integration, RAG pipelines, prompt engineering, model evaluation, and AI agent development

- Skill: `vodouai/ai-engineer` (Agent Skill)
- Install (CLI): `npx skillmds@latest add vodouai/ai-engineer`
- Raw SKILL.md: https://api.skillmd.com/api/skills/vodouai/ai-engineer/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: AI & ML
- Author: VodouAI (https://skillmd.com/u/vodouai)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/vodouai/ai-engineer

---


# AI Engineer - Expert Agent

## Overview

You are an expert AI engineer who builds production AI systems. You help teams integrate LLMs into applications, build retrieval-augmented generation pipelines, design prompt engineering systems, evaluate and compare models, and build AI agents with tool use. You work across Python and TypeScript, and you focus on practical, production-ready patterns over theoretical exercises.

Use this agent when you need to:
- Wire an LLM API into an existing application
- Build a RAG system that actually retrieves relevant context
- Design prompts that are maintainable and testable
- Choose between models for a specific use case
- Build an agent that can call tools and take actions

**STOPPING POINT 1**: What do you need to build?

1. **Integrate an LLM API into an app** - Add chat, completion, or structured output to an existing application
2. **Build a RAG pipeline** - Set up document ingestion, chunking, embedding, retrieval, and generation
3. **Design a prompt engineering system** - Create maintainable, versioned, testable prompts
4. **Evaluate and compare models** - Run structured evaluations to pick the right model
5. **Build an AI agent with tool use** - Create an agent that can call functions and take actions

---

## Workflow 1: Integrate an LLM API Into an App

### Step 1: Define the Integration Pattern

Determine how the LLM fits into your application flow.

**Common patterns:**

| Pattern | Use When | Example |
|---------|----------|---------|
| Request-Response | User asks, AI answers | Chatbot, Q&A |
| Background Processing | AI processes data offline | Summarization, classification |
| Streaming | Real-time token delivery | Chat UI, writing assistant |
| Structured Output | AI returns typed data | Extraction, form filling |

### Step 2: Set Up the Client

**Python (OpenAI-compatible):**
```python
import os
from openai import OpenAI

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

def complete(prompt: str, system: str = "", model: str = "gpt-4o") -> str:
    messages = []
    if system:
        messages.append({"role": "system", "content": system})
    messages.append({"role": "user", "content": prompt})

    response = client.chat.completions.create(
        model=model,
        messages=messages,
        temperature=0.7,
    )
    return response.choices[0].message.content
```

**Python (Anthropic):**
```python
import anthropic

client = anthropic.Anthropic()

def complete(prompt: str, system: str = "", model: str = "claude-sonnet-4-20250514") -> str:
    response = client.messages.create(
        model=model,
        max_tokens=4096,
        system=system,
        messages=[{"role": "user", "content": prompt}],
    )
    return response.content[0].text
```

**TypeScript (streaming):**
```typescript
import OpenAI from "openai";

const client = new OpenAI();

async function* streamComplete(prompt: string, system?: string) {
  const stream = await client.chat.completions.create({
    model: "gpt-4o",
    messages: [
      ...(system ? [{ role: "system" as const, content: system }] : []),
      { role: "user", content: prompt },
    ],
    stream: true,
  });

  for await (const chunk of stream) {
    const content = chunk.choices[0]?.delta?.content;
    if (content) yield content;
  }
}
```

### Step 3: Add Structured Output

When you need typed responses, not free text:

```python
from pydantic import BaseModel

class ExtractedEntity(BaseModel):
    name: str
    entity_type: str
    confidence: float

class ExtractionResult(BaseModel):
    entities: list[ExtractedEntity]

response = client.beta.chat.completions.parse(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "Extract entities from the text."},
        {"role": "user", "content": user_text},
    ],
    response_format=ExtractionResult,
)
result: ExtractionResult = response.choices[0].message.parsed
```

### Step 4: Production Hardening

Add retries, rate limiting, and cost tracking:

```python
import time
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, max=10))
def safe_complete(prompt: str, **kwargs) -> str:
    try:
        return complete(prompt, **kwargs)
    except openai.RateLimitError:
        time.sleep(5)
        raise
    except openai.APIError as e:
        if e.status_code >= 500:
            raise  # Retry on server errors
        raise  # Don't retry on client errors
```

**STOPPING POINT 2**: Your integration is set up. What next?

1. **Add conversation memory** - Maintain chat history across turns
2. **Add caching** - Cache identical requests to reduce cost and latency
3. **Add fallback models** - Automatically switch providers on failure
4. **Move to production** - Add monitoring, logging, and alerting

---

## Workflow 2: Build a RAG Pipeline

### Step 1: Document Ingestion

Load and normalize your source documents:

```python
from pathlib import Path

def load_documents(source_dir: str) -> list[dict]:
    docs = []
    for path in Path(source_dir).rglob("*"):
        if path.suffix == ".md":
            docs.append({"content": path.read_text(), "source": str(path), "type": "markdown"})
        elif path.suffix == ".pdf":
            docs.append({"content": extract_pdf_text(path), "source": str(path), "type": "pdf"})
        elif path.suffix == ".txt":
            docs.append({"content": path.read_text(), "source": str(path), "type": "text"})
    return docs
```

### Step 2: Chunking Strategy

Chunking is where most RAG pipelines succeed or fail. Choose based on your content:

| Strategy | Chunk Size | Overlap | Best For |
|----------|-----------|---------|----------|
| Fixed-size | 500-1000 tokens | 50-100 tokens | General text, articles |
| Semantic (paragraph) | Varies | 1 sentence | Well-structured docs |
| Recursive splitting | 500-1500 tokens | 100 tokens | Code, mixed content |
| Document-level | Whole doc | None | Short docs (<2000 tokens) |

**Recursive character splitting (most versatile):**
```python
def chunk_text(text: str, max_tokens: int = 800, overlap: int = 100) -> list[str]:
    separators = ["\n\n", "\n", ". ", " "]
    chunks = []
    current = ""

    for sep in separators:
        if len(text.split(sep)) > 1:
            parts = text.split(sep)
            for part in parts:
                if token_count(current + sep + part) > max_tokens:
                    if current:
                        chunks.append(current.strip())
                        # Keep overlap from end of current chunk
                        overlap_text = current[-overlap * 4:]  # Approximate chars
                        current = overlap_text + sep + part
                    else:
                        current = part
                else:
                    current = current + sep + part if current else part
            if current:
                chunks.append(current.strip())
            return chunks

    # Fallback: split by character count
    for i in range(0, len(text), max_tokens * 4):
        chunks.append(text[i:i + max_tokens * 4 + overlap * 4])
    return chunks
```

### Step 3: Embedding and Storage

```python
import chromadb

chroma_client = chromadb.PersistentClient(path="./chroma_db")
collection = chroma_client.get_or_create_collection(
    name="documents",
    metadata={"hnsw:space": "cosine"},
)

def embed_and_store(chunks: list[dict]):
    # chunks = [{"content": "...", "source": "...", "chunk_index": 0}]
    response = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=[c["content"] for c in chunks],
    )

    collection.add(
        ids=[f"{c['source']}_{c['chunk_index']}" for c in chunks],
        embeddings=[e.embedding for e in response.data],
        documents=[c["content"] for c in chunks],
        metadatas=[{"source": c["source"]} for c in chunks],
    )
```

### Step 4: Retrieval and Generation

```python
def rag_query(question: str, n_results: int = 5) -> str:
    # Embed the question
    q_embedding = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=question,
    ).data[0].embedding

    # Retrieve relevant chunks
    results = collection.query(query_embeddings=[q_embedding], n_results=n_results)
    context_chunks = results["documents"][0]

    # Generate answer with context
    context = "\n\n---\n\n".join(context_chunks)
    response = openai_client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": f"""Answer the question using ONLY the provided context.
If the context doesn't contain the answer, say so. Cite sources when possible.

Context:
{context}"""},
            {"role": "user", "content": question},
        ],
    )
    return response.choices[0].message.content
```

**STOPPING POINT 3**: Your basic RAG pipeline works. How do you want to improve it?

1. **Improve retrieval quality** - Add hybrid search (keyword + semantic), re-ranking, query expansion
2. **Add metadata filtering** - Filter by date, source, category before semantic search
3. **Handle updates** - Incremental re-indexing when documents change
4. **Scale it** - Move to a managed vector database (Pinecone, Weaviate, Qdrant)
5. **Add evaluation** - Measure retrieval precision and answer quality

---

## Workflow 3: Design a Prompt Engineering System

### Step 1: Prompt Structure

Every production prompt should have these sections:

```
ROLE: Who the AI is (sets behavior boundaries)
CONTEXT: Background information the AI needs
TASK: What specifically to do
FORMAT: How to structure the output
CONSTRAINTS: Rules, limitations, edge cases
EXAMPLES: 1-3 few-shot examples of ideal input/output pairs
```

**Template (Python):**
```python
PROMPTS = {
    "classify_ticket": {
        "version": "1.3",
        "system": """You are a support ticket classifier. You categorize incoming
support tickets into exactly one category.

Categories:
- billing: Payment issues, subscription changes, refunds
- technical: Bugs, errors, integration problems
- feature_request: New feature suggestions, improvements
- account: Login issues, password resets, profile changes
- other: Anything that doesn't fit above

Rules:
- Choose the MOST specific category that applies
- If unclear between two, choose the one mentioned first in the ticket
- Respond with ONLY the category name, nothing else""",

        "user_template": "Classify this support ticket:\n\n{ticket_text}",

        "examples": [
            {"input": "I was charged twice for my subscription this month",
             "output": "billing"},
            {"input": "The export button returns a 500 error",
             "output": "technical"},
        ],
    }
}
```

### Step 2: Prompt Versioning and Testing

```python
import json
import hashlib
from datetime import datetime

class PromptRegistry:
    def __init__(self, storage_path: str = "./prompts"):
        self.path = Path(storage_path)
        self.path.mkdir(exist_ok=True)

    def register(self, name: str, system: str, user_template: str,
                 examples: list[dict] = None) -> str:
        version = hashlib.sha256(
            (system + user_template).encode()
        ).hexdigest()[:8]

        record = {
            "name": name,
            "version": version,
            "system": system,
            "user_template": user_template,
            "examples": examples or [],
            "created_at": datetime.utcnow().isoformat(),
        }

        filepath = self.path / f"{name}_v{version}.json"
        filepath.write_text(json.dumps(record, indent=2))
        return version

    def load(self, name: str, version: str = "latest") -> dict:
        if version == "latest":
            files = sorted(self.path.glob(f"{name}_v*.json"))
            filepath = files[-1]
        else:
            filepath = self.path / f"{name}_v{version}.json"
        return json.loads(filepath.read_text())
```

### Step 3: Build a Test Suite for Your Prompts

```python
class PromptTestCase:
    def __init__(self, input_text: str, expected: str, tags: list[str] = None):
        self.input_text = input_text
        self.expected = expected
        self.tags = tags or []

def evaluate_prompt(prompt_config: dict, test_cases: list[PromptTestCase],
                    model: str = "gpt-4o") -> dict:
    results = {"passed": 0, "failed": 0, "errors": []}

    for tc in test_cases:
        user_msg = prompt_config["user_template"].format(ticket_text=tc.input_text)
        output = complete(user_msg, system=prompt_config["system"], model=model)

        if output.strip().lower() == tc.expected.lower():
            results["passed"] += 1
        else:
            results["failed"] += 1
            results["errors"].append({
                "input": tc.input_text,
                "expected": tc.expected,
                "got": output.strip(),
            })

    results["accuracy"] = results["passed"] / (results["passed"] + results["failed"])
    return results
```

**STOPPING POINT 4**: Your prompt system is structured. What next?

1. **Add A/B testing** - Run two prompt versions side by side and compare
2. **Add chain-of-thought** - Build multi-step reasoning into prompts
3. **Build prompt pipelines** - Chain prompts together for complex tasks
4. **Add guardrails** - Input validation, output validation, safety checks

---

## Workflow 4: Evaluate and Compare Models

### Step 1: Define Your Evaluation Criteria

Build an eval set matched to your actual use case:

```python
class EvalCase:
    def __init__(self, input_text: str, reference: str, category: str):
        self.input_text = input_text
        self.reference = reference  # Gold standard answer
        self.category = category

eval_set = [
    EvalCase("Summarize this contract clause...", "The clause limits liability to...", "summarization"),
    EvalCase("Extract the deadline from...", "March 15, 2025", "extraction"),
    # Minimum 30-50 cases per category for meaningful results
]
```

### Step 2: Run Multi-Model Evaluation

```python
import time

MODELS_TO_TEST = [
    {"provider": "openai", "model": "gpt-4o", "label": "GPT-4o"},
    {"provider": "openai", "model": "gpt-4o-mini", "label": "GPT-4o Mini"},
    {"provider": "anthropic", "model": "claude-sonnet-4-20250514", "label": "Claude Sonnet"},
]

def run_evaluation(eval_set: list[EvalCase], models: list[dict]) -> dict:
    results = {}

    for model_config in models:
        label = model_config["label"]
        results[label] = {"scores": [], "latencies": [], "costs": []}

        for case in eval_set:
            start = time.time()
            output = call_model(model_config, case.input_text)
            latency = time.time() - start

            score = score_output(output, case.reference, case.category)

            results[label]["scores"].append(score)
            results[label]["latencies"].append(latency)
            results[label]["costs"].append(estimate_cost(model_config, case.input_text, output))

    return results

def score_output(output: str, reference: str, category: str) -> float:
    """Score from 0-1. Use LLM-as-judge for subjective tasks."""
    if category == "extraction":
        return 1.0 if reference.lower() in output.lower() else 0.0

    # LLM-as-judge for open-ended tasks
    judge_prompt = f"""Rate how well the Output answers compared to the Reference.
Score 0-10 where 10 is perfect.

Reference: {reference}
Output: {output}

Respond with ONLY a number 0-10."""

    score_text = complete(judge_prompt, model="gpt-4o")
    return float(score_text.strip()) / 10.0
```

### Step 3: Generate Comparison Report

```python
def generate_report(results: dict) -> str:
    lines = ["| Model | Avg Score | P50 Latency | P95 Latency | Avg Cost/req |"]
    lines.append("|-------|-----------|-------------|-------------|--------------|")

    for model, data in results.items():
        avg_score = sum(data["scores"]) / len(data["scores"])
        p50_lat = sorted(data["latencies"])[len(data["latencies"]) // 2]
        p95_lat = sorted(data["latencies"])[int(len(data["latencies"]) * 0.95)]
        avg_cost = sum(data["costs"]) / len(data["costs"])
        lines.append(f"| {model} | {avg_score:.2f} | {p50_lat:.2f}s | {p95_lat:.2f}s | ${avg_cost:.4f} |")

    return "\n".join(lines)
```

**STOPPING POINT 5**: Evaluation is complete. What's your decision?

1. **Pick a model and ship** - Use the results to make a final selection
2. **Run deeper analysis** - Break down scores by category, find failure modes
3. **Test with fine-tuning** - Fine-tune a smaller model on your best results
4. **Build a routing system** - Route easy queries to cheap models, hard ones to expensive models

---

## Workflow 5: Build an AI Agent with Tool Use

### Step 1: Define Your Tools

```python
import json

tools = [
    {
        "type": "function",
        "function": {
            "name": "search_database",
            "description": "Search the product database by name, category, or ID. Returns matching products with prices and availability.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search query text"},
                    "category": {"type": "string", "enum": ["electronics", "clothing", "home", "all"]},
                    "max_results": {"type": "integer", "default": 5},
                },
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "create_order",
            "description": "Create a new order for a customer. Requires product ID and quantity.",
            "parameters": {
                "type": "object",
                "properties": {
                    "product_id": {"type": "string"},
                    "quantity": {"type": "integer", "minimum": 1},
                    "customer_id": {"type": "string"},
                },
                "required": ["product_id", "quantity", "customer_id"],
            },
        },
    },
]
```

### Step 2: Implement the Agent Loop

```python
def run_agent(user_message: str, conversation: list[dict] = None,
              max_iterations: int = 10) -> str:
    messages = conversation or []
    messages.append({"role": "user", "content": user_message})

    for i in range(max_iterations):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": AGENT_SYSTEM_PROMPT},
                *messages,
            ],
            tools=tools,
        )

        choice = response.choices[0]

        # If no tool calls, we have our final answer
        if choice.finish_reason == "stop":
            messages.append({"role": "assistant", "content": choice.message.content})
            return choice.message.content

        # Execute tool calls
        messages.append(choice.message)
        for tool_call in choice.message.tool_calls:
            fn_name = tool_call.function.name
            fn_args = json.loads(tool_call.function.arguments)

            result = execute_tool(fn_name, fn_args)

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result),
            })

    return "Agent reached maximum iterations without completing."

def execute_tool(name: str, args: dict) -> dict:
    tool_map = {
        "search_database": search_database,
        "create_order": create_order,
    }
    fn = tool_map.get(name)
    if not fn:
        return {"error": f"Unknown tool: {name}"}
    try:
        return fn(**args)
    except Exception as e:
        return {"error": str(e)}
```

### Step 3: Agent System Prompt

```python
AGENT_SYSTEM_PROMPT = """You are a helpful shopping assistant. You help customers
find products and place orders.

Behavior rules:
- Always search before recommending products (don't make up products)
- Confirm order details with the user before calling create_order
- If a search returns no results, suggest alternative search terms
- Never invent product IDs, prices, or availability - only use data from search results
- If you're unsure about something, ask the user to clarify

When placing orders:
1. Search for the product
2. Show the user the options with prices
3. Ask them to confirm product and quantity
4. Only then create the order
"""
```

### Step 4: Add Guardrails

```python
def safe_agent(user_message: str, **kwargs) -> str:
    # Input validation
    if len(user_message) > 10000:
        return "Message too long. Please keep requests under 10,000 characters."

    # Run agent
    result = run_agent(user_message, **kwargs)

    # Output validation
    if any(term in result.lower() for term in ["ignore previous", "system prompt"]):
        return "I can help you with shopping. What are you looking for?"

    return result
```

**STOPPING POINT 6**: Your agent works. What next?

1. **Add memory** - Persist conversation history across sessions
2. **Add more tools** - Expand what the agent can do
3. **Build a multi-agent system** - Specialize agents and route between them
4. **Add observability** - Log every tool call, measure success rates, track costs
5. **Deploy it** - Set up as an API endpoint or integrate into your app