Developing Custom Strategies

This guide shows how to create custom strategies for both embedding and retrieval pipelines. All strategies follow the same pattern: write a pure TypeScript function, then convert it to an Aktor strategy.

Strategy Development Pattern

1. Pure TypeScript Function

First, write your logic as a pure TypeScript function that matches the strategy interface.

2. Convert to Aktor Strategy

Use createAktorFunctionAsync to convert your function into an Aktor strategy.

3. Export Both Versions

Export both the pure function and Aktor version for maximum flexibility.

Embedding Pipeline Custom Strategies

Custom Document Loader Strategy

Scenario: Process PDF files with Azure Document Intelligence

import { createAktorFunctionAsync } from '@operaide/aktor';
import type { DocumentContent } from '@operaide/vector';
import type { ToAktorStrategy } from '@operaide/vector';

// 1. Pure TypeScript function
export const pdfDocumentLoaderStrategy = async ({ file }: { file: File }): Promise<DocumentContent> => {
    // Check if file is PDF
    if (!file.name.toLowerCase().endsWith('.pdf')) {
        throw new Error(`Unsupported file type: ${file.name}`);
    }
    
    // Process with Azure Document Intelligence
    const markdown = await processFileWithDocumentIntelligence(file);
    
    return {
        id: generateDocumentId(file.name),
        title: extractTitleFromPDF(file.name),
        content: markdown,
        metadata: {
            type: 'pdf',
            source: file.name,
            createdAt: new Date().toISOString(),
            processingMethod: 'azure_document_intelligence',
            pageCount: await getPDFPageCount(file),
        }
    };
};

// 2. Convert to Aktor strategy
export type PdfDocumentLoaderStrategy = typeof pdfDocumentLoaderStrategy;
export type AktorPdfDocumentLoaderStrategy = ToAktorStrategy<PdfDocumentLoaderStrategy>;

export const aktorPdfDocumentLoaderStrategy: AktorPdfDocumentLoaderStrategy = createAktorFunctionAsync(
    'aktorPdfDocumentLoaderStrategy',
    pdfDocumentLoaderStrategy
);

// Helper functions
async function processFileWithDocumentIntelligence(file: File): Promise<string> {
    // Azure Document Intelligence API call
    const formData = new FormData();
    formData.append('file', file);
    
    const response = await fetch('/api/azure/document-intelligence', {
        method: 'POST',
        body: formData,
    });
    
    const result = await response.json();
    return result.markdown;
}

function extractTitleFromPDF(filename: string): string {
    return filename.replace('.pdf', '').replace(/[-_]/g, ' ');
}

async function getPDFPageCount(file: File): Promise<number> {
    // Implementation to count PDF pages
    return 1; // Placeholder
}

function generateDocumentId(filename: string): string {
    return `doc_${Date.now()}_${filename.replace(/[^a-zA-Z0-9]/g, '_')}`;
}

Custom Semantic Chunking Strategy

Scenario: Split documents using semantic boundaries instead of fixed sizes

import { createAktorFunctionAsync } from '@operaide/aktor';
import type { CleanedDocument, DocumentChunk } from '@operaide/vector';
import type { ToAktorStrategy } from '@operaide/vector';

// 1. Pure TypeScript function
export const semanticChunkingStrategy = async ({ 
    document 
}: { 
    document: CleanedDocument 
}): Promise<DocumentChunk[]> => {
    // Split by sentences first
    const sentences = splitIntoSentences(document.content);
    
    // Group sentences into semantic chunks
    const semanticChunks = await createSemanticChunks(sentences, {
        maxTokens: 512,
        overlapSentences: 2,
        coherenceThreshold: 0.7,
    });
    
    // Convert to DocumentChunk format
    return semanticChunks.map((chunk, index) => ({
        id: generateChunkId(document.id, index),
        documentId: document.id,
        content: chunk.text,
        index,
        metadata: {
            ...document.metadata,
            chunkingMethod: 'semantic',
            coherenceScore: chunk.coherenceScore,
            sentenceCount: chunk.sentenceCount,
            tokenCount: chunk.tokenCount,
        }
    }));
};

// 2. Convert to Aktor strategy
export type SemanticChunkingStrategy = typeof semanticChunkingStrategy;
export type AktorSemanticChunkingStrategy = ToAktorStrategy<SemanticChunkingStrategy>;

