Janady13/AI_Brain / the_tired_agi/core/src/ingest_pipeline.cpp

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/AI_Brain / the_tired_agi/core/src/ingest_pipeline.cpp
Date	Not specified in source
Repository	Janady13/AI_Brain
Revision	d6c309d3fa439714ca46b9d3bb94caf478ea5376
Branch	main
Path or archive member	the_tired_agi/core/src/ingest_pipeline.cpp
Original SHA-256	53f02c63ecb1ea0d0a036b72b8e087a939ef4477cb173f854190a16cadb1db7a
Git blob SHA	38765cc255b0398450f4f394bd6b70a5e89a290c
Reading-copy SHA-256	53f02c63ecb1ea0d0a036b72b8e087a939ef4477cb173f854190a16cadb1db7a
Original source ↗
Every recorded occurrence

These 2 records contain identical original source bytes. Their paths and revisions remain part of the archive.
Original record	Branch or archive	Revision / identifier	Original location
Janady13/AI_Brain / the_tired_agi/core/src/ingest_pipeline.cpp	main	d6c309d3fa439714ca46b9d3bb94caf478ea5376	Open original
Janady13/AI_Brain / the_tired_agi/core/src/ingest_pipeline.cpp	claude/binary-helix-fibonacci-hive-011CUj7dx1WWJLVYcWCjBbYv	527d4882c71eb041393f9d8945996177c5ac9e86	Open original
Recovered source transcript

Download complete source text ↓618 lines · 19.2 KB
// ==============================================================================
// The-Tired-AGI Core: Ingest Pipeline Implementation
// Read-Only Text Processing & Embedding Generation
// ==============================================================================

#include "ingest_pipeline.hpp"
#include <fstream>
#include <cstring>
#include <cmath>
#include <algorithm>
#include <execution>
#include <numeric>
#include <chrono>
#include <sys/mman.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>

namespace tired_agi {

// ==============================================================================
// Constructor / Destructor
// ==============================================================================

IngestPipeline::IngestPipeline(IngestConfig config) noexcept
    : config_(config)
{
    stat_start_time_ms_.store(get_time_ms(), std::memory_order_relaxed);
}

IngestPipeline::~IngestPipeline() = default;

// ==============================================================================
// File Processing
// ==============================================================================

std::vector<Embedding> IngestPipeline::process_file(const std::filesystem::path& path) noexcept {
    std::vector<Embedding> embeddings;

    stat_files_scanned_.fetch_add(1, std::memory_order_relaxed);

    // Check if file is allowed
    if (!is_allowed_file(path)) {
        report_error("process_file", "File extension not allowed: " + path.string());
        return embeddings;
    }

    // Check file size
    std::error_code ec;
    auto file_size = std::filesystem::file_size(path, ec);
    if (ec) {
        report_error("process_file", "Cannot get file size: " + path.string());
        return embeddings;
    }

    if (file_size > config_.max_file_size_mb * 1024 * 1024) {
        report_error("process_file", "File too large: " + path.string());
        return embeddings;
    }

    // Check if binary
    if (config_.skip_binary_files && is_binary_file(path)) {
        return embeddings;  // Skip silently
    }

    // Read and chunk
    auto chunks = read_and_chunk(path);
    if (chunks.empty()) {
        report_error("process_file", "No chunks generated: " + path.string());
        return embeddings;
    }

    // Generate embeddings
    embeddings = chunks_to_embeddings(chunks);

    stat_files_processed_.fetch_add(1, std::memory_order_relaxed);

    return embeddings;
}

std::vector<Embedding> IngestPipeline::process_directory(
    const std::filesystem::path& dir,
    bool recursive
) noexcept {
    std::vector<Embedding> all_embeddings;
    std::vector<std::filesystem::path> files;

    // Collect all files
    try {
        if (recursive) {
            for (const auto& entry : std::filesystem::recursive_directory_iterator(dir)) {
                if (entry.is_regular_file()) {
                    if (config_.skip_hidden_files && entry.path().filename().string()[0] == '.') {
                        continue;
                    }
                    files.push_back(entry.path());
                }
            }
        } else {
            for (const auto& entry : std::filesystem::directory_iterator(dir)) {
                if (entry.is_regular_file()) {
                    if (config_.skip_hidden_files && entry.path().filename().string()[0] == '.') {
                        continue;
                    }
                    files.push_back(entry.path());
                }
            }
        }
    } catch (const std::exception& e) {
        report_error("process_directory", std::string("Directory iteration failed: ") + e.what());
        return all_embeddings;
    }

    // Process files in parallel
    all_embeddings = process_files_parallel(files);

    return all_embeddings;
}

std::vector<Embedding> IngestPipeline::process_files_parallel(
    const std::vector<std::filesystem::path>& paths
) noexcept {
    std::vector<std::vector<Embedding>> per_file_embeddings(paths.size());

    // Process in parallel
    std::vector<size_t> indices(paths.size());
    std::iota(indices.begin(), indices.end(), 0);

    std::for_each(
        std::execution::par_unseq,
        indices.begin(),
        indices.end(),
        [&](size_t i) {
            per_file_embeddings[i] = process_file(paths[i]);
        }
    );

    // Flatten results
    std::vector<Embedding> all_embeddings;
    for (const auto& file_embs : per_file_embeddings) {
        all_embeddings.insert(all_embeddings.end(), file_embs.begin(), file_embs.end());
    }

    return all_embeddings;
}

// ==============================================================================
// Low-Level Processing
// ==============================================================================

std::vector<TextChunk> IngestPipeline::read_and_chunk(const std::filesystem::path& path) noexcept {
    std::vector<TextChunk> chunks;

    if (config_.use_mmap) {
        // Memory-mapped I/O (faster for large files)
        int fd = open(path.c_str(), O_RDONLY);
        if (fd < 0) {
            report_error("read_and_chunk", "Cannot open file: " + path.string());
            return chunks;
        }

        struct stat sb;
        if (fstat(fd, &sb) < 0) {
            close(fd);
            report_error("read_and_chunk", "Cannot stat file: " + path.string());
            return chunks;
        }

        size_t file_size = sb.st_size;
        if (file_size == 0) {
            close(fd);
            return chunks;
        }

        void* mapped = mmap(nullptr, file_size, PROT_READ, MAP_PRIVATE, fd, 0);
        if (mapped == MAP_FAILED) {
            close(fd);
            report_error("read_and_chunk", "Cannot mmap file: " + path.string());
            return chunks;
        }

        // Advise sequential access
        madvise(mapped, file_size, MADV_SEQUENTIAL);

        // Chunk the data
        chunks = chunk_data(
            static_cast<const uint8_t*>(mapped),
            file_size,
            path.string(),
            0
        );

        munmap(mapped, file_size);
        close(fd);

        stat_bytes_read_.fetch_add(file_size, std::memory_order_relaxed);

    } else {
        // Standard file I/O
        std::ifstream file(path, std::ios::binary);
        if (!file) {
            report_error("read_and_chunk", "Cannot open file: " + path.string());
            return chunks;
        }

        // Read entire file
        file.seekg(0, std::ios::end);
        size_t file_size = file.tellg();
        file.seekg(0, std::ios::beg);

        std::vector<uint8_t> buffer(file_size);
        file.read(reinterpret_cast<char*>(buffer.data()), file_size);

        if (!file) {
            report_error("read_and_chunk", "Read error: " + path.string());
            return chunks;
        }

        chunks = chunk_data(buffer.data(), buffer.size(), path.string(), 0);

        stat_bytes_read_.fetch_add(file_size, std::memory_order_relaxed);
    }

    return chunks;
}

std::vector<TextChunk> IngestPipeline::chunk_data(
    const uint8_t* data,
    size_t size,
    const std::string& source_file,
    uint64_t file_offset
) noexcept {
    std::vector<TextChunk> chunks;

    size_t pos = 0;
    uint64_t chunk_id = 0;

    while (pos < size) {
        TextChunk chunk;

        // Determine chunk size
        size_t remaining = size - pos;
        size_t chunk_size = std::min(remaining, config_.chunk_size);

        if (chunk_size < config_.min_chunk_size && pos > 0) {
            // Skip small trailing chunk, merge with previous
            break;
        }

        // Copy data
        std::memcpy(chunk.data.data(), data + pos, chunk_size);
        chunk.size = chunk_size;
        chunk.chunk_id = chunk_id++;
        chunk.source_file = source_file;
        chunk.file_offset = file_offset + pos;
        chunk.timestamp = get_time_ms();

        // Compute hash
        chunk.hash = compute_hash(chunk.data.data(), chunk.size);

        // Compute entropy
        chunk.entropy = compute_entropy(chunk.data.data(), chunk.size);

        // Filter low-entropy chunks
        if (chunk.entropy < config_.min_entropy_threshold) {
            pos += config_.chunk_size - config_.chunk_overlap;
            continue;
        }

        chunks.push_back(chunk);
        update_statistics(chunk);

        // Invoke callback if set
        if (chunk_callback_) {
            chunk_callback_(chunk);
        }

        // Advance position with overlap
        pos += config_.chunk_size - config_.chunk_overlap;
    }

    return chunks;
}

Embedding IngestPipeline::chunk_to_embedding(const TextChunk& chunk) noexcept {
    return simple_embedding_generation(chunk);
}

std::vector<Embedding> IngestPipeline::chunks_to_embeddings(
    const std::vector<TextChunk>& chunks
) noexcept {
    std::vector<Embedding> embeddings(chunks.size());

    // Parallel embedding generation
    std::vector<size_t> indices(chunks.size());
    std::iota(indices.begin(), indices.end(), 0);

    std::for_each(
        std::execution::par_unseq,
        indices.begin(),
        indices.end(),
        [&](size_t i) {
            embeddings[i] = chunk_to_embedding(chunks[i]);
        }
    );

    return embeddings;
}

// ==============================================================================
// Utilities
// ==============================================================================

uint64_t IngestPipeline::compute_hash(const uint8_t* data, size_t size) noexcept {
    return XXHash64::hash(data, size, 0);
}

double IngestPipeline::compute_entropy(const uint8_t* data, size_t size) noexcept {
    if (size == 0) return 0.0;

    // Build frequency histogram (256 bins for bytes)
    std::array<uint64_t, 256> histogram{};
    for (size_t i = 0; i < size; ++i) {
        histogram[data[i]]++;
    }

    // Compute Shannon entropy: H = -Σ p_i * log2(p_i)
    double entropy = 0.0;
    double size_inv = 1.0 / static_cast<double>(size);

    for (uint64_t count : histogram) {
        if (count > 0) {
            double p = static_cast<double>(count) * size_inv;
            entropy -= p * std::log2(p);
        }
    }

    // Normalize to [0,1] (max entropy for uniform distribution is log2(256) = 8)
    return entropy / 8.0;
}

bool IngestPipeline::is_binary_file(const std::filesystem::path& path) noexcept {
    // Read first 8KB and check for binary content
    std::ifstream file(path, std::ios::binary);
    if (!file) return true;  // Assume binary if can't open

    constexpr size_t SAMPLE_SIZE = 8192;
    std::array<uint8_t, SAMPLE_SIZE> buffer;
    file.read(reinterpret_cast<char*>(buffer.data()), SAMPLE_SIZE);
    size_t bytes_read = file.gcount();

    if (bytes_read == 0) return false;  // Empty file is text

    // Count non-printable characters
    size_t non_printable = 0;
    for (size_t i = 0; i < bytes_read; ++i) {
        uint8_t c = buffer[i];
        // Allow common text characters: printable ASCII, newline, tab, carriage return
        if (c < 32 && c != '\n' && c != '\t' && c != '\r') {
            non_printable++;
        }
        if (c >= 127 && c < 160) {
            non_printable++;
        }
    }

    // If more than 10% non-printable, consider binary
    return (static_cast<double>(non_printable) / bytes_read) > 0.10;
}

bool IngestPipeline::is_allowed_file(const std::filesystem::path& path) const noexcept {
    std::string ext = path.extension().string();
    return std::find(
        config_.allowed_extensions.begin(),
        config_.allowed_extensions.end(),
        ext
    ) != config_.allowed_extensions.end();
}

// ==============================================================================
// Statistics
// ==============================================================================

IngestStatistics IngestPipeline::get_statistics() const noexcept {
    IngestStatistics stats;
    stats.total_files_scanned = stat_files_scanned_.load(std::memory_order_relaxed);
    stats.total_files_processed = stat_files_processed_.load(std::memory_order_relaxed);
    stats.total_bytes_read = stat_bytes_read_.load(std::memory_order_relaxed);
    stats.total_chunks_generated = stat_chunks_generated_.load(std::memory_order_relaxed);
    stats.total_embeddings_generated = stat_embeddings_generated_.load(std::memory_order_relaxed);
    stats.total_errors = stat_errors_.load(std::memory_order_relaxed);

    if (stats.total_chunks_generated > 0) {
        stats.average_entropy = stat_sum_entropy_.load(std::memory_order_relaxed) / stats.total_chunks_generated;
        stats.average_chunk_size = stat_sum_chunk_size_.load(std::memory_order_relaxed) / stats.total_chunks_generated;
    }

    uint64_t start_time = stat_start_time_ms_.load(std::memory_order_relaxed);
    uint64_t current_time = get_time_ms();
    stats.elapsed_time_ms = current_time - start_time;

    if (stats.elapsed_time_ms > 0) {
        double seconds = stats.elapsed_time_ms / 1000.0;
        double megabytes = stats.total_bytes_read / (1024.0 * 1024.0);
        stats.processing_rate_mbps = megabytes / seconds;
    }

    return stats;
}

void IngestPipeline::reset_statistics() noexcept {
    stat_files_scanned_.store(0, std::memory_order_relaxed);
    stat_files_processed_.store(0, std::memory_order_relaxed);
    stat_bytes_read_.store(0, std::memory_order_relaxed);
    stat_chunks_generated_.store(0, std::memory_order_relaxed);
    stat_embeddings_generated_.store(0, std::memory_order_relaxed);
    stat_errors_.store(0, std::memory_order_relaxed);
    stat_sum_entropy_.store(0.0, std::memory_order_relaxed);
    stat_sum_chunk_size_.store(0.0, std::memory_order_relaxed);
    stat_start_time_ms_.store(get_time_ms(), std::memory_order_relaxed);
}

// ==============================================================================
// Helper Functions
// ==============================================================================

Embedding IngestPipeline::simple_embedding_generation(const TextChunk& chunk) noexcept {
    Embedding emb(config_.embedding_dim);
    emb.chunk_id = chunk.chunk_id;
    emb.hash = chunk.hash;
    emb.timestamp = get_time_ms();

    // Simple embedding: statistical features of text
    // This is a placeholder - a real system would use a neural network
    // For now, generate deterministic features from hash and content

    // Use hash to seed the embedding
    uint64_t seed = chunk.hash;

    for (size_t i = 0; i < config_.embedding_dim; ++i) {
        // Simple pseudo-random generation from hash
        seed = seed * 6364136223846793005ULL + 1442695040888963407ULL;
        emb.vector[i] = static_cast<double>(seed) / static_cast<double>(UINT64_MAX) * 2.0 - 1.0;
    }

    // Mix in entropy and chunk size
    if (config_.embedding_dim > 0) {
        emb.vector[0] += chunk.entropy;
    }
    if (config_.embedding_dim > 1) {
        emb.vector[1] += static_cast<double>(chunk.size) / TextChunk::MAX_SIZE;
    }

    // Normalize to unit sphere if requested
    if (config_.normalize_embeddings) {
        double norm = 0.0;
        for (double val : emb.vector) {
            norm += val * val;
        }
        norm = std::sqrt(norm);

        if (norm > 1e-10) {
            for (double& val : emb.vector) {
                val /= norm;
            }
        }
        emb.magnitude = 1.0;
    } else {
        double norm = 0.0;
        for (double val : emb.vector) {
            norm += val * val;
        }
        emb.magnitude = std::sqrt(norm);
    }

    update_statistics(emb);

    // Invoke callback if set
    if (embedding_callback_) {
        embedding_callback_(emb);
    }

    return emb;
}

void IngestPipeline::update_statistics(const TextChunk& chunk) noexcept {
    stat_chunks_generated_.fetch_add(1, std::memory_order_relaxed);
    stat_sum_entropy_.fetch_add(chunk.entropy, std::memory_order_relaxed);
    stat_sum_chunk_size_.fetch_add(chunk.size, std::memory_order_relaxed);
}

void IngestPipeline::update_statistics(const Embedding& embedding) noexcept {
    stat_embeddings_generated_.fetch_add(1, std::memory_order_relaxed);
}

void IngestPipeline::report_error(const std::string& context, const std::string& message) noexcept {
    stat_errors_.fetch_add(1, std::memory_order_relaxed);
    if (error_callback_) {
        error_callback_(context, message);
    }
}

uint64_t IngestPipeline::get_time_ms() const noexcept {
    using namespace std::chrono;
    return duration_cast<milliseconds>(steady_clock::now().time_since_epoch()).count();
}

// ==============================================================================
// xxHash64 Implementation
// ==============================================================================

XXHash64::XXHash64(uint64_t seed) noexcept
    : total_len_(0)
    , seed_(seed)
    , buffer_size_(0)
{
    state_[0] = seed + PRIME64_1 + PRIME64_2;
    state_[1] = seed + PRIME64_2;
    state_[2] = seed;
    state_[3] = seed - PRIME64_1;
    std::memset(buffer_, 0, sizeof(buffer_));
}

void XXHash64::update(const void* data, size_t size) noexcept {
    const uint8_t* ptr = static_cast<const uint8_t*>(data);
    total_len_ += size;

    // Process buffered data first
    if (buffer_size_ > 0) {
        size_t to_copy = std::min(size, 32 - buffer_size_);
        std::memcpy(buffer_ + buffer_size_, ptr, to_copy);
        buffer_size_ += to_copy;
        ptr += to_copy;
        size -= to_copy;

        if (buffer_size_ == 32) {
            const uint64_t* lanes = reinterpret_cast<const uint64_t*>(buffer_);
            state_[0] = round(state_[0], lanes[0]);
            state_[1] = round(state_[1], lanes[1]);
            state_[2] = round(state_[2], lanes[2]);
            state_[3] = round(state_[3], lanes[3]);
            buffer_size_ = 0;
        }
    }

    // Process 32-byte blocks
    while (size >= 32) {
        const uint64_t* lanes = reinterpret_cast<const uint64_t*>(ptr);
        state_[0] = round(state_[0], lanes[0]);
        state_[1] = round(state_[1], lanes[1]);
        state_[2] = round(state_[2], lanes[2]);
        state_[3] = round(state_[3], lanes[3]);
        ptr += 32;
        size -= 32;
    }

    // Buffer remaining data
    if (size > 0) {
        std::memcpy(buffer_, ptr, size);
        buffer_size_ = size;
    }
}

uint64_t XXHash64::digest() noexcept {
    uint64_t h64;

    if (total_len_ >= 32) {
        h64 = rotl64(state_[0], 1) + rotl64(state_[1], 7) +
              rotl64(state_[2], 12) + rotl64(state_[3], 18);
        h64 = merge_round(h64, state_[0]);
        h64 = merge_round(h64, state_[1]);
        h64 = merge_round(h64, state_[2]);
        h64 = merge_round(h64, state_[3]);
    } else {
        h64 = seed_ + PRIME64_5;
    }

    h64 += total_len_;

    // Process remaining bytes
    const uint8_t* ptr = buffer_;
    size_t remaining = buffer_size_;

    while (remaining >= 8) {
        uint64_t lane = *reinterpret_cast<const uint64_t*>(ptr);
        lane = round(0, lane);
        h64 ^= lane;
        h64 = rotl64(h64, 27) * PRIME64_1 + PRIME64_4;
        ptr += 8;
        remaining -= 8;
    }

    if (remaining >= 4) {
        uint32_t lane = *reinterpret_cast<const uint32_t*>(ptr);
        h64 ^= static_cast<uint64_t>(lane) * PRIME64_1;
        h64 = rotl64(h64, 23) * PRIME64_2 + PRIME64_3;
        ptr += 4;
        remaining -= 4;
    }

    while (remaining > 0) {
        h64 ^= static_cast<uint64_t>(*ptr) * PRIME64_5;
        h64 = rotl64(h64, 11) * PRIME64_1;
        ptr++;
        remaining--;
    }

    return avalanche(h64);
}

uint64_t XXHash64::hash(const void* data, size_t size, uint64_t seed) noexcept {
    XXHash64 hasher(seed);
    hasher.update(data, size);
    return hasher.digest();
}

} // namespace tired_agi
Mathematical records from this source

ingest_pipeline · Compute Shannon entropy: H

This relation defines Shannon entropy: H in the source section “ingest_pipeline · Compute Shannon entropy: H”. Uses logarithmic scaling; its base and the positivity of its…