如何在LangChain中使用已部署的FastAPI嵌入模型构建检索器

核心问题说明 #

你提供的代码直接将requests.post的响应对象传给向量存储，这不符合LangChain对嵌入模型的要求——LangChain需要的是实现了Embeddings抽象类的实例，而非HTTP响应对象。正确的做法是自定义一个嵌入类，封装对FastAPI接口的调用。

实现步骤与代码示例 #

1. 自定义FastAPI嵌入类
   继承LangChain的Embeddings基类，实现embed_documents（批量嵌入文档）和embed_query（嵌入单个查询）两个核心方法，内部通过HTTP请求调用你的FastAPI嵌入服务：

from langchain_core.embeddings import Embeddings
import requests
from typing import List

class FastAPIEmbeddings(Embeddings):
    def __init__(self, api_url: str):
        self.api_url = api_url

    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        # 调用FastAPI接口批量嵌入文档
        response = requests.post(
            self.api_url,
            json={"texts": texts}
        )
        response.raise_for_status()  # 捕获请求异常
        return response.json()["embeddings"]

    def embed_query(self, text: str) -> List[float]:
        # 调用FastAPI接口嵌入单个查询文本
        response = requests.post(
            self.api_url,
            json={"texts": [text]}
        )
        response.raise_for_status()
        return response.json()["embeddings"][0]

2. 初始化嵌入模型并构建检索器
   使用自定义的嵌入类实例，结合LangChain的向量存储（以DocArrayInMemorySearch为例）构建检索器：

from langchain_community.vectorstores import DocArrayInMemorySearch

# 初始化自定义嵌入模型，替换为你的FastAPI服务地址
embeddings_model = FastAPIEmbeddings(api_url="http://internal-server/embeddings/")

# 从分块文档创建向量存储
db = DocArrayInMemorySearch.from_documents(chunked_docs, embeddings_model)

# 转为检索器
retriever = db.as_retriever()

注意事项 #

• 确保你的FastAPI服务接受{"texts": [str]}格式的POST请求，并返回包含embeddings字段的JSON响应（embeddings是二维浮点数组）。
• 可以根据需要添加请求超时、重试逻辑，或者在FastAPI服务端添加API密钥验证，对应在自定义类中添加请求头。

内容的提问来源于stack exchange，提问作者Christian