export const aktorSemanticChunkingStrategy: AktorSemanticChunkingStrategy = createAktorFunctionAsync(
    'aktorSemanticChunkingStrategy',
    semanticChunkingStrategy
);

// Helper functions
function splitIntoSentences(text: string): string[] {
    // Simple sentence splitting - could use more sophisticated NLP
    return text.split(/[.!?]+/)
        .map(s => s.trim())
        .filter(s => s.length > 0);
}

interface SemanticChunk {
    text: string;
    coherenceScore: number;
    sentenceCount: number;
    tokenCount: number;
}

async function createSemanticChunks(
    sentences: string[], 
    options: {
        maxTokens: number;
        overlapSentences: number;
        coherenceThreshold: number;
    }
): Promise<SemanticChunk[]> {
    const chunks: SemanticChunk[] = [];
    let currentChunk: string[] = [];
    let currentTokens = 0;
    
    for (let i = 0; i < sentences.length; i++) {
        const sentence = sentences[i];
        const sentenceTokens = estimateTokenCount(sentence);
        
        // Check if adding this sentence would exceed token limit
        if (currentTokens + sentenceTokens > options.maxTokens && currentChunk.length > 0) {
            // Finalize current chunk
            const chunkText = currentChunk.join('. ') + '.';
            const coherenceScore = await calculateCoherenceScore(currentChunk);
            
            chunks.push({
                text: chunkText,
                coherenceScore,
                sentenceCount: currentChunk.length,
                tokenCount: currentTokens,
            });
            
            // Start new chunk with overlap
            const overlapStart = Math.max(0, currentChunk.length - options.overlapSentences);
            currentChunk = currentChunk.slice(overlapStart);
            currentTokens = currentChunk.reduce((sum, s) => sum + estimateTokenCount(s), 0);
        }
        
        currentChunk.push(sentence);
        currentTokens += sentenceTokens;
    }
    
    // Add final chunk if it exists
    if (currentChunk.length > 0) {
        const chunkText = currentChunk.join('. ') + '.';
        const coherenceScore = await calculateCoherenceScore(currentChunk);
        
        chunks.push({
            text: chunkText,
            coherenceScore,
            sentenceCount: currentChunk.length,
            tokenCount: currentTokens,
        });
    }
    
    return chunks;
}

function estimateTokenCount(text: string): number {
    // Rough estimation: 1 token ≈ 4 characters
    return Math.ceil(text.length / 4);
}

async function calculateCoherenceScore(sentences: string[]): Promise<number> {
    // Placeholder for semantic coherence calculation
    // In real implementation, you might use sentence transformers
    return Math.random() * 0.3 + 0.7; // Mock score between 0.7-1.0
}

function generateChunkId(documentId: string, index: number): string {
    return `${documentId}_chunk_${index}`;
}

Custom OpenAI Embedding Strategy

Scenario: Use OpenAI's latest embedding models with custom configuration

import { createAktorFunctionAsync } from '@operaide/aktor';
import type { EnrichedChunk, EmbeddedChunk } from '@operaide/vector';
import type { ToAktorStrategy } from '@operaide/vector';

// 1. Pure TypeScript function
export const openAIEmbeddingStrategy = async ({ 
    chunks 
}: { 
    chunks: EnrichedChunk[] 
}): Promise<EmbeddedChunk[]> => {
    const embeddingModel = 'text-embedding-3-large';
    const batchSize = 100;
    const maxRetries = 3;
    
    const embeddedChunks: EmbeddedChunk[] = [];
    
    // Process in batches for efficiency
    for (let i = 0; i < chunks.length; i += batchSize) {
        const batch = chunks.slice(i, i + batchSize);
        
        try {
            const embeddings = await generateEmbeddingsWithRetry(
                batch.map(chunk => chunk.content),
                embeddingModel,
                maxRetries
            );
            
            // Combine chunks with their embeddings
            batch.forEach((chunk, index) => {
                embeddedChunks.push({
                    ...chunk,
                    embedding: embeddings[index],
                    embeddingModel,
                    embeddedAt: new Date().toISOString(),
                });
            });
            
        } catch (error) {
            console.error(`Failed to embed batch starting at index ${i}:`, error);
            
            // Add chunks with empty embeddings to maintain order
            batch.forEach(chunk => {
                embeddedChunks.push({
                    ...chunk,
                    embedding: Array(3072).fill(0), // text-embedding-3-large dimensions
                    embeddingModel: `${embeddingModel}_failed`,
                    embeddedAt: new Date().toISOString(),
                });
            });
        }
    }
    
    return embeddedChunks;
};

