基于LLM的文档搜索引擎开发【Ray+LangChain】

Ray 是一个非常强大的 ML 编排框架，但强大的功能伴随着大量的文档。事实上120兆字节。我们如何才能使该文档更易于访问？

答案：使其可搜索！过去，创建自己的高质量搜索结果很困难。但通过使用 LangChain，我们可以用大约 100 行代码来构建它。

这就是 LangChain 的用武之地。LangChain 为LLM相关的一切提供了一套令人惊叹的工具。它有点像 HuggingFace，但专门针对LLM。有用于提示、索引、生成和总结文本的工具（链）。 Ray虽然是一个令人惊叹的工具，但与它配合使用可以让LangChain变得更加强大。特别是，它可以：

简单快速地帮助你部署LangChain服务。
允许 Chains 与 LLM 本身共同运行并自动扩展，而不是依赖远程 API 调用。这带来了我们在之前的博客文章中讨论的所有优势：更低的成本、更低的延迟以及对数据的控制。

NSDT工具推荐： Three.js AI纹理开发包 - YOLO合成数据生成器 - GLTF/GLB在线编辑 - 3D模型格式在线转换 - 可编程3D场景编辑器 - REVIT导出3D模型插件 - 3D模型语义搜索引擎 - Three.js虚拟轴心开发包

1、建立LangChain索引

首先，我们将通过以下步骤构建索引。

下载我们想要在本地建立索引的内容。
阅读内容并将其切成小块（每块大约一个句子）。这是因为将查询与页面的各个部分而不是整个页面进行匹配更容易。
使用 HuggingFace 的 Sentence Transformers 库生成每个句子的向量表示。
将这些向量嵌入到向量数据库中（我们使用 FAISS，但您可以使用您喜欢的任何内容）。

这段代码的神奇之处在于它非常简单 - 请参阅此处。正如您将看到的，感谢浪链，所有繁重的工作都为我们完成了。我们来摘录几段吧。

假设我们已经下载了 Ray 文档，那么我们只需执行以下操作即可阅读以下位置的所有文档：

loader = ReadTheDocsLoader("docs.ray.io/en/master/")
docs = loader.load()

下一步是将每个文档分成小块。 LangChain 使用分割器来做到这一点。所以我们所要做的就是：

chunks = text_splitter.create_documents(
    [doc.page_content for doc in docs], 
    metadatas=[doc.metadata for doc in docs])

我们希望保留原始 URL 的元数据，因此我们确保将元数据与这些文档一起保留。

现在我们有了块，我们可以将它们嵌入为向量。 LLM 提供商确实提供了用于远程执行此操作的 API（这就是大多数人使用 LangChain 的方式）。但是，只需一点点胶水，我们就可以从 HuggingFace 下载 Sentence Transformers 并在本地运行它们（受到 LangChain 对 llama.cpp 支持的启发）。这是胶水代码。

通过这样做，我们可以减少延迟，继续使用开源技术，并且不需要 HuggingFace 密钥或支付 API 使用费用。

最后，我们有了嵌入，现在我们可以使用向量数据库（在本例中为 FAISS）来存储嵌入。矢量数据库经过优化，可以在高维空间中进行快速搜索。 LangChain 再次让这一切变得毫不费力。

from langchain.vectorstores import FAISS

db = FAISS.from_documents(chunks, embeddings)
db.save_local(FAISS_INDEX_PATH)

就是这样。代码在这里。现在我们可以建立存储库了。

% python build_vector_store.py

执行此操作大约需要 8 分钟。大部分时间都花在嵌入上。当然，在这种情况下这不是什么大问题，但想象一下，如果您要索引数百 GB 而不是数百兆字节。

2、使用 Ray 加速索引

[注意：这是一个稍微高级的主题，第一次阅读时可以跳过。它只是展示了我们如何更快地做到这一点 – 速度提高 4 倍到 8 倍]

我们怎样才能加快索引速度呢？伟大的事情是嵌入很容易并行化。如果我们：

将块列表切成 8 个分片。
分别嵌入 8 个分片。
合并分片。

需要意识到的一件关键事情是嵌入是 GPU 加速的，所以如果我们想做到这一点，我们需要 8 个 GPU。感谢 Ray，这 8 个 GPU 不必位于同一台机器上。但即使在单台机器上，使用 Ray 也有显着的优势。而且不必考虑设置 Ray 集群的复杂性，您所需要做的就是 pip install ray[default]，然后导入 ray。

