feat: 代码选定叙事轴并注入 Agent 开场约束

Co-authored-by: Cursor <cursoragent@cursor.com>

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。"

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