如何通过LangChain查询Elasticsearch现有索引数据实现RAG

1. 安装依赖 #

先安装所需的Python包：

pip install langchain langchain-elasticsearch elasticsearch langchain-openai

注：如果不用OpenAI，可替换为对应LLM的包，比如langchain-google-genai或本地模型相关依赖

2. 连接Elasticsearch并创建检索器 #

LangChain的ElasticsearchRetriever可直接对接已有索引，无需重新写入数据。以下是适配你现有索引的代码：

from elasticsearch import Elasticsearch
from langchain_elasticsearch import ElasticsearchRetriever

# 1. 连接Elasticsearch集群（和你的原生代码一致）
es_client = Elasticsearch(['http://10.100.102.107:9200'])

# 2. 创建检索器：指定索引名和查询逻辑
# 这里用match查询匹配text字段，可根据你的字段名和需求修改
retriever = ElasticsearchRetriever.from_es_client(
    es_client=es_client,
    index_name="instagram_index",
    query_body="{\"query\": {\"match\": {\"text\": \"{question}\"}}}"
)

# 测试检索器：输入问题获取相关文档
docs = retriever.invoke("找一下提到美食的帖子")
for doc in docs:
    print(doc.page_content)

3. 构建完整RAG问答流程 #

将检索器与LLM结合，实现基于索引数据的问答：

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 1. 初始化LLM（替换为你的API密钥或本地模型）
llm = ChatOpenAI(model="gpt-3.5-turbo", api_key="你的API密钥")

# 2. 定义问答Prompt模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是基于Instagram帖子的问答助手，只能根据提供的上下文信息回答问题。上下文：\n{context}"),
    ("human", "{question}")
])

# 3. 组装RAG链
rag_chain = (
    {"context": retriever, "question": lambda x: x}
    | prompt
    | llm
    | StrOutputParser()
)

# 测试问答
response = rag_chain.invoke("哪些用户提到了披萨？")
print(response)

4. 自定义优化 #

• 多字段上下文展示：如果需要在返回结果中包含用户名等字段，可自定义page_content_field：

retriever = ElasticsearchRetriever.from_es_client(
    es_client=es_client,
    index_name="instagram_index",
    query_body="{\"query\": {\"match\": {\"text\": \"{question}\"}}}",
    page_content_field=lambda hit: f"用户: {hit['_source']['username']}\n内容: {hit['_source']['text']}"
)

• 复杂查询逻辑：支持布尔查询、时间过滤等复杂Elasticsearch语法，比如只查询最近30天的帖子：

query_body = """
{
    "query": {
        "bool": {
            "must": [{"match": {"text": "{question}"}}],
            "filter": [{"range": {"timestamp": {"gte": "now-30d/d"}}}]
        }
    }
}
"""

• 向量检索适配：如果你的索引存储了向量字段，可改用ElasticsearchVectorStore实现语义检索：

from langchain_openai import OpenAIEmbeddings
from langchain_elasticsearch import ElasticsearchVectorStore

embeddings = OpenAIEmbeddings()
vector_store = ElasticsearchVectorStore(
    es_client=es_client,
    index_name="instagram_index",
    embedding=embeddings,
    vector_field="embedding"  # 替换为你的向量字段名
)
retriever = vector_store.as_retriever()

内容的提问来源于stack exchange，提问作者Sardar Muhammad Raheem Tariq