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