Skip to content
BranchesTags

Repository files navigation

The First Info Bender Logo

The First Info Bender

Intelligent Multi-Stage Data Processing Framework

A general-purpose, highly configurable data processing framework that transforms diverse data inputs into structured, categorized, and relevance-filtered information through multi-stage LLM processing.

Table of Contents

Overview

Organizations receive data from multiple sources in varying formats. This framework solves the challenge of:

  1. Normalizing diverse inputs into consistent structure
  2. Categorizing content based on domain taxonomy
  3. Filtering for relevance to specific use cases

The framework is wildly configurable - adaptable to any domain (marine science, finance, legal, etc.) through configuration alone, without code changes.

Key Features

🎯 3-Layer Processing Pipeline

  1. Layer 1 (Ingestion): Handles 4 input types

    • Structured Single (e.g., one API object)
    • Unstructured Single (e.g., one email)
    • Structured Multiple (e.g., API array of 50 papers)
    • Unstructured Multiple (e.g., HTML with embedded articles)

  2. Layer 2 (Structuring): LLM-based extraction & categorization

    • Input-type-specific prompts
    • Configurable LLM settings per input type
    • Optional parallel validation (majority voting)

  3. Layer 3 (Relevance Filtering): LLM-based relevance determination

    • Batch processing with configurable triggers
    • Auto-rules for common patterns
    • Quarantine system for uncertain cases

🔄 LLM Validation Through Parallel Calls

  • Make N parallel LLM calls for critical decisions
  • Majority voting, unanimous, or 2/3 agreement
  • Reduces inconsistency in categorization
  • Configurable per-stage and per-input-type

📊 Complete Traceability

Every operation is logged:

  • Raw data backups
  • Exact prompts sent
  • All LLM responses (including parallel calls)
  • Agreement analysis
  • API costs
  • Processing durations

For any database entry, you can trace:

  • Where it came from
  • What prompts processed it
  • What the LLM decided (with voting record)
  • Why it was categorized/filtered as such

⚙️ Configuration-Driven

  • No code changes for new domains
  • Different LLM settings per stage
  • Pluggable prompt templates
  • Flexible category taxonomies
  • Auto-rules and post-processing

🛡️ Fault Tolerance

  • Raw data always preserved
  • Configurable retry logic
  • Failed items quarantined
  • Duplicate detection
  • Graceful degradation

Architecture

┌─────────────────────────────────────────────────────────────┐
│                    LAYER 1: INGESTION                       │
│  ┌─────────────┐  ┌──────────────┐  ┌──────────────────┐   │
│  │  API Fetch  │  │ Web Scraping │  │ Webhook Receipt │   │
│  └─────────────┘  └──────────────┘  └──────────────────┘   │
│         │                │                    │              │
│         └────────────────┴────────────────────┘              │
│                          │                                   │
│              Input Classification (4 types)                  │
│                          │                                   │
│               Store Raw Data + Metadata                      │
└──────────────────────────┬──────────────────────────────────┘
                           │
┌──────────────────────────┴──────────────────────────────────┐
│              LAYER 2: LLM STRUCTURING                       │
│                                                              │
│  ┌────────────────────────────────────────────────────┐    │
│  │  Input-Type-Specific Prompt Templates              │    │
│  └────────────────────────────────────────────────────┘    │
│                          │                                   │
│  ┌────────────────────────────────────────────────────┐    │
│  │  LLM Call(s) with Optional Validation              │    │
│  │  - Single call OR                                   │    │
│  │  - N parallel calls + majority voting              │    │
│  └────────────────────────────────────────────────────┘    │
│                          │                                   │
│  ┌────────────────────────────────────────────────────┐    │
│  │  Extract Fields + Assign Categories                │    │
│  └────────────────────────────────────────────────────┘    │
│                          │                                   │
│         Store Structured Entries + Processing Log           │
└──────────────────────────┬──────────────────────────────────┘
                           │
┌──────────────────────────┴──────────────────────────────────┐
│            LAYER 3: RELEVANCE FILTERING                     │
│                                                              │
│  Triggered when: N new entries OR max wait time             │
│                          │                                   │
│  ┌────────────────────────────────────────────────────┐    │
│  │  Apply Auto-Rules (optional)                       │    │
│  │  - Auto-relevant patterns                          │    │
│  │  - Auto-exclude patterns                           │    │
│  └────────────────────────────────────────────────────┘    │
│                          │                                   │
│  ┌────────────────────────────────────────────────────┐    │
│  │  LLM Relevance Evaluation                          │    │
│  │  - Single call OR                                   │    │
│  │  - N parallel calls + majority voting              │    │
│  └────────────────────────────────────────────────────┘    │
│                          │                                   │
│  ┌────────────────────────────────────────────────────┐    │
│  │  Mark: Relevant / Not Relevant / Quarantined       │    │
│  └────────────────────────────────────────────────────┘    │
│                          │                                   │
│            Update Entries + Processing Log                   │
└──────────────────────────┬──────────────────────────────────┘
                           │
                    ┌──────┴──────┐
                    │   DATABASE  │
                    │  + LINEAGE  │
                    └─────────────┘

