feat: make article summary prompt independent from daily digest

- Add outputs/prompts/article-summary-prompt.txt with deep knowledge
  note schema (core_conclusion, main_argument, key_methods, etc.)
- Add ArticleSummaryResult Pydantic model for the new schema
- Add validate_article_summary_payload() validator
- Add optional validator param to run_loop_payload(); defaults to
  existing validate_llm_result_payload so daily digest is unaffected
- Update article_summary.py: new prompt path, new validator, new
  Markdown output with Chinese section headers
This commit is contained in:
wdm
2026-03-29 16:49:05 +08:00
parent 79509fd1ba
commit ed0e3a9c7d
6 changed files with 194 additions and 14 deletions
@@ -0,0 +1,41 @@
你是一个专业的知识沉淀助手。你的任务是对提供的文章正文做深度分析,输出一份结构化的知识沉淀笔记,而不是简短的摘要卡片。
要求:
- 基于文章完整正文(article.plain_text)进行分析
- 所有输出字段使用中文
- 如果文章没有相关内容(如无技术方法、无具体细节),对应数组字段返回空数组 []
- 输出必须是单个合法 JSON 对象,不要加任何解释文字
输出 JSON schema:
{
"title": "文章标题(与原文一致)",
"url": "文章 URL(与原文一致)",
"core_conclusion": "作者最核心的结论,1-2 句,精准概括",
"main_argument": "文章的主要论点或主张,可展开,允许多句",
"key_methods": [
"关键方法、机制或技术手段,每条一句,3-6 条;无相关内容时返回空数组"
],
"important_details": [
"值得记录的细节、数据或案例,每条一句,3-6 条;无相关内容时返回空数组"
],
"reusable_insights": [
"可复用于其他场景的启发或观点,2-4 条"
],
"keywords": [
"具体实体、工具名、方法名,5-8 个"
],
"topics": [
"更高层的主题标签,3-5 个"
],
"category": "内容分类,从以下选项中选择一个:资讯 / 方法论 / 工具实践 / 观点评论",
"worth_keeping": true,
"reason": "沉淀理由,一句话说明为什么值得长期保留"
}
注意:
- keywords 和 topics 不能有重叠
- keywords 侧重具体实体(工具名、框架名、人名、产品名)
- topics 侧重抽象主题(如「知识管理」「系统设计」「AI Agent」)
- core_conclusion 必须是作者的核心观点,不是文章描述
- 只输出 JSON,不要输出任何其他内容
@@ -0,0 +1,40 @@
# 规划:单篇文章总结 Prompt 独立化
## 背景
单篇精读总结(article_summary.py)目前复用日报摘要的同一份 prompt,
导致输出是 2-3 句短摘要卡片,没有体现正文深度。
目标是让单篇精读总结彻底独立,输出知识沉淀笔记。
## 已确认决策
- Validator 方案:方案 B,新增 ArticleSummaryResult 模型 + 专用校验函数
- 输出语言:中文
- 无技术内容时 key_methods / important_details 返回空数组 []
- run_loop_payload 新增可选 validator 参数,None 时保持原有行为,日报链路不受影响
## 涉及文件
- outputs/prompts/article-summary-prompt.txt 新建
- src/summary_mcp/models/article_summary_result.py 新建
- src/summary_mcp/validators/article_summary.py 新建
- src/summary_mcp/core/summary_loop.py 修改
- src/summary_mcp/workflows/article_summary.py 修改
## 新增字段(ArticleSummaryResult)
- core_conclusion:核心结论,1-2 句
- main_argument:主要论点
- key_methods:关键方法,无内容返回 []
- important_details:重要细节,无内容返回 []
- reusable_insights:可复用启发
- keywords / topics / category / worth_keeping / reason
## Markdown 输出格式
核心结论 / 主要论点 / 关键方法 / 重要细节 / 可复用启发 / 关键词 / 主题
## 约束
- freshrss_pipeline.py 不变
- summary_loop.py 只加参数,默认行为不变
+5 -2
View File
@@ -7,7 +7,7 @@ import json
import os import os
import re import re
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any, Callable
import httpx import httpx
@@ -183,7 +183,10 @@ def run_loop_payload(
model: str | None, model: str | None,
api_url: str | None, api_url: str | None,
output_path: Path | None = None, output_path: Path | None = None,
validator: Callable[[dict[str, Any], dict[str, Any] | None], ValidationReport] | None = None,
) -> tuple[int, dict[str, Any] | None, ValidationReport | None]: ) -> tuple[int, dict[str, Any] | None, ValidationReport | None]:
# validator 为 None 时使用日报摘要默认校验器;传入自定义 validator 时使用传入的
resolved_validator = validator if validator is not None else validate_llm_result_payload
prompt_template = load_text(prompt_path) prompt_template = load_text(prompt_path)
if output_path is not None: if output_path is not None:
output_path.parent.mkdir(parents=True, exist_ok=True) output_path.parent.mkdir(parents=True, exist_ok=True)
@@ -223,7 +226,7 @@ def run_loop_payload(
if output_path is not None: if output_path is not None:
save_json(output_path, result_payload) save_json(output_path, result_payload)
report = validate_llm_result_payload(result_payload, extracted_payload) report = resolved_validator(result_payload, extracted_payload)
_save_attempt_artifact( _save_attempt_artifact(
output_path, output_path,
f"attempt-{attempt}.validation.json", f"attempt-{attempt}.validation.json",
@@ -0,0 +1,25 @@
from __future__ import annotations
# 单篇文章深度沉淀的结构化输出模型,与日报摘要的 LlmSummaryResult 完全独立。
from typing import Literal
from pydantic import BaseModel, HttpUrl
ArticleSummaryCategory = Literal["资讯", "方法论", "工具实践", "观点评论"]
class ArticleSummaryResult(BaseModel):
title: str
url: HttpUrl
core_conclusion: str
main_argument: str
key_methods: list[str] = []
important_details: list[str] = []
reusable_insights: list[str] = []
keywords: list[str] = []
topics: list[str] = []
category: ArticleSummaryCategory
worth_keeping: bool
reason: str
@@ -0,0 +1,40 @@
from __future__ import annotations
# 单篇文章深度沉淀的校验器,与日报摘要 validator 完全独立。
# 只校验 ArticleSummaryResult schema,不做日报摘要的业务规则检查。
from typing import Any
from pydantic import ValidationError
from summary_mcp.models.article_summary_result import ArticleSummaryResult
from summary_mcp.validators.llm_result import ValidationReport
def validate_article_summary_payload(
result_payload: dict[str, Any],
extracted_payload: dict[str, Any] | None = None,
) -> ValidationReport:
# 校验 LLM 输出是否符合 ArticleSummaryResult schema
# extracted_payload 暂未使用,保留参数与 run_loop_payload validator 签名一致
try:
parsed = ArticleSummaryResult.model_validate(result_payload)
except ValidationError as exc:
errors = [
".".join(str(part) for part in error["loc"]) + ": " + error["msg"]
for error in exc.errors()
]
return ValidationReport(valid=False, errors=errors)
# keywords 和 topics 不能重叠
keyword_overlap = set(parsed.keywords) & set(parsed.topics)
errors: list[str] = []
if keyword_overlap:
errors.append(f"`keywords` and `topics` must not overlap: {sorted(keyword_overlap)}")
return ValidationReport(
valid=not errors,
errors=errors,
warnings=[],
normalized_result=parsed.model_dump(mode="json"),
)
+41 -10
View File
@@ -5,12 +5,13 @@ from pathlib import Path
from typing import Iterable, Mapping, Sequence from typing import Iterable, Mapping, Sequence
from summary_mcp.core.summary_loop import run_loop_payload from summary_mcp.core.summary_loop import run_loop_payload
from summary_mcp.validators.article_summary import validate_article_summary_payload
REPO_ROOT = Path(__file__).resolve().parents[3] REPO_ROOT = Path(__file__).resolve().parents[3]
OUTPUT_ROOT = REPO_ROOT / "outputs" OUTPUT_ROOT = REPO_ROOT / "outputs"
FRESHRSS_OUTPUT_ROOT = OUTPUT_ROOT / "freshrss" FRESHRSS_OUTPUT_ROOT = OUTPUT_ROOT / "freshrss"
DEFAULT_PROMPT_PATH = OUTPUT_ROOT / "prompts" / "llm-summary-prompt.txt" DEFAULT_PROMPT_PATH = OUTPUT_ROOT / "prompts" / "article-summary-prompt.txt"
@dataclass @dataclass
@@ -181,28 +182,58 @@ def summarize_selected_articles(
model=resolved_model, model=resolved_model,
api_url=resolved_api_url, api_url=resolved_api_url,
output_path=None, output_path=None,
validator=validate_article_summary_payload,
) )
if summary_exit_code != 0 or summary_payload is None: if summary_exit_code != 0 or summary_payload is None:
continue continue
# Render a simple Markdown file summarizing the article. # Render a Markdown knowledge note using the new article summary schema.
title = article.get("title") or summary_payload.get("title") or item_id title = article.get("title") or summary_payload.get("title") or item_id
url = article.get("url") url = article.get("url")
summary_text = summary_payload.get("summary") or "" category = summary_payload.get("category") or ""
highlights = summary_payload.get("highlights") or [] core_conclusion = summary_payload.get("core_conclusion") or ""
main_argument = summary_payload.get("main_argument") or ""
key_methods = summary_payload.get("key_methods") or []
important_details = summary_payload.get("important_details") or []
reusable_insights = summary_payload.get("reusable_insights") or []
keywords = summary_payload.get("keywords") or []
topics = summary_payload.get("topics") or []
lines: list[str] = [] lines: list[str] = []
lines.append(f"# {title}") lines.append(f"# {title}")
lines.append("")
if url: if url:
lines.append("")
lines.append(f"Source: {url}") lines.append(f"Source: {url}")
if category:
lines.append(f"Category: {category}")
lines.append("") lines.append("")
if summary_text: if core_conclusion:
lines.append(summary_text) lines.append("## 核心结论")
lines.append(core_conclusion)
lines.append("") lines.append("")
if highlights: if main_argument:
lines.append("## Highlights") lines.append("## 主要论点")
lines.extend(f"- {h}" for h in highlights) lines.append(main_argument)
lines.append("")
if key_methods:
lines.append("## 关键方法 / 机制")
lines.extend(f"- {m}" for m in key_methods)
lines.append("")
if important_details:
lines.append("## 重要细节")
lines.extend(f"- {d}" for d in important_details)
lines.append("")
if reusable_insights:
lines.append("## 可复用启发")
lines.extend(f"- {i}" for i in reusable_insights)
lines.append("")
if keywords:
lines.append("## 关键词")
lines.append("、".join(keywords))
lines.append("")
if topics:
lines.append("## 主题")
lines.append("、".join(topics))
lines.append("") lines.append("")
safe_title = "-".join( safe_title = "-".join(