--- name: ai-engineering description: "Comprehensive 16-skill suite for AI engineers: RAG pipelines, Vector DBs, Multi-Agent orchestration, MCP servers, Guardrails, and LLM evaluation." license: MIT metadata: author: jgamaraalv repo: jgamaraalv/delivery-loop version: "1.0.0" ---

A complete production-grade toolkit for building modern AI applications, autonomous agents, and LLM systems. Covers 16 key engineering areas from RAG and vector databases to MCP servers, multi-agent orchestration, guardrails, and automated evaluation.

Design end-to-end RAG pipelines for accurate document retrieval and generation.

Documents → Chunking → Embedding → Vector Store → Retrieval → Reranking → Generation

# Semantic chunking (recommended)
from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter( chunk_size=1000, # Characters per chunk chunk_overlap=200, # Overlap between chunks separators=["\n\n", "\n", ". ", " ", ""], length_function=len, )

chunks = splitter.split_text(document.text)

interface ChunkMetadata {
  // Source information
  document_id: string;
  source: string;
  url?: string;

// Location page?: number; section?: string; chunk_index: number;

// Content classification content_type: "text" | "code" | "table" | "list"; language?: string;

// Timestamps created_at: Date; updated_at: Date;

// Retrieval optimization keywords: string[]; summary?: string; importance_score?: number; } ```

# Pinecone example
import pinecone
from langchain.vectorstores import Pinecone
from langchain.embeddings import OpenAIEmbeddings

pinecone.init(api_key="...", environment="...")

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

vectorstore = Pinecone.from_documents( documents=chunks, embedding=embeddings, index_name="knowledge-base", namespace="production", ) ```

# Hybrid search (dense + sparse)
def hybrid_retrieval(query: str, k: int = 5):
    # Dense retrieval (semantic)
    dense_results = vectorstore.similarity_search(query, k=k*2)

return combined[:k]

from sentence_transformers import CrossEncoder

reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

def rerank_results(query: str, results: List[Document], top_k: int = 3): # Score each result against query pairs = [(query, doc.page_content) for doc in results] scores = reranker.predict(pairs)

return [doc for doc, score in scored_results[:top_k]] ```

# Query expansion
def expand_query(query: str) -> str:
    expansion_prompt = f"""
    Generate 3 alternative phrasings of this query:
    "{query}"

Return as JSON array of strings. """ alternatives = llm(expansion_prompt) return [query] + alternatives

for q in queries: results = vectorstore.similarity_search(q, k=k) all_results.extend(results)

# Define golden dataset
golden_dataset = [
    {
        "query": "How do I authenticate users?",
        "expected_docs": ["auth_guide.md", "user_management.md"],
        "relevant_chunks": ["chunk_123", "chunk_456"],
    },
]

for item in dataset: retrieved = retrieval_fn(item["query"]) retrieved_ids = [doc.metadata["chunk_id"] for doc in retrieved]

precision = len(relevant & retrieved_set) / len(retrieved_set) recall = len(relevant & retrieved_set) / len(relevant)

results["precision"].append(precision) results["recall"].append(recall)

return {k: sum(v)/len(v) for k, v in results.items()} ```

def fit_context_window(chunks: List[Document], max_tokens: int = 4000):
    """Select chunks that fit in context window"""
    total_tokens = 0
    selected_chunks = []

for chunk in chunks: chunk_tokens = count_tokens(chunk.page_content) if total_tokens + chunk_tokens <= max_tokens: selected_chunks.append(chunk) total_tokens += chunk_tokens else: break

return selected_chunks ```

1. Chunk size: 500-1000 chars for general text
2. Overlap: 10-20% overlap between chunks
3. Metadata: Rich metadata for filtering
4. Hybrid search: Combine semantic + keyword
5. Reranking: Cross-encoder for final ranking
6. Evaluation: Golden dataset with metrics
7. Context management: Don't exceed model limits

• Chunking strategy defined
• Metadata schema documented
• Vector store configured
• Retrieval algorithm implemented
• Reranking pipeline added
• Query enhancement (optional)
• Context window management
• Evaluation dataset created
• Metrics implementation
• Performance baseline established

Configure vector databases for semantic search and AI applications.

1. Choose database: Select based on requirements
2. Setup connection: Configure client
3. Generate embeddings: Create vector representations
4. Index documents: Store with metadata
5. Query vectors: Semantic similarity search
6. Optimize: Tune for performance

// embeddings/openai.ts
import OpenAI from 'openai';

const openai = new OpenAI();

export async function generateEmbedding(text: string): Promise<number[]> { const response = await openai.embeddings.create({ model: 'text-embedding-3-small', // or text-embedding-3-large input: text, });

return response.data[0].embedding; }

export async function generateEmbeddings(texts: string[]): Promise<number[][]> { const response = await openai.embeddings.create({ model: 'text-embedding-3-small', input: texts, });

return response.data.map((d) => d.embedding); } ```

// embeddings/batch.ts
const BATCH_SIZE = 100;

export async function batchGenerateEmbeddings( texts: string[] ): Promise<number[][]> { const embeddings: number[][] = [];

for (let i = 0; i < texts.length; i += BATCH_SIZE) { const batch = texts.slice(i, i + BATCH_SIZE); const batchEmbeddings = await generateEmbeddings(batch); embeddings.push(...batchEmbeddings);

// Rate limiting if (i + BATCH_SIZE < texts.length) { await new Promise((resolve) => setTimeout(resolve, 100)); } }

return embeddings; } ```

npm install @pinecone-database/pinecone

// db/pinecone.ts
import { Pinecone } from '@pinecone-database/pinecone';

const pinecone = new Pinecone({ apiKey: process.env.PINECONE_API_KEY!, });

// Get or create index export async function getIndex(indexName: string) { const indexes = await pinecone.listIndexes();

if (!indexes.indexes?.find((i) => i.name === indexName)) { await pinecone.createIndex({ name: indexName, dimension: 1536, // OpenAI embedding dimension metric: 'cosine', spec: { serverless: { cloud: 'aws', region: 'us-east-1', }, }, });

// Wait for index to be ready await new Promise((resolve) => setTimeout(resolve, 60000)); }

return pinecone.Index(indexName); } ```

// db/pinecone-ops.ts
import { getIndex } from './pinecone';
import { generateEmbedding, generateEmbeddings } from '../embeddings/openai';

const index = await getIndex('my-index');

interface Document { id: string; content: string; metadata: Record<string, any>; }

// Upsert documents export async function upsertDocuments( documents: Document[], namespace = 'default' ) { const embeddings = await generateEmbeddings(documents.map((d) => d.content));

const vectors = documents.map((doc, i) => ({ id: doc.id, values: embeddings[i], metadata: { content: doc.content, ...doc.metadata, }, }));

// Upsert in batches const BATCH_SIZE = 100; for (let i = 0; i < vectors.length; i += BATCH_SIZE) { const batch = vectors.slice(i, i + BATCH_SIZE); await index.namespace(namespace).upsert(batch); } }

// Query similar documents export async function querySimilar( query: string, options: { topK?: number; namespace?: string; filter?: Record<string, any>; } = {} ) { const { topK = 5, namespace = 'default', filter } = options;

const queryEmbedding = await generateEmbedding(query);

const results = await index.namespace(namespace).query({ vector: queryEmbedding, topK, includeMetadata: true, filter, });

return results.matches?.map((match) => ({ id: match.id, score: match.score, content: match.metadata?.content, metadata: match.metadata, })); }

// Delete documents export async function deleteDocuments(ids: string[], namespace = 'default') { await index.namespace(namespace).deleteMany(ids); }

// Delete by filter export async function deleteByFilter( filter: Record<string, any>, namespace = 'default' ) { await index.namespace(namespace).deleteMany({ filter }); } ```

npm install chromadb

// db/chroma.ts
import { ChromaClient, OpenAIEmbeddingFunction } from 'chromadb';

const client = new ChromaClient({ path: process.env.CHROMA_URL || 'http://localhost:8000', });

const embedder = new OpenAIEmbeddingFunction({ openai_api_key: process.env.OPENAI_API_KEY!, openai_model: 'text-embedding-3-small', });

export async function getCollection(name: string) { return client.getOrCreateCollection({ name, embeddingFunction: embedder, metadata: { 'hnsw:space': 'cosine' }, }); } ```

// db/chroma-ops.ts
import { getCollection } from './chroma';

const collection = await getCollection('documents');

// Add documents (Chroma generates embeddings) export async function addDocuments(documents: Document[]) { await collection.add({ ids: documents.map((d) => d.id), documents: documents.map((d) => d.content), metadatas: documents.map((d) => d.metadata), }); }

// Query export async function query(queryText: string, nResults = 5) { const results = await collection.query({ queryTexts: [queryText], nResults, });

return results.ids[0].map((id, i) => ({ id, content: results.documents?.[0][i], metadata: results.metadatas?.[0][i], distance: results.distances?.[0][i], })); }

// Query with filter export async function queryWithFilter( queryText: string, filter: Record<string, any>, nResults = 5 ) { const results = await collection.query({ queryTexts: [queryText], nResults, where: filter, });

return results; }

// Update document export async function updateDocument(id: string, content: string, metadata?: Record<string, any>) { await collection.update({ ids: [id], documents: [content], metadatas: metadata ? [metadata] : undefined, }); }

// Delete export async function deleteDocuments(ids: string[]) { await collection.delete({ ids }); } ```

npm install pg pgvector

-- Enable extension
CREATE EXTENSION vector;

-- Create table CREATE TABLE documents ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), content TEXT NOT NULL, metadata JSONB, embedding vector(1536), created_at TIMESTAMP DEFAULT NOW() );

-- Create index for similarity search CREATE INDEX ON documents USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);

-- Or use HNSW (better for production) CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops); ```

// db/pgvector.ts
import { Pool } from 'pg';
import pgvector from 'pgvector/pg';

const pool = new Pool({ connectionString: process.env.DATABASE_URL, });

// Register pgvector type await pgvector.registerType(pool);

// Insert document export async function insertDocument( content: string, embedding: number[], metadata?: Record<string, any> ) { const result = await pool.query( INSERT INTO documents (content, embedding, metadata) VALUES ($1, $2, $3) RETURNING id, [content, pgvector.toSql(embedding), metadata] );

return result.rows[0].id; }

// Similarity search export async function searchSimilar( queryEmbedding: number[], limit = 5, threshold = 0.7 ) { const result = await pool.query( SELECT id, content, metadata, 1 - (embedding <=> $1) as similarity FROM documents WHERE 1 - (embedding <=> $1) > $2 ORDER BY embedding <=> $1 LIMIT $3, [pgvector.toSql(queryEmbedding), threshold, limit] );

return result.rows; }

// Search with metadata filter export async function searchWithFilter( queryEmbedding: number[], filter: Record<string, any>, limit = 5 ) { const result = await pool.query( SELECT id, content, metadata, 1 - (embedding <=> $1) as similarity FROM documents WHERE metadata @> $2 ORDER BY embedding <=> $1 LIMIT $3, [pgvector.toSql(queryEmbedding), filter, limit] );

return result.rows; }

// Hybrid search (vector + full-text) export async function hybridSearch( queryEmbedding: number[], textQuery: string, limit = 5 ) { const result = await pool.query( SELECT id, content, metadata, (1 - (embedding <=> $1)) * 0.7 + ts_rank(to_tsvector(content), plainto_tsquery($2)) * 0.3 as score FROM documents WHERE to_tsvector(content) @@ plainto_tsquery($2) OR 1 - (embedding <=> $1) > 0.5 ORDER BY score DESC LIMIT $3, [pgvector.toSql(queryEmbedding), textQuery, limit] );

return result.rows; } ```

npm install @qdrant/js-client-rest

// db/qdrant.ts
import { QdrantClient } from '@qdrant/js-client-rest';

const client = new QdrantClient({ url: process.env.QDRANT_URL, apiKey: process.env.QDRANT_API_KEY, });

// Create collection export async function createCollection(name: string) { await client.createCollection(name, { vectors: { size: 1536, distance: 'Cosine', }, }); }

// Upsert points export async function upsertPoints( collection: string, points: Array<{ id: string; vector: number[]; payload: Record<string, any>; }> ) { await client.upsert(collection, { points: points.map((p) => ({ id: p.id, vector: p.vector, payload: p.payload, })), }); }

// Search export async function search( collection: string, vector: number[], limit = 5, filter?: Record<string, any> ) { const results = await client.search(collection, { vector, limit, filter: filter ? { must: Object.entries(filter).map(([key, value]) => ({ key, match: { value }, })), } : undefined, with_payload: true, });

return results; } ```

// pipeline/ingest.ts
import { RecursiveCharacterTextSplitter } from 'langchain/text_splitter';
import { generateEmbeddings } from '../embeddings/openai';
import { upsertDocuments } from '../db/pinecone-ops';

interface RawDocument { id: string; content: string; source: string; metadata?: Record<string, any>; }

export async function ingestDocuments(documents: RawDocument[]) { const splitter = new RecursiveCharacterTextSplitter({ chunkSize: 1000, chunkOverlap: 200, });

const chunks: Array<{ id: string; content: string; metadata: Record<string, any>; }> = [];

for (const doc of documents) { const splits = await splitter.splitText(doc.content);

splits.forEach((text, index) => { chunks.push({ id: ${doc.id}-chunk-${index}, content: text, metadata: { source: doc.source, documentId: doc.id, chunkIndex: index, ...doc.metadata, }, }); }); }

// Upsert in batches await upsertDocuments(chunks);

return { totalChunks: chunks.length }; } ```

1. Choose the right dimension: Match embedding model
2. Use namespaces: Organize data logically
3. Add metadata: Enable filtering
4. Batch operations: Reduce API calls
5. Index appropriately: HNSW for speed, IVF for memory
6. Monitor performance: Track latency and recall
7. Cache embeddings: Avoid regenerating
8. Use hybrid search: Combine vector and keyword

Every vector database setup should include:

• Database client configured
• Embedding generation function
• Collection/index creation
• Document upsert with metadata
• Similarity search function
• Filtered search capability
• Batch processing for large datasets
• Delete/update operations
• Error handling
• Performance monitoring

Build production-ready document embedding and retrieval pipelines.

1. Load documents: Ingest from various sources
2. Preprocess text: Clean and normalize
3. Chunk documents: Split into optimal sizes
4. Generate embeddings: Create vector representations
5. Index vectors: Store in vector database
6. Optimize retrieval: Tune for accuracy

┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│   Loader    │───▶│ Preprocessor │───▶│   Chunker   │
└─────────────┘    └─────────────┘    └─────────────┘
                                            │
                                            ▼
┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│  Retriever  │◀───│   Indexer   │◀───│  Embedder   │
└─────────────┘    └─────────────┘    └─────────────┘

// pipeline/loaders.ts
import { readFile, readdir } from 'fs/promises';
import { join, extname } from 'path';
import pdf from 'pdf-parse';
import mammoth from 'mammoth';

interface LoadedDocument { id: string; content: string; metadata: { source: string; type: string; title?: string; createdAt?: Date; [key: string]: any; }; }

export class DocumentLoader { async loadFile(filePath: string): Promise<LoadedDocument> { const ext = extname(filePath).toLowerCase(); const content = await this.extractContent(filePath, ext);

return { id: this.generateId(filePath), content, metadata: { source: filePath, type: ext.slice(1), }, }; }

async loadDirectory(dirPath: string): Promise<LoadedDocument[]> { const files = await readdir(dirPath, { recursive: true }); const documents: LoadedDocument[] = [];

for (const file of files) { const filePath = join(dirPath, file); try { const doc = await this.loadFile(filePath); documents.push(doc); } catch (error) { console.error(Failed to load ${filePath}:, error); } }

return documents; }

private async extractContent(filePath: string, ext: string): Promise<string> { const buffer = await readFile(filePath);

switch (ext) { case '.txt': case '.md': return buffer.toString('utf-8');

case '.pdf': const pdfData = await pdf(buffer); return pdfData.text;

case '.docx': const result = await mammoth.extractRawText({ buffer }); return result.value;

case '.json': const json = JSON.parse(buffer.toString('utf-8')); return this.flattenJson(json);

default: throw new Error(Unsupported file type: ${ext}); } }

private flattenJson(obj: any, prefix = ''): string { const parts: string[] = [];

for (const [key, value] of Object.entries(obj)) { const path = prefix ? ${prefix}.${key} : key;

if (typeof value === 'object' && value !== null) { parts.push(this.flattenJson(value, path)); } else { parts.push(${path}: ${value}); } }

return parts.join('\n'); }

private generateId(source: string): string { return doc_${Buffer.from(source).toString('base64url').slice(0, 16)}; } } ```

// pipeline/web-loader.ts
import { JSDOM } from 'jsdom';
import { Readability } from '@mozilla/readability';

export class WebLoader { async loadUrl(url: string): Promise<LoadedDocument> { const response = await fetch(url); const html = await response.text();

const dom = new JSDOM(html, { url }); const reader = new Readability(dom.window.document); const article = reader.parse();

return { id: this.generateId(url), content: article?.textContent || '', metadata: { source: url, type: 'webpage', title: article?.title, byline: article?.byline, }, }; }

async loadSitemap(sitemapUrl: string): Promise<LoadedDocument[]> { const response = await fetch(sitemapUrl); const xml = await response.text();

const urlMatches = xml.match(/<loc>([^<]+)<\/loc>/g) || []; const urls = urlMatches.map((match) => match.replace('<loc>', '').replace('</loc>', '') );

const documents: LoadedDocument[] = []; for (const url of urls.slice(0, 100)) { // Limit for safety try { const doc = await this.loadUrl(url); documents.push(doc); await new Promise((r) => setTimeout(r, 1000)); // Rate limit } catch (error) { console.error(Failed to load ${url}:, error); } }

return documents; } } ```

// pipeline/preprocessor.ts
export class TextPreprocessor {
  process(text: string): string {
    return this.pipeline(text, [
      this.normalizeWhitespace,
      this.removeSpecialCharacters,
      this.normalizeUnicode,
      this.removeExcessiveNewlines,
    ]);
  }

private pipeline(text: string, transforms: Array<(t: string) => string>): string { return transforms.reduce((t, fn) => fn(t), text); }

private normalizeWhitespace(text: string): string { return text .replace(/\t/g, ' ') .replace(/[ ]+/g, ' ') .trim(); }

private removeSpecialCharacters(text: string): string { return text .replace(/[^\w\s\n.,!?;:'"()-]/g, '') .replace(/\s+/g, ' '); }

private normalizeUnicode(text: string): string { return text.normalize('NFKC'); }

private removeExcessiveNewlines(text: string): string { return text.replace(/\n{3,}/g, '\n\n'); } } ```

// pipeline/chunker.ts
export interface Chunk {
  id: string;
  content: string;
  metadata: {
    documentId: string;
    chunkIndex: number;
    startChar: number;
    endChar: number;
    [key: string]: any;
  };
}

export interface ChunkerOptions { chunkSize: number; chunkOverlap: number; separators?: string[]; }

export class RecursiveChunker { private options: ChunkerOptions; private separators: string[];

constructor(options: ChunkerOptions) { this.options = options; this.separators = options.separators || [ '\n\n', // Paragraphs '\n', // Lines '. ', // Sentences ', ', // Clauses ' ', // Words '', // Characters ]; }

chunk(document: LoadedDocument): Chunk[] { const chunks: Chunk[] = []; const textChunks = this.splitText(document.content, 0);

textChunks.forEach((text, index) => { chunks.push({ id: ${document.id}_chunk_${index}, content: text.content, metadata: { documentId: document.id, chunkIndex: index, startChar: text.start, endChar: text.end, ...document.metadata, }, }); });

return chunks; }

private splitText( text: string, separatorIndex: number ): Array<{ content: string; start: number; end: number }> { const separator = this.separators[separatorIndex]; const { chunkSize, chunkOverlap } = this.options;

// If text fits in chunk size, return as is if (text.length <= chunkSize) { return [{ content: text, start: 0, end: text.length }]; }

// Try splitting with current separator const parts = separator ? text.split(separator) : text.split(''); const results: Array<{ content: string; start: number; end: number }> = []; let currentChunk = ''; let currentStart = 0; let position = 0;

for (const part of parts) { const partWithSep = part + (separator || '');

if ((currentChunk + partWithSep).length > chunkSize) { if (currentChunk) { // If chunk is still too large, try next separator if (currentChunk.length > chunkSize && separatorIndex < this.separators.length - 1) { const subChunks = this.splitText(currentChunk, separatorIndex + 1); results.push(...subChunks.map((c) => ({ ...c, start: c.start + currentStart, end: c.end + currentStart, }))); } else { results.push({ content: currentChunk.trim(), start: currentStart, end: position, }); }

// Start new chunk with overlap const overlapText = this.getOverlapText(currentChunk, chunkOverlap); currentChunk = overlapText + partWithSep; currentStart = position - overlapText.length; } else { currentChunk = partWithSep; } } else { currentChunk += partWithSep; }

position += partWithSep.length; }

// Don't forget the last chunk if (currentChunk.trim()) { results.push({ content: currentChunk.trim(), start: currentStart, end: position, }); }

return results; }

private getOverlapText(text: string, overlapSize: number): string { if (overlapSize === 0) return ''; return text.slice(-overlapSize); } } ```

// pipeline/semantic-chunker.ts
import { generateEmbedding } from '../embeddings';

export class SemanticChunker { private similarityThreshold: number; private minChunkSize: number; private maxChunkSize: number;

constructor(options: { similarityThreshold?: number; minChunkSize?: number; maxChunkSize?: number; } = {}) { this.similarityThreshold = options.similarityThreshold || 0.8; this.minChunkSize = options.minChunkSize || 100; this.maxChunkSize = options.maxChunkSize || 2000; }

async chunk(document: LoadedDocument): Promise<Chunk[]> { // Split into sentences const sentences = this.splitIntoSentences(document.content);

// Generate embeddings for each sentence const embeddings = await Promise.all( sentences.map((s) => generateEmbedding(s)) );

// Group similar sentences const groups = this.groupBySimilarity(sentences, embeddings);

// Create chunks from groups return groups.map((group, index) => ({ id: ${document.id}_semantic_${index}, content: group.join(' '), metadata: { documentId: document.id, chunkIndex: index, chunkType: 'semantic', ...document.metadata, }, })); }

private splitIntoSentences(text: string): string[] { return text .split(/(?<=[.!?])\s+/) .filter((s) => s.length > 10); }

private groupBySimilarity(sentences: string[], embeddings: number[][]): string[][] { const groups: string[][] = []; let currentGroup: string[] = []; let currentLength = 0;

for (let i = 0; i < sentences.length; i++) { const sentence = sentences[i];

if (currentGroup.length === 0) { currentGroup.push(sentence); currentLength = sentence.length; continue; }

// Check similarity with previous sentence const similarity = this.cosineSimilarity(embeddings[i], embeddings[i - 1]);

if ( similarity > this.similarityThreshold && currentLength + sentence.length < this.maxChunkSize ) { currentGroup.push(sentence); currentLength += sentence.length; } else { if (currentLength >= this.minChunkSize) { groups.push(currentGroup); } currentGroup = [sentence]; currentLength = sentence.length; } }

if (currentGroup.length > 0) { groups.push(currentGroup); }

return groups; }

private cosineSimilarity(a: number[], b: number[]): number { let dotProduct = 0; let normA = 0; let normB = 0;

for (let i = 0; i < a.length; i++) { dotProduct += a[i] * b[i]; normA += a[i] * a[i]; normB += b[i] * b[i]; }

return dotProduct / (Math.sqrt(normA) * Math.sqrt(normB)); } } ```

// pipeline/embedder.ts
import OpenAI from 'openai';
import pLimit from 'p-limit';

const openai = new OpenAI();

export interface EmbeddedChunk extends Chunk { embedding: number[]; }

export class Embedder { private model: string; private batchSize: number; private concurrency: number;

constructor(options: { model?: string; batchSize?: number; concurrency?: number; } = {}) { this.model = options.model || 'text-embedding-3-small'; this.batchSize = options.batchSize || 100; this.concurrency = options.concurrency || 5; }

async embedChunks(chunks: Chunk[]): Promise<EmbeddedChunk[]> { const limit = pLimit(this.concurrency); const batches = this.createBatches(chunks);

const embeddedBatches = await Promise.all( batches.map((batch) => limit(() => this.embedBatch(batch)) ) );

return embeddedBatches.flat(); }

private createBatches(chunks: Chunk[]): Chunk[][] { const batches: Chunk[][] = []; for (let i = 0; i < chunks.length; i += this.batchSize) { batches.push(chunks.slice(i, i + this.batchSize)); } return batches; }

private async embedBatch(chunks: Chunk[]): Promise<EmbeddedChunk[]> { const response = await openai.embeddings.create({ model: this.model, input: chunks.map((c) => c.content), });

return chunks.map((chunk, i) => ({ ...chunk, embedding: response.data[i].embedding, })); } } ```

// pipeline/index.ts
import { DocumentLoader } from './loaders';
import { TextPreprocessor } from './preprocessor';
import { RecursiveChunker } from './chunker';
import { Embedder } from './embedder';
import { VectorStore } from './vectorstore';

export interface PipelineOptions { chunkSize?: number; chunkOverlap?: number; embeddingModel?: string; namespace?: string; }

export class EmbeddingPipeline { private loader: DocumentLoader; private preprocessor: TextPreprocessor; private chunker: RecursiveChunker; private embedder: Embedder; private vectorStore: VectorStore;

constructor(options: PipelineOptions = {}) { this.loader = new DocumentLoader(); this.preprocessor = new TextPreprocessor(); this.chunker = new RecursiveChunker({ chunkSize: options.chunkSize || 1000, chunkOverlap: options.chunkOverlap || 200, }); this.embedder = new Embedder({ model: options.embeddingModel, }); this.vectorStore = new VectorStore(options.namespace); }

async ingestFile(filePath: string): Promise<{ chunks: number }> { // Load const document = await this.loader.loadFile(filePath);

return this.processDocument(document); }

async ingestDirectory(dirPath: string): Promise<{ documents: number; chunks: number }> { const documents = await this.loader.loadDirectory(dirPath); let totalChunks = 0;

for (const doc of documents) { const result = await this.processDocument(doc); totalChunks += result.chunks; }

return { documents: documents.length, chunks: totalChunks }; }

async ingestUrl(url: string): Promise<{ chunks: number }> { const webLoader = new WebLoader(); const document = await webLoader.loadUrl(url);

return this.processDocument(document); }

private async processDocument(document: LoadedDocument): Promise<{ chunks: number }> { // Preprocess document.content = this.preprocessor.process(document.content);

// Chunk const chunks = this.chunker.chunk(document);

// Embed const embeddedChunks = await this.embedder.embedChunks(chunks);

// Store await this.vectorStore.upsert(embeddedChunks);

return { chunks: embeddedChunks.length }; }

async query(query: string, topK = 5): Promise<Chunk[]> { return this.vectorStore.search(query, topK); } }

// Usage const pipeline = new EmbeddingPipeline({ chunkSize: 1000, chunkOverlap: 200, namespace: 'knowledge-base', });

// Ingest documents await pipeline.ingestDirectory('./documents'); await pipeline.ingestUrl('https://docs.example.com');

// Query const results = await pipeline.query('How do I configure authentication?'); ```

1. Chunk size: Balance context vs noise (500-1500 chars)
2. Overlap: 10-20% prevents context loss
3. Preprocessing: Clean but preserve meaning
4. Batch embeddings: Reduce API calls
5. Add metadata: Enable filtering
6. Semantic chunking: For high-quality retrieval
7. Hybrid search: Combine vector and keyword
8. Monitor quality: Test retrieval accuracy

Every embedding pipeline should include:

• Multi-format document loading
• Text preprocessing and cleaning
• Smart chunking with overlap
• Batch embedding generation
• Metadata preservation
• Vector store indexing
• Semantic search capability
• Error handling and logging
• Progress tracking
• Retrieval quality testing

Transform documents into high-quality vector search datasets.

1. Extract text from various formats (PDF, DOCX, HTML)
2. Clean text (remove noise, normalize)
3. Chunk strategically (semantic boundaries)
4. Add metadata (source, timestamps, classification)
5. Deduplicate (near-duplicate detection)
6. Quality check (length, content validation)
7. Export JSONL (one chunk per line)

# PDF extraction
import pymupdf

def extract_pdf(filepath: str) -> str: doc = pymupdf.open(filepath) text = "" for page in doc: text += page.get_text() return text

import re

def clean_text(text: str) -> str: # Remove extra whitespace text = re.sub(r'\s+', ' ', text)

return text.strip() ```

def semantic_chunk(text: str, max_chunk_size: int = 1000) -> List[str]:
    """Chunk at semantic boundaries (paragraphs, sentences)"""
    # Split by paragraphs first
    paragraphs = text.split('\n\n')

chunks = [] current_chunk = ""

for para in paragraphs: if len(current_chunk) + len(para) <= max_chunk_size: current_chunk += para + "\n\n" else: if current_chunk: chunks.append(current_chunk.strip()) current_chunk = para + "\n\n"

if current_chunk: chunks.append(current_chunk.strip())

return chunks ```

def extract_metadata(filepath: str, chunk: str, chunk_idx: int) -> dict:
    return {
        "source": filepath,
        "chunk_id": f"{hash(filepath)}_{chunk_idx}",
        "chunk_index": chunk_idx,
        "char_count": len(chunk),
        "word_count": len(chunk.split()),
        "created_at": datetime.now().isoformat(),

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

def deduplicate_chunks(chunks: List[dict], threshold: float = 0.95) -> List[dict]: """Remove near-duplicate chunks""" texts = [c["text"] for c in chunks]

def quality_check(chunk: dict) -> bool:
    """Validate chunk quality"""
    text = chunk["text"]

return True ```

import json

def export_jsonl(chunks: List[dict], output_path: str): """Export chunks as JSONL (one JSON object per line)""" with open(output_path, 'w') as f: for chunk in chunks: f.write(json.dumps(chunk) + '\n')

def process_documents(input_dir: str, output_path: str):
    all_chunks = []

print(f"Processed {len(unique_chunks)} chunks") ```

• Chunk at semantic boundaries
• Rich metadata for filtering
• Deduplicate aggressively
• Quality checks prevent garbage
• JSONL format for streaming
• Version your datasets

• Text extraction from all formats
• Cleaning pipeline implemented
• Semantic chunking strategy
• Metadata schema defined
• Deduplication logic
• Quality validation checks
• JSONL export format
• Dataset statistics logged

Build coordinated multi-agent systems for complex task automation.

1. Define agents: Create specialized agents
2. Design workflow: Plan agent coordination
3. Implement handoffs: Agent-to-agent communication
4. Add shared memory: Persistent context
5. Create supervisor: Orchestrate execution
6. Monitor execution: Track agent activities

// agents/base.ts
import { ChatOpenAI } from '@langchain/openai';
import { SystemMessage, HumanMessage, AIMessage } from '@langchain/core/messages';

export interface AgentConfig { name: string; role: string; systemPrompt: string; tools?: Tool[]; model?: string; }

export interface AgentResponse { content: string; toolCalls?: ToolCall[]; nextAgent?: string; completed?: boolean; }

export class Agent { private model: ChatOpenAI; private config: AgentConfig; private messageHistory: BaseMessage[] = [];

constructor(config: AgentConfig) { this.config = config; this.model = new ChatOpenAI({ modelName: config.model || 'gpt-4-turbo-preview', temperature: 0.7, }); }

async execute(input: string, context?: Record<string, any>): Promise<AgentResponse> { const systemMessage = new SystemMessage( this.buildSystemPrompt(context) );

const messages = [ systemMessage, ...this.messageHistory, new HumanMessage(input), ];

const response = await this.model.invoke(messages, { tools: this.config.tools, });

this.messageHistory.push(new HumanMessage(input)); this.messageHistory.push(new AIMessage(response.content as string));

return this.parseResponse(response); }

private buildSystemPrompt(context?: Record<string, any>): string { let prompt = this.config.systemPrompt;

if (context) { prompt += \n\nContext:\n${JSON.stringify(context, null, 2)}; }

return prompt; }

private parseResponse(response: any): AgentResponse { // Parse tool calls and determine next actions return { content: response.content as string, toolCalls: response.tool_calls, completed: response.content?.includes('[TASK_COMPLETE]'), }; }

clearHistory() { this.messageHistory = []; } } ```

// agents/specialists.ts
import { Agent, AgentConfig } from './base';

• export const ResearchAgent = new Agent({
• name: 'researcher',
• role: 'Research Specialist',
• systemPrompt: `You are a research specialist. Your job is to:
• Search for and gather relevant information
• Analyze sources and extract key insights
• Summarize findings clearly
• Cite sources when possible

When you have gathered sufficient information, include [TASK_COMPLETE] in your response. If you need help from another agent, specify: [HANDOFF:agent_name]`, tools: [searchTool, webScrapeTool], });

• export const WriterAgent = new Agent({
• name: 'writer',
• role: 'Content Writer',
• systemPrompt: `You are a professional content writer. Your job is to:
• Create engaging, well-structured content
• Adapt tone and style to the target audience
• Incorporate research and data effectively
• Edit and refine for clarity

Use the research provided to create compelling content. When complete, include [TASK_COMPLETE].`, });

• export const ReviewerAgent = new Agent({
• name: 'reviewer',
• role: 'Quality Reviewer',
• systemPrompt: `You are a quality reviewer. Your job is to:
• Review content for accuracy and clarity
• Check for errors and inconsistencies
• Suggest improvements
• Approve or request revisions

Provide specific feedback. If approved, include [APPROVED]. If revisions needed, include [REVISIONS_NEEDED] with specific changes.`, });

• export const PlannerAgent = new Agent({
• name: 'planner',
• role: 'Task Planner',
• systemPrompt: `You are a task planner. Your job is to:
• Break down complex tasks into subtasks
• Identify which specialist agent should handle each subtask
• Create an execution order
• Track progress

Output a structured plan in JSON format: { "goal": "...", "steps": [ { "step": 1, "agent": "researcher", "task": "..." }, { "step": 2, "agent": "writer", "task": "..." } ] }, });

// orchestrator/sequential.ts
import { Agent } from '../agents/base';

interface WorkflowStep { agent: Agent; task: string; inputFrom?: string; }

export class SequentialOrchestrator { private agents: Map<string, Agent> = new Map(); private results: Map<string, string> = new Map();

registerAgent(name: string, agent: Agent) { this.agents.set(name, agent); }

async execute(workflow: WorkflowStep[]): Promise<Record<string, string>> { for (const step of workflow) { const agent = step.agent;

// Get input from previous step if specified let input = step.task; if (step.inputFrom && this.results.has(step.inputFrom)) { input = ${step.task}\n\nPrevious output:\n${this.results.get(step.inputFrom)}; }

console.log(Executing: ${agent.name} - ${step.task});

const result = await agent.execute(input); this.results.set(agent.name, result.content);

console.log(Completed: ${agent.name}); }

return Object.fromEntries(this.results); } }

// Usage const orchestrator = new SequentialOrchestrator(); orchestrator.registerAgent('researcher', ResearchAgent); orchestrator.registerAgent('writer', WriterAgent); orchestrator.registerAgent('reviewer', ReviewerAgent);

const results = await orchestrator.execute([ { agent: ResearchAgent, task: 'Research the latest AI trends in 2024' }, { agent: WriterAgent, task: 'Write a blog post about AI trends', inputFrom: 'researcher' }, { agent: ReviewerAgent, task: 'Review the blog post', inputFrom: 'writer' }, ]); ```

// orchestrator/supervisor.ts
import { ChatOpenAI } from '@langchain/openai';
import { Agent } from '../agents/base';

interface AgentRegistry { [name: string]: { agent: Agent; description: string; }; }

export class SupervisorOrchestrator { private supervisor: ChatOpenAI; private agents: AgentRegistry = {}; private sharedContext: Record<string, any> = {}; private maxIterations = 10;

constructor() { this.supervisor = new ChatOpenAI({ modelName: 'gpt-4-turbo-preview', temperature: 0, }); }

registerAgent(name: string, agent: Agent, description: string) { this.agents[name] = { agent, description }; }

async execute(task: string): Promise<string> { let iteration = 0; let currentTask = task; const history: string[] = [];

while (iteration < this.maxIterations) { iteration++;

// Supervisor decides next action const decision = await this.supervise(currentTask, history);

if (decision.complete) { return decision.finalResponse!; }

// Execute selected agent const { agent } = this.agents[decision.nextAgent!]; const result = await agent.execute(decision.agentTask!, this.sharedContext);

// Update shared context this.sharedContext[decision.nextAgent!] = result.content; history.push(${decision.nextAgent}: ${result.content});

// Check for handoff if (result.nextAgent) { currentTask = Continue with: ${result.content}; } }

throw new Error('Max iterations reached'); }

private async supervise( task: string, history: string[] ): Promise<{ complete: boolean; finalResponse?: string; nextAgent?: string; agentTask?: string; }> { const agentList = Object.entries(this.agents) .map(([name, { description }]) => - ${name}: ${description}) .join('\n');

const prompt = `You are a supervisor coordinating AI agents.

Available agents: ${agentList}

Task: ${task}

History: ${history.join('\n')}

Decide the next action. Respond in JSON: { "thought": "your reasoning", "complete": false, "nextAgent": "agent_name", "agentTask": "specific task for the agent" }

Or if the task is complete: { "thought": "your reasoning", "complete": true, "finalResponse": "the final answer" }`;

const response = await this.supervisor.invoke([{ role: 'user', content: prompt }]); return JSON.parse(response.content as string); } } ```

// orchestrator/parallel.ts
export class ParallelOrchestrator {
  private agents: Map<string, Agent> = new Map();

async executeParallel( tasks: Array<{ agentName: string; task: string }> ): Promise<Map<string, string>> { const results = new Map<string, string>();

await Promise.all( tasks.map(async ({ agentName, task }) => { const agent = this.agents.get(agentName); if (!agent) throw new Error(Agent ${agentName} not found);

const result = await agent.execute(task); results.set(agentName, result.content); }) );

return results; }

async fanOutFanIn( task: string, agentNames: string[], aggregator: Agent ): Promise<string> { // Fan out: same task to multiple agents const parallelResults = await this.executeParallel( agentNames.map((name) => ({ agentName: name, task })) );

// Fan in: aggregate results const aggregatedInput = Array.from(parallelResults.entries()) .map(([name, result]) => ${name}:\n${result}) .join('\n\n---\n\n');

const finalResult = await aggregator.execute( Synthesize these perspectives:\n\n${aggregatedInput} );

return finalResult.content; } } ```

// memory/shared.ts
import { Redis } from 'ioredis';

export class SharedMemory { private redis: Redis; private prefix: string;

constructor(sessionId: string) { this.redis = new Redis(process.env.REDIS_URL!); this.prefix = agent:${sessionId}:; }

async set(key: string, value: any, ttl?: number): Promise<void> { const serialized = JSON.stringify(value); if (ttl) { await this.redis.setex(this.prefix + key, ttl, serialized); } else { await this.redis.set(this.prefix + key, serialized); } }

async get<T>(key: string): Promise<T | null> { const value = await this.redis.get(this.prefix + key); return value ? JSON.parse(value) : null; }

async append(key: string, item: any): Promise<void> { const list = (await this.get<any[]>(key)) || []; list.push(item); await this.set(key, list); }

async getConversation(): Promise<Message[]> { return (await this.get<Message[]>('conversation')) || []; }

async addMessage(message: Message): Promise<void> { await this.append('conversation', message); }

async getAgentOutputs(): Promise<Record<string, string>> { return (await this.get<Record<string, string>>('outputs')) || {}; }

async setAgentOutput(agent: string, output: string): Promise<void> { const outputs = await this.getAgentOutputs(); outputs[agent] = output; await this.set('outputs', outputs); }

async clear(): Promise<void> { const keys = await this.redis.keys(this.prefix + '*'); if (keys.length > 0) { await this.redis.del(...keys); } } } ```

// events/agent-events.ts
import { EventEmitter } from 'events';

export class AgentEventBus extends EventEmitter { private static instance: AgentEventBus;

static getInstance(): AgentEventBus { if (!this.instance) { this.instance = new AgentEventBus(); } return this.instance; }

emitAgentMessage(from: string, to: string, message: any) { this.emit(message:${to}, { from, message, timestamp: new Date() }); }

emitAgentComplete(agent: string, result: any) { this.emit('agent:complete', { agent, result, timestamp: new Date() }); }

emitAgentError(agent: string, error: Error) { this.emit('agent:error', { agent, error, timestamp: new Date() }); }

onMessage(agentName: string, handler: (message: any) => void) { this.on(message:${agentName}, handler); }

onAnyComplete(handler: (event: any) => void) { this.on('agent:complete', handler); } }

// Usage const eventBus = AgentEventBus.getInstance();

// Agent listens for messages eventBus.onMessage('writer', async ({ from, message }) => { console.log(Writer received from ${from}:, message); const result = await WriterAgent.execute(message); eventBus.emitAgentComplete('writer', result); });

// Orchestrator listens for completions eventBus.onAnyComplete(({ agent, result }) => { console.log(${agent} completed:, result.content); }); ```

// workflows/definition.ts
interface WorkflowDefinition {
  name: string;
  description: string;
  agents: string[];
  steps: WorkflowStep[];
  errorHandling: 'retry' | 'fallback' | 'abort';
}

const ContentCreationWorkflow: WorkflowDefinition = { name: 'content-creation', description: 'Create and publish content', agents: ['planner', 'researcher', 'writer', 'reviewer'], steps: [ { id: 'plan', agent: 'planner', input: '{task}', outputKey: 'plan', }, { id: 'research', agent: 'researcher', input: 'Research for: {plan.topic}', outputKey: 'research', parallel: true, }, { id: 'write', agent: 'writer', input: 'Write about {plan.topic} using research: {research}', outputKey: 'draft', dependsOn: ['research'], }, { id: 'review', agent: 'reviewer', input: 'Review: {draft}', outputKey: 'review', loop: { condition: 'review.approved === false', maxIterations: 3, backTo: 'write', }, }, ], errorHandling: 'retry', }; ```

1. Single responsibility: Each agent has one clear role
2. Clear handoffs: Explicit agent-to-agent communication
3. Shared context: Use memory for persistent state
4. Iteration limits: Prevent infinite loops
5. Error handling: Graceful degradation
6. Observability: Log all agent actions
7. Testing: Test agents individually and together
8. Timeout handling: Prevent stuck agents

Every agent system should include:

• Agent base class with common functionality
• Specialized agents with clear roles
• Orchestrator for coordination
• Shared memory system
• Event-based communication
• Workflow definitions
• Error handling and retries
• Iteration limits
• Logging and monitoring
• Agent handoff protocol

Design robust multi-step agent systems with tools and error handling.

User Query → Planning → Tool Selection → Tool Execution → Result Synthesis → Response
              ↓            ↓                ↓                    ↓
           Memory      Retry Logic      Validation         Cost Tracking

from typing import List, Dict, Any

class Agent: def __init__(self, tools: List[Tool], max_iterations: int = 5): self.tools = tools self.max_iterations = max_iterations self.memory = [] self.cost_tracker = CostTracker()

def run(self, query: str) -> str: self.memory.append({"role": "user", "content": query})

for iteration in range(self.max_iterations): # Decide next action action = self.plan_next_action()

if action["type"] == "final_answer": return action["content"]

return "Max iterations reached"

def plan_next_action(self) -> Dict: prompt = self.build_planning_prompt() response = llm(prompt) return parse_action(response) ```

TOOL_ORDER = {
    "search_web": 1,        # Always try search first
    "query_database": 2,    # Then database
    "call_api": 3,          # Then external APIs
    "generate_content": 4,  # Finally generate
}

def select_tools(query: str, available_tools: List[Tool]) -> List[Tool]: """Select and order tools based on query""" # Use LLM to select relevant tools tool_selection_prompt = f""" Given this query: "{query}"

Which of these tools are needed? {[t.name for t in available_tools]}

Return JSON array of tool names in execution order. """

selected_names = json.loads(llm(tool_selection_prompt)) selected_tools = [t for t in available_tools if t.name in selected_names]

return selected_tools ```

def execute_with_retry(tool: Tool, params: Dict, max_retries: int = 3):
    """Execute tool with exponential backoff retry"""
    for attempt in range(max_retries):
        try:
            result = tool.execute(params)
            return {"success": True, "data": result}
        except ToolError as e:
            if attempt == max_retries - 1:
                # Try fallback tool
                fallback = get_fallback_tool(tool.name)
                if fallback:
                    return execute_with_retry(fallback, params, 1)

return {"success": False, "error": str(e)}

FALLBACK_TOOLS = { "search_web": "query_database", "call_api": "use_cached_data", } ```

class AgentState:
    def __init__(self):
        self.memory = []
        self.tool_results = {}
        self.costs = 0.0
        self.iteration = 0

def add_message(self, role: str, content: str): self.memory.append({"role": role, "content": content})

def add_tool_result(self, tool_name: str, result: Any): self.tool_results[tool_name] = result

def get_context(self) -> str: """Build context from memory for next LLM call""" return "\n".join([ f"{msg['role']}: {msg['content']}" for msg in self.memory[-5:] # Last 5 messages ]) ```

class CostTracker:
    def __init__(self, max_cost: float = 1.0):
        self.max_cost = max_cost
        self.total_cost = 0.0
        self.breakdown = {}

def add(self, cost: float, category: str = "llm"): self.total_cost += cost self.breakdown[category] = self.breakdown.get(category, 0) + cost

def exceeds_limit(self) -> bool: return self.total_cost >= self.max_cost

def remaining(self) -> float: return self.max_cost - self.total_cost

graph TD
    A[User Query] --> B[Plan Action]
    B --> C{Action Type?}
    C -->|Tool Call| D[Execute Tool]
    C -->|Final Answer| E[Return Response]
    D --> F[Validate Result]
    F -->|Success| G[Update Memory]
    F -->|Failure| H[Retry/Fallback]
    H --> D
    G --> I{Budget OK?}
    I -->|Yes| B
    I -->|No| J[Budget Exceeded]
    J --> E

def build_planning_prompt(state: AgentState) -> str:
    return f"""
You are an agent that can use tools to answer questions.

Available tools: {json.dumps([t.schema for t in tools], indent=2)}

Conversation history: {state.get_context()}

1. Based on the conversation, decide your next action:
2. Call a tool (specify tool name and parameters)
3. Provide final answer

If calling a tool, respond with: {{"action": "tool_call", "tool": "tool_name", "params": {{...}}}}

If providing final answer, respond with: {{"action": "final_answer", "content": "your answer"}}

Think step by step about what information you need. """ ```

class MultiAgentSystem:
    def __init__(self):
        self.agents = {
            "researcher": ResearchAgent(),
            "coder": CodeAgent(),
            "reviewer": ReviewAgent(),
        }

def run(self, task: str): # Researcher gathers information context = self.agents["researcher"].run(task)

if review["approved"]: return code else: # Iterate with feedback return self.agents["coder"].run( f"Fix this code based on feedback:\n{review['feedback']}" ) ```

1. Limit iterations: Prevent infinite loops
2. Budget controls: Track and limit costs
3. Tool validation: Verify tool outputs
4. Error handling: Graceful fallbacks
5. State persistence: Save progress
6. Observability: Log all actions
7. Human-in-loop: Critical decisions

• Agent loop implementation
• Tool selection logic
• Retry & fallback strategies
• State management
• Cost tracking
• Budget limits
• Orchestration diagram
• Planning prompts
• Error handling
• Observability/logging

Design robust tool schemas that LLMs can reliably invoke.

// OpenAI function calling format
const searchDocsTool = {
  type: "function",
  function: {
    name: "search_documentation",
    description:
      "Search through product documentation using semantic search. Use this when the user asks about features, how-tos, or troubleshooting.",
    parameters: {
      type: "object",
      properties: {
        query: {
          type: "string",
          description: "The search query, phrased as a question or keywords",
        },
        filters: {
          type: "object",
          properties: {
            category: {
              type: "string",
              enum: ["api", "guides", "tutorials", "troubleshooting"],
              description: "Filter by documentation category",
            },
            version: {
              type: "string",
              description: "Filter by product version (e.g., 'v2.0')",
            },
          },
        },
        max_results: {
          type: "integer",
          minimum: 1,
          maximum: 10,
          default: 5,
          description: "Maximum number of results to return",
        },
      },
      required: ["query"],
    },
  },
};

// TypeScript types matching schema
interface SearchDocsParams {
  query: string;
  filters?: {
    category?: "api" | "guides" | "tutorials" | "troubleshooting";
    version?: string;
  };
  max_results?: number;
}

// Implementation async function search_documentation( params: SearchDocsParams ): Promise<SearchResult[]> { const { query, filters = {}, max_results = 5 } = params;

// Implementation return await vectorStore.search(query, { filter: filters, limit: max_results, }); } ```

import { z } from "zod";

// Zod schema for runtime validation const searchDocsSchema = z.object({ query: z.string().min(1, "Query cannot be empty"), filters: z .object({ category: z .enum(["api", "guides", "tutorials", "troubleshooting"]) .optional(), version: z.string().optional(), }) .optional(), max_results: z.number().int().min(1).max(10).default(5), });

// Validate before execution function validateAndExecute(toolName: string, params: unknown) { const validated = searchDocsSchema.parse(params); return search_documentation(validated); } ```

export const TOOLS = {
  search_documentation: {
    schema: searchDocsTool,
    implementation: search_documentation,
    validator: searchDocsSchema,
  },
  create_ticket: {
    schema: createTicketTool,
    implementation: create_ticket,
    validator: createTicketSchema,
  },
  // ... more tools
};

// Execute tool safely async function executeTool(name: string, params: unknown) { const tool = TOOLS[name]; if (!tool) throw new Error(Unknown tool: ${name});

const validated = tool.validator.parse(params); return tool.implementation(validated); } ```

// Example 1: Simple search
{
  "name": "search_documentation",
  "parameters": {
    "query": "How do I authenticate API requests?"
  }
}

// Example 2: With filters { "name": "search_documentation", "parameters": { "query": "rate limiting", "filters": { "category": "api", "version": "v2.0" }, "max_results": 3 } } ```

interface ToolResult {
  success: boolean;
  data?: any;
  error?: {
    code: string;
    message: string;
  };
}

async function safeExecuteTool( name: string, params: unknown ): Promise<ToolResult> { try { const data = await executeTool(name, params); return { success: true, data }; } catch (error) { if (error instanceof z.ZodError) { return { success: false, error: { code: "VALIDATION_ERROR", message: Invalid parameters: ${error.message}, }, }; } return { success: false, error: { code: "EXECUTION_ERROR", message: error.message, }, }; } } ```

1. Clear descriptions: Explain when to use the tool
2. Specific types: Use enums, ranges, patterns
3. Sensible defaults: Reduce required parameters
4. Validate rigorously: Don't trust LLM output
5. Error messages: Help LLM correct mistakes
6. Example calls: Show success cases
7. Type safety: TypeScript interfaces

• JSON schema defined
• TypeScript interface
• Validation with Zod
• Implementation function
• Error handling
• Example calls (3+)
• Tool registry entry
• Documentation

Extract reliable, typed data from LLM responses.

1. Define schema: Create data structure
2. Choose method: Function calling vs prompting
3. Generate response: Call LLM with structure
4. Validate output: Parse and verify
5. Handle errors: Retry or fallback

// extractors/json-mode.ts
import OpenAI from 'openai';

const openai = new OpenAI();

interface ExtractedData { name: string; email: string; phone?: string; company?: string; }

export async function extractContactInfo(text: string): Promise<ExtractedData> { const response = await openai.chat.completions.create({ model: 'gpt-4-turbo-preview', response_format: { type: 'json_object' }, messages: [ { role: 'system', content: Extract contact information from text. Return JSON with: { "name": "string", "email": "string", "phone": "string or null", "company": "string or null" }, }, { role: 'user', content: text }, ], });

return JSON.parse(response.choices[0].message.content!); } ```

// extractors/structured.ts
import OpenAI from 'openai';
import { z } from 'zod';
import { zodResponseFormat } from 'openai/helpers/zod';

const ContactSchema = z.object({ name: z.string().describe('Full name of the contact'), email: z.string().email().describe('Email address'), phone: z.string().nullable().describe('Phone number if available'), company: z.string().nullable().describe('Company name if mentioned'), role: z.string().nullable().describe('Job title or role'), });

type Contact = z.infer<typeof ContactSchema>;

export async function extractContact(text: string): Promise<Contact> { const response = await openai.beta.chat.completions.parse({ model: 'gpt-4o-2024-08-06', messages: [ { role: 'system', content: 'Extract contact information from the provided text.', }, { role: 'user', content: text }, ], response_format: zodResponseFormat(ContactSchema, 'contact'), });

return response.choices[0].message.parsed!; } ```

// extractors/function-calling.ts
import OpenAI from 'openai';

const openai = new OpenAI();

const functions = [ { name: 'extract_entities', description: 'Extract named entities from text', parameters: { type: 'object', properties: { people: { type: 'array', items: { type: 'object', properties: { name: { type: 'string' }, role: { type: 'string' }, organization: { type: 'string' }, }, required: ['name'], }, }, organizations: { type: 'array', items: { type: 'object', properties: { name: { type: 'string' }, type: { type: 'string', enum: ['company', 'nonprofit', 'government', 'other'] }, }, required: ['name'], }, }, locations: { type: 'array', items: { type: 'object', properties: { name: { type: 'string' }, type: { type: 'string', enum: ['city', 'country', 'address', 'other'] }, }, required: ['name'], }, }, dates: { type: 'array', items: { type: 'object', properties: { text: { type: 'string' }, normalized: { type: 'string', format: 'date' }, }, required: ['text'], }, }, }, required: ['people', 'organizations', 'locations', 'dates'], }, }, ];

export async function extractEntities(text: string) { const response = await openai.chat.completions.create({ model: 'gpt-4-turbo-preview', messages: [ { role: 'system', content: 'Extract all named entities from the provided text.', }, { role: 'user', content: text }, ], functions, function_call: { name: 'extract_entities' }, });

const functionCall = response.choices[0].message.function_call; return JSON.parse(functionCall!.arguments); } ```

// extractors/tools.ts
import OpenAI from 'openai';

const tools: OpenAI.Chat.ChatCompletionTool[] = [ { type: 'function', function: { name: 'classify_intent', description: 'Classify the user intent from their message', parameters: { type: 'object', properties: { intent: { type: 'string', enum: ['question', 'complaint', 'feedback', 'request', 'other'], }, confidence: { type: 'number', minimum: 0, maximum: 1, }, entities: { type: 'object', properties: { product: { type: 'string' }, issue: { type: 'string' }, sentiment: { type: 'string', enum: ['positive', 'negative', 'neutral'] }, }, }, suggestedAction: { type: 'string', }, }, required: ['intent', 'confidence'], }, }, }, ];

export async function classifyMessage(message: string) { const response = await openai.chat.completions.create({ model: 'gpt-4-turbo-preview', messages: [{ role: 'user', content: message }], tools, tool_choice: { type: 'function', function: { name: 'classify_intent' } }, });

const toolCall = response.choices[0].message.tool_calls?.[0]; return JSON.parse(toolCall!.function.arguments); } ```

// extractors/zod-extractor.ts
import { z } from 'zod';
import OpenAI from 'openai';

const openai = new OpenAI();

// Define schema const ProductReviewSchema = z.object({ productName: z.string(), rating: z.number().min(1).max(5), pros: z.array(z.string()), cons: z.array(z.string()), summary: z.string().max(200), wouldRecommend: z.boolean(), sentiment: z.enum(['positive', 'negative', 'mixed', 'neutral']), });

type ProductReview = z.infer<typeof ProductReviewSchema>;

// Generate JSON schema from Zod function zodToJsonSchema(schema: z.ZodObject<any>) { // Simplified - use zod-to-json-schema in production const shape = schema.shape; const properties: Record<string, any> = {}; const required: string[] = [];

for (const [key, value] of Object.entries(shape)) { const zodType = value as z.ZodTypeAny; properties[key] = zodTypeToJson(zodType); if (!zodType.isOptional()) { required.push(key); } }

return { type: 'object', properties, required }; }

export async function extractReview(reviewText: string): Promise<ProductReview> { const schema = zodToJsonSchema(ProductReviewSchema);

const response = await openai.chat.completions.create({ model: 'gpt-4-turbo-preview', response_format: { type: 'json_object' }, messages: [ { role: 'system', content: Extract a structured product review. Return JSON matching this schema: ${JSON.stringify(schema, null, 2)}, }, { role: 'user', content: reviewText }, ], });

const data = JSON.parse(response.choices[0].message.content!);

// Validate with Zod return ProductReviewSchema.parse(data); } ```

// extractors/retry.ts
export async function extractWithRetry<T>(
  schema: z.ZodSchema<T>,
  prompt: string,
  text: string,
  maxRetries = 3
): Promise<T> {
  let lastError: Error | null = null;

for (let attempt = 0; attempt < maxRetries; attempt++) { try { const response = await openai.chat.completions.create({ model: 'gpt-4-turbo-preview', response_format: { type: 'json_object' }, messages: [ { role: 'system', content: prompt }, { role: 'user', content: text }, // Add error context on retry ...(lastError ? [ { role: 'user' as const, content: Previous attempt failed validation: ${lastError.message}. Please fix., }, ] : []), ], });

const data = JSON.parse(response.choices[0].message.content!); return schema.parse(data); } catch (error) { if (error instanceof z.ZodError) { lastError = new Error( error.errors.map((e) => ${e.path.join('.')}: ${e.message}).join(', ') ); } else { throw error; } } }

throw new Error(Failed after ${maxRetries} attempts: ${lastError?.message}); } ```

// extractors/hierarchical.ts
const DocumentSchema = z.object({
  title: z.string(),
  authors: z.array(
    z.object({
      name: z.string(),
      affiliation: z.string().optional(),
    })
  ),
  abstract: z.string(),
  sections: z.array(
    z.object({
      heading: z.string(),
      content: z.string(),
      subsections: z
        .array(
          z.object({
            heading: z.string(),
            content: z.string(),
          })
        )
        .optional(),
    })
  ),
  references: z.array(
    z.object({
      authors: z.array(z.string()),
      title: z.string(),
      year: z.number(),
      source: z.string().optional(),
    })
  ),
  keywords: z.array(z.string()),
});

// extractors/multistep.ts
export async function extractComplex(document: string) {
  // Step 1: Extract structure
  const structure = await extract(
    z.object({
      sections: z.array(z.string()),
      hasReferences: z.boolean(),
    }),
    'Identify the document structure',
    document
  );

// Step 2: Extract each section const sections = await Promise.all( structure.sections.map((section) => extract( z.object({ heading: z.string(), summary: z.string(), keyPoints: z.array(z.string()), }), Extract details from section: ${section}, document ) ) );

// Step 3: Extract references if present let references = []; if (structure.hasReferences) { references = await extract( z.array( z.object({ authors: z.array(z.string()), title: z.string(), year: z.number(), }) ), 'Extract all references', document ); }

return { sections, references }; } ```

// extractors/streaming.ts
import { zodToJsonSchema } from 'zod-to-json-schema';

export async function* streamExtract<T>( schema: z.ZodSchema<T>, prompt: string, text: string ): AsyncGenerator<{ partial: any; complete: boolean }> { const jsonSchema = zodToJsonSchema(schema);

const stream = await openai.chat.completions.create({ model: 'gpt-4-turbo-preview', response_format: { type: 'json_object' }, messages: [ { role: 'system', content: ${prompt}\n\nReturn JSON matching: ${JSON.stringify(jsonSchema)}, }, { role: 'user', content: text }, ], stream: true, });

let fullContent = '';

for await (const chunk of stream) { const content = chunk.choices[0]?.delta?.content || ''; fullContent += content;

// Try to parse partial JSON try { const partial = JSON.parse(fullContent); yield { partial, complete: false }; } catch { // JSON not complete yet } }

// Final parse and validate const final = JSON.parse(fullContent); const validated = schema.parse(final); yield { partial: validated, complete: true }; } ```

// extractors/error-handling.ts
export class ExtractionError extends Error {
  constructor(
    message: string,
    public readonly raw: string,
    public readonly validationErrors?: z.ZodError
  ) {
    super(message);
    this.name = 'ExtractionError';
  }
}

export async function safeExtract<T>( schema: z.ZodSchema<T>, prompt: string, text: string ): Promise<{ success: true; data: T } | { success: false; error: ExtractionError }> { try { const data = await extractWithRetry(schema, prompt, text); return { success: true, data }; } catch (error) { if (error instanceof z.ZodError) { return { success: false, error: new ExtractionError('Validation failed', text, error), }; } return { success: false, error: new ExtractionError( error instanceof Error ? error.message : 'Unknown error', text ), }; } } ```

1. Use structured outputs: OpenAI's native feature when available
2. Define clear schemas: Descriptive field names and descriptions
3. Validate with Zod: Type-safe runtime validation
4. Retry on failure: With error context
5. Use function calling: For complex schemas
6. Handle edge cases: Null values, empty arrays
7. Stream for UX: Partial results during extraction
8. Test with variations: Different input formats

Every structured extraction should include:

• Zod schema with descriptions
• Appropriate extraction method (JSON/function/tools)
• Runtime validation
• Retry logic with error context
• Error handling and typing
• Streaming support (if needed)
• Edge case handling
• Type-safe return values
• Comprehensive tests

Create Model Context Protocol servers to extend Claude's capabilities with custom tools and resources.

1. Define purpose: Identify what capabilities to add
2. Choose transport: stdio (local) or HTTP/SSE (remote)
3. Design tools: Define tool schemas and handlers
4. Add resources: Optional file/data access
5. Create prompts: Optional reusable prompts
6. Test locally: Verify with MCP inspector
7. Deploy: Configure for Claude Desktop or API

┌─────────────┐     MCP Protocol      ┌─────────────┐
│   Claude    │◄────────────────────►│ MCP Server  │
│   (Host)    │  JSON-RPC over stdio  │  (Your App) │
└─────────────┘                       └─────────────┘
                                            │
                                            ▼
                                      ┌───────────┐
                                      │  Tools    │
                                      │ Resources │
                                      │  Prompts  │
                                      └───────────┘

# Create project
mkdir my-mcp-server && cd my-mcp-server
npm init -y

// package.json
{
  "name": "my-mcp-server",
  "version": "1.0.0",
  "type": "module",
  "main": "dist/index.js",
  "bin": {
    "my-mcp-server": "./dist/index.js"
  },
  "scripts": {
    "build": "tsc",
    "dev": "tsx watch src/index.ts",
    "start": "node dist/index.js"
  }
}

// tsconfig.json
{
  "compilerOptions": {
    "target": "ES2022",
    "module": "NodeNext",
    "moduleResolution": "NodeNext",
    "outDir": "./dist",
    "rootDir": "./src",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "declaration": true
  },
  "include": ["src/**/*"]
}

// src/index.ts
#!/usr/bin/env node

import { Server } from "@modelcontextprotocol/sdk/server/index.js"; import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js"; import { CallToolRequestSchema, ListToolsRequestSchema, ListResourcesRequestSchema, ReadResourceRequestSchema, ListPromptsRequestSchema, GetPromptRequestSchema, } from "@modelcontextprotocol/sdk/types.js";

// Create server instance const server = new Server( { name: "my-mcp-server", version: "1.0.0", }, { capabilities: { tools: {}, resources: {}, prompts: {}, }, } );

// Define tools const TOOLS = [ { name: "get_weather", description: "Get current weather for a location", inputSchema: { type: "object" as const, properties: { location: { type: "string", description: "City name or coordinates", }, units: { type: "string", enum: ["celsius", "fahrenheit"], default: "celsius", }, }, required: ["location"], }, }, { name: "search_database", description: "Search the internal database", inputSchema: { type: "object" as const, properties: { query: { type: "string", description: "Search query", }, limit: { type: "number", default: 10, }, }, required: ["query"], }, }, ];

// List tools handler server.setRequestHandler(ListToolsRequestSchema, async () => { return { tools: TOOLS }; });

// Call tool handler server.setRequestHandler(CallToolRequestSchema, async (request) => { const { name, arguments: args } = request.params;

switch (name) { case "get_weather": { const { location, units = "celsius" } = args as { location: string; units?: string; };

// Implement your logic here const weather = await fetchWeather(location, units);

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

case "search_database": { const { query, limit = 10 } = args as { query: string; limit?: number };

const results = await searchDatabase(query, limit);

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

default: throw new Error(Unknown tool: ${name}); } });

// Start server async function main() { const transport = new StdioServerTransport(); await server.connect(transport); console.error("MCP Server running on stdio"); }

main().catch(console.error);

// Helper functions (implement your logic) async function fetchWeather(location: string, units: string) { // Your implementation return { location, temperature: 22, units, condition: "sunny" }; }

async function searchDatabase(query: string, limit: number) { // Your implementation return { query, results: [], total: 0 }; } ```

// src/tools/index.ts
import { z } from "zod";

// Define input schemas with Zod for validation export const GetWeatherSchema = z.object({ location: z.string().describe("City name or coordinates"), units: z.enum(["celsius", "fahrenheit"]).default("celsius"), });

export const SearchSchema = z.object({ query: z.string().min(1).describe("Search query"), filters: z .object({ category: z.string().optional(), dateFrom: z.string().optional(), dateTo: z.string().optional(), }) .optional(), limit: z.number().min(1).max(100).default(10), });

// Convert Zod schema to JSON Schema for MCP export function zodToJsonSchema(schema: z.ZodObject<any>) { // Use zod-to-json-schema package in production return schema; } ```

// src/tools/handlers.ts
import { z } from "zod";

type ToolHandler<T extends z.ZodSchema> = ( args: z.infer<T> ) => Promise<{ content: Array<{ type: string; text: string }> }>;

export function createToolHandler<T extends z.ZodSchema>( schema: T, handler: (args: z.infer<T>) => Promise<any> ): ToolHandler<T> { return async (rawArgs) => { // Validate input const args = schema.parse(rawArgs);

// Execute handler const result = await handler(args);

// Format response return { content: [ { type: "text", text: typeof result === "string" ? result : JSON.stringify(result, null, 2), }, ], }; }; }

// Usage export const handleGetWeather = createToolHandler( GetWeatherSchema, async ({ location, units }) => { const response = await fetch( https://api.weather.com/v1/current?location=${location}&units=${units} ); return response.json(); } ); ```

// src/resources/index.ts
const RESOURCES = [
  {
    uri: "config://app-settings",
    name: "Application Settings",
    description: "Current application configuration",
    mimeType: "application/json",
  },
  {
    uri: "file://docs/readme",
    name: "Documentation",
    description: "Project documentation",
    mimeType: "text/markdown",
  },
];

// List resources handler server.setRequestHandler(ListResourcesRequestSchema, async () => { return { resources: RESOURCES }; });

// Read resource handler server.setRequestHandler(ReadResourceRequestSchema, async (request) => { const { uri } = request.params;

switch (uri) { case "config://app-settings": return { contents: [ { uri, mimeType: "application/json", text: JSON.stringify(getAppSettings(), null, 2), }, ], };

case "file://docs/readme": const content = await fs.readFile("./README.md", "utf-8"); return { contents: [ { uri, mimeType: "text/markdown", text: content, }, ], };

default: throw new Error(Unknown resource: ${uri}); } }); ```

// Resources with templates (e.g., database records)
const RESOURCE_TEMPLATES = [
  {
    uriTemplate: "db://users/{userId}",
    name: "User Profile",
    description: "Get user by ID",
    mimeType: "application/json",
  },
];

server.setRequestHandler(ListResourceTemplatesRequestSchema, async () => { return { resourceTemplates: RESOURCE_TEMPLATES }; });

server.setRequestHandler(ReadResourceRequestSchema, async (request) => { const { uri } = request.params;

// Parse URI to extract parameters const userMatch = uri.match(/^db:\/\/users\/(\w+)$/); if (userMatch) { const userId = userMatch[1]; const user = await db.users.findById(userId);

return { contents: [ { uri, mimeType: "application/json", text: JSON.stringify(user, null, 2), }, ], }; }

throw new Error(Unknown resource: ${uri}); }); ```

// src/prompts/index.ts
const PROMPTS = [
  {
    name: "code_review",
    description: "Review code for best practices and issues",
    arguments: [
      {
        name: "language",
        description: "Programming language",
        required: true,
      },
      {
        name: "focus",
        description: "What to focus on (security, performance, style)",
        required: false,
      },
    ],
  },
  {
    name: "explain_error",
    description: "Explain an error message and suggest fixes",
    arguments: [
      {
        name: "error",
        description: "The error message",
        required: true,
      },
      {
        name: "context",
        description: "Additional context about what you were doing",
        required: false,
      },
    ],
  },
];

// List prompts handler server.setRequestHandler(ListPromptsRequestSchema, async () => { return { prompts: PROMPTS }; });

// Get prompt handler server.setRequestHandler(GetPromptRequestSchema, async (request) => { const { name, arguments: args } = request.params;

switch (name) { case "code_review": return { description: "Code review prompt", messages: [ { role: "user", content: { type: "text", text: Please review the following ${args?.language || "code"} code. ${args?.focus ? Focus particularly on: ${args.focus}` : ""}

• Look for:
• Bugs and potential issues
• Security vulnerabilities
• Performance improvements
• Code style and best practices
• Suggestions for improvement`,
• },
• },
• ],
• };

case "explain_error": return { description: "Error explanation prompt", messages: [ { role: "user", content: { type: "text", text: `I encountered this error:

\\\ ${args?.error} \\\

${args?.context ? Context: ${args.context} : ""}

1. Please explain:
2. What this error means
3. Common causes
4. How to fix it
5. How to prevent it in the future`,
6. },
7. },
8. ],
9. };

default: throw new Error(Unknown prompt: ${name}); } }); ```

// src/utils/errors.ts
import { McpError, ErrorCode } from "@modelcontextprotocol/sdk/types.js";

export class ToolError extends Error { constructor( message: string, public code: ErrorCode = ErrorCode.InternalError ) { super(message); this.name = "ToolError"; }

toMcpError(): McpError { return new McpError(this.code, this.message); } }

// Usage in handlers server.setRequestHandler(CallToolRequestSchema, async (request) => { try { const { name, arguments: args } = request.params; // ... handle tool } catch (error) { if (error instanceof ToolError) { throw error.toMcpError(); } if (error instanceof z.ZodError) { throw new McpError( ErrorCode.InvalidParams, Invalid parameters: ${error.errors.map((e) => e.message).join(", ")} ); } throw new McpError( ErrorCode.InternalError, error instanceof Error ? error.message : "Unknown error" ); } }); ```

// ~/Library/Application Support/Claude/claude_desktop_config.json (macOS)
// %APPDATA%\Claude\claude_desktop_config.json (Windows)
{
  "mcpServers": {
    "my-mcp-server": {
      "command": "node",
      "args": ["/path/to/my-mcp-server/dist/index.js"],
      "env": {
        "API_KEY": "your-api-key"
      }
    },
    "npx-server": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/allowed/dir"]
    }
  }
}

// src/http-server.ts
import express from "express";
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";

const app = express(); const server = new Server(/* ... */);

// SSE endpoint for MCP app.get("/mcp", async (req, res) => { const transport = new SSEServerTransport("/mcp/messages", res); await server.connect(transport); });

// Message endpoint app.post("/mcp/messages", express.json(), async (req, res) => { // Handle incoming messages });

app.listen(3000, () => { console.log("MCP HTTP server running on port 3000"); }); ```

# Install MCP inspector
npx @modelcontextprotocol/inspector

// tests/tools.test.ts
import { describe, it, expect } from "vitest";
import { handleGetWeather } from "../src/tools/handlers";

describe("get_weather tool", () => { it("returns weather data for valid location", async () => { const result = await handleGetWeather({ location: "New York", units: "celsius", });

expect(result.content[0].type).toBe("text"); const data = JSON.parse(result.content[0].text); expect(data).toHaveProperty("temperature"); });

it("throws error for invalid location", async () => { await expect( handleGetWeather({ location: "", units: "celsius" }) ).rejects.toThrow(); }); }); ```

// src/index.ts
#!/usr/bin/env node

import { Server } from "@modelcontextprotocol/sdk/server/index.js"; import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js"; import { CallToolRequestSchema, ListToolsRequestSchema, } from "@modelcontextprotocol/sdk/types.js"; import { Octokit } from "@octokit/rest";

const octokit = new Octokit({ auth: process.env.GITHUB_TOKEN });

const server = new Server( { name: "github-mcp", version: "1.0.0" }, { capabilities: { tools: {} } } );

const TOOLS = [ { name: "list_issues", description: "List issues in a GitHub repository", inputSchema: { type: "object" as const, properties: { owner: { type: "string", description: "Repository owner" }, repo: { type: "string", description: "Repository name" }, state: { type: "string", enum: ["open", "closed", "all"], default: "open" }, }, required: ["owner", "repo"], }, }, { name: "create_issue", description: "Create a new issue", inputSchema: { type: "object" as const, properties: { owner: { type: "string" }, repo: { type: "string" }, title: { type: "string" }, body: { type: "string" }, labels: { type: "array", items: { type: "string" } }, }, required: ["owner", "repo", "title"], }, }, ];

server.setRequestHandler(ListToolsRequestSchema, async () => ({ tools: TOOLS }));

server.setRequestHandler(CallToolRequestSchema, async (request) => { const { name, arguments: args } = request.params;

switch (name) { case "list_issues": { const { owner, repo, state = "open" } = args as any; const { data } = await octokit.issues.listForRepo({ owner, repo, state });

return { content: [{ type: "text", text: JSON.stringify( data.map((i) => ({ number: i.number, title: i.title, state: i.state })), null, 2 ), }], }; }

case "create_issue": { const { owner, repo, title, body, labels } = args as any; const { data } = await octokit.issues.create({ owner, repo, title, body, labels });

return { content: [{ type: "text", text: Created issue #${data.number}: ${data.html_url}, }], }; }

default: throw new Error(Unknown tool: ${name}); } });

const transport = new StdioServerTransport(); server.connect(transport); ```

1. Validate inputs: Use Zod or similar for schema validation
2. Handle errors gracefully: Return meaningful error messages
3. Use environment variables: Never hardcode secrets
4. Rate limit external calls: Respect API limits
5. Log appropriately: Use stderr for logs (stdout is for MCP)
6. Document tools well: Clear descriptions help Claude use them correctly
7. Keep tools focused: Single responsibility per tool
8. Test with inspector: Verify behavior before deployment

Every MCP server should include:

• Server with name and version
• Capabilities declaration (tools/resources/prompts)
• Tool schemas with descriptions
• Input validation with clear error messages
• Proper error handling and MCP error codes
• Environment variable configuration
• Claude Desktop config example
• MCP inspector testing
• Unit tests for handlers
• README with setup instructions

Build powerful LLM applications with chains, agents, and retrieval-augmented generation.

1. Setup LangChain: Install and configure
2. Create chains: Build processing pipelines
3. Add memory: Enable conversation context
4. Define tools: Extend agent capabilities
5. Implement RAG: Add knowledge retrieval
6. Deploy: Production-ready setup

npm install langchain @langchain/openai @langchain/community

// chains/simple.ts
import { ChatOpenAI } from '@langchain/openai';
import { ChatPromptTemplate } from '@langchain/core/prompts';
import { StringOutputParser } from '@langchain/core/output_parsers';

const model = new ChatOpenAI({ modelName: 'gpt-4-turbo-preview', temperature: 0.7, });

const prompt = ChatPromptTemplate.fromMessages([ ['system', 'You are a helpful assistant that {task}.'], ['human', '{input}'], ]);

const chain = prompt.pipe(model).pipe(new StringOutputParser());

// Usage const result = await chain.invoke({ task: 'summarizes text concisely', input: 'Summarize this article: ...', }); ```

// chains/sequential.ts
import { RunnableSequence } from '@langchain/core/runnables';

// Chain 1: Extract key points const extractChain = ChatPromptTemplate.fromMessages([ ['system', 'Extract the key points from the following text.'], ['human', '{text}'], ]).pipe(model).pipe(new StringOutputParser());

// Chain 2: Summarize key points const summarizeChain = ChatPromptTemplate.fromMessages([ ['system', 'Create a brief summary from these key points.'], ['human', '{keyPoints}'], ]).pipe(model).pipe(new StringOutputParser());

// Combined chain const fullChain = RunnableSequence.from([ { keyPoints: extractChain, originalText: (input) => input.text, }, { summary: summarizeChain, keyPoints: (input) => input.keyPoints, }, ]);

const result = await fullChain.invoke({ text: 'Long article...' }); // { summary: '...', keyPoints: '...' } ```

// chains/branching.ts
import { RunnableBranch } from '@langchain/core/runnables';

const classifyChain = ChatPromptTemplate.fromMessages([ ['system', 'Classify the query as: question, complaint, or feedback'], ['human', '{query}'], ]).pipe(model).pipe(new StringOutputParser());

const questionChain = ChatPromptTemplate.fromMessages([ ['system', 'Answer this question helpfully.'], ['human', '{query}'], ]).pipe(model).pipe(new StringOutputParser());

const complaintChain = ChatPromptTemplate.fromMessages([ ['system', 'Respond empathetically to this complaint.'], ['human', '{query}'], ]).pipe(model).pipe(new StringOutputParser());

const feedbackChain = ChatPromptTemplate.fromMessages([ ['system', 'Thank the user for their feedback.'], ['human', '{query}'], ]).pipe(model).pipe(new StringOutputParser());

const routingChain = RunnableSequence.from([ { classification: classifyChain, query: (input) => input.query, }, RunnableBranch.from([ [(input) => input.classification.includes('question'), questionChain], [(input) => input.classification.includes('complaint'), complaintChain], feedbackChain, // Default ]), ]); ```

// memory/conversation.ts
import { BufferMemory } from 'langchain/memory';
import { ConversationChain } from 'langchain/chains';

const memory = new BufferMemory({ returnMessages: true, memoryKey: 'history', });

const chain = new ConversationChain({ llm: model, memory, prompt: ChatPromptTemplate.fromMessages([ ['system', 'You are a helpful assistant.'], new MessagesPlaceholder('history'), ['human', '{input}'], ]), });

// Conversation maintains context await chain.invoke({ input: 'My name is Alice' }); await chain.invoke({ input: 'What is my name?' }); // Remembers Alice ```

// memory/window.ts
import { BufferWindowMemory } from 'langchain/memory';

const memory = new BufferWindowMemory({ k: 5, // Keep last 5 exchanges returnMessages: true, memoryKey: 'history', }); ```

// memory/summary.ts
import { ConversationSummaryMemory } from 'langchain/memory';

const memory = new ConversationSummaryMemory({ llm: model, memoryKey: 'history', }); // Summarizes conversation to save tokens ```

// memory/redis.ts
import { BufferMemory } from 'langchain/memory';
import { RedisChatMessageHistory } from '@langchain/community/stores/message/redis';

const memory = new BufferMemory({ chatHistory: new RedisChatMessageHistory({ sessionId: user:${userId}:session:${sessionId}, client: redisClient, ttl: 3600, // 1 hour }), returnMessages: true, memoryKey: 'history', }); ```

// tools/custom.ts
import { DynamicTool, DynamicStructuredTool } from '@langchain/core/tools';
import { z } from 'zod';

// Simple tool const searchTool = new DynamicTool({ name: 'search', description: 'Search the web for information', func: async (query: string) => { const results = await searchAPI.search(query); return JSON.stringify(results); }, });

// Structured tool with schema const calculatorTool = new DynamicStructuredTool({ name: 'calculator', description: 'Perform mathematical calculations', schema: z.object({ expression: z.string().describe('Mathematical expression to evaluate'), }), func: async ({ expression }) => { try { const result = eval(expression); // Use safer math parser in production return String(result); } catch { return 'Error: Invalid expression'; } }, });

// Database query tool const dbQueryTool = new DynamicStructuredTool({ name: 'query_database', description: 'Query the database for user or order information', schema: z.object({ table: z.enum(['users', 'orders', 'products']), filter: z.record(z.string()).optional(), limit: z.number().default(10), }), func: async ({ table, filter, limit }) => { const results = await db[table].findMany({ where: filter, take: limit, }); return JSON.stringify(results); }, }); ```

// agents/react.ts
import { createReactAgent, AgentExecutor } from 'langchain/agents';
import { pull } from 'langchain/hub';

// Get standard ReAct prompt const prompt = await pull('hwchase17/react');

// Create agent const agent = await createReactAgent({ llm: model, tools: [searchTool, calculatorTool, dbQueryTool], prompt, });

// Create executor const executor = new AgentExecutor({ agent, tools: [searchTool, calculatorTool, dbQueryTool], verbose: true, maxIterations: 5, });

// Run agent const result = await executor.invoke({ input: 'What is the square root of the number of users in our database?', }); ```

// agents/openai-functions.ts
import { createOpenAIFunctionsAgent, AgentExecutor } from 'langchain/agents';

const prompt = ChatPromptTemplate.fromMessages([ ['system', 'You are a helpful assistant with access to tools.'], new MessagesPlaceholder('chat_history'), ['human', '{input}'], new MessagesPlaceholder('agent_scratchpad'), ]);

const agent = await createOpenAIFunctionsAgent({ llm: new ChatOpenAI({ modelName: 'gpt-4-turbo-preview' }), tools: [searchTool, calculatorTool], prompt, });

const executor = new AgentExecutor({ agent, tools: [searchTool, calculatorTool], memory: new BufferMemory({ returnMessages: true, memoryKey: 'chat_history', }), }); ```

// rag/loader.ts
import { DirectoryLoader } from 'langchain/document_loaders/fs/directory';
import { PDFLoader } from 'langchain/document_loaders/fs/pdf';
import { TextLoader } from 'langchain/document_loaders/fs/text';
import { JSONLoader } from 'langchain/document_loaders/fs/json';

const loader = new DirectoryLoader('./documents', { '.pdf': (path) => new PDFLoader(path), '.txt': (path) => new TextLoader(path), '.json': (path) => new JSONLoader(path), });

const docs = await loader.load(); ```

// rag/splitter.ts
import { RecursiveCharacterTextSplitter } from 'langchain/text_splitter';

const splitter = new RecursiveCharacterTextSplitter({ chunkSize: 1000, chunkOverlap: 200, separators: ['\n\n', '\n', ' ', ''], });

const chunks = await splitter.splitDocuments(docs); ```

// rag/vectorstore.ts
import { OpenAIEmbeddings } from '@langchain/openai';
import { PineconeStore } from '@langchain/pinecone';
import { Pinecone } from '@pinecone-database/pinecone';

const embeddings = new OpenAIEmbeddings();

const pinecone = new Pinecone(); const index = pinecone.Index(process.env.PINECONE_INDEX!);

// Create vector store from documents const vectorStore = await PineconeStore.fromDocuments(chunks, embeddings, { pineconeIndex: index, namespace: 'documents', });

// Or connect to existing const existingStore = await PineconeStore.fromExistingIndex(embeddings, { pineconeIndex: index, namespace: 'documents', }); ```

// rag/chain.ts
import { createRetrievalChain } from 'langchain/chains/retrieval';
import { createStuffDocumentsChain } from 'langchain/chains/combine_documents';

const retriever = vectorStore.asRetriever({ k: 5, // Return top 5 results searchType: 'similarity', });

const prompt = ChatPromptTemplate.fromMessages([ [ 'system', `Answer questions based on the following context. If the answer is not in the context, say so.

Context: {context}`, ], ['human', '{input}'], ]);

const combineDocsChain = await createStuffDocumentsChain({ llm: model, prompt, });

const ragChain = await createRetrievalChain({ retriever, combineDocsChain, });

const result = await ragChain.invoke({ input: 'What is our refund policy?', }); // { answer: '...', context: [Document, Document, ...] } ```

// rag/conversational.ts
import { createHistoryAwareRetriever } from 'langchain/chains/history_aware_retriever';

const contextualizePrompt = ChatPromptTemplate.fromMessages([ ['system', 'Given chat history and question, reformulate as standalone question.'], new MessagesPlaceholder('chat_history'), ['human', '{input}'], ]);

const historyAwareRetriever = await createHistoryAwareRetriever({ llm: model, retriever, rephrasePrompt: contextualizePrompt, });

const qaPrompt = ChatPromptTemplate.fromMessages([ ['system', 'Answer based on context:\n\n{context}'], new MessagesPlaceholder('chat_history'), ['human', '{input}'], ]);

const questionAnswerChain = await createStuffDocumentsChain({ llm: model, prompt: qaPrompt, });

const conversationalRagChain = await createRetrievalChain({ retriever: historyAwareRetriever, combineDocsChain: questionAnswerChain, }); ```

// streaming/stream.ts
const stream = await chain.stream({ input: 'Tell me a story' });

for await (const chunk of stream) { process.stdout.write(chunk); }

// With callbacks const result = await chain.invoke( { input: 'Tell me a story' }, { callbacks: [ { handleLLMNewToken(token: string) { process.stdout.write(token); }, }, ], } ); ```

// config/langchain.ts
import { ChatOpenAI } from '@langchain/openai';

export const model = new ChatOpenAI({ modelName: process.env.OPENAI_MODEL || 'gpt-4-turbo-preview', temperature: 0.7, maxTokens: 4096, timeout: 60000, maxRetries: 3, cache: true, });

// With rate limiting import Bottleneck from 'bottleneck';

const limiter = new Bottleneck({ minTime: 100, // 10 requests per second maxConcurrent: 5, });

export async function invokeWithRateLimit(chain: any, input: any) { return limiter.schedule(() => chain.invoke(input)); } ```

1. Use LCEL: LangChain Expression Language for composability
2. Add memory strategically: Balance context vs tokens
3. Define clear tool descriptions: Help the agent choose correctly
4. Chunk documents properly: Optimize for retrieval
5. Use streaming: Better UX for long responses
6. Cache embeddings: Reduce API costs
7. Monitor token usage: Track and optimize
8. Handle errors gracefully: Retry and fallback

Every LangChain implementation should include:

• Model configuration with retries
• Prompt templates with variables
• Chain composition with LCEL
• Memory for conversation context
• Tools with clear descriptions
• Agent with iteration limits
• RAG with proper chunking
• Streaming for UX
• Error handling
• Rate limiting for production

Build robust, reusable prompt templates with clear contracts and consistent outputs.

System Prompt: Role, persona, constraints, output format User Prompt: Task, context, variables, examples Few-Shot Examples: Input/output pairs demonstrating desired behavior Output Contract: Strict format specification (JSON schema, Markdown structure) Style Rules: Tone, verbosity, formatting preferences Guardrails: Do's and don'ts, safety constraints

# System Prompt: Code Review Assistant

You are an expert code reviewer specializing in {language} and {framework}. Your role is to provide constructive, actionable feedback on code quality, best practices, and potential issues.

Provide your review in the following JSON structure:

{
  "summary": "Brief 1-2 sentence overview",
  "issues": [
    {
      "severity": "critical|major|minor",
      "line": number,
      "message": "Description of the issue",
      "suggestion": "How to fix it"
    }
  ],
  "strengths": ["List of positive aspects"],
  "overall_score": 1-10
}
```

• Be constructive and specific
• Cite line numbers for issues
• Provide actionable suggestions
• Balance criticism with praise
• Use professional, respectful tone

• Do NOT suggest unnecessary refactors
• Do focus on correctness, security, performance
• Do NOT be overly pedantic about style
• Do consider the context and project requirements

// prompt-templates/code-review.ts
export const codeReviewPrompt = (variables: {
  language: string;
  framework: string;
  code: string;
  context?: string;
}) => `
Please review the following ${variables.language} code:

${variables.context ? Context: ${variables.context}\n : ''}

\\\${variables.language} ${variables.code} \\\

Provide a thorough code review following the output format specified in the system prompt. `;

// Usage const prompt = codeReviewPrompt({ language: 'typescript', framework: 'React', code: userSubmittedCode, context: 'This is a production component for user authentication', }); ````

# Few-Shot Examples

Input:

function calculateTotal(items: Item[]): number {
  return items.reduce((sum, item) => sum + item.price, 0);
}
```

Output:

{
  "summary": "Clean, type-safe implementation with no issues found.",
  "issues": [],
  "strengths": [
    "Type safety with TypeScript",
    "Functional approach with reduce",
    "Clear, descriptive naming"
  ],
  "overall_score": 9
}

Input:

function calc(arr) {
  let total = 0;
  for (var i = 0; i < arr.length; i++) {
    total = total + arr[i].price;
  }
  return total;
}

Output:

{
  "summary": "Functional but has type safety and modern syntax issues.",
  "issues": [
    {
      "severity": "major",
      "line": 1,
      "message": "Missing type annotations for parameters and return value",
      "suggestion": "Add types: function calc(arr: Item[]): number"
    },
    {
      "severity": "minor",
      "line": 3,
      "message": "Using 'var' instead of 'let' or 'const'",
      "suggestion": "Replace 'var' with 'let': for (let i = 0; ...)"
    }
  ],
  "strengths": ["Logic is correct", "Handles empty array case"],
  "overall_score": 6
}

// Define strict output schema
import { z } from 'zod';

export const codeReviewSchema = z.object({ summary: z.string().min(10).max(200), issues: z.array(z.object({ severity: z.enum(['critical', 'major', 'minor']), line: z.number().int().positive(), message: z.string(), suggestion: z.string(), })), strengths: z.array(z.string()), overall_score: z.number().int().min(1).max(10), });

// Validate LLM output export const parseCodeReview = (output: string) => { try { const parsed = JSON.parse(output); return codeReviewSchema.parse(parsed); } catch (error) { throw new Error('Invalid code review output format'); } }; ````

export interface PromptVariables {
  // Required
  required_field: string;

// Optional with defaults optional_field?: string;

// Constrained values severity_level: "low" | "medium" | "high";

// Numeric with ranges max_tokens: number; // 1-4096 }

export const buildPrompt = (vars: PromptVariables): string => { // Validate variables if (!vars.required_field) { throw new Error("required_field is required"); }

// Set defaults const optional = vars.optional_field ?? "default value";

// Build prompt return Task: ${vars.required_field} Options: ${optional} Severity: ${vars.severity_level}; }; ```

## Tone Guidelines

• Professional: Formal language, no slang
• Friendly: Conversational but respectful
• Technical: Precise terminology, assume expertise
• Educational: Explain concepts, teach as you go

• Concise: 1-2 sentences, bullet points
• Standard: 1 paragraph per point
• Detailed: Full explanations with examples
• Comprehensive: Deep dive with references

• Use markdown headers for structure
• Bold important terms
• Code blocks for technical content
• Lists for enumeration
• Tables for comparisons
• `

## Do's

✓ Provide specific, actionable feedback ✓ Include code examples when relevant ✓ Reference line numbers for issues ✓ Suggest concrete improvements ✓ Balance criticism with praise ✓ Consider context and constraints

✗ Don't be vague ("this is bad") ✗ Don't suggest unnecessary rewrites ✗ Don't ignore security issues ✗ Don't be overly pedantic ✗ Don't assume unlimited resources ✗ Don't make assumptions without context ```

// Multi-step prompts
export const chainedPrompts = {
  step1_analyze: (code: string) => `
    Analyze this code and identify potential issues:
    ${code}

List issues in JSON array format with severity and description. `,

step2_suggest: (issues: Issue[]) => ` Given these code issues: ${JSON.stringify(issues)}

Provide detailed fix suggestions for each issue. `,

step3_summarize: (suggestions: Suggestion[]) => Summarize these code review suggestions into a final report: ${JSON.stringify(suggestions)} , };

// Execute chain const issues = await llm(chainedPrompts.step1_analyze(code)); const suggestions = await llm(chainedPrompts.step2_suggest(issues)); const report = await llm(chainedPrompts.step3_summarize(suggestions)); ```

// Track prompt versions
export const PROMPT_VERSIONS = {
  "v1.0": {
    system: "Original system prompt...",
    user: (vars) => `Original user prompt...`,
    deprecated: false,
  },
  "v1.1": {
    system: "Improved system prompt with better constraints...",
    user: (vars) => `Updated user prompt...`,
    deprecated: false,
    changes: "Added JSON schema validation, improved examples",
  },
  "v1.0-deprecated": {
    system: "...",
    user: (vars) => `...`,
    deprecated: true,
    deprecation_reason: "Replaced by v1.1 with better output format",
  },
};

// Use specific version const prompt = PROMPT_VERSIONS["v1.1"]; ```

// Test cases for prompt validation
const testCases = [
  {
    input: { code: "function test() {}", language: "javascript" },
    expected: {
      hasIssues: false,
      scoreRange: [8, 10],
    },
  },
  {
    input: { code: "func test(arr) { return arr[0] }", language: "javascript" },
    expected: {
      hasIssues: true,
      minIssues: 2,
      severities: ["major", "minor"],
    },
  },
];

// Run tests for (const test of testCases) { const output = await llm(buildPrompt(test.input)); const parsed = parseCodeReview(output);

if (test.expected.hasIssues) { assert(parsed.issues.length >= test.expected.minIssues); } if (test.expected.scoreRange) { assert(parsed.overall_score >= test.expected.scoreRange[0]); assert(parsed.overall_score <= test.expected.scoreRange[1]); } } ```

1. Clear instructions: Be explicit about what you want
2. Output contracts: Define strict schemas
3. Few-shot examples: Show, don't just tell
4. Variable validation: Check inputs before building prompts
5. Version tracking: Maintain prompt history
6. Test thoroughly: Validate against edge cases
7. Iterate: Improve based on real outputs
8. Document constraints: Explain limitations

• System prompt with role and constraints
• User prompt template with variables
• Output format specification (JSON schema)
• 3+ few-shot examples (good and bad)
• Style guidelines documented
• Do's and don'ts list
• Variable validation logic
• Output parsing/validation
• Test cases for prompt
• Version tracking system

Systematically test prompt changes to prevent regressions.

{
  "test_cases": [
    {
      "id": "test_001",
      "input": "Summarize this article",
      "context": "Article text here...",
      "expected_behavior": "Concise 2-3 sentence summary",
      "baseline_output": "Output from v1.0 prompt",
      "must_include": ["main point", "conclusion"],
      "must_not_include": ["opinion", "speculation"]
    }
  ]
}

def compare_prompts(old_prompt, new_prompt, test_cases):
    results = {
        "test_cases": [],
        "summary": {
            "total": len(test_cases),
            "improvements": 0,
            "regressions": 0,
            "unchanged": 0,
        },
        "breakages": []
    }

for test in test_cases: old_output = llm(old_prompt.format(test)) new_output = llm(new_prompt.format(test))

comparison = { "test_id": test["id"], "old_output": old_output, "new_output": new_output, "diff": compute_diff(old_output, new_output), "scores": { "old": score_output(old_output, test), "new": score_output(new_output, test), }, "verdict": classify_change(old_output, new_output, test) }

results["test_cases"].append(comparison) results["summary"][comparison["verdict"]] += 1

if comparison["verdict"] == "regressions": results["breakages"].append(analyze_breakage(comparison, test))

return results ```

def calculate_stability_metrics(results):
    return {
        "output_stability": measure_output_consistency(results),
        "format_stability": check_format_preservation(results),
        "constraint_adherence": check_constraints(results),
        "behavioral_consistency": measure_behavior_delta(results),
    }

def measure_output_consistency(results): """How similar are outputs between versions?""" similarities = [] for result in results["test_cases"]: sim = semantic_similarity( result["old_output"], result["new_output"] ) similarities.append(sim) return sum(similarities) / len(similarities) ```

def analyze_breakage(comparison, test_case):
    """Identify why the new prompt failed"""
    reasons = []

new_out = comparison["new_output"]

return { "test_id": test_case["id"], "reasons": reasons, "old_output": comparison["old_output"][:100], "new_output": comparison["new_output"][:100], } ```

def suggest_fixes(breakages):
    """Generate fix suggestions based on breakage patterns"""
    suggestions = []

return suggestions ```

# Prompt Regression Report

• Total tests: 50
• Improvements: 5 (10%)
• Regressions: 3 (6%)
• Unchanged: 42 (84%)

• Output stability: 0.87
• Format stability: 0.95
• Constraint adherence: 0.94

Old output: "The main benefit is cost savings..." New output: "This approach provides flexibility..." Issue: Missing required keyword 'cost' Fix: Add explicit instruction: "Mention cost implications in your response"

1. Revert changes that caused regressions (tests: 005, 012, 023)
2. Add format constraints for JSON output
3. Run full test suite before deployment
4. `

• Test with diverse inputs
• Compare across multiple runs (LLMs are stochastic)
• Track metrics over time
• Automate in CI/CD
• Review all regressions before deploy
• Maintain test case library

• Test cases defined (30+)
• Comparison runner
• Stability metrics
• Breakage analyzer
• Fix suggestions
• Diff visualizer
• Automated report
• CI integration

Build comprehensive safety systems for LLM applications.

1. Input filtering: Block malicious prompts
2. Output filtering: Redact sensitive data
3. Topic constraints: Policy-based refusals
4. PII detection: Mask personal information
5. Prompt injection: Detect manipulation attempts

import re
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

analyzer = AnalyzerEngine() anonymizer = AnonymizerEngine()

def redact_pii(text: str) -> str: # Detect PII results = analyzer.analyze( text=text, language='en', entities=["EMAIL_ADDRESS", "PHONE_NUMBER", "CREDIT_CARD", "SSN"] )

def detect_prompt_injection(user_input: str) -> bool:
    """Detect common prompt injection patterns"""
    patterns = [
        r'ignore (previous|above) instructions',
        r'disregard (all|any) (prior|previous)',
        r'you are now',
        r'new instructions',
        r'system:',
        r'override',
    ]

for pattern in patterns: if re.search(pattern, user_input, re.IGNORECASE): return True

return False

# Define allowed/disallowed topics
POLICY = {
    "allowed_topics": [
        "product_features",
        "troubleshooting",
        "billing",
        "account_management"
    ],
    "disallowed_topics": [
        "medical_advice",
        "legal_advice",
        "financial_advice",
        "politics",
        "violence"
    ],
    "requires_disclaimer": [
        "security_practices",
        "data_privacy"
    ]
}

Query: {query}

Return only the topic name. """ return llm(classification_prompt)

if topic in POLICY["disallowed_topics"]: return { "allowed": False, "reason": f"Cannot provide {topic}", "refusal": REFUSAL_TEMPLATES[topic] }

return {"allowed": True, "topic": topic} ```

REFUSAL_TEMPLATES = {
    "medical_advice": """
        I cannot provide medical advice. Please consult with a healthcare
        professional for medical concerns.
    """,
    "legal_advice": """
        I cannot provide legal advice. For legal matters, please consult
        with a qualified attorney.
    """,
    "out_of_scope": """
        I'm designed to help with product documentation and support.
        This question is outside my area of expertise.
    """,
}

def refuse_safely(reason: str) -> str: return REFUSAL_TEMPLATES.get(reason, REFUSAL_TEMPLATES["out_of_scope"]) ```

def validate_output(output: str) -> dict:
    """Check output before returning to user"""
    issues = []

return { "safe": len(issues) == 0, "issues": issues, "sanitized_output": output } ```

def apply_guardrails(user_input: str) -> dict:
    # 1. Input validation
    if detect_prompt_injection(user_input):
        return {
            "allowed": False,
            "response": "Invalid request detected."
        }

return { "allowed": True, "response": response } ```

• Layer multiple defenses
• Log all blocked requests
• Provide helpful refusals
• Redact, don't reject when possible
• Regular pattern updates
• Human review of edge cases

• PII detection implemented
• Prompt injection detection
• Topic classification
• Policy constraints defined
• Refusal templates written
• Output validation
• Logging/monitoring
• Test cases for bypasses

Build systematic evaluation frameworks for LLM applications.

[
  {
    "id": "test_001",
    "category": "code_generation",
    "input": "Write a Python function to reverse a string",
    "expected_output": "def reverse_string(s: str) -> str:\n    return s[::-1]",
    "rubric": {
      "correctness": 1.0,
      "style": 0.8,
      "documentation": 0.5
    },
    "metadata": {
      "difficulty": "easy",
      "tags": ["python", "strings"]
    }
  }
]

from typing import Dict, Any

def score_exact_match(actual: str, expected: str) -> float: """Binary score: 1.0 if exact match, 0.0 otherwise""" return 1.0 if actual.strip() == expected.strip() else 0.0

def score_semantic_similarity(actual: str, expected: str) -> float: """Cosine similarity of embeddings""" actual_emb = get_embedding(actual) expected_emb = get_embedding(expected) return cosine_similarity(actual_emb, expected_emb)

def score_contains_keywords(actual: str, keywords: List[str]) -> float: """Percentage of required keywords present""" found = sum(1 for kw in keywords if kw.lower() in actual.lower()) return found / len(keywords)

def score_with_llm(actual: str, expected: str, rubric: Dict[str, float]) -> Dict[str, float]: """Use LLM as judge""" prompt = f""" Grade this output on a scale of 0-1 for each criterion:

Expected: {expected} Actual: {actual}

Criteria: {', '.join(rubric.keys())}

Return JSON with scores. """ return json.loads(llm(prompt)) ```

class EvaluationHarness:
    def __init__(self, dataset_path: str):
        self.dataset = self.load_dataset(dataset_path)
        self.results = []

def run_evaluation(self, model_fn): for test_case in self.dataset: # Generate output actual = model_fn(test_case["input"])

return self.generate_report()

def score_output(self, actual, expected, rubric): return { "exact_match": score_exact_match(actual, expected), "semantic_similarity": score_semantic_similarity(actual, expected), **score_with_llm(actual, expected, rubric) }

def check_threshold(self, scores, test_case): min_scores = test_case.get("min_scores", {}) for metric, threshold in min_scores.items(): if scores.get(metric, 0) < threshold: return False return True ```

# Define thresholds per category
THRESHOLDS = {
    "code_generation": {
        "correctness": 0.9,
        "style": 0.7,
    },
    "summarization": {
        "semantic_similarity": 0.8,
        "brevity": 0.7,
    },
    "classification": {
        "exact_match": 1.0,
    }
}

def check_test_passed(result: Dict) -> bool: category = result["category"] thresholds = THRESHOLDS.get(category, {})

for metric, threshold in thresholds.items(): if result["scores"].get(metric, 0) < threshold: return False

return True ```

def generate_regression_report(baseline_results, current_results):
    report = {
        "summary": {},
        "regressions": [],
        "improvements": [],
        "unchanged": 0
    }

for baseline, current in zip(baseline_results, current_results): assert baseline["test_id"] == current["test_id"]

baseline_passed = baseline["passed"] current_passed = current["passed"]

if baseline_passed and not current_passed: report["regressions"].append({ "test_id": baseline["test_id"], "category": baseline["category"], "baseline_scores": baseline["scores"], "current_scores": current["scores"], }) elif not baseline_passed and current_passed: report["improvements"].append(baseline["test_id"]) else: report["unchanged"] += 1

report["summary"] = { "total_tests": len(baseline_results), "regressions": len(report["regressions"]), "improvements": len(report["improvements"]), "unchanged": report["unchanged"], }

return report ```

# Run evaluation on every commit
def ci_evaluation():
    harness = EvaluationHarness("golden_dataset.json")
    results = harness.run_evaluation(production_model)

print("✅ All tests passed!") ```

1. Representative dataset: Cover edge cases
2. Multiple metrics: Don't rely on one score
3. Human validation: Review LLM judge scores
4. Version datasets: Track changes over time
5. Automate in CI: Catch regressions early
6. Regular updates: Add new test cases

• Golden dataset created (50+ examples)
• Multiple scoring functions
• Pass/fail thresholds defined
• Test runner implemented
• Regression comparison
• Report generation
• CI integration
• Baseline established

Optimize LLM applications for cost and performance.

class CostAnalyzer:
    def __init__(self):
        self.costs = {
            "llm_calls": 0,
            "embeddings": 0,
            "tool_calls": 0,
        }
        self.counts = {
            "llm_calls": 0,
            "embeddings": 0,
        }

def track_llm_call(self, tokens_in: int, tokens_out: int): # GPT-4 pricing cost = (tokens_in / 1000) * 0.03 + (tokens_out / 1000) * 0.06 self.costs["llm_calls"] += cost self.counts["llm_calls"] += 1

def report(self): return { "total_cost": sum(self.costs.values()), "breakdown": self.costs, "avg_cost_per_call": self.costs["llm_calls"] / self.counts["llm_calls"], } ```

import hashlib
from functools import lru_cache

class LLMCache: def __init__(self, redis_client): self.cache = redis_client self.ttl = 3600 # 1 hour

def get_cache_key(self, prompt: str, model: str) -> str: content = f"{model}:{prompt}" return f"llm_cache:{hashlib.sha256(content.encode()).hexdigest()}"

def get(self, prompt: str, model: str): key = self.get_cache_key(prompt, model) return self.cache.get(key)

def set(self, prompt: str, model: str, response: str): key = self.get_cache_key(prompt, model) self.cache.setex(key, self.ttl, response)

def cached_llm_call(prompt: str, model: str = "gpt-4"): # Check cache cached = cache.get(prompt, model) if cached: return cached

return response ```

MODEL_PRICING = {
    "gpt-4": {"input": 0.03, "output": 0.06},
    "gpt-3.5-turbo": {"input": 0.0005, "output": 0.0015},
    "claude-3-opus": {"input": 0.015, "output": 0.075},
    "claude-3-sonnet": {"input": 0.003, "output": 0.015},
}

def select_model_by_complexity(query: str) -> str: """Use cheaper models for simple queries""" # Classify complexity complexity = classify_complexity(query)

if complexity == "simple": return "gpt-3.5-turbo" # 60x cheaper elif complexity == "medium": return "claude-3-sonnet" else: return "gpt-4"

def classify_complexity(query: str) -> str: # Simple heuristics if len(query) < 100 and "?" in query: return "simple" elif any(word in query.lower() for word in ["analyze", "complex", "detailed"]): return "complex" return "medium" ```

def optimize_prompt(prompt: str) -> str:
    """Reduce token count while preserving meaning"""
    optimizations = [
        # Remove extra whitespace
        lambda p: re.sub(r'\s+', ' ', p),

for optimize in optimizations: prompt = optimize(prompt)

return prompt.strip()

async def batch_llm_calls(prompts: List[str], batch_size: int = 5):
    """Process multiple prompts in parallel"""
    results = []

for i in range(0, len(prompts), batch_size): batch = prompts[i:i + batch_size]

results.extend(batch_results)

return results

import time

class LatencyTracker: def __init__(self): self.timings = {}

def track(self, operation: str): def decorator(func): def wrapper(*args, **kwargs): start = time.time() result = func(*args, **kwargs) duration = time.time() - start

if operation not in self.timings: self.timings[operation] = [] self.timings[operation].append(duration)

return result return wrapper return decorator

def report(self): return { op: { "count": len(times), "total": sum(times), "avg": sum(times) / len(times), "p95": sorted(times)[int(len(times) * 0.95)] } for op, times in self.timings.items() }

@tracker.track("llm_call") def call_llm(prompt): return llm(prompt)

def generate_recommendations(cost_analysis, latency_analysis):
    recs = []

if cost_analysis["avg_cost_per_call"] > 0.01: recs.append({ "issue": "Using expensive model for all queries", "recommendation": "Use gpt-3.5-turbo for simple queries", "impact": "60% cost reduction", })

return recs ```

async def streaming_llm(prompt: str):
    """Stream tokens as they're generated"""
    async for chunk in llm_stream(prompt):
        yield chunk
        # User sees partial response immediately

1. Cache aggressively: Identical queries cached
2. Model selection: Use cheaper models when possible
3. Prompt optimization: Reduce unnecessary tokens
4. Batching: Parallel execution for throughput
5. Streaming: Faster perceived latency
6. Monitor costs: Track per-user, per-feature
7. Set budgets: Alert on anomalies

• Cost tracking implementation
• Caching layer
• Model selection logic
• Prompt optimization
• Batching for parallel calls
• Latency tracking
• Hotspot analysis
• Optimization recommendations
• Budget alerts
• Performance dashboard

Systematically diagnose and fix LLM output issues.

class FailureType(Enum):
    HALLUCINATION = "hallucination"
    FORMAT_VIOLATION = "format_violation"
    CONSTRAINT_BREAK = "constraint_break"
    REASONING_ERROR = "reasoning_error"
    TOOL_MISUSE = "tool_misuse"
    REFUSAL = "unexpected_refusal"
    INCOMPLETE = "incomplete_output"

def diagnose_failure(input: str, output: str, expected: dict) -> dict:
    """Identify why LLM output failed"""

issues = []

return { "has_issues": len(issues) > 0, "issues": issues, "primary_issue": issues[0] if issues else None }

def contains_hallucination_markers(output: str) -> bool: """Detect common hallucination patterns""" markers = [ r'According to.*that doesn\'t exist', r'In \d{4}.*before that year', r'contradicts itself', ] return any(re.search(marker, output, re.IGNORECASE) for marker in markers) ```

PROMPT_FIXES = {
    FailureType.FORMAT_VIOLATION: """
Add strict format instructions:
"Output MUST be valid JSON with this exact structure:
```json
{{"field1": "value", "field2": 123}}

Do not add any text before or after the JSON." """,

FailureType.HALLUCINATION: """

Add grounding instructions: "Base your response ONLY on the provided context. If information is not in the context, say 'I don't have that information.' Never make up facts or details." """,

FailureType.CONSTRAINT_BREAK: """

Add explicit constraints: "Your response must be:

• Maximum 200 words
• No code examples
• Professional tone only"
• """,
• FailureType.REASONING_ERROR: """
• Add step-by-step reasoning:
• "Think through this step by step:

1. First, identify...
2. Then, consider...
3. Finally, conclude..."
4. """,
5. }

def suggest_prompt_fix(diagnosis: dict, current_prompt: str) -> str: """Generate improved prompt based on diagnosis""" primary_issue = diagnosis["primary_issue"] if not primary_issue: return current_prompt

fix = PROMPT_FIXES[primary_issue["type"]]

return f"""{current_prompt}

IMPORTANT INSTRUCTIONS: {fix} """

def suggest_tool_fixes(diagnosis: dict, tool_schema: dict) -> dict:
    """Improve tool schema to prevent misuse"""

fixes = {}

for issue in diagnosis["issues"]: if issue["type"] == FailureType.TOOL_MISUSE: # Make required parameters more explicit fixes["parameters"] = { **tool_schema["parameters"], "description": "REQUIRED: " + tool_schema["parameters"].get("description", "") }

return {tool_schema, fixes} ````

def generate_test_cases(diagnosis: dict, failed_input: str, failed_output: str) -> List[dict]:
    """Create regression test cases from failures"""

test_cases = []

return test_cases ```

def debug_llm_output(
    prompt: str,
    output: str,
    expected: dict,
    context: dict = {}
) -> dict:
    """Complete debugging workflow"""

if not diagnosis["has_issues"]: return {"status": "ok", "diagnosis": diagnosis}

return { "status": "issues_found", "diagnosis": diagnosis, "fixes": { "prompt": fixed_prompt, "fix_verified": fix_works, }, "test_cases": test_cases, "recommendations": generate_recommendations(diagnosis, context) } ```

def interactive_debug():
    """Interactive debugging session"""
    print("LLM Debugger - Interactive Mode")

prompt = input("Enter prompt: ") expected_output = input("Expected output format (json/text): ")

output = llm(prompt) print(f"\nGenerated output:\n{output}\n")

if input("Does this look correct? (y/n): ").lower() == 'n': diagnosis = diagnose_failure(prompt, output, {"format": expected_output})

print("\nDiagnosis:") for issue in diagnosis["issues"]: print(f"- {issue['type'].value}: {issue['details']}")

print("\nSuggested fix:") print(suggest_prompt_fix(diagnosis, prompt)) ```

1. Reproduce consistently: Multiple runs
2. Isolate variables: Test prompt changes one at a time
3. Check examples: Are few-shot examples clear?
4. Validate constraints: Are they explicit enough?
5. Test edge cases: Boundary conditions
6. Log everything: Inputs, outputs, issues
7. Iterate systematically: Track what fixes work

• Failure taxonomy defined
• Root cause analyzer
• Prompt fix suggestions
• Tool schema improvements
• Test case generator
• Fix verification
• Debugging workflow
• Recommendations engine
• Interactive debugger
• Logging system