简介¶

这是一款名为‘FunctionAgentWithRetrieval检索增强小助手’的演示应用，展示了如何结合LangChain的工具和ERNIEBot的function agent来回答用户的专业知识问题。首先，应用通过构建FaissSearch在知识库中检索相关内容，然后评估检索内容与问题的相关度。如果内容与问题高度相关，大模型将采用检索增强方式回答问题；否则，大模型将调用工具列表中的工具回答用户的问题。这种设计不仅扩展了大模型的专业领域知识，还保持了其在领域知识之外的通用对话能力

构建流程如下：

1. 使用langchain¶

In [ ]:

pip install langchain

pip install langchain

2. 导入第三方库¶

主要是在导入一些必要的Python库和模块，以便实现FunctionAgentWithRetrival的功能。

• os: Python的标准库，用于与操作系统进行交互，如读写文件、管理路径等。
• SpacyTextSplitter: 一个文本分割工具，用于将文本分割为更小的部分，如句子或短语。
• FAISS: 用于向量存储的模块，是用于存储和检索经过嵌入处理的文本或图像的向量表示。
• PyPDFDirectoryLoader: 用于从PDF文件中加载数据的工具。
• ErnieEmbeddings: 用于文本嵌入的工具，将文本转换为可以在模型中使用的向量表示。
• FunctionAgent: 这个类实现function calling功能的Agent的类，如问答、对话等。
• WholeMemory: 用于存储和管理代理的记忆的类。
• ERNIEBot: 实现ERNIE Bot的主要类，包含了实现对话功能的主要逻辑。
• cosine_similarity: 用于计算余弦相似度的函数

这里使用EB_AGENT_ACCESS_TOKEN, 申请地址请参考accessToken

In [14]:

import os
from erniebot_agent.agents.function_agent_with_retrieval import FunctionAgentWithRetrieval
from erniebot_agent.memory.whole_memory import WholeMemory
from erniebot_agent.chat_models.erniebot import ERNIEBot
from langchain.text_splitter import SpacyTextSplitter
from langchain.vectorstores import FAISS
from langchain.document_loaders import PyPDFDirectoryLoader
from erniebot_agent.extensions.langchain.embeddings import ErnieEmbeddings
from sklearn.metrics.pairwise import cosine_similarity
from erniebot_agent.tools import RemoteToolkit

os.environ["EB_AGENT_ACCESS_TOKEN"] = "your access token"

import os from erniebot_agent.agents.function_agent_with_retrieval import FunctionAgentWithRetrieval from erniebot_agent.memory.whole_memory import WholeMemory from erniebot_agent.chat_models.erniebot import ERNIEBot from langchain.text_splitter import SpacyTextSplitter from langchain.vectorstores import FAISS from langchain.document_loaders import PyPDFDirectoryLoader from erniebot_agent.extensions.langchain.embeddings import ErnieEmbeddings from sklearn.metrics.pairwise import cosine_similarity from erniebot_agent.tools import RemoteToolkit os.environ["EB_AGENT_ACCESS_TOKEN"] = "your access token"

3. 预处理¶

3.1 下载数据集¶

In [2]:

! wget https://paddlenlp.bj.bcebos.com/datasets/examples/construction_regulations.tar
! tar xvf construction_regulations.tar

! wget https://paddlenlp.bj.bcebos.com/datasets/examples/construction_regulations.tar ! tar xvf construction_regulations.tar

--2023-12-27 05:50:15--  https://paddlenlp.bj.bcebos.com/datasets/examples/construction_regulations.tar
Connecting to 172.19.57.45:3128... connected.
Proxy request sent, awaiting response... 200 OK
Length: 1239040 (1.2M) [application/x-tar]
Saving to: ‘construction_regulations.tar’

construction_regula 100%[===================>]   1.18M  94.2KB/s    in 14s     

2023-12-27 05:50:31 (89.5 KB/s) - ‘construction_regulations.tar’ saved [1239040/1239040]

construction_regulations/
construction_regulations/城市管理执法办法.pdf
construction_regulations/建筑工程设计招标投标管理办法.pdf
construction_regulations/建筑业企业资质管理规定.pdf
construction_regulations/城市照明管理规定.pdf
construction_regulations/城市设计管理办法.pdf
construction_regulations/建筑工程施工发包与承包计价管理办法.pdf
construction_regulations/市政公用设施抗灾设防管理规定.pdf

