目标

了解EnsembleRetriever的使用场景

EnsembleRetriever是一个工具，它将多个检索器的列表作为输入，并将这些检索器返回的相关文档结果进行整合。然后，它使用Reciprocal Rank Fusion算法对这些结果进行重新排序。 通过结合不同算法的优点，EnsembleRetriever的表现可以超过任何一个单独的算法。 最常见的方法是将一个基于稀疏特征的检索器（比如BM25）与一个基于密集特征的检索器（比如嵌入向量相似性）结合起来，因为它们的优点可以互相补充。这种方法也被称为“混合搜索”。稀疏检索器擅长根据关键词找到相关文档，而密集检索器则擅长根据语义相似度找到相关文档。

BM25搜索 + 向量搜索

ElasticSearch BM25： langchain 实现的ES 数据库的BM25搜索

直接调用Langchain 的 ElasticSearchBM25Retriever 搜索方法，什么也查不到！？

代码分析：

就是调用了ES的 search 方法，在 content 这个域搜索query

1
2
3
4
5
6
7
8
9
10
def _get_relevant_documents(
    self, query: str, *, run_manager: CallbackManagerForRetrieverRun
 ) -> List[Document]:
     query_dict = {"query": {"match": {"content": query}}}
     res = self.client.search(index=self.index_name, body=query_dict)

     docs = []
     for r in res["hits"]["hits"]:
         docs.append(Document(page_content=r["_source"]["content"]))
     return docs

回忆之前我们直接调用ES 底层进行搜素：

1
2
3
4
5
6
7
8
9
query = 'Susirial'
query_dict = {"query": {"match": {"text": query}}}
res = vectorstore.client.search(index='index_sd_1024_vectors', body=query_dict)

docs = []
for r in res["hits"]["hits"]:
    docs.append(Document(page_content=r['_source']['text']))
    print('找到： {}'.format(r['_source']['text']))
pass

{“query”: {“match”: {“content”: query}}} VS {“query”: {“match”: {“text”: query}}}

重点： vectorstore.add_documents()

add_documents 方法默认把我们的数据上传到 数据的’text’ 这个域

我们需要修改 Langchain 的 ElasticSearch BM25 的搜索函数

注意： 下面这个方法只适用于 ES

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
#原来
def _get_relevant_documents(
    self, query: str, *, run_manager: CallbackManagerForRetrieverRun
) -> List[Document]:
    query_dict = {"query": {"match": {"content": query}}}
    res = self.client.search(index=self.index_name, body=query_dict)

    docs = []
    for r in res["hits"]["hits"]:
        docs.append(Document(page_content=r["_source"]["content"]))
    return docs
  
# 修改为
def _get_relevant_documents(
      self, query: str, *, run_manager: CallbackManagerForRetrieverRun
  ) -> List[Document]:
      query_dict = {"query": {"match": {"text": query}}}
      res = self.client.search(index=self.index_name, body=query_dict)

      docs = []
      for r in res["hits"]["hits"]:
          docs.append(Document(page_content=r["_source"]["text"]))
      return docs

搜索内容对比，比如我们搜 ‘ReAct’

BM25:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
vectorstore = ElasticsearchStore(
    es_url=os.environ['ELASTIC_HOST_HTTP'],
    index_name="index_sd_1024_vectors",
    embedding=embeddings_model,
    es_user="elastic",
    vector_query_field='question_vectors',
    es_password=os.environ['ELASTIC_ACCESS_PASSWORD']
)

# 建立 BM25 检索器
bm25_retriever = ElasticSearchBM25Retriever(client=vectorstore.client, index_name="index_sd_1024_vectors")

# 测试 BM25
bm25_docs = bm25_retriever.get_relevant_documents("ReAct")

向量搜索

1
2
3
4
5
6
7
8
9
10
vectorstore = ElasticsearchStore(
  es_url=os.environ['ELASTIC_HOST_HTTP'],
  index_name="index_sd_1024_vectors",
  embedding=embeddings_model,
  es_user="elastic",
  vector_query_field='question_vectors',
  es_password=os.environ['ELASTIC_ACCESS_PASSWORD']
)  

vector_docs = vector_retriever.get_relevant_documents("ReAct ")

文本块在分割时，按照chunk_size=500

1
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=0)

关键字搜索准确度，BM25 搜索结果 优于 向量搜索

EnsembleRetriever： BM25搜索 + 向量搜索

设置权重weights=[0.5, 0.5]

1
2
3
4
5
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever], weights=[0.5, 0.5]
)

