# 掌柜智库项目(RAG)实战

## 5. 导入数据节点实现与测试

### 5.8 进行主图调用测试 (main_graph)

主图添加测试代码，进行流程完成测试！

```python
if __name__ == "__main__":
    from app.utils.path_util import PROJECT_ROOT
    import os

    # 全流程测试：验证PDF导入→Milvus入库→KG导入完整链路
    logger.info("===== 开始执行知识图谱导入全流程测试 =====")
    # 1. 构造测试文件路径（复用你项目的doc目录，和pdf2md测试文件一致）
    test_pdf_name = os.path.join("doc", "万用表RS-12的使用.pdf")
    test_pdf_path = os.path.join(PROJECT_ROOT, test_pdf_name)
    # 2. 构造输出目录（存放MD/图片等中间文件）
    test_output_dir = os.path.join(PROJECT_ROOT, "output")
    os.makedirs(test_output_dir, exist_ok=True)  # 不存在则创建

    # 3. 校验测试PDF文件是否存在
    if not os.path.exists(test_pdf_path):
        logger.error(f"全流程测试失败：测试PDF文件不存在，路径：{test_pdf_path}")
        logger.info("请检查文件路径，或手动将测试文件放入项目根目录的doc文件夹中")
    else:
        # 4. 构造测试状态（贴合实际业务入参，开启PDF解析开关）
        test_state = ImportGraphState({
            "task_id": "test_kg_import_workflow_001",  # 测试任务ID
            "user_id": "test_user",  # 测试用户ID
            "local_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_state = None
            for step in kb_import_app.stream(test_state, stream_mode="values"):
                # 打印当前执行完成的节点（流式输出更直观）
                current_node = list(step.keys())[-1] if step else "未知节点"
                logger.info(f"✅ 节点执行完成：{current_node}")
                final_state = step  # 保存最终状态

            # 6. 全流程执行完成，结果预览和核心指标打印
            if final_state:
                logger.info("-" * 80)
                logger.info("===== 全流程测试执行成功，核心结果预览 =====")
                # 提取核心结果指标
                chunks = final_state.get("chunks", [])
                chunk_count = len(chunks)
                md_content = final_state.get("md_content", "")[:150]  # MD内容前150字符
                has_embedding = all("dense_vector" in c and "sparse_vector" in c for c in chunks) if chunks else False
                has_chunk_id = all("chunk_id" in c for c in chunks) if chunks else False
                kg_id = final_state.get("kg_id", "未生成")  # KG导入生成的ID（按实际业务字段调整）

                # 打印核心指标
                logger.info(f"📄 PDF转MD内容预览（前150字符）：{md_content}...")
                logger.info(f"📝 文档切分总切片数：{chunk_count}")
                logger.info(f"🔍 所有切片是否完成向量化：{'是' if has_embedding else '否'}")
                logger.info(f"🗄️  所有切片是否完成Milvus入库（含chunk_id）：{'是' if has_chunk_id else '否'}")
                logger.info(f"🧠 知识图谱导入ID：{kg_id}")
                logger.info(f"📂 最终状态包含的核心键：{list(final_state.keys())}")
                logger.info("-" * 80)
        except Exception as e:
            # 7. 异常捕获，打印详细错误信息
            logger.error(f"===== 全流程测试运行失败 =====", exc_info=True)
            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_name=HAK 180 烫金机 的旧切片...
> 幂等清理完成：item_name=HAK 180 烫金机
> 准备了 6 条数据，正在执行插入操作...
> 成功插入 6 条数据。
> [Stub]执行节点：node_import_kg
> === 流程执行结束 ===
> ✅ 任务ID: full_graph_test_001
> ✅ 文件标题: hak180产品安全手册
> ✅ 识别商品: HAK 180 烫金机
> ✅ 切片数量: 6
> ✅ 向量化结果: 成功 (检测到 6 条带向量的数据)

效果体现：

<img src="images/1769398909789.png" alt="1769398909789"  />

