如何通过pysolr将Apache Solr集成为LangChain的向量存储？

1. Solr前置配置 #

先在Solr中创建支持向量存储的core，修改managed-schema（或schema.xml）添加向量相关字段：

<field name="embedding" type="dense_vector" indexed="true" stored="true" vectorDimension="1536"/> <!-- 匹配OpenAI ada-002的向量维度 -->
<field name="content" type="text_general" indexed="true" stored="true"/>
<field name="id" type="string" indexed="true" stored="true" required="true"/>

同时确保solrconfig.xml中启用了向量搜索组件，配置支持KNN查询的解析器。

2. 安装依赖 #

pip install pysolr langchain openai python-dotenv

3. 自定义Solr向量存储类 #

基于LangChain的VectorStore抽象类实现核心功能，通过pysolr完成Solr的读写操作：

import pysolr
from langchain.vectorstores.base import VectorStore
from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.schema import Document
import os
from dotenv import load_dotenv

load_dotenv()

class SolrVectorStore(VectorStore):
    def __init__(self, solr_url: str, embedding: OpenAIEmbeddings, core_name: str = "vector_core"):
        self.solr = pysolr.Solr(f"{solr_url}/{core_name}", always_commit=True)
        self.embedding = embedding

    def add_texts(self, texts: list[str], metadatas: list[dict] = None, **kwargs) -> list[str]:
        # 生成文本向量嵌入
        embeddings = self.embedding.embed_documents(texts)
        docs = []
        for idx, text in enumerate(texts):
            doc = {
                "id": f"doc_{idx}",
                "content": text,
                "embedding": embeddings[idx],
                **(metadatas[idx] if metadatas else {})
            }
            docs.append(doc)
        # 批量写入Solr
        self.solr.add(docs)
        return [f"doc_{idx}" for idx in range(len(texts))]

    def similarity_search(self, query: str, k: int = 4, **kwargs) -> list[Document]:
        # 生成查询语句的向量嵌入
        query_embedding = self.embedding.embed_query(query)
        # 构造Solr KNN相似度查询（余弦相似度）
        solr_query = f'{!knn f=embedding topK={k}}[{",".join(map(str, query_embedding))}]'
        results = self.solr.search(solr_query, **kwargs)
        # 转换为LangChain标准Document格式
        return [
            Document(page_content=res["content"], metadata={k: v for k, v in res.items() if k not in ["content", "embedding", "id"]})
            for res in results
        ]

    @classmethod
    def from_texts(cls, texts: list[str], embedding: OpenAIEmbeddings, solr_url: str, core_name: str = "vector_core", metadatas: list[dict] = None, **kwargs):
        store = cls(solr_url, embedding, core_name)
        store.add_texts(texts, metadatas)
        return store

4. 集成到LangChain聊天机器人 #

from langchain.chat_models import ChatOpenAI
from langchain.chains import RetrievalQA

# 初始化OpenAI嵌入模型
embeddings = OpenAIEmbeddings(openai_api_key=os.getenv("OPENAI_API_KEY"))

# 初始化Solr向量存储
solr_store = SolrVectorStore(
    solr_url="http://localhost:8983/solr",
    embedding=embeddings,
    core_name="vector_core"
)

# 首次运行时添加示例文本（后续可注释）
# texts = ["LangChain是用于构建大语言模型应用的开发框架", "Apache Solr是开源企业级搜索平台，支持向量检索功能"]
# solr_store.add_texts(texts)

# 创建检索器
retriever = solr_store.as_retriever(search_kwargs={"k": 2})

# 构建检索增强生成（RAG）链
qa_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(model="gpt-3.5-turbo"),
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True
)

# 测试查询
response = qa_chain({"query": "LangChain的作用是什么？"})
print(response["result"])
print("参考文档：", [doc.page_content for doc in response["source_documents"]])

关键注意事项 #

• 推荐使用Solr 9.x版本，确保对DenseVectorField和KNN查询的支持完整
• 嵌入模型的向量维度必须与Solr中vectorDimension配置值完全匹配
• 批量写入大数量文本时，可通过pysolr的batch参数分批次提交，避免性能问题
• 可根据需求调整Solr的相似度算法，替换查询语句中的KNN参数

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