@aid-on/fractop

日本語 | English

FractoP (Fractal Processor) - Elegant text processing for LLMs with streaming, batching, and fractal chunking.

🚨 The Problem

LLMs have context limits. GPT-4 caps at 128K tokens. Claude at 200K. Even Gemini's 2M context fills up fast.

What happens when you need to:

• Summarize a 500-page PDF?
• Analyze a codebase with 10,000 files?
• Translate an entire book?
• Process millions of customer reviews?

// ❌ This fails
const summary = await llm.process(entire500PagePDF);
// Error: Context length exceeded (400,000 tokens > 128,000 limit)

✅ The Solution: FractoP

FractoP intelligently chunks your text, processes each piece, and merges results - all while preserving context.

// ✅ This works for ANY size document
const summary = await fractop()
  .withLLM(llm)
  .chunking({ size: 3000, overlap: 300 })
  .parallel(5)
  .run(entire500PagePDF);

✨ Features

• 🎯 Fluent API: Elegant chainable interface for building processing pipelines
• 🌊 Nagare Streaming: Reactive stream processing with Stream<T> integration
• 🔄 Smart Chunking: Intelligent text splitting with overlap for context preservation
• ⚡ Parallel Processing: Concurrent chunk processing for maximum performance
• 🛡️ Enterprise Reliability: Timeouts, retries, and circuit breaker patterns built-in
• 🎨 UnillM Integration: Works seamlessly with any LLM through UnillM adapters
• 📦 Batch Processing: Process multiple documents efficiently
• 🔁 Auto-retry: Exponential backoff for transient failures

Installation

npm install @aid-on/fractop

🚀 Quick Start

Primary Interface - Fluent API

The most elegant way to use FractoP:

import { fractop } from '@aid-on/fractop';

// Simple and elegant
const results = await fractop()
  .withLLM(async (chunk) => {
    // Your LLM logic here
    const response = await callYourLLM(chunk);
    return response;
  })
  .chunking({ size: 3000, overlap: 300 })
  .parallel(5)
  .retry(3, 1000)
  .timeout(30000)
  .run(longText);

With GROQ/OpenAI

const summaries = await fractop()
  .withLLM(async (chunk) => {
    const response = await fetch('https://api.groq.com/openai/v1/chat/completions', {
      method: 'POST',
      headers: {
        'Authorization': `Bearer ${GROQ_API_KEY}`,
        'Content-Type': 'application/json'
      },
      body: JSON.stringify({
        model: 'llama-3.1-8b-instant',
        messages: [
          { role: 'system', content: 'Summarize concisely.' },
          { role: 'user', content: chunk }
        ]
      })
    });
    const data = await response.json();
    return data.choices[0].message.content;
  })
  .chunking({ size: 2000, overlap: 200 })
  .run(document);

With UnillM

// UnillM configuration object
const results = await fractop()
  .withLLM({
    model: 'groq:llama-3.1-70b',
    credentials: { groqApiKey: process.env.GROQ_API_KEY },
    messages: (chunk) => [
      { role: 'system', content: 'Extract key points.' },
      { role: 'user', content: chunk }
    ],
    options: { temperature: 0.7 }
  })
  .chunking({ size: 3000 })
  .parallel(3)
  .run(text);

// Or with custom transform
const entities = await fractop<Entity[]>()
  .withLLM({
    model: 'anthropic:claude-3-5-haiku',
    credentials: { anthropicApiKey: process.env.ANTHROPIC_API_KEY },
    messages: (chunk) => [
      { role: 'system', content: 'Extract entities as JSON.' },
      { role: 'user', content: chunk }
    ],
    transform: (response) => JSON.parse(response.text)
  })
  .run(document);

🌊 Streaming with Nagare

Process large documents with memory-efficient streaming:

import { fractopStream } from '@aid-on/fractop';

// Stream results as they're processed
const stream = fractopStream(largeDocument)
  .withLLM(async (chunk) => await processChunk(chunk))
  .chunking({ size: 2000, overlap: 200 })
  .parallel(3)
  .stream();

