管理系統(tǒng)深度解析：從PDF轉(zhuǎn)Markdown到多模態(tài)處理的全流程技術(shù))
項(xiàng)目背景為企業(yè)客戶提供一站式智能知識(shí)管理解決方案涵蓋文檔自動(dòng)化解析、多模態(tài)內(nèi)容提取、語義檢索、MCP檢索等功能。技術(shù)棧FastAPI Uvicorn LangGraph LangChain BGE-M3 Milvus MongoDB MinIO MinerU整體流程圖單節(jié)點(diǎn)數(shù)據(jù)流圖1、任務(wù)分發(fā)關(guān)鍵代碼展示def process(self, state: ImportGraphState): # 1、參數(shù)非空校驗(yàn) local_file_path state.get(local_file_path) if not local_file_path: raise ValueError(請(qǐng)指定文件路徑) # 2、提取文件名稱 file_title splitext(os.path.basename(local_file_path))[0] # 3、文件類型檢查并分發(fā) 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轉(zhuǎn)MarkdownStep2上傳 輪詢Step3下載 解壓關(guān)鍵代碼展示文件上傳與輪詢def _step2_upload_and_poll(self, pdf_path_obj: Path): # 1、申請(qǐng)上傳鏈接 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、執(zhí)行文件上傳 with open(pdf_path_obj, rb) as f: res_upload requests.put(signed_url, dataf) # 3、輪詢解析結(jié)果最大600秒 start_time time.time() while True: elapsed_time time.time() - start_time if elapsed_time 600: raise RuntimeError(任務(wù)超時(shí)) 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(任務(wù)失敗) else: time.sleep(3) # 繼續(xù)輪詢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、多模態(tài)圖片處理Step2掃描圖片Step3生成摘要 速率限制Step4上傳 替換關(guān)鍵代碼展示圖片掃描與上下文提取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限速器滑動(dòng)窗口def _apply_api_rate_limit(self, request_deque: Deque[float], max_requests: int, window_size: int 60): current_time time.time() # 移除過期的請(qǐng)求時(shí)間戳 while request_deque and current_time - request_deque[0] window_size: request_deque.popleft() # 如果達(dá)到上限等待 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() # 入隊(duì)當(dāng)前請(qǐng)求 request_deque.append(current_time)VLM多模態(tài)圖片理解def _summarize_image(self, image_path: str, file_title: str, context: Tuple[str, str]): # 1、圖片轉(zhuǎn)base64 with open(image_path, rb) as f: image_data f.read() image_base64 base64.b64encode(image_data).decode(utf-8) # 2、構(gòu)造多模態(tài)消息 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、調(diào)用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、構(gòu)造MinIO目錄 upload_dir f{minio_config.img_dir}/{file_title}.replace( , ) # 2、清理舊數(shù)據(jù)冪等性 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按標(biāo)題切分Step4-1長(zhǎng)內(nèi)容切分Step4-2短內(nèi)容合并關(guān)鍵代碼展示按標(biāo)題初切識(shí)別標(biāo)題和代碼塊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 無標(biāo)題, content: \n.join(current_lines) }) content content.replace(\r\n, \n).replace(\r, \n) for line in content.split(\n): stripped_line line.strip() # 識(shí)別代碼圍欄切換狀態(tài) code_match re.match(code_pattern, stripped_line) if code_match: in_code_block not in_code_block continue # 識(shí)別標(biāo)題不在代碼塊內(nèi) 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))長(zhǎng)內(nèi)容二次切分def _split_long_section(self, section: Dict[str, str]) - List[Dict[str, str]]: content section.get(content) title section.get(title) # 長(zhǎng)度不足或包含表格則不切分 if len(content) 500 or tablein content.lower(): return [section] # 計(jì)算可用長(zhǎng)度扣除標(biāo)題 prefix f{title}\n\n available_len 500 - len(prefix) if available_len 0: return [section] # 去除標(biāo)題前綴 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短內(nèi)容合并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 # 合并條件當(dāng)前chunk短100且同父標(biāo)題 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: # 去除重復(fù)標(biāo)題前綴后合并 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)})注本文只提供項(xiàng)目思路及關(guān)鍵代碼不提供項(xiàng)目完整代碼。01什么是AI大模型應(yīng)用開發(fā)工程師如果說AI大模型是蘊(yùn)藏著巨大能量的“后臺(tái)超級(jí)能力”那么AI大模型應(yīng)用開發(fā)工程師就是將這種能量轉(zhuǎn)化為實(shí)用工具的執(zhí)行者。AI大模型應(yīng)用開發(fā)工程師是基于AI大模型設(shè)計(jì)開發(fā)落地業(yè)務(wù)的應(yīng)用工程師。這個(gè)職業(yè)的核心價(jià)值在于打破技術(shù)與用戶之間的壁壘把普通人難以理解的算法邏輯、模型參數(shù)轉(zhuǎn)化為人人都能輕松操作的產(chǎn)品形態(tài)。無論是日常寫作時(shí)用到的AI文案生成器、修圖軟件里的智能美化功能還是辦公場(chǎng)景中的自動(dòng)記賬工具、會(huì)議記錄用的語音轉(zhuǎn)文字APP這些看似簡(jiǎn)單的應(yīng)用背后都是應(yīng)用開發(fā)工程師在默默搭建技術(shù)與需求之間的橋梁。他們不追求創(chuàng)造全新的大模型而是專注于讓已有的大模型“聽懂”業(yè)務(wù)需求“學(xué)會(huì)”解決具體問題最終形成可落地、可使用的產(chǎn)品。CSDN粉絲獨(dú)家福利給大家整理了一份AI大模型全套學(xué)習(xí)資料這份完整版的 AI 大模型學(xué)習(xí)資料已經(jīng)上傳CSDN朋友們?nèi)绻枰梢話呙柘路蕉S碼點(diǎn)擊下方CSDN官方認(rèn)證鏈接免費(fèi)領(lǐng)取【保證100%免費(fèi)】02AI大模型應(yīng)用開發(fā)工程師的核心職責(zé)需求分析與拆解是工作的起點(diǎn)也是確保開發(fā)不偏離方向的關(guān)鍵。應(yīng)用開發(fā)工程師需要直接對(duì)接業(yè)務(wù)方深入理解其核心訴求——不僅要明確“要做什么”更要厘清“為什么要做”以及“做到什么程度算合格”。在此基礎(chǔ)上他們會(huì)將模糊的業(yè)務(wù)需求拆解為具體的技術(shù)任務(wù)明確每個(gè)環(huán)節(jié)的執(zhí)行標(biāo)準(zhǔn)并評(píng)估技術(shù)實(shí)現(xiàn)的可行性同時(shí)定義清晰的核心指標(biāo)為后續(xù)開發(fā)、測(cè)試提供依據(jù)。這一步就像建筑前的圖紙?jiān)O(shè)計(jì)若出現(xiàn)偏差后續(xù)所有工作都可能白費(fèi)。技術(shù)選型與適配是銜接需求與開發(fā)的核心環(huán)節(jié)。工程師需要根據(jù)業(yè)務(wù)場(chǎng)景的特點(diǎn)選擇合適的基礎(chǔ)大模型、開發(fā)框架和工具——不同的業(yè)務(wù)對(duì)模型的響應(yīng)速度、精度、成本要求不同選型的合理性直接影響最終產(chǎn)品的表現(xiàn)。同時(shí)他們還要對(duì)行業(yè)相關(guān)數(shù)據(jù)進(jìn)行預(yù)處理通過提示詞工程優(yōu)化模型輸出或在必要時(shí)進(jìn)行輕量化微調(diào)讓基礎(chǔ)模型更好地適配具體業(yè)務(wù)。此外設(shè)計(jì)合理的上下文管理規(guī)則確保模型理解連貫需求建立敏感信息過濾機(jī)制保障數(shù)據(jù)安全也是這一環(huán)節(jié)的重要內(nèi)容。應(yīng)用開發(fā)與對(duì)接則是將方案轉(zhuǎn)化為產(chǎn)品的實(shí)操階段。工程師會(huì)利用選定的開發(fā)框架構(gòu)建應(yīng)用的核心功能同時(shí)聯(lián)動(dòng)各類外部系統(tǒng)——比如將AI模型與企業(yè)現(xiàn)有的客戶管理系統(tǒng)、數(shù)據(jù)存儲(chǔ)系統(tǒng)打通確保數(shù)據(jù)流轉(zhuǎn)順暢。在這一過程中他們還需要配合設(shè)計(jì)團(tuán)隊(duì)打磨前端交互界面讓技術(shù)功能以簡(jiǎn)潔易懂的方式呈現(xiàn)給用戶實(shí)現(xiàn)從技術(shù)方案到產(chǎn)品形態(tài)的轉(zhuǎn)化。測(cè)試與優(yōu)化是保障產(chǎn)品質(zhì)量的關(guān)鍵步驟。工程師會(huì)開展全面的功能測(cè)試找出并修復(fù)開發(fā)過程中出現(xiàn)的漏洞同時(shí)針對(duì)模型的響應(yīng)速度、穩(wěn)定性等性能指標(biāo)進(jìn)行優(yōu)化。安全合規(guī)性也是測(cè)試的重點(diǎn)需要確保應(yīng)用符合數(shù)據(jù)保護(hù)、隱私安全等相關(guān)規(guī)定。此外他們還會(huì)收集用戶反饋通過調(diào)整模型參數(shù)、優(yōu)化提示詞等方式持續(xù)提升產(chǎn)品體驗(yàn)讓應(yīng)用更貼合用戶實(shí)際使用需求。部署運(yùn)維與迭代則貫穿產(chǎn)品的整個(gè)生命周期。工程師會(huì)通過云服務(wù)器或私有服務(wù)器將應(yīng)用部署上線并實(shí)時(shí)監(jiān)控運(yùn)行狀態(tài)及時(shí)處理突發(fā)故障確保應(yīng)用穩(wěn)定運(yùn)行。隨著業(yè)務(wù)需求的變化他們還需要對(duì)應(yīng)用功能進(jìn)行迭代更新同時(shí)編寫完善的開發(fā)文檔和使用手冊(cè)為后續(xù)的維護(hù)和交接提供支持。03薪資情況與職業(yè)價(jià)值市場(chǎng)對(duì)這一職業(yè)的高度認(rèn)可直接體現(xiàn)在薪資待遇上。據(jù)獵聘最新在招崗位數(shù)據(jù)顯示AI大模型應(yīng)用開發(fā)工程師的月薪最高可達(dá)60k。在AI技術(shù)加速落地的當(dāng)下這種“技術(shù)業(yè)務(wù)”的復(fù)合型能力尤為稀缺讓該職業(yè)成為當(dāng)下極具吸引力的就業(yè)選擇。AI大模型應(yīng)用開發(fā)工程師是AI技術(shù)落地的關(guān)鍵橋梁。他們用專業(yè)能力將抽象的技術(shù)轉(zhuǎn)化為具體的產(chǎn)品讓大模型的價(jià)值真正滲透到各行各業(yè)。隨著AI場(chǎng)景化應(yīng)用的不斷深化這一職業(yè)的重要性將更加凸顯也必將吸引更多人才投身其中推動(dòng)AI技術(shù)更好地服務(wù)于社會(huì)發(fā)展。CSDN粉絲獨(dú)家福利給大家整理了一份AI大模型全套學(xué)習(xí)資料這份完整版的 AI 大模型學(xué)習(xí)資料已經(jīng)上傳CSDN朋友們?nèi)绻枰梢話呙柘路蕉S碼點(diǎn)擊下方CSDN官方認(rèn)證鏈接免費(fèi)領(lǐng)取【保證100%免費(fèi)】 AI知識(shí)管理系統(tǒng)深度解析：從PDF轉(zhuǎn)Markdown到多模態(tài)處理的全流程技術(shù)
項(xiàng)目背景為企業(yè)客戶提供一站式智能知識(shí)管理解決方案涵蓋文檔自動(dòng)化解析、多模態(tài)內(nèi)容提取、語義檢索、MCP檢索等功能。技術(shù)棧FastAPI Uvicorn LangGraph LangChain BGE-M3 Milvus MongoDB MinIO MinerU整體流程圖單節(jié)點(diǎn)數(shù)據(jù)流圖1、任務(wù)分發(fā)關(guān)鍵代碼展示def process(self, state: ImportGraphState): # 1、參數(shù)非空校驗(yàn) local_file_path state.get(local_file_path) if not local_file_path: raise ValueError(請(qǐng)指定文件路徑) # 2、提取文件名稱 file_title splitext(os.path.basename(local_file_path))[0] # 3、文件類型檢查并分發(fā) 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轉(zhuǎn)MarkdownStep2上傳 輪詢Step3下載 解壓關(guān)鍵代碼展示文件上傳與輪詢def _step2_upload_and_poll(self, pdf_path_obj: Path): # 1、申請(qǐng)上傳鏈接 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、執(zhí)行文件上傳 with open(pdf_path_obj, rb) as f: res_upload requests.put(signed_url, dataf) # 3、輪詢解析結(jié)果最大600秒 start_time time.time() while True: elapsed_time time.time() - start_time if elapsed_time 600: raise RuntimeError(任務(wù)超時(shí)) 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(任務(wù)失敗) else: time.sleep(3) # 繼續(xù)輪詢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、多模態(tài)圖片處理Step2掃描圖片Step3生成摘要 速率限制Step4上傳 替換關(guān)鍵代碼展示圖片掃描與上下文提取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限速器滑動(dòng)窗口def _apply_api_rate_limit(self, request_deque: Deque[float], max_requests: int, window_size: int 60): current_time time.time() # 移除過期的請(qǐng)求時(shí)間戳 while request_deque and current_time - request_deque[0] window_size: request_deque.popleft() # 如果達(dá)到上限等待 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() # 入隊(duì)當(dāng)前請(qǐng)求 request_deque.append(current_time)VLM多模態(tài)圖片理解def _summarize_image(self, image_path: str, file_title: str, context: Tuple[str, str]): # 1、圖片轉(zhuǎn)base64 with open(image_path, rb) as f: image_data f.read() image_base64 base64.b64encode(image_data).decode(utf-8) # 2、構(gòu)造多模態(tài)消息 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、調(diào)用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、構(gòu)造MinIO目錄 upload_dir f{minio_config.img_dir}/{file_title}.replace( , ) # 2、清理舊數(shù)據(jù)冪等性 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按標(biāo)題切分Step4-1長(zhǎng)內(nèi)容切分Step4-2短內(nèi)容合并關(guān)鍵代碼展示按標(biāo)題初切識(shí)別標(biāo)題和代碼塊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 無標(biāo)題, content: \n.join(current_lines) }) content content.replace(\r\n, \n).replace(\r, \n) for line in content.split(\n): stripped_line line.strip() # 識(shí)別代碼圍欄切換狀態(tài) code_match re.match(code_pattern, stripped_line) if code_match: in_code_block not in_code_block continue # 識(shí)別標(biāo)題不在代碼塊內(nèi) 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))長(zhǎng)內(nèi)容二次切分def _split_long_section(self, section: Dict[str, str]) - List[Dict[str, str]]: content section.get(content) title section.get(title) # 長(zhǎng)度不足或包含表格則不切分 if len(content) 500 or tablein content.lower(): return [section] # 計(jì)算可用長(zhǎng)度扣除標(biāo)題 prefix f{title}\n\n available_len 500 - len(prefix) if available_len 0: return [section] # 去除標(biāo)題前綴 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短內(nèi)容合并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 # 合并條件當(dāng)前chunk短100且同父標(biāo)題 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: # 去除重復(fù)標(biāo)題前綴后合并 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)})注本文只提供項(xiàng)目思路及關(guān)鍵代碼不提供項(xiàng)目完整代碼。01什么是AI大模型應(yīng)用開發(fā)工程師如果說AI大模型是蘊(yùn)藏著巨大能量的“后臺(tái)超級(jí)能力”那么AI大模型應(yīng)用開發(fā)工程師就是將這種能量轉(zhuǎn)化為實(shí)用工具的執(zhí)行者。AI大模型應(yīng)用開發(fā)工程師是基于AI大模型設(shè)計(jì)開發(fā)落地業(yè)務(wù)的應(yīng)用工程師。這個(gè)職業(yè)的核心價(jià)值在于打破技術(shù)與用戶之間的壁壘把普通人難以理解的算法邏輯、模型參數(shù)轉(zhuǎn)化為人人都能輕松操作的產(chǎn)品形態(tài)。無論是日常寫作時(shí)用到的AI文案生成器、修圖軟件里的智能美化功能還是辦公場(chǎng)景中的自動(dòng)記賬工具、會(huì)議記錄用的語音轉(zhuǎn)文字APP這些看似簡(jiǎn)單的應(yīng)用背后都是應(yīng)用開發(fā)工程師在默默搭建技術(shù)與需求之間的橋梁。他們不追求創(chuàng)造全新的大模型而是專注于讓已有的大模型“聽懂”業(yè)務(wù)需求“學(xué)會(huì)”解決具體問題最終形成可落地、可使用的產(chǎn)品。CSDN粉絲獨(dú)家福利給大家整理了一份AI大模型全套學(xué)習(xí)資料這份完整版的 AI 大模型學(xué)習(xí)資料已經(jīng)上傳CSDN朋友們?nèi)绻枰梢話呙柘路蕉S碼點(diǎn)擊下方CSDN官方認(rèn)證鏈接免費(fèi)領(lǐng)取【保證100%免費(fèi)】02AI大模型應(yīng)用開發(fā)工程師的核心職責(zé)需求分析與拆解是工作的起點(diǎn)也是確保開發(fā)不偏離方向的關(guān)鍵。應(yīng)用開發(fā)工程師需要直接對(duì)接業(yè)務(wù)方深入理解其核心訴求——不僅要明確“要做什么”更要厘清“為什么要做”以及“做到什么程度算合格”。在此基礎(chǔ)上他們會(huì)將模糊的業(yè)務(wù)需求拆解為具體的技術(shù)任務(wù)明確每個(gè)環(huán)節(jié)的執(zhí)行標(biāo)準(zhǔn)并評(píng)估技術(shù)實(shí)現(xiàn)的可行性同時(shí)定義清晰的核心指標(biāo)為后續(xù)開發(fā)、測(cè)試提供依據(jù)。這一步就像建筑前的圖紙?jiān)O(shè)計(jì)若出現(xiàn)偏差后續(xù)所有工作都可能白費(fèi)。技術(shù)選型與適配是銜接需求與開發(fā)的核心環(huán)節(jié)。工程師需要根據(jù)業(yè)務(wù)場(chǎng)景的特點(diǎn)選擇合適的基礎(chǔ)大模型、開發(fā)框架和工具——不同的業(yè)務(wù)對(duì)模型的響應(yīng)速度、精度、成本要求不同選型的合理性直接影響最終產(chǎn)品的表現(xiàn)。同時(shí)他們還要對(duì)行業(yè)相關(guān)數(shù)據(jù)進(jìn)行預(yù)處理通過提示詞工程優(yōu)化模型輸出或在必要時(shí)進(jìn)行輕量化微調(diào)讓基礎(chǔ)模型更好地適配具體業(yè)務(wù)。此外設(shè)計(jì)合理的上下文管理規(guī)則確保模型理解連貫需求建立敏感信息過濾機(jī)制保障數(shù)據(jù)安全也是這一環(huán)節(jié)的重要內(nèi)容。應(yīng)用開發(fā)與對(duì)接則是將方案轉(zhuǎn)化為產(chǎn)品的實(shí)操階段。工程師會(huì)利用選定的開發(fā)框架構(gòu)建應(yīng)用的核心功能同時(shí)聯(lián)動(dòng)各類外部系統(tǒng)——比如將AI模型與企業(yè)現(xiàn)有的客戶管理系統(tǒng)、數(shù)據(jù)存儲(chǔ)系統(tǒng)打通確保數(shù)據(jù)流轉(zhuǎn)順暢。在這一過程中他們還需要配合設(shè)計(jì)團(tuán)隊(duì)打磨前端交互界面讓技術(shù)功能以簡(jiǎn)潔易懂的方式呈現(xiàn)給用戶實(shí)現(xiàn)從技術(shù)方案到產(chǎn)品形態(tài)的轉(zhuǎn)化。測(cè)試與優(yōu)化是保障產(chǎn)品質(zhì)量的關(guān)鍵步驟。工程師會(huì)開展全面的功能測(cè)試找出并修復(fù)開發(fā)過程中出現(xiàn)的漏洞同時(shí)針對(duì)模型的響應(yīng)速度、穩(wěn)定性等性能指標(biāo)進(jìn)行優(yōu)化。安全合規(guī)性也是測(cè)試的重點(diǎn)需要確保應(yīng)用符合數(shù)據(jù)保護(hù)、隱私安全等相關(guān)規(guī)定。此外他們還會(huì)收集用戶反饋通過調(diào)整模型參數(shù)、優(yōu)化提示詞等方式持續(xù)提升產(chǎn)品體驗(yàn)讓應(yīng)用更貼合用戶實(shí)際使用需求。部署運(yùn)維與迭代則貫穿產(chǎn)品的整個(gè)生命周期。工程師會(huì)通過云服務(wù)器或私有服務(wù)器將應(yīng)用部署上線并實(shí)時(shí)監(jiān)控運(yùn)行狀態(tài)及時(shí)處理突發(fā)故障確保應(yīng)用穩(wěn)定運(yùn)行。隨著業(yè)務(wù)需求的變化他們還需要對(duì)應(yīng)用功能進(jìn)行迭代更新同時(shí)編寫完善的開發(fā)文檔和使用手冊(cè)為后續(xù)的維護(hù)和交接提供支持。03薪資情況與職業(yè)價(jià)值市場(chǎng)對(duì)這一職業(yè)的高度認(rèn)可直接體現(xiàn)在薪資待遇上。據(jù)獵聘最新在招崗位數(shù)據(jù)顯示AI大模型應(yīng)用開發(fā)工程師的月薪最高可達(dá)60k。在AI技術(shù)加速落地的當(dāng)下這種“技術(shù)業(yè)務(wù)”的復(fù)合型能力尤為稀缺讓該職業(yè)成為當(dāng)下極具吸引力的就業(yè)選擇。AI大模型應(yīng)用開發(fā)工程師是AI技術(shù)落地的關(guān)鍵橋梁。他們用專業(yè)能力將抽象的技術(shù)轉(zhuǎn)化為具體的產(chǎn)品讓大模型的價(jià)值真正滲透到各行各業(yè)。隨著AI場(chǎng)景化應(yīng)用的不斷深化這一職業(yè)的重要性將更加凸顯也必將吸引更多人才投身其中推動(dòng)AI技術(shù)更好地服務(wù)于社會(huì)發(fā)展。CSDN粉絲獨(dú)家福利給大家整理了一份AI大模型全套學(xué)習(xí)資料這份完整版的 AI 大模型學(xué)習(xí)資料已經(jīng)上傳CSDN朋友們?nèi)绻枰梢話呙柘路蕉S碼點(diǎn)擊下方CSDN官方認(rèn)證鏈接免費(fèi)領(lǐng)取【保證100%免費(fèi)】