掌柜智库项目(RAG)实战5. 导入数据节点实现与测试5.8 进行主图调用测试 (main_graph)主图添加测试代码进行流程完成测试if__name____main__:fromapp.utils.path_utilimportPROJECT_ROOTimportos# 全流程测试验证PDF导入→Milvus入库→KG导入完整链路logger.info( 开始执行知识图谱导入全流程测试 )# 1. 构造测试文件路径复用你项目的doc目录和pdf2md测试文件一致test_pdf_nameos.path.join(doc,万用表RS-12的使用.pdf)test_pdf_pathos.path.join(PROJECT_ROOT,test_pdf_name)# 2. 构造输出目录存放MD/图片等中间文件test_output_diros.path.join(PROJECT_ROOT,output)os.makedirs(test_output_dir,exist_okTrue)# 不存在则创建# 3. 校验测试PDF文件是否存在ifnotos.path.exists(test_pdf_path):logger.error(f全流程测试失败测试PDF文件不存在路径{test_pdf_path})logger.info(请检查文件路径或手动将测试文件放入项目根目录的doc文件夹中)else:# 4. 构造测试状态贴合实际业务入参开启PDF解析开关test_stateImportGraphState({task_id:test_kg_import_workflow_001,# 测试任务IDuser_id:test_user,# 测试用户IDlocal_file_path:test_pdf_path,# 测试PDF文件路径local_dir:test_output_dir,# 中间文件输出目录is_pdf_read_enabled:False,# 开启PDF解析核心开关is_md_read_enabled:False# 关闭MD解析})try:logger.info(f测试任务启动PDF文件路径{test_pdf_path})logger.info(f中间文件输出目录{test_output_dir})logger.info(开始执行全流程节点依次执行entry→pdf2md→md_img→split→item_name→embedding→milvus→kg)# 5. 执行LangGraph全流程流式执行打印节点执行进度final_stateNoneforstepinkb_import_app.stream(test_state,stream_modevalues):# 打印当前执行完成的节点流式输出更直观current_nodelist(step.keys())[-1]ifstepelse未知节点logger.info(f✅ 节点执行完成{current_node})final_statestep# 保存最终状态# 6. 全流程执行完成结果预览和核心指标打印iffinal_state:logger.info(-*80)logger.info( 全流程测试执行成功核心结果预览 )# 提取核心结果指标chunksfinal_state.get(chunks,[])chunk_countlen(chunks)md_contentfinal_state.get(md_content,)[:150]# MD内容前150字符has_embeddingall(dense_vectorincandsparse_vectorincforcinchunks)ifchunkselseFalsehas_chunk_idall(chunk_idincforcinchunks)ifchunkselseFalsekg_idfinal_state.get(kg_id,未生成)# KG导入生成的ID按实际业务字段调整# 打印核心指标logger.info(f PDF转MD内容预览前150字符{md_content}...)logger.info(f 文档切分总切片数{chunk_count})logger.info(f 所有切片是否完成向量化{是ifhas_embeddingelse否})logger.info(f️ 所有切片是否完成Milvus入库含chunk_id{是ifhas_chunk_idelse否})logger.info(f 知识图谱导入ID{kg_id})logger.info(f 最终状态包含的核心键{list(final_state.keys())})logger.info(-*80)exceptExceptionase:# 7. 异常捕获打印详细错误信息logger.error(f 全流程测试运行失败 ,exc_infoTrue)logger.error(f异常原因{str(e)})logger.info( 知识图谱导入全流程测试结束 )输出日志—识别完成: HAK 180 烫金机—[Stub]执行节点node_bge_embedding—开始 BGE-M3 向量化处理 (node_bge_embedding)—成功获取第 1-5 项的嵌入。成功获取第 6-6 项的嵌入。— 向量化处理完成共处理 6 条数据 —[Stub] 执行节点node_import_milvus—开始导入 Milvus—从数据中检测到向量维度为: 1024从数据中检测到 item_name 为: HAK 180 烫金机正在连接到 Milvus 并准备集合 ‘kb_chunks’…幂等清理正在删除 Milvus 集合 ‘kb_chunks’ 中 item_nameHAK 180 烫金机 的旧切片…幂等清理完成item_nameHAK 180 烫金机准备了 6 条数据正在执行插入操作…成功插入 6 条数据。[Stub]执行节点node_import_kg 流程执行结束 ✅ 任务ID: full_graph_test_001✅ 文件标题: hak180产品安全手册✅ 识别商品: HAK 180 烫金机✅ 切片数量: 6✅ 向量化结果: 成功 (检测到 6 条带向量的数据)效果体现