// Reactive stream operations
await stream
  .map(result => result.toUpperCase())
  .filter(result => result.length > 100)
  .take(10)
  .collect();

Batch Processing

Process multiple documents efficiently:

import { fractopBatch } from '@aid-on/fractop';

const documents = ['doc1.txt', 'doc2.txt', 'doc3.txt'];

const results = await fractopBatch(documents)
  .withLLM(async (chunk) => await summarize(chunk))
  .chunking({ size: 2000 })
  .collectAll();

// Results is a Map<string, T[]>
for (const [doc, summaries] of results) {
  console.log(`${doc}: ${summaries.length} chunks processed`);
}

🔧 Advanced Features

Custom Processing Pipeline

const pipeline = await fractop<ExtractedEntity>()
  .withLLM(llmProcessor)
  .chunking({ size: 4000, overlap: 400 })
  .parallel(5)
  .retry(3, 2000)
  .timeout(60000, true)  // per-chunk timeout
  .minResults(50)
  .merge('simple')  // or provide custom merger
  .run(text);

Context-Aware Processing

const results = await fractop()
  .withLLM(llmProcessor)
  .context(async (text) => {
    // Generate global context from full document
    return await generateSummary(text.substring(0, 5000));
  })
  .process(async (chunk, context) => {
    // Process each chunk with context
    return await extractWithContext(chunk, context);
  })
  .merge((results) => customMergeLogic(results))
  .run(document);

Stream Processing

Process documents as reactive streams:

// Stream individual results
const stream = fractopStream(document)
  .withLLM(async (chunk) => await analyze(chunk))
  .chunking({ size: 2000, overlap: 200 })
  .stream();

// Use Nagare's reactive operators
const processed = await stream
  .map(result => transform(result))
  .filter(result => result.score > 0.8)
  .collect();

🏗️ Architecture

FractoP's fractal architecture enables processing of unlimited document sizes:

graph TD
    A[Input Text] --> B[Smart Chunking]
    B --> C[Parallel/Sequential Processing]
    C --> D[Stream Results]
    D --> E[Merge & Deduplicate]
    E --> F[Final Output]

Key Concepts

1. Fractal Chunking: Intelligent text splitting that preserves context
2. Stream Processing: Memory-efficient processing with Nagare streams
3. Context Propagation: Maintains document understanding across chunks
4. Result Merging: Smart deduplication and aggregation

📊 Performance

// Optimize for speed
const fast = await fractop()
  .withLLM(llm)
  .chunking({ size: 5000 })  // Larger chunks
  .parallel(10)               // High concurrency
  .run(text);

// Optimize for quality
const quality = await fractop()
  .withLLM(llm)
  .chunking({ size: 2000, overlap: 500 })  // More overlap
  .retry(5, 2000)                          // More retries
  .timeout(120000)                         // Longer timeout
  .run(text);

🛡️ Reliability Features

Automatic Retries

fractop()
  .withLLM(llm)
  .retry(3, 1000)  // 3 retries with exponential backoff
  .run(text);

Timeouts

fractop()
  .withLLM(llm)
  .timeout(60000)        // Overall timeout
  .timeout(5000, true)   // Per-chunk timeout
  .run(text);

Circuit Breaker

Automatically stops processing after consecutive failures to prevent cascade failures.

💡 Real-World Examples

📚 Summarize a 500-Page Research Paper

const paper = readFileSync('quantum-computing-thesis.pdf', 'utf-8');
// 200,000+ characters - would fail with direct LLM call

const summary = await fractop()
  .withLLM({
    model: 'groq:llama-3.1-70b',
    credentials: { groqApiKey: process.env.GROQ_API_KEY },
    messages: (chunk) => [
      { role: 'system', content: 'Summarize key findings. Be concise.' },
      { role: 'user', content: chunk }
    ]
  })
  .chunking({ size: 3000, overlap: 300 })
  .parallel(5)  // Process 5 chunks simultaneously
  .run(paper);

// Merge summaries into final document
const finalSummary = summary.join('\n\n');

