你有没有这种感觉每天在微信、知乎、网页之间来回跳看到好的文章要么随手点个赞要么丢进收藏夹吃灰。等过几天想找回来翻遍十几个 App脑子一片空白。这篇文章想聊的不是又一个稍后读工具而是我最近把整套信息收集 自动整理流水线搬到本地之后沉淀下来的一些工程细节。整个项目的核心思路只有一句话“先存下来再让它自己整理”。下面我会按动机、架构、关键模块、踩过的坑的顺序讲清楚。一、先说动机为什么不做 App先做收集箱我以前试过 Notion、Obsidian、各种浏览器插件最后都死在两个地方入口太重要让随手收藏这件事足够轻最好两步以内搞定看到 → 分享 → 完事。AI 整理是异步的摘要、分类、关联主题这些动作不应该阻塞收藏动作本身。所以目标很明确入口必须是飞书消息这种已经在用的工具——我只是把文章链接发到飞书群。收藏那一刻只做一件事把原始数据落盘。哪怕 AI 挂了、网络挂了、内容提取挂了原始消息也不能丢。后处理异步跑失败可以重试可以降级绝不能反过来影响入口。design.md里那张图基本就是这个思路的源头微信/知乎/网页/公众号 ↓ 分享 飞书群机器人 ↓ 拉取消息 落盘原始 Markdown ↓ 后处理 提取正文 / 多模态理解 / 关联 ↓ ~/my-kb 本地知识库听起来没啥新意确实。但真正落地的时候里面藏了一堆工程细节比如飞书消息分页、自机器人消息过滤、增量游标这些写着写着就发现原型和能稳定跑的版本完全是两回事。二、整体架构四个文件一千多行代码代码不长总共 4 个 Python 文件刚好对应 4 个职责文件职责main.py监控入口拉消息、去重、调度后处理feishu_bot.py飞书 API 封装拿 token、拉消息、解析消息体message_post_processor.py后处理文字 / 图片 / 链接 → Markdown 落盘image_content_extractor.py多模态调用视觉大模型提取图片语义启动方式就一行python main.py然后它就一直跑着5 秒拉一次。入口main.py的循环结构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# 自己发的消息直接跳过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 条时最新那几条会被直接漏掉等下次再拉的时候又因为时间窗口错位整段卡死。解决方案很直接循环翻页直到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跳过上一轮已经处理过的最新一条不然会无限循环处理同一条消息。四、踩过的坑 2增量游标要用毫秒时间戳不是消息 ID我第一版用last_message_id作为游标结果发现飞书的message_id是不保证全局有序的跨页之后会乱。改成时间戳之后稳了last_timestampmax(int(m.get(create_time,0))forminmessages)orlast_timestamp注意两个细节create_time是毫秒级字符串转成秒要/1000。用or last_timestamp是为了防御如果本批消息为空max(...)会返回0会回退到上次的游标而不是把窗口推到 1970 年。五、踩过的坑 3自机器人消息过滤群聊里如果还有别的机器人监控就会把它们的回复也当成收藏处理掉——这是真实踩过的坑当时日志里一堆机器人欢迎语刷屏。飞书消息的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{}这是飞书生态里很常见的问题API 升级时新老结构并存半年写客户端必须同时支持否则一更新就全量报错。六、消息后处理三种类型走三条路径MessagePostProcessor是整个流水线的核心。它根据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 图片消息下载 多模态理解图片路径稍微复杂一点从body.content.image_key拿到飞书的图片 key。调bot.download_message_resource()拉二进制。用 Content-Type 推断扩展名写到本地。同时把图片 base64 塞给多模态大模型让它描述图片内容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 抓取是整个系统里最容易翻车的环节。原因大家都知道现代网站大量使用 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。快速、轻量80% 的博客和文档站能搞定。第二策略opencli browser。一个 Node 写的浏览器自动化 CLI能跑完整 JS 渲染。代价是慢每页要 60 秒超时且吃内存。判定标准清洗后的纯文本超过 120 字符就算成功。太短基本是 JS 没渲染完。整个过程我特地做了三件事过滤无关标签脚本、样式、导航、页脚、侧栏全decompose()掉。HTML 转 Markdown 用白名单只处理h1-h4 / p / li / blockquote / pre / table其它标签丢弃——避免抓到一堆乱七八糟的 div。正文提取的优先级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这个集合有三个数据来源当前进程的内存跑得越久越大。磁盘 JSON~/.feishu-monitor-seen.json最多保留 5000 条滑动窗口扔掉最老的。已有 Markdown 文件的开头用正则从每个.md的 frontmatter 里把message_id抠出来重启程序自动恢复。第三个特别关键——意味着哪怕状态文件丢了重启时也能从知识库反向重建已处理集合永远不会重复处理同一条消息。九、消息内容解析富文本和纯文本要走两条路飞书的富文本消息msg_type postbody 是个嵌套 JSON长这样{zh_cn:{content:[[{tag:text,text:这是第一段},{tag:a,text:链接,href:https://...}],[{tag:text,text:这是第二段}]]}}要把里面的纯文本按顺序提出来递归遍历所有节点把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标签里的链接。十、多模态图片理解只调一次绝不阻塞入口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图片笔记永远会生成。多模态提取只是个增强字段——拿到内容就塞进去没拿到就空着下一轮可以人工补。请求体用的是 OpenAI 多模态兼容格式image_url.data:...base64...所以换任何一家支持视觉的模型都能直接对接。模型 prompt 我留成了环境变量PROMPT方便后期接不同的视觉任务比如 OCR 专用、表格专用、截图专用。十一、给未来的自己还要补什么写完这套之后我又回看了design.md发现当初规划的三阶段只落地了 1.5✅ 第一阶段分享 → 落盘 → 摘要骨架✅ 第二阶段图片多模态、网页兜底提取、去重⏳ 第三阶段还没做飞书多维表格双向同步目前只在本地向量检索 RAG 问答自动生成周报 / 月度综述与已有项目、论文的关联最想做的其实是关联性判断——收藏一篇文章时自动和已有笔记做语义匹配告诉我这篇和你之前收藏的 X 是同一个话题建议合并。这才是真正的个人知识库而不只是一个更大的收藏夹。但这一步要等数据量上去才有意义。先让收集箱跑半年攒够 1000 篇再说。十二、一些工程上的小取舍最后讲几个写代码过程中的判断也许对你做类似项目有参考入口极简后台丰富。监控主循环只做拉消息 → 落盘所有重活都在process_message里。这样主循环稳定、不容易因某个异常崩溃。失败即文件。任何处理失败都生成一个.md错误笔记记录原始消息 ID 和错误原因。这样日志 文件 状态文件三件套任何一个丢失都能从其它两个恢复。绝不引入强依赖。整套系统只用requests beautifulsoup4 python-dotenv连数据库都没有。~/my-kb就是一个文件夹所有内容都是普通 Markdown可以直接用 VS Code、Obsidian 打开。配置驱动。config.env、config_mimo.env分别管飞书凭证和多模态 API 凭证更换服务只改环境变量不改代码。说到底做这个项目的初衷不是AI 改变生活而是承认自己会忘、承认 AI 会挂、承认网络会断。在这个基础上能跑起来的最小系统比永远画不完的架构图更有价值。
我用飞书机器人搭了一个“个人知识收集箱“，顺便把收藏自动化了
你有没有这种感觉每天在微信、知乎、网页之间来回跳看到好的文章要么随手点个赞要么丢进收藏夹吃灰。等过几天想找回来翻遍十几个 App脑子一片空白。这篇文章想聊的不是又一个稍后读工具而是我最近把整套信息收集 自动整理流水线搬到本地之后沉淀下来的一些工程细节。整个项目的核心思路只有一句话“先存下来再让它自己整理”。下面我会按动机、架构、关键模块、踩过的坑的顺序讲清楚。一、先说动机为什么不做 App先做收集箱我以前试过 Notion、Obsidian、各种浏览器插件最后都死在两个地方入口太重要让随手收藏这件事足够轻最好两步以内搞定看到 → 分享 → 完事。AI 整理是异步的摘要、分类、关联主题这些动作不应该阻塞收藏动作本身。所以目标很明确入口必须是飞书消息这种已经在用的工具——我只是把文章链接发到飞书群。收藏那一刻只做一件事把原始数据落盘。哪怕 AI 挂了、网络挂了、内容提取挂了原始消息也不能丢。后处理异步跑失败可以重试可以降级绝不能反过来影响入口。design.md里那张图基本就是这个思路的源头微信/知乎/网页/公众号 ↓ 分享 飞书群机器人 ↓ 拉取消息 落盘原始 Markdown ↓ 后处理 提取正文 / 多模态理解 / 关联 ↓ ~/my-kb 本地知识库听起来没啥新意确实。但真正落地的时候里面藏了一堆工程细节比如飞书消息分页、自机器人消息过滤、增量游标这些写着写着就发现原型和能稳定跑的版本完全是两回事。二、整体架构四个文件一千多行代码代码不长总共 4 个 Python 文件刚好对应 4 个职责文件职责main.py监控入口拉消息、去重、调度后处理feishu_bot.py飞书 API 封装拿 token、拉消息、解析消息体message_post_processor.py后处理文字 / 图片 / 链接 → Markdown 落盘image_content_extractor.py多模态调用视觉大模型提取图片语义启动方式就一行python main.py然后它就一直跑着5 秒拉一次。入口main.py的循环结构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# 自己发的消息直接跳过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 条时最新那几条会被直接漏掉等下次再拉的时候又因为时间窗口错位整段卡死。解决方案很直接循环翻页直到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跳过上一轮已经处理过的最新一条不然会无限循环处理同一条消息。四、踩过的坑 2增量游标要用毫秒时间戳不是消息 ID我第一版用last_message_id作为游标结果发现飞书的message_id是不保证全局有序的跨页之后会乱。改成时间戳之后稳了last_timestampmax(int(m.get(create_time,0))forminmessages)orlast_timestamp注意两个细节create_time是毫秒级字符串转成秒要/1000。用or last_timestamp是为了防御如果本批消息为空max(...)会返回0会回退到上次的游标而不是把窗口推到 1970 年。五、踩过的坑 3自机器人消息过滤群聊里如果还有别的机器人监控就会把它们的回复也当成收藏处理掉——这是真实踩过的坑当时日志里一堆机器人欢迎语刷屏。飞书消息的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{}这是飞书生态里很常见的问题API 升级时新老结构并存半年写客户端必须同时支持否则一更新就全量报错。六、消息后处理三种类型走三条路径MessagePostProcessor是整个流水线的核心。它根据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 图片消息下载 多模态理解图片路径稍微复杂一点从body.content.image_key拿到飞书的图片 key。调bot.download_message_resource()拉二进制。用 Content-Type 推断扩展名写到本地。同时把图片 base64 塞给多模态大模型让它描述图片内容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 抓取是整个系统里最容易翻车的环节。原因大家都知道现代网站大量使用 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。快速、轻量80% 的博客和文档站能搞定。第二策略opencli browser。一个 Node 写的浏览器自动化 CLI能跑完整 JS 渲染。代价是慢每页要 60 秒超时且吃内存。判定标准清洗后的纯文本超过 120 字符就算成功。太短基本是 JS 没渲染完。整个过程我特地做了三件事过滤无关标签脚本、样式、导航、页脚、侧栏全decompose()掉。HTML 转 Markdown 用白名单只处理h1-h4 / p / li / blockquote / pre / table其它标签丢弃——避免抓到一堆乱七八糟的 div。正文提取的优先级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这个集合有三个数据来源当前进程的内存跑得越久越大。磁盘 JSON~/.feishu-monitor-seen.json最多保留 5000 条滑动窗口扔掉最老的。已有 Markdown 文件的开头用正则从每个.md的 frontmatter 里把message_id抠出来重启程序自动恢复。第三个特别关键——意味着哪怕状态文件丢了重启时也能从知识库反向重建已处理集合永远不会重复处理同一条消息。九、消息内容解析富文本和纯文本要走两条路飞书的富文本消息msg_type postbody 是个嵌套 JSON长这样{zh_cn:{content:[[{tag:text,text:这是第一段},{tag:a,text:链接,href:https://...}],[{tag:text,text:这是第二段}]]}}要把里面的纯文本按顺序提出来递归遍历所有节点把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标签里的链接。十、多模态图片理解只调一次绝不阻塞入口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图片笔记永远会生成。多模态提取只是个增强字段——拿到内容就塞进去没拿到就空着下一轮可以人工补。请求体用的是 OpenAI 多模态兼容格式image_url.data:...base64...所以换任何一家支持视觉的模型都能直接对接。模型 prompt 我留成了环境变量PROMPT方便后期接不同的视觉任务比如 OCR 专用、表格专用、截图专用。十一、给未来的自己还要补什么写完这套之后我又回看了design.md发现当初规划的三阶段只落地了 1.5✅ 第一阶段分享 → 落盘 → 摘要骨架✅ 第二阶段图片多模态、网页兜底提取、去重⏳ 第三阶段还没做飞书多维表格双向同步目前只在本地向量检索 RAG 问答自动生成周报 / 月度综述与已有项目、论文的关联最想做的其实是关联性判断——收藏一篇文章时自动和已有笔记做语义匹配告诉我这篇和你之前收藏的 X 是同一个话题建议合并。这才是真正的个人知识库而不只是一个更大的收藏夹。但这一步要等数据量上去才有意义。先让收集箱跑半年攒够 1000 篇再说。十二、一些工程上的小取舍最后讲几个写代码过程中的判断也许对你做类似项目有参考入口极简后台丰富。监控主循环只做拉消息 → 落盘所有重活都在process_message里。这样主循环稳定、不容易因某个异常崩溃。失败即文件。任何处理失败都生成一个.md错误笔记记录原始消息 ID 和错误原因。这样日志 文件 状态文件三件套任何一个丢失都能从其它两个恢复。绝不引入强依赖。整套系统只用requests beautifulsoup4 python-dotenv连数据库都没有。~/my-kb就是一个文件夹所有内容都是普通 Markdown可以直接用 VS Code、Obsidian 打开。配置驱动。config.env、config_mimo.env分别管飞书凭证和多模态 API 凭证更换服务只改环境变量不改代码。说到底做这个项目的初衷不是AI 改变生活而是承认自己会忘、承认 AI 会挂、承认网络会断。在这个基础上能跑起来的最小系统比永远画不完的架构图更有价值。
)
)
)