Installation

Prerequisites

  • Python 3.8+
  • PostgreSQL (or SQLite for development)
  • Access to LLM Router (bkvy) at http://localhost:10006

Setup

  1. Clone the repository:

    git clone https://github.com/NitroxHead/thefirstinfobender.git
cd thefirstinfobender

  2. Install dependencies:

    pip install -r requirements.txt

  3. Configure environment variables:

    cp .env.example .env
# Edit .env with your API keys and database password

  4. Set up database:

    For PostgreSQL:

    createdb thefirstinfobender

    For SQLite (development):

    # Edit config/database.yaml to use SQLite
# Database will be created automatically

  5. Configure your domain:

    Edit configuration files in config/:

    • sources.yaml - Define your data sources
    • categories.yaml - Define your category taxonomy
    • layer2_config.yaml - Configure structuring prompts
    • layer3_config.yaml - Configure relevance criteria

Quick Start

Process a single source:

python main.py process-source job_api_single

Process all enabled sources:

python main.py process-all

Manually trigger relevance filtering:

python main.py run-layer3 --batch-size 20

View statistics:

python main.py stats

Configuration

1. Data Sources (config/sources.yaml)

Define where data comes from:

sources:
  my_api_source:
    enabled: true
    source_type: "api"
    input_type: "structured_multiple"
    description: "Research papers API"
    fetch_config:
      url: "https://api.example.com/papers"
      method: "GET"
      headers:
        X-API-Key: "${RESEARCH_API_KEY}"
      polling_interval_seconds: 3600
    layer2_config_name: "paper_structuring"

2. Categories (config/categories.yaml)

Define your domain taxonomy:

categories:
  RESEARCH:
    description: "Research positions and projects"
    subcategories:
      PHD:
        description: "PhD positions"
      POSTDOC:
        description: "Postdoctoral positions"

3. Layer 2 Configuration (config/layer2_config.yaml)

Configure LLM structuring:

configurations:
  paper_structuring:
    input_type: "structured_multiple"

    llm_settings:
      endpoint: "http://localhost:10006/llm/intelligence"
      intelligence_level: "medium"
      max_wait_seconds: 30
      options:
        max_tokens: 2000
        temperature: 0.3

    llm_validation:
      enabled: true
      parallel_calls: 3
      agreement_threshold: "majority"
      compare_fields:
        - "category"
        - "subcategory"

    prompt_template: |
      Extract and categorize research papers.

      INPUT: {raw_data}
      CATEGORIES: {categories}

      Output JSON with fields: title, authors, category, etc.

4. Layer 3 Configuration (config/layer3_config.yaml)

Configure relevance filtering:

use_case:
  name: "student_opportunities"
  description: "Opportunities for students and ECRs"

trigger:
  batch_size: 10
  max_wait_seconds: 300

llm_validation:
  enabled: true
  parallel_calls: 5
  agreement_threshold: "majority"

relevance_criteria:
  target_audience:
    - "Students"
    - "Early-career researchers"
  exclude:
    - "Senior positions (>5 years experience)"
    - "Faculty positions"

Usage

Command-Line Interface

# Process specific source
python main.py process-source <source_id>

# Process all sources
python main.py process-all

# Run Layer 3 batch
python main.py run-layer3 --batch-size 20

# View statistics
python main.py stats

# Set log level
python main.py --log-level DEBUG process-source my_source

Programmatic Usage

from src.pipeline import PipelineOrchestrator
from src.logging import setup_logging

# Initialize
setup_logging(log_level="INFO")
pipeline = PipelineOrchestrator(config_dir="config")

# Process source with custom data
custom_data = {"title": "Marine Biology PhD", ...}
result = pipeline.process_source("job_api_single", data=custom_data)

# Run Layer 3
layer3_result = pipeline.run_layer3_batch(batch_size=20)

# Shutdown
pipeline.shutdown()

See example_usage.py for more examples.

Logging & Traceability

What Gets Logged

Layer 1 (Ingestion):

  • Source ID and fetch timestamp
  • Raw data (full backup)
  • Input type classification
  • Content hash (deduplication)

Layer 2 (Structuring):

  • Prompt sent to LLM (exact text)
  • LLM settings used
  • If validation enabled:
    • All N parallel responses
    • Agreement analysis
    • Consensus decision
  • Model used, processing time, API cost

Layer 3 (Relevance):

  • Prompt sent to LLM
  • All parallel responses (if validation enabled)
  • Relevance decision + reasoning
  • Auto-rules applied
  • Processing time, API cost

System Events:

  • Pipeline starts/stops
  • Configuration changes
  • Errors and exceptions

Database Schema

Key tables:

  • raw_data - All raw ingested data
  • entries - Structured entries with categories
  • processing_log - Complete processing history
  • system_log - System events

