Elasticsearch：使用 Gemini、Langchain 和 Elasticsearch 进行问答

本教程演示如何使用 Gemini API创建 embeddings 并将其存储在 Elasticsearch 中。我们将学习如何将 Gemini 连接到 Elasticsearch 中存储的私有数据，并使用 Langchian 构建问答功能。

准备

Elasticsearch 及 Kibana

如果你还没有安装好自己的 Elasticsearch 及 Kibana 的话，请参阅如下的文章来进行安装：

如何在 Linux，MacOS 及 Windows 上进行安装 Elasticsearch
Kibana：如何在 Linux，MacOS 及 Windows 上安装 Elastic 栈中的 Kibana

在安装的时候，请参照 Elastic Stack 8.x 的文章来进行安装。

Gemini 开发者 key

你可以参考文章来申请一个免费的 key 供下面的开发。你也可以直接去地址进行申请。

设置环境变量

我们在 termnial 中打入如下的命令来设置环境变量：

export ES_USER=elastic
export ES_PASSWORD=-M3aD_m3MHCZNYyJi_V2
export GOOGLE_API_KEY=YourGoogleAPIkey

拷贝 Elasticsearch 证书

我们把 Elasticsearch 的证书拷贝到当前的目录下：

$ pwd
/Users/liuxg/python/elser
$ cp ~/elastic/elasticsearch-8.12.0/config/certs/http_ca.crt .

安装 Python 依赖包

pip3 install -q -U google-generativeai elasticsearch langchain langchain_google_genai

应用设计

我们在当前的工作目录下打入命令：

jupyter notebook

导入包

import google.generativeai as genai
import google.ai.generativelanguage as glm
from elasticsearch import Elasticsearch, helpers
from langchain.vectorstores import ElasticsearchStore
from langchain.text_splitter import CharacterTextSplitter
from langchain_google_genai import GoogleGenerativeAIEmbeddings
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.prompts import ChatPromptTemplate
from langchain.prompts import PromptTemplate
from langchain.schema.output_parser import StrOutputParser
from langchain.schema.runnable import RunnablePassthrough
from langchain.schema.runnable import RunnableLambda
from langchain.schema import HumanMessage
from urllib.request import urlopen
from dotenv import load_dotenv
import json, os

读取环境变量

load_dotenv()

GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")
ES_USER = os.getenv("ES_USER")
ES_PASSWORD = os.getenv("ES_PASSWORD")
elastic_index_name='gemini-qa'

写入文档

让我们下载示例数据集并反序列化文档

我们首先在地址下载示例数据集：

wget https://raw.githubusercontent.com/liu-xiao-guo/semantic_search_es/main/datasets/data.json

其中的一个文档的内容如下：

$ pwd
/Users/liuxg/python/elser
$ ls datasets/
data.json

# Load data into a JSON object
with open('./datasets/data.json') as f:
   workplace_docs = json.load(f)

print(f"Successfully loaded {len(workplace_docs)} documents")

将文档拆分为段落

metadata = []
content = []

for doc in workplace_docs:
  content.append(doc["content"])
  metadata.append({
      "name": doc["name"],
      "summary": doc["summary"],
      "rolePermissions":doc["rolePermissions"]
  })

text_splitter = CharacterTextSplitter(chunk_size=50, chunk_overlap=0)
docs = text_splitter.create_documents(content, metadatas=metadata)

使用 Gemini Embeddings 将文档索引到 Elasticsearch

url = f"https://{ES_USER}:{ES_PASSWORD}@192.168.0.3:9200"

connection = Elasticsearch(
        hosts=[url], 
        ca_certs = "./http_ca.crt", 
        verify_certs = True
)
print(connection.info())

embeddings = GoogleGenerativeAIEmbeddings(
    model="models/embedding-001", task_type="retrieval_document"
)

es = ElasticsearchStore.from_documents( 
                            docs,
                            embedding = embeddings, 
                            es_url = url, 
                            es_connection = connection,
                            index_name = elastic_index_name, 
                            es_user = ES_USER,
                            es_password = ES_PASSWORD)

运行完上面的代码后，我们可以去 Kibana 中进行查看：

创建 retriever

更多搜索的方法可以参考 “Elasticsearch：使用 Open AI 和 Langchain 的 RAG - Retrieval Augmented Generation （四）”。

embeddings = GoogleGenerativeAIEmbeddings(
    model="models/embedding-001", task_type="retrieval_query"
)

retriever = es.as_retriever(search_kwargs={"k": 3})

如果你不用去写入文档（没有上一步的 es），那么你可以使用如下的方法创建 es：

es = ElasticsearchStore(
    es_connection=connection,
    embedding=embedding,
    index_name=elastic_index_name
)

格式化文档

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

使用 Prompt Template+gemini-pro 模型创建一条链

template = """Answer the question based only on the following context:\n

{context}

Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)


chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()} 
    | prompt 
    | ChatGoogleGenerativeAI(model="gemini-pro", temperature=0.7) 
    | StrOutputParser()
)

chain.invoke("what is our sales goals?")