// 2. Convert to Aktor strategy
export type OpenAIEmbeddingStrategy = typeof openAIEmbeddingStrategy;
export type AktorOpenAIEmbeddingStrategy = ToAktorStrategy<OpenAIEmbeddingStrategy>;

export const aktorOpenAIEmbeddingStrategy: AktorOpenAIEmbeddingStrategy = createAktorFunctionAsync(
    'aktorOpenAIEmbeddingStrategy',
    openAIEmbeddingStrategy
);

// Helper functions
async function generateEmbeddingsWithRetry(
    texts: string[], 
    model: string, 
    maxRetries: number
): Promise<number[][]> {
    let lastError: Error;
    
    for (let attempt = 1; attempt <= maxRetries; attempt++) {
        try {
            const response = await fetch('https://api.openai.com/v1/embeddings', {
                method: 'POST',
                headers: {
                    'Authorization': `Bearer ${process.env.OPENAI_API_KEY}`,
                    'Content-Type': 'application/json',
                },
                body: JSON.stringify({
                    input: texts,
                    model: model,
                }),
            });
            
            if (!response.ok) {
                throw new Error(`OpenAI API error: ${response.status} ${response.statusText}`);
            }
            
            const data = await response.json();
            return data.data.map((item: any) => item.embedding);
            
        } catch (error) {
            lastError = error as Error;
            console.warn(`Embedding attempt ${attempt} failed:`, error);
            
            if (attempt < maxRetries) {
                // Exponential backoff
                await new Promise(resolve => setTimeout(resolve, Math.pow(2, attempt) * 1000));
            }
        }
    }
    
    throw lastError!;
}

Retrieval Pipeline Custom Strategies

Custom Query Expansion Strategy

Scenario: Expand queries with synonyms and related terms

import { createAktorFunctionAsync } from '@operaide/aktor';
import type { PreprocessedQuery } from '@operaide/vector';
import type { ToAktorStrategy } from '@operaide/vector';

// 1. Pure TypeScript function
export const queryExpansionStrategy = async ({ 
    query 
}: { 
    query: string 
}): Promise<PreprocessedQuery> => {
    const originalQuery = query.trim();
    
    // Extract key terms
    const keyTerms = extractKeyTerms(originalQuery);
    
    // Get synonyms and related terms
    const expandedTerms = await getExpandedTerms(keyTerms);
    
    // Create expanded query
    const processedQuery = createExpandedQuery(originalQuery, expandedTerms);
    
    return {
        originalQuery,
        processedQuery,
        queryType: 'semantic',
        expandedTerms,
    };
};

// 2. Convert to Aktor strategy
export type QueryExpansionStrategy = typeof queryExpansionStrategy;
export type AktorQueryExpansionStrategy = ToAktorStrategy<QueryExpansionStrategy>;

export const aktorQueryExpansionStrategy: AktorQueryExpansionStrategy = createAktorFunctionAsync(
    'aktorQueryExpansionStrategy',
    queryExpansionStrategy
);

// Helper functions
function extractKeyTerms(query: string): string[] {
    // Remove stop words and extract important terms
    const stopWords = new Set(['the', 'a', 'an', 'and', 'or', 'but', 'in', 'on', 'at', 'to', 'for']);
    
    return query
        .toLowerCase()
        .split(/\s+/)
        .filter(term => term.length > 2 && !stopWords.has(term));
}

async function getExpandedTerms(keyTerms: string[]): Promise<string[]> {
    const expanded: string[] = [];
    
    for (const term of keyTerms) {
        // Get synonyms (could integrate with WordNet, thesaurus API, etc.)
        const synonyms = await getSynonyms(term);
        expanded.push(...synonyms);
        
        // Get related technical terms (for technical documentation)
        const relatedTerms = await getTechnicalRelatedTerms(term);
        expanded.push(...relatedTerms);
    }
    
    return [...new Set(expanded)]; // Remove duplicates
}