🔍 Analyze 1000+ Files in a Codebase

const files = globSync('src/**/*.ts');  // 1000+ TypeScript files
const fullCode = files.map(f => readFileSync(f)).join('\n');
// Millions of characters - impossible with single LLM call

// Extract all API endpoints
const endpoints = await fractop()
  .withLLM({
    model: 'anthropic:claude-3-5-haiku',
    credentials: { anthropicApiKey: API_KEY },
    messages: (chunk) => [
      { role: 'system', content: 'Extract REST API endpoints as JSON.' },
      { role: 'user', content: chunk }
    ],
    transform: (res) => JSON.parse(res.text)
  })
  .chunking({ size: 4000, overlap: 500 })  // Overlap prevents missing endpoints
  .parallel(10)  // Analyze 10 files simultaneously
  .run(fullCode);

// Deduplicate results
const uniqueEndpoints = [...new Set(endpoints.flat())];
console.log(`Found ${uniqueEndpoints.length} API endpoints`);

🌐 Translate an Entire Book

const book = await fetch('https://gutenberg.org/files/2600/2600-0.txt')
  .then(r => r.text());  // War and Peace - 3.2 million characters!

const translatedBook = await fractop()
  .withLLM({
    model: 'gemini:gemini-2.5-pro',
    credentials: { geminiApiKey: process.env.GEMINI_API_KEY },
    messages: (chunk) => [
      { role: 'system', content: 'Translate to Japanese. Keep literary style.' },
      { role: 'user', content: chunk }
    ]
  })
  .chunking({ 
    size: 2000,      // Smaller chunks for quality
    overlap: 200     // Preserve sentence flow
  })
  .retry(3, 2000)    // Retry failed chunks
  .timeout(120000)   // 2 min timeout per chunk
  .run(book);

writeFileSync('war-and-peace-ja.txt', translatedBook.join(''));

📊 Process 50,000 Customer Reviews

// 50,000 support tickets from database
const tickets = await db.query('SELECT * FROM tickets LIMIT 50000');
const ticketTexts = tickets.map(t => t.content);

// Process in batches with streaming
const analysis = await fractopBatch(ticketTexts)
  .withLLM({
    model: 'groq:llama-3.1-8b-instant',  // Fast model for high volume
    credentials: { groqApiKey: API_KEY },
    messages: (ticket) => [
      { role: 'system', content: 'Output: sentiment|category|priority' },
      { role: 'user', content: ticket }
    ],
    transform: (res) => {
      const [sentiment, category, priority] = res.text.split('|');
      return { sentiment, category, priority };
    }
  })
  .collectAll();

// Aggregate insights
const insights = {
  sentiments: { positive: 0, negative: 0, neutral: 0 },
  categories: new Map(),
  highPriority: []
};

for (const [ticket, results] of analysis) {
  results.forEach(r => {
    insights.sentiments[r.sentiment]++;
    if (r.priority === 'high') insights.highPriority.push(ticket);
  });
}

🤖 Generate Tests for Large Components

const component = readFileSync('src/Dashboard.tsx', 'utf-8');
// 5000-line React component

const tests = await fractop()
  .withLLM({
    model: 'openai:gpt-4o',
    credentials: { openaiApiKey: process.env.OPENAI_API_KEY },
    messages: (chunk) => [
      { role: 'system', content: 'Generate Jest unit tests with React Testing Library.' },
      { role: 'user', content: chunk }
    ]
  })
  .chunking({ size: 1500 })  // Each chunk gets targeted tests
  .parallel(3)
  .run(component);

// Combine into test suite
const testFile = `
describe('Dashboard Component', () => {
  ${tests.join('\n\n')}
});
`;
writeFileSync('Dashboard.test.tsx', testFile);

💬 Real-time Document Q&A