3.2 构建FaissSearch¶

这段代码定义了一个名为 FaissSearch 的类，用于在一个数据库中执行相似性搜索。这个类的主要功能是使用 FAISS (Facebook AI Similarity Search) 库或类似技术，对文档数据库进行高效的相似性搜索。它首先找到与查询最相似的文档，然后计算每个文档与查询的余弦相似度，最后返回包含内容、相似度得分和文档标题的搜索结果。

In [15]:

class FaissSearch:
    def __init__(self, db):
        # 类的初始化方法，接收一个数据库实例并将其存储在类的实例变量 self.db 中
        self.db = db

    def search(self, query: str, top_k: int = 10, **kwargs):
        # 定义一个搜索方法，接受一个查询字符串 'query' 和一个整数 'top_k'，默认为 10
        docs = self.db.similarity_search(query, top_k)
        # 调用数据库的 similarity_search 方法来获取与查询最相关的文档
        para_result = embeddings.embed_documents([i.page_content for i in docs])
        # 对获取的文档内容进行嵌入（embedding），以便进行相似性比较
        query_result = embeddings.embed_query(query)
        # 对查询字符串也进行嵌入
        similarities = cosine_similarity([query_result], para_result).reshape((-1,))
        # 计算查询嵌入和文档嵌入之间的余弦相似度
        retrieval_results = []
        for index, doc in enumerate(docs):
            retrieval_results.append(
                {"content": doc.page_content, "score": similarities[index], "title": doc.metadata["source"]}
            )
        # 遍历每个文档，将内容、相似度得分和来源标题作为字典添加到结果列表中
        return retrieval_results  # 返回包含搜索结果的列表

class FaissSearch: def __init__(self, db): # 类的初始化方法，接收一个数据库实例并将其存储在类的实例变量 self.db 中 self.db = db def search(self, query: str, top_k: int = 10, **kwargs): # 定义一个搜索方法，接受一个查询字符串 'query' 和一个整数 'top_k'，默认为 10 docs = self.db.similarity_search(query, top_k) # 调用数据库的 similarity_search 方法来获取与查询最相关的文档 para_result = embeddings.embed_documents([i.page_content for i in docs]) # 对获取的文档内容进行嵌入（embedding），以便进行相似性比较 query_result = embeddings.embed_query(query) # 对查询字符串也进行嵌入 similarities = cosine_similarity([query_result], para_result).reshape((-1,)) # 计算查询嵌入和文档嵌入之间的余弦相似度 retrieval_results = [] for index, doc in enumerate(docs): retrieval_results.append( {"content": doc.page_content, "score": similarities[index], "title": doc.metadata["source"]} ) # 遍历每个文档，将内容、相似度得分和来源标题作为字典添加到结果列表中 return retrieval_results # 返回包含搜索结果的列表

4.FunctionAgentWithRetrieval¶

4.1 建索引库¶

这段代码主要功能是用于创建或加载一个FAISS索引来进行文档相似度匹配。首先，创建一个ErnieEmbeddings对象，用于生成文档的嵌入向量。aistudio_access_token是访问AI Studio的令牌，chunk_size是用于嵌入的文档块的数量。

In [16]:

aistudio_access_token = os.environ.get("EB_AGENT_ACCESS_TOKEN", "")
embeddings = ErnieEmbeddings(aistudio_access_token=aistudio_access_token, chunk_size=16)

aistudio_access_token = os.environ.get("EB_AGENT_ACCESS_TOKEN", "") embeddings = ErnieEmbeddings(aistudio_access_token=aistudio_access_token, chunk_size=16)

接下来利用ErnieEmbeddings来抽取向量构建索引。

• 如果FAISS索引文件已经存在，就使用FAISS.load_local方法加载这个索引，这个索引文件的名字就是定义的faiss_name。
• 如果FAISS索引不存在，则需要建索引。
  • 第一步，使用PyPDFDirectoryLoader来从"construction_regulations"这个文件夹中加载PDF文档。
  • 第二步，使用SpacyTextSplitter来将加载的文档分割成更小的部分，以便于生成嵌入向量。这个分割器主要用于中文文本，因为这里使用的pipeline是'zh_core_web_sm'，如果是初次运行，需要安装spacy并运行python -m spacy download zh_core_web_sm来下载中文分句模型。
  • 第三步，通过分割后的文档创建一个新的FAISS索引，并将这个索引保存为之前定义的faiss_name。

