当前位置：首页 > news >正文

【译】数据摄取构建模块简介（预览版）(二)

news 2026/5/12 0:59:00

原文 | Luis, Adam

翻译 | 郑子铭

丰富和处理您的数据

下一步是处理数据IngestionChunk块。数据块处理器在数据块级别进行操作，可以丰富内容或执行其他转换。与文档处理器类似，选择使用哪些处理器取决于您的具体应用场景。

此示例使用内置功能SummaryEnricher，通过 AI 服务为每个数据块添加摘要：

IngestionChunkProcessor<string> summaryEnricher = new SummaryEnricher(enricherOptions);

存储您的数据

存储已处理的数据块是数据摄取管道的最后一步。s是一个用于将数据块存储在任何存储中的IngestionChunkWriter抽象概念，但它是一个使用向量存储的实现。它构建于 Microsoft.Extensions.VectorData.Abstractions抽象之上，因此您可以使用任何受支持的向量存储。IngestionChunkVectorStoreWriter

在这个例子中，我们将使用 SQLiteSqliteVectorStore将数据块存储在本地 SQLite 数据库中：

using SqliteVectorStore vectorStore = new("Data Source=vectors.db;Pooling=false",new(){EmbeddingGenerator = embeddingGenerator});// The writer requires the embedding dimension count to be specified.
// For OpenAI's `text-embedding-3-small`, the dimension count is 1536.
using VectorStoreWriter<string> writer = new(vectorStore, dimensionCount: 1536);

作者将自动：

使用默认架构创建向量存储集合。
使用提供的嵌入生成器为每个数据块生成嵌入。
完成后，删除之前为具有相同 ID 的文档存储的所有数据块（以支持对同一文档的不同版本进行增量分块）。

编写并运行您的管道

使用IngestionPipeline先前配置的读取器、分块器、增强器和写入器来处理当前目录中的所有文件。

using IngestionPipeline<string> pipeline = new(reader, chunker, writer, loggerFactory: loggerFactory)
{DocumentProcessors = { imageAlternativeTextEnricher },ChunkProcessors = { summaryEnricher }
};await foreach (var result in pipeline.ProcessAsync(new DirectoryInfo("."), searchPattern: "*.md"))
{Console.WriteLine($"Completed processing '{result.DocumentId}'. Succeeded: '{result.Succeeded}'.");
}

重要的

单个文档导入失败不应导致整个流程失败。该机制IngestionPipeline.ProcessAsync通过返回部分成功来实现IAsyncEnumerable。调用者负责处理任何失败情况（例如，重试导入失败的文档或在遇到第一个错误时停止）。

检索数据

VectorStoreWriter公开了可用于对存储的数据块执行向量搜索的底层机制VectorStoreCollection。此示例提示用户输入查询，并从向量存储中返回最相似的 3 个数据块：

var collection = writer.VectorStoreCollection;
while (true)  
{Console.Write("Enter your question (or 'exit' to quit): ");string? searchValue = Console.ReadLine();if (string.IsNullOrEmpty(searchValue) || searchValue == "exit"){break;}Console.WriteLine("Searching...\n");await foreach (var result in collection.SearchAsync(searchValue, top: 3)){Console.WriteLine($"Score: {result.Score}\n\tContent: {result.Record["content"]}");}
}

端到端场景

想看看实际效果吗？试试全新的 .NET AI Web 聊天模板，体验完整的端到端流程。您可以使用MarkItDown MCP 服务器解析文档，使用语义感知分块器将其分块，并将分块存储在您选择的矢量数据库中。示例应用程序包含一个 Web 聊天功能，该功能使用已接收的数据进行 RAG（红绿灯）分析。

dotnet new install Microsoft.Extensions.AI.Templates

构建您的分布式应用程序

以下来自模板的代码片段展示了如何在 Aspire 中配置应用程序的不同组件，包括用于托管模型的 Ollama、用于矢量存储的 Qdrant、用于文档解析的 MarkItDown 以及 Web 应用程序本身。

var builder = DistributedApplication.CreateBuilder(args);var ollama = builder.AddOllama("ollama").WithDataVolume();
var chat = ollama.AddModel("chat", "llama3.2");
var embeddings = ollama.AddModel("embeddings", "all-minilm");var vectorDB = builder.AddQdrant("vectordb").WithDataVolume().WithLifetime(ContainerLifetime.Persistent);var markitdown = builder.AddContainer("markitdown", "mcp/markitdown").WithArgs("--http", "--host", "0.0.0.0", "--port", "3001").WithHttpEndpoint(targetPort: 3001, name: "http");var webApp = builder.AddProject<Projects.RagSample_Web>("aichatweb-app");
webApp.WithReference(chat).WithReference(embeddings).WaitFor(chat).WaitFor(embeddings);
webApp.WithReference(vectorDB).WaitFor(vectorDB);
webApp.WithEnvironment("MARKITDOWN_MCP_URL", markitdown.GetEndpoint("http"));builder.Build().Run();

可观测性

使用Aspire，您可以获得丰富的数据摄取管道可观测性体验。

该模板已经包含了对数据摄取过程和 Web 应用程序的OpenTelemetry 跟踪。

public static TBuilder ConfigureOpenTelemetry<TBuilder>(this TBuilder builder) where TBuilder : IHostApplicationBuilder
{// The rest is omitted for brevity.builder.Services.AddOpenTelemetry().WithTracing(tracing =>{tracing.AddSource("Experimental.Microsoft.Extensions.AI").AddSource("Experimental.Microsoft.Extensions.DataIngestion");});return builder;
}