async function askDocument(doc: string, question: string) {
  // Stream through document to find answers
  return await fractopStream(doc)
    .withLLM({
      model: 'gemini:gemini-2.5-flash',  // Fast for real-time
      credentials: { geminiApiKey: API_KEY },
      messages: (chunk) => [
        { role: 'user', content: `Answer "${question}" from: ${chunk}` }
      ]
    })
    .chunking({ size: 3000, overlap: 500 })
    .stream()
    .filter(answer => answer.length > 20)  // Filter relevant answers
    .take(3)  // First 3 good answers
    .collect();
}

// Usage
const manual = readFileSync('kubernetes-manual.txt', 'utf-8');
const answers = await askDocument(manual, "How to set up auto-scaling?");
// Returns in seconds, not minutes!

🎯 Common Use Cases

Document Summarization

const summary = await fractop()
  .withLLM(async (chunk) => await summarize(chunk))
  .chunking({ size: 3000, overlap: 300 })
  .merge(results => results.join('\n'))
  .run(document);

Keyword Extraction

const keywords = await fractop<string[]>()
  .withLLM(async (chunk) => await extractKeywords(chunk))
  .chunking({ size: 2000 })
  .merge(results => [...new Set(results.flat())])
  .run(document);

Translation

const translated = await fractop()
  .withLLM(async (chunk) => await translate(chunk, 'ja'))
  .chunking({ size: 2000, overlap: 200 })
  .run(document);

Question Answering

const answers = await fractop()
  .withLLM(async (chunk) => await answerQuestion(chunk, question))
  .chunking({ size: 3000 })
  .parallel(5)
  .run(document);

⚙️ Configuration

Default Settings

{
  chunkSize: 3000,        // Optimized for LLM token limits
  overlapSize: 300,       // Context preservation
  concurrency: 3,         // Parallel processing threads
  maxRetries: 2,          // Retry attempts
  retryDelay: 1000,       // Initial retry delay (ms)
  chunkTimeout: 30000     // Per-chunk timeout (ms)
}

Advanced Configuration

const processor = fractop()
  .withLLM(llm)
  .chunking({ 
    size: 4000,      // Larger chunks for better context
    overlap: 500     // More overlap for continuity
  })
  .parallel(5)       // 5 concurrent processors
  .retry(3, 2000)    // 3 retries, 2s initial delay
  .timeout(60000)    // 1 minute overall timeout
  .minResults(100)   // Minimum 100 results
  .build();

📦 API Reference

Main Exports

// Primary fluent interface
import { fractop } from '@aid-on/fractop';

// Streaming interfaces
import { fractopStream, fractopBatch } from '@aid-on/fractop';

// Core processor (advanced use)
import { FractalProcessor } from '@aid-on/fractop';

// Utilities
import { simpleMerge, weightedMerge } from '@aid-on/fractop';

Fluent API Methods

| Method | Description | |--------|-------------| | .withLLM(fn\|config) | Set LLM processor (function or UnillM config) | | .chunking(opts) | Configure chunk size and overlap | | .parallel(n) | Enable parallel processing | | .retry(n, delay) | Configure retry behavior | | .timeout(ms, perChunk?) | Set timeout limits | | .context(fn) | Set context generator | | .process(fn) | Set chunk processor | | .merge(strategy) | Set merge strategy | | .minResults(n) | Set minimum result count | | .run(text) | Execute processing |

🔬 TypeScript

Full TypeScript support with generics:

interface Analysis {
  sentiment: 'positive' | 'negative' | 'neutral';
  score: number;
  keywords: string[];
}

const results = await fractop<Analysis>()
  .withLLM(async (chunk): Promise<Analysis> => {
    // Type-safe processing
    return analyzeChunk(chunk);
  })
  .merge((results: Analysis[][]) => {
    // Type-safe merging
    return combineAnalyses(results);
  })
  .run(document);

// results is Analysis[]

🚀 Performance Tips

1. Chunk Size: Balance between context and token limits (2000-4000 chars recommended)
2. Overlap: 10-20% of chunk size for good context preservation
3. Concurrency: Match your LLM rate limits (3-5 for most providers)
4. Streaming: Use fractopStream for documents > 100KB
5. Batching: Use fractopBatch for multiple documents

License

MIT