•{"id": 34, "question": "根据武汉兴图新科电子股份有限公司招股意向书，电子信息行业的上游涉及哪些企业？"}
通过查看日志，检索器没有检索到相关信息：
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解析器。
第一步：安装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注意：
•上述的/Users/deadwalk/Code/smart-finance-bot请根据本地路径修改；
•运行完毕后请使用docker ps确认容器已经启动。
第六步：进入es容器
dockerexec-ites/bin/bash
第七步：命令行输入重置密码命令（此处我们重置密码为123abc）
bin/elasticsearch-reset-password-i-uelastic
第八步：使用浏览器访问http://localhost:9200/，验证服务可以使用
编写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())
运行结果：
代码文件：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()说明：
•elasticsearch后续的插入操作中，使用到了nltk分词，其代码已经封装在UtilNltk类中，具体代码请查看Github仓库代码，本文不再赘述。
•LoggerManager是代码重构时，封装的一个日志管理类，具体代码请查看Github仓库代码，本文不再赘述。
在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的金融问答系统：代码重构已做介绍，所以本文不再赘述。
在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()运行结果：
代码文件：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[]在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
•作为传统搜索引擎，可以通过keyword_search检索到相关内容。
•使用时需要使用Docker搭建ES服务。
•数据文件需要添加到ES服务中，方便检索。
•MultiQueryRetriever
•多路召回，将问题拆分成多个问题，然后进行检索，最终合并结果。
| 欢迎光临 链载Ai (https://www.lianzai.com/) | Powered by Discuz! X3.5 |