作者来自 Elastic Ajay Krishnan Gopalan了解如何使用 NeMo Retriever、Unstructured Platform 和 Elasticsearch 为 RAG 应用构建可扩展的非结构化文档数据处理流水线。在本博客中我们将讨论如何使用 NVIDIA NeMo Retriever 提取模型、Unstructured Platform 和 Elasticsearch 构建一个可扩展的数据处理流水线。该流水线能够将来自数据源的非结构化数据转换为结构化、可搜索的内容为 RAG 等下游 AI 应用做好准备。检索增强生成 RAG 是一种 AI 技术它为大型语言模型 LLM 提供外部知识以生成对用户查询的回答。这使 LLM 能够结合特定上下文生成回答从而使答案更加准确且更具相关性。在开始之前我们先来了解一下支撑这一流水线的关键组件以及它们各自发挥的作用。Pipeline 组件NeMo Retriever extraction 是一组用于将非结构化文档转换为结构化内容和元数据的微服务。它能够大规模处理文档解析、视觉结构识别以及 OCR。RAG NVIDIA AI Blueprint 提供了一个起点展示了如何在高性能提取流水线中使用 NeMo Retriever 微服务。Unstructured 是一个 ETL 平台用于编排整个非结构化数据处理流程包括从多个数据源采集非结构化数据通过可配置的工作流引擎将原始非结构化文件转换为结构化数据利用额外的数据转换进行丰富处理以及最终将结果上传到向量存储、数据库和搜索引擎。它提供可视化 UI、API 和可扩展的后端基础设施在单一工作流中完成文档解析、数据丰富和嵌入生成。Elasticsearch 是业界领先的搜索与分析引擎目前已原生支持向量搜索能力。它既可以作为传统文本数据库也可以作为向量数据库支持包括 k-NN 相似度搜索在内的大规模语义搜索。介绍完这些核心组件之后我们先来看一下它们在典型工作流中是如何协同工作的然后再深入具体实现。使用 NeMo Retriever Unstructured Elasticsearch 构建 RAG这里仅介绍关键内容完整 Notebook 请参阅对应链接。本文分为三个部分配置源连接器和目标连接器使用 Unstructured API 配置工作流基于处理后的数据构建 RAGUnstructured 工作流采用 DAG有向无环图表示其中的节点称为连接器 connector 用于控制数据从哪里采集以及处理后的结果上传到哪里。这些节点是任何工作流都必需的。源连接器负责配置从数据源采集原始数据目标连接器负责将处理后的数据上传到向量存储、搜索引擎或数据库。在本文中我们将研究论文存储在 Amazon S3 中并希望将处理后的数据写入 Elasticsearch以供下游应用使用。这意味着在构建数据处理工作流之前需要先使用 Unstructured API 创建一个 Amazon S3 源连接器以及一个 Elasticsearch 目标连接器。步骤 1配置 S3 源连接器创建源连接器时需要为其指定一个唯一名称、连接器类型例如 S3 或 Google Drive并提供相应配置。配置通常包括所连接数据源的位置例如 S3 Bucket URI 或 Google Drive 文件夹以及认证信息。source_connector_response unstructured_client.sources.create_source( requestCreateSourceRequest( create_source_connectorCreateSourceConnector( namedemo_source1, typeSourceConnectorType.S3, configS3SourceConnectorConfigInput( keyos.environ[S3_AWS_KEY], secretos.environ[S3_AWS_SECRET], remote_urlos.environ[S3_REMOTE_URL], recursiveFalse #True/False ) ) ) ) pretty_print_model(source_connector_response.source_connector_information)步骤 2配置 Elasticsearch 目标连接器接下来配置 Elasticsearch 目标连接器。你所使用的 Elasticsearch 索引必须具有与 Unstructured 生成文档模式 schema 兼容的映射。有关详细要求请参阅相关文档。destination_connector_response unstructured_client.destinations.create_destination( requestCreateDestinationRequest( create_destination_connectorCreateDestinationConnector( namedemo_dest-3, typeDestinationConnectorType.ELASTICSEARCH, configElasticsearchConnectorConfigInput( hosts[os.environ[es_host]], es_api_keyos.environ[es_api_key], index_namedemo-index ) ) ) )步骤 3使用 Unstructured 创建工作流完成源连接器和目标连接器的配置后就可以创建一个新的数据处理工作流。我们将使用以下节点构建工作流 DAG使用 NeMo Retriever 对文档进行分区 partitioning 使用 Unstructured 的 Image Summarizer、Table Summarizer 和 Named Entity Recognition 节点对内容进行丰富 enrichment 使用 Chunker 和 Embedder 节点为内容做好相似性搜索的准备from unstructured_client.models.shared import ( WorkflowNode, WorkflowNodeType, WorkflowType, Schedule ) # Partition the content by using NV-Ingest parition_node WorkflowNode( nameIngest, subtypenvingest, typepartition, settings{nvingest_host: userdata.get(NV-Ingest-host-address)}, ) # Summarize each detected image. image_summarizer_node WorkflowNode( nameImage summarizer, subtypeopenai_image_description, typeWorkflowNodeType.PROMPTER, settings{} ) # Summarize each detected table. table_summarizer_node WorkflowNode( nameTable summarizer, subtypeanthropic_table_description, typeWorkflowNodeType.PROMPTER, settings{} ) # Label each recognized named entity. named_entity_recognizer_node WorkflowNode( nameNamed entity recognizer, subtypeopenai_ner, typeWorkflowNodeType.PROMPTER, settings{ prompt_interface_overrides: None } ) # Chunk the partitioned content. chunk_node WorkflowNode( nameChunker, subtypechunk_by_title, typeWorkflowNodeType.CHUNK, settings{ unstructured_api_url: None, unstructured_api_key: None, multipage_sections: False, combine_text_under_n_chars: 0, include_orig_elements: True, max_characters: 1537, overlap: 160, overlap_all: False, contextual_chunking_strategy: None } ) # Generate vector embeddings. embed_node WorkflowNode( nameEmbedder, subtypeazure_openai, typeWorkflowNodeType.EMBED, settings{ model_name: text-embedding-3-large } ) response unstructured_client.workflows.create_workflow( request{ create_workflow: { name: fs3-to-es-NV-Ingest-custom-workflow, source_id: source_connector_response.source_connector_information.id, destination_id: a72838a4-bb72-4e93-972d-22dc0403ae9e, workflow_type: WorkflowType.CUSTOM, workflow_nodes: [ parition_node, image_summarizer_node, table_summarizer_node, named_entity_recognizer_node, chunk_node, embed_node ], } } ) workflow_id response.workflow_information.id pretty_print_model(response.workflow_information) job unstructured_client.workflows.run_workflow( request{ workflow_id: workflow_id, } ) pretty_print_model(job.job_information)工作流任务完成后数据将上传到 Elasticsearch接下来我们就可以开始构建一个基础的 RAG 应用。步骤 4配置 RAG接下来我们将实现一个简单的检索器 retriever 。它会连接到数据源接收用户查询使用与原始数据生成嵌入时相同的模型对查询进行嵌入 embedding 然后计算余弦相似度 cosine similarity 检索出最相关的前 3 个文档。from langchain_elasticsearch import ElasticsearchStore from langchain.embeddings import OpenAIEmbeddings import os embeddings OpenAIEmbeddings( modeltext-embedding-3-large, openai_api_keyos.environ[OPENAI_API_KEY] ) vector_store ElasticsearchStore( es_urlos.environ[es_host], index_namedemo-index, embeddingembeddings, es_api_keyos.environ[es_api_key], query_fieldtext, vector_query_fieldembeddings, distance_strategyCOSINE ) retriever vector_store.as_retriever( search_typesimilarity, search_kwargs{k: 3} # Number of results to return )接下来我们将配置一个工作流用于接收用户查询从 Elasticsearch 检索相似文档并将这些文档作为上下文来回答用户的问题。from openai import OpenAI client OpenAI(api_keyos.getenv(OPENAI_API_KEY)) def generate_answer(question: str, documents: str): prompt You are an assistant that can answer user questions given provided context. Your answer should be thorough and technical. If you dont know the answer, or no documents are provided, say I do not have enough context to answer the question. augmented_prompt ( f{prompt} fUser question: {question}\n\n f{documents} ) response client.chat.completions.create( messages[ {role: system, content: You answer users questions.}, {role: user, content: augmented_prompt}, ], modelgpt-4o-2024-11-20, temperature0, ) return response.choices[0].message.content def format_docs(docs): seen_texts set() useful_content [doc.page_content for doc in docs] return \nRetrieved documents:\n .join( [ f\n\n Document {str(i)} \n doc for i, doc in enumerate(useful_content) ] ) def rag(query): docs retriever.invoke(query) documents format_docs(docs) answer generate_answer(query, documents) return documents, answer将所有步骤组合起来后我们得到如下流程query How did the response lengths change with training? docs, answer rag(query) print(answer)以及一个响应Based on the provided context, the response lengths during training for the DeepSeek-R1-Zero model showed a clear trend of increasing as the number of training steps progressed. This is evident from the graphs described in Document 0 and Document 1, which both depict the average length per response on the y-axis and training steps on the x-axis. ### Key Observations: 1. **Increasing Trend**: The average response length consistently increased as training steps advanced. This suggests that the model naturally learned to allocate more thinking time (i.e., generate longer responses) as it improved its reasoning capabilities during the reinforcement learning (RL) process. 2. **Variability**: Both graphs include a shaded area around the average response length, indicating some variability in response lengths during training. However, the overall trend remained upward. 3. **Quantitative Range**: The y-axis for response length ranged from 0 to 12,000 tokens, and the graphs show a steady increase in the average response length over the course of training, though specific numerical values at different steps are not provided in the descriptions. ### Implications: The increase in response length aligns with the models goal of solving reasoning tasks more effectively. Longer responses likely reflect the models ability to provide more detailed and comprehensive reasoning, which is critical for tasks requiring complex problem-solving. In summary, the response lengths increased during training, indicating that the model adapted to allocate more resources (in terms of response length) to improve its reasoning performance.Elasticsearch 提供了各种增强搜索的策略包括混合搜索Hybrid search它结合了近似语义搜索和基于关键词的搜索。这种方法可以提高 RAG 架构中作为上下文使用的顶部文档的相关性。要启用它你需要如下修改 vector_store 初始化from langchain_elasticsearch import DenseVectorStrategy vector_store ElasticsearchStore( es_urlos.environ[es_host], index_namedemo-index, embeddingembeddings, es_api_keyos.environ[es_api_key], query_fieldtext, vector_query_fieldembeddings, strategyDenseVectorStrategy(hybridTrue) // -- here the change )结论良好的 RAG 始于经过良好准备的数据而 Unstructured 简化了这一关键的第一步。通过使用 NeMo Retriever 启用分区处理、对非结构化数据进行元数据增强以及高效地将数据摄取到 Elasticsearch 中它确保你的 RAG 流程建立在坚实的基础之上为所有下游任务释放其全部潜力。常见问题什么是非结构化数据非结构化数据指的是没有预定义格式或可搜索结构的原始文件例如 PDF、研究论文和图像。NVIDIA NeMo Retriever 做什么NVIDIA NeMo Retriever 使用 OCR 和视觉工具来识别页面中的不同部分例如文本块、表格和图像。NeMo Retriever 如何与 Elasticsearch 协作NeMo Retriever 通过对数据进行分区处理和清理来准备数据然后将其保存为 Elasticsearch 中的“向量”这样 AI 就可以立即搜索其中的内容。原文https://www.elastic.co/search-labs/blog/unstructured-data-processing-with-nvidia-nemo-retriever-unstructured-and-elasticsearch