| name | ai-agent-super-skill |
| description | 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. |
| metadata | {"author":"anavvanzin"} |
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
- Gap Analysis Table
- Agent Architecture & Design Patterns
- MCP Server Development
- RAG System Construction
- Subagent Coordination
- Execution Planning & Verification
- Prompt Engineering & Optimization
- ML Integration for Agents
- Skill & Capability Creation
- Backend Infrastructure for Agents
- Agent Deployment & Monitoring
- 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.
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.
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.
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.
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
- Fetch MCP spec:
https://modelcontextprotocol.io/sitemap.xml then pages with .md suffix
- Study the target API's documentation — auth requirements, rate limits, key endpoints
- Decide: TypeScript (recommended) or Python (FastMCP)
- 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
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
#!/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:
3.5 MCP Error Message Patterns
Good error messages are diagnostic and actionable:
// 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
#!/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 |
# 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
# 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
# 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: Implementation that contradicts the spec
## Output Format
STATUS: OK COMPLIANT or NOT COMPLIANT
If non-compliant, list each issue as:
- MISSING: [description]
- EXTRA: [description]
- WRONG: [description]
Do NOT comment on code quality, style, or performance.
5.4 Code Quality Reviewer Prompt Template
# Code Quality Reviewer
## Your Role
You are a code quality reviewer. The spec compliance reviewer has already confirmed
this implementation matches the spec — your job is code quality ONLY.
## Implementation to Review
Git SHAs: {commit_shas}
## Review Criteria
For each finding, classify as:
- CRITICAL: Must fix before merge (security, correctness, data loss)
- IMPORTANT: Should fix (maintainability, performance)
- SUGGESTION: Nice to have (style, naming)
## What to Check
- Error handling completeness
- Edge cases (null, empty, boundary values)
- Naming clarity
- Magic numbers/strings (extract to constants)
- DRY violations
- Security issues (injection, auth bypass, data exposure)
- Test coverage adequacy
## Output Format
STRENGTHS: [what's well done]
CRITICAL ISSUES: [list or "None"]
IMPORTANT ISSUES: [list or "None"]
VERDICT: OK APPROVED or CHANGES REQUIRED
5.5 Red Flags in Subagent Coordination
Never do these:
- Start code quality review before spec compliance passes — wrong order produces wasted cycles
- Dispatch multiple implementer subagents in parallel on the same codebase — merge conflicts guaranteed
- Let subagent read plan files — provide full task text in the prompt instead (eliminates file-reading overhead)
- Accept "close enough" spec compliance — reviewer found issues means the task is not done
- Skip the re-review after fixes — don't trust the fix without verification
- Skip scene-setting context in subagent prompts — subagent needs to understand where the task fits
6. Execution Planning & Verification
6.1 Parallel Agent Dispatch Pattern
Use when 2+ independent tasks can proceed without shared state or sequential dependencies.
Decision tree:
Multiple independent tasks?
+-- YES: Can they write to the same files/resources?
| +-- YES -> Sequential agents (avoid conflict)
| +-- NO -> Parallel dispatch OK
+-- NO: Tasks are related -> Single agent investigates all
Parallel dispatch template:
import asyncio
from typing import Callable, Any
async def dispatch_parallel_agents(
tasks: list[dict],
agent_fn: Callable[[dict], Any],
max_concurrent: int = 5,
) -> list[dict]:
"""
Dispatch multiple agent tasks in parallel with a concurrency limit.
tasks: list of dicts, each with 'id', 'description', 'context', 'constraints'
agent_fn: async function(task) -> result
"""
semaphore = asyncio.Semaphore(max_concurrent)
async def run_with_semaphore(task: dict) -> dict:
async with semaphore:
try:
result = await agent_fn(task)
return {"task_id": task["id"], "status": "success", "result": result}
except Exception as e:
return {"task_id": task["id"], "status": "failed", "error": str(e)}
results = await asyncio.gather(*[run_with_semaphore(t) for t in tasks])
return list(results)
def check_result_conflicts(results: list[dict]) -> list[str]:
"""
Scan parallel agent results for potential conflicts before integration.
Checks: same files modified, same database records mutated.
"""
conflicts = []
files_modified = {}
for result in results:
if result["status"] != "success":
continue
files = result.get("result", {}).get("files_modified", [])
for f in files:
if f in files_modified:
conflicts.append(
f"Conflict: '{f}' modified by both task "
f"{files_modified[f]} and {result['task_id']}"
)
else:
files_modified[f] = result["task_id"]
return conflicts
6.2 Focused Agent Task Prompt Structure
Good parallel agent task prompts are self-contained, specific, and constrained:
## Task: {task_title}
### Problem Statement
{specific_error_messages_or_failure_description}
### Scope
Files/subsystems in scope: {explicit_list}
Files/subsystems OUT of scope: {explicit_exclusions}
### Goal
{single_clear_success_criterion}
### Constraints
- Do NOT modify: {protected_files}
- Do NOT add new dependencies without flagging it
- {other_constraints}
### Required Output
Return:
1. Root cause analysis
2. What you changed and why
3. Verification output (test results, logs)
4. Summary of changes as a git diff or commit SHA
6.3 Batch Execution with Checkpoints
EXECUTING-PLANS PROCESS:
1. LOAD & REVIEW
- Read plan file once
- Identify concerns or blockers -> raise with human BEFORE starting
- Create TodoList from all tasks
- Announce: "Using executing-plans to implement this plan."
2. EXECUTE BATCH (default: 3 tasks per batch)
For each task in batch:
- Mark in_progress
- Follow steps exactly as written
- Run all verifications specified in plan
- Mark completed
3. CHECKPOINT REPORT
- Show: what was implemented, verification output
- Say: "Ready for feedback."
- Wait for approval before next batch
4. REPEAT until all tasks complete
5. FINISH
- Use finishing-a-development-branch workflow
- Verify all tests pass, no regressions
STOP IMMEDIATELY if:
- A blocker appears mid-batch
- Verification fails repeatedly
- Instructions are ambiguous
-> Ask for clarification; never guess.
6.4 Verification Workflow
Every implementation task should define explicit verification steps:
VERIFICATION_LEVELS = {
"smoke": [
"Application starts without errors",
"All previously-passing tests still pass",
"No new lint errors",
],
"functional": [
"New unit tests written and passing",
"Integration tests pass against staging",
"Edge cases tested (null, empty, boundary)",
],
"acceptance": [
"Feature works end-to-end as specified",
"Performance within targets (latency, throughput)",
"Security review passed",
"Documentation updated",
],
}
def build_verification_prompt(task: str, level: str = "functional") -> str:
checks = "\n".join(f"- [ ] {c}" for c in VERIFICATION_LEVELS[level])
return f"""
After implementing: {task}
Run these verification checks:
{checks}
Report each check as OK PASS, FAIL, or SKIPPED (with reason).
If any check FAILS, stop and report before proceeding.
"""
7. Prompt Engineering & Optimization
7.1 Core Prompt Patterns
| Pattern | When to Use | Token Cost | Quality Gain |
|---|
| Zero-shot | Simple, well-defined tasks | Lowest | Baseline |
| Few-shot (3-5 examples) | Complex tasks, consistent format needed | Medium | High |
| Chain-of-Thought (CoT) | Reasoning, math, multi-step logic | Medium | High |
| Role Prompting | Domain expertise, specific perspective | Low | Medium |
| Structured Output | Need parseable JSON/XML | Low | High (reliability) |
| Tree-of-Thought | Complex problem solving, backtracking | High | Very High |
| Meta-prompting | Generating/optimizing other prompts | High | Very High |
| Self-consistency | High-stakes decisions (majority vote) | Very High | High |
7.2 Chain-of-Thought Implementation
COT_TEMPLATES = {
# Standard CoT
"standard": """
{task}
Think step by step:
1. First, identify what information is given
2. Determine what needs to be found
3. Work through the reasoning systematically
4. State your conclusion
Reasoning:
""",
# Few-shot CoT
"few_shot": """
Solve problems by thinking step by step.
Example 1:
Problem: {example_problem_1}
Reasoning: {example_reasoning_1}
Answer: {example_answer_1}
Example 2:
Problem: {example_problem_2}
Reasoning: {example_reasoning_2}
Answer: {example_answer_2}
Now solve:
Problem: {problem}
Reasoning:
""",
# Zero-shot CoT (Kojima et al.)
"zero_shot": "{task}\n\nLet's think step by step.",
# Plan-then-execute CoT
"plan_execute": """
{task}
Step 1 - Make a plan: List the sub-problems you need to solve, in order.
Step 2 - Execute: Work through each sub-problem, showing your reasoning.
Step 3 - Verify: Check your answer against the original question.
Begin:
""",
}
7.3 Structured Output Design
from typing import Literal
from pydantic import BaseModel, Field
# --- Define Output Schema -----------------------------------------------------
class SentimentAnalysis(BaseModel):
summary: str = Field(..., max_length=200, description="Brief content summary")
sentiment: Literal["positive", "negative", "neutral", "mixed"]
confidence: float = Field(..., ge=0.0, le=1.0, description="Confidence 0-1")
key_points: list[str] = Field(..., max_items=5, description="Up to 5 key points")
# --- Build Prompt with Schema -------------------------------------------------
def build_structured_prompt(content: str, schema: type[BaseModel]) -> str:
schema_json = schema.model_json_schema()
return f"""Analyze the following content.
Respond ONLY with valid JSON matching this schema:
{json.dumps(schema_json, indent=2)}
IMPORTANT:
- Start your response with {{
- End your response with }}
- No markdown code fences, no explanation outside the JSON
Content to analyze:
{content}
JSON response:"""
# --- Parse and Validate Output ------------------------------------------------
def parse_structured_output(response: str, schema: type[BaseModel]) -> BaseModel:
# Strip markdown fences if present
import re
cleaned = re.sub(r"```(?:json)?\s*|\s*```", "", response).strip()
# Find outermost JSON object
start = cleaned.find("{")
end = cleaned.rfind("}") + 1
if start == -1 or end == 0:
raise ValueError(f"No JSON object found in response:\n{response}")
data = json.loads(cleaned[start:end])
return schema.model_validate(data)
7.4 Prompt Optimization Workflow
STEP 1: Baseline
python scripts/prompt_optimizer.py current_prompt.txt --analyze --output baseline.json
Capture: token count, clarity score, issues found
STEP 2: Identify Problems
| Issue | Apply This Pattern |
|--------------------|----------------------------------|
| Ambiguous output | Add explicit format specification |
| Too verbose | Extract to few-shot examples |
| Inconsistent results| Add role/persona framing |
| Missing edge cases | Add constraint boundaries |
| Poor reasoning | Add chain-of-thought trigger |
| Wrong format | Add schema + format enforcement |
STEP 3: Apply Optimizations
python scripts/prompt_optimizer.py current_prompt.txt --optimize --output optimized.txt
STEP 4: Compare
python scripts/prompt_optimizer.py optimized.txt --analyze --compare baseline.json
STEP 5: A/B Test
Run both prompts against held-out evaluation set.
Accept optimization only if: quality up AND cost <= 1.2x baseline.
7.5 Meta-Prompting (Prompt Generation)
META_PROMPT_GENERATOR = """
You are an expert prompt engineer. Generate an optimized prompt for the following use case.
## Use Case
Task: {task_description}
Model: {model}
Expected input format: {input_format}
Expected output format: {output_format}
Edge cases to handle: {edge_cases}
Constraints: {constraints}
## Generate a prompt that:
1. Uses role framing appropriate for the task
2. Provides clear, unambiguous instructions
3. Includes 2-3 few-shot examples if appropriate
4. Specifies exact output format
5. Handles the listed edge cases
6. Is token-efficient (no redundancy)
Return the complete prompt, ready to use.
"""
def generate_prompt(
task_description: str,
model: str = "claude-opus-4-5",
output_format: str = "JSON",
edge_cases: str = "empty input, ambiguous cases",
constraints: str = "respond in English only",
input_format: str = "plain text",
) -> str:
"""Use an LLM to generate an optimized prompt for a given task."""
import anthropic
client = anthropic.Anthropic()
response = client.messages.create(
model=model,
max_tokens=2000,
messages=[{
"role": "user",
"content": META_PROMPT_GENERATOR.format(
task_description=task_description,
model=model,
input_format=input_format,
output_format=output_format,
edge_cases=edge_cases,
constraints=constraints,
)
}]
)
return response.content[0].text
7.6 Few-Shot Example Design
EXAMPLE DESIGN CHECKLIST:
[ ] 3-5 examples (more = diminishing returns + token cost)
[ ] Covers: simple case, edge case, complex case, negative case
[ ] Consistent format across all examples
[ ] Output format matches expected production output exactly
[ ] Examples do NOT appear in test set (data contamination)
[ ] Ordered: simple -> complex (progressive difficulty)
EXAMPLE TEMPLATE:
Input: {diverse_input}
Output: {correctly_formatted_output}
[Repeat for each example with blank line between]
Now apply to:
Input: {actual_input}
Output:
8. ML Integration for Agents
8.1 LLM Provider Abstraction Layer
from abc import ABC, abstractmethod
from dataclasses import dataclass
import time
import anthropic
import openai
from tenacity import retry, stop_after_attempt, wait_exponential
@dataclass
class LLMResponse:
content: str
model: str
input_tokens: int
output_tokens: int
cost_usd: float
latency_ms: float
# Cost per 1K tokens (update as pricing changes)
PRICING = {
"claude-opus-4-5": {"input": 0.015, "output": 0.075},
"claude-haiku-3": {"input": 0.00025, "output": 0.00125},
"gpt-4o": {"input": 0.005, "output": 0.015},
"gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
}
class LLMProvider(ABC):
@abstractmethod
def complete(self, prompt: str, **kwargs) -> LLMResponse:
pass
def estimate_cost(self, prompt: str, output_tokens_estimate: int = 500) -> float:
model = getattr(self, "model", "unknown")
pricing = PRICING.get(model, {"input": 0.01, "output": 0.03})
input_tokens = len(prompt.split()) * 1.3 # rough approximation
return (input_tokens / 1000 * pricing["input"] +
output_tokens_estimate / 1000 * pricing["output"])
class AnthropicProvider(LLMProvider):
def __init__(self, model: str = "claude-opus-4-5"):
self.client = anthropic.Anthropic()
self.model = model
@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
def complete(self, prompt: str, max_tokens: int = 2048, **kwargs) -> LLMResponse:
start = time.time()
response = self.client.messages.create(
model=self.model,
max_tokens=max_tokens,
messages=[{"role": "user", "content": prompt}],
**kwargs,
)
latency = (time.time() - start) * 1000
pricing = PRICING.get(self.model, {"input": 0.015, "output": 0.075})
cost = (response.usage.input_tokens / 1000 * pricing["input"] +
response.usage.output_tokens / 1000 * pricing["output"])
return LLMResponse(
content=response.content[0].text,
model=self.model,
input_tokens=response.usage.input_tokens,
output_tokens=response.usage.output_tokens,
cost_usd=cost,
latency_ms=latency,
)
class OpenAIProvider(LLMProvider):
def __init__(self, model: str = "gpt-4o-mini"):
self.client = openai.OpenAI()
self.model = model
@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
def complete(self, prompt: str, max_tokens: int = 2048, **kwargs) -> LLMResponse:
start = time.time()
response = self.client.chat.completions.create(
model=self.model,
max_tokens=max_tokens,
messages=[{"role": "user", "content": prompt}],
**kwargs,
)
latency = (time.time() - start) * 1000
usage = response.usage
pricing = PRICING.get(self.model, {"input": 0.005, "output": 0.015})
cost = (usage.prompt_tokens / 1000 * pricing["input"] +
usage.completion_tokens / 1000 * pricing["output"])
return LLMResponse(
content=response.choices[0].message.content,
model=self.model,
input_tokens=usage.prompt_tokens,
output_tokens=usage.completion_tokens,
cost_usd=cost,
latency_ms=latency,
)
class FallbackProvider(LLMProvider):
"""Try primary provider; fall back to secondary on failure."""
def __init__(self, primary: LLMProvider, fallback: LLMProvider):
self.primary = primary
self.fallback = fallback
def complete(self, prompt: str, **kwargs) -> LLMResponse:
try:
return self.primary.complete(prompt, **kwargs)
except Exception as e:
print(f"Primary LLM failed ({e}), falling back...")
return self.fallback.complete(prompt, **kwargs)
8.2 Model Deployment for Agent Serving
# Dockerfile for agent service
FROM python:3.11-slim
WORKDIR /app
# Install dependencies first (layer caching)
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy source
COPY src/ ./src/
COPY config/ ./config/
# Health check endpoint
HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
CMD curl -f http://localhost:8080/health || exit 1
EXPOSE 8080
CMD ["uvicorn", "src.agent_server:app", "--host", "0.0.0.0", "--port", "8080", "--workers", "4"]
# src/agent_server.py — FastAPI agent serving wrapper
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
import asyncio
import uuid
app = FastAPI(title="Agent Service", version="1.0.0")
class AgentRequest(BaseModel):
query: str
session_id: str | None = None
max_steps: int = 10
metadata: dict = {}
class AgentResponse(BaseModel):
result: str
session_id: str
steps_used: int
cost_usd: float
latency_ms: float
@app.get("/health")
async def health():
return {"status": "ok", "version": "1.0.0"}
@app.post("/agent/run", response_model=AgentResponse)
async def run_agent(request: AgentRequest):
session_id = request.session_id or str(uuid.uuid4())
# ... agent execution logic ...
return AgentResponse(
result="...",
session_id=session_id,
steps_used=0,
cost_usd=0.0,
latency_ms=0.0,
)
@app.post("/agent/run-async")
async def run_agent_async(request: AgentRequest, background: BackgroundTasks):
job_id = str(uuid.uuid4())
background.add_task(_run_agent_task, job_id, request)
return {"job_id": job_id, "status": "queued"}
@app.get("/agent/status/{job_id}")
async def get_status(job_id: str):
# Fetch from job store (Redis, DB, etc.)
return {"job_id": job_id, "status": "unknown"}
async def _run_agent_task(job_id: str, request: AgentRequest):
# Run agent and persist result
pass
8.3 Model Monitoring for Agents
# Drift detection for agent input distributions
from scipy.stats import ks_2samp
import numpy as np
def detect_input_drift(
reference_inputs: list[str],
current_inputs: list[str],
threshold_p: float = 0.05,
) -> dict:
"""
Detect distribution shift in agent input queries using
token length distribution as a proxy metric.
"""
ref_lengths = np.array([len(t.split()) for t in reference_inputs])
cur_lengths = np.array([len(t.split()) for t in current_inputs])
stat, p_value = ks_2samp(ref_lengths, cur_lengths)
return {
"drift_detected": p_value < threshold_p,
"ks_statistic": float(stat),
"p_value": float(p_value),
"ref_mean_tokens": float(ref_lengths.mean()),
"cur_mean_tokens": float(cur_lengths.mean()),
"recommendation": (
"Retrain or re-evaluate agent prompts — input distribution shifted significantly."
if p_value < threshold_p else
"No significant drift detected."
),
}
# Alert thresholds for agent monitoring
AGENT_ALERT_THRESHOLDS = {
"p95_latency_ms": {"warning": 2000, "critical": 5000},
"error_rate_pct": {"warning": 1.0, "critical": 5.0},
"cost_per_query_usd": {"warning": 0.05, "critical": 0.20},
"tool_failure_rate": {"warning": 0.05, "critical": 0.15},
"token_overflow_rate":{"warning": 0.02, "critical": 0.10},
}
8.4 Serving Strategy Selection
| Strategy | Latency | Throughput | Cost | Use Case |
|---|
| FastAPI + Uvicorn | Low | Medium | Low | REST agent APIs, single-model |
| Ray Serve | Medium | Very High | Medium | Multi-model pipelines, scaling |
| Triton Inference | Very Low | Very High | Medium | GPU batch inference |
| Serverless (Lambda/Cloud Run) | Cold-start medium | Auto-scale | Pay-per-use | Bursty agent tasks |
| Streaming (SSE/WebSocket) | Apparent Low | Medium | Low | Conversational agents |
9. Skill & Capability Creation
9.1 SKILL.md Format Specification
Every Perplexity Computer skill must follow this exact format:
---
name: skill-name-with-hyphens
description: One or two sentences describing when to use this skill. Start with "Use when..." or describe the trigger conditions clearly.
license: MIT
metadata:
author: your-username
version: '1.0'
---
# Skill Title
Brief one-paragraph overview of the skill's purpose.
## When to Use
...
## Core Concepts
...
## Step-by-Step Process
...
## Examples
...
## Common Mistakes
...
9.2 Validation Pipeline
# Validate skill structure and frontmatter
cd /home/user/workspace && uvx --from skills-ref agentskills validate <skill-name>
# What the validator checks:
# OK First line is exactly ---
# OK YAML frontmatter present and parseable
# OK Required fields: name, description, license, metadata.author, metadata.version
# OK name matches directory name
# OK version is quoted string ('1.0' not 1.0)
# OK Skill directory exists at workspace/<skill-name>/SKILL.md
# OK No syntax errors in YAML block
9.3 Skill Quality Checklist
Before publishing any skill:
9.4 Skill Packaging for Distribution
skill-directory/
+-- SKILL.md <- Main skill file (required)
+-- README.md <- Human-readable docs (optional)
+-- examples/
| +-- basic.md <- Annotated simple example
| +-- advanced.md <- Complex workflow example
+-- templates/
| +-- prompt_template.md
| +-- config.yaml
+-- scripts/
+-- validate.py <- Skill-specific validation helpers
9.5 Skill Trigger Design
The description field determines when the skill is loaded. Write it to trigger on the right signals:
# Bad — too vague, triggers on everything
description: Helps build things with AI.
# Bad — too narrow, misses many triggers
description: Use when the user types "build an MCP server".
# Good — triggers on intent, not exact phrasing
description: >
Use when building AI agents, creating MCP servers, designing RAG systems,
coordinating subagents, optimizing prompts, or architecting any AI-powered
automation workflow. Covers agent design patterns, multi-agent orchestration,
and production deployment.
10. Backend Infrastructure for Agents
10.1 Agent Memory Persistence with SQLite (CGI-bin)
Agents need persistent memory across sessions. The CGI-bin pattern lets agents store and retrieve state via HTTP endpoints without a dedicated backend server.
#!/usr/bin/env python3
# cgi-bin/agent_memory.py
# Agent memory store: conversations, tool results, learned facts
import json
import os
import sqlite3
import sys
from datetime import datetime
DB_PATH = "agent_memory.db"
def init_db(conn):
conn.executescript("""
CREATE TABLE IF NOT EXISTS sessions (
session_id TEXT PRIMARY KEY,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
metadata TEXT DEFAULT '{}'
);
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
role TEXT NOT NULL CHECK(role IN ('user','assistant','tool','system')),
content TEXT NOT NULL,
tool_name TEXT,
tool_input TEXT,
tool_result TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (session_id) REFERENCES sessions(session_id)
);
CREATE TABLE IF NOT EXISTS facts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT,
key TEXT NOT NULL,
value TEXT NOT NULL,
confidence REAL DEFAULT 1.0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
expires_at TIMESTAMP,
UNIQUE(session_id, key)
);
CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id);
CREATE INDEX IF NOT EXISTS idx_facts_key ON facts(key);
""")
conn.commit()
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
init_db(conn)
method = os.environ.get("REQUEST_METHOD", "GET")
query = os.environ.get("QUERY_STRING", "")
path_info = os.environ.get("PATH_INFO", "")
def respond(data, status=200):
print(f"Status: {status}")
print("Content-Type: application/json")
print()
print(json.dumps(data))
def parse_qs(qs):
params = {}
for part in qs.split("&"):
if "=" in part:
k, v = part.split("=", 1)
params[k] = v
return params
# -- Routes --------------------------------------------------------------------
if path_info == "/sessions" and method == "POST":
body = json.loads(sys.stdin.read() or "{}")
sid = body.get("session_id") or f"sess_{datetime.utcnow().strftime('%Y%m%d_%H%M%S_%f')}"
conn.execute("INSERT OR IGNORE INTO sessions (session_id, metadata) VALUES (?,?)",
[sid, json.dumps(body.get("metadata", {}))])
conn.commit()
respond({"session_id": sid}, 201)
elif path_info == "/messages" and method == "POST":
body = json.loads(sys.stdin.read())
conn.execute(
"INSERT INTO messages (session_id, role, content, tool_name, tool_input, tool_result) "
"VALUES (?,?,?,?,?,?)",
[body["session_id"], body["role"], body["content"],
body.get("tool_name"), body.get("tool_input"), body.get("tool_result")]
)
conn.commit()
msg_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
respond({"id": msg_id}, 201)
elif path_info == "/messages" and method == "GET":
params = parse_qs(query)
sid = params.get("session_id", "")
limit = int(params.get("limit", 50))
rows = conn.execute(
"SELECT * FROM messages WHERE session_id=? ORDER BY created_at LIMIT ?",
[sid, limit]
).fetchall()
respond([dict(r) for r in rows])
elif path_info == "/facts" and method == "PUT":
body = json.loads(sys.stdin.read())
conn.execute(
"INSERT OR REPLACE INTO facts (session_id, key, value, confidence) VALUES (?,?,?,?)",
[body.get("session_id"), body["key"], json.dumps(body["value"]),
body.get("confidence", 1.0)]
)
conn.commit()
respond({"status": "ok"})
elif path_info == "/facts" and method == "GET":
params = parse_qs(query)
sid = params.get("session_id", "")
rows = conn.execute(
"SELECT key, value, confidence FROM facts WHERE session_id=? OR session_id IS NULL",
[sid]
).fetchall()
respond({r["key"]: {"value": json.loads(r["value"]), "confidence": r["confidence"]}
for r in rows})
else:
respond({"error": f"Unknown route: {method} {path_info}"}, 400)
10.2 Webhook Receiver for Agent Triggers
#!/usr/bin/env python3
# cgi-bin/webhook_receiver.py
# Receives external events and queues them for agent processing
import hashlib
import hmac
import json
import os
import sqlite3
import sys
import time
DB_PATH = "webhook_events.db"
WEBHOOK_SECRET = os.environ.get("WEBHOOK_SECRET", "")
conn = sqlite3.connect(DB_PATH)
conn.execute("""
CREATE TABLE IF NOT EXISTS events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
source TEXT NOT NULL,
event_type TEXT NOT NULL,
payload TEXT NOT NULL,
processed INTEGER DEFAULT 0,
received_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.commit()
method = os.environ.get("REQUEST_METHOD", "GET")
def verify_signature(body: str, signature: str, secret: str) -> bool:
"""Verify HMAC-SHA256 webhook signature."""
if not secret:
return True # Skip verification if no secret configured
expected = "sha256=" + hmac.new(
secret.encode(), body.encode(), hashlib.sha256
).hexdigest()
return hmac.compare_digest(signature, expected)
if method == "POST":
raw_body = sys.stdin.read()
sig = os.environ.get("HTTP_X_HUB_SIGNATURE_256", "")
if not verify_signature(raw_body, sig, WEBHOOK_SECRET):
print("Status: 401")
print("Content-Type: application/json")
print()
print('{"error": "Invalid signature"}')
sys.exit(0)
body = json.loads(raw_body)
conn.execute(
"INSERT INTO events (source, event_type, payload) VALUES (?,?,?)",
[body.get("source", "unknown"), body.get("type", "unknown"), raw_body]
)
conn.commit()
print("Status: 202")
print("Content-Type: application/json")
print()
print('{"status": "accepted"}')
elif method == "GET":
# Dequeue unprocessed events for agent polling
rows = conn.execute(
"SELECT * FROM events WHERE processed=0 ORDER BY received_at LIMIT 50"
).fetchall()
events = [
{"id": r[0], "source": r[1], "event_type": r[2],
"payload": json.loads(r[3]), "received_at": r[5]}
for r in rows
]
print("Content-Type: application/json")
print()
print(json.dumps(events))
10.3 Agent-to-Agent Communication via Message Bus
#!/usr/bin/env python3
# cgi-bin/message_bus.py
# Simple pub/sub message bus for multi-agent coordination
import json
import os
import sqlite3
import sys
import uuid
DB_PATH = "message_bus.db"
conn = sqlite3.connect(DB_PATH)
conn.executescript("""
CREATE TABLE IF NOT EXISTS messages (
id TEXT PRIMARY KEY,
from_agent TEXT NOT NULL,
to_agent TEXT, -- NULL = broadcast
topic TEXT NOT NULL,
payload TEXT NOT NULL,
ack INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS subscriptions (
agent_id TEXT NOT NULL,
topic TEXT NOT NULL,
PRIMARY KEY (agent_id, topic)
);
""")
conn.commit()
method = os.environ.get("REQUEST_METHOD", "GET")
path_info = os.environ.get("PATH_INFO", "")
query = os.environ.get("QUERY_STRING", "")
def respond(data, status=200):
print(f"Status: {status}")
print("Content-Type: application/json")
print()
print(json.dumps(data))
if path_info == "/publish" and method == "POST":
body = json.loads(sys.stdin.read())
msg_id = str(uuid.uuid4())
conn.execute(
"INSERT INTO messages (id, from_agent, to_agent, topic, payload) VALUES (?,?,?,?,?)",
[msg_id, body["from_agent"], body.get("to_agent"),
body["topic"], json.dumps(body["payload"])]
)
conn.commit()
respond({"message_id": msg_id}, 201)
elif path_info == "/subscribe" and method == "POST":
body = json.loads(sys.stdin.read())
conn.execute(
"INSERT OR IGNORE INTO subscriptions VALUES (?,?)",
[body["agent_id"], body["topic"]]
)
conn.commit()
respond({"status": "subscribed"})
elif path_info == "/poll" and method == "GET":
params = dict(p.split("=") for p in query.split("&") if "=" in p)
agent_id = params.get("agent_id", "")
# Get messages for this agent (direct + subscribed topics)
rows = conn.execute("""
SELECT m.* FROM messages m
LEFT JOIN subscriptions s ON s.agent_id=? AND s.topic=m.topic
WHERE m.ack=0 AND (m.to_agent=? OR (m.to_agent IS NULL AND s.agent_id IS NOT NULL))
ORDER BY m.created_at LIMIT 20
""", [agent_id, agent_id]).fetchall()
messages = [{"id": r[0], "from": r[1], "topic": r[3],
"payload": json.loads(r[4])} for r in rows]
# Mark as acked
if messages:
ids = [m["id"] for m in messages]
conn.execute(f"UPDATE messages SET ack=1 WHERE id IN ({','.join('?'*len(ids))})", ids)
conn.commit()
respond(messages)
10.4 JavaScript Client for Agent Backend
// Agent backend client — use __CGI_BIN__ placeholder in production
const CGI_BIN = "__CGI_BIN__";
class AgentMemoryClient {
async createSession(metadata = {}) {
const res = await fetch(`${CGI_BIN}/agent_memory.py/sessions`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ metadata }),
});
return res.json();
}
async addMessage(sessionId, role, content, toolData = {}) {
const res = await fetch(`${CGI_BIN}/agent_memory.py/messages`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ session_id: sessionId, role, content, ...toolData }),
});
return res.json();
}
async getHistory(sessionId, limit = 50) {
const res = await fetch(
`${CGI_BIN}/agent_memory.py/messages?session_id=${sessionId}&limit=${limit}`
);
return res.json();
}
async setFact(sessionId, key, value, confidence = 1.0) {
const res = await fetch(`${CGI_BIN}/agent_memory.py/facts`, {
method: "PUT",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ session_id: sessionId, key, value, confidence }),
});
return res.json();
}
async getFacts(sessionId) {
const res = await fetch(
`${CGI_BIN}/agent_memory.py/facts?session_id=${sessionId}`
);
return res.json();
}
}
class MessageBusClient {
async publish(fromAgent, topic, payload, toAgent = null) {
const res = await fetch(`${CGI_BIN}/message_bus.py/publish`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ from_agent: fromAgent, topic, payload, to_agent: toAgent }),
});
return res.json();
}
async subscribe(agentId, topic) {
const res = await fetch(`${CGI_BIN}/message_bus.py/subscribe`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ agent_id: agentId, topic }),
});
return res.json();
}
async poll(agentId) {
const res = await fetch(`${CGI_BIN}/message_bus.py/poll?agent_id=${agentId}`);
return res.json();
}
}
11. Agent Deployment & Monitoring
11.1 Deployment Checklist
Before deploying any agent to production:
Infrastructure
Observability
Security
Cost Controls
11.2 Structured Agent Logging
import json
import logging
import time
import uuid
from contextlib import contextmanager
from dataclasses import asdict, dataclass
@dataclass
class AgentTraceEvent:
trace_id: str
span_id: str
event_type: str # "llm_call" | "tool_call" | "agent_start" | "agent_end"
timestamp: float
agent_id: str
model: str | None = None
tool_name: str | None = None
input_tokens: int = 0
output_tokens: int = 0
cost_usd: float = 0.0
latency_ms: float = 0.0
error: str | None = None
metadata: dict | None = None
class AgentTracer:
def __init__(self, agent_id: str):
self.agent_id = agent_id
self.logger = logging.getLogger("agent.trace")
def log(self, event: AgentTraceEvent):
self.logger.info(json.dumps(asdict(event)))
@contextmanager
def trace_llm_call(self, trace_id: str, model: str):
span_id = str(uuid.uuid4())[:8]
start = time.time()
event = AgentTraceEvent(
trace_id=trace_id, span_id=span_id,
event_type="llm_call", timestamp=start,
agent_id=self.agent_id, model=model,
)
try:
yield event
except Exception as e:
event.error = str(e)
raise
finally:
event.latency_ms = (time.time() - start) * 1000
self.log(event)
@contextmanager
def trace_tool_call(self, trace_id: str, tool_name: str):
span_id = str(uuid.uuid4())[:8]
start = time.time()
event = AgentTraceEvent(
trace_id=trace_id, span_id=span_id,
event_type="tool_call", timestamp=start,
agent_id=self.agent_id, tool_name=tool_name,
)
try:
yield event
except Exception as e:
event.error = str(e)
raise
finally:
event.latency_ms = (time.time() - start) * 1000
self.log(event)
11.3 Agent Health Dashboard Schema
# Metrics to expose at /metrics (Prometheus-compatible)
AGENT_METRICS = """
# HELP agent_requests_total Total requests processed
# TYPE agent_requests_total counter
agent_requests_total{agent_id="{agent_id}",status="success"} {success_count}
agent_requests_total{agent_id="{agent_id}",status="error"} {error_count}
# HELP agent_latency_seconds Request latency
# TYPE agent_latency_seconds histogram
agent_latency_seconds_bucket{{agent_id="{agent_id}",le="0.5"}} {bucket_0_5}
agent_latency_seconds_bucket{{agent_id="{agent_id}",le="1.0"}} {bucket_1_0}