In [17]:

embeddings = ErnieEmbeddings(aistudio_access_token=aistudio_access_token, chunk_size=16)
faiss_name = "faiss_index"
if os.path.exists(faiss_name):
    db = FAISS.load_local(faiss_name, embeddings)
else:
    loader = PyPDFDirectoryLoader("construction_regulations")
    documents = loader.load()
    text_splitter = SpacyTextSplitter(pipeline="zh_core_web_sm", chunk_size=320, chunk_overlap=0)
    docs = text_splitter.split_documents(documents)
    db = FAISS.from_documents(docs, embeddings)
    db.save_local(faiss_name)

embeddings = ErnieEmbeddings(aistudio_access_token=aistudio_access_token, chunk_size=16) faiss_name = "faiss_index" if os.path.exists(faiss_name): db = FAISS.load_local(faiss_name, embeddings) else: loader = PyPDFDirectoryLoader("construction_regulations") documents = loader.load() text_splitter = SpacyTextSplitter(pipeline="zh_core_web_sm", chunk_size=320, chunk_overlap=0) docs = text_splitter.split_documents(documents) db = FAISS.from_documents(docs, embeddings) db.save_local(faiss_name)

以下代码是使用FaissSearc进行搜索的一个例子，流程大致如下：

1. 创建FaissSearch对象，并传入数据库对象db。
2. 调用FaissSearch对象的search函数，并传入查询字符串"城市管理执法主管部门的职责是什么？"。
3. 将搜索结果以格式化的形式进行打印，这里使用了pprint模块进行美化打印。最终的结果存储在变量res中。

In [6]:

faiss_search = FaissSearch(db=db)
res = faiss_search.search(query="城市管理执法主管部门的职责是什么？")
from pprint import pprint

pprint(res)

faiss_search = FaissSearch(db=db) res = faiss_search.search(query="城市管理执法主管部门的职责是什么？") from pprint import pprint pprint(res)

