diff --git a/daily/agent_workflow.py b/daily/agent_workflow.py index 01b0d6f..8b95e03 100644 --- a/daily/agent_workflow.py +++ b/daily/agent_workflow.py @@ -59,11 +59,41 @@ def _extract_markdown(text: str) -> str: def analyze_trends(llm_input: dict[str, Any], *, date_str: str) -> dict[str, Any] | None: + from daily.config import theme_ban_days + from daily.narrative_axis import ( + enforce_narrative_axis, + load_recent_axes, + load_recent_theme_summaries, + pick_narrative_axis, + ) + skill = _load_skill() + featured_note = "" + if llm_input.get("featured_pick"): + featured_note = ( + "\n输入已含 **featured_pick**（编辑指定今日首推）；" + "top_picks.skill 必须以 featured_pick 为准；" + "why/opening 不得向读者提及「编辑指定」。\n" + ) + used_axes = set(load_recent_axes(date_str)) + axis = pick_narrative_axis(used_axes) + llm_input["required_narrative_axis"] = axis + llm_input["narrative_axis"] = axis + theme_ban = load_recent_theme_summaries(date_str, theme_ban_days()) + ban_note = "" + if theme_ban: + ban_note = ( + "\n近几日已用过的主题/导语（请软避开同类开场，勿原样复用）：\n- " + + "\n- ".join(theme_ban) + + "\n" + ) system = ( f"{skill}\n\n" + f"{featured_note}" + f"{ban_note}" "当前执行 **Step 1：趋势分析**。\n" - "只输出 trends JSON（headline, opening, themes, top_picks, signals），不要 Markdown。" + f"**required_narrative_axis** = `{axis}`；输出 JSON 必须含 `narrative_axis` 且等于该值。\n" + "只输出 trends JSON（headline, opening, themes, top_picks, signals, narrative_axis），不要 Markdown。" ) user = json.dumps(llm_input, ensure_ascii=False, indent=2) try: @@ -77,8 +107,9 @@ def analyze_trends(llm_input: dict[str, Any], *, date_str: str) -> dict[str, Any if not parsed.get("headline") and not parsed.get("opening"): logger.warning("Agent Step1 JSON 无效") return None + parsed = enforce_narrative_axis(parsed, axis) save_json(trends_json_path(date_str), parsed) - logger.info("Agent Step1 完成：%s", parsed.get("headline", "?")) + logger.info("Agent Step1 完成：%s [%s]", parsed.get("headline", "?"), axis) return parsed @@ -91,8 +122,17 @@ def write_wecom_report( updated: str, ) -> str | None: skill = _load_skill() + featured_note = "" + if llm_input.get("featured_pick"): + featured_note = ( + "\n输入 data 已含 **featured_pick**；" + "今日首推区块须使用 featured_pick.why_today；" + "链接行用 Markdown [标题](URL)，勿用反引号裸 URL；" + "读者可见文案不得出现「编辑指定」等元信息。\n" + ) system = ( f"{skill}\n\n" + f"{featured_note}" "当前执行 **Step 2：撰写企微早报**。\n" f"日期={date_str}，时间={time_str}，数据截至={updated}。\n" "只输出企微 Markdown 正文，不要代码块，不要 JSON。" diff --git a/daily/narrative_axis.py b/daily/narrative_axis.py new file mode 100644 index 0000000..7ed4195 --- /dev/null +++ b/daily/narrative_axis.py @@ -0,0 +1,103 @@ +"""叙事轴硬互斥：代码选定轴，注入 Agent Step1 并强制覆写。""" + +from __future__ import annotations + +import json +import logging +import random +from datetime import datetime, timedelta +from typing import Any + +from daily.config import OUTPUT_DIR, narrative_axis_days + +logger = logging.getLogger(__name__) + +NARRATIVE_AXES: tuple[str, ...] = ( + "政策监管", + "模型发布", + "工具链/Agent", + "芯片算力", + "开源生态", + "应用落地", + "安全/诉讼", +) + + +def pick_narrative_axis( + used: set[str], + *, + rng: random.Random | None = None, +) -> str: + """从固定轴枚举中排除已用轴后随机选取；全用尽则回退全表。""" + available = [a for a in NARRATIVE_AXES if a not in used] + pool = available or list(NARRATIVE_AXES) + picker = rng or random.Random() + return picker.choice(pool) + + +def load_recent_axes(date_str: str, days: int | None = None) -> list[str]: + """近 N 日 data.narrative_axis（不含当日，按时间从近到远）。""" + lookback = days if days is not None else narrative_axis_days() + try: + dt = datetime.strptime(date_str, "%Y-%m-%d") + except ValueError: + return [] + axes: list[str] = [] + for day_offset in range(1, lookback + 1): + prev = (dt - timedelta(days=day_offset)).strftime("%Y-%m-%d") + path = OUTPUT_DIR / f"{prev}.data.json" + if not path.exists(): + continue + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + logger.warning("读取 narrative_axis %s 失败：%s", path, exc) + continue + data = payload.get("data") if isinstance(payload, dict) else None + if not isinstance(data, dict): + continue + axis = str(data.get("narrative_axis") or "").strip() + if axis: + axes.append(axis) + return axes + + +def load_recent_theme_summaries(date_str: str, days: int) -> list[str]: + """近 N 日 theme/opening 摘要，供 Step1 软禁参考。""" + try: + dt = datetime.strptime(date_str, "%Y-%m-%d") + except ValueError: + return [] + summaries: list[str] = [] + for day_offset in range(1, days + 1): + prev = (dt - timedelta(days=day_offset)).strftime("%Y-%m-%d") + path = OUTPUT_DIR / f"{prev}.data.json" + if not path.exists(): + continue + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + continue + data = payload.get("data") if isinstance(payload, dict) else None + if not isinstance(data, dict): + continue + theme = str(data.get("theme") or data.get("editorial_theme") or "").strip() + opening = "" + trends = data.get("trends") if isinstance(data.get("trends"), dict) else {} + if isinstance(trends, dict): + opening = str(trends.get("opening") or "").strip() + if not theme: + themes = trends.get("themes") or [] + if themes and isinstance(themes[0], dict): + theme = str(themes[0].get("title") or "").strip() + bit = " · ".join(x for x in (prev, theme, opening[:40]) if x) + if bit: + summaries.append(bit) + return summaries + + +def enforce_narrative_axis(trends: dict[str, Any], axis: str) -> dict[str, Any]: + """强制 trends['narrative_axis'] = axis。""" + out = dict(trends) + out["narrative_axis"] = axis + return out diff --git a/skills/daily-agent/SKILL.md b/skills/daily-agent/SKILL.md index 8196547..aed8b7d 100644 --- a/skills/daily-agent/SKILL.md +++ b/skills/daily-agent/SKILL.md @@ -30,8 +30,9 @@ Step 2 基于趋势 + 原始数据 → 写企微 Markdown 早报 | `github_topic.topic` | Topic 名称，用于区块标题（如 `llm`） | | `movement.*_moves` | 较昨日新增（**仅**用于 opening / signals；`effective_wecom_mode=delta` 时列表由 Python 插入） | | `movement.*_summary` | 新增摘要（可选写入 signals） | -| `ai_news` | 国际 AI 时讯 Top N | -| `cn_ai_news` | 国内 AI 时讯 Top N | +| `ai_news` | 国际 AI 时讯 Top N（RSS 模式） | +| `cn_ai_news` | 国内 AI 时讯 Top N（RSS 模式） | +| `ai_news_mode` | `research` 时仅 `ai_news` 有 10 条合并精选，`cn_ai_news` 为空 | | `featured_pick` | **可选**，编辑指定今日首推（Step 0 产出；含 command / why_today / evidence） | **禁止**使用已合并的 `skills_moves` / `github_moves` 自行扩写；**禁止**排名变化、安装涨跌。 @@ -51,6 +52,7 @@ Step 1 的 `signals` 与 `top_picks` **优先引用 Top 榜榜首/前列条目 { "headline": "8–16字焦点标题", "opening": "2–3句中文导语；首句必须是具体证据（榜首 skill+安装量 / 头条新闻 / GitHub #1），再解释为什么值得看", + "narrative_axis": "必须等于输入 required_narrative_axis（政策监管|模型发布|工具链/Agent|芯片算力|开源生态|应用落地|安全/诉讼）", "themes": [ { "title": "主题名", @@ -73,6 +75,7 @@ Step 1 的 `signals` 与 `top_picks` **优先引用 Top 榜榜首/前列条目 要求： - 所有结论必须能在输入 JSON 中找到依据，禁止编造 +- `narrative_axis` **必填**，且必须等于输入中的 `required_narrative_axis`（代码已选定；勿自选其它轴） - `opening` 遵循 **article-writing Newsletter** 规则：首句用数字/条目名/新闻标题开头，不用「今天有三条线」「值得关注」等空框架 - `signals` 3–5 条，每条单行，可含 emoji 前缀；与 `opening` 不重复同一句信息 - `top_picks.why` 用「事实/数字 + 一句判断」，不用空泛形容词 @@ -117,9 +120,13 @@ Step 1 的 `signals` 与 `top_picks` **优先引用 Top 榜榜首/前列条目 1. [{title_zh}]({link}) — {why 或摘要} 2. ...（**必须 10 条**，来自 `ai_news`，按重要性排序） -🇨🇳 **国内 AI · 精选 8** +🇨🇳 **国内 AI · 精选 10** 1. [{title}]({link}) — {why 或摘要} -2. ...（**必须 8 条**，来自 `cn_ai_news`，按重要性排序；标题已是中文，可微调润色） +2. ...（**必须 10 条**，来自 `cn_ai_news`；RSS 模式） + +📰 **AI 时讯精选 · 15**（当 `ai_news_mode=research`） +1. [{title}]({link}) — {why 或摘要} +2. ...（**必须 15 条**：前 10 条综合精选 + 后 5 条偏工程技术，来自 `ai_news` 与 `tech_ai_news`；**不要**再写 🌍/🇨🇳/🔧 分块） @@ -144,9 +151,10 @@ Step 1 的 `signals` 与 `top_picks` **优先引用 Top 榜榜首/前列条目 4. **禁止**在条目后写 `（新入 … #n）` 类括号标注 5. `effective_wecom_mode=full` 时：即使某榜较昨日无新增，仍须完整列出 Top 榜条目 6. **国际 AI 必须 10 条**（来自 `ai_news`） -7. **国内 AI 必须 8 条**（来自 `cn_ai_news`；无数据时写「暂无可用条目」） -8. 禁止排名变化、安装涨跌、连霸描述 -9. 榜全稳（movement 各 `*_stable` 为 true）时，signals 聚焦新闻与首推，不编造榜单变化 +7. **国内 AI 必须 10 条**（来自 `cn_ai_news`；无数据时写「暂无可用条目」） +8. **`ai_news_mode=research` 时**：只写 **📰 AI 时讯精选 15 条**（`ai_news` 10 条 + `tech_ai_news` 5 条合并展示），不写 🌍/🇨🇳/🔧 分块；Python 会用调研结果覆盖该区块 +9. 禁止排名变化、安装涨跌、连霸描述 +10. 榜全稳（movement 各 `*_stable` 为 true）时，signals 聚焦新闻与首推，不编造榜单变化 ```markdown 📈 **Skills Trending Top 10** diff --git a/tests/test_narrative_axis.py b/tests/test_narrative_axis.py new file mode 100644 index 0000000..54eaa6f --- /dev/null +++ b/tests/test_narrative_axis.py @@ -0,0 +1,34 @@ +# tests/test_narrative_axis.py +from __future__ import annotations + +import random +import unittest + +from daily.narrative_axis import ( + NARRATIVE_AXES, + enforce_narrative_axis, + pick_narrative_axis, +) + + +class NarrativeAxisTests(unittest.TestCase): + def test_pick_excludes_used(self): + used = {"政策监管", "模型发布", "工具链/Agent"} + for _ in range(20): + axis = pick_narrative_axis(used, rng=random.Random(1)) + self.assertNotIn(axis, used) + self.assertIn(axis, NARRATIVE_AXES) + + def test_enforce_overwrites_llm(self): + trends = {"narrative_axis": "开源生态", "opening": "..."} + out = enforce_narrative_axis(trends, "芯片算力") + self.assertEqual(out["narrative_axis"], "芯片算力") + + def test_pick_when_all_used_falls_back(self): + used = set(NARRATIVE_AXES) + axis = pick_narrative_axis(used, rng=random.Random(0)) + self.assertIn(axis, NARRATIVE_AXES) + + +if __name__ == "__main__": + unittest.main()