企业实战:导入数据节点

目录1 入口与类型判断 (node_entry)2 节点作用与实现思路3 步骤分解4 工具类解读任务追踪5 代码实现6 单元测试1 入口与类型判断 (node_entry)文件:app/import_process/agent/nodes/node_entry.py相关工具类位置:app/utils/task_utils.py2 节点作用与实现思路节点作用: 作为数据加载流程的“总调度员”负责接收外部输入的文件识别文件类型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_tasknode_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_pathstate[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_osos.path.basename(local_file_path).split(.)[0]# 基于pathlib处理file_titlePath(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.stemsplitext(file_name)拆分文件名和后缀test.pdf→(test, .pdf)state.get(key, )安全提取状态值无 key 时返回默认值state.get(a, )→ 无 “a” 则返回 “”sys._getframe().f_code.co_name动态获取当前函数名本节点中返回node_entryadd_running_task/add_done_task记录任务的节点运行状态用于任务监控面板展示节点执行进度(fastapi使用)6 单元测试您可以在node_entry.py文件底部直接运行以下测试代码if__name____main__:# 单元测试覆盖不支持类型、MD、PDF三种场景logger.info( 开始node_entry节点单元测试 )# 测试1: 不支持的TXT文件test_state1create_default_state(task_idtest_task_001,local_file_path联想海豚用户手册.txt)node_entry(test_state1)# 测试2: MD文件test_state2create_default_state(task_idtest_task_002,local_file_path小米用户手册.md)node_entry(test_state2)# 测试3: PDF文件test_state3create_default_state(task_idtest_task_003,local_file_path万用表的使用.pdf)node_entry(test_state3)logger.info( 结束node_entry节点单元测试 )