项目背景为企业客户提供一站式智能知识管理解决方案涵盖文档自动化解析、多模态内容提取、语义检索、MCP检索等功能。技术栈FastAPI Uvicorn LangGraph LangChain BGE-M3 Milvus MongoDB MinIO MinerU整体流程图单节点数据流图1、任务分发关键代码展示def process(self, state: ImportGraphState): # 1、参数非空校验 local_file_path state.get(local_file_path) if not local_file_path: raise ValueError(请指定文件路径) # 2、提取文件名称 file_title splitext(os.path.basename(local_file_path))[0] # 3、文件类型检查并分发 if local_file_path.endswith(.pdf): return { is_pdf_read_enabled: True, pdf_path: local_file_path, file_title: file_title, } elif local_file_path.endswith(.md): return { is_md_read_enabled: True, md_path: local_file_path, file_title: file_title, } else: raise ValueError(f不支持的文件类型{current_type})2、PDF转MarkdownStep2上传 轮询Step3下载 解压关键代码展示文件上传与轮询def _step2_upload_and_poll(self, pdf_path_obj: Path): # 1、申请上传链接 url f{self._get_base_url()}/file-urls/batch header {Authorization: fBearer {token}} data {files: [{name: pdf_path_obj.name}], model_version: vlm} response requests.post(url, headersheader, jsondata) result response.json() batch_id result[data][batch_id] signed_url result[data][file_urls][0] # 2、执行文件上传 with open(pdf_path_obj, rb) as f: res_upload requests.put(signed_url, dataf) # 3、轮询解析结果最大600秒 start_time time.time() while True: elapsed_time time.time() - start_time if elapsed_time 600: raise RuntimeError(任务超时) res requests.get(f{base_url}/extract-results/batch/{batch_id}, headersheader) poll_data res.json() data_state poll_data[data][extract_result][0][state] if data_state done: return poll_data[data][extract_result][0][full_zip_url] elif data_state failed: raise RuntimeError(任务失败) else: time.sleep(3) # 继续轮询ZIP下载与解压def _step3_download_and_extract(self, zip_url, output_dir_obj, file_title): # 1、下载ZIP response requests.get(zip_url) zip_save_path output_dir_obj / f{file_title}.zip with open(zip_save_path, wb) as f: f.write(response.content) # 2、删除旧目录并解压 unzip_dir_obj output_dir_obj / file_title if unzip_dir_obj.exists(): shutil.rmtree(unzip_dir_obj) unzip_dir_obj.mkdir(parentsTrue, exist_okTrue) with ZipFile(zip_save_path, r) as zip_file: zip_file.extractall(unzip_dir_obj) # 3、重命名MD文件 md_file_obj unzip_dir_obj / full.md new_md_path md_file_obj.with_name(file_title .md) md_file_obj.rename(new_md_path) return str(new_md_path.absolute())3、多模态图片处理Step2扫描图片Step3生成摘要 速率限制Step4上传 替换关键代码展示图片扫描与上下文提取def _step2_scan_images(self, md_content: str, images_dir: Path): images [] for image_file in os.listdir(images_dir): # 过滤图片格式 if Path(image_file).suffix.lower() not in {.jpg, .jpeg, .png, .gif, .bmp, .webp}: continue # 在MD中查找图片引用获取上下文 context self._find_image_in_md(md_content, image_file) image_path str(images_dir / image_file) images.append((image_file, image_path, context)) return imagesdef _find_image_in_md(self, md_content: str, image_file: str): # 正则匹配图片引用提取前后文各100字符 pattern re.compile(r!\[.*?\]\(.*? re.escape(image_file) r.*?\)) match pattern.search(md_content) if not match: return None start, end match.span() pre_text md_content[max(0, start - 100):start] post_text md_content[end:min(end 100, len(md_content))] return pre_text, post_textAPI限速器滑动窗口def _apply_api_rate_limit(self, request_deque: Deque[float], max_requests: int, window_size: int 60): current_time time.time() # 移除过期的请求时间戳 while request_deque and current_time - request_deque[0] window_size: request_deque.popleft() # 如果达到上限等待 if len(request_deque) max_requests: sleep_duration window_size - (current_time - request_deque[0]) if sleep_duration 0: time.sleep(sleep_duration) current_time time.time() while request_deque and current_time - request_deque[0] window_size: request_deque.popleft() # 入队当前请求 request_deque.append(current_time)VLM多模态图片理解def _summarize_image(self, image_path: str, file_title: str, context: Tuple[str, str]): # 1、图片转base64 with open(image_path, rb) as f: image_data f.read() image_base64 base64.b64encode(image_data).decode(utf-8) # 2、构造多模态消息 prompt IMAGE_SUMMARY.format(file_titlefile_title, contextcontext) messages [ { role: user, content: [ {type: text, text: prompt}, {type: image_url, image_url: {url: fdata:image/jpg;base64,{image_base64}}} ] } ] # 3、调用VLM模型 chat_model ChatOpenAI(modellm_config.vl_model, api_keylm_config.api_key, base_urllm_config.base_url) response chat_model.invoke(messages) return response.content.strip().replace(\n, )MinIO上传与MD替换def _step4_upload_and_replace(self, file_title, images, summaries, md_content): # 1、构造MinIO目录 upload_dir f{minio_config.img_dir}/{file_title}.replace( , ) # 2、清理旧数据幂等性 self._clean_minio_directory(upload_dir) # 3、批量上传图片 urls {} for image_file, image_path, context in images: object_name f{upload_dir}/{image_file} minio_client.fput_object(minio_config.bucket_name, object_name, image_path) urls[image_file] fhttp://{minio_config.endpoint}/{minio_config.bucket_name}/{object_name} # 4、替换MD中的图片引用 for image_file, (summary, url) in image_info.items(): pattern re.compile(r!\[.*?\]\(.*? re.escape(image_file) r\)) md_content pattern.sub(lambda x: f, md_content) return md_content4、文档切分Step2按标题切分Step4-1长内容切分Step4-2短内容合并关键代码展示按标题初切识别标题和代码块def _step2_split_by_titles(self, content, file_title): title_pattern r^\s*#{1,6}\s. code_pattern r^({3,}|~{3,}) in_code_block False current_lines [] sections [] current_title title_count 0 def _flush_section(): if not current_lines: return nonlocal title_count title_count 1 sections.append({ file_title: file_title, title: current_title or 无标题, content: \n.join(current_lines) }) content content.replace(\r\n, \n).replace(\r, \n) for line in content.split(\n): stripped_line line.strip() # 识别代码围栏切换状态 code_match re.match(code_pattern, stripped_line) if code_match: in_code_block not in_code_block continue # 识别标题不在代码块内 if not in_code_block and re.match(title_pattern, stripped_line): _flush_section() current_title stripped_line current_lines [current_title] else: current_lines.append(stripped_line) _flush_section() return sections, title_count, len(content.split(\n))长内容二次切分def _split_long_section(self, section: Dict[str, str]) - List[Dict[str, str]]: content section.get(content) title section.get(title) # 长度不足或包含表格则不切分 if len(content) 500 or tablein content.lower(): return [section] # 计算可用长度扣除标题 prefix f{title}\n\n available_len 500 - len(prefix) if available_len 0: return [section] # 去除标题前缀 body content if title and body.startswith(title): body body[body.find(title) len(title):].lstrip() # 使用LangChain递归切分器 splitter RecursiveCharacterTextSplitter( chunk_sizeavailable_len, chunk_overlap50, separators[\n\n, \n, 。, , , , ., !, ?, ;, ] ) sub_sections [] for idx, chunk in enumerate(splitter.split_text(body), start1): text chunk.strip() if not text: continue sub_sections.append({ title: f{title} - {idx}, content: prefix text, parent_title: title, part: idx, file_title: section.get(file_title) }) return sub_sections短内容合并def _merge_short_sections(self, sections: List[Dict[str, str]]) - List[Dict[str, str]]: if not sections: return [] merged_sections [] current_chunk None for sec in sections: if current_chunk is None: current_chunk sec continue # 合并条件当前chunk短100且同父标题 is_current_short len(current_chunk[content]) 100 is_same_parent current_chunk.get(parent_title) sec.get(parent_title) if is_current_short and is_same_parent: # 去除重复标题前缀后合并 parent_title sec.get(parent_title, ) next_content sec.get(content) if parent_title and next_content.startswith(parent_title): next_content next_content[len(parent_title):].lstrip() current_chunk[content] \n\n next_content current_chunk[part] sec[part] else: merged_sections.append(current_chunk) current_chunk sec if current_chunk is not None: merged_sections.append(current_chunk) return merged_sectionsJSON备份def _step6_backup(self, state, sections): try: backup_path Path(state.get(local_dir)) / state.get(file_title) / chunks.json with open(backup_path, w, encodingutf-8) as f: json.dump(sections, f, ensure_asciiFalse, indent2) logger.info(f备份成功{backup_path}) except Exception as e: logger.error(f备份失败{str(e)})注本文只提供项目思路及关键代码不提供项目完整代码。01什么是AI大模型应用开发工程师如果说AI大模型是蕴藏着巨大能量的“后台超级能力”那么AI大模型应用开发工程师就是将这种能量转化为实用工具的执行者。AI大模型应用开发工程师是基于AI大模型设计开发落地业务的应用工程师。这个职业的核心价值在于打破技术与用户之间的壁垒把普通人难以理解的算法逻辑、模型参数转化为人人都能轻松操作的产品形态。无论是日常写作时用到的AI文案生成器、修图软件里的智能美化功能还是办公场景中的自动记账工具、会议记录用的语音转文字APP这些看似简单的应用背后都是应用开发工程师在默默搭建技术与需求之间的桥梁。他们不追求创造全新的大模型而是专注于让已有的大模型“听懂”业务需求“学会”解决具体问题最终形成可落地、可使用的产品。CSDN粉丝独家福利给大家整理了一份AI大模型全套学习资料这份完整版的 AI 大模型学习资料已经上传CSDN朋友们如果需要可以扫描下方二维码点击下方CSDN官方认证链接免费领取【保证100%免费】02AI大模型应用开发工程师的核心职责需求分析与拆解是工作的起点也是确保开发不偏离方向的关键。应用开发工程师需要直接对接业务方深入理解其核心诉求——不仅要明确“要做什么”更要厘清“为什么要做”以及“做到什么程度算合格”。在此基础上他们会将模糊的业务需求拆解为具体的技术任务明确每个环节的执行标准并评估技术实现的可行性同时定义清晰的核心指标为后续开发、测试提供依据。这一步就像建筑前的图纸设计若出现偏差后续所有工作都可能白费。技术选型与适配是衔接需求与开发的核心环节。工程师需要根据业务场景的特点选择合适的基础大模型、开发框架和工具——不同的业务对模型的响应速度、精度、成本要求不同选型的合理性直接影响最终产品的表现。同时他们还要对行业相关数据进行预处理通过提示词工程优化模型输出或在必要时进行轻量化微调让基础模型更好地适配具体业务。此外设计合理的上下文管理规则确保模型理解连贯需求建立敏感信息过滤机制保障数据安全也是这一环节的重要内容。应用开发与对接则是将方案转化为产品的实操阶段。工程师会利用选定的开发框架构建应用的核心功能同时联动各类外部系统——比如将AI模型与企业现有的客户管理系统、数据存储系统打通确保数据流转顺畅。在这一过程中他们还需要配合设计团队打磨前端交互界面让技术功能以简洁易懂的方式呈现给用户实现从技术方案到产品形态的转化。测试与优化是保障产品质量的关键步骤。工程师会开展全面的功能测试找出并修复开发过程中出现的漏洞同时针对模型的响应速度、稳定性等性能指标进行优化。安全合规性也是测试的重点需要确保应用符合数据保护、隐私安全等相关规定。此外他们还会收集用户反馈通过调整模型参数、优化提示词等方式持续提升产品体验让应用更贴合用户实际使用需求。部署运维与迭代则贯穿产品的整个生命周期。工程师会通过云服务器或私有服务器将应用部署上线并实时监控运行状态及时处理突发故障确保应用稳定运行。随着业务需求的变化他们还需要对应用功能进行迭代更新同时编写完善的开发文档和使用手册为后续的维护和交接提供支持。03薪资情况与职业价值市场对这一职业的高度认可直接体现在薪资待遇上。据猎聘最新在招岗位数据显示AI大模型应用开发工程师的月薪最高可达60k。在AI技术加速落地的当下这种“技术业务”的复合型能力尤为稀缺让该职业成为当下极具吸引力的就业选择。AI大模型应用开发工程师是AI技术落地的关键桥梁。他们用专业能力将抽象的技术转化为具体的产品让大模型的价值真正渗透到各行各业。随着AI场景化应用的不断深化这一职业的重要性将更加凸显也必将吸引更多人才投身其中推动AI技术更好地服务于社会发展。CSDN粉丝独家福利给大家整理了一份AI大模型全套学习资料这份完整版的 AI 大模型学习资料已经上传CSDN朋友们如果需要可以扫描下方二维码点击下方CSDN官方认证链接免费领取【保证100%免费】