From cecf4d3ae7398cf95c754afcd4dc846cb2dc0bbd Mon Sep 17 00:00:00 2001 From: zhuyongxin Date: Wed, 25 Mar 2026 10:25:26 +0800 Subject: [PATCH] Add markdown sink for filtered knowledge notes --- README.md | 14 +- TODO.md | 13 +- docs/README.md | 10 +- docs/context-reset-brief.md | 27 +- docs/markdown-sink-design.md | 105 +++++++ ...03-25-信息过载时代-我的漏斗式阅读工作流.md | 282 ++++++++++++++++++ scripts/run_markdown_sink.py | 61 ++++ src/summary_mcp/models/sink.py | 35 +++ src/summary_mcp/sinks/__init__.py | 3 + src/summary_mcp/sinks/markdown.py | 127 ++++++++ 10 files changed, 656 insertions(+), 21 deletions(-) create mode 100644 docs/markdown-sink-design.md create mode 100644 knowledge-base/inbox/unknown-date/2026-03-25-信息过载时代-我的漏斗式阅读工作流.md create mode 100644 scripts/run_markdown_sink.py create mode 100644 src/summary_mcp/models/sink.py create mode 100644 src/summary_mcp/sinks/__init__.py create mode 100644 src/summary_mcp/sinks/markdown.py diff --git a/README.md b/README.md index a129d6e..4890853 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # Content Extract MCP -Python MCP scaffold for article content extraction, structured summary validation, and deterministic filtering. +Python MCP scaffold for article content extraction, structured summary validation, deterministic filtering, and Markdown sink output. ## Run @@ -77,3 +77,15 @@ python scripts/run_filter_rules.py ^ --context outputs/filter-context.json ^ --output outputs/filter-decision.with-context.json ``` + +Write a filtered result into the Markdown sink: + +```bash +python scripts/run_markdown_sink.py ^ + --summary outputs/result.json ^ + --extracted outputs/read-flow-2026.extracted.json ^ + --filter outputs/filter-decision.json +``` + +The script writes markdown notes under `knowledge-base/`. + diff --git a/TODO.md b/TODO.md index 3f40640..3e4d232 100644 --- a/TODO.md +++ b/TODO.md @@ -2,7 +2,7 @@ ## 当前状态 -项目当前处于 `Content Extract MCP` 的 MVP 完成并进入规则过滤阶段。 +项目当前处于 `Content Extract MCP` 的 MVP 完成，已进入规则过滤与 Markdown sink 阶段。 已完成： @@ -23,6 +23,8 @@ - [x] 跑通 `FreshRSS -> item -> content extraction` 单条链路 - [x] 定义规则过滤层 schema - [x] 实现第一版规则引擎与过滤 MCP tool +- [x] 定义 sink 输入输出 schema +- [x] 实现第一版 Markdown sink 与本地写入脚本 --- @@ -42,7 +44,7 @@ - [x] 补充 `item` 的实际样例文件 - [x] 跑通 `FreshRSS -> item -> content extraction` 单条链路 - [x] 定义规则过滤层的输入输出 schema -- [ ] 设计知识库入库格式 +- [x] 设计知识库入库格式 - [ ] 设计 webhook / 推送格式 - [ ] 给提取结果增加更多正文清洗策略，比如尾部噪音清理 - [ ] 细化过滤规则并引入更多个性化上下文 @@ -57,6 +59,7 @@ - [ ] 增加提取缓存、重试和更细粒度日志 - [ ] 重命名包和项目名，从 `summary_mcp` 调整为更符合当前职责的名称 - [ ] 与 OpenClaw 做更正式的工作流编排整合 +- [ ] 扩展 Notion / Webhook / 其他 sink --- @@ -64,8 +67,8 @@ 优先做这两件事： -1. 设计知识库入库格式 -2. 设计 webhook / 推送格式 +1. 设计 webhook / 推送格式 +2. 细化过滤规则并引入更多个性化上下文 --- @@ -73,5 +76,5 @@ - docs/context-reset-brief.md - docs/filter-rule-engine-design.md -- docs/content-extract-mcp-mvp-archive.md +- docs/markdown-sink-design.md - TODO.md diff --git a/docs/README.md b/docs/README.md index bfff397..128e36f 100644 --- a/docs/README.md +++ b/docs/README.md @@ -8,11 +8,13 @@ - 当前 MCP 服务的职责、接口和边界 3. `filter-rule-engine-design.md` - 过滤层的输入输出、规则结构与当前实现 -4. `source-schema-design.md` +4. `markdown-sink-design.md` + - 第一版 Markdown sink 的输入输出、目录结构与落地方式 +5. `source-schema-design.md` - `source -> item -> document` 的对象设计 -5. `reading-pipeline-design-notes.md` +6. `reading-pipeline-design-notes.md` - 更上层的阅读流方案与阶段划分 -6. `summary-loop-explained.md` +7. `summary-loop-explained.md` - 当前 LLM 摘要校验闭环的解释 ## 当前文档分层 @@ -40,6 +42,8 @@ - 提取 JSON -> LLM 摘要 JSON -> 校验 的闭环说明 - `docs/filter-rule-engine-design.md` - 第一版规则过滤引擎设计与落地位置 +- `docs/markdown-sink-design.md` + - 第一版 Markdown sink 设计与落地位置 ### 3. 上下游方案设计 diff --git a/docs/context-reset-brief.md b/docs/context-reset-brief.md index d0fbd6f..9b64556 100644 --- a/docs/context-reset-brief.md +++ b/docs/context-reset-brief.md @@ -24,6 +24,8 @@ - 已跑通 `FreshRSS -> item -> content extraction` 单条链路 - 已定义过滤层输入输出 schema - 已完成第一版规则引擎、本地脚本和 MCP tool +- 已定义 sink 输入输出 schema +- 已完成第一版 Markdown sink 与本地写入脚本 ## 当前关键文件 @@ -39,6 +41,10 @@ - `src/summary_mcp/models/filtering.py` - 过滤引擎： - `src/summary_mcp/filters/engine.py` +- sink 模型： + - `src/summary_mcp/models/sink.py` +- Markdown sink： + - `src/summary_mcp/sinks/markdown.py` - 默认过滤规则： - `configs/filter_rules.json` - 校验 CLI： @@ -51,20 +57,16 @@ - `scripts/run_freshrss_extract.py` - 过滤脚本： - `scripts/run_filter_rules.py` +- Markdown sink 脚本： + - `scripts/run_markdown_sink.py` +- Markdown sink 文档： + - `docs/markdown-sink-design.md` - 当前提示词： - `outputs/llm-summary-prompt.txt` -- FreshRSS 原始响应样例： - - `outputs/freshrss.raw.json` -- FreshRSS item 样例： - - `outputs/freshrss.items.json` -- FreshRSS 提取结果： - - `outputs/freshrss.extracted.json` - 过滤结果样例： - `outputs/filter-decision.json` - 文档索引： - `docs/README.md` -- MVP 归档： - - `docs/content-extract-mcp-mvp-archive.md` - 当前 TODO： - `TODO.md` @@ -79,21 +81,22 @@ - 标准化 `item` 可继续进入内容提取流程 - 规则引擎可对结构化摘要结果输出 `keep / drop / review` 决策 - MCP tool 可承接“由上层 LLM/Agent 调用过滤”的模式 +- Markdown sink 可将过滤后的结果写入本地知识库目录 ## 当前未开始的下一阶段 -- 设计知识库入库格式 - 设计 webhook / 推送格式 - 把 FreshRSS 拉取与提取流程进一步批量化/调度化 - 迭代更细的过滤规则与个性化上下文 +- 决定是否扩展 Notion / Webhook / 其他 sink ## 收束后建议从这里继续 优先从这两个问题继续： -1. 先设计知识库 sink 的输入输出格式 -2. 再决定先接 webhook / 推送还是继续细化过滤规则 +1. 先设计 webhook / 推送格式 +2. 再决定是否扩展其他 sink，而不是继续深化 Markdown sink 本身 ## 一句话结论 -当前 MVP 已完成，FreshRSS 上游和第一版规则过滤层都已接通，下一阶段应转向下游 sink 与推送设计。 +当前 MVP 已完成，FreshRSS 上游、第一版规则过滤层和第一版 Markdown sink 都已接通，下一阶段应转向推送层与更完整的下游集成。 diff --git a/docs/markdown-sink-design.md b/docs/markdown-sink-design.md new file mode 100644 index 0000000..1d150a5 --- /dev/null +++ b/docs/markdown-sink-design.md @@ -0,0 +1,105 @@ +# Markdown Sink 设计 + +## 1. 目标 + +第一版 sink 不直接绑定外部平台，而是先落一个稳定的 `Markdown sink`。 + +原因： + +- 可审计 +- 可版本化 +- 易迁移 +- 可直接兼容 Obsidian / 通用 Markdown 知识库 + +## 2. 统一输入 + +Markdown sink 读取统一的 `SinkInput`： + +- `item` +- `article` +- `summary` +- `filter_decision` +- `metadata` + +这样未来扩展 Notion sink、Webhook sink 时，可以继续复用同一套上游输入结构。 + +## 3. 目录结构 + +当前建议目录结构： + +```text +knowledge-base/ + inbox/ + review/ + archive/ +``` + +具体落盘规则： + +- `keep` -> `knowledge-base/inbox/YYYY/YYYY-MM/` +- `review` -> `knowledge-base/review/YYYY/YYYY-MM/` +- `drop` -> `knowledge-base/archive/YYYY/YYYY-MM/` + +## 4. 文件命名 + +文件名格式： + +```text +YYYY-MM-DD-title-slug.md +``` + +例如： + +```text +2026-03-21-kimi-cursor.md +``` + +## 5. Markdown 内容结构 + +输出采用： + +- YAML frontmatter +- 摘要区 +- 要点区 +- 过滤决策区 +- 原文信息区 +- 正文摘录区 + +frontmatter 中保留： + +- `title` +- `url` +- `source_id` +- `item_id` +- `extract_id` +- `category` +- `worth_keeping` +- `decision` +- `priority` +- `topics` +- `keywords` +- `labels` +- `published_at` +- `saved_at` +- `quality_flags` + +## 6. 当前落地位置 + +- sink 模型： + - `src/summary_mcp/models/sink.py` +- Markdown sink： + - `src/summary_mcp/sinks/markdown.py` +- 本地运行脚本： + - `scripts/run_markdown_sink.py` + +## 7. 当前阶段结论 + +第一版知识库不先绑定 Notion、Lumina 或数据库，而是先让过滤后的内容稳定进入 Markdown 知识库。 + +只要这层格式稳定，后续可以继续扩展： + +- Obsidian Vault +- Logseq +- Notion +- Webhook +- 自定义数据库 sink diff --git a/knowledge-base/inbox/unknown-date/2026-03-25-信息过载时代-我的漏斗式阅读工作流.md b/knowledge-base/inbox/unknown-date/2026-03-25-信息过载时代-我的漏斗式阅读工作流.md new file mode 100644 index 0000000..4397852 --- /dev/null +++ b/knowledge-base/inbox/unknown-date/2026-03-25-信息过载时代-我的漏斗式阅读工作流.md @@ -0,0 +1,282 @@ +--- +title: "信息过载时代，我的漏斗式阅读工作流" +url: "https://shawnxie.top/blogs/tools/read-flow-2026.html" +source_id: "" +item_id: "" +extract_id: "sha256:1ee177246bce46534c81718bee693fea43a4b491f545e61b3c41f22e20994057" +category: "方法论" +worth_keeping: true +decision: "keep" +priority: 80 +topics: + - "个人信息管理" + - "阅读工作流" + - "知识沉淀" +keywords: + - "RSS" + - "FreshRSS" + - "OpenClaw" + - "Digest" + - "Daily Review" + - "Lumina" + - "信息漏斗" + - "人在回路" +labels: + - "durable" + - "needs-review" + - "summary" +published_at: "" +saved_at: "2026-03-25T02:24:05.936615+00:00" +quality_flags: + is_paywalled: false + is_truncated: false + is_low_content: false +--- + +# 摘要 + +作者为解决信息过载问题，构建了一套以RSS为上游、FreshRSS为聚合池、OpenClaw为编排层的漏斗式工作流，通过分层筛选、AI精选和人工精读，将信息逐步沉淀至Lumina知识库，形成稳定可控的个人信息处理闭环。 + +# 要点 + +- 以RSS为主统一信息源，通过转换工具将公众号、社交动态等非RSS内容标准化接入。 +- Digest预处理负责去重、抓取正文和生成摘要，Daily Review结合AI将内容分类为日报栏目。 +- 人工精读是长期价值判断的核心环节，只有经过筛选的内容才进入Lumina知识库沉淀。 +- 通过轻量兴趣画像和反馈机制，系统能逐步优化筛选排序，同时避免形成信息茧房。 +- 沉淀后的内容可进一步生成周刊或主题文章，将信息流转化为长期资产。 + +# 过滤决策 + +- decision: `keep` +- priority: `80` +- matched rules: keep-worth-keeping-method, review-worth-keeping-other + +## reasons + +- Structured summary marked the content as worth keeping in a durable category. +- Worth-keeping signal is positive but no stronger keep rule matched. + +# 原文信息 + +- 标题: 信息过载时代，我的漏斗式阅读工作流 +- 链接: https://shawnxie.top/blogs/tools/read-flow-2026.html +- 作者: unknown +- 发布时间: unknown +- 内容类型: article + +# 正文摘录 + +信息过载时代，我的漏斗式阅读工作流 +这两年我越来越强烈地感觉到，信息问题早就不是获取不到，而是处理不过来。 +真正让我疲惫的，不是没东西看，而是每天都有太多东西值得看：公众号文章、技术博客、GitHub Release、AI 新闻、社区讨论、长文、短讯、碎片化观点等等，全都在争夺我的注意力。 +如果不做点什么，一个人的信息生活很容易退化成这样： +- 收藏夹里躺满了文章，真正读完的却寥寥无几 +- 每天浏览大量内容，能留在脑子里的却屈指可数 +- 输入看似充实，输出却异常薄弱 +- 以为自己一直在吸收信息，实际上只是在不同窗口间疲于切换 +2026年，我开始认真整理自己的一套信息处理工作流。我不需要大而全的平台，也不想要那种号称能“一键替我读完互联网”的 AI 产品。 +我只需要一套围绕自己运转的信息吸收漏斗：既能尽可能广泛地捕捉信息，又能提前过滤掉噪音和重复内容，只把真正值得投入时间的内容呈现在我面前。更重要的是，这套系统还要能将我精读后的高价值内容沉淀下来，反过来优化下一轮的信息筛选。 +信息处理不追求看得更多，而是在于更稳定地吸收、判断和沉淀。 +这篇文章，我将把这套工作流思路完整写下来，包括： +- 为什么我要这样做 +- 它是怎么搭起来的 +- 每一层分别承担什么职责 +- 未来优化方向 +如果你也在被信息过载折磨，或者你已经有不少输入工具，却还是觉得每天看了很多，脑子里没留下什么，那也许这篇文章会有点参考价值。 +为什么要做信息吸收漏斗？ +我以前也尝试过很多方法： +- 订阅很多 RSS，然后每天刷 +- 把值得看的东西先扔进稍后阅读 +- 用收藏、标签、笔记软件把文章存起来 +- 靠搜索和回忆去找以前读过的内容 +这些工具各有用处，但通常只能解决信息处理中的某一个环节。而个人信息处理真正的难点，从来都不在于单点突破，而在于如何将这些分散的环节串联起来。信息系统如果不能同时解决这两个问题，就很容易失控： +输入太多，内容太杂 +几乎每一个你关心的领域，都在持续产生新内容。技术突破、产品迭代、AI进展、工具更新、商业动态、研究成果、行业变化…… +而且在未经过滤之前，重要新闻、工具发布、经验复盘、广告软文、标题党、重复报道等往往都混在一起，让人难以分辨。 +如果每一条内容都需要你从零开始判断其价值，那日积月累下来的认知负担会非常高。很多时候，人真正感到疲惫的不是读一篇长文，而是反复做这些微小但无意义的判断： +- 这篇文章值不值得点开？ +- 这条消息是不是和刚才那条重复？ +- 这个标题是不是在夸大其词？ +- 这条内容到底在说什么？ +- 这篇文章我只是需要了解一下，还是值得保存？ +缺少精读回路 +很多内容看完就过去了，既没有标记，也没有沉淀，更没有真正融入自己的知识体系。 +如果信息系统永远只知道最新内容是什么，却不知道什么内容真正对你有帮助，那最多只能算是一个信息输送管道，而不是一个能越来越懂你的个性化系统。 +所以我要的不是做一个更强的信息池，而是做一个更稳的信息漏斗。 +桶的思路是往里装更多，漏斗的思路是让信息在往下走的过程中不断收窄，最后留下真正值得进入大脑和长期记忆的部分。 +从实践效果来看，这套漏斗式信息处理工作流至少给我带来了几个显著的变化： +- 我不再需要每天从海量标题中艰难地筛选重点内容 +- 低质量和重复的信息大部分在上游就被过滤掉了 +- 真正值得精读的内容，会在更靠后的环节中呈现在我面前 +- 我读过并认为有价值的内容，终于开始能够反过来优化后续的筛选逻辑 +不是追求更快地刷完信息，而是实现更稳定地吸收知识。 +为什么基于 OpenClaw 搭建？ +由于流程尚处探索阶段，选择先用 OpenClaw 将整条链路串联起来，让系统先运行起来，验证可行性，发现问题后再逐步优化和完善。 +OpenClaw 主要扮演编排层的角色，负责以下工作： +- 定时触发各类任务 +- 串联不同的外部工具和系统 +- 与飞书文档进行交互 +- 与 Lumina 知识库进行联动 +- 在需要的环节引入 AI 进行结构化内容生成 +这种方式起步快、迭代快，非常适合边运行边优化，不需要一开始就投入大量精力构建完整的前后端系统。 +如果你也想搭建一套类似的信息处理流程，需要做这些准备： +1. 可持续维护的信息源 +这是整个系统的上游基础，需要先想清楚几个问题： +- 自己真正长期关注的主题领域是什么 +- 这些领域的主要信息来源有哪些 +- 哪些信息源值得长期订阅，哪些只是偶尔查看即可 +这一步不需要追求大而全，但要尽量保持稳定和高质量。 +如果你还没有构建自己的信息源，可以先参考我收集的 RSS 订阅源。 +2. 统一的聚合池 +我选择的聚合池是 FreshRSS，一个开源的RSS阅读器，支持多平台，支持自定义订阅，内容分类管理和API，非常适合用来进行内容存储和同步。 +它并非最终阅读工具，而是信息汇聚的中转站，让所有信息源先汇聚到同一个地方，形成可持续消费的内容候选池。 +3. 自动化编排层 +这是 OpenClaw 最能发挥价值的地方，极大地降低了自动化流程实现的难度，通过自然语言描述诉求，在对话中逐步落地想法。我主要用它来完成以下任务： +- 定时执行 digest（自定义的预处理Skill） +- 定时执行 daily-review（自定义的内容精选Skill） +- 发布飞书文档 +- 调用 Lumina（个人开源的知识库项目） +- 串联各个技能和脚本 +4. 长期知识沉淀层 +在我的工作流中，这一层使用 Lumina——个人开发的信息管理工作台。当然也可以使用别的笔记工具，如Notion、Obsidian等。 +Lumina 只承接我经过筛选后确认值得长期保留的优质内容，这一步直接决定了后续反馈机制的质量。 +一个系统最终能学到什么，很大程度上取决于你为它提供了什么样的正反馈样本。 +信息处理工作流 +一句话描述：用 RSS 尽可能广泛地捕捉信息，用 FreshRSS 进行稳定聚合，用 Digest 完成预处理，用 Daily Review 实现每日精选，通过人工精读判断内容的长期价值，用 Lumina 进行知识沉淀，最后将这些沉淀的价值反向转化为轻量的个性化信号。 +这套流程的核心优势不是追求全自动，而是在于分层处理的设计理念。这意味着： +- 并非所有信息都值得投入时间精读 +- 并非所有信息都值得长期留存 +- 并非所有信息都需要进行个性化加权 +- 并非所有信息都适合直接作为输出内容 +一旦层次划分清晰，系统就不会退化为单一维度的推荐流，而会演变成一个真正的认知加工流程。漏斗的真正价值不在于让信息越来越少，而在于实现了这几个关键目标： +- 上游宽广：确保不会错过真正重要的信息变化 +- 中游稳定：有效过滤噪音，避免其直接干扰注意力系统 +- 下游精准：让我不必在不值得精读的内容上浪费时间 +- 回流轻柔：通过轻量反馈机制，避免系统演变成封闭的信息茧房 +个人信息系统核心能力不是如何接入更多信息源，而是要明确：哪些信息值得进入系统，哪些信息值得投入时间精读，哪些内容值得长期留存，哪些知识最终真正融入了你的思考和行动？ +当能够清晰地回答这些问题时，就不再是互联网信息的被动接收者，你将拥有一套真正属于自己的认知处理系统。接下来将按顺序详细介绍每一层的工作原理。 +信息源：以 RSS 为主，但不局限于原生 RSS +整个系统的最上游是信息源，主要依赖 RSS。 +RSS 是最被低估的个人信息基础设施。其天然符合个人信息系统最核心的几个要求： +- 订阅权完全掌握在自己手中 +- 信息来源清晰明确 +- 更新内容结构化 +- 不受平台推荐算法直接支配 +- 便于程序接入和自动化处理 +然而在现实中，有一个不可避免的问题：并非所有值得关注的信息源都提供 RSS 订阅。例如：部分公众号内容、某些社区的特定栏目、垂直网站的更新页面和社交平台上的账号动态。 +因此，需要在信息源进入流程之前，尽可能将其统一转换为 RSS 或类似 feed 的格式。常见的方案有： +- RSSHub / RSS-Bridge 等转换工具：能够将大量原本不提供 RSS 订阅的内容源，转换为可以被订阅和程序自动化消费的 feed 格式； +- GitHub feed：项目的 Releases、Commits、Discussions 等，都天然提供了结构化的 feed 接口； +- wewe-rss：能够将公众号文章转换成RSS订阅源； +- nitter：将 Twitter/X 动态转换为RSS源； +- 自定义抓取脚本：对于没有现成 RSS 解决方案的页面，也可以自行编写轻量级的抓取脚本，将其转换为内部可用的 feed 格式。 +这一步的原则很简单：上游信息源可以多种多样，但在进入系统之前，格式必须尽可能统一。只有这样，下游的预处理和筛选环节才能稳定可靠地运行。 +更多信息源归一处理方案可参考之前文章碎片时间刷文章！懒人阅读方案分享。 +聚合池：用 FreshRSS 打造稳定的“中间水库” +所有订阅源最终都会汇聚到 FreshRSS 中，它并非"我每天真正坐下来阅读的地方"，而是流程的缓冲层，主要有以下作用： +实现信息来源的统一管理 +无论内容来自博客、社区、GitHub 还是转换后的 feed，最终都以统一的格式呈现，成为可被消费的标准化对象。 +让下游环节无需直接对接互联网 +digest 模块无需再逐个访问各个网站抓取今日更新内容，只需从 FreshRSS 这个统一的内容池中获取未读候选即可。大大降低了系统各环节之间的耦合度。 +确保了信息处理的时间连续性 +信息系统最忌讳的是“今天临时查看一下、明天就忘了、后天又重新开始”的碎片化处理方式。FreshRSS 提供了一个稳定的时间窗口，使后续任务能够按照固定节奏有序运行。 +我的目标不是追求无边界的信息获取，而是实现有边界的信息处理。 +预处理：Digest 将海量候选内容转化为可判断对象 +如果把 FreshRSS 比作蓄水池，那么 Digest 就是这套系统中的第一道加工厂，专注于完成内容预处理任务。 +让人每天真正感到疲惫的，往往不是精读一篇高质量文章，而是反复判断大量低质量内容是否值得投入时间。 +核心处理任务 +- URL 精确去重 +- 相似内容去重 +- 正文抓取 +- 质量检查 +- 噪音过滤 +- 摘要生成 +- 初步排序与文档输出 +在真正的“阅读”开始之前，先将互联网上天然混乱的信息整理成一批更具可读性的候选内容。 +为什么重要？ +如果没有预处理环节，后续的所有精选工作都将建立在一堆未经整理的原始标题之上。而经过 Digest 处理后，情况会改善： +- 标题党内容大幅减少 +- 重复报道得到有效过滤 +- 无法获取正文的无效数据明显减少 +- 每条候选内容至少附带一个可供快速判断的摘要 +这将显著降低后续阅读的认知成本。Digest 的职责并非编辑终稿，而是将候选内容池整理干净、结构化，为后续的精选环节奠定基础。 +AI 精选：Daily Review 为我呈现重点关注内容 +如果说 Digest 的作用是将原始候选内容处理得更具可读性，那么 Daily Review 则是从这些经过预处理的候选中，进一步提炼出当天真正值得关注的精华内容。 +这一步让流程从预处理迈向编辑阶段。不再追求内容的广度，而是致力于打造重点更突出、结构更清晰的阅读体验，让最终产物更像一份真正意义上的日报，而非简单堆砌的摘要集合。 +目前，我将 Daily Review 设计为几个固定栏目，将不同类型的信息分配到不同的认知槽位中，包含： +- 今日大事：聚焦具有公共重要性的事件 +- 变更与实践：关注对个人有直接操作价值的内容 +- 安全与风险：警惕潜在的风险因素 +- 开源与工具：追踪工具生态的发展变化 +- 洞察与数据点：把握行业趋势和关键数据 +- 主题深挖：将单一新闻事件提升到趋势层面进行分析 +LLM 赋能 +从 Digest 输出的候选内容，到 Daily Review 最终的成稿层，中间有很多工作适合 AI 来完成： +- 合并同一事件的多个信息来源 +- 提炼核心主题 +- 为内容分类并分配到对应栏目 +- 识别值得深入探讨的话题 +- 将零散的候选内容重组为人类可以快速阅读的日报结构 +但我对 AI 在这一环节的应用始终保持克制。不是让 AI 全自动生成日报，而是扮演结构化整理者的角色，从候选内容中筛选出当天和我相关的精华部分。 +精读留存：将 Human in the loop 置于系统核心 +前面所有流程，都是为将信息筛选到值得精读的阶段。真正让系统不至于退化为另一种自动化信息流的，正是这一关键环节：Human in the loop（人在回路中）。 +我坚信，在个人知识系统中，最不能完全外包的是长期价值判断能力。 +系统可以协助我完成许多任务，如信息收集、去重、摘要、聚类、排序和精选等。但它无法完全替我做出关键决策： +- 哪些内容真正值得纳入长期知识库 +- 哪些内容在未来会持续发挥价值 +- 哪些内容会对我的写作和判断产生深远影响 +因此，在我的工作流中，Lumina 前面始终设有一道人工筛选门槛。我会从之前各个环节产生的内容中，挑选出真正值得精读和长期留存的内容。 +这是整套系统中最关键的价值确认环节。能够进入 Lumina 的内容，不仅代表我看过，更意味着我认为这篇内容值得在未来持续为我所用。 +个人画像：让系统学会识别对我真正有价值的内容 +如果流程到 Lumina 就戛然而止，那么它仍然只是一个单纯的过滤和沉淀系统。只有当沉淀的内容开始反过来影响上游的信息选择时，整个系统才真正形成了闭环。 +为此我加入了一层设计较为克制的兴趣画像逻辑。 +刻意控制了影响力，不希望整个系统演变成另一个猜你喜欢的推荐引擎。我只需要它能稍微更懂我，但又不过度迎合我的偏好，保留对公共重要性内容的关注和探索未知领域的空间。 +这层画像目前主要承担轻量 rerank 的功能，作为辅助信号，轻微影响 Digest 和 Daily Review 环节中候选内容的排序。主要从以下几个维度逐步学习我的偏好： +- 长期精读的主题领域 +- 能稳定提供价值的信息源 +- 偏好的内容格式 +- 最终存入 Lumina 知识库的内容类型 +轻量引导的设计，旨在减少无效信息对注意力的浪费，同时避免构建封闭的信息茧房。个性化推荐应帮助我们减少无意义的判断，而非让我们躲进舒适区。 +沉淀：将信息流转化为长期内容资产 +信息漏斗的终点不是读完，而是沉淀。目前从两个方向拓展沉淀的价值： +1. 周刊生成 +当系统积累了一周的高质量内容后，就不应再局限于每天生成一份日报。一周的时间跨度非常适合进行复盘总结。此时，系统已经拥有了丰富的素材： +- 一周的 Digest 预处理结果 +- 一周的 Daily Review 精选内容 +- 若干经过价值确认的 Lumina 知识库内容 +- 一些开始反复出现的热门主题 +此时生成周刊，比单纯从网页上抓取热点更有价值。因为这些内容已经经过了个人筛选和沉淀，带有明显的个性化痕迹。周刊不应仅仅是本周发生了什么的简单罗列，更应该具备深度和价值： +- 本周有哪些主题值得重点关注和记忆 +- 哪些变化只是短期噪音，哪些是值得重视的长期信号 +- 哪些内容具有长期参考价值，值得反复回看 +周刊合集👉🏻：肖恩技术周刊 +2. 主题文章生成 +另一个方向是让系统能够逐渐识别哪些主题已经积累了足够的素材，值得写成长篇文章。 +虽然信息流中的内容看起来是离散的，但如果拉长时间维度观察，就会发现很多内容其实都在指向同一个核心主题，例如： +- AI Agent 工程的发展趋势 +- 开源工具链的演变 +- 内容平台分发机制的变革 +- 隐私保护、合规要求与数据治理 +一旦某个主题在一段时间内反复出现，并且我多次对相关内容进行精读、收藏和沉淀，那么它就不应再仅仅是多条零散的新闻，而应该逐渐发展成为一个可以深入挖掘和输出的长期主题。 +内容合集👉🏻：今日观察 +未来迭代方向 +尽管工作流目前已经能够稳定运行，但它远未达到最终完成的状态。可以继续完善的方面有： +细化反馈机制 +目前系统中最强的反馈信号是这篇内容是否被存入 Lumina。但在理想状态下，我希望系统能够逐渐识别更多层次的用户行为： +- 点开内容但未读完 +- 读完内容但未收藏 +- 内容值得精读 +- 内容值得长期沉淀 +- 内容最终影响了写作、决策或实际实现 +- 不喜欢的内容（负反馈） +一旦这些层次的反馈机制得以完善，兴趣画像将变得更加精准，不再只是一个粗粒度的偏好集合。 +进一步抽象流程 +目前，我更倾向于继续在现有的 OpenClaw 体系内迭代优化，这是探索阶段最适合的方式。 +但如果这套流程能够变得更加稳定，职责边界也更加清晰，那么抽象出其中的信息处理内核，会更有利于后续工程化迭代。 +不过比起将其产品化，还是先让这套漏斗系统持续稳定地运转，越来越懂我。 +结语 +信息处理的关键，从来不是看到更多，而是让真正重要的信息被自己接住。 +当信息经过筛选、精读、沉淀和反馈，真正融入知识结构时，我们不再是信息的被动消费者，将真正成为自己信息环境的主人。 +感谢阅读 +微信公众号「肖恩聊技术」 +如果这篇文章对你有帮助，欢迎扫码关注，获取原创文章推送。 diff --git a/scripts/run_markdown_sink.py b/scripts/run_markdown_sink.py new file mode 100644 index 0000000..bd0aa9e --- /dev/null +++ b/scripts/run_markdown_sink.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +import argparse +import json +import sys +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + + +REPO_ROOT = Path(__file__).resolve().parents[1] +SRC_ROOT = REPO_ROOT / "src" + +if str(SRC_ROOT) not in sys.path: + sys.path.insert(0, str(SRC_ROOT)) + +from summary_mcp.models.document import ExtractedArticle +from summary_mcp.models.filtering import FilterDecisionResult +from summary_mcp.models.item import Item +from summary_mcp.models.llm_result import LlmSummaryResult +from summary_mcp.models.sink import SinkInput, SinkMetadata +from summary_mcp.sinks.markdown import MarkdownSink + + +def _load_json(path: Path) -> dict[str, Any]: + return json.loads(path.read_text(encoding="utf-8-sig")) + + +def main() -> None: + parser = argparse.ArgumentParser(description="Write a filtered result into the Markdown sink.") + parser.add_argument("--summary", type=Path, required=True, help="Structured LLM summary JSON file") + parser.add_argument("--extracted", type=Path, required=True, help="Extracted article JSON file") + parser.add_argument("--filter", type=Path, required=True, help="Filter decision JSON file") + parser.add_argument("--item", type=Path, default=None, help="Optional normalized item JSON file") + parser.add_argument( + "--base-dir", + type=Path, + default=REPO_ROOT / "knowledge-base", + help="Markdown sink base directory", + ) + args = parser.parse_args() + + summary = LlmSummaryResult.model_validate(_load_json(args.summary)) + extracted_payload = _load_json(args.extracted) + article = ExtractedArticle.model_validate(extracted_payload.get("article", extracted_payload)) + decision = FilterDecisionResult.model_validate(_load_json(args.filter)) + item = Item.model_validate(_load_json(args.item)) if args.item else None + + sink_input = SinkInput( + item=item, + article=article, + summary=summary, + filter_decision=decision, + metadata=SinkMetadata(generated_at=datetime.now(tz=UTC), source="run_markdown_sink.py"), + ) + result = MarkdownSink(args.base_dir).write(sink_input) + print(json.dumps(result.model_dump(mode="json"), ensure_ascii=False, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/src/summary_mcp/models/sink.py b/src/summary_mcp/models/sink.py new file mode 100644 index 0000000..1f3a8d5 --- /dev/null +++ b/src/summary_mcp/models/sink.py @@ -0,0 +1,35 @@ +from __future__ import annotations + +from datetime import datetime +from typing import Literal + +from pydantic import BaseModel, Field + +from .document import ExtractedArticle +from .filtering import FilterDecisionResult +from .item import Item +from .llm_result import LlmSummaryResult + + +SinkTarget = Literal["markdown"] + + +class SinkMetadata(BaseModel): + generated_at: datetime | None = None + pipeline_version: str = "v1" + source: str | None = None + + +class SinkInput(BaseModel): + item: Item | None = None + article: ExtractedArticle + summary: LlmSummaryResult + filter_decision: FilterDecisionResult + metadata: SinkMetadata = Field(default_factory=SinkMetadata) + + +class SinkResult(BaseModel): + target: SinkTarget + success: bool + output_path: str | None = None + message: str | None = None diff --git a/src/summary_mcp/sinks/__init__.py b/src/summary_mcp/sinks/__init__.py new file mode 100644 index 0000000..4f9226e --- /dev/null +++ b/src/summary_mcp/sinks/__init__.py @@ -0,0 +1,3 @@ +from .markdown import MarkdownSink + +__all__ = ["MarkdownSink"] diff --git a/src/summary_mcp/sinks/markdown.py b/src/summary_mcp/sinks/markdown.py new file mode 100644 index 0000000..7b3c90c --- /dev/null +++ b/src/summary_mcp/sinks/markdown.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +from datetime import UTC, datetime +from pathlib import Path +import re + +from summary_mcp.models.sink import SinkInput, SinkResult + + +_DECISION_DIR = { + "keep": "inbox", + "review": "review", + "drop": "archive", +} + + +def _slugify(text: str) -> str: + slug = text.lower() + slug = re.sub(r"[^a-z0-9\u4e00-\u9fff]+", "-", slug) + slug = re.sub(r"-+", "-", slug).strip("-") + return slug or "untitled" + + +def _yaml_scalar(value: str) -> str: + escaped = value.replace('"', '\\"') + return f'"{escaped}"' + + +def _yaml_list(values: list[str]) -> str: + if not values: + return "[]" + return "\n".join(f" - {_yaml_scalar(value)}" for value in values) + + +def _frontmatter(payload: SinkInput) -> str: + article = payload.article + summary = payload.summary + decision = payload.filter_decision + generated_at = payload.metadata.generated_at or datetime.now(tz=UTC) + published_at = article.published_at.isoformat() if article.published_at else "" + source_id = article.source_id or (payload.item.source_id if payload.item else "") + item_id = article.item_id or (payload.item.item_id if payload.item else "") + + return "\n".join( + [ + "---", + f"title: {_yaml_scalar(summary.title)}", + f"url: {_yaml_scalar(str(summary.url))}", + f"source_id: {_yaml_scalar(source_id)}", + f"item_id: {_yaml_scalar(item_id)}", + f"extract_id: {_yaml_scalar(article.extract_id)}", + f"category: {_yaml_scalar(summary.category)}", + f"worth_keeping: {'true' if summary.worth_keeping else 'false'}", + f"decision: {_yaml_scalar(decision.decision)}", + f"priority: {decision.priority}", + "topics:", + _yaml_list(summary.topics), + "keywords:", + _yaml_list(summary.keywords), + "labels:", + _yaml_list(decision.labels), + f"published_at: {_yaml_scalar(published_at)}", + f"saved_at: {_yaml_scalar(generated_at.isoformat())}", + "quality_flags:", + f" is_paywalled: {'true' if article.quality_flags.is_paywalled else 'false'}", + f" is_truncated: {'true' if article.quality_flags.is_truncated else 'false'}", + f" is_low_content: {'true' if article.quality_flags.is_low_content else 'false'}", + "---", + ] + ) + + +def _render_markdown(payload: SinkInput) -> str: + summary = payload.summary + article = payload.article + decision = payload.filter_decision + highlights = "\n".join(f"- {item}" for item in summary.highlights) + reasons = "\n".join(f"- {item}" for item in decision.reasons) + + return ( + f"{_frontmatter(payload)}\n\n" + f"# 摘要\n\n{summary.summary}\n\n" + f"# 要点\n\n{highlights}\n\n" + f"# 过滤决策\n\n" + f"- decision: `{decision.decision}`\n" + f"- priority: `{decision.priority}`\n" + f"- matched rules: {', '.join(decision.matched_rules) if decision.matched_rules else 'none'}\n\n" + f"## reasons\n\n{reasons}\n\n" + f"# 原文信息\n\n" + f"- 标题: {summary.title}\n" + f"- 链接: {summary.url}\n" + f"- 作者: {article.author or 'unknown'}\n" + f"- 发布时间: {article.published_at.isoformat() if article.published_at else 'unknown'}\n" + f"- 内容类型: {article.content_kind}\n\n" + f"# 正文摘录\n\n{article.plain_text.strip()}\n" + ) + + +class MarkdownSink: + def __init__(self, base_dir: Path) -> None: + self.base_dir = base_dir + + def _target_dir(self, decision: str, published_at: datetime | None) -> Path: + bucket = _DECISION_DIR.get(decision, "archive") + if published_at is None: + return self.base_dir / bucket / "unknown-date" + return self.base_dir / bucket / published_at.strftime("%Y") / published_at.strftime("%Y-%m") + + def _target_name(self, payload: SinkInput) -> str: + article = payload.article + published_at = article.published_at or datetime.now(tz=UTC) + title = payload.summary.title or article.title or "untitled" + return f"{published_at.strftime('%Y-%m-%d')}-{_slugify(title)}.md" + + def write(self, payload: SinkInput) -> SinkResult: + target_dir = self._target_dir(payload.filter_decision.decision, payload.article.published_at) + target_dir.mkdir(parents=True, exist_ok=True) + output_path = target_dir / self._target_name(payload) + output_path.write_text(_render_markdown(payload), encoding="utf-8") + return SinkResult( + target="markdown", + success=True, + output_path=str(output_path), + message="Markdown note written successfully.", + ) + +