生產(chǎn)級(jí)大模型RAG應(yīng)用：可擴(kuò)展的文檔處理管道

大家好，我是 Ai 學(xué)習(xí)的老章

隨著組織產(chǎn)生的數(shù)據(jù)量不斷增長(zhǎng)，從非結(jié)構(gòu)化文檔中提取有價(jià)值的洞察已成為一項(xiàng)重大挑戰(zhàn)。本文介紹了一種高級(jí)架構(gòu)，利用云存儲(chǔ)、流式技術(shù)、機(jī)器學(xué)習(xí)和數(shù)據(jù)庫(kù)，構(gòu)建出穩(wěn)健高效的文檔處理管道。

引言：現(xiàn)代文檔處理面臨的挑戰(zhàn)

如今，企業(yè)正被海量文檔淹沒。PDF、報(bào)告、合同及其他文本密集型文件中蘊(yùn)含寶貴信息，但高效提取這些知識(shí)卻面臨巨大挑戰(zhàn)。傳統(tǒng)文檔處理方式往往存在以下局限：

• 處理大規(guī)模文檔時(shí)的擴(kuò)展性問(wèn)題。

• 解析復(fù)雜文檔結(jié)構(gòu)的智能程度有限。

• 批處理延遲阻礙及時(shí)、實(shí)時(shí)的洞察。

• 難以集成處理后的數(shù)據(jù)到下游應(yīng)用。

現(xiàn)代企業(yè)需要能夠大規(guī)模處理文檔、提取其語(yǔ)義含義，并使這些信息立即可用于搜索、推薦系統(tǒng)或商業(yè)智能等應(yīng)用的解決方案。我們的架構(gòu)通過(guò)基于流的方法應(yīng)對(duì)這些挑戰(zhàn)，利用先進(jìn)的前沿技術(shù)。

架構(gòu)概覽：從原始文件到結(jié)構(gòu)化數(shù)據(jù)

我們的解決方案采用了一套精密的實(shí)時(shí)數(shù)據(jù)處理流水線，融合了云存儲(chǔ)、流處理、機(jī)器學(xué)習(xí)和持久化數(shù)據(jù)庫(kù)，構(gòu)建出面向文檔富化與分析的全棧系統(tǒng)。

其核心架構(gòu)遵循流式數(shù)據(jù)模式，信息從源頭持續(xù)流向目的地，并在途中被不斷轉(zhuǎn)換與增強(qiáng)。下面逐一梳理關(guān)鍵組件：

• AWS S3 存儲(chǔ)桶：作為主力數(shù)據(jù)湖，用于存放原始 PDF 文檔。

• Python 攝取腳本：從 S3 讀取文件并協(xié)調(diào)文檔處理流程。

• LlamaParse：提供智能文檔解析與分塊功能。

• Confluent：作為“中樞神經(jīng)系統(tǒng)”，包含兩個(gè)核心主題：

• “raw”：存放已解析的文檔分塊。

• “summary_embedding”：存儲(chǔ)已處理并帶有嵌入向量的分塊。

• Apache Flink：處理流式數(shù)據(jù)，并通過(guò) ML 生成嵌入。

• Confluent Schema Registry：管理數(shù)據(jù)契約，確保數(shù)據(jù)格式一致。

• MongoDB：存儲(chǔ)最終處理后的文檔及其嵌入。

該架構(gòu)特別適用于需要實(shí)時(shí)文檔處理并結(jié)合機(jī)器學(xué)習(xí)增強(qiáng)的場(chǎng)景，例如語(yǔ)義搜索應(yīng)用、內(nèi)容分類或知識(shí)管理系統(tǒng)。

數(shù)據(jù)攝?。豪?LlamaParse 的高效文檔分塊

旅程始于存儲(chǔ)在 AWS S3 存儲(chǔ)桶中的 PDF 文檔。由 Python 構(gòu)建的攝取層負(fù)責(zé)以下任務(wù)：

• 文檔檢索：Python 腳本使用配置的憑證連接 AWS S3，以訪問(wèn)存儲(chǔ)的 PDF 文檔。