ensemble_docs = ensemble_retriever.invoke("what is a LLM-powered autonomous agent system")

用Agent 轻松实现复杂逻辑：先搜向量数据库，没有相关内容，再BM25搜索

安装：langchainhub （从langsmith 上拿提示词，也可自己写）

1
pip install langchainhub

只需要写提示词

你是茉卷知识库管理员,你掌握2个工具:search_vector_docs, 功能:在茉卷向量数据库中,搜索问题,并返回文档列表,里面是Document对象;search_bm25_docs,功能:在茉卷BM25数据库中,搜索问题,并返回文档列表,里面是Document对象.你需要先在茉卷向量数据库中进行搜索,如果搜到的内容无法回答用户的问题,你再到茉卷BM25数据库中搜索.用户的问题是:ReAct“

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
# chat 模型
os.environ["AZURE_OPENAI_API_KEY"] = os.getenv('MY_AZURE_OPENAI_API_KEY')
os.environ["AZURE_OPENAI_ENDPOINT"] = os.getenv('MY_AZURE_OPENAI_ENDPOINT')
DEPLOYMENT_NAME_GPT3P5 = os.getenv('MY_DEPLOYMENT_NAME_GPT3P5')
chat = AzureChatOpenAI(
    openai_api_version="2023-05-15",
    azure_deployment=DEPLOYMENT_NAME_GPT3P5,
    temperature=0
)

# 向量检索器
vector_tool = create_retriever_tool(
    vector_retriever,
    "search_vector_docs",
    "在茉卷向量数据库中,搜索问题,并返回文档列表,里面是Document对象",
)

# bm25 检索器
bm25_tool = create_retriever_tool(
    bm25_retriever,
    "search_bm25_docs",
    "在茉卷BM25数据库中,搜索问题,并返回文档列表,里面是Document对象",
)

tools = [vector_tool,bm25_tool]

agent = create_openai_tools_agent(chat, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)
  user_input = ("你是茉卷知识库管理员,你掌握2个工具:search_vector_docs, 功能:在茉卷向量数据库中,搜索问题,并返回文档列表,里面是Document对象;"
              "search_bm25_docs,功能:在茉卷BM25数据库中,搜索问题,并返回文档列表,里面是Document对象."
              "你需要先在茉卷向量数据库中进行搜索,如果搜到的内容无法回答用户的问题,你再到茉卷BM25数据库中搜索."
              "用户的问题是:ReAct")
res = agent_executor.invoke({"input": user_input})

Agent 按照我们的要求执行，得到了结果

1
{'input': '你是茉卷知识库管理员,你掌握2个工具:search_vector_docs, 功能:在茉卷向量数据库中,搜索问题,并返回文档列表,里面是Document对象;search_bm25_docs,功能:在茉卷BM25数据库中,搜索问题,并返回文档列表,里面是Document对象.你需要先在茉卷向量数据库中进行搜索,如果搜到的内容无法回答用户的问题,你再到茉卷BM25数据库中搜索.用户的问题是:ReAct', 'output': '根据在茉卷BM25数据库中的搜索结果，找到了关于"ReAct"的文档。这些文档提到了ReAct模型，它在LLM中集成了推理和行动，并扩展了行动空间，使其成为任务特定离散行动和语言空间的组合。ReAct模型通过与环境交互（例如使用维基百科搜索API）和生成自然语言推理轨迹的方式，使LLM能够进行推理和行动。ReAct模型还提供了一个模板，用于指导LLM进行思考、行动和观察。\n\n如果在茉卷向量数据库中的搜索结果无法回答用户的问题，可以尝试在茉卷BM25数据库中进行搜索。在BM25数据库中的搜索结果显示，ReAct模型在知识密集型任务和决策任务上的实验中表现优于仅行动的基准模型。\n\n另外，还有一个案例研究提到了一个名为ChemCrow的领域特定示例，其中LLM通过13个专家设计的工具来完成有机合成、药物发现和材料设计等任务。这个案例研究中的工作流程与之前描述的ReAct和MRKLs相一致，并结合了与任务相关的工具和CoT推理。\n\n希望以上信息能够帮助回答用户的问题。如果还有其他问题，请随时提问。'}

完整代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
import os
from uuid import uuid4

from langchain import hub
from langchain.agents import create_openai_tools_agent, AgentExecutor
from langchain.retrievers import EnsembleRetriever
from langchain.tools.retriever import create_retriever_tool
from langchain_community.chat_models.azure_openai import AzureChatOpenAI
from langchain_community.document_loaders.web_base import WebBaseLoader

