企业实战:导入数据节点
目录
- 1 入口与类型判断 (node_entry)
- 2 节点作用与实现思路
- 3 步骤分解
- 4 工具类解读：任务追踪
- 5 代码实现
- 6 单元测试
1 入口与类型判断 (node_entry)
文件:app/import_process/agent/nodes/node_entry.py
相关工具类位置:app/utils/task_utils.py
2 节点作用与实现思路
节点作用: 作为数据加载流程的“总调度员”，负责接收外部输入的文件，识别文件类型（PDF或Markdown），并根据类型开启相应的处理分支。同时，它提取文件名作为全局元数据，并初始化任务追踪状态，确保后续流程可追溯、可监控。
实现思路:
- 路由分发: 采用轻量级的条件判断逻辑，通过文件后缀 (
.pdf/.md) 决定激活is_pdf_read_enabled还是is_md_read_enabled状态位，实现不同格式文件的差异化处理。 - 元数据提取: 在入口处统一提任务标记(
task_id)，作为贯穿整个知识库构建流程的唯一标识，避免后续节点重复解析。 - 任务监控初始化: 集成
task_utils，记录当前任务 ID 和初始状态，为前端提供实时的进度反馈。
3 步骤分解
- 接收状态: 获取
local_file_path。 - 判断类型: 检查文件后缀是
.pdf还是.md。 - 设置标记: 更新 state 中的
is_pdf_read_enabled或is_md_read_enabled，供主图路由使用。 - 提取标题: 从文件名中提取
file_title，后续作为元数据。
4 工具类解读：任务追踪
文件:app/utils/task_utils.py
实现思路:
- 内存管理: 使用简单的内存字典
_tasks_running_list和_tasks_done_list记录任务状态，轻量高效。 - 状态映射: 维护
_NODE_NAME_TO_CN字典，将技术性的节点名称（如node_entry）映射为用户友好的中文名称（如检查文件），方便前端展示。 - SSE 集成: 集成 SSE (Server-Sent Events) 推送机制，允许实时将任务进度推送到前端。
- 操作封装: 提供
add_running_task和add_done_task接口，方便各节点调用，屏蔽底层状态管理细节。
5 代码实现
importosfrompathlibimportPathfromapp.core.loggerimportlogger,node_logfromapp.import_process.agent.stateimportImportGraphState,create_default_statefromapp.utils.task_utilsimportadd_running_task,add_done_task@node_log("node_entry")defnode_entry(state:ImportGraphState)->ImportGraphState:""" 节点: 入口节点 (node_entry) 为什么叫这个名字: 作为图的 Entry Point，负责接收外部输入并决定流程走向。 未来要实现: 1. 进行任务状态记录,开始和结束列表记录 2. 根据state中 local_file_path属性判断数据类型进而修改 相关参数, is_md_read_enabled 或者 is_pdf_read_enabled md_path 或者 pdf_path 3. 不可解析结果类型不可用,直接输出对应警告日志! 逻辑路由节点会自动处理 4. 获取file_tile标识,用于后期识别pdf对应的主体(item_name)进行兜底 """# 1. 任务状态记录处理add_running_task(state['task_id'],'node_entry')# 2. 判断文件类型local_file_path=state['local_file_path']ifnotlocal_file_path:logger.warning(f"没有输入文件地址,无法处理,直接跳转到结束节点!")add_done_task(state['task_id'],'node_entry')returnstateiflocal_file_path.endswith(".md"):state['is_md_read_enabled']=Truestate['md_path']=local_file_patheliflocal_file_path.endswith(".pdf"):state['is_pdf_read_enabled']=Truestate['pdf_path']=local_file_pathelse:logger.warning(f"虽然输出了loclal_file_path,但是无法识别文件类型,请检查输入文件类型是否正确,目前只支持md和pdf文件,请检查!{local_file_path}")add_done_task(state['task_id'],'node_entry')returnstate# 3. 获取文件标识# 基于os.path处理file_title_os=os.path.basename(local_file_path).split(".")[0]# 基于pathlib处理file_title=Path(local_file_path).stem# 文件名 .name 文件夹名 .parent 文件后缀 .suffixstate['file_title']=file_title add_done_task(state['task_id'],'node_entry')returnstate关键语法补充说明
| 语法 / 函数 | 具体作用 | 执行示例 |
|---|---|---|
os.path.basename(document_path) | 从完整路径提取文件名 | /data/kb/test.pdf→test.pdfp.stem |
splitext(file_name) | 拆分文件名和后缀 | test.pdf→('test', '.pdf') |
state.get("key", "") | 安全提取状态值，无 key 时返回默认值 | state.get("a", "")→ 无 “a” 则返回 “” |
sys._getframe().f_code.co_name | 动态获取当前函数名 | 本节点中返回node_entry |
add_running_task/add_done_task | 记录任务的节点运行状态 | 用于任务监控面板，展示节点执行进度(fastapi使用) |
6 单元测试
您可以在node_entry.py文件底部直接运行以下测试代码：
if__name__=='__main__':# 单元测试：覆盖不支持类型、MD、PDF三种场景logger.info("===== 开始node_entry节点单元测试 =====")# 测试1: 不支持的TXT文件test_state1=create_default_state(task_id="test_task_001",local_file_path="联想海豚用户手册.txt")node_entry(test_state1)# 测试2: MD文件test_state2=create_default_state(task_id="test_task_002",local_file_path="小米用户手册.md")node_entry(test_state2)# 测试3: PDF文件test_state3=create_default_state(task_id="test_task_003",local_file_path="万用表的使用.pdf")node_entry(test_state3)logger.info("===== 结束node_entry节点单元测试 =====")