• 借助 LlamaParse 的智能解析：該系統(tǒng)從根本上改變了 PDF 的處理方式。它不再將復(fù)雜文檔視為簡(jiǎn)單的文本序列進(jìn)行提取，而是借助 LlamaParse 的強(qiáng)大功能。這一先進(jìn)的文檔解析工具不僅限于字符識(shí)別，還能智能理解文檔的結(jié)構(gòu)與布局。LlamaParse 能夠細(xì)致地識(shí)別并解讀以下關(guān)鍵格式元素：

• 表格：準(zhǔn)確區(qū)分行、列和單元格內(nèi)容，保持表格數(shù)據(jù)完整性。

• 圖像：在文本中識(shí)別圖像，并根據(jù)圖像在整體布局中的位置提供額外上下文。

• 標(biāo)題：識(shí)別層級(jí)標(biāo)題和子標(biāo)題對(duì)于有效組織文檔至關(guān)重要。

• 其他格式元素：包括列表、加粗文本、斜體以及各種布局組件，確保在解析過(guò)程中語(yǔ)義含義和視覺呈現(xiàn)得以保留。

通過(guò)利用 LlamaParse，系統(tǒng)確保不會(huì)在文檔中丟失上下文，采用結(jié)合傳統(tǒng) OCR、LLMs 和 LVMs 的解析策略。

以下 Python 代碼演示如何初始化解析器并提取相關(guān)信息。

# Initialize LlamaParse with your API key parser = LlamaParse( api_key=os.getenv("LLAMA_PARSE_API_KEY"), result_type="json"  # Get JSON output  ) # Parse the PDF with file name in extra_info parsed_doc = parser.parse(file_bytes, extra_info={"file_name": file_name})

• 文檔分塊：LlamaParse 會(huì)將文檔拆分為可管理的塊，通常在頁(yè)面級(jí)別進(jìn)行，同時(shí)保留每個(gè)塊的上下文和元數(shù)據(jù)。這種分塊方法帶來(lái)多項(xiàng)好處：

• 更高效地處理大型文檔。

• 更精準(zhǔn)的上下文用于生成嵌入。

• 最終應(yīng)用的搜索粒度更高。

處理后的分塊隨即進(jìn)入管道的下一階段。Python 腳本會(huì)優(yōu)雅地處理解析錯(cuò)誤，確保在遇到格式錯(cuò)誤的文檔時(shí)管道依然穩(wěn)健。

流式基礎(chǔ)設(shè)施：利用 Confluent Cloud

Confluent Cloud 是一項(xiàng)全托管的 Apache Kafka 服務(wù)，作為我們架構(gòu)的骨干。該流處理平臺(tái)具備多項(xiàng)優(yōu)勢(shì)：

• 解耦組件：Kafka 將數(shù)據(jù)生產(chǎn)者（文檔解析器）與數(shù)據(jù)消費(fèi)者（處理引擎）分離，使它們各自按自己的節(jié)奏運(yùn)行。同樣，LlamaParse、Flink 和 MongoDB 也通過(guò)解耦，以不同的吞吐量獨(dú)立處理和寫入數(shù)據(jù)。

• 可擴(kuò)展性：平臺(tái)可處理高吞吐，并支持可配置的分區(qū)（在我們的實(shí)現(xiàn)中，每個(gè)主題 6 個(gè)分區(qū)）。

• 數(shù)據(jù)韌性：Kafka 的復(fù)制機(jī)制確保在處理過(guò)程中不會(huì)丟失任何文檔分片。

• 模式管理：Confluent Schema Registry 為模式演進(jìn)提供了強(qiáng)有力的保證（前向和后向兼容性）。

我們的實(shí)現(xiàn)使用了兩個(gè)主要的 Kafka 主題：

• raw:包含來(lái)自 LlamaParse 的已解析文檔塊。

• summary_embedding:存儲(chǔ)已處理的分塊及其生成的向量嵌入。

嵌入消息的 Avro 模式確保一致性：

{   "type": "record", "name": "summary_embedding_value", "namespace": "org.apache.flink.avro.generated.record", "fields": [     {       "name": "content",       "type": ["null", "string"],       "default": null     },     {       "name": "embeddings",       "type": ["null", {"type": "array", "items": ["null", "float"]}],       "default": null     }   ] }

該模式定義了每條消息的結(jié)構(gòu)，包含原始文本內(nèi)容及其對(duì)應(yīng)的向量嵌入。