Every entry has full lineage traceable back to source.

Examples

Example 1: Process Custom Job Posting

from src.pipeline import PipelineOrchestrator

pipeline = PipelineOrchestrator()

job_data = {
    "title": "Marine Biologist Position",
    "organization": "Ocean Institute",
    "requirements": ["PhD in marine biology", "5 years experience"],
    "deadline": "2025-06-30"
}

result = pipeline.process_source("job_api_single", data=job_data)
print(f"Entry created: {result['layer2']['entry_ids']}")

Example 2: Batch Process for Relevance

pipeline = PipelineOrchestrator()

# Process 50 entries
result = pipeline.run_layer3_batch(batch_size=50)

print(f"Relevant: {result['relevant']}")
print(f"Not relevant: {result['not_relevant']}")
print(f"Needs review: {result['quarantined']}")

Example 3: LLM Validation

Configuration with 5 parallel calls:

llm_validation:
  enabled: true
  parallel_calls: 5
  agreement_threshold: "majority"
  compare_fields:
    - "category"
    - "is_relevant"

If 3+ calls agree: accept consensus If no agreement: quarantine for manual review

Advanced Features

Parallel LLM Validation

Reduce inconsistency through redundant calls:

llm_validation:
  enabled: true
  parallel_calls: 3  # or 5, 7, etc.
  agreement_threshold: "majority"  # or "unanimous", "2/3"
  compare_fields:
    - "category"
    - "subcategory"

How it works:

  1. Send same prompt to LLM N times in parallel
  2. Compare responses on specified fields
  3. If agreement threshold met: accept consensus
  4. If no agreement: flag for manual review

Auto-Rules

Skip LLM calls for obvious cases:

post_processing:
  auto_relevant_rules:
    - field: "category"
      value: "EDUCATION.INTERNSHIPS"
      reason: "All internships relevant for students"

  auto_exclude_rules:
    - field: "experience_level"
      value: "senior"
      reason: "Senior positions not for students"

Quarantine System

Flag uncertain cases for review:

quarantine_rules:
  - field: "confidence"
    operator: "<"
    value: 0.6
    reason: "Low confidence decision"

Deduplication

Automatic duplicate detection via content hashing:

deduplication:
  enabled: true
  strategy: "content_hash"
  fields_to_hash:
    - "title"
    - "description"

Database Schema

raw_data

  • Stores all ingested data
  • Content hash for deduplication
  • Processing status tracking

entries

  • Structured, categorized entries
  • Flexible JSON for domain-specific fields
  • Relevance status and score

processing_log

  • Complete audit trail
  • All prompts and responses
  • Validation details (parallel calls, voting)
  • Performance metrics

system_log

  • System events
  • Configuration changes
  • Error tracking

Adapting to Your Domain

The framework is designed to be domain-agnostic. To adapt:

  1. Define your categories (config/categories.yaml)
  2. Configure your sources (config/sources.yaml)
  3. Write prompt templates (config/layer2_config.yaml)
  4. Set relevance criteria (config/layer3_config.yaml)

No code changes required!

Example Domains

  • Legal Documents: Categorize by case type, jurisdiction, relevance to practice area
  • Financial News: Structure market updates, filter by portfolio relevance
  • Scientific Literature: Extract metadata, categorize by field, filter by research focus
  • Job Postings: Structure requirements, categorize by industry/level, filter by candidate profile

Performance Considerations

  • LLM Validation Cost: Parallel calls multiply costs. Use selectively for critical decisions.
  • Batch Size: Larger Layer 3 batches improve throughput but increase latency.
  • Intelligence Level: Use "low" for simple tasks, "high" only when needed.
  • Deduplication: Reduces redundant processing.

Troubleshooting

Database connection fails:

  • Check config/database.yaml settings
  • Verify database exists and credentials are correct
  • Check .env file for DB_PASSWORD

LLM calls failing:

  • Verify LLM router is running at http://localhost:10006
  • Check /health endpoint
  • Review LLM router logs

No agreement in validation:

  • Review prompt clarity
  • Check if task is too subjective
  • Consider lowering threshold or reducing parallel calls
  • Review quarantined items in database

Processing too slow:

  • Disable validation for simple tasks
  • Reduce batch sizes
  • Use lower intelligence levels
  • Check database query performance

Contributing

This framework is designed for extensibility:

  • Custom input classifiers
  • Plugin prompt templates
  • Additional post-processing rules
  • Alternative storage backends
  • Custom LLM providers

License

MIT License

Acknowledgments

Built to integrate with the bkvy LLM Router for intelligent, cost-optimized LLM routing.

Questions? Check the examples or review the configuration files for detailed options.

About

Domain-agnostic LLM pipeline framework for ingesting, structuring, categorizing, and relevance-filtering data from any source; fully configurable, with parallel validation and complete processing traceability.

Resources

Readme
MIT license
Activity

Stars

0 stars

Watchers

0 watching

Forks

0 forks
Report repository

Releases

Packages

Contributors

Languages