你有沒有這種感覺每天在微信、知乎、網頁之間來回跳看到好的文章要么隨手點個贊要么丟進收藏夾吃灰。等過幾天想找回來翻遍十幾個 App腦子一片空白。這篇文章想聊的不是又一個稍后讀工具而是我最近把整套信息收集 自動整理流水線搬到本地之后沉淀下來的一些工程細節(jié)。整個項目的核心思路只有一句話“先存下來再讓它自己整理”。下面我會按動機、架構、關鍵模塊、踩過的坑的順序講清楚。一、先說動機為什么不做 App先做收集箱我以前試過 Notion、Obsidian、各種瀏覽器插件最后都死在兩個地方入口太重要讓隨手收藏這件事足夠輕最好兩步以內搞定看到 → 分享 → 完事。AI 整理是異步的摘要、分類、關聯(lián)主題這些動作不應該阻塞收藏動作本身。所以目標很明確入口必須是飛書消息這種已經在用的工具——我只是把文章鏈接發(fā)到飛書群。收藏那一刻只做一件事把原始數(shù)據(jù)落盤。哪怕 AI 掛了、網絡掛了、內容提取掛了原始消息也不能丟。后處理異步跑失敗可以重試可以降級絕不能反過來影響入口。design.md里那張圖基本就是這個思路的源頭微信/知乎/網頁/公眾號 ↓ 分享 飛書群機器人 ↓ 拉取消息 落盤原始 Markdown ↓ 后處理 提取正文 / 多模態(tài)理解 / 關聯(lián) ↓ ~/my-kb 本地知識庫聽起來沒啥新意確實。但真正落地的時候里面藏了一堆工程細節(jié)比如飛書消息分頁、自機器人消息過濾、增量游標這些寫著寫著就發(fā)現(xiàn)原型和能穩(wěn)定跑的版本完全是兩回事。二、整體架構四個文件一千多行代碼代碼不長總共 4 個 Python 文件剛好對應 4 個職責文件職責main.py監(jiān)控入口拉消息、去重、調度后處理feishu_bot.py飛書 API 封裝拿 token、拉消息、解析消息體message_post_processor.py后處理文字 / 圖片 / 鏈接 → Markdown 落盤image_content_extractor.py多模態(tài)調用視覺大模型提取圖片語義啟動方式就一行python main.py然后它就一直跑著5 秒拉一次。入口main.py的循環(huán)結構whileTrue:messagesfetch_new_messages(bot,chat_id,last_timestamp,PAGE_SIZE)formessageinmessages:ifmessage_idinseen_message_ids:continueseen_message_ids.add(message_id)ifis_current_bot_message(message,bot):continue# 自己發(fā)的消息直接跳過saved_pathsprocessor.process_message(message)last_timestampmax(int(m.get(create_time,0))forminmessages)time.sleep(CHECK_INTERVAL_SECONDS)邏輯很簡單但有幾個坑值得單獨拎出來說。三、踩過的坑 1飛書消息必須翻頁到底飛書的GET /im/v1/messages接口按時間從舊到新返回但每頁最多 20 條。如果你只調一次拿第一頁看起來好像收到了其實群聊積壓超過 20 條時最新那幾條會被直接漏掉等下次再拉的時候又因為時間窗口錯位整段卡死。解決方案很直接循環(huán)翻頁直到has_moreFalsedeffetch_new_messages(bot,chat_id,last_timestamp,page_size):all_messages[]page_tokenNonewhileTrue:resultbot.get_chat_messages(chat_id,page_sizepage_size,page_tokenpage_token,start_timelast_timestamp//10001,end_timeint(time.time()),)itemsresult.get(items,[])ifnotitems:breakall_messages.extend(items)ifnotresult.get(has_more):breakpage_tokenresult.get(page_token)returnall_messages另外注意start_time是秒級時間戳且包含邊界——所以要1跳過上一輪已經處理過的最新一條不然會無限循環(huán)處理同一條消息。四、踩過的坑 2增量游標要用毫秒時間戳不是消息 ID我第一版用last_message_id作為游標結果發(fā)現(xiàn)飛書的message_id是不保證全局有序的跨頁之后會亂。改成時間戳之后穩(wěn)了last_timestampmax(int(m.get(create_time,0))forminmessages)orlast_timestamp注意兩個細節(jié)create_time是毫秒級字符串轉成秒要/1000。用or last_timestamp是為了防御如果本批消息為空max(...)會返回0會回退到上次的游標而不是把窗口推到 1970 年。五、踩過的坑 3自機器人消息過濾群聊里如果還有別的機器人監(jiān)控就會把它們的回復也當成收藏處理掉——這是真實踩過的坑當時日志里一堆機器人歡迎語刷屏。飛書消息的sender.sender_type字段會標記user/app/bot而sender.sender_id里帶著具體 IDdefis_current_bot_message(message,bot):sendermessage.get(sender,{})or{}ifsender.get(sender_type)notin{app,bot}:returnFalsesender_idsget_sender_ids(message)returnbot.app_idinsender_ids.values()但飛書 API 改版之后sender的結構也變了——新版是sender.id sender.id_type舊版是sender.sender_id.open_id / union_id / user_id。為了不踩坑做了一層兼容defget_sender_ids(message):sendermessage.get(sender,{})or{}sender_idsender.get(sender_id)or{}ifsender_id:return{k:vfork,vinsender_id.items()ifv}ifsender.get(id):return{sender.get(id_type,id):sender[id]}return{}這是飛書生態(tài)里很常見的問題API 升級時新老結構并存半年寫客戶端必須同時支持否則一更新就全量報錯。六、消息后處理三種類型走三條路徑MessagePostProcessor是整個流水線的核心。它根據(jù)msg_type分流defprocess_message(self,message):saved_paths[]msg_typemessage.get(msg_type,)ifmsg_typein{text,post}:# 文字 / 富文本saved_paths.append(self.save_text_message(message))elifmsg_typeimage:# 圖片saved_paths.extend(self.save_image_message(message))forurlinself.extract_urls_from_message(message):saved_paths.append(self.save_url_content(url,message))returnsaved_paths注意最后那段——即使消息是文字只要里面帶了鏈接也會被再走一次 URL 抓取。也就是說一條帶鏈接的文字消息會同時產出兩個文件一個是文字原文一個是網頁正文。這是特意設計的文字消息負責上下文網頁抓取負責完整內容。6.1 文字消息直接落 Markdown文字消息最簡單直接拼 frontmatter 內容markdown(render_frontmatter({type:feishu_text,message_id:message_id,msg_type:msg_type,created_at:created_at,sender:format_sender(message),})# 飛書文字消息\n\ncontent)文件命名用feishu-text-20260829-153022-a1b2c3d4e5.md時間 短哈希確保不會撞名。6.2 圖片消息下載 多模態(tài)理解圖片路徑稍微復雜一點從body.content.image_key拿到飛書的圖片 key。調bot.download_message_resource()拉二進制。用 Content-Type 推斷擴展名寫到本地。同時把圖片 base64 塞給多模態(tài)大模型讓它描述圖片內容OCR、看圖說話。最終落兩個文件圖片本體 Markdown 筆記里面嵌入圖片 提取的語義文本。content,content_typeself.bot.download_message_resource(message_id,image_key)image_path.write_bytes(content)ifself.image_extractor:responseself.image_extractor.extract_content_from_image(content)extractedself.image_extractor.extract_text_from_response(response)note_content(render_frontmatter({...})# 飛書圖片消息\n\nf\n(f\n## 提取的內容\n\n{extracted}\nifextractedelse))這一步是整個流水線里唯一會調外部 LLM 的地方。而且是同步的——為了保證圖片筆記里一定有內容會等模型返回。七、網頁正文提取雙策略兜底URL 抓取是整個系統(tǒng)里最容易翻車的環(huán)節(jié)。原因大家都知道現(xiàn)代網站大量使用 JS 渲染、懶加載、反爬、登錄墻。requests BeautifulSoup這一套在十年前夠用今天 50% 的網站抓不到正文。我做了雙策略自動降級deffetch_web_article(url):try:articlefetch_with_requests(url)ifis_valid_article(article):returnarticleexceptExceptionasexc:request_errorstr(exc)try:articlefetch_with_opencli(url)ifis_valid_article(article):returnarticleexceptExceptionasexc:...第一策略requests BeautifulSoup?？焖佟⑤p量80% 的博客和文檔站能搞定。第二策略opencli browser。一個 Node 寫的瀏覽器自動化 CLI能跑完整 JS 渲染。代價是慢每頁要 60 秒超時且吃內存。判定標準清洗后的純文本超過 120 字符就算成功。太短基本是 JS 沒渲染完。整個過程我特地做了三件事過濾無關標簽腳本、樣式、導航、頁腳、側欄全decompose()掉。HTML 轉 Markdown 用白名單只處理h1-h4 / p / li / blockquote / pre / table其它標簽丟棄——避免抓到一堆亂七八糟的 div。正文提取的優(yōu)先級article→main→body按語義主干層層回退。rootsoup.find(article)orsoup.find(main)orsoup.bodyorsoup八、去重機制寧可重復抓不能漏消息消息 ID 的去重放在了最前面ifnotmessage_idormessage_idinseen_message_ids:continueseen_message_ids.add(message_id)seen_message_ids這個集合有三個數(shù)據(jù)來源當前進程的內存跑得越久越大。磁盤 JSON~/.feishu-monitor-seen.json最多保留 5000 條滑動窗口扔掉最老的。已有 Markdown 文件的開頭用正則從每個.md的 frontmatter 里把message_id摳出來重啟程序自動恢復。第三個特別關鍵——意味著哪怕狀態(tài)文件丟了重啟時也能從知識庫反向重建已處理集合永遠不會重復處理同一條消息。九、消息內容解析富文本和純文本要走兩條路飛書的富文本消息msg_type postbody 是個嵌套 JSON長這樣{zh_cn:{content:[[{tag:text,text:這是第一段},{tag:a,text:鏈接,href:https://...}],[{tag:text,text:這是第二段}]]}}要把里面的純文本按順序提出來遞歸遍歷所有節(jié)點把tag text的部分拼接起來forparagraphincontent_obj[content]:ifisinstance(paragraph,list):forelementinparagraph:ifisinstance(element,dict)andelement.get(tag)text:text_parts.append(element.get(text,))return .join(text_parts)URL 提取也是同樣的道理——不只從可讀文本里抓還要遞歸遍歷整個 body JSON 結構用flatten_strings把所有字符串攤平避免漏掉嵌在富文本a標簽里的鏈接。十、多模態(tài)圖片理解只調一次絕不阻塞入口ImageContentExtractor的設計有個核心原則調用失敗不能影響主流程。defsave_image_message(self,message):try:content,content_typeself.bot.download_message_resource(...)exceptExceptionasexc:return[self.save_error_note(message,f圖片下載失敗:{exc})]...try:extractedself.image_extractor.extract_content_from_image(content)exceptExceptionase:logging.error(...)extracted圖片筆記永遠會生成。多模態(tài)提取只是個增強字段——拿到內容就塞進去沒拿到就空著下一輪可以人工補。請求體用的是 OpenAI 多模態(tài)兼容格式image_url.data:...base64...所以換任何一家支持視覺的模型都能直接對接。模型 prompt 我留成了環(huán)境變量PROMPT方便后期接不同的視覺任務比如 OCR 專用、表格專用、截圖專用。十一、給未來的自己還要補什么寫完這套之后我又回看了design.md發(fā)現(xiàn)當初規(guī)劃的三階段只落地了 1.5? 第一階段分享 → 落盤 → 摘要骨架? 第二階段圖片多模態(tài)、網頁兜底提取、去重? 第三階段還沒做飛書多維表格雙向同步目前只在本地向量檢索 RAG 問答自動生成周報 / 月度綜述與已有項目、論文的關聯(lián)最想做的其實是關聯(lián)性判斷——收藏一篇文章時自動和已有筆記做語義匹配告訴我這篇和你之前收藏的 X 是同一個話題建議合并。這才是真正的個人知識庫而不只是一個更大的收藏夾。但這一步要等數(shù)據(jù)量上去才有意義。先讓收集箱跑半年攢夠 1000 篇再說。十二、一些工程上的小取舍最后講幾個寫代碼過程中的判斷也許對你做類似項目有參考入口極簡后臺豐富。監(jiān)控主循環(huán)只做拉消息 → 落盤所有重活都在process_message里。這樣主循環(huán)穩(wěn)定、不容易因某個異常崩潰。失敗即文件。任何處理失敗都生成一個.md錯誤筆記記錄原始消息 ID 和錯誤原因。這樣日志 文件 狀態(tài)文件三件套任何一個丟失都能從其它兩個恢復。絕不引入強依賴。整套系統(tǒng)只用requests beautifulsoup4 python-dotenv連數(shù)據(jù)庫都沒有。~/my-kb就是一個文件夾所有內容都是普通 Markdown可以直接用 VS Code、Obsidian 打開。配置驅動。config.env、config_mimo.env分別管飛書憑證和多模態(tài) API 憑證更換服務只改環(huán)境變量不改代碼。說到底做這個項目的初衷不是AI 改變生活而是承認自己會忘、承認 AI 會掛、承認網絡會斷。在這個基礎上能跑起來的最小系統(tǒng)比永遠畫不完的架構圖更有價值。