一旦文檔分塊流入 Apache Kafka，真正的魔法就在處理層上演。Apache Flink 作為強(qiáng)大的流處理框架，從原始主題消費(fèi)數(shù)據(jù)并應(yīng)用轉(zhuǎn)換以生成嵌入。Flink 持續(xù)處理來(lái)自 Kafka 的文檔分塊流，并持續(xù)將富化的 summary_embedding 流回寫到 Kafka。

嵌入是捕捉文本語(yǔ)義含義的數(shù)值向量表示。它們賦予以下強(qiáng)大能力：

• 語(yǔ)義搜索（按含義而非僅按關(guān)鍵詞查找文檔）。

• 文檔聚類與分類。

• 文檔間的相似度檢測(cè)。

• 構(gòu)建復(fù)雜 AI 應(yīng)用的基礎(chǔ)。

我們的實(shí)現(xiàn)通過(guò) Flink SQL 調(diào)用 AWS Bedrock 生成嵌入：

-- Create the embedding model CREATEMODEL AWSBedrockEmbedding INPUT (textSTRING) OUTPUT (embeddings ARRAY

 ) WITH (     'bedrock.connection' = 'bedrock-connection',     'task' = 'embedding',     'provider' = 'BEDROCK' ); -- Create the destination table CREATETABLE summary_embedding (     contentSTRING,     embeddings ARRAY

 ); -- Insert transformed data INSERTINTO summary_embedding SELECT     CAST(val asSTRING),     embeddings FROM     raw,     LATERALTABLE (ML_PREDICT('AWSBedrockEmbedding', CAST(val asSTRING)));

該 SQL 定義了 Flink 應(yīng)如何：

• 連接到 AWS Bedrock 以獲取 ML 功能。

• 定義嵌入的目標(biāo)結(jié)構(gòu)。

• 通過(guò) ML_PREDICT 函數(shù)生成嵌入，從而轉(zhuǎn)換傳入的文本。

結(jié)果是一個(gè)持續(xù)的數(shù)據(jù)流，其中文檔塊與其語(yǔ)義向量表示成對(duì)出現(xiàn)。

數(shù)據(jù)消費(fèi)：支持 schema 演進(jìn)的 Avro 反序列化

在消費(fèi)端，一個(gè)專用消費(fèi)者應(yīng)用從 embedded_data 主題讀取數(shù)據(jù)。該應(yīng)用負(fù)責(zé)多項(xiàng)關(guān)鍵任務(wù)：

• 消息消費(fèi)：通過(guò)完善的偏移量管理，高效地從 Kafka 讀取消息。

• Avro 反序列化：借助 Schema Registry，將二進(jìn)制 Avro 格式還原為可用對(duì)象。

• 錯(cuò)誤處理與重試：處理消費(fèi)或處理過(guò)程中可能出現(xiàn)的故障。

Avro 反序列化對(duì)于保持管道演進(jìn)時(shí)的兼容性尤為關(guān)鍵。Schema Registry 確保即使模式隨時(shí)間變化（例如新增字段），消費(fèi)者仍能正確解析使用舊模式生成的消息。

消費(fèi)者應(yīng)用采用多線程實(shí)現(xiàn)，以最大化吞吐量，允許并行處理來(lái)自不同分區(qū)的消息。

存儲(chǔ)策略：使用 MongoDB 進(jìn)行靈活的文檔存儲(chǔ)

處理后的文檔分塊的最終目的地是 MongoDB，這是一個(gè)面向文檔的數(shù)據(jù)庫(kù)，非常適合存儲(chǔ)復(fù)雜的嵌套數(shù)據(jù)結(jié)構(gòu)（包括向量嵌入）。

MongoDB 為該架構(gòu)提供了多項(xiàng)優(yōu)勢(shì)：

• 靈活的模式：適應(yīng)不同的文檔結(jié)構(gòu)和元數(shù)據(jù)。

• 向量存儲(chǔ)：高效存儲(chǔ)并索引高維嵌入向量。

• 查詢能力：通過(guò)向量相似度查詢支持語(yǔ)義搜索。

• 可擴(kuò)展性：通過(guò)分片處理大規(guī)模文檔集合。

• 集成選項(xiàng)：可輕松連接下游應(yīng)用和可視化工具。

消費(fèi)者應(yīng)用將每個(gè)處理后的文檔片段插入 MongoDB，同時(shí)保留原始文本內(nèi)容和生成的嵌入向量，使數(shù)據(jù)立即可供需要搜索或分析文檔集合的應(yīng)用使用。