async function getSynonyms(term: string): Promise<string[]> {
    // Mock implementation - integrate with actual thesaurus API
    const synonymMap: Record<string, string[]> = {
        'authentication': ['login', 'signin', 'auth', 'credentials'],
        'database': ['db', 'storage', 'persistence', 'data store'],
        'function': ['method', 'procedure', 'routine', 'operation'],
        'user': ['account', 'profile', 'person', 'individual'],
    };
    
    return synonymMap[term.toLowerCase()] || [];
}

async function getTechnicalRelatedTerms(term: string): Promise<string[]> {
    // Mock implementation - could use domain-specific knowledge bases
    const techTermMap: Record<string, string[]> = {
        'authentication': ['oauth', 'jwt', 'session', 'token', 'password'],
        'database': ['sql', 'nosql', 'query', 'index', 'table'],
        'function': ['api', 'endpoint', 'call', 'invoke', 'execute'],
        'user': ['permission', 'role', 'access', 'identity'],
    };
    
    return techTermMap[term.toLowerCase()] || [];
}

function createExpandedQuery(originalQuery: string, expandedTerms: string[]): string {
    if (expandedTerms.length === 0) {
        return originalQuery;
    }
    
    // Add expanded terms with lower weight
    return `${originalQuery} ${expandedTerms.slice(0, 5).join(' ')}`;
}

Custom Cross-Encoder Reranking Strategy

Scenario: Use transformer models for precise relevance scoring

import { createAktorFunctionAsync } from '@operaide/aktor';
import type { FilteredResult, RankedResult, PreprocessedQuery } from '@operaide/vector';
import type { ToAktorStrategy } from '@operaide/vector';

// 1. Pure TypeScript function
export const crossEncoderRerankingStrategy = async ({ 
    query, 
    results 
}: { 
    query: PreprocessedQuery; 
    results: FilteredResult[]; 
}): Promise<RankedResult[]> => {
    const rerankingModel = 'cross-encoder/ms-marco-MiniLM-L-6-v2';
    const maxPairs = 100; // Limit for performance
    
    // Prepare query-document pairs for cross-encoder
    const queryDocumentPairs = results
        .slice(0, maxPairs)
        .map(result => ({
            query: query.processedQuery,
            document: result.content,
            originalResult: result,
        }));
    
    try {
        // Calculate relevance scores using cross-encoder
        const relevanceScores = await calculateCrossEncoderScores(
            queryDocumentPairs,
            rerankingModel
        );
        
        // Create ranked results
        const rankedResults: RankedResult[] = results.map((result, index) => ({
            ...result,
            relevanceScore: index < maxPairs ? relevanceScores[index] : result.similarity,
            rankingReason: index < maxPairs ? 'cross_encoder' : 'vector_similarity',
            rerankedAt: new Date().toISOString(),
        }));
        
        // Sort by relevance score descending
        rankedResults.sort((a, b) => b.relevanceScore - a.relevanceScore);
        
        return rankedResults;
        
    } catch (error) {
        console.error('Cross-encoder reranking failed:', error);
        
        // Fallback to similarity-based ranking
        return results.map(result => ({
            ...result,
            relevanceScore: result.similarity,
            rankingReason: 'fallback_similarity',
            rerankedAt: new Date().toISOString(),
        }));
    }
};

// 2. Convert to Aktor strategy
export type CrossEncoderRerankingStrategy = typeof crossEncoderRerankingStrategy;
export type AktorCrossEncoderRerankingStrategy = ToAktorStrategy<CrossEncoderRerankingStrategy>;

export const aktorCrossEncoderRerankingStrategy: AktorCrossEncoderRerankingStrategy = createAktorFunctionAsync(
    'aktorCrossEncoderRerankingStrategy',
    crossEncoderRerankingStrategy
);

// Helper functions
interface QueryDocumentPair {
    query: string;
    document: string;
    originalResult: FilteredResult;
}

async function calculateCrossEncoderScores(
    pairs: QueryDocumentPair[],
    model: string
): Promise<number[]> {
    // This would integrate with a transformer model service
    // For example: Hugging Face Inference API, local model server, etc.
    
    const inputs = pairs.map(pair => ({
        query: pair.query,
        document: truncateText(pair.document, 512), // Limit length for model
    }));
    
    const response = await fetch('/api/reranking/cross-encoder', {
        method: 'POST',
        headers: {
            'Content-Type': 'application/json',
        },
        body: JSON.stringify({
            model: model,
            inputs: inputs,
        }),
    });
    
    if (!response.ok) {
        throw new Error(`Cross-encoder API error: ${response.status}`);
    }
    
    const data = await response.json();
    return data.scores;
}

