|
ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 15.4px;font-weight: bold;display: table;margin-right: auto;margin-bottom: 2em;margin-left: auto;padding-right: 0.2em;padding-left: 0.2em;background: rgb(15, 76, 129);color: rgb(255, 255, 255);">前言ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;margin: 1.5em 8px;letter-spacing: 0.1em;color: rgb(63, 63, 63);">本章内容，我们将在已经构建的agent框架基础上，优化检索器，为检索器搭建ElasticSearch服务，实现问答系统的检索增强。ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 15.4px;font-weight: bold;display: table;margin: 4em auto 2em;padding-right: 0.2em;padding-left: 0.2em;background: rgb(15, 76, 129);color: rgb(255, 255, 255);">检索问题ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;margin: 1.5em 8px;letter-spacing: 0.1em;color: rgb(63, 63, 63);">通过测试天池大赛数据集的前100个问题，我们发现有很多问题RAG检索不到，例如：ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;padding-left: 1em;list-style: circle;color: rgb(63, 63, 63);" class="list-paddingleft-1"> •{"id": 34, "question": "根据武汉兴图新科电子股份有限公司招股意向书，电子信息行业的上游涉及哪些企业？"} 通过查看日志，检索器没有检索到相关信息： ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;margin: 1.5em 8px;color: rgb(63, 63, 63);">![]() ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;border-radius: 4px;display: block;margin: 0.1em auto 0.5em;height: auto !important;" title="null" src="https://api.ibos.cn/v4/weapparticle/accesswximg?aid=92350&url=aHR0cHM6Ly9tbWJpei5xcGljLmNuL3N6X21tYml6X3BuZy9EYUJHdmE2bmljUGlia3Mxa2NqYllOcUZwOFIxSVl4MUs1TEVDa0h3elBpY0dvUjdaS1pBcDhXenQyaWNNUW16eUxxdWNjVlBZQURJdmRVeUFNSlpOTWliU3dnLzY0MD93eF9mbXQ9cG5nJmFtcA==;from=appmsg"/>ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 15.4px;font-weight: bold;display: table;margin: 4em auto 2em;padding-right: 0.2em;padding-left: 0.2em;background: rgb(15, 76, 129);color: rgb(255, 255, 255);">优化方案ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;margin: 1.5em 8px;letter-spacing: 0.1em;color: rgb(63, 63, 63);">分析上述case原因，检索器太过简单所致ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;overflow-x: auto;border-radius: 8px;padding: 1em;margin: 10px 8px;">
classSimpleRetrieverWrapper():
"""自定义检索器实现"""
def__init__(self,store,llm,**kwargs):
self.store=store
self.llm=llm
logger.info(f'检索器所使用的Chat模型：{self.llm}')
defcreate_retriever(self):
logger.info(f'初始化自定义的Retriever')
chromadb_retriever=self.store.as_retriever()
returnchromadb_retriever基于以上问题，我们计划使用集成检索器，方案如下：  说明： •将检索器改为使用EnsembleRetriever •集成检索器其中之一使用ElasticSearch检索器，这个检索器通过连接ElasticSearch服务，通过关键字查询相关信息。 •集成检索器另外一个使用MultiQueryRetriever检索器，这个检索器通过连接Chroma向量库查询信息。
关于MultiQueryRetriever和ElasticSearch，之前有文章做过基本内容的总结，详情请查看课程总结】day29：大模型之深入了解Retrievers解析器。
优化步骤1、搭建ES服务第一步：安装Docker，该内容不再赘述，具体请见10分钟学会Docker的安装和使用 第二步：创建网络 dockernetworkcreatees-net  第三步：拉取镜像 dockerpullelasticsearch:8.6.0  第四步：创建挂载点目录 smart-finance-bot\
|-app\
|-docker\
|-elasticsearch\#创建elasticsearch挂载目录
|-data\#创建数据目录
|-config\#创建配置目录
|-plugins\#创建插件目录
第五步：命令行中输入命令启动Docker容器 dockerrun-d\
--restart=always\
--namees\
--networkes-net\
-p9200:9200\
-p9300:9300\
--privileged\
-v/Users/deadwalk/Code/smart-finance-bot/docker/elasticsearch/data:/usr/share/elasticsearch/data\
-v/Users/deadwalk/Code/smart-finance-bot/docker/elasticsearch/plugins:/usr/share/elasticsearch/plugins\
-e"discovery.type=single-node"\
-e"ES_JAVA_OPTS=-Xms512m-Xmx512m"\
elasticsearch:8.6.0
注意： 第六步：进入es容器 dockerexec-ites/bin/bash 第七步：命令行输入重置密码命令（此处我们重置密码为123abc） bin/elasticsearch-reset-password-i-uelastic  第八步：使用浏览器访问http://localhost:9200/，验证服务可以使用  2、添加数据到ES服务2.1、测试ES的连接编写ES连接测试代码，验证ES服务连接。 deftest_es_connect():
fromelasticsearchimportElasticsearch
ELASTIC_PASSWORD="123abc"
host="localhost"
port=9200
schema="https"
url=f"{schema}://elastic:{ELASTIC_PASSWORD}@{host}:{port}"
client=Elasticsearch(
url,
verify_certs=False,
)
print(client.info())
运行结果：  2.2、实现ElasticSearch连接代码代码文件：app/rag/elasticsearch_db.py #引入
fromlangchain_core.retrieversimportBaseRetriever
fromlangchain_core.documentsimportDocument
#ES需要导入的库
fromtypingimportList
importre
importjieba
importnltk
fromnltk.corpusimportstopwords
importtime
fromelasticsearchimportElasticsearch
fromelasticsearch.exceptionsimportConnectionError,AuthenticationException
fromelasticsearchimporthelpers
importsettings
fromutils.logger_configimportLoggerManager
fromutils.util_nltkimportUtilNltk
importos
importwarnings
warnings.simplefilter("ignore")#屏蔽ES的一些Warnings
utilnltk=UtilNltk()
logger=LoggerManager().logger
classTraditionDB:
defadd_documents(self,docs):
"""
将文档添加到数据库
"""
raiseNotImplementedError("Subclassesshouldimplementthismethod!")
defget_store(self):
"""
获得向量数据库的对象实例
"""
raiseNotImplementedError("Subclassesshouldimplementthismethod!")
classElasticsearchDB(TraditionDB):
def__init__(self,
schema=settings.ELASTIC_SCHEMA,
host=settings.ELASTIC_HOST,
port=settings.ELASTIC_PORT,
index_name=settings.ELASTIC_INDEX_NAME,
k=3
#docs=docs
):
#定义索引名称
self.index_name=index_name
self.k=k
try:
url=f"{schema}://elastic:{settings.ELASTIC_PASSWORD}@{host}:{port}"
logger.info(f'初始化ES服务连接：{url}')
self.es=Elasticsearch(
url,
verify_certs=False,
#ca_certs="./docker/elasticsearch/certs/ca/ca.crt",
#basic_auth=("elastic",settings.ELASTIC_PASSWORD)
)
response=self.es.info()#尝试获取信息
logger.info(f'ES服务响应:{response}')
except(ConnectionError,AuthenticationException)ase:
logger.error(f'连接Elasticsearch失败:{e}')
raise
exceptExceptionase:
logger.error(f'发生其他错误:{e}')
logger.error(f'异常类型:{type(e).__name__}')#记录异常类型
raise
defto_keywords(self,input_string):
"""将句子转成检索关键词序列"""
#按搜索引擎模式分词
word_tokens=jieba.cut_for_search(input_string)
#加载停用词表
stop_words=set(stopwords.words('chinese'))
#去除停用词
filtered_sentence=[wforwinword_tokensifnotwinstop_words]
return''.join(filtered_sentence)
defsent_tokenize(self,input_string):
"""按标点断句,没有用到"""
#按标点切分
sentences=re.split(r'(?<=[。！？；?!])',input_string)
#去掉空字符串
return[sentenceforsentenceinsentencesifsentence.strip()]
defcreate_index(self):
"""如果索引不存在，则创建索引"""
ifnotself.es.indices.exists(index=self.index_name):
#创建索引
self.es.indices.create(index=self.index_name,ignore=400)
defbluk_data(self,paragraphs):
"""批量进行数据灌库"""
#灌库指令
actions=[
{
"_index":self.index_name,
"_source":{
"keywords":self.to_keywords(para.page_content),
"text":para.page_content
}
}
forparainparagraphs
]
#文本灌库
helpers.bulk(self.es,actions)
##灌库是异步的
#time.sleep(2)
defflush(self):
#刷新数据,数据入库完成以后刷新数据
self.es.indices.flush()
defsearch(self,query_string):
"""关键词检索"""
#ES的查询语言
search_query={
"match":{
"keywords":self.to_keywords(query_string)
}
}
res=self.es.search(index=self.index_name,query=search_query,size=self.k)
return[hit["_source"]["text"]forhitinres["hits"]["hits"]]
defdelete(self):
"""如果索引存在，则删除索引"""
ifself.es.indices.exists(index=self.index_name):
#创建索引
self.es.indices.delete(index=self.index_name,ignore=400)
defadd_documents(self,docs):
self.bluk_data(docs)
self.flush()
说明： 2.3、修改PDF文件导入代码在settings.py中添加elasticsearch配置信息：
"""
ES数据库相关的配置
"""
#ES服务开关：True表示开启ES服务，False表示关闭ES服务
ELASTIC_ENABLE_ES=True
ELASTIC_PASSWORD=os.getenv("ELASTIC_PASSWORD","123abc")
ELASTIC_HOST=os.getenv("ELASTIC_HOST","localhost")
ELASTIC_PORT=os.getenv("ELASTIC_PORT",9200)
ELASTIC_SCHEMA="https"
ELASTIC_INDEX_NAME="smart_test_index"
确认PDFProcessor.py中已经添加了对于Elasticsearch的插入操作支持，具体代码在【项目实战】基于Agent的金融问答系统：代码重构已做介绍，所以本文不再赘述。 2.4、测试PDF文件导入代码在test_framework.py中添加如下代码
deftest_import_elasticsearch():
#fromrag.elasticsearch_dbimportTraditionDB
fromrag.elasticsearch_dbimportElasticsearchDB
fromrag.pdf_processorimportPDFProcessor
llm,chat,embed=settings.LLM,settings.CHAT,settings.EMBED
#导入文件的文件目录
directory="./dataset/pdf"
#创建Elasticsearch数据库实例
es_db=ElasticsearchDB()
#创建PDFProcessor实例
pdf_processor=PDFProcessor(directory=directory,
db_type="es",
es_client=es_db,
embed=embed)
#处理PDF文件
pdf_processor.process_pdfs()
运行结果：  3、修改检索器增加Elasticsearch检索代码文件：app/rag/retrievers.py fromlangchain_core.callbacksimportCallbackManagerForRetrieverRun
fromutils.logger_configimportLoggerManager
fromlangchain_core.retrieversimportBaseRetriever
fromlangchain_core.documentsimportDocument
fromlangchain.retrieversimportEnsembleRetriever
fromlangchain.retrievers.multi_queryimportMultiQueryRetriever
fromrag.elasticsearch_dbimportElasticsearchDB
#ES需要导入的库
fromtypingimportList
importlogging
importsettings
logger=LoggerManager().logger
classSimpleRetrieverWrapper():
"""自定义检索器实现"""
def__init__(self,store,llm,**kwargs):
self.store=store
self.llm=llm
logger.info(f'检索器所使用的Chat模型：{self.llm}')
defcreate_retriever(self):
logger.info(f'初始化自定义的Retriever')
#初始化一个空的检索器列表
retrievers=[]
weights=[]
#Step1：创建一个多路召回检索器MultiQueryRetriever
chromadb_retriever=self.store.as_retriever()
mq_retriever=MultiQueryRetriever.from_llm(retriever=chromadb_retriever,llm=self.llm)
retrievers.append(mq_retriever)
weights.append(0.5)
logger.info(f'已启用MultiQueryRetriever')
#Step2：创建一个ES检索器
ifsettings.ELASTIC_ENABLE_ESisTrue:
es_retriever=ElasticsearchRetriever()
retrievers.append(es_retriever)
weights.append(0.5)
logger.info(f'已启用ElasticsearchRetriever')
#使用集成检索器，将所有启用的检索器集合在一起
ensemble_retriever=EnsembleRetriever(retrievers=retrievers,weights=weights)
returnensemble_retriever
classElasticsearchRetriever(BaseRetriever):
def_get_relevant_documents(self,query:str,)->List[Document]:
"""Returnthefirstkdocumentsfromthelistofdocuments"""
es_connector=ElasticsearchDB()
query_result=es_connector.search(query)
logger.info(f"ElasticSearch检索到资料文件个数：{len(query_result)}")
ifquery_result:
return[Document(page_content=doc)fordocinquery_result]
return[]
asyncdef_aget_relevant_documents(self,query:str)->List[Document]:
"""(Optional)asyncnativeimplementation."""
es_connector=ElasticsearchDB()
query_result=es_connector.search(query)
ifquery_result:
return[Document(page_content=doc)fordocinquery_result]
return[]
4、测试验证在test_framework.py中运行test_financebot_ex()函数，测试检索功能。 deftest_financebot_ex():
fromfinance_bot_eximportFinanceBotEx
#使用Chroma的向量库
financebot=FinanceBotEx()
example_query="根据武汉兴图新科电子股份有限公司招股意向书，电子信息行业的上游涉及哪些企业？"
financebot.handle_query(example_query)
运行结果： 连接ES后检索到3个资料文件  使用多路召回，生成3个检索问题  最终通过集成检索器检索到答案 
优化效果通过对天池大赛前100个问题的对比测试，我们最终得到如下对比验证结果：  内容小结•集成检索器： •Elasticsearch •MultiQueryRetriever
|
