ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;display: table;padding: 0.3em 1em;color: rgb(255, 255, 255);background: rgb(250, 81, 81);border-radius: 8px;box-shadow: rgba(0, 0, 0, 0.1) 0px 4px 6px;">背景ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 14px;letter-spacing: 0.1em;color: rgb(63, 63, 63);">上次讲解代码以后，把rag这块遗留了下来，rag的代码相对来说比较复杂，一环套一环。我们今天先来拆解下分片的整体流程。ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 14px;letter-spacing: 0.1em;color: rgb(63, 63, 63);">整体分为3个阶段ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 14px;text-indent: -1em;display: block;margin: 0.5em 8px;color: rgb(63, 63, 63);">ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 14px;text-indent: -1em;display: block;margin: 0.5em 8px;color: rgb(63, 63, 63);">ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 14px;text-indent: -1em;display: block;margin: 0.5em 8px;color: rgb(63, 63, 63);"> ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 14px;letter-spacing: 0.1em;color: rgb(63, 63, 63);">我们挨个来解析下。ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 14px;letter-spacing: 0.1em;color: rgb(63, 63, 63);">注意：代码为当天更新的代码，部分功能还未发布。ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;display: table;padding: 0.3em 1em;color: rgb(255, 255, 255);background: rgb(250, 81, 81);border-radius: 8px;box-shadow: rgba(0, 0, 0, 0.1) 0px 4px 6px;">切片设置ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;padding-left: 12px;color: rgb(63, 63, 63);">查找接口
 
我们从接口的请求参数和界面对照，大致了解下- •
layout_recognize使用哪种解析器，默认用的DeepDOC - •
use_raptor使用召回增强RAPTOR策略
代码我们按照上一篇解剖RAGFlow！全网最硬核源码架构解析讲的，找到documnet_app.py文件，搜索change_parser方法。 @manager.route('/change_parser', methods=['POST']) # noqa: F821
@login_required
@validate_request("doc_id","parser_id")
defchange_parser():
req = request.json
#权限校验，看下你有没有操作的权限
ifnotDocumentService.accessible(req["doc_id"], current_user.id):
try:
# 解析器切换验证，
e, doc = DocumentService.get_by_id(req["doc_id"])
if((doc.type== FileType.VISUALandreq["parser_id"] !="picture")
or(re.search(
r"\.(ppt|pptx|pages)$", doc.name)andreq["parser_id"] !="presentation")):
returnget_data_error_result(message="Not supported yet!")
e = DocumentService.update_by_id(doc.id,
{"parser_id": req["parser_id"],"progress":0,"progress_msg":"",
"run": TaskStatus.UNSTART.value})
ifnote:
returnget_data_error_result(message="Document not found!")
if"parser_config"inreq:
DocumentService.update_parser_config(doc.id, req["parser_config"])
ifdoc.token_num >0:
e = DocumentService.increment_chunk_num(doc.id, doc.kb_id, doc.token_num * -1, doc.chunk_num * -1,
doc.process_duation * -1)
ifnote:
returnget_data_error_result(message="Document not found!")
tenant_id = DocumentService.get_tenant_id(req["doc_id"])
ifnottenant_id:
returnget_data_error_result(message="Tenant not found!")
ifsettings.docStoreConn.indexExist(search.index_name(tenant_id), doc.kb_id):
settings.docStoreConn.delete({"doc_id": doc.id}, search.index_name(tenant_id), doc.kb_id)
returnget_json_result(data=True)
exceptExceptionase:
returnserver_error_response(e)
这段代码 - • 视觉文件(FileType.VISUAL)只能使用"picture"解析器
- • PPT类文件只能使用"presentation"解析器
- • 重置文档统计信息(token_num/chunk_num清零)
这里只是把切片的配置做了设置。整体流程如下： IndexDBServerClientIndexDBServerClientPOST/change_parser权限检查权限结果解析器类型验证更新解析器配置删除旧索引返回操作结果 启动
点击文档后面的启动按钮。可以看到调用的具体接口
重点是这个方法,在api/db/services/task_service.py中
 - • 第一段代码在进行任务分割，不通类型的文档最后分割策略一样
- • 第二段代码是生成任务，包括：唯一task_digest 和初始进度，优先
在这里我们要了解一个概念，一个task，会拆出多个chuck。使用的是redis的消息队列,根据优先级使用不同的队列名称，默认用rag_flow_svr_queue。这段代码看图片里的注释吧。
最后这块的整体流程如下： PDFExcel其他开始文档类型判断PDF分片处理表格分片处理单任务处理生成任务摘要历史任务检查结果复用处理数据清理任务持久化未完成任务入队结束 任务消费与处理在rag/svr/task_executor.py中
任务执行器启动以后，死循环执行asyncdefhandle_task():
#通过redis消息队列从`rag_flow_svr_queue`获取任务
redis_msg, task =awaitcollect()
ifnottask:
# 没有获取到，休眠5秒
awaittrio.sleep(5)
return
try:
# 状态记录
CURRENT_TASKS[task["id"]] = copy.deepcopy(task)
# 核心处理
awaitdo_handle_task(task)
# 成功处理
DONE_TASKS +=1
exceptExceptionase:
# 异常处理
FAILED_TASKS +=1
set_progress(task_id, -1,str(e))
我们接着看do_handle_task方法, # 绑定进度回调
progress_callback = partial(set_progress, task_id, ...)
# 任务状态检查
ifTaskService.do_cancel(task_id):
# 主动取消
progress_callback(-1,"Task canceled")
return
标准模式GraphRAG模式RAPTOR模式raptorgraphragdefault任务类型判断类型层次化摘要知识图谱构建标准分片初始化Chat模型执行run_raptor调整并发限制加载配置执行run_graphrag生成原始分片向量编码 我们看下默认分片中的关键方法 # 分片，这个是核心
chunks =awaitbuild_chunks(task, progress_callback)
# 所有的分片向量化，并向量结果写入到每个chunk的["q_%d_vec" % len(v)] = v
token_count, vector_size =awaitembedding(chunks, embedding_model, task_parser_config, progress_callback)
简单看下build_chunks方法 asyncwithchunk_limiter:
cks =awaittrio.to_thread.run_sync(lambda: chunker.chunk(task["name"], binary=binary, from_page=task["from_page"],
to_page=task["to_page"], lang=task["language"], callback=progress_callback,
kb_id=task["kb_id"], parser_config=task["parser_config"], tenant_id=task["tenant_id"]))
最后到了rag/app/naive.py文件中的chunk方法。在这个方法里根据切片配置进行了处理。整体流程如下： DOCXPDFExcelTXT/CodeMarkdownHTML/JSON是否输入文件格式判断DOCX解析器PDF解析器+布局识别表格解析器文本分割器MD表格提取结构化解析原始分片生成是否视觉增强?视觉模型处理图表基础分片处理分片合并Token化处理输出结构化分片 在最新的版本中，使用视觉模型，对图表进行增强。该代码还有发布。 整个异步处理如下： 资源清理异常处理任务处理任务收集初始化阶段有消息无消息异常main入口初始化系统启动监控线程任务处理循环获取任务处理任务清理资源打印Banner加载配置设置信号处理初始化内存追踪检查未ACK消息获取任务详情拉取新消息验证任务有效性记录开始状态执行核心处理更新进度记录完成状态捕获错误构造错误信息更新失败状态释放内存引用发送ACK更新计数器 后记通过最近的源码解析，ragflow后面的升级有几块 |