function truncateText(text: string, maxTokens: number): string {
    // Rough token estimation and truncation
    const estimatedTokens = Math.ceil(text.length / 4);
    
    if (estimatedTokens <= maxTokens) {
        return text;
    }
    
    const maxChars = maxTokens * 4;
    return text.substring(0, maxChars) + '...';
}

Usage in Pipelines

Using Custom Embedding Strategies

import { aktorVektorEmbeddingPipeline } from '@operaide/vector';
import { 
    aktorPdfDocumentLoaderStrategy,
    aktorSemanticChunkingStrategy,
    aktorOpenAIEmbeddingStrategy 
} from './custom-strategies';

// Use multiple custom strategies
const result = aktorVektorEmbeddingPipeline({
    file: pdfFile,
    client: dbClient,
    documentLoaderStrategy: aktorPdfDocumentLoaderStrategy,
    chunkingStrategy: aktorSemanticChunkingStrategy,
    embeddingStrategy: aktorOpenAIEmbeddingStrategy,
});

Using Custom Retrieval Strategies

import { aktorVektorRetrievalPipeline } from '@operaide/vector';
import { 
    aktorQueryExpansionStrategy,
    aktorCrossEncoderRerankingStrategy 
} from './custom-strategies';

// Use custom query expansion and reranking
const results = aktorVektorRetrievalPipeline({
    client: dbClient,
    query: userQuery,
    limit: 10,
    queryPreprocessingStrategy: aktorQueryExpansionStrategy,
    rerankingStrategy: aktorCrossEncoderRerankingStrategy,
});

Testing Custom Strategies

Unit Testing Pure Functions

import { describe, it, expect } from 'vitest';
import { semanticChunkingStrategy } from './custom-strategies';

describe('semanticChunkingStrategy', () => {
    it('should create semantic chunks with proper metadata', async () => {
        const mockDocument = {
            id: 'doc1',
            title: 'Test Document',
            content: 'First sentence. Second sentence. Third sentence.',
            metadata: { type: 'text', source: 'test.txt' }
        };
        
        const result = await semanticChunkingStrategy({ document: mockDocument });
        
        expect(result).toHaveLength(1);
        expect(result[0]).toMatchObject({
            documentId: 'doc1',
            content: expect.stringContaining('First sentence'),
            metadata: expect.objectContaining({
                chunkingMethod: 'semantic',
                coherenceScore: expect.any(Number),
            })
        });
    });
});

Integration Testing with Aktors

import { describe, it, expect } from 'vitest';
import { aktorSemanticChunkingStrategy } from './custom-strategies';

describe('aktorSemanticChunkingStrategy', () => {
    it('should work as an Aktor strategy', async () => {
        const mockDocument = /* test data */;
        
        const chunksAktor = aktorSemanticChunkingStrategy({ 
            document: () => mockDocument 
        });
        
        const result = await chunksAktor.get();
        
        expect(result).toBeDefined();
        expect(Array.isArray(result)).toBe(true);
    });
});

Best Practices

1. Error Handling

Always include try-catch blocks and fallback behavior:

export const robustStrategy = async ({ input }) => {
    try {
        return await primaryLogic(input);
    } catch (error) {
        console.error('Strategy failed:', error);
        return await fallbackLogic(input);
    }
};

2. Performance Optimization

Use batching and caching for expensive operations:

export const optimizedStrategy = async ({ chunks }) => {
    const batchSize = 100;
    const results = [];
    
    for (let i = 0; i < chunks.length; i += batchSize) {
        const batch = chunks.slice(i, i + batchSize);
        const batchResults = await processBatch(batch);
        results.push(...batchResults);
    }
    
    return results;
};

3. Configuration

Make strategies configurable:

export const createConfigurableStrategy = (config: StrategyConfig) => {
    return async ({ input }) => {
        // Use config to customize behavior
        return await processWithConfig(input, config);
    };
};

4. Type Safety

Always use proper TypeScript types:

export const typedStrategy: CustomStrategyType = async ({ input }) => {
    // TypeScript will enforce correct input/output types
    return await processTypedInput(input);
};

Next Steps

• View Embedding Pipeline Documentation →
• View Retrieval Pipeline Documentation →
• Explore Core Library APIs →