使用 Redpanda、OpenAI 和 MongoDB 在流文本上生成向量嵌入

大家好，欢迎来到IT知识分享网。

使用 Redpanda、OpenAI 和 MongoDB 在流文本上生成向量嵌入

用于检索增强生成（RAG）的流式文本嵌入

这不仅仅是另一篇关于检索增强生成（RAG）的文章。它更侧重于如何在流数据上创建文本嵌入。我将提供一个示例，展示如何将文档流式传输到 Redpanda（一个功能强大的兼容Kafka的流数据平台），然后使用 Redpanda Connect 通过OpenAI的文本嵌入来丰富这些文档。Redpanda Connect是我们在 Redpanda收购Benthos（一个简单的流处理框架）后宣布的一款新产品。

示例随后将丰富的文档添加到向量数据库中，以通过查询相关文本来补充检索管道，从而为大型语言模型（LLM）提示添加上下文。在深入之前，让我们建立一些背景。

快速了解RAG

OpenAI将RAG定义为如下：

“RAG是从数据源中检索相关上下文信息并将该信息与用户的提示一起传递给大型语言模型的过程。这些信息用于通过增强模型的基础知识来改善模型的输出。”

RAG 包含两部分：

0. 获取和持久化LLM之前没有知识的新信息。这些新信息（文档、网页等）被拆分成较小的文本块，并与其向量嵌入一起存储在向量数据库中。这就是我们将使用流处理的部分。
1. 从向量数据库中检索相关的上下文信息（语义搜索），并将该上下文与用户提示一起传递给LLM，以提高模型生成的答案。

向量嵌入是文本的数学表示，能够在多维向量空间中编码其语义意义。具有相似含义的词在这空间中聚集得更近（即它们具有相似的坐标）。向量数据库能够查询向量嵌入，以检索具有相似语义意义的文本。一个著名的例子是 王 – 男人 + 女人 = 女王。 — 格兰特·桑德森

现在上下文已经建立（双关语），让我们通过一个示例来探讨如何构建流处理管道。本文中包含的所有代码可以在这个GitHub中找到：jrkinley/benthos-embeddings。

使用 Redpanda、OpenAI 和 MongoDB 构建流处理管道

要使其正常工作，有几个前提条件，但所有说明都在README中提供，且易于遵循。总之，此示例使用了三个在线服务，我们需要每个服务的身份验证凭据：

0. OpenAI API: 用于请求文本嵌入和模型生成。
1. Redpanda Serverless: 用于快速流式传输文本文件。
2. MongoDB Atlas: 用于向量存储和语义搜索。

克隆 GitHub 仓库，并在 demo 目录中添加所需的凭据到名为 .env 的文件中，如下所示：

 REDPANDA_SERVERS=" 
  
    " REDPANDA_USER=" 
   
     " REDPANDA_PASS=" 
    
      " REDPANDA_TOPICS="documents" OPENAI_API_KEY=" 
     
       " OPENAI_EMBEDDING_MODEL="text-embedding-3-small" OPENAI_MODEL="gpt-4o" ATLAS_CONNECTION_STRING="mongodb+srv:// 
      
        : 
       
         @<...>.mongodb.net/?retryWrites=false" ATLAS_DB="VectorStore" ATLAS_COLLECTION="Embeddings" ATLAS_INDEX="vector_index" 
        
       
      
     
    
  

在正式开始之前，完成设置的最后一步是创建一个Python虚拟环境并安装requirements.txt中列出的包：

% cd demo % python3 -m venv env % source env/bin/activate (env) % pip install -r requirements.txt

1. 查询 ChatGPT 以获取基准

我们在这个例子中试图实现的目标是展示如何使用流式文本来改善来自ChatGPT的生成响应。2024年5月13日，OpenAI发布了其最新模型 gpt-4o，在撰写本文时，该模型已经训练了截至2023年10月的数据。

所以模型不太可能对这个日期之后的时事有任何直觉，因此不太可能对提及近期事件的问题生成有意义的回应。这就是为什么RAG存在的原因！

首先，让我们询问ChatGPT，以获得针对最近事件问题的基准答案。2024年5月21日，英格兰国家足球队宣布了其即将参加的欧洲足球锦标赛2024比赛的临时男队名单。

让我们请ChatGPT列出队伍。

