链载Ai

标题: 使用RAG技术构建企业级文档问答系统之QA抽取 [打印本页]
作者: 链载Ai    时间: 2025-12-2 10:44
标题: 使用RAG技术构建企业级文档问答系统之QA抽取

1 概述

从本文开始，将开一个大坑，陆续介绍企业级文档问答系统构建的全流程，以及关键环节的优化手段。重点介绍算法流程。

构建一个基础版的RAG是非常简单的，甚至使用扣子、Dify等平台，熟练的情况下都用不了5分钟，即使使用Langchain、LlamaIndex等框架，搭建完整流程，代码也不会超过100行。但基础版的问答效果往往较差。

下面这张图是OpenAI介绍的RAG优化经验，这个准确率当然随不同的数据集会有不同，但基本上优化后的准确率比优化前有显著提升这个基本上是一致的。


问答系统构建完成后，总的流程是先对文档进行解析、切分，然后使用问题检索相关知识片段，最后将问题和知识片段输入LLM，生成答案。

在构建的过程中也是一样的，这三个环节是可以分别独立优化的，如下图所示：



本篇首先专注在如何获取QA数据，所谓的QA数据，就是“问题-回答”数据，理想情况下，如果包含回答所用到的文档片段是更好的。部分系统（如客服系统）是有这方面数据的，但绝大多数情况下是没有的，这时就需要首先构造一批问答数据，这是后续所有环节最重要的一步。

本系列将会使用中国银行所发布的《2024全球经济金融展望报告》作为文档，围绕针对这个文档的问答效果优化展开。

本文所介绍的方法，会使用千问官方的qwen-long模型，对《2024全球经济金融展望报告》这个文档抽取QA，这个模型足够便宜，抽取的结果质量也还不错。QA抽取包含如下3个步骤：

整个过程花费不到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

2 准备环境

代码在Google Colab环境下进行了测试，正常情况下，安装Anaconda基本上会包含大部分所用到的包，再安装如下包即可：

pipinstalllangchainlangchain_communitypypdfopenai

为了便于大家复现，打印所安装的版本：

,,,

module(langchain,langchain_community,pypdf,openai):
()
langchain0.2.8
langchain_community0.2.7
pypdf4.3.0
openai1.35.14

设置API key


os.environ[]=
os.environ[]=

3 文档解析与切分

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}

4 QA抽取

既然是构造QA，那最好是保留回答问题时所使用的上下文，方便后续环节的优化。

4.1 QA抽取Prompt

这一步核心的2个Prompt如下：

qa_gen_prompt_tmpl=













{{document}}





qa_gen_prompt_tmpl_large_context=







{{document}}

4.2 QA抽取代码

抽取核心代码，此处使用多线程加速抽取，考虑到网络请求异常情况会比较多，因此增加失败重试机制，同时考虑到这是一个耗时操作，并保存中间结果，以确保失败或者再次运行时，已经执行过的部分不会被重复执行：

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,))

4.3 抽取样例

[
{
:,
:,
:
},
{
:,
:,
:
},
{
:,
:,
:
},
{
:,
:,
:
},
{
:,
:,
:
},
{
:,
:,
:
},
{
:,
:,
:
},
{
:,
:,
:
}
]

4.4 后置处理

从上面的样例可以看出，结果是被json...包裹的，没有办法直接解析为JSON，使用正则表达式进行后置处理，提取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])

5 QA质量检查

这部分就是对qa_df中的问题-回答对，再打一次分，然后过滤低分结果，Prompt如下：

qa_check_prompt_tmpl=










{{question}}




{{answer}}

总体又是一个循环，与QA抽取部分非常相似，此处不再粘贴代码，需要的朋友们请访问代码仓库。

5.1 打分结果样例

5.2 3分样例

5.3 2分样例

可以看出，低分问答对，质量确实相对较低

5.4 最终数据集构建

这部分首先保留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=)





欢迎光临 链载Ai (http://www.lianzai.com/) Powered by Discuz! X3.5