网站设计的含义,黄石网站开发,网站建设帝国,网站的建设合同是否交印花税参考自科大讯飞AI大赛#xff08;多模态RAG方向#xff09; - Datawhale
赛题意义#xff1a;
我们正处在一个信息爆炸的时代#xff0c;但这些信息并非以整洁的纯文本形式存在。它们被封装在各种各样的载体中#xff1a;公司的年度财报、市场研究报告、产品手册、学术论…参考自科大讯飞AI大赛多模态RAG方向 - Datawhale
赛题意义
我们正处在一个信息爆炸的时代但这些信息并非以整洁的纯文本形式存在。它们被封装在各种各样的载体中公司的年度财报、市场研究报告、产品手册、学术论文以及无数的网页。这些载体的共同特点是 图文混排 ——文字、图表、照片、流程图等元素交织在一起共同承载着完整的信息。
传统的AI技术如搜索引擎或基于文本的问答系统在处理这类复杂文档时显得力不从心。它们能很好地理解文字但对于图表中蕴含的趋势、数据和关系却是“视而不见”的。这就造成了一个巨大的信息鸿沟AI无法回答那些需要结合视觉内容才能解决的问题例如“根据这张条形图哪个产品的市场份额最高”或“请解释一下这张流程图的工作原理”。
多模态技术方案选择
基于图片描述 对所有图片生成文本描述将这些描述与原文的文本块统一处理。这能将多模态问题简化为纯文本问题最适合快速构建Baseline。
具体工具栈
PDF解析 这个环节我们选择的mineru但是task1里面为了降低大家的学习门槛使用的是pymupdf作为平替方案。 Embedding实现 我最初考虑使用 sentence-transformer 库。但在进一步查阅资料时我发现了 Xinference 它能将模型部署为服务并通过兼容OpenAI的API来调用。我立即决定采用这种方式因为 服务化能让我的Embedding模块与主应用逻辑解耦更利于调试和未来的扩展。 Baseline执行流程 时间消耗问题
解决 不要在一个脚本里完成所有事 。强烈建议使用 Jupyter Notebook 进行开发调试并将流程拆分 第一阶段解析 。在一个Notebook中专门负责调用pymupdf将所有PDF解析为JSON并 保存到本地 。这个阶段成功运行一次后就不再需要重复执行。 第二阶段预处理与Embedding 。在另一个Notebook中读取第一步生成的JSON文件进行图片描述生成、数据清洗并调用Embedding模型。将最终包含向量的知识库 保存为持久化文件 。 第三阶段检索与生成 。在第三个Notebook中加载第二步保存好的知识库专注于调试检索逻辑和Prompt工程。
通过这种 分步执行、缓存中间结果 的方式可以极大地提高调试效率每次修改只需运行对应的、耗时较短的模块。 核心问题 LLM需要什么样的信息 检索模块是LLM的眼睛和耳朵。我们是应该只把和问题最相似的那一小块知识喂给LLM还是应该提供更丰富的周边信息例如找到一个关键段落后是否应该把它的上下文前后段落、所属章节标题也一并提供来帮助LLM更好地理解
Baseline代码
1、将压缩包解压到指定目录
unzip -n datas/财报数据库.zip -d datas/ -n → 不覆盖已存在的文件已有文件会跳过解压 datas/财报数据库.zip → 压缩包位置 -d datas/ → 解压到 datas/ 目录
2、文档分块没有涉及多模态的理解对图表类也仅仅是提取文本
批量遍历./datas目录下含子目录即./datas/财报数据库的所有 PDF 文件然后用 PyMuPDFfitz逐页提取文本内容最终把每一页的文字和相关信息整理成 JSON 文件保存到all_pdf_page_chunks.json中即按页的方式来分块即我们RAG系统的最终知识库而非向量数据库。
fitz_pipeline_all.py
import fitz # PyMuPDF用于操作 PDF 文件提取文字、图片等
import json
from pathlib import Path # Python 标准库用于更方便地处理路径def process_pdfs_to_chunks(datas_dir: Path, output_json_path: Path):扫描指定目录下的所有 PDF 文件将每个 PDF 按页提取文本内容转换成“页面数据块chunk”最终统一保存到一个 JSON 文件中。Args:datas_dir (Path): 存放 PDF 文件的目录可以包含子目录。output_json_path (Path): 保存最终 JSON 文件的路径。all_chunks [] # 用于存储所有 PDF 页面的数据块# 递归搜索 datas_dir 目录下的所有 .pdf 文件pdf_files list(datas_dir.rglob(*.pdf))if not pdf_files:print(f警告在目录 {datas_dir} 中未找到任何 PDF 文件。)returnprint(f找到 {len(pdf_files)} 个 PDF 文件开始处理...)# 遍历找到的每一个 PDF 文件for pdf_path in pdf_files:file_name_stem pdf_path.stem # 文件名不包含扩展名full_file_name pdf_path.name # 完整文件名含扩展名print(f - 正在处理: {full_file_name})try:# 打开 PDF 文件with 确保用完后自动关闭with fitz.open(pdf_path) as doc:# 遍历 PDF 的每一页for page_idx, page in enumerate(doc):# 提取当前页面的文字text 模式只取文本不含图片content page.get_text(text)# 如果该页没有任何文字则跳过if not content.strip():continue# 构造当前页面的“数据块”字典chunk {id: f{file_name_stem}_page_{page_idx}, # 唯一 ID文件名 页码content: content, # 当前页的文字内容metadata: { # 元信息page: page_idx, # 页码从 0 开始file_name: full_file_name # 源文件名}}all_chunks.append(chunk) # 添加到总列表中except Exception as e:# 如果某个 PDF 处理出错则打印错误信息并继续处理下一个print(f处理文件 {pdf_path} 时发生错误: {e})# 确保输出目录存在没有就创建output_json_path.parent.mkdir(parentsTrue, exist_okTrue)# 将收集到的所有数据块写入 JSON 文件with open(output_json_path, w, encodingutf-8) as f:json.dump(all_chunks, f, ensure_asciiFalse, indent2)print(f\n处理完成所有内容已保存至: {output_json_path})def main():# 获取当前脚本所在目录base_dir Path(__file__).parent# 假设 PDF 数据在当前目录的 datas 文件夹中datas_dir base_dir / datas# 生成的 JSON 存放在当前目录文件名为 all_pdf_page_chunks.jsonchunk_json_path base_dir / all_pdf_page_chunks.json# 执行 PDF 批量处理process_pdfs_to_chunks(datas_dir, chunk_json_path)if __name__ __main__:main()两层循环所有pdf文档进行循环然后是一个pdf文档中所有页进行循环最终JSON文件以每一页内容为单位进行记录。即从一个pdf文件的第一页开始直到最后一个pdf的最后一页即按页的方式来分块。
每一个chunk或者说每一页字典输出格式 chunk { id: f{file_name_stem}_page_{page_idx}, content: content, metadata: { page: page_idx, # 0-based page index file_name: full_file_name } }
如下所示
[{id: file1_page_0,content: 第一页的文字内容...,metadata: {page: 0,file_name: file1.pdf}},{id: file1_page_1,content: 第二页的文字内容...,metadata: {page: 1,file_name: file1.pdf}},{id: file2_page_0,content: 第一页的文字内容...,metadata: {page: 0,file_name: file2.pdf}}
]3、构建一个简易的多文档 RAG系统
步骤如下 读取 PDF 页内容的 JSON 文件之前我们提取的 all_pdf_page_chunks.json。 调用本地的 embedding 接口把每个页面内容键content对应的值转成向量embeddings self.embedding_model.embed_texts([c[content] for c in chunks]) 把向量和对应的chunkpage一起存进一个简单的向量库SimpleVectorStore本Baseline为了简化用一个内存列表 SimpleVectorStore 模拟了向量数据库的功能可以看详细代码直接定义了两个列表去储存也就是直接在内存中存 embeddings chunks然后用 numpy 实现余弦相似度搜索self.vector_store.add_chunks(chunks, embeddings) 接收一个问题 → 检索最相似的页面 → 调用大模型生成结构化回答JSON 格式 answer / filename / page检索步骤如下 generate_answer()先检索回Top-k个chunk再把所有召回的chunk依次拼接成上下文文件名 页码 内容全部join起来成为一个大的上下文context生成一个 prompt要求模型用 JSON 格式返回调用本地的大模型 APILOCAL_TEXT_MODEL用 extract_json_array 解析模型输出的 JSON如果解析失败就用原始文本第一条检索结果的文件名和页码返回结构化字典question, answer, filename, page, retrieval_chunks。 prompt如下 prompt (f你是一名专业的金融分析助手请根据以下检索到的内容回答用户问题。\nf请严格按照如下JSON格式输出\nf{{answer: 你的简洁回答, filename: 来源文件名, page: 来源页码}}\nf检索内容\n{context}\n\n问题{question}\nf请确保输出内容为合法JSON字符串不要输出多余内容。) 支持批量测试可从 datas/test.json 读取一堆问题并并发处理最后输出两个结果文件。rag_top1_pred_raw.json原始结果包含 retrieval_chunks 和原始 idxrag_top1_pred.json过滤结果去掉 retrieval_chunks未回答的补空白用于评测结果的提交包含test.json里面的全部问题和针对其中10个问题的答案。 采用线程池并发调用的方式with concurrent.futures.ThreadPoolExecutor(max_workers10) as executor:因为在RAG检索和LLM回答的瓶颈不是 CPU 计算而是 IO 等待网络延迟 模型响应时间而现在一次性有10个问题因此线程池并发而不是顺序执行的方式把多个题目同时处理大幅减少总耗时。
rag_from_page_chunks.py
import json
import osimport hashlib
from typing import List, Dict, Any
from tqdm import tqdm
import sys
import concurrent.futures
import randomfrom get_text_embedding import get_text_embeddingfrom dotenv import load_dotenv
from openai import OpenAI
# 统一加载项目根目录的.env
load_dotenv()class PageChunkLoader:def __init__(self, json_path: str):self.json_path json_pathdef load_chunks(self) - List[Dict[str, Any]]:with open(self.json_path, r, encodingutf-8) as f:return json.load(f)class EmbeddingModel:def __init__(self, batch_size: int 64):self.api_key os.getenv(LOCAL_API_KEY)self.base_url os.getenv(LOCAL_BASE_URL)self.embedding_model os.getenv(LOCAL_EMBEDDING_MODEL)self.batch_size batch_sizeif not self.api_key or not self.base_url:raise ValueError(请在.env中配置LOCAL_API_KEY和LOCAL_BASE_URL)def embed_texts(self, texts: List[str]) - List[List[float]]:return get_text_embedding(texts,api_keyself.api_key,base_urlself.base_url,embedding_modelself.embedding_model,batch_sizeself.batch_size)def embed_text(self, text: str) - List[float]:return self.embed_texts([text])[0]class SimpleVectorStore:def __init__(self):self.embeddings []self.chunks []
# 直接内存中存 embeddings chunksdef add_chunks(self, chunks: List[Dict[str, Any]], embeddings: List[List[float]]):self.chunks.extend(chunks)self.embeddings.extend(embeddings)def search(self, query_embedding: List[float], top_k: int 3) - List[Dict[str, Any]]:from numpy import dotfrom numpy.linalg import normimport numpy as npif not self.embeddings:return []emb_matrix np.array(self.embeddings)query_emb np.array(query_embedding)sims emb_matrix query_emb / (norm(emb_matrix, axis1) * norm(query_emb) 1e-8)idxs sims.argsort()[::-1][:top_k]return [self.chunks[i] for i in idxs]class SimpleRAG:def __init__(self, chunk_json_path: str, model_path: str None, batch_size: int 32):self.loader PageChunkLoader(chunk_json_path)self.embedding_model EmbeddingModel(batch_sizebatch_size)self.vector_store SimpleVectorStore()def setup(self):print(加载所有页chunk...)chunks self.loader.load_chunks()print(f共加载 {len(chunks)} 个chunk)print(生成嵌入...)embeddings self.embedding_model.embed_texts([c[content] for c in chunks])print(存储向量...)self.vector_store.add_chunks(chunks, embeddings)print(RAG向量库构建完成)def query(self, question: str, top_k: int 3) - Dict[str, Any]:q_emb self.embedding_model.embed_text(question)results self.vector_store.search(q_emb, top_k)return {question: question,chunks: results}def generate_answer(self, question: str, top_k: int 3) - Dict[str, Any]:检索大模型生成式回答返回结构化结果qwen_api_key os.getenv(LOCAL_API_KEY)qwen_base_url os.getenv(LOCAL_BASE_URL)qwen_model os.getenv(LOCAL_TEXT_MODEL)if not qwen_api_key or not qwen_base_url or not qwen_model:raise ValueError(请在.env中配置LOCAL_API_KEY、LOCAL_BASE_URL、LOCAL_TEXT_MODEL)q_emb self.embedding_model.embed_text(question)chunks self.vector_store.search(q_emb, top_k)# 拼接检索内容带上元数据context \n.join([f[文件名]{c[metadata][file_name]} [页码]{c[metadata][page]}\n{c[content]} for c in chunks])# 明确要求输出JSON格式 answer/page/filenameprompt (f你是一名专业的金融分析助手请根据以下检索到的内容回答用户问题。\nf请严格按照如下JSON格式输出\nf{{answer: 你的简洁回答, filename: 来源文件名, page: 来源页码}}\nf检索内容\n{context}\n\n问题{question}\nf请确保输出内容为合法JSON字符串不要输出多余内容。)client OpenAI(api_keyqwen_api_key, base_urlqwen_base_url)completion client.chat.completions.create(modelqwen_model,messages[{role: system, content: 你是一名专业的金融分析助手。},{role: user, content: prompt}],temperature0.2,max_tokens1024)import json as pyjsonfrom extract_json_array import extract_json_arrayraw completion.choices[0].message.content.strip()# 用 extract_json_array 提取 JSON 对象json_str extract_json_array(raw, modeobjects)if json_str:try:arr pyjson.loads(json_str)# 只取第一个对象if isinstance(arr, list) and arr:j arr[0]answer j.get(answer, )filename j.get(filename, )page j.get(page, )else:answer rawfilename chunks[0][metadata][file_name] if chunks else page chunks[0][metadata][page] if chunks else except Exception:answer rawfilename chunks[0][metadata][file_name] if chunks else page chunks[0][metadata][page] if chunks else else:answer rawfilename chunks[0][metadata][file_name] if chunks else page chunks[0][metadata][page] if chunks else # 结构化输出return {question: question,answer: answer,filename: filename,page: page,retrieval_chunks: chunks}if __name__ __main__:# 路径可根据实际情况调整chunk_json_path ./all_pdf_page_chunks.jsonrag SimpleRAG(chunk_json_path)rag.setup()# 控制测试时读取的题目数量默认只随机抽取10个实际跑全部时设为NoneTEST_SAMPLE_NUM 10 # 设置为None则全部跑FILL_UNANSWERED True # 未回答的也输出默认内容# 批量评测脚本读取测试集检索大模型生成输出结构化结果test_path ./datas/test.jsonif os.path.exists(test_path):with open(test_path, r, encodingutf-8) as f:test_data json.load(f)# 记录所有原始索引all_indices list(range(len(test_data)))# 随机抽取部分题目用于测试selected_indices all_indicesif TEST_SAMPLE_NUM is not None and TEST_SAMPLE_NUM 0:if len(test_data) TEST_SAMPLE_NUM:selected_indices sorted(random.sample(all_indices, TEST_SAMPLE_NUM))def process_one(idx):item test_data[idx]question item[question]tqdm.write(f[{selected_indices.index(idx)1}/{len(selected_indices)}] 正在处理: {question[:30]}...)result rag.generate_answer(question, top_k5)return idx, resultresults []if selected_indices:with concurrent.futures.ThreadPoolExecutor(max_workers10) as executor:results list(tqdm(executor.map(process_one, selected_indices), totallen(selected_indices), desc并发批量生成))# 先输出一份未过滤的原始结果含 idxraw_out_path ./rag_top1_pred_raw.jsonwith open(raw_out_path, w, encodingutf-8) as f:json.dump(results, f, ensure_asciiFalse, indent2)print(f已输出原始未过滤结果到: {raw_out_path})# 只保留结果部分并去除 retrieval_chunks 字段idx2result {idx: {k: v for k, v in r.items() if k ! retrieval_chunks} for idx, r in results}filtered_results []for idx, item in enumerate(test_data):if idx in idx2result:filtered_results.append(idx2result[idx])elif FILL_UNANSWERED:# 未被回答的补默认内容filtered_results.append({question: item.get(question, ),answer: ,filename: ,page: ,})# 输出结构化结果到jsonout_path ./rag_top1_pred.jsonwith open(out_path, w, encodingutf-8) as f:json.dump(filtered_results, f, ensure_asciiFalse, indent2)print(f已输出结构化检索大模型生成结果到: {out_path})else:print(datas/test.json 不存在) 目前Baseline存在的问题
1、在第二步的文档解析没有涉及多模态的理解对图表类也仅仅是提取文本即目前采用的是图像文本化这个方法会造成信息损失 没有提取图片里面的内容。
2、Embedding那里针对4263个chunkpage共耗时3分钟在做向量相似度的search的时候耗时11分钟随机抽取 TEST_SAMPLE_NUM10 个问题并做了并发调用每个问题召回的chunk数目TOP-K是5如何进行加速
3、将文档按照页面切分成独立的块进行检索可能会破坏原文中段落与段落、段落与图表之间的上下文关联。检索出的知识块可能是孤立的缺乏上下文。
4、检索策略单一 仅基于语义相似度的检索对于一些包含特定关键词或需要大范围信息整合的问题可能不是最优解。 提分策略 分块策略 (Chunking Strategy) : 目前是按“页”分块这样做简单但粗糙。是否可以尝试更细粒度的分块比如按段落、甚至固定长度的句子分块这会如何影响检索的精度和召回率 如何处理跨越多个块的表格或段落是否可以引入重叠Overlap分块的策略 检索优化 (Retrieval Optimization) : 当前的Top-K检索策略很简单。如果检索回来的5个块中只有1个是真正相关的其他4个都是噪音这会严重干扰LLM。如何提高检索结果的信噪比 可以引入重排Re-ranking模型吗即在初步检索召回出20个候选块后用一个更强的模型对这20个块进行重新排序选出最相关的5个。 Prompt工程 (Prompt Engineering) : rag_from_page_chunks.py 中的Prompt是整个生成环节的灵魂。你能设计出更好的Prompt吗 比如如何更清晰地指示LLM在多个来源中选择最精确的那一个如何让它在信息不足时回答“根据现有信息无法回答”而不是产生幻觉 多模态融合 (Multimodal Fusion) : “图片-文字描述”的方案有信息损失。有没有办法做得更好 可以尝试 多路召回 吗即文本问题同时去检索文本库和图片库使用CLIP等多模态向量模型将检索到的文本和图片信息都提供给一个多模态大模型如Qwen-VL让它自己去融合信息并作答。 升级数据解析方案从 fitz 到 MinerU 这是至关重要的进阶环节。基础方案所使用的 fitz_pipeline_all.py 仅能提取文本会遗漏表格、图片等关键信息。 提升攻略1Prompt工程设计
1、明确的来源选择指令
告诉 LLM即使有多个来源也要进行判断和筛选选择质量最高的那个来作为最终的引用。
prompt (f你是一名专业的金融分析助手请根据以下检索到的内容回答用户问题。\nf---检索内容---\nf{context}\nf---用户问题---\nf{question}\nf请严格遵循以下规则\nf1. **只使用**上述检索内容中的信息进行回答不要使用你的外部知识。\nf2. 如果有多个来源文件名和页码提供了相似或互补的信息请综合它们。但在最终回答中**只引用最直接、最完整的那个来源**的文件名和页码。\nf3. 你的回答必须是简洁、精确的并且直接引用或概括检索到的内容。\nf4. 严格按照如下JSON格式输出不要输出多余内容\nf{{answer: 你的简洁回答, filename: 来源文件名, page: 来源页码}}
)
2、处理信息不足的策略
防止 LLM 在信息不足时产生幻觉是 RAG 系统设计中一个重要挑战与其让它“编造”一个看似合理的答案不如明确告诉它在这种情况下该怎么做比如在 Prompt 中加入无法回答的指令
prompt (# ... (前面的 Prompt 内容不变) ...f请严格遵循以下规则\nf1. **只使用**上述检索内容中的信息进行回答不要使用你的外部知识。\nf2. 如果检索内容中**没有足够的信息来回答问题**请直接在 answer 字段中回答根据现有信息无法回答。此时filename 和 page 字段都应为空。\nf3. 如果有多个来源... (其余规则不变) ...
)