这需要对代码进行一些小改动。这是我们必须做的。

首先，创建一个任务来创建嵌入，然后使用它来索引分片。请注意 Ray 注释，我们告诉我们每个任务都需要整个 GPU。

@ray.remote(num_gpus=1)
def process_shard(shard): 
    embeddings = LocalHuggingFaceEmbeddings('multi-qa-mpnet-base-dot-v1')
    result = FAISS.from_documents(shard, embeddings)
    return result

接下来，将工作负载拆分到分片中。 NumPy 来救援！这是一行：

shards = np.array_split(chunks, db_shards)

然后为每个分片创建一个任务并等待结果：

futures = [process_shard.remote(shards[i]) for i in range(db_shards)]
results = ray.get(futures)

最后，让我们将分片合并在一起。我们使用简单的线性合并来做到这一点。

db = results[0]
for i in range(1,db_shards):
    db.merge_from(results[i])

你可能想知道，这真的有效吗？我们在具有 8 个 GPU 的 g4dn.metal 实例上运行了一些测试。原始代码需要 313 秒才能创建嵌入，新代码需要 70 秒，大约提高了 4.5 倍。创建任务、设置 GPU 等仍然有一些一次性开销。随着数据的增加，这种情况会减少。例如，我们用 4 倍的数据做了一个简单的测试，它大约是理论最大性能的 80%（即 6.5 倍快，而 8 个 GPU 的理论最大速度快 8 倍）。

我们可以使用 Ray Dashboard 来查看这些 GPU 的工作强度。果然，它们都接近 100% 运行我们刚刚编写的 process_shard 方法。

事实证明，合并向量数据库的速度非常快，只需 0.3 秒即可合并所有 8 个分片。

3、服务提供

服务是LangChain和Ray Serve结合的另一个领域。这只是皮毛：我们将在本系列的下一篇文章中探讨独立自动扩展和请求批处理等惊人功能。

执行此操作所需的步骤如下：

加载我们创建的 FAISS 数据库并实例化嵌入
开始使用 FAISS 进行相似性搜索。

Ray Serve 让这一切变得神奇地简单。Ray 使用“部署”来包装一个简单的 python 类。 __init__方法执行加载，然后 __call__实际执行工作。Ray 负责将其连接到互联网，启动服务、http 等。

下面是代码的简化版本：

@serve.deployment
class VectorSearchDeployment:
    def __init__(self):
        self.embeddings = … 
        self.db = FAISS.load_local(FAISS_INDEX_PATH, self.embeddings)

    def search(self,query): 
        results = self.db.max_marginal_relevance_search(query)
        retval = <some string processing of the results>
        return retval

    async def __call__(self, request: Request) -> List[str]:
        return self.search(request.query_params["query"])

deployment = VectorSearchDeployment.bind()

就是这样！

现在让我们使用命令行启动此服务（当然 Serve 的部署选项比这更多，但这是一种简单的方法）：

% serve run serve_vector_store:deployment

现在我们可以编写一个简单的 python 脚本来查询服务以获取相关向量（它只是一个运行在端口 8000 上的 Web 服务器）。

import requests
import sys
query = sys.argv[1]
response = requests.post(f'http://localhost:8000/?query={query}')
print(response.content.decode())

现在让我们试试看：

$ python query.py 'Does Ray Serve support batching?'
From http://docs.ray.io/en/master/serve/performance.html

You can check out our microbenchmark instructions
to benchmark Ray Serve on your hardware.
Request Batching#
====

From http://docs.ray.io/en/master/serve/performance.html

You can enable batching by using the ray.serve.batch decorator. Let’s take a look at a simple example by modifying the MyModel class to accept a batch.
from ray import serve
import ray
@serve.deployment
class Model:
    def __call__(self, single_sample: int) -> int:
        return single_sample * 2
====

From http://docs.ray.io/en/master/ray-air/api/doc/ray.train.lightgbm.LightGBMPredictor.preferred_batch_format.html

native batch format.
DeveloperAPI: This API may change across minor Ray releases.
====

From http://docs.ray.io/en/master/serve/performance.html

Machine Learning (ML) frameworks such as Tensorflow, PyTorch, and Scikit-Learn support evaluating multiple samples at the same time.
Ray Serve allows you to take advantage of this feature via dynamic request batching.
====