(env) % python retrieve_generate.py -q """ 哪些足球运动员被列入2024年欧洲杯的英格兰国家队临时名单，谁没有入选？""" 初步答案： 截至我上次更新的2023年10月，2024年欧洲杯的英格兰国家队临时名单尚未公布。 像欧洲杯这样的大型赛事的阵容选择通常是在赛事开始前的几个月进行， 通常是在赛事年的春季或初夏。 要获取关于临时阵容的最准确和最新的信息，以及哪些球员入选或被 排除，我建议查看官方来源的最新消息，例如足协（FA）网站、 可靠的体育新闻媒体，或英格兰国家队的官方社交媒体渠道。 这些来源将提供关于球员选择的最新细节以及与阵容相关的任何更新。

如预期的那样，ChatGPT 对这些信息没有先前的知识，因此对这个问题生成了一个有帮助但有些通用的回答。现在，让我们构建流式 RAG 管道，以增强 ChatGPT 的知识背景，并看看它是否能够生成改进的答案。

2. 使用LangChain将文本流式传输到Redpanda

LangChain是一个构建AI应用的框架和工具集合。

在这个例子中，我们使用LangChain的WebBaseLoader从BBC体育网站加载一个HTML页面，该页面记录了英格兰的欧洲杯2024阵容公告，以及关于阵容中惊讶缺席者的一些其他新闻。LangChain使用Python的BeautifulSoup库解析HTML以提取文本。

from langchain_community.document_loaders import WebBaseLoader from langchain.text_splitter import RecursiveCharacterTextSplitter ... loader = WebBaseLoader(url) loader.requests_kwargs = {"verify": True} docs = loader.load() ... splitter = RecursiveCharacterTextSplitter( chunk_size=2000, chunk_overlap=200) chunks = splitter.split_documents(docs)

使用LangChain的RecursiveCharacterTextSplitter将文本拆分为更小的块，以确保它适合ChatGPT的上下文窗口，然后将这些块发送到Redpanda，准备添加文本嵌入。

(env) % python produce_documents.py \ --url "https://www.bbc.co.uk/sport/football/articles/c3gglr8mpzdo" 准备了 7 个文档 正在生成到 Kafka: 100%|#| 7/7 [00:00<00:00, 17199.84it/s] 已发送 7 个文档到 Kafka 主题: documents

3. 使用 Redpanda Connect 生成文本嵌入

现在，将文本片段流式传输到Redpanda主题中后，下一步是使用Benthos（现称为Redpanda Connect）来消费该主题中的事件，并通过一个流处理器传递文本值，该处理器会调用OpenAI的嵌入API以获取向量嵌入。然后，将丰富后的事件（文本 + 嵌入 + 元数据）插入到MongoDB Atlas数据库集合中，在那里它们被索引并可用于语义搜索。

% export $(grep -v '^#' ./demo/.env | xargs) % go test PASS ok benthos-embeddings 2.837s % go build % ./benthos-embeddings -c demo/rag_demo.yaml --log.level debug INFO 从指定文件运行主配置 @service=benthos benthos_version=v4.27.0 path=demo/rag_demo.yaml INFO 正在监听HTTP请求: http://0.0.0.0:4195 @service=benthos DEBU url: https://api.openai.com/v1/embeddings, model: text-embedding-3-small @service=benthos label="" path=root.pipeline.processors.0 INFO 启动了一个benthos实例, 使用CTRL+C关闭 @service=benthos INFO 输入类型kafka现在已激活 @service=benthos label="" path=root.input DEBU 启动消费者组 @service=benthos label="" path=root.input INFO 输出类型mongodb现在已激活 @service=benthos label="" path=root.output DEBU 正在从主题'documents'分区'0'消费消息 @service=benthos label="" path=root.input 

请注意，Benthos 配置 rag_demo.yaml 使用了变量插值，因此跳过设置 demo/.env 中列出的环境变量的 export 命令是很重要的。

这完成了RAG管道的第一部分。文本现在正在流入Redpanda，正在从OpenAI获取文本嵌入进行增强，并被插入到向量存储中，其中数据库包含如下所示的文档：

_id: c20b49dd898e29105 embedding: Array (1536) metadata: Object description: "马库斯·拉什福德和乔丹·亨德森被排除在加areth·索斯盖特的队伍之外…" document_id: "TtX1mLEFGYEVXYG2Ue0yfJC3OeKTmrsNbt/F5eIhGlE=" language: "en-GB" source: "https://www.bbc.co.uk/sport/football/articles/c3gglr8mpzdo" title: "英格兰欧元2024阵容：马库斯·拉什福德和乔丹·亨德森被排除在外…" text: "英格兰欧元2024阵容：马库斯·拉什福德和乔丹·亨德森被排除在外…"

embedding 字段是一个包含 1,536 个双精度浮点数的数组，这是 OpenAI 的 text-embedding-3-small 模型的嵌入向量的长度。

4. 增强模型生成

拼图的最后一块是将原始问题重新发送给ChatGPT，但这次通过从向量存储中检索相似的文本，将一些相关的上下文添加到提示中。这个示例再次使用LangChain来帮助我们构建检索链：

from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser from langchain_mongodb import MongoDBAtlasVectorSearch from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain.chains.chat_vector_db import prompts vector_search = MongoDBAtlasVectorSearch.from_connection_string( os.getenv("ATLAS_CONNECTION_STRING"), f"{os.getenv("ATLAS_DB")}.{os.getenv("ATLAS_COLLECTION")}", OpenAIEmbeddings(model=os.getenv("OPENAI_EMBEDDING_MODEL")), index_name=os.getenv("ATLAS_INDEX") ) retriever = vector_search.as_retriever( search_type="similarity", search_kwargs={"score_threshold": 0.75} ) llm = ChatOpenAI(model=os.getenv("OPENAI_MODEL")) def format_docs(docs): return "\n".join(doc.page_content for doc in docs) # 构建 RAG 链 rag_chain = ( {"context": retriever | format_docs, "question": RunnablePassthrough()} | prompts.QA_PROMPT | llm | StrOutputParser() ) answer = rag_chain.invoke(query) 

我们能得到更好的答案吗？

是的！ChatGPT使用提示中提供的上下文来生成一个回应，列出进入球队的英格兰足球运动员，以及那些未能入选的显著球员：

(env) % python retrieve_generate.py -q """ 哪位足球运动员入选了2024年欧洲杯 英格兰国家队的初选大名单，谁又未能入选？""" 增强答案： 入选2024年欧洲杯英格兰国家队初选大名单的足球运动员有： 门将： - Dean Henderson (水晶宫) - Jordan Pickford (埃弗顿) - Aaron Ramsdale (阿森纳) - James Trafford (伯恩利) 后卫： - Jarrad Branthwaite (埃弗顿) - Lewis Dunk (布赖顿) - Joe Gomez (利物浦) - Marc Guehi (水晶宫) - Ezri Konsa (阿斯顿维拉) - Harry Maguire (曼联) - Jarell Quansah (利物浦) - Luke Shaw (曼联) - John Stones (曼城) - Kieran Trippier (纽卡斯尔) - Kyle Walker (曼城) 中场： - Trent Alexander-Arnold (利物浦) - Conor Gallagher (切尔西) - Curtis Jones (利物浦) - Kobbie Mainoo (曼联) - Declan Rice (阿森纳) - Adam Wharton (水晶宫) 前锋： - Jude Bellingham (皇家马德里) - Jarrod Bowen (西汉姆联) - Eberechi Eze (水晶宫) - Phil Foden (曼城) - Jack Grealish (曼城) - Anthony Gordon (纽卡斯尔) - Harry Kane (拜仁慕尼黑) - James Maddison (热刺) - Cole Palmer (切尔西) - Bukayo Saka (阿森纳) - Ivan Toney (布伦特福德) - Ollie Watkins (阿斯顿维拉) 未能入选的 notable 球员： - Marcus Rashford - Jordan Henderson - Ben Chilwell - Eric Dier - Reece James - Jadon Sancho - Dominic Solanke - Raheem Sterling - Ben White

结论

如果你从这篇文章中领悟到一件事，那就是RAG管道可以，并且可以说应该包含一个流数据组件。信息不会静止不动，你能够多快地收集、准备并使新信息可用于提示工程，你选择的LLM生成的结果就会更加准确和相关。

这里呈现的例子有效但简单。我们完全可以仅使用 LangChain 和 MongoDB 组件构建管道，结果将是相同的，但这并不是我们的想法。新闻网站、社交媒体应用程序和其他基于事件的系统可以以惊人的速度生成信息，而你的 RAG 管道将达到一个拐点，你需要解耦这些组件以处理生成的大量数据。

将流数据平台插入到您的RAG架构中可以解耦数据收集、处理和存储组件。像Redpanda这样的工具提供一个简单、快速且持久的事件缓冲区，可以支撑您的端到端AI应用，确保它们在扩展时的未来适应性。