from langchain_community.embeddings import HuggingFaceEmbeddings, QianfanEmbeddingsEndpoint
from langchain_community.retrievers import ElasticSearchBM25Retriever
from langchain_community.vectorstores.elasticsearch import ElasticsearchStore
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 设置一个随机ID
unique_id = uuid4().hex[0:8]

# 设置一个Log 名称
os.environ["LANGCHAIN_PROJECT"] = f" [Agent]BM25+向量  - {unique_id}"

# 设置 生成Langsmith 轨迹
os.environ["LANGCHAIN_TRACING_V2"] = 'true'

# 填写你的 API KEY
os.environ["LANGCHAIN_API_KEY"] = os.getenv('MY_LANGCHAIN_API_KEY')

if __name__ == '__main__':

    # 千帆 bge_large_en
    os.environ["QIANFAN_ACCESS_KEY"] = os.getenv('MY_QIANFAN_ACCESS_KEY')
    os.environ["QIANFAN_SECRET_KEY"] = os.getenv('MY_QIANFAN_SECRET_KEY')

    embeddings_model = QianfanEmbeddingsEndpoint(model="bge_large_en", endpoint="bge_large_en")

    # 使用ES
    vectorstore = ElasticsearchStore(
        es_url=os.environ['ELASTIC_HOST_HTTP'],
        index_name="index_sd_1024_vectors",
        embedding=embeddings_model,
        es_user="elastic",
        vector_query_field='question_vectors',
        es_password=os.environ['ELASTIC_ACCESS_PASSWORD']
    )

    # 解析并载入url
    loader = WebBaseLoader("https://lilianweng.github.io/posts/2023-06-23-agent/")
    data = loader.load()

    # # 分割文本
    text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=0)
    splits = text_splitter.split_documents(data)

    # 写入数据库，运行一次就行
    #vectorstore.add_documents(splits)

    # 建立向量检索器
    vector_retriever = vectorstore.as_retriever(search_kwargs={'k': 10})

    # 建立 BM25 检索器
    bm25_retriever = ElasticSearchBM25Retriever(client=vectorstore.client, index_name="index_sd_1024_vectors",search_kwargs={'k': 10})

    # 测试 BM25
    # bm25_docs = bm25_retriever.get_relevant_documents("what is a LLM-powered autonomous agent system")
    # pass

    # 测试 向量搜索
    # vector_docs = vector_retriever.get_relevant_documents("what is a LLM-powered autonomous agent system")
    # pass

    ensemble_retriever = EnsembleRetriever(
        retrievers=[bm25_retriever, vector_retriever], weights=[0.5, 0.5]
    )

    # ensemble_docs = ensemble_retriever.invoke("what is a LLM-powered autonomous agent system")
    pass

    vector_tool = create_retriever_tool(
        vector_retriever,
        "search_vector_docs",
        "在茉卷向量数据库中,搜索问题,并返回文档列表,里面是Document对象",
    )

    bm25_tool = create_retriever_tool(
        bm25_retriever,
        "search_bm25_docs",
        "在茉卷BM25数据库中,搜索问题,并返回文档列表,里面是Document对象",
    )

    tools = [vector_tool,bm25_tool]

    prompt = hub.pull("hwchase17/openai-tools-agent")

    # 定义agent
    os.environ["AZURE_OPENAI_API_KEY"] = os.getenv('MY_AZURE_OPENAI_API_KEY')
    os.environ["AZURE_OPENAI_ENDPOINT"] = os.getenv('MY_AZURE_OPENAI_ENDPOINT')
    DEPLOYMENT_NAME_GPT3P5 = os.getenv('MY_DEPLOYMENT_NAME_GPT3P5')
    chat = AzureChatOpenAI(
        openai_api_version="2023-05-15",
        azure_deployment=DEPLOYMENT_NAME_GPT3P5,
        temperature=0
    )
    agent = create_openai_tools_agent(chat, tools, prompt)
    agent_executor = AgentExecutor(agent=agent, tools=tools)
    user_input = ("你是茉卷知识库管理员,你掌握2个工具:search_vector_docs, 功能:在茉卷向量数据库中,搜索问题,并返回文档列表,里面是Document对象;"
                  "search_bm25_docs,功能:在茉卷BM25数据库中,搜索问题,并返回文档列表,里面是Document对象."
                  "你需要先在茉卷向量数据库中进行搜索,如果搜到的内容无法回答用户的问题,你再到茉卷BM25数据库中搜索."
                  "用户的问题是:ReAct")
    res = agent_executor.invoke({"input": user_input})
    print(res)