[{'content': '第十条 城市管理执法主管部门依法相对集中行使行政处罚权的， 可以实施法律法规规定的与行政处罚权相关的行政强制措施。',
  'score': 0.6833730421296541,
  'title': 'construction_regulations/城市管理执法办法.pdf'},
 {'content': '第六条 城市管理执法主管部门应当加强城市管理法律法规规章的宣传普及工作，增强全民守法意识，共同维护城市管理秩序。\n'
             '\n'
             '  \n'
             '\n'
             '第七条 城市管理执法主管部门应当积极为公众监督城市管理执法活动提供条件。',
  'score': 0.645987793263916,
  'title': 'construction_regulations/城市管理执法办法.pdf'},
 {'content': '其他违反法律法规和本办法规定的。   第四十条 '
             '非城市管理执法人员着城市管理执法制式服装的，城市管理执法主管部门应当予以纠正，依法追究法律责任。',
  'score': 0.628888526926951,
  'title': 'construction_regulations/城市管理执法办法.pdf'},
 {'content': '第三章 执法主体     第十三条 城市管理执法主管部门按照权责清晰、事权统一、精简效能的原则设置执法队伍。   第十四条 '
             '直辖市、 设区的市城市管理执法推行市级执法或者区级执法。   '
             '直辖市、设区的市的城市管理执法事项，市辖区人民政府城市管理执法主管部门能够承担的，可以实行区级执法。   直辖市、 '
             '设区的市人民政府城市管理执法主管部门可以承担跨区域和重大复杂违法案件的查处。',
  'score': 0.6045487776924248,
  'title': 'construction_regulations/城市管理执法办法.pdf'},
 {'content': '第五章 执法规范     第二十五条 '
             '城市管理执法主管部门依照法定程序开展执法活动，应当保障当事人依法享有的陈述、申辩、听证等权利。   第二十六条 '
             '城市管理执法主管部门开展执法活动， 应当根据违法行为的性质和危害后果依法给予相应的行政处罚。\n'
             '\n'
             '  \n'
             '\n'
             '对违法行为轻微的，可以采取教育、劝诫、疏导等方式予以纠正。',
  'score': 0.5855412211996395,
  'title': 'construction_regulations/城市管理执法办法.pdf'},
 {'content': '住房和城乡建设部规章 \n'
             '   X住房和城乡建设部发布  \n'
             '-\n'
             '\n'
             '4 -   第十一条 城市管理执法事项范围确定后，应当向社会公开。\n'
             '\n'
             '  \n'
             '\n'
             '第十二条 城市管理执法主管部门集中行使原由其他部门行使的行政处罚权的，应当与其他部门明确职责权限和工作机制。',
  'score': 0.5779289002003206,
  'title': 'construction_regulations/城市管理执法办法.pdf'},
 {'content': '住房和城乡建设部规章 \n'
             '   X住房和城乡建设部发布  \n'
             '-\n'
             '\n'
             '2 -   第三条 城市管理执法应当遵循以人为本、依法治理、源头治理、 权责一致、 协调创新的原则， '
             '坚持严格规范公正文明执法。   第四条 国务院住房城乡建设主管部门负责全国城市管理执法的指导监督协调工作。\n'
             '\n'
             '  \n'
             '\n'
             '各省、 自治区人民政府住房城乡建设主管部门负责本行政区域内城市管理执法的指导监督考核协调工作。\n'
             '\n'
             '  \n'
             '\n'
             '城市、 县人民政府城市管理执法主管部门负责本行政区域内的城市管理执法工作。\n'
             '\n'
             '  \n'
             '\n'
             '第五条 城市管理执法主管部门应当推动建立城市管理协调机制，协调有关部门做好城市管理执法工作。',
  'score': 0.5762151842611781,
  'title': 'construction_regulations/城市管理执法办法.pdf'},
 {'content': '第二十九条 城市管理执法主管部门对查封、扣押的物品，应当妥善保管，不得使用、截留、损毁或者擅自处置。\n'
             '\n'
             '查封、扣押的物品属非法物品的，移送有关部门处理。',
  'score': 0.5753635808026318,
  'title': 'construction_regulations/城市管理执法办法.pdf'},
 {'content': '住房和城乡建设部规章 \n'
             '   X住房和城乡建设部发布  \n'
             '- 10 -   第三十四条 城市管理执法主管部门应当通过门户网站、 '
             '办事窗口等渠道或者场所，公开行政执法职责、权限、依据、监督方式等行政执法信息。   \n'
             '\n'
             '第六章 协作与配合     第三十五条 城市管理执法主管部门应当与有关部门建立行政执法信息互通共享机制， '
             '及时通报行政执法信息和相关行政管理信息。   第三十六条 城市管理执法主管部门可以对城市管理执法事项实行网格化管理。\n'
             '\n'
             '  \n'
             '\n'
             '第三十七条 城市管理执法主管部门在执法活动中发现依法应当由其他部门查处的违法行为， 应当及时告知或者移送有关部门。   \n'
             '\n'
             '第七章 执法监督',
  'score': 0.5731701893090992,
  'title': 'construction_regulations/城市管理执法办法.pdf'},
 {'content': '住房和城乡建设部规章 \n'
             '   X住房和城乡建设部发布  \n'
             '- 1 -   城市管理执法办法 （2017年1月24日中华人民共和国住房和城乡建设部令第34号公布 '
             '自2017年5月1日起施行）  第一章 总  则     第一条 为了规范城市管理执法工作，提高执法和服务水平， '
             '维护城市管理秩序， 保护公民、 法人和其他组织的合法权益，根据行政处罚法、行政强制法等法律法规的规定，制定本办法。   第二条 '
             '城市、 县人民政府所在地镇建成区内的城市管理执法活动以及执法监督活动，适用本办法。   本办法所称城市管理执法， '
             '是指城市管理执法主管部门在城市管理领域根据法律法规规章规定履行行政处罚、 行政强制等行政执法职责的行为。',
  'score': 0.5722312939985383,
  'title': 'construction_regulations/城市管理执法办法.pdf'}]

4.2 构建FunctionAgentWithRetrieval¶

FunctionAgentWithRetrieval可以是否进行检索增强回答用户问题，以下是其示例。

4.2.1 检索增强¶

In [30]:

# 创建一个ERNIEBot实例，使用"ernie-3.5"模型。
llm = ERNIEBot(model="ernie-3.5")
# 创建一个WholeMemory实例。这是一个用于存储对话历史和上下文信息的类，有助于模型理解和持续对话。
memory = WholeMemory()
# 调用一个文本转语音的工具。
tts_tool = RemoteToolkit.from_aistudio("texttospeech").get_tools()[0]
# 创建一个FunctionAgentWithRetrieval实例。这个代理将使用上面创建的ERNIEBot模型、WholeMemory和faiss_search，同时传入了一个名为tts_tool的工具。
agent = FunctionAgentWithRetrieval(
    llm=llm, tools=[tts_tool], memory=memory, knowledge_base=faiss_search, threshold=0.9
)

