|
从本文开始，将开一个大坑，陆续介绍企业级文档问答系统构建的全流程，以及关键环节的优化手段。重点介绍算法流程。 构建一个基础版的RAG是非常简单的，甚至使用扣子、Dify等平台，熟练的情况下都用不了5分钟，即使使用Langchain、LlamaIndex等框架，搭建完整流程，代码也不会超过100行。但基础版的问答效果往往较差。 下面这张图是OpenAI介绍的RAG优化经验，这个准确率当然随不同的数据集会有不同，但基本上优化后的准确率比优化前有显著提升这个基本上是一致的。 
问答系统构建完成后，总的流程是先对文档进行解析、切分，然后使用问题检索相关知识片段，最后将问题和知识片段输入LLM，生成答案。 在构建的过程中也是一样的，这三个环节是可以分别独立优化的，如下图所示： 
本篇首先专注在如何获取QA数据，所谓的QA数据，就是“问题-回答”数据，理想情况下，如果包含回答所用到的文档片段是更好的。部分系统（如客服系统）是有这方面数据的，但绝大多数情况下是没有的，这时就需要首先构造一批问答数据，这是后续所有环节最重要的一步。 本系列将会使用中国银行所发布的《2024全球经济金融展望报告》作为文档，围绕针对这个文档的问答效果优化展开。 本文所介绍的方法，会使用千问官方的qwen-long模型，对《2024全球经济金融展望报告》这个文档抽取QA，这个模型足够便宜，抽取的结果质量也还不错。QA抽取包含如下3个步骤： 短文档片段QA抽取：这部分模拟日常情况下，经常会询问细节性问题的使用场景 长文档片段QA抽取：这部分模拟需要综合较多上下文才能回答的使用场景 QA质量打分：使用LLM再次对抽取的QA进行质量评估，这一步算是借鉴了微软phi-1.5模型Textbooks Are All You Need论文中的方法，就是借助模型对数据质量进行评估
整个过程花费不到1元，结果已经抽取好了，大家可以直接使用。 本文所对应代码已开源，地址在：https://github.com/Steven-Luo/MasteringRAG/blob/main/00_PDF%E8%A7%A3%E6%9E%90%E4%B8%8EQA%E6%8A%BD%E5%8F%96_v1.ipynb 代码在Google Colab环境下进行了测试，正常情况下，安装Anaconda基本上会包含大部分所用到的包，再安装如下包即可： pipinstalllangchainlangchain_communitypypdfopenai ,,,
module(langchain,langchain_community,pypdf,openai):
() langchain0.2.8
langchain_community0.2.7
pypdf4.3.0
openai1.35.14
os.environ[]=
os.environ[]= PyPDFLoader
Document
RecursiveCharacterTextSplitter
uuid4
(documents,filepath,chunk_size=,chunk_overlap=,seperators=[,],force_split=):
os.path.exists(filepath)force_split:
()
pickle.load((filepath,))
splitter=RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
separators=seperators
)
split_docs=splitter.split_documents(documents)
chunksplit_docs:
chunk.metadata[]=(uuid4())
pickle.dump(split_docs,(filepath,))
split_docs
loader=PyPDFLoader()
documents=loader.load()
pattern=
merged_docs=[Document(page_content=.join(re.sub(pattern,,doc.page_content)docdocuments))]
splitted_docs=split_docs(documents,os.path.join(output_dir,),chunk_size=,chunk_overlap=)
splitted_docs_large=split_docs(merged_docs,os.path.join(output_dir,),chunk_size=,chunk_overlap=)
uuid2doc={doc.metadata[]:doc.page_contentdocsplitted_docs}
uuid2large_doc={doc.metadata[]:doc.page_contentdocsplitted_docs_large}既然是构造QA，那最好是保留回答问题时所使用的上下文，方便后续环节的优化。 qa_gen_prompt_tmpl=
{{document}}
qa_gen_prompt_tmpl_large_context=
{{document}}抽取核心代码，此处使用多线程加速抽取，考虑到网络请求异常情况会比较多，因此增加失败重试机制，同时考虑到这是一个耗时操作，并保存中间结果，以确保失败或者再次运行时，已经执行过的部分不会被重复执行： OpenAI
tqdm
client=OpenAI(
api_key=os.environ[],
base_url=os.environ[]
)
(prompt_tmpl,text):
prompt=prompt_tmpl.replace(,text).strip()
prompt
(prompt,max_retry=,debug=,top_p=,temperature=):
(prompt):
completion=client.chat.completions.create(
model=,
messages=[
{:,:prompt}
],
top_p=top_p,
temperature=temperature
)
completion.choices[].message.content
max_retry>:
:
do_chat(prompt)
e:
max_retry-=
sleep_seconds=random.randint(,)
debug:
()
time.sleep(sleep_seconds)
(splitted_docs,prompt_tmpl,qa_ckpt_filename):
qa_ckpt={}
os.path.exists(qa_ckpt_filename):
qa_ckpt=(qa_ckpt_filename).readlines()
qa_ckpt=[json.loads(line.strip())lineqa_ckptline.strip()!=]
qa_ckpt={item[]:itemitemqa_ckpt}
()
file_lock=threading.Lock()
max_workers=
concurrent.futures.ThreadPoolExecutor(max_workers=max_workers)executor:
futures={doc.metadata[]:executor.submit(chat,build_qa_prompt(prompt_tmpl,doc.page_content),,)docsplitted_docs(doc.page_content.replace(,))>=doc.metadata[]qa_ckpt}
uuidtqdm(futures):
future=futures[uuid]
result=future.result()
result:
item={:uuid,:result}
qa_ckpt[uuid]=item
file_lock.acquire()
:
(qa_ckpt_filename,)f:
f.write(json.dumps(item,ensure_ascii=)+)
e:
(e)
:
file_lock.release()
qa_ckpt
detailed_qa_dict=gen_qa(splitted_docs,qa_gen_prompt_tmpl,os.path.join(output_dir,))
large_context_qa_dict=gen_qa(splitted_docs_large,qa_gen_prompt_tmpl_large_context,os.path.join(output_dir,))
[
{
:,
:,
:
},
{
:,
:,
:
},
{
:,
:,
:
},
{
:,
:,
:
},
{
:,
:,
:
},
{
:,
:,
:
},
{
:,
:,
:
},
{
:,
:,
:
}
]从上面的样例可以看出，结果是被json...
(text):
pattern=
text=text.replace(,)
:
json.loads(text)
:
match=re.search(pattern,text,re.DOTALL)
:
matched=match.group()
json.loads(matched)
e:
()
[]
(qa_ckpt,uuid2doc_map):
data=[]
key,valuetqdm(qa_ckpt.items()):
text=value[]
qa_list=convert2json(text)
itemqa_list:
question=item.get(,).strip()
answer=item.get(,).strip()
context=item.get(,).strip()
question==answer==:
(qa_list)
data.append({
:key,
:question,
:answer,
:context,
:uuid2doc_map[key]
})
qa_df=pd.DataFrame(data)
qa_df
qa_df=build_qa_df(detailed_qa_dict,uuid2doc)
qa_df.drop_duplicates(,inplace=)
qa_df[]=
large_context_qa_df=build_qa_df(large_context_qa_dict,uuid2large_doc)
large_context_qa_df.drop_duplicates(,inplace=)
large_context_qa_df[]=
qa_df=pd.concat([qa_df,large_context_qa_df])这部分就是对qa_df中的问题-回答对，再打一次分，然后过滤低分结果，Prompt如下： qa_check_prompt_tmpl=
{{question}}
{{answer}}总体又是一个循环，与QA抽取部分非常相似，此处不再粘贴代码，需要的朋友们请访问代码仓库。 这部分首先保留4分及以上的问答对，然后随机挑选100条数据作为后续的测试集。至此，准备工作完成。 hq_qa_df=qa_df[qa_df[]>=]
test_q=hq_qa_df.sample(,replace=)[].values.tolist()
hq_qa_df[]=
hq_qa_df.loc[hq_qa_df[].isin(test_q),]=
hq_qa_df.to_excel(os.path.join(output_dir,),index=) |
