# AI Agent Super Skill

> Comprehensive AI agent building skill merging Perplexity Computer's skill creation, webserver, and automation capabilities with Claude Code's agent orchestration, MCP server building, RAG system construction, subagent coordination, parallel agent dispatching, prompt optimization, and execution planning. Covers designing AI agents, building MCP servers, creating RAG pipelines, orchestrating multi-agent systems, optimizing prompts, and deploying AI-powered workflows. Use when building AI agents, creating MCP servers, designing RAG systems, coordinating subagents, optimizing prompts, or architecting any AI-powered automation. Use when this capability is needed.

- Skill: `tomevault-io/ai-agent-super-skill` (Agent Skill, multi-file: 2 files)
- Install (CLI): `npx skillmds@latest add tomevault-io/ai-agent-super-skill`
- Raw SKILL.md: https://api.skillmd.com/api/skills/tomevault-io/ai-agent-super-skill/raw
- Safety review: pending (external: skill-scanner WARNING, skillspector PASS)
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: AI & ML
- Author: tomevault-io (https://skillmd.com/u/tomevault-io)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/tomevault-io/ai-agent-super-skill

---


# AI Agent Builder Super-Skill

A comprehensive reference for designing, building, and deploying AI agents — from single-tool bots to production multi-agent systems — merging best practices from Claude Code's agent orchestration patterns with Perplexity Computer's deployment infrastructure.

---

## Table of Contents

1. [Gap Analysis Table](#1-gap-analysis-table)
2. [Agent Architecture & Design Patterns](#2-agent-architecture--design-patterns)
3. [MCP Server Development](#3-mcp-server-development)
4. [RAG System Construction](#4-rag-system-construction)
5. [Subagent Coordination](#5-subagent-coordination)
6. [Execution Planning & Verification](#6-execution-planning--verification)
7. [Prompt Engineering & Optimization](#7-prompt-engineering--optimization)
8. [ML Integration for Agents](#8-ml-integration-for-agents)
9. [Skill & Capability Creation](#9-skill--capability-creation)
10. [Backend Infrastructure for Agents](#10-backend-infrastructure-for-agents)
11. [Agent Deployment & Monitoring](#11-agent-deployment--monitoring)
12. [Unique Perplexity Computer Capabilities](#12-unique-perplexity-computer-capabilities)

---

## 1. Gap Analysis Table

This table maps each capability domain to its source skill, coverage level, and any gaps filled by this super-skill.

| Capability | Source Skill(s) | Coverage | Gaps Filled Here |
|------------|----------------|----------|-----------------|
| Agent architecture (ReAct, Plan-Execute) | senior-prompt-engineer | Partial — workflow diagrams only | Full pattern library with code |
| Multi-agent orchestration | subagent-driven-development, dispatching-parallel-agents | Strong process, no code | Integration patterns, conflict detection |
| MCP server building | mcp-builder | Full (4-phase process) | Perplexity-compatible CGI deployment |
| RAG pipeline construction | senior-ml-engineer, senior-prompt-engineer | Chunking + DB selection tables | End-to-end pipeline code |
| Prompt engineering | senior-prompt-engineer | Pattern reference table | Advanced chain-of-thought + meta-prompting |
| Subagent task dispatch | subagent-driven-development | Process diagrams | Template prompts with full context injection |
| Parallel agent dispatch | dispatching-parallel-agents | Decision tree + examples | Conflict detection, state isolation |
| Plan execution with checkpoints | executing-plans | Step-by-step process | Batch sizing, rollback strategies |
| MLOps / model deployment | senior-ml-engineer | Docker + k8s templates | Agent-specific serving patterns |
| Backend webhooks/SQLite | webserver | Full CGI-bin reference | Agent memory persistence layer |
| Skill packaging (SKILL.md) | create-skill (Perplexity) | YAML frontmatter format | Validation pipeline, versioning |
| Deployment & observability | website-building (Perplexity) | UI deployment only | Agent health checks, trace logging |
| Perplexity 400+ integrations | Perplexity Computer native | Available but undocumented | Integration mapping for agent use |
| Scheduled monitoring | Perplexity Computer native | Not in any skill | Agent heartbeat and drift triggers |

---

## 2. Agent Architecture & Design Patterns

### 2.1 Core Agent Loop

Every agent follows a fundamental observe → think → act → observe loop. The differences between architectures lie in how deeply they plan before acting and how they handle tool results.

```
+---------------------------------------------+
|                 AGENT LOOP                  |
|                                             |
|   Input/Observation                         |
|         |                                   |
|         v                                   |
|   +-------------+                           |
|   |    Think    | <---------------------+   |
|   |  (Reason)   |                       |   |
|   +------+------+                       |   |
|          |                              |   |
|          v                              |   |
|   +-------------+    No more tools      |   |
|   |  Select     |------------------>    |   |
|   |  Action     |                  |    |   |
|   +------+------+                  |    |   |
|          |                         |    |   |
|          v                         v    |   |
|   +-------------+           +----------+|   |
|   |  Execute    |           |  Final   ||   |
|   |  Tool/API   |           |  Answer  ||   |
|   +------+------+           +----------+|   |
|          |                              |   |
|          v                              |   |
|   +-------------+                       |   |
|   |  Observe    |-----------------------+   |
|   |  Result     |                           |
|   +-------------+                           |
+---------------------------------------------+
```

### 2.2 Architecture Patterns

#### Pattern A: ReAct (Reason + Act)

Best for: open-ended research, customer support, tool-use agents.

**How it works:** The agent interleaves reasoning traces (Thought:) with actions (Action:) and observations (Observation:) in a single conversation thread.

```python
REACT_SYSTEM_PROMPT = """
You are a research agent. For every task:

1. THOUGHT: Reason about what you know and what you need
2. ACTION: Choose one tool to call
3. OBSERVATION: Read the tool result
4. Repeat until you have enough information
5. FINAL ANSWER: Synthesize and respond

Available tools: {tool_list}

Format strictly:
Thought: <your reasoning>
Action: <tool_name>
Action Input: <tool_arguments as JSON>
Observation: <tool result — filled by system>
... (repeat)
Final Answer: <your complete response>
"""

def react_agent(query: str, tools: dict, llm, max_iterations: int = 10) -> str:
    messages = [
        {"role": "system", "content": REACT_SYSTEM_PROMPT.format(
            tool_list="\n".join(f"- {k}: {v['description']}" for k, v in tools.items())
        )},
        {"role": "user", "content": query}
    ]
    
    for iteration in range(max_iterations):
        response = llm.complete(messages)
        
        if "Final Answer:" in response:
            return response.split("Final Answer:")[-1].strip()
        
        # Parse Action / Action Input
        action_line = [l for l in response.split("\n") if l.startswith("Action:")]
        input_line  = [l for l in response.split("\n") if l.startswith("Action Input:")]
        
        if not action_line:
            break
        
        tool_name = action_line[0].replace("Action:", "").strip()
        tool_input = json.loads(input_line[0].replace("Action Input:", "").strip())
        
        # Execute tool
        if tool_name in tools:
            observation = tools[tool_name]["fn"](**tool_input)
        else:
            observation = f"Error: Unknown tool '{tool_name}'"
        
        messages.append({"role": "assistant", "content": response})
        messages.append({"role": "user", "content": f"Observation: {observation}"})
    
    return "Agent reached max iterations without a final answer."
```

#### Pattern B: Plan-and-Execute

Best for: complex multi-step workflows, code generation, structured report creation.

**How it works:** A planner LLM generates a complete task list first; executor agents complete each step sequentially or in parallel.

```python
PLANNER_PROMPT = """
Given this goal: {goal}

Create a numbered execution plan. Each step must be:
- Atomic: one clear action
- Verifiable: has a concrete success criterion
- Independent (where possible): can run without other steps completing first

Output format:
PLAN:
1. [Step description] | SUCCESS: [verification criterion] | DEPS: [step numbers or NONE]
2. ...
"""

EXECUTOR_PROMPT = """
Execute this step exactly:
{step}

Context from previous steps:
{context}

Available tools: {tools}

Return:
- RESULT: what you produced
- STATUS: SUCCESS or FAILED
- NOTES: any issues or observations
"""

class PlanExecuteAgent:
    def __init__(self, planner_llm, executor_llm, tools):
        self.planner = planner_llm
        self.executor = executor_llm
        self.tools = tools
    
    def run(self, goal: str) -> dict:
        # Phase 1: Plan
        plan_response = self.planner.complete(
            PLANNER_PROMPT.format(goal=goal)
        )
        steps = self._parse_plan(plan_response)
        
        # Phase 2: Execute
        results = {}
        for step in self._topological_sort(steps):
            context = {k: v["result"] for k, v in results.items() if v["status"] == "SUCCESS"}
            result = self.executor.complete(
                EXECUTOR_PROMPT.format(
                    step=step["description"],
                    context=json.dumps(context, indent=2),
                    tools=list(self.tools.keys())
                )
            )
            results[step["id"]] = self._parse_result(result)
        
        return results
```

#### Pattern C: Reflexion

Best for: code debugging, essay writing, tasks that benefit from self-critique.

**How it works:** After each attempt, the agent evaluates its own output, stores a reflection in memory, and retries.

```python
REFLEXION_EVALUATOR_PROMPT = """
Task: {task}
Attempt: {attempt}

Evaluate this attempt:
1. What did it get RIGHT? (be specific)
2. What did it get WRONG or MISS? (be specific)
3. What should the NEXT attempt do differently?

Score (0-10): 
Reflection: 
"""

class ReflexionAgent:
    def __init__(self, llm, max_attempts: int = 3, pass_threshold: float = 8.0):
        self.llm = llm
        self.max_attempts = max_attempts
        self.threshold = pass_threshold
        self.memory = []  # Persisted reflections
    
    def run(self, task: str) -> str:
        for attempt_num in range(self.max_attempts):
            # Inject prior reflections into context
            reflection_context = "\n".join(
                f"Attempt {i+1} reflection: {r}" for i, r in enumerate(self.memory)
            )
            
            attempt = self.llm.complete(
                f"Task: {task}\n\nPrior attempt learnings:\n{reflection_context}\n\nYour attempt:"
            )
            
            # Evaluate
            eval_response = self.llm.complete(
                REFLEXION_EVALUATOR_PROMPT.format(task=task, attempt=attempt)
            )
            score = float(re.search(r"Score \(0-10\):\s*([\d.]+)", eval_response).group(1))
            reflection = re.search(r"Reflection:\s*(.+)", eval_response, re.DOTALL).group(1).strip()
            
            self.memory.append(reflection)
            
            if score >= self.threshold:
                return attempt
        
        return attempt  # Return best attempt after max tries
```

#### Pattern D: Tool-Use Agent (Function Calling)

Best for: API integrations, data retrieval, modern LLM APIs that support native tool calling.

```python
import anthropic

def build_tool_agent(tools: list[dict], system: str = "") -> callable:
    """
    tools: list of Anthropic-format tool definitions
    Returns a function that runs the agent for a given query.
    """
    client = anthropic.Anthropic()
    
    def run(query: str, tool_executors: dict[str, callable]) -> str:
        messages = [{"role": "user", "content": query}]
        
        while True:
            response = client.messages.create(
                model="claude-opus-4-5",
                max_tokens=4096,
                system=system,
                tools=tools,
                messages=messages
            )
            
            # No tool calls — final answer
            if response.stop_reason == "end_turn":
                return response.content[0].text
            
            # Process tool calls
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    executor = tool_executors.get(block.name)
                    if executor:
                        result = executor(**block.input)
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": str(result)
                        })
            
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})
    
    return run
```

### 2.3 Architecture Selection Guide

| Goal | Pattern | Reason |
|------|---------|--------|
| Open-ended research | ReAct | Flexible, self-correcting, handles unknown paths |
| Multi-step report generation | Plan-Execute | Predictable, auditable, checkpointable |
| Code writing / debugging | Reflexion | Self-critique loop improves quality over iterations |
| API integration / tool calling | Tool-Use | Native LLM feature, lower latency, less prompt engineering |
| Customer support bot | ReAct + Tool-Use | Hybrid: structured tools with flexible reasoning |
| Batch data processing | Plan-Execute with parallel dispatch | Speed via parallelism, structured output |
| Creative tasks (writing, design) | Reflexion | Quality improves with each self-critique cycle |

### 2.4 Multi-Agent System Topologies

```
TOPOLOGY 1: Hub-and-Spoke (Orchestrator + Specialists)

                    +----------------+
                    |  Orchestrator  |
                    |  (Coordinator) |
                    +-------+--------+
          +-----------------+-----------------+
          v                 v                 v
   +------------+   +------------+   +------------+
   |  Research  |   |   Coder    |   |  Writer    |
   |   Agent    |   |   Agent    |   |   Agent    |
   +------------+   +------------+   +------------+

Use for: complex tasks needing specialized expertise per sub-domain.
Orchestrator decomposes goal -> dispatches -> aggregates results.

TOPOLOGY 2: Pipeline (Assembly Line)

  Input -> [Extractor] -> [Transformer] -> [Validator] -> [Writer] -> Output

Use for: ETL, document processing, multi-stage generation tasks.
Each agent only sees the previous stage's output.

TOPOLOGY 3: Competitive / Debate

  Query -> Agent A --+
                    +---> Judge Agent ---> Final Answer
  Query -> Agent B --+

Use for: decisions requiring multiple perspectives, factual verification,
high-stakes outputs where consensus improves reliability.

TOPOLOGY 4: Peer Network (Gossip/Consensus)

  Agent 1 <---> Agent 2
     ^  \          ^
     |   \         |
     v    \        v
  Agent 4 <---> Agent 3

Use for: simulation, emergent behavior research, distributed problem solving.
High coordination overhead — avoid for production automation.
```

---

## 3. MCP Server Development

### 3.1 Four-Phase MCP Build Process

Building a production MCP server follows four phases. Do not skip phases — each builds on the previous.

**Phase 1: Research & Planning**
1. Fetch MCP spec: `https://modelcontextprotocol.io/sitemap.xml` then pages with `.md` suffix
2. Study the target API's documentation — auth requirements, rate limits, key endpoints
3. Decide: TypeScript (recommended) or Python (FastMCP)
4. List all tools, prioritizing comprehensive API coverage over convenience wrappers

**Phase 2: Implementation**

**Phase 3: Review & Test** — use MCP Inspector

**Phase 4: Create Evaluations** — 10 read-only, complex, verifiable questions

### 3.2 TypeScript MCP Server Template

```typescript
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

// --- Server Initialization ---------------------------------------------------
const server = new McpServer({
  name: "my-service-mcp",
  version: "1.0.0",
});

// --- Shared API Client --------------------------------------------------------
interface ApiConfig {
  baseUrl: string;
  apiKey: string;
}

class ServiceClient {
  constructor(private config: ApiConfig) {}

  async request<T>(
    endpoint: string,
    options: RequestInit = {}
  ): Promise<T> {
    const url = `${this.config.baseUrl}${endpoint}`;
    const response = await fetch(url, {
      ...options,
      headers: {
        "Authorization": `Bearer ${this.config.apiKey}`,
        "Content-Type": "application/json",
        ...options.headers,
      },
    });

    if (!response.ok) {
      const error = await response.text();
      throw new Error(
        `API error ${response.status}: ${error}. ` +
        `Suggestion: Check your API key or verify the resource exists.`
      );
    }

    return response.json() as Promise<T>;
  }
}

const client = new ServiceClient({
  baseUrl: process.env.SERVICE_BASE_URL ?? "https://api.example.com",
  apiKey: process.env.SERVICE_API_KEY ?? "",
});

// --- Tool: List Items ---------------------------------------------------------
server.registerTool(
  "service_list_items",
  {
    description: "List items with optional filtering and pagination.",
    inputSchema: z.object({
      page:     z.number().int().min(1).default(1).describe("Page number (1-indexed)"),
      per_page: z.number().int().min(1).max(100).default(20).describe("Items per page"),
      filter:   z.string().optional().describe("Optional keyword filter"),
    }),
    annotations: {
      readOnlyHint:  true,
      destructiveHint: false,
      idempotentHint: true,
    },
  },
  async (params) => {
    const query = new URLSearchParams({
      page:     String(params.page),
      per_page: String(params.per_page),
      ...(params.filter ? { q: params.filter } : {}),
    });

    const data = await client.request<{ items: unknown[]; total: number }>(
      `/items?${query}`
    );

    return {
      content: [{
        type: "text",
        text: JSON.stringify(data, null, 2),
      }],
      structuredContent: data,
    };
  }
);

// --- Tool: Get Item -----------------------------------------------------------
server.registerTool(
  "service_get_item",
  {
    description: "Get a single item by ID.",
    inputSchema: z.object({
      id: z.string().describe("Item ID"),
    }),
    annotations: {
      readOnlyHint: true,
      destructiveHint: false,
      idempotentHint: true,
    },
  },
  async (params) => {
    const item = await client.request(`/items/${params.id}`);
    return {
      content: [{ type: "text", text: JSON.stringify(item, null, 2) }],
      structuredContent: item,
    };
  }
);

// --- Tool: Create Item --------------------------------------------------------
server.registerTool(
  "service_create_item",
  {
    description: "Create a new item. Returns the created item with its assigned ID.",
    inputSchema: z.object({
      name:        z.string().min(1).describe("Item name"),
      description: z.string().optional().describe("Optional item description"),
      tags:        z.array(z.string()).optional().describe("Tag list"),
    }),
    annotations: {
      readOnlyHint:    false,
      destructiveHint: false,
      idempotentHint:  false,
    },
  },
  async (params) => {
    const item = await client.request("/items", {
      method: "POST",
      body: JSON.stringify(params),
    });
    return {
      content: [{ type: "text", text: JSON.stringify(item, null, 2) }],
      structuredContent: item,
    };
  }
);

// --- Transport ----------------------------------------------------------------
const transport = new StdioServerTransport();
await server.connect(transport);
console.error("MCP server running on stdio");
```

### 3.3 Python FastMCP Server Template

```python
#!/usr/bin/env python3
"""FastMCP server template for Python-based MCP servers."""

import json
import os
from typing import Any

import httpx
from fastmcp import FastMCP
from pydantic import BaseModel, Field

# --- Server Initialization ---------------------------------------------------
mcp = FastMCP("my-service-mcp")

# --- API Client ---------------------------------------------------------------
BASE_URL = os.environ.get("SERVICE_BASE_URL", "https://api.example.com")
API_KEY  = os.environ.get("SERVICE_API_KEY", "")

async def api_request(endpoint: str, method: str = "GET", body: dict | None = None) -> Any:
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    }
    async with httpx.AsyncClient() as client:
        response = await client.request(
            method,
            f"{BASE_URL}{endpoint}",
            headers=headers,
            json=body,
        )
        if not response.is_success:
            raise ValueError(
                f"API error {response.status_code}: {response.text}. "
                "Check your credentials or verify the resource exists."
            )
        return response.json()

# --- Input Models -------------------------------------------------------------
class ListParams(BaseModel):
    page:     int   = Field(default=1,  ge=1,          description="Page number (1-indexed)")
    per_page: int   = Field(default=20, ge=1,  le=100, description="Items per page")
    filter:   str | None = Field(default=None,         description="Keyword filter")

class CreateParams(BaseModel):
    name:        str        = Field(...,       description="Item name")
    description: str | None = Field(None,      description="Optional description")
    tags:        list[str]  = Field(default=[], description="Tag list")

# --- Tools --------------------------------------------------------------------
@mcp.tool(description="List items with optional filtering and pagination.")
async def service_list_items(params: ListParams) -> str:
    query = f"?page={params.page}&per_page={params.per_page}"
    if params.filter:
        query += f"&q={params.filter}"
    data = await api_request(f"/items{query}")
    return json.dumps(data, indent=2)


@mcp.tool(description="Get a single item by ID.")
async def service_get_item(id: str) -> str:
    """id: Item ID to retrieve"""
    item = await api_request(f"/items/{id}")
    return json.dumps(item, indent=2)


@mcp.tool(description="Create a new item. Returns the created item with its assigned ID.")
async def service_create_item(params: CreateParams) -> str:
    item = await api_request("/items", method="POST", body=params.model_dump())
    return json.dumps(item, indent=2)


if __name__ == "__main__":
    mcp.run()
```

### 3.4 MCP Tool Design Checklist

Before shipping any MCP server, verify every tool against this checklist:

- [ ] Tool name uses `service_verb_noun` convention (e.g., `github_create_issue`)
- [ ] Description is a single sentence — concise, action-oriented
- [ ] All input fields have `description` populated
- [ ] Required vs. optional fields are correctly marked
- [ ] Numeric fields have `min`/`max` constraints
- [ ] Enum fields use `z.enum()`/`Literal` instead of free strings
- [ ] Annotations set: `readOnlyHint`, `destructiveHint`, `idempotentHint`
- [ ] Error messages suggest a remediation action
- [ ] Pagination supported for list endpoints
- [ ] `structuredContent` returned alongside text content
- [ ] Build compiles without errors: `npm run build` or `python -m py_compile`
- [ ] Tested with MCP Inspector: `npx @modelcontextprotocol/inspector`

### 3.5 MCP Error Message Patterns

Good error messages are diagnostic and actionable:

```typescript
// Bad
throw new Error("Not found");

// Good
throw new Error(
  `Item '${id}' not found. ` +
  `Use service_list_items to find valid IDs, or verify the item exists in the service.`
);

// Bad
throw new Error("Unauthorized");

// Good
throw new Error(
  `Authentication failed. ` +
  `Check that SERVICE_API_KEY is set and has the required 'items:read' scope. ` +
  `Generate a new key at https://service.example.com/settings/api-keys`
);
```

---

## 4. RAG System Construction

### 4.1 Complete RAG Pipeline

```
+---------------------------------------------------------------------+
|                       INGESTION PIPELINE                            |
|                                                                     |
|  Documents -> [Loader] -> [Chunker] -> [Embedder] -> [Vector Store] |
|                                          |                          |
|                                   [Metadata Store]                  |
+---------------------------------------------------------------------+
                                    |
                              (index built)
                                    |
+---------------------------------------------------------------------+
|                        QUERY PIPELINE                               |
|                                                                     |
|  Query -> [Query Embed] -> [Vector Search] -> [Reranker] -> [LLM]  |
|              |                  |                |                  |
|         [HyDE opt.]        [Metadata          [Context             |
|                              Filter]           Format]              |
+---------------------------------------------------------------------+
```

### 4.2 Full Python RAG Implementation

```python
#!/usr/bin/env python3
"""
Production RAG pipeline with chunking, embedding, retrieval, and reranking.
"""

import hashlib
import json
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

# --- Data Models -------------------------------------------------------------
@dataclass
class Document:
    content: str
    metadata: dict[str, Any] = field(default_factory=dict)
    doc_id: str = field(default="")
    
    def __post_init__(self):
        if not self.doc_id:
            self.doc_id = hashlib.md5(self.content.encode()).hexdigest()[:12]

@dataclass
class Chunk:
    content: str
    doc_id: str
    chunk_index: int
    metadata: dict[str, Any] = field(default_factory=dict)
    embedding: list[float] = field(default_factory=list)

@dataclass
class RetrievedChunk:
    chunk: Chunk
    score: float
    rerank_score: float = 0.0

# --- Chunking Strategies -----------------------------------------------------
class ChunkingStrategy:
    """Base class — override chunk()."""
    def chunk(self, doc: Document) -> list[Chunk]:
        raise NotImplementedError

class FixedSizeChunker(ChunkingStrategy):
    """Fixed token-count chunks with overlap. Good for general text."""
    
    def __init__(self, chunk_size: int = 512, overlap: int = 64):
        self.chunk_size = chunk_size
        self.overlap    = overlap
    
    def chunk(self, doc: Document) -> list[Chunk]:
        # Simple word-based split (use tiktoken for production token counting)
        words = doc.content.split()
        chunks = []
        step = self.chunk_size - self.overlap
        
        for i in range(0, len(words), step):
            chunk_words = words[i:i + self.chunk_size]
            if not chunk_words:
                break
            chunks.append(Chunk(
                content=" ".join(chunk_words),
                doc_id=doc.doc_id,
                chunk_index=len(chunks),
                metadata=doc.metadata,
            ))
        return chunks

class SentenceChunker(ChunkingStrategy):
    """Sentence-boundary-aware chunking. Better for structured prose."""
    
    def __init__(self, sentences_per_chunk: int = 5, overlap_sentences: int = 1):
        self.n  = sentences_per_chunk
        self.ov = overlap_sentences
    
    def chunk(self, doc: Document) -> list[Chunk]:
        import re
        sentences = re.split(r'(?<=[.!?])\s+', doc.content.strip())
        chunks = []
        step = self.n - self.ov
        
        for i in range(0, len(sentences), step):
            batch = sentences[i:i + self.n]
            if not batch:
                break
            chunks.append(Chunk(
                content=" ".join(batch),
                doc_id=doc.doc_id,
                chunk_index=len(chunks),
                metadata=doc.metadata,
            ))
        return chunks

class RecursiveChunker(ChunkingStrategy):
    """Hierarchical chunking — preserves section structure. Best for long docs."""
    
    SEPARATORS = ["\n\n", "\n", ". ", " "]
    
    def __init__(self, max_chunk_size: int = 800, min_chunk_size: int = 100):
        self.max_size = max_chunk_size
        self.min_size = min_chunk_size
    
    def chunk(self, doc: Document) -> list[Chunk]:
        chunks = []
        self._split(doc.content, doc.doc_id, doc.metadata, 0, chunks, [])
        return chunks
    
    def _split(self, text, doc_id, metadata, depth, chunks, idx_counter):
        if len(text.split()) <= self.max_size or depth >= len(self.SEPARATORS):
            idx_counter.append(None)
            chunks.append(Chunk(
                content=text,
                doc_id=doc_id,
                chunk_index=len(chunks),
                metadata=metadata,
            ))
            return
        
        sep = self.SEPARATORS[depth]
        parts = text.split(sep)
        current = ""
        
        for part in parts:
            candidate = (current + sep + part).strip() if current else part
            if len(candidate.split()) <= self.max_size:
                current = candidate
            else:
                if current and len(current.split()) >= self.min_size:
                    self._split(current, doc_id, metadata, depth + 1, chunks, idx_counter)
                current = part
        
        if current:
            self._split(current, doc_id, metadata, depth + 1, chunks, idx_counter)

# --- Vector Store (using Chroma for local dev) --------------------------------
class VectorStore:
    """Minimal abstract vector store interface."""
    
    def upsert(self, chunks: list[Chunk]) -> None:
        raise NotImplementedError
    
    def query(self, embedding: list[float], top_k: int = 10,
              filter_metadata: dict | None = None) -> list[RetrievedChunk]:
        raise NotImplementedError

class ChromaVectorStore(VectorStore):
    """Local development store using Chroma."""
    
    def __init__(self, collection_name: str, persist_dir: str = "./chroma_db"):
        import chromadb
        client = chromadb.PersistentClient(path=persist_dir)
        self.collection = client.get_or_create_collection(
            name=collection_name,
            metadata={"hnsw:space": "cosine"},
        )
    
    def upsert(self, chunks: list[Chunk]) -> None:
        self.collection.upsert(
            ids=[f"{c.doc_id}_{c.chunk_index}" for c in chunks],
            documents=[c.content for c in chunks],
            embeddings=[c.embedding for c in chunks],
            metadatas=[c.metadata for c in chunks],
        )
    
    def query(self, embedding: list[float], top_k: int = 10,
              filter_metadata: dict | None = None) -> list[RetrievedChunk]:
        kwargs: dict[str, Any] = {
            "query_embeddings": [embedding],
            "n_results": top_k,
            "include": ["documents", "metadatas", "distances"],
        }
        if filter_metadata:
            kwargs["where"] = filter_metadata
        
        results = self.collection.query(**kwargs)
        retrieved = []
        for doc, meta, dist in zip(
            results["documents"][0],
            results["metadatas"][0],
            results["distances"][0],
        ):
            retrieved.append(RetrievedChunk(
                chunk=Chunk(content=doc, doc_id=meta.get("doc_id",""), chunk_index=0, metadata=meta),
                score=1.0 - dist,  # cosine: distance -> similarity
            ))
        return retrieved

# --- Embedder ----------------------------------------------------------------
class Embedder:
    """Embed text using OpenAI-compatible API."""
    
    def __init__(self, model: str = "text-embedding-3-small"):
        import openai
        self.client = openai.OpenAI()
        self.model  = model
    
    def embed(self, texts: list[str]) -> list[list[float]]:
        response = self.client.embeddings.create(model=self.model, input=texts)
        return [item.embedding for item in response.data]
    
    def embed_one(self, text: str) -> list[float]:
        return self.embed([text])[0]

# --- Reranker ----------------------------------------------------------------
class CrossEncoderReranker:
    """Rerank retrieved chunks with a cross-encoder for precision improvement."""
    
    def __init__(self, model: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"):
        from sentence_transformers import CrossEncoder
        self.model = CrossEncoder(model)
    
    def rerank(self, query: str, chunks: list[RetrievedChunk], top_k: int = 5) -> list[RetrievedChunk]:
        pairs  = [(query, c.chunk.content) for c in chunks]
        scores = self.model.predict(pairs)
        for chunk, score in zip(chunks, scores):
            chunk.rerank_score = float(score)
        return sorted(chunks, key=lambda c: c.rerank_score, reverse=True)[:top_k]

# --- RAG Pipeline ------------------------------------------------------------
class RAGPipeline:
    def __init__(
        self,
        chunker: ChunkingStrategy,
        embedder: Embedder,
        vector_store: VectorStore,
        reranker: CrossEncoderReranker | None = None,
        llm_fn: callable = None,
        retrieval_top_k: int = 10,
        rerank_top_k: int = 5,
    ):
        self.chunker      = chunker
        self.embedder     = embedder
        self.store        = vector_store
        self.reranker     = reranker
        self.llm          = llm_fn
        self.retrieval_k  = retrieval_top_k
        self.rerank_k     = rerank_top_k
    
    # -- Ingestion ------------------------------------------------------------
    def ingest(self, documents: list[Document]) -> int:
        all_chunks: list[Chunk] = []
        
        for doc in documents:
            chunks = self.chunker.chunk(doc)
            texts  = [c.content for c in chunks]
            embeddings = self.embedder.embed(texts)
            for chunk, emb in zip(chunks, embeddings):
                chunk.embedding = emb
            all_chunks.extend(chunks)
        
        self.store.upsert(all_chunks)
        return len(all_chunks)
    
    # -- Query -----------------------------------------------------------------
    def retrieve(
        self,
        query: str,
        metadata_filter: dict | None = None,
        use_hyde: bool = False,
    ) -> list[RetrievedChunk]:
        # Optional HyDE: generate a hypothetical document to improve retrieval
        embed_target = query
        if use_hyde and self.llm:
            hypothetical = self.llm(
                f"Write a short passage that would answer this question:\n{query}"
            )
            embed_target = hypothetical
        
        q_embedding = self.embedder.embed_one(embed_target)
        candidates  = self.store.query(q_embedding, top_k=self.retrieval_k,
                                       filter_metadata=metadata_filter)
        
        if self.reranker and candidates:
            return self.reranker.rerank(query, candidates, top_k=self.rerank_k)
        return candidates[:self.rerank_k]
    
    def answer(self, query: str, metadata_filter: dict | None = None) -> dict:
        chunks = self.retrieve(query, metadata_filter=metadata_filter)
        
        context = "\n\n---\n\n".join(
            f"[Source {i+1}] {c.chunk.content}"
            for i, c in enumerate(chunks)
        )
        
        prompt = f"""Answer the question using ONLY the provided context.
If the context doesn't contain the answer, say "I don't have enough information."

Context:
{context}

Question: {query}

Answer:"""
        
        response = self.llm(prompt) if self.llm else "[No LLM configured]"
        
        return {
            "answer": response,
            "sources": [
                {
                    "content":  c.chunk.content[:200] + "...",
                    "score":    c.score,
                    "rerank":   c.rerank_score,
                    "metadata": c.chunk.metadata,
                }
                for c in chunks
            ],
        }
```

### 4.3 Vector Database Selection

| Database | Hosting | Scale | Latency | Best For |
|----------|---------|-------|---------|----------|
| Pinecone | Managed cloud | High | Low | Production, zero-ops |
| Qdrant | Self-hosted / cloud | High | Very Low | Performance-critical |
| Weaviate | Both | High | Low | Hybrid (keyword + vector) |
| Chroma | Self-hosted | Medium | Low | Local dev / prototyping |
| pgvector | Self-hosted (Postgres) | Medium | Medium | Existing Postgres stacks |
| Redis VSS | Both | Medium | Very Low | Real-time / cache-adjacent |
| Milvus | Self-hosted / cloud | Very High | Low | Enterprise scale |

### 4.4 Chunking Strategy Selection

| Strategy | Chunk Size | Overlap | Best For |
|----------|------------|---------|----------|
| Fixed-size | 500-1000 tokens | 50-100 tokens | General text, unknown structure |
| Sentence | 3-5 sentences | 1 sentence | News articles, documentation |
| Semantic | Variable | Meaning-based | Research papers, books |
| Recursive | Hierarchical | Parent-child | Long documents with headers |

### 4.5 RAG Evaluation Metrics

| Metric | Definition | Target |
|--------|------------|--------|
| Context Relevance | % of retrieved chunks relevant to query | > 0.80 |
| Answer Faithfulness | % of answer grounded in context | > 0.90 |
| Retrieval Precision@5 | Relevant chunks in top 5 / 5 | > 0.70 |
| Context Coverage | % of questions with >=1 relevant chunk in top-5 | > 0.85 |
| End-to-end Accuracy | Correct answers / total questions | > 0.80 |

```bash
# Evaluate a RAG pipeline
python scripts/rag_evaluator.py \
  --contexts retrieved_contexts.json \
  --questions eval_questions.json \
  --metrics relevance,faithfulness,coverage \
  --output report.json
```

---

## 5. Subagent Coordination

### 5.1 Subagent-Driven Development

**Core principle:** Fresh subagent per task + two-stage review (spec compliance, then code quality) = high quality, fast iteration.

This pattern runs entirely within the current session — no context switch to parallel sessions.

```
PROCESS FLOW:

1. Read plan -> extract all tasks with full text -> create TodoList

2. FOR EACH TASK:
   a. Dispatch Implementer subagent (full task text + context injected)
      +-> Subagent asks questions? -> Answer -> Re-dispatch
      +-> Subagent implements, tests, self-reviews, signals done
   
   b. Dispatch Spec Compliance Reviewer
      +-> Reviewer finds issues? -> Implementer fixes -> Re-review
      +-> OK Spec compliant -> proceed
   
   c. Dispatch Code Quality Reviewer (ONLY after spec review passes)
      +-> Reviewer finds issues? -> Implementer fixes -> Re-review
      +-> OK Quality approved -> mark task complete
   
3. After all tasks: Dispatch Final Code Reviewer for full implementation

4. Use finishing-a-development-branch workflow
```

### 5.2 Implementer Subagent Prompt Template

```markdown
# Implementer Subagent

## Context
You are implementing one task from a larger plan. You have been given full task text below.
Do NOT read plan files — the controller has already provided all necessary context.

## Project Context
{project_description}
Repository: {repo_path}
Branch: {branch_name}
Tech stack: {stack}

## Your Task
{full_task_text}

## Requirements
1. Ask questions BEFORE beginning if anything is unclear
2. Follow TDD: write failing test first, then implementation
3. Run all tests and verify they pass
4. Self-review: check for edge cases, naming, error handling
5. Commit with a descriptive message

## Output When Done
- Summary of what you implemented
- Test results (pass/fail counts)
- Any concerns or trade-offs you made
- Commit SHA
```

### 5.3 Spec Reviewer Prompt Template

```markdown
# Spec Compliance Reviewer

## Your Role
You are a spec compliance reviewer — NOT a code quality reviewer.
Your ONLY job: verify the implementation matches the spec. Nothing more.

## Task Spec
{task_spec}

## Implementation to Review
Git SHAs of new commits: {commit_shas}

## Review Criteria
Check for:
1. MISSING: Requirements in the spec not implemented
2. EXTRA: Features implemented that were NOT requested (scope creep)
3. WRONG: Imp

…(truncated)