# 创建一个ERNIEBot实例，使用"ernie-3.5"模型。 llm = ERNIEBot(model="ernie-3.5") # 创建一个WholeMemory实例。这是一个用于存储对话历史和上下文信息的类，有助于模型理解和持续对话。 memory = WholeMemory() # 调用一个文本转语音的工具。 tts_tool = RemoteToolkit.from_aistudio("texttospeech").get_tools()[0] # 创建一个FunctionAgentWithRetrieval实例。这个代理将使用上面创建的ERNIEBot模型、WholeMemory和faiss_search，同时传入了一个名为tts_tool的工具。 agent = FunctionAgentWithRetrieval( llm=llm, tools=[tts_tool], memory=memory, knowledge_base=faiss_search, threshold=0.9 )

In [31]:

# 定义一个查询字符串，这个查询是关于"城乡建设部规章中，城市管理执法第三章，第十三条"的内容。
query = "城乡建设部规章中，城市管理执法第三章，第十三条是什么？"
# 使用agent的async_run方法来异步执行查询。由于这是异步操作，因此需要使用'await'关键字。
response = await agent._run(query)
messages = response.chat_history
for item in messages:
    print(item.to_dict())

# 定义一个查询字符串，这个查询是关于"城乡建设部规章中，城市管理执法第三章，第十三条"的内容。 query = "城乡建设部规章中，城市管理执法第三章，第十三条是什么？" # 使用agent的async_run方法来异步执行查询。由于这是异步操作，因此需要使用'await'关键字。 response = await agent._run(query) messages = response.chat_history for item in messages: print(item.to_dict())

{'role': 'user', 'content': '城乡建设部规章中，城市管理执法第三章，第十三条是什么？'}
{'role': 'assistant', 'content': '城市管理执法第三章第十三条规定，城市管理执法主管部门应当按照权责清晰、事权统一、精简效能的原则设置执法队伍。\n\n如需获取更多具体信息，建议查阅城乡建设部官方网站或者相关规章文件获取更详细、准确的信息。', 'function_call': None, 'plugin_info': None, 'search_info': None}

4.2.2 调用tool示例¶

在这个例子中，FunctionAgentWithRetrieval没有利用其检索信息，而是直接调用工具回答用户的问题

In [29]:

# 定义一个查询字符串，这个查询是关于"11-2="的内容。
response = await agent._run("把上一轮的检索内容转换为语音")
# 使用agent的async_run方法来异步执行查询。由于这是异步操作，因此需要使用'await'关键字。
messages = response.chat_history
for item in messages:
    print(item.to_dict())

# 定义一个查询字符串，这个查询是关于"11-2="的内容。 response = await agent._run("把上一轮的检索内容转换为语音") # 使用agent的async_run方法来异步执行查询。由于这是异步操作，因此需要使用'await'关键字。 messages = response.chat_history for item in messages: print(item.to_dict())

{'role': 'user', 'content': '把上一轮的检索内容转换为语音'}
{'role': 'assistant', 'content': '', 'function_call': {'name': 'texttospeech/v1.6/tts', 'thoughts': '用户需要将上一轮的检索内容转换为语音。', 'arguments': '{"tex":"城市管理执法主管部门应当按照权责清晰、事权统一、精简效能的原则设置执法队伍。"}'}, 'plugin_info': None, 'search_info': None}
{'role': 'function', 'name': 'texttospeech/v1.6/tts', 'content': '{"audio": "file-local-b603cbd6-a47d-11ee-9fbb-fa280016e4f3", "prompt": "参考工具说明中对各个结果字段的描述，提取工具调用结果中的信息，生成一段通顺的文本满足用户的需求。请务必确保每个符合\'file-\'格式的字段只出现一次，无需将其转换为链接，也无需添加任何HTML、Markdown或其他格式化元素。"}'}
{'role': 'assistant', 'content': '根据您的要求，我为您检索到的第十三条规定已转换为语音。您可以通过点击以下链接来听取：file-local-b603cbd6-a47d-11ee-9fbb-fa280016e4f3。如果您需要再次听取或使用其他功能，请随时告诉我。', 'function_call': None, 'plugin_info': None, 'search_info': None}

In [ ]: