Proposal: Add LLM-Powered Ingestion Processors (CommunityToolkit.DataIngestion)

[luisquintanilla]

Proposal: CommunityToolkit.DataIngestion — LLM-Powered Chunk Processors

Summary

Add a CommunityToolkit.DataIngestion package that provides LLM-powered chunk enrichment processors for Microsoft.Extensions.DataIngestion pipelines. These processors enhance indexed chunks with semantic metadata at ingestion time, significantly improving downstream retrieval quality.

Motivation

The Problem with Basic RAG Ingestion

A basic RAG (Retrieval-Augmented Generation) ingestion pipeline works like this:

1. Read a document (PDF, markdown, HTML, etc.)
2. Split it into chunks (paragraphs, pages, or fixed-size windows)
3. Embed each chunk into a vector (a numeric representation of its meaning)
4. Store those vectors in a database

At query time, the user's question is embedded and you find the "closest" chunks by vector similarity. The problem? This approach loses a lot of context:

• A chunk might mention "John" but the embedding doesn't know that's a person entity you might want to filter on
• A chunk about security best practices looks the same as one about performance tuning — there's no topic label to route queries
• Users ask questions ("How do I configure auth?") but chunks are statements ("Auth is configured via appsettings.json") — the embeddings live in different semantic spaces
• A user asking a broad question ("What does this codebase do?") gets back tiny specific chunks — there's no high-level summary to retrieve

How These Processors Solve It

Each processor in this package runs at ingestion time (before storing) and enriches chunks with metadata or generates additional chunks that dramatically improve retrieval quality:

1. Entity Extraction (EntityExtractionProcessor)

What it does: Asks an LLM to identify named entities (people, organizations, technologies, version numbers) in each chunk and stores them as structured metadata.

Why it helps: Enables filtered vector search. Instead of hoping vector similarity alone finds the right chunk, you can combine "semantically similar" with "mentions Azure Functions" or "references v3.0". This turns fuzzy retrieval into precise retrieval.

Example: A chunk about dependency injection gets metadata like:

entities_technologies: "Microsoft.Extensions.DependencyInjection, IServiceCollection"
entities_organizations: "Microsoft"

2. Topic Classification (TopicClassificationProcessor)

What it does: Classifies each chunk into one primary topic and zero or more secondary topics from a taxonomy you define (e.g., "web", "data", "security", "performance", "architecture").

Why it helps: Enables topic-based routing and faceted search. If a user asks a security question, you can boost or exclusively search chunks labeled "security" — reducing noise from unrelated content. Also useful for analytics ("what % of our docs cover performance?").

Example: You define ["web", "data", "security", "performance", "architecture"] as your taxonomy. A chunk about SQL injection prevention gets:

topic_primary: "security"
topic_secondary: "data"

3. Hypothetical Query Generation (HypotheticalQueryProcessor)

What it does: For each chunk, asks an LLM: "What questions would this text answer?" Then stores those questions as additional chunks in the same collection, linked back to the original.

Why it helps: This is sometimes called "reverse HyDE" (Hypothetical Document Embeddings, but in reverse). The core insight: users ask questions, but documents contain answers. These live in different "semantic neighborhoods" — a question embedding and its answer embedding aren't always close in vector space. By generating question-chunks at ingestion time, you create entries that are already in "question space" — directly matching how users actually query.

Example: A chunk explaining how to configure CORS gets 3 synthetic question-chunks:

• "How do I enable CORS in ASP.NET Core?"
• "What middleware is needed for cross-origin requests?"
• "Where do I configure allowed origins?"

When a user asks something similar, these question-chunks match much better than the original declarative text would.

4. Tree Index (TreeIndexProcessor)

What it does: After processing all chunks, generates summary chunks at multiple levels — one summary per document (branch level), and one summary for the entire corpus (root level). These summaries are stored in the same collection as the original chunks.

Why it helps: This is based on the RAPTOR technique (Recursive Abstractive Processing for Tree-Organized Retrieval). Standard chunking only gives you fine-grained detail — great for specific questions, terrible for broad ones. By adding summaries at different abstraction levels, vector search naturally returns a mix of specific details and broader context. A broad question retrieves the root/branch summaries; a specific question retrieves leaf chunks. No special retrieval logic needed — it "just works" because the summaries are stored alongside everything else.

Example: You ingest 50 pages of documentation. The tree index creates:

• 50+ leaf chunks (the originals)
• ~5 branch summaries (one per logical document, 2-3 sentences each)
• 1 root summary (corpus overview)

A question like "What is this project about?" now has a great match in the root summary, while "How do I configure the retry policy?" still matches the specific leaf chunk.

Why At Ingestion Time?

All of these techniques run once when you ingest documents — not on every query. This is important because:

• LLM calls are expensive; paying once at ingest is far cheaper than paying per-query
• The enriched metadata persists in your vector store permanently
• Query latency stays fast (no LLM calls in the retrieval hot path)
• You can re-ingest selectively when documents change

Composability

These processors are independent and composable. Use one, some, or all depending on your needs:

// Just entity extraction for filtered search
.UseEntityExtraction()

// Full enrichment pipeline
.UseEntityExtraction()
.UseTopicClassification()
.UseHypotheticalQueries()
.UseTreeIndex()

Each processor receives chunks from the previous one (or from the chunker if first), enriches them, and passes them along. They implement the standard IngestionChunkProcessor<string> interface from Microsoft.Extensions.DataIngestion.

API Surface

Package: CommunityToolkit.DataIngestion

Processors:

namespace CommunityToolkit.DataIngestion;

// Entity extraction: people, organizations, technologies, versions → metadata
public sealed class EntityExtractionProcessor : IngestionChunkProcessor<string>
{
    public EntityExtractionProcessor(IChatClient chatClient, ILogger<EntityExtractionProcessor>? logger = null);
}

// Topic classification: configurable taxonomy → primary + secondary labels
public sealed class TopicClassificationProcessor : IngestionChunkProcessor<string>
{
    public TopicClassificationProcessor(IChatClient chatClient, string[] taxonomy, ILogger<TopicClassificationProcessor>? logger = null);
}

// Reverse HyDE: generates N questions per chunk for question-space embeddings
public sealed class HypotheticalQueryProcessor : IngestionChunkProcessor<string>
{
    public HypotheticalQueryProcessor(IChatClient chatClient, int questionsPerChunk = 3, ILogger<HypotheticalQueryProcessor>? logger = null);
}

// RAPTOR tree: leaf → branch (doc-level) → root (corpus-level) summaries
public sealed class TreeIndexProcessor : IngestionChunkProcessor<string>
{
    public TreeIndexProcessor(IChatClient chatClient, ILogger<TreeIndexProcessor>? logger = null);
}

Metadata Constants:

public static class MetadataKeys
{
    // EntityExtraction
    public const string EntitiesPeople = "entities_people";
    public const string EntitiesOrganizations = "entities_organizations";
    public const string EntitiesTechnologies = "entities_technologies";
    public const string EntitiesVersions = "entities_versions";
    
    // TopicClassification
    public const string TopicPrimary = "topic_primary";
    public const string TopicSecondary = "topic_secondary";
    
    // HypotheticalQuery + TreeIndex
    public const string ChunkType = "chunk_type";
    public const string ParentChunkId = "parent_chunk_id";
    public const string HypotheticalQuestions = "hypothetical_questions";
    public const string Level = "level";
    public const string ParentId = "parent_id";
}

Builder + DI Extensions:

public class IngestionPipelineBuilder<T>
{
    public IngestionPipelineBuilder<T> UseDocumentProcessor(Func<IServiceProvider, IngestionDocumentProcessor> factory);
    public IngestionPipelineBuilder<T> UseDocumentProcessor<TProcessor>() where TProcessor : IngestionDocumentProcessor;
    public IngestionPipelineBuilder<T> UseChunkProcessor(Func<IServiceProvider, IngestionChunkProcessor<T>> factory);
    public IngestionPipelineBuilder<T> UseChunkProcessor<TProcessor>() where TProcessor : IngestionChunkProcessor<T>;
    
    // Convenience methods
    public IngestionPipelineBuilder<T> UseEntityExtraction();
    public IngestionPipelineBuilder<T> UseTopicClassification(Action<TopicClassificationOptions>? configure = null);
    public IngestionPipelineBuilder<T> UseHypotheticalQueries(Action<HypotheticalQueryOptions>? configure = null);
    public IngestionPipelineBuilder<T> UseTreeIndex();
    
    public IngestionPipeline<T> Build(IServiceProvider sp, IngestionDocumentReader reader, IngestionChunker<T> chunker, IngestionChunkWriter<T> writer, ILoggerFactory? loggerFactory = null);
}

public static class DataIngestionServiceCollectionExtensions
{
    public static IngestionPipelineBuilder<T> AddIngestionPipeline<T>(this IServiceCollection services);
    public static IngestionPipelineBuilder<string> AddIngestionPipeline(this IServiceCollection services);
}

Usage Example

builder.Services.AddChatClient(new OllamaChatClient(...));

builder.Services.AddIngestionPipeline<string>()
    .UseEntityExtraction()
    .UseTopicClassification(o => o.Taxonomy = ["web", "data", "security", "performance"])
    .UseHypotheticalQueries(o => o.QuestionsPerChunk = 5)
    .UseTreeIndex();

// Later, build the pipeline with reader/chunker/writer:
var pipeline = pipelineBuilder.Build(serviceProvider, reader, chunker, writer);
await pipeline.RunAsync(documents, cancellationToken);

Design Decisions

1. IngestionChunkProcessor<string> base — directly implements the stable MEDI abstraction; no custom interfaces needed
2. IChatClient dependency — model-agnostic via M.E.AI; works with OpenAI, Ollama, Azure AI, etc.
3. MetadataKeys constants — FDG compliance; avoids magic strings, enables IntelliSense
4. Optional ILogger — observability without forcing DI; defaults to NullLogger
5. Separate from PdfPig PR — different concern domain (enrichment vs. reading); different dependency chains
6. Processors are composable — chain any combination; each is independent

Relationship to Other Work

Package	Role	Status
Microsoft.Extensions.DataIngestion	Upstream abstractions	✅ Shipped (stable)
CommunityToolkit.DocumentProcessing.PdfPig.*	PDF reading + chunking	✅ PR #3 filed
CommunityToolkit.DataIngestion	LLM chunk enrichment	This proposal
CommunityToolkit.DataRetrieval	Retrieval processors	Separate (depends on extensions design review)

Dependencies

• Microsoft.Extensions.DataIngestion (stable, shipped)
• Microsoft.Extensions.DataIngestion.Abstractions (stable, shipped)
• Microsoft.Extensions.AI.Abstractions (stable)
• Microsoft.Extensions.DependencyInjection.Abstractions (stable)
• Microsoft.Extensions.Logging.Abstractions (stable)

No preview or unstable dependencies.