MongoDB 與其他向量數(shù)據(jù)庫(kù)的區(qū)別

MongoDB 作為向量存儲(chǔ)的通用之選，在與專用向量數(shù)據(jù)庫(kù)對(duì)比時(shí)尤為突出，原因如下：

• 原生集成：MongoDB 的核心優(yōu)勢(shì)在于能夠在同一平臺(tái)內(nèi)同時(shí)存儲(chǔ)和管理結(jié)構(gòu)化與非結(jié)構(gòu)化數(shù)據(jù)（包括向量嵌入）。不同于通常需要額外數(shù)據(jù)同步與管理的獨(dú)立向量數(shù)據(jù)庫(kù)，MongoDB Atlas Vector Search 讓你把原始數(shù)據(jù)及其對(duì)應(yīng)的嵌入向量存儲(chǔ)在同一份文檔中，從而消除數(shù)據(jù)孤島并簡(jiǎn)化架構(gòu)。

• 靈活的數(shù)據(jù)模型：MongoDB 的文檔模型提供前所未有的靈活性。你可以將原始文本、元數(shù)據(jù)和向量嵌入全部存儲(chǔ)在一個(gè)類似 JSON 的文檔里，無(wú)需跨多張表或多個(gè)系統(tǒng)對(duì)數(shù)據(jù)進(jìn)行規(guī)范化，讓模式演進(jìn)更簡(jiǎn)單，并降低開發(fā)復(fù)雜度。

• 全面的查詢能力：除了簡(jiǎn)單的向量相似度搜索，MongoDB 還允許你將向量搜索與其他強(qiáng)大的查詢操作符結(jié)合使用，例如按元數(shù)據(jù)過(guò)濾、地理空間查詢或全文搜索。這使得信息檢索更加細(xì)致和精準(zhǔn)，對(duì)高級(jí) AI 應(yīng)用至關(guān)重要。

• 運(yùn)營(yíng)成熟度和生態(tài)系統(tǒng)：MongoDB 是一款成熟、久經(jīng)考驗(yàn)的數(shù)據(jù)庫(kù)，擁有豐富的工具、驅(qū)動(dòng)和集成生態(tài)。它提供企業(yè)級(jí)功能，如可擴(kuò)展性、高可用性、安全性以及豐富的開發(fā)者工具。專門的向量數(shù)據(jù)庫(kù)雖然在其細(xì)分領(lǐng)域表現(xiàn)良好，但可能缺乏像 MongoDB 這樣的通用數(shù)據(jù)庫(kù)所具備的廣泛運(yùn)營(yíng)能力和社區(qū)支持。

• 成本效益與簡(jiǎn)化：通過(guò)將數(shù)據(jù)存儲(chǔ)和向量搜索能力整合到單一數(shù)據(jù)庫(kù)中，你可以降低運(yùn)營(yíng)開銷和成本。無(wú)需管理和擴(kuò)展獨(dú)立的數(shù)據(jù)庫(kù)系統(tǒng)，從而簡(jiǎn)化基礎(chǔ)設(shè)施并優(yōu)化開發(fā)流程。

本質(zhì)上，專用向量數(shù)據(jù)庫(kù)只擅長(zhǎng)單一任務(wù)，而 MongoDB 提供的是端到端解決方案：在同一集成平臺(tái)內(nèi)，完成從數(shù)據(jù)攝取、存儲(chǔ)到高級(jí)查詢與分析的整個(gè)生命周期。

采用這一架構(gòu)，組織可將文檔處理能力從靜態(tài)批處理系統(tǒng)轉(zhuǎn)變?yōu)閷?shí)時(shí)動(dòng)態(tài)管道，從非結(jié)構(gòu)化內(nèi)容中提取有價(jià)值的洞察。云存儲(chǔ)、流式處理、機(jī)器學(xué)習(xí)與靈活存儲(chǔ)的結(jié)合，為以文檔為中心、驅(qū)動(dòng)業(yè)務(wù)價(jià)值的應(yīng)用奠定了強(qiáng)大基礎(chǔ)。

來(lái)源：https://www.mongodb.com/company/blog/technical/building-scalable-document-processing-pipeline-llamaparse-confluent-cloud