增加需求拓展

.env.example

@@ -2,8 +2,8 @@
 OPENROUTER_API_KEY=sk-or-v1-...
 
 # Default model (any OpenRouter-supported model)
-# Examples: openai/gpt-4o, anthropic/claude-sonnet-5, google/gemini-2.5-pro-preview
-OPENROUTER_MODEL=anthropic/claude-sonnet-5
+# Examples: google/gemini-2.5-flash, google/gemini-2.5-flash-lite, anthropic/claude-sonnet-5
+OPENROUTER_MODEL=google/gemini-2.5-flash
 
 # Agent defaults
 AGENT_MAX_ITERATIONS=20

README.md

@@ -136,6 +136,7 @@ async for event in agent.astream("Your question"):
 
 通过 OpenRouter 可使用任意支持的模型，例如：
 
+- `google/gemini-2.5-flash`（默认）
 - `anthropic/claude-sonnet-5`
 - `openai/gpt-4o`
 - `google/gemini-2.5-pro-preview`
@@ -148,7 +149,7 @@ async for event in agent.astream("Your question"):
 | 变量 | 说明 | 默认值 |
 |------|------|--------|
 | `OPENROUTER_API_KEY` | OpenRouter API 密钥 | (必填) |
-| `OPENROUTER_MODEL` | 默认模型 | `anthropic/claude-sonnet-5` |
+| `OPENROUTER_MODEL` | 默认模型 | `google/gemini-2.5-flash` |
 | `AGENT_MAX_ITERATIONS` | 最大循环次数 | `20` |
 | `AGENT_TEMPERATURE` | 生成温度 | `0.7` |
 | `SKILLS_DIR` | Skills 目录 | `skills` |

agents/demand_video_expand_agent/__init__.py

@@ -0,0 +1,9 @@
+"""
+demand_video_expand_agent — S/A 需求视频点位拓展判断 Agent
+
+职责：对任务层已组装好的「需求 + 视频点位」做语义判断，
+筛选可作为拓展需求的点位并落库。不负责查库。
+"""
+from agents.demand_video_expand_agent.agent import create_demand_video_expand_agent
+
+__all__ = ["create_demand_video_expand_agent"]

agents/demand_video_expand_agent/agent.py

@@ -0,0 +1,30 @@
+"""
+demand_video_expand_agent 工厂 — 组装需求视频点位拓展判断 Agent。
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+from supply_agent import Agent
+from supply_agent.config import Settings
+from agents.demand_video_expand_agent.tools import register_all_tools
+
+_PROMPT_PATH = Path(__file__).parent / "prompt" / "system_prompt.md"
+DEMAND_VIDEO_EXPAND_AGENT_SYSTEM_PROMPT = _PROMPT_PATH.read_text(encoding="utf-8")
+
+
+def create_demand_video_expand_agent(
+    settings: Settings | None = None,
+    *,
+    model: str | None = None,
+) -> Agent:
+    """创建 demand_video_expand_agent 实例，注册专属工具。"""
+    agent = Agent(
+        settings=settings,
+        name="demand_video_expand_agent",
+        model=model,
+        system_prompt=DEMAND_VIDEO_EXPAND_AGENT_SYSTEM_PROMPT,
+        max_iterations=10,
+    )
+    register_all_tools(agent.tools)
+    return agent

agents/demand_video_expand_agent/prompt/system_prompt.md

@@ -0,0 +1,45 @@
+## 角色与任务
+你是需求拓展判断专家。用户会提供一个已评级的需求（demand_name、grade、demand_grade_id），
+以及其关联视频的全部点位列表（point_type / point_data / point_desc / video_id）。
+
+你只负责：
+1. 判断哪些点位可作为「拓展需求」
+2. 为每条候选写清 reason（为何与原需求相近）
+3. 调用 `batch_save_demand_expansions` 落库
+
+禁止调用任何查询工具；所有数据已在用户消息中给出。
+
+## 点位类型含义
+- **purpose**（目的点）：用户观看该视频的目的，拓展价值最高
+- **key**（关键点）：视频核心话题，常可作为子需求/细分需求
+- **inspiration**（灵感点）：创作灵感/场景，仅在与原需求强相关时采纳
+
+同等相近时，优先级：**purpose > key > inspiration**。
+
+## 相近可采纳（满足其一即可，但 reason 须写清依据）
+1. **语义包含/被包含**：原需求是上位概念，点位是下位具体话题（或相反但意图一致）
+2. **同场景细分**：同一使用场景下的更细需求表达
+3. **同意图不同表达**：措辞不同但用户意图一致
+
+## 必须剔除
+- 与 demand_name 完全相同或仅差标点/空格
+- 过于宽泛、无法作为具体供给方向（如单独的「健康饮食」「减肥」且无场景）
+- 与原需求无关的点位（同一视频但语义脱节）
+- 纯描述性语句而非需求表达（如「视频展示了三种做法」）
+- point_data 为空或无意义片段
+
+## 工作流程
+1. 阅读 demand_name 与各点位
+2. 筛选可拓展候选，整理 items
+3. 调用 `batch_save_demand_expansions`，将用户消息中的 biz_dt、demand_grade_id、
+   demand_name、grade、run_id 原样传入
+4. 若无合适拓展，传 `items=[]`，并在回复中说明原因
+5. 简要总结：采纳几条、主要剔除原因
+
+## 落库字段说明
+items 每项：
+- `expanded_text`：来自 point_data，作为拓展需求文本
+- `point_type`：inspiration / purpose / key
+- `point_desc`：点位描述（有则带上）
+- `video_id`：来源视频
+- `reason`：为何可作为拓展（必填，具体可读）

agents/demand_video_expand_agent/run.py

@@ -0,0 +1,91 @@
+#!/usr/bin/env python3
+"""单需求视频点位拓展判断 — 由任务层调用。"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from supply_agent.types import AgentResult, Role
+
+_POINT_TYPE_LABEL = {
+    "inspiration": "灵感点",
+    "purpose": "目的点",
+    "key": "关键点",
+}
+
+
+@dataclass
+class VideoPoint:
+    video_id: str
+    point_type: str
+    point_data: str | None
+    point_desc: str | None
+
+
+@dataclass
+class DemandExpandContext:
+    biz_dt: str
+    run_id: str
+    demand_grade_id: int
+    demand_name: str
+    grade: str
+    score: float | None
+    video_ids: list[str]
+    points: list[VideoPoint]
+
+
+def build_expand_user_input(ctx: DemandExpandContext) -> str:
+    """构建传给 Agent 的用户消息。"""
+    score_text = f"{ctx.score:.2f}" if ctx.score is not None else "—"
+    lines = [
+        f"biz_dt={ctx.biz_dt}",
+        f"run_id={ctx.run_id}",
+        f"demand_grade_id={ctx.demand_grade_id}",
+        f"demand_name={ctx.demand_name}",
+        f"grade={ctx.grade}",
+        f"score={score_text}",
+        f"video_count={len(ctx.video_ids)}",
+        "",
+        "以下是从关联视频中提取的点位，请判断哪些可作为拓展需求：",
+    ]
+    for index, point in enumerate(ctx.points, start=1):
+        type_label = _POINT_TYPE_LABEL.get(point.point_type, point.point_type)
+        data_text = point.point_data or "—"
+        desc_text = point.point_desc or "—"
+        lines.append(
+            f"[{index}|{type_label}|video={point.video_id}] "
+            f"{data_text} | 描述: {desc_text}"
+        )
+    lines.extend(
+        [
+            "",
+            "判断完成后调用 batch_save_demand_expansions 落库；",
+            "将上述 biz_dt、demand_grade_id、demand_name、grade、run_id 原样传入工具。",
+            "若无合适拓展，传 items=[]。",
+        ]
+    )
+    return "\n".join(lines)
+
+
+def judge_demand_expansion(ctx: DemandExpandContext) -> AgentResult:
+    """对单个需求执行拓展判断。"""
+    from agents.demand_video_expand_agent.agent import create_demand_video_expand_agent
+
+    agent = create_demand_video_expand_agent()
+    user_input = build_expand_user_input(ctx)
+    return agent.run(user_input)
+
+
+def extract_saved_count(result: AgentResult) -> int:
+    """从 Agent 工具返回消息中解析保存条数。"""
+    for msg in reversed(result.messages):
+        if msg.role != Role.TOOL or not msg.content:
+            continue
+        text = str(msg.content)
+        if "成功保存" not in text:
+            continue
+        try:
+            part = text.split("成功保存", 1)[1].strip()
+            return int(part.split("条", 1)[0].strip())
+        except (IndexError, ValueError):
+            continue
+    return 0

agents/demand_video_expand_agent/tools/__init__.py

@@ -0,0 +1,27 @@
+"""
+demand_video_expand_agent 工具包
+"""
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Any
+
+from agents.demand_video_expand_agent.tools.batch_save_demand_expansions import (
+    batch_save_demand_expansions,
+)
+from supply_agent.tools.registry import ToolRegistry
+
+ALL_TOOLS: list[Callable[..., Any]] = [
+    batch_save_demand_expansions,
+]
+
+__all__ = [
+    "ALL_TOOLS",
+    "batch_save_demand_expansions",
+    "register_all_tools",
+]
+
+
+def register_all_tools(registry: ToolRegistry) -> ToolRegistry:
+    """将 demand_video_expand_agent 包内的所有工具注册到 ToolRegistry。"""
+    return registry.from_decorated(*ALL_TOOLS)

agents/demand_video_expand_agent/tools/batch_save_demand_expansions.py

@@ -0,0 +1,222 @@
+"""
+批量保存视频点位拓展判断结果。
+"""
+from __future__ import annotations
+
+import logging
+import re
+import uuid
+from typing import Any
+
+from supply_agent.tools import tool
+from supply_infra.db.models.multi_demand_video_point import POINT_TYPES
+from supply_infra.db.repositories.demand_video_expansion_repo import (
+    DemandVideoExpansionRepository,
+)
+from supply_infra.db.repositories.multi_demand_video_point_repo import (
+    MultiDemandVideoPointRepository,
+)
+from supply_infra.db.session import get_session
+
+logger = logging.getLogger(__name__)
+
+_VALID_GRADES = frozenset({"S", "A"})
+
+
+def _optional_str(value: Any) -> str | None:
+    if value is None:
+        return None
+    text = str(value).strip()
+    return text or None
+
+
+def _normalize_items(
+    items: list[dict[str, Any]],
+    *,
+    default_run_id: str,
+    biz_dt: str,
+    source_demand_grade_id: int,
+    source_demand_name: str,
+    source_grade: str,
+) -> tuple[list[dict[str, Any]], list[str]]:
+    rows: list[dict[str, Any]] = []
+    errors: list[str] = []
+    seen_keys: set[tuple[str, str]] = set()
+
+    for idx, item in enumerate(items):
+        if not isinstance(item, dict):
+            errors.append(f"第 {idx} 项不是对象")
+            continue
+
+        expanded_text = _optional_str(item.get("expanded_text"))
+        point_type = _optional_str(item.get("point_type"))
+        video_id = _optional_str(item.get("video_id"))
+        reason = _optional_str(item.get("reason"))
+
+        if not expanded_text:
+            errors.append(f"第 {idx} 项缺少 expanded_text")
+            continue
+        if not point_type or point_type not in POINT_TYPES:
+            allowed = " / ".join(POINT_TYPES)
+            errors.append(f"第 {idx} 项 point_type 无效（只能是 {allowed}）: {point_type!r}")
+            continue
+        if not video_id:
+            errors.append(f"第 {idx} 项缺少 video_id")
+            continue
+        if not reason:
+            errors.append(f"第 {idx} 项缺少 reason")
+            continue
+
+        dedupe_key = (expanded_text, video_id)
+        if dedupe_key in seen_keys:
+            errors.append(f"第 {idx} 项在本次请求中重复: {expanded_text!r} / {video_id!r}")
+            continue
+        seen_keys.add(dedupe_key)
+
+        rows.append(
+            {
+                "biz_dt": biz_dt,
+                "run_id": _optional_str(item.get("run_id")) or default_run_id,
+                "source_demand_grade_id": source_demand_grade_id,
+                "source_demand_name": source_demand_name,
+                "source_grade": source_grade,
+                "expanded_text": expanded_text,
+                "point_type": point_type,
+                "point_desc": _optional_str(item.get("point_desc")),
+                "video_id": video_id,
+                "reason": reason,
+                "is_delete": 0,
+            }
+        )
+
+    return rows, errors
+
+
+def _normalize_demand_name(name: str) -> str:
+    text = re.sub(r"\s+", "", name.strip().lower())
+    return re.sub(r"[，,。．.！!？?；;：:""''\"'、/\\|·—_（）()【】\[\]《》<>]", "", text)
+
+
+def _fill_missing_point_descs(rows: list[dict[str, Any]], session) -> None:
+    """point_desc 为空时，从 multi_demand_video_point 按 video_id/point_type/point_data 补全。
+
+    查不到或源表 point_desc 也为空时，保持原值（仍为 None）。
+    """
+    missing = [row for row in rows if not row.get("point_desc")]
+    if not missing:
+        return
+
+    video_ids = sorted({str(row["video_id"]) for row in missing})
+    points_by_vid = MultiDemandVideoPointRepository(session).list_by_video_ids(video_ids)
+
+    lookup: dict[tuple[str, str, str], str] = {}
+    for vid, points in points_by_vid.items():
+        for point in points:
+            point_data = _optional_str(point.get("point_data"))
+            point_type = _optional_str(point.get("point_type"))
+            point_desc = _optional_str(point.get("point_desc"))
+            if not point_data or not point_type or not point_desc:
+                continue
+            lookup[(vid, point_type, point_data)] = point_desc
+
+    for row in missing:
+        key = (str(row["video_id"]), str(row["point_type"]), str(row["expanded_text"]))
+        desc = lookup.get(key)
+        if desc:
+            row["point_desc"] = desc
+        # 未命中或源表无描述：不改动，保留空值
+
+
+@tool
+def batch_save_demand_expansions(
+    items: list[dict[str, Any]],
+    biz_dt: str,
+    source_demand_grade_id: int,
+    source_demand_name: str,
+    source_grade: str,
+    run_id: str | None = None,
+) -> str:
+    """
+    保存视频点位拓展判断结果到 demand_video_expansion 表。
+
+    Args:
+        items: 拓展候选列表。每项必填：
+            - expanded_text: 拓展需求文本（来自 point_data）
+            - point_type: inspiration / purpose / key
+            - video_id: 来源视频 id
+            - reason: 为何与原需求相近、可作为拓展
+          选填：
+            - point_desc: 点位描述快照
+        biz_dt: 业务日 YYYYMMDD。
+        source_demand_grade_id: 来源 demand_grade.id（由用户消息提供，原样传入）。
+        source_demand_name: 来源需求名。
+        source_grade: 来源等级 S 或 A。
+        run_id: 任务批次 id；省略则自动生成。
+
+    Returns:
+        保存结果摘要。若无合适拓展，传 items=[] 即可。
+    """
+    biz_dt_text = _optional_str(biz_dt)
+    if not biz_dt_text or len(biz_dt_text) != 8 or not biz_dt_text.isdigit():
+        return f"biz_dt 格式无效，应为 YYYYMMDD: {biz_dt!r}"
+
+    demand_name = _optional_str(source_demand_name)
+    if not demand_name:
+        return "source_demand_name 不能为空"
+
+    grade = _optional_str(source_grade)
+    if grade not in _VALID_GRADES:
+        return f"source_grade 无效，只能是 S 或 A: {source_grade!r}"
+
+    try:
+        grade_id = int(source_demand_grade_id)
+    except (TypeError, ValueError):
+        return f"source_demand_grade_id 无效: {source_demand_grade_id!r}"
+
+    if not items:
+        return "无拓展候选，跳过落库"
+
+    default_run_id = _optional_str(run_id) or uuid.uuid4().hex
+    rows, errors = _normalize_items(
+        items,
+        default_run_id=default_run_id,
+        biz_dt=biz_dt_text,
+        source_demand_grade_id=grade_id,
+        source_demand_name=demand_name,
+        source_grade=grade,
+    )
+
+    normalized_demand = _normalize_demand_name(demand_name)
+    filtered_rows: list[dict[str, Any]] = []
+    skipped_same: list[str] = []
+    for row in rows:
+        if _normalize_demand_name(row["expanded_text"]) == normalized_demand:
+            skipped_same.append(row["expanded_text"])
+            continue
+        filtered_rows.append(row)
+
+    if not filtered_rows:
+        detail = "；".join(errors) if errors else "无有效数据"
+        same_note = ""
+        if skipped_same:
+            same_note = f"；剔除与原需求相同 {len(skipped_same)} 条"
+        return f"没有可保存的数据: {detail}{same_note}"
+
+    try:
+        with get_session() as session:
+            _fill_missing_point_descs(filtered_rows, session)
+            saved = DemandVideoExpansionRepository(session).bulk_upsert(filtered_rows)
+
+        parts = [f"成功保存 {saved} 条拓展需求", f"biz_dt={biz_dt_text}", f"run_id={default_run_id}"]
+        if skipped_same:
+            parts.append(f"剔除与原需求相同 {len(skipped_same)} 条")
+        if errors:
+            parts.append(f"校验失败 {len(errors)} 条: " + "；".join(errors[:10]))
+
+        message = "。".join(parts)
+        logger.info("batch_save_demand_expansions completed: %s", message)
+        return message
+
+    except Exception as e:
+        logger.error("batch_save_demand_expansions failed: %s", e, exc_info=True)
+        return f"保存拓展需求失败: {e}"

api/app.py

@@ -94,7 +94,7 @@ def demand_grade(
 
 @app.get("/api/demand-grade/{demand_grade_id}/videos")
 def demand_grade_videos(demand_grade_id: int) -> dict:
-    """Return videos linked to a demand_grade row (vid + title + points JSON)."""
+    """Return expansion videos/points for a demand_grade row (by its biz_dt)."""
     result = list_videos_for_demand_grade(demand_grade_id)
     if result is None:
         raise HTTPException(status_code=404, detail="demand_grade not found")

api/services/demand_grade_videos.py

@@ -1,16 +1,17 @@
-"""Resolve demand_grade.video_list → multi_demand_video_detail."""
+"""Resolve demand_video_expansion → videos + expansion points for the web UI."""
 from __future__ import annotations
 
 import json
 from typing import Any
 
 from supply_infra.db.repositories.demand_grade_repo import DemandGradeRepository
+from supply_infra.db.repositories.demand_video_expansion_repo import (
+    DemandVideoExpansionRepository,
+    DemandVideoExpansionRunRepository,
+)
 from supply_infra.db.repositories.multi_demand_video_detail_repo import (
     MultiDemandVideoDetailRepository,
 )
-from supply_infra.db.repositories.multi_demand_video_point_repo import (
-    MultiDemandVideoPointRepository,
-)
 from supply_infra.db.session import get_session
 
 
@@ -24,49 +25,72 @@ def _parse_json_list(raw: str | None) -> list[Any]:
     return parsed if isinstance(parsed, list) else []
 
 
-def _parse_video_ids(raw: str | None) -> list[str]:
-    return [str(v).strip() for v in _parse_json_list(raw) if v is not None and str(v).strip()]
-
-
 def list_videos_for_demand_grade(demand_grade_id: int) -> dict[str, Any] | None:
     """
-    按 demand_grade.id 返回关联视频详情。
+    按 demand_grade.id 返回该需求在对应 biz_dt 下的拓展视频与选题结果。
 
-    顺序与 video_list 一致；详情表缺失的 vid 仍返回，title/三点 JSON 为空。
-    category_ids 返回该分级需求归属的全部全局树节点 id（可能不止一个）。
+    数据来源：
+    - demand_video_expansion_run：判断是否已完成拓展（含零结果）
+    - demand_video_expansion：真实视频实例与拓展点位
     """
     with get_session() as session:
         grade = DemandGradeRepository(session).get_by_id(demand_grade_id)
         if grade is None:
             return None
 
-        vids = _parse_video_ids(grade.video_list)
-        details = MultiDemandVideoDetailRepository(session).list_by_vids(vids)
-        points_by_vid = MultiDemandVideoPointRepository(session).json_fields_by_video_ids(
-            vids
+        biz_dt = str(grade.biz_dt)
+        run = DemandVideoExpansionRunRepository(session).get_by_demand_grade(
+            biz_dt, demand_grade_id
+        )
+        expansions = DemandVideoExpansionRepository(session).list_by_demand_grade(
+            biz_dt, demand_grade_id
         )
 
+        vids: list[str] = []
+        seen_vids: set[str] = set()
+        points_by_vid: dict[str, list[dict[str, Any]]] = {}
+        for row in expansions:
+            vid = str(row.video_id).strip()
+            if not vid:
+                continue
+            if vid not in seen_vids:
+                seen_vids.add(vid)
+                vids.append(vid)
+            points_by_vid.setdefault(vid, []).append(
+                {
+                    "expanded_text": row.expanded_text,
+                    "point_type": row.point_type,
+                    "point_desc": row.point_desc,
+                    "reason": row.reason,
+                }
+            )
+
+        details = MultiDemandVideoDetailRepository(session).list_by_vids(vids)
         videos: list[dict[str, Any]] = []
         for vid in vids:
             row = details.get(vid)
-            point_fields = points_by_vid.get(vid, {})
             videos.append(
                 {
                     "vid": vid,
                     "title": row.title if row else None,
-                    "inspiration_points_json": point_fields.get(
-                        "inspiration_points_json"
-                    ),
-                    "purpose_points_json": point_fields.get("purpose_points_json"),
-                    "key_points_json": point_fields.get("key_points_json"),
+                    "expansion_points": points_by_vid.get(vid, []),
                 }
             )
 
+        expansion_status: str | None = None
+        expansion_saved_count = 0
+        if run is not None:
+            expansion_status = str(run.status)
+            expansion_saved_count = int(run.saved_count or 0)
+
         return {
             "demand_grade_id": grade.id,
             "demand_name": grade.demand_name,
+            "biz_dt": biz_dt,
             "category_ids": [int(c) for c in _parse_json_list(grade.category_ids)],
             "grade": grade.grade,
             "strategies": _parse_json_list(grade.strategies),
+            "expansion_status": expansion_status,
+            "expansion_saved_count": expansion_saved_count,
             "videos": videos,
         }

jobs/backfill_demand_video_expansion_point_desc.py

@@ -0,0 +1,36 @@
+#!/usr/bin/env python3
+"""手动补全 demand_video_expansion 缺失的 point_desc。
+
+用法:
+    python jobs/backfill_demand_video_expansion_point_desc.py
+    python jobs/backfill_demand_video_expansion_point_desc.py 20260721
+    python jobs/backfill_demand_video_expansion_point_desc.py 20260721 --dry-run
+"""
+from __future__ import annotations
+
+import logging
+import sys
+from pathlib import Path
+
+_ROOT = Path(__file__).resolve().parents[1]
+if str(_ROOT) not in sys.path:
+    sys.path.insert(0, str(_ROOT))
+
+from scripts.backfill_demand_video_expansion_point_desc import backfill_missing_point_descs
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+)
+
+
+def main(biz_dt: str | None = None, *, dry_run: bool = False) -> dict:
+    result = backfill_missing_point_descs(biz_dt, dry_run=dry_run)
+    print(result)
+    return result
+
+
+if __name__ == "__main__":
+    args = sys.argv[1:]
+    biz_dt_arg = args[0] if args and not args[0].startswith("-") else None
+    main(biz_dt_arg, dry_run="--dry-run" in args)

jobs/expand_demand_from_video_points.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python3
+"""手动执行 S/A 需求视频点位拓展任务。
+
+用法:
+    python jobs/expand_demand_from_video_points.py                  # 最新/当天 biz_dt
+    python jobs/expand_demand_from_video_points.py 20260721          # 指定业务日
+    python jobs/expand_demand_from_video_points.py 20260721 5        # 指定业务日 + 5 并发
+    python jobs/expand_demand_from_video_points.py 20260721 --force   # 忽略已完成记录重跑
+"""
+from __future__ import annotations
+
+import logging
+import sys
+
+from supply_infra.scheduler.jobs.expand_demand_from_video_points import (
+    expand_demand_from_video_points,
+)
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+)
+
+
+def main(
+    biz_dt: str | None = None,
+    workers_arg: str | None = None,
+    *,
+    skip_finished: bool = True,
+) -> dict:
+    workers = int(workers_arg) if workers_arg else 5
+    result = expand_demand_from_video_points(
+        biz_dt,
+        skip_finished=skip_finished,
+        workers=workers,
+    )
+    print(result)
+    return result
+
+
+if __name__ == "__main__":
+    args = sys.argv[1:]
+    biz_dt_arg = args[0] if args and not args[0].startswith("-") else None
+    workers_arg = None
+    if biz_dt_arg and len(args) > 1 and not args[1].startswith("-"):
+        workers_arg = args[1]
+    skip_finished = "--force" not in args
+    main(biz_dt_arg, workers_arg, skip_finished=skip_finished)

jobs/retry_failed_grade_plan_items.py

@@ -0,0 +1,45 @@
+#!/usr/bin/env python3
+"""手动重试 demand_grade_plan_group_item 中失败的分级任务。
+
+用法:
+    python jobs/retry_failed_grade_plan_items.py
+    python jobs/retry_failed_grade_plan_items.py 20260721
+    python jobs/retry_failed_grade_plan_items.py 20260721 5
+    python jobs/retry_failed_grade_plan_items.py 20260721 5 --dry-run
+"""
+from __future__ import annotations
+
+import logging
+import sys
+
+from supply_infra.scheduler.jobs.grade_demand_pool import retry_failed_plan_group_items
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+)
+
+
+def main(
+    biz_dt: str | None = None,
+    workers_arg: str | None = None,
+    *,
+    dry_run: bool = False,
+) -> dict:
+    workers = int(workers_arg) if workers_arg else 5
+    result = retry_failed_plan_group_items(
+        biz_dt,
+        workers=workers,
+        dry_run=dry_run,
+    )
+    print(result)
+    return result
+
+
+if __name__ == "__main__":
+    args = sys.argv[1:]
+    biz_dt_arg = args[0] if args and not args[0].startswith("-") else None
+    workers_arg = None
+    if biz_dt_arg and len(args) > 1 and not args[1].startswith("-"):
+        workers_arg = args[1]
+    main(biz_dt_arg, workers_arg, dry_run="--dry-run" in args)

jobs/run_supply_pipeline.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""手动执行供给数据流水线（全局树 → 需求池 → 分级）。"""
+"""手动执行供给数据流水线（全局树 → 需求池 → 分级 → 视频点位拓展）。"""
 
 import logging
 import sys

scripts/backfill_demand_video_expansion_point_desc.py

@@ -0,0 +1,162 @@
+#!/usr/bin/env python3
+"""补全 demand_video_expansion 表中缺失的 point_desc。
+
+从 multi_demand_video_point 按 (video_id, point_type, expanded_text=point_data) 匹配；
+查不到则保持空值。
+
+用法:
+  .venv/bin/python scripts/backfill_demand_video_expansion_point_desc.py
+  .venv/bin/python scripts/backfill_demand_video_expansion_point_desc.py --biz-dt 20260721
+  .venv/bin/python scripts/backfill_demand_video_expansion_point_desc.py --biz-dt 20260721 --dry-run
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import sys
+from pathlib import Path
+from typing import Any
+
+from sqlalchemy import or_, select, update
+
+_ROOT = Path(__file__).resolve().parents[1]
+if str(_ROOT) not in sys.path:
+    sys.path.insert(0, str(_ROOT))
+
+from agents.demand_video_expand_agent.tools.batch_save_demand_expansions import (
+    _fill_missing_point_descs,
+)
+from supply_infra.db.models.demand_video_expansion import DemandVideoExpansion
+from supply_infra.db.session import get_session
+
+logger = logging.getLogger(__name__)
+_BATCH_SIZE = 500
+
+
+def _list_rows_missing_point_desc(biz_dt: str | None) -> list[dict[str, Any]]:
+    stmt = select(DemandVideoExpansion).where(
+        DemandVideoExpansion.is_delete == 0,
+        or_(
+            DemandVideoExpansion.point_desc.is_(None),
+            DemandVideoExpansion.point_desc == "",
+        ),
+    )
+    if biz_dt:
+        stmt = stmt.where(DemandVideoExpansion.biz_dt == biz_dt)
+    stmt = stmt.order_by(DemandVideoExpansion.id)
+
+    with get_session() as session:
+        rows = session.scalars(stmt).all()
+        return [
+            {
+                "id": int(row.id),
+                "biz_dt": str(row.biz_dt),
+                "video_id": str(row.video_id),
+                "point_type": str(row.point_type),
+                "expanded_text": str(row.expanded_text),
+                "point_desc": row.point_desc,
+            }
+            for row in rows
+        ]
+
+
+def backfill_missing_point_descs(
+    biz_dt: str | None = None,
+    *,
+    dry_run: bool = False,
+) -> dict[str, Any]:
+    rows = _list_rows_missing_point_desc(biz_dt)
+    result: dict[str, Any] = {
+        "biz_dt": biz_dt,
+        "dry_run": dry_run,
+        "missing_total": len(rows),
+        "filled": 0,
+        "still_empty": 0,
+        "updated": 0,
+        "samples": [],
+    }
+    if not rows:
+        return result
+
+    with get_session() as session:
+        _fill_missing_point_descs(rows, session)
+
+    to_update: list[dict[str, Any]] = []
+    for row in rows:
+        if row.get("point_desc"):
+            to_update.append(row)
+            result["filled"] += 1
+            if len(result["samples"]) < 10:
+                result["samples"].append(
+                    {
+                        "id": row["id"],
+                        "video_id": row["video_id"],
+                        "point_type": row["point_type"],
+                        "expanded_text": row["expanded_text"],
+                        "point_desc": row["point_desc"][:80]
+                        if len(str(row["point_desc"])) > 80
+                        else row["point_desc"],
+                    }
+                )
+        else:
+            result["still_empty"] += 1
+
+    if dry_run or not to_update:
+        result["updated"] = 0
+        return result
+
+    with get_session() as session:
+        for i in range(0, len(to_update), _BATCH_SIZE):
+            batch = to_update[i : i + _BATCH_SIZE]
+            for row in batch:
+                session.execute(
+                    update(DemandVideoExpansion)
+                    .where(DemandVideoExpansion.id == int(row["id"]))
+                    .values(point_desc=row["point_desc"])
+                )
+            result["updated"] += len(batch)
+
+    return result
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        description="补全 demand_video_expansion 缺失的 point_desc",
+    )
+    parser.add_argument("--biz-dt", default=None, help="业务日期 YYYYMMDD，默认全表")
+    parser.add_argument("--dry-run", action="store_true", help="仅统计，不写库")
+    parser.add_argument("--json", action="store_true", help="以 JSON 输出结果")
+    args = parser.parse_args(argv)
+
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    )
+
+    result = backfill_missing_point_descs(args.biz_dt, dry_run=bool(args.dry_run))
+
+    if args.json:
+        print(json.dumps(result, ensure_ascii=False, indent=2, default=str))
+    else:
+        print("\n=== point_desc 补全 ===")
+        print(f"biz_dt={result.get('biz_dt') or '全部'}")
+        print(f"dry_run={result.get('dry_run')}")
+        print(f"缺失记录={result.get('missing_total')}")
+        print(f"可补全={result.get('filled')}")
+        print(f"仍为空={result.get('still_empty')}")
+        print(f"已更新={result.get('updated')}")
+        if result.get("samples"):
+            print("\n示例:")
+            for item in result["samples"]:
+                print(
+                    f"  id={item['id']} video={item['video_id']} "
+                    f"type={item['point_type']} text={item['expanded_text']!r} "
+                    f"desc={item['point_desc']!r}"
+                )
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/retry_failed_grade_plan_items.py

@@ -0,0 +1,100 @@
+#!/usr/bin/env python3
+"""重试 demand_grade_plan_group_item 中 status=failed 的分级任务。
+
+用法:
+  .venv/bin/python scripts/retry_failed_grade_plan_items.py
+  .venv/bin/python scripts/retry_failed_grade_plan_items.py --biz-dt 20260721
+  .venv/bin/python scripts/retry_failed_grade_plan_items.py --biz-dt 20260721 --workers 5
+  .venv/bin/python scripts/retry_failed_grade_plan_items.py --biz-dt 20260721 --dry-run
+  .venv/bin/python scripts/retry_failed_grade_plan_items.py --group-id 12 --group-id 15
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import sys
+from pathlib import Path
+
+_ROOT = Path(__file__).resolve().parents[1]
+if str(_ROOT) not in sys.path:
+    sys.path.insert(0, str(_ROOT))
+
+from supply_infra.scheduler.jobs.grade_demand_pool import retry_failed_plan_group_items
+from supply_infra.scheduler.plan_group_batch import MAX_DEMANDS_PER_BATCH
+
+logger = logging.getLogger(__name__)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        description="重试 demand_grade_plan_group_item 中失败的分级任务",
+    )
+    parser.add_argument("--biz-dt", default=None, help="业务日期 YYYYMMDD，默认当天")
+    parser.add_argument("--workers", type=int, default=5, help="并发执行的 plan_group 数")
+    parser.add_argument(
+        "--max-demands-per-batch",
+        type=int,
+        default=MAX_DEMANDS_PER_BATCH,
+        help=f"每个 Agent 子批次最多处理的需求条数，默认 {MAX_DEMANDS_PER_BATCH}",
+    )
+    parser.add_argument(
+        "--group-id",
+        type=int,
+        action="append",
+        dest="group_ids",
+        help="仅重试指定 group_id，可重复传入",
+    )
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="仅列出将要重试的 failed 记录，不实际执行",
+    )
+    parser.add_argument("--json", action="store_true", help="以 JSON 打印结果")
+    args = parser.parse_args(argv)
+
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    )
+
+    result = retry_failed_plan_group_items(
+        args.biz_dt,
+        workers=max(1, int(args.workers)),
+        max_demands_per_batch=max(1, min(int(args.max_demands_per_batch), MAX_DEMANDS_PER_BATCH)),
+        group_ids=args.group_ids,
+        dry_run=bool(args.dry_run),
+    )
+
+    if args.json:
+        print(json.dumps(result, ensure_ascii=False, indent=2, default=str))
+    else:
+        reset = result.get("reset") or {}
+        print("\n=== 失败任务重试 ===")
+        print(f"biz_dt={result.get('biz_dt')}")
+        print(f"dry_run={result.get('dry_run')}")
+        print(f"failed_items={result.get('failed_items', 0)}")
+        print(f"reset_items={reset.get('reset_items', 0)}")
+        print(f"reset_groups={reset.get('reset_groups', 0)}")
+        if reset.get("group_ids"):
+            print(f"group_ids={reset.get('group_ids')}")
+        if not result.get("dry_run"):
+            print(f"graded: {result.get('graded_before')} -> {result.get('graded_after')}")
+            print(f"remaining_failed={result.get('remaining_failed', 0)}")
+            print(f"group_status={result.get('group_status')}")
+            print(f"success={result.get('success')}")
+        elif reset.get("items"):
+            print("\n待重试明细:")
+            for item in reset["items"][:20]:
+                print(
+                    f"  item_id={item['item_id']} group_id={item['group_id']} "
+                    f"demand={item['demand_name']!r}"
+                )
+            if len(reset["items"]) > 20:
+                print(f"  ... 另有 {len(reset['items']) - 20} 条")
+
+    return 0 if result.get("success") else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

supply_agent/agent/core.py

@@ -58,7 +58,7 @@ class Agent:
         skills: SkillRegistry | None = None,
         max_iterations: int | None = None,
         temperature: float | None = None,
-        reasoning_effort: str | None = "medium",
+        reasoning_effort: str | None = None,
         logger: AgentLogger | None = None,
     ) -> None:
         self.name = name

supply_agent/config.py

@@ -22,7 +22,7 @@ class Settings(BaseSettings):
     # OpenRouter
     openrouter_api_key: str = Field(..., alias="OPENROUTER_API_KEY")
     openrouter_model: str = Field(
-        default="anthropic/claude-sonnet-5",
+        default="google/gemini-2.5-flash",
         alias="OPENROUTER_MODEL",
     )
     openrouter_base_url: str = Field(

supply_agent/llm/client.py

@@ -20,7 +20,7 @@ class LLMClient:
         settings: Settings,
         logger: AgentLogger | None = None,
         *,
-        reasoning_effort: str | None = "medium",
+        reasoning_effort: str | None = None,
     ) -> None:
         self.settings = settings
         self.model = settings.openrouter_model

supply_infra/db/models/__init__.py

@@ -11,6 +11,10 @@ from supply_infra.db.models.demand_grade_plan import (
     DemandGradePlanGroupItem,
 )
 from supply_infra.db.models.demand_popularity_stats import DemandPopularityStats
+from supply_infra.db.models.demand_video_expansion import (
+    DemandVideoExpansion,
+    DemandVideoExpansionRun,
+)
 from supply_infra.db.models.generated_demand import GeneratedDemand
 from supply_infra.db.models.global_tree_category import GlobalTreeCategory
 from supply_infra.db.models.global_tree_element import GlobalTreeElement
@@ -30,6 +34,8 @@ __all__ = [
     "DemandGradePlanGroup",
     "DemandGradePlanGroupItem",
     "DemandPopularityStats",
+    "DemandVideoExpansion",
+    "DemandVideoExpansionRun",
     "GeneratedDemand",
     "GlobalTreeCategory",
     "GlobalTreeElement",

supply_infra/db/models/demand_video_expansion.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import BigInteger, Index, Integer, String, Text, UniqueConstraint, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from supply_infra.db.base import Base
+
+
+class DemandVideoExpansion(Base):
+    """S/A 需求关联视频点位拓展结果。"""
+
+    __tablename__ = "demand_video_expansion"
+    __table_args__ = (
+        UniqueConstraint(
+            "biz_dt",
+            "source_demand_grade_id",
+            "expanded_text",
+            "video_id",
+            name="uk_demand_video_expansion",
+        ),
+        Index("idx_dve_biz_dt", "biz_dt"),
+        Index("idx_dve_source_grade", "source_demand_grade_id"),
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    biz_dt: Mapped[str] = mapped_column(String(32), nullable=False, comment="业务日 YYYYMMDD")
+    run_id: Mapped[str] = mapped_column(String(64), nullable=False, comment="任务批次 run_id")
+    source_demand_grade_id: Mapped[int] = mapped_column(
+        BigInteger, nullable=False, comment="来源 demand_grade.id"
+    )
+    source_demand_name: Mapped[str] = mapped_column(
+        String(256), nullable=False, comment="来源需求名"
+    )
+    source_grade: Mapped[str] = mapped_column(String(4), nullable=False, comment="来源等级 S/A")
+    expanded_text: Mapped[str] = mapped_column(
+        String(512), nullable=False, comment="拓展需求文本（来自 point_data）"
+    )
+    point_type: Mapped[str] = mapped_column(
+        String(32), nullable=False, comment="点类型：inspiration / purpose / key"
+    )
+    point_desc: Mapped[str | None] = mapped_column(Text, nullable=True, comment="点位描述快照")
+    video_id: Mapped[str] = mapped_column(String(64), nullable=False, comment="来源视频 id")
+    reason: Mapped[str] = mapped_column(Text, nullable=False, comment="相近判断依据")
+    is_delete: Mapped[int] = mapped_column(
+        Integer, default=0, nullable=False, comment="是否删除 0-正常 1-删除"
+    )
+    create_time: Mapped[datetime] = mapped_column(
+        nullable=False,
+        server_default=func.now(),
+        comment="创建时间",
+    )
+    update_time: Mapped[datetime] = mapped_column(
+        nullable=False,
+        server_default=func.now(),
+        onupdate=func.now(),
+        comment="更新时间",
+    )
+
+
+class DemandVideoExpansionRun(Base):
+    """记录每个需求是否已完成拓展判断（含零结果），用于幂等跳过。"""
+
+    __tablename__ = "demand_video_expansion_run"
+    __table_args__ = (
+        UniqueConstraint(
+            "biz_dt",
+            "source_demand_grade_id",
+            name="uk_demand_video_expansion_run",
+        ),
+        Index("idx_dver_biz_dt", "biz_dt"),
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    biz_dt: Mapped[str] = mapped_column(String(32), nullable=False, comment="业务日 YYYYMMDD")
+    run_id: Mapped[str] = mapped_column(String(64), nullable=False, comment="任务批次 run_id")
+    source_demand_grade_id: Mapped[int] = mapped_column(
+        BigInteger, nullable=False, comment="来源 demand_grade.id"
+    )
+    saved_count: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, comment="落库拓展条数"
+    )
+    status: Mapped[str] = mapped_column(
+        String(16), nullable=False, default="finished", comment="finished / failed"
+    )
+    error_message: Mapped[str | None] = mapped_column(
+        Text, nullable=True, comment="失败原因"
+    )
+    create_time: Mapped[datetime] = mapped_column(
+        nullable=False,
+        server_default=func.now(),
+        comment="创建时间",
+    )
+    update_time: Mapped[datetime] = mapped_column(
+        nullable=False,
+        server_default=func.now(),
+        onupdate=func.now(),
+        comment="更新时间",
+    )

supply_infra/db/repositories/__init__.py

@@ -18,6 +18,10 @@ from supply_infra.db.repositories.demand_grade_plan_repo import DemandGradePlanR
 from supply_infra.db.repositories.demand_popularity_stats_repo import (
     DemandPopularityStatsRepository,
 )
+from supply_infra.db.repositories.demand_video_expansion_repo import (
+    DemandVideoExpansionRepository,
+    DemandVideoExpansionRunRepository,
+)
 from supply_infra.db.repositories.generated_demand_repo import GeneratedDemandRepository
 from supply_infra.db.repositories.global_tree_category_repo import GlobalTreeCategoryRepository
 from supply_infra.db.repositories.global_tree_element_repo import GlobalTreeElementRepository
@@ -42,6 +46,8 @@ __all__ = [
     "DemandGradeRepository",
     "DemandGradePlanRepository",
     "DemandPopularityStatsRepository",
+    "DemandVideoExpansionRepository",
+    "DemandVideoExpansionRunRepository",
     "GeneratedDemandRepository",
     "GlobalTreeCategoryRepository",
     "GlobalTreeElementRepository",

supply_infra/db/repositories/demand_grade_plan_repo.py

@@ -14,7 +14,6 @@ from supply_infra.db.models.demand_grade_plan import (
 )
 from supply_infra.db.repositories.base import BaseRepository
 from supply_infra.db.repositories.demand_grade_repo import DemandGradeRepository
-from supply_infra.scheduler.plan_group_batch import resolve_demands_for_category_ids
 
 
 class DemandGradePlanRepository(BaseRepository[DemandGradePlan]):
@@ -190,6 +189,8 @@ class DemandGradePlanRepository(BaseRepository[DemandGradePlan]):
             return 0
 
         graded = graded_names or set()
+        from supply_infra.scheduler.plan_group_batch import resolve_demands_for_category_ids
+
         demands = resolve_demands_for_category_ids(biz_dt, category_ids)
         created = 0
         for sort_order, demand in enumerate(demands, start=1):
@@ -274,3 +275,72 @@ class DemandGradePlanRepository(BaseRepository[DemandGradePlan]):
             .where(DemandGradePlanGroupItem.id.in_([int(item_id) for item_id in item_ids]))
             .values(**values)
         )
+
+    def list_failed_group_items(
+        self,
+        biz_dt: str | None = None,
+        *,
+        group_ids: list[int] | None = None,
+    ) -> list[dict[str, Any]]:
+        """返回 status=failed 的组内需求明细。"""
+        stmt = select(DemandGradePlanGroupItem).where(DemandGradePlanGroupItem.status == "failed")
+        if biz_dt:
+            stmt = stmt.where(DemandGradePlanGroupItem.biz_dt == biz_dt)
+        if group_ids:
+            stmt = stmt.where(
+                DemandGradePlanGroupItem.group_id.in_([int(group_id) for group_id in group_ids])
+            )
+        stmt = stmt.order_by(
+            DemandGradePlanGroupItem.biz_dt,
+            DemandGradePlanGroupItem.group_id,
+            DemandGradePlanGroupItem.sort_order,
+            DemandGradePlanGroupItem.id,
+        )
+        rows = self.session.scalars(stmt).all()
+        return [
+            {
+                "item_id": int(row.id),
+                "group_id": int(row.group_id),
+                "biz_dt": str(row.biz_dt),
+                "pool_id": int(row.pool_id),
+                "demand_name": str(row.demand_name),
+                "error_message": row.error_message,
+            }
+            for row in rows
+        ]
+
+    def reset_failed_items_to_pending(
+        self,
+        biz_dt: str | None = None,
+        *,
+        group_ids: list[int] | None = None,
+    ) -> dict[str, Any]:
+        """将 failed 明细重置为 pending，并将所属计划组重置为 pending 以便重新领取。"""
+        failed_rows = self.list_failed_group_items(biz_dt, group_ids=group_ids)
+        if not failed_rows:
+            return {"reset_items": 0, "reset_groups": 0, "group_ids": [], "items": []}
+
+        item_ids = [int(row["item_id"]) for row in failed_rows]
+        affected_group_ids = sorted({int(row["group_id"]) for row in failed_rows})
+
+        self.session.execute(
+            update(DemandGradePlanGroupItem)
+            .where(DemandGradePlanGroupItem.id.in_(item_ids))
+            .values(status="pending", error_message=None, finished_at=None)
+        )
+        self.session.execute(
+            update(DemandGradePlanGroup)
+            .where(DemandGradePlanGroup.id.in_(affected_group_ids))
+            .values(
+                status="pending",
+                error_message=None,
+                started_at=None,
+                finished_at=None,
+            )
+        )
+        return {
+            "reset_items": len(item_ids),
+            "reset_groups": len(affected_group_ids),
+            "group_ids": affected_group_ids,
+            "items": failed_rows,
+        }

supply_infra/db/repositories/demand_grade_repo.py

@@ -54,6 +54,22 @@ class DemandGradeRepository(BaseRepository[DemandGrade]):
         )
         return list(self.session.scalars(stmt).all())
 
+    def list_by_biz_dt_and_grades(
+        self,
+        biz_dt: str,
+        grades: Iterable[str] = ("S", "A"),
+    ) -> list[DemandGrade]:
+        """返回指定业务日、指定等级的分级结果，按等级、需求名排序。"""
+        grade_list = [g for g in grades if g]
+        if not grade_list:
+            return []
+        stmt = (
+            select(DemandGrade)
+            .where(DemandGrade.biz_dt == biz_dt, DemandGrade.grade.in_(grade_list))
+            .order_by(DemandGrade.grade, DemandGrade.demand_name)
+        )
+        return list(self.session.scalars(stmt).all())
+
     def get_latest_biz_dt(self) -> str | None:
         """返回 demand_grade 中最新的业务日期。"""
         stmt = select(func.max(DemandGrade.biz_dt))

supply_infra/db/repositories/demand_video_expansion_repo.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+from collections.abc import Iterable
+
+from sqlalchemy import select
+from sqlalchemy.dialects.mysql import insert
+
+from supply_infra.db.models.demand_video_expansion import (
+    DemandVideoExpansion,
+    DemandVideoExpansionRun,
+)
+from supply_infra.db.repositories.base import BaseRepository
+
+_BATCH_SIZE = 500
+
+
+class DemandVideoExpansionRepository(BaseRepository[DemandVideoExpansion]):
+    """需求视频点位拓展结果 repository。"""
+
+    model = DemandVideoExpansion
+
+    def list_by_demand_grade(
+        self, biz_dt: str, source_demand_grade_id: int
+    ) -> list[DemandVideoExpansion]:
+        """按业务日与来源需求 id 查询拓展结果（未删除）。"""
+        stmt = (
+            select(DemandVideoExpansion)
+            .where(
+                DemandVideoExpansion.biz_dt == biz_dt,
+                DemandVideoExpansion.source_demand_grade_id == int(source_demand_grade_id),
+                DemandVideoExpansion.is_delete == 0,
+            )
+            .order_by(
+                DemandVideoExpansion.video_id,
+                DemandVideoExpansion.point_type,
+                DemandVideoExpansion.id,
+            )
+        )
+        return list(self.session.scalars(stmt).all())
+
+    def bulk_upsert(self, rows: list[dict]) -> int:
+        """按唯一键批量 upsert，冲突时更新 reason / point_desc。"""
+        if not rows:
+            return 0
+
+        affected = 0
+        for i in range(0, len(rows), _BATCH_SIZE):
+            batch = rows[i : i + _BATCH_SIZE]
+            stmt = insert(DemandVideoExpansion).values(batch)
+            stmt = stmt.on_duplicate_key_update(
+                reason=stmt.inserted.reason,
+                point_desc=stmt.inserted.point_desc,
+                run_id=stmt.inserted.run_id,
+                is_delete=0,
+            )
+            result = self.session.execute(stmt)
+            affected += result.rowcount or 0
+        return affected
+
+
+class DemandVideoExpansionRunRepository(BaseRepository[DemandVideoExpansionRun]):
+    """拓展任务执行记录 repository。"""
+
+    model = DemandVideoExpansionRun
+
+    def list_finished_grade_ids(self, biz_dt: str) -> set[int]:
+        """返回指定业务日已成功完成拓展判断的 demand_grade.id 集合。"""
+        stmt = select(DemandVideoExpansionRun.source_demand_grade_id).where(
+            DemandVideoExpansionRun.biz_dt == biz_dt,
+            DemandVideoExpansionRun.status == "finished",
+        )
+        return {int(v) for v in self.session.scalars(stmt).all() if v is not None}
+
+    def upsert_run(
+        self,
+        *,
+        biz_dt: str,
+        run_id: str,
+        source_demand_grade_id: int,
+        saved_count: int,
+        status: str = "finished",
+        error_message: str | None = None,
+    ) -> None:
+        """记录单条需求的拓展判断完成状态。"""
+        row = {
+            "biz_dt": biz_dt,
+            "run_id": run_id,
+            "source_demand_grade_id": int(source_demand_grade_id),
+            "saved_count": int(saved_count),
+            "status": status,
+            "error_message": error_message,
+        }
+        stmt = insert(DemandVideoExpansionRun).values(row)
+        stmt = stmt.on_duplicate_key_update(
+            run_id=stmt.inserted.run_id,
+            saved_count=stmt.inserted.saved_count,
+            status=stmt.inserted.status,
+            error_message=stmt.inserted.error_message,
+        )
+        self.session.execute(stmt)
+
+    def list_by_biz_dt(self, biz_dt: str) -> list[DemandVideoExpansionRun]:
+        stmt = (
+            select(DemandVideoExpansionRun)
+            .where(DemandVideoExpansionRun.biz_dt == biz_dt)
+            .order_by(DemandVideoExpansionRun.source_demand_grade_id)
+        )
+        return list(self.session.scalars(stmt).all())
+
+    def get_by_demand_grade(
+        self, biz_dt: str, source_demand_grade_id: int
+    ) -> DemandVideoExpansionRun | None:
+        stmt = select(DemandVideoExpansionRun).where(
+            DemandVideoExpansionRun.biz_dt == biz_dt,
+            DemandVideoExpansionRun.source_demand_grade_id == int(source_demand_grade_id),
+        )
+        return self.session.scalars(stmt).first()

supply_infra/scheduler/jobs/expand_demand_from_video_points.py

@@ -0,0 +1,335 @@
+"""
+从 S/A 级需求关联视频中挖掘拓展需求。
+
+任务层负责查库与组装上下文；Agent 仅做语义判断与落库。
+"""
+from __future__ import annotations
+
+import json
+import logging
+import uuid
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from datetime import datetime
+from typing import Any
+from zoneinfo import ZoneInfo
+
+from agents.demand_video_expand_agent.run import (
+    DemandExpandContext,
+    VideoPoint,
+    extract_saved_count,
+    judge_demand_expansion,
+)
+from supply_infra.config import get_infra_settings
+from supply_infra.db.models.demand_grade import DemandGrade
+from supply_infra.db.repositories.demand_grade_repo import DemandGradeRepository
+from supply_infra.db.repositories.demand_video_expansion_repo import (
+    DemandVideoExpansionRunRepository,
+)
+from supply_infra.db.repositories.multi_demand_video_point_repo import (
+    MultiDemandVideoPointRepository,
+)
+from supply_infra.db.session import get_session
+
+logger = logging.getLogger(__name__)
+
+_DEFAULT_WORKERS = 5
+
+
+def _resolve_biz_dt(biz_dt: str | None) -> str:
+    if biz_dt:
+        text = str(biz_dt).strip()
+        if len(text) == 8 and text.isdigit():
+            return text
+        raise ValueError(f"biz_dt 格式无效，应为 YYYYMMDD: {biz_dt!r}")
+
+    with get_session() as session:
+        latest = DemandGradeRepository(session).get_latest_biz_dt()
+    if latest:
+        return str(latest)
+
+    timezone = ZoneInfo(get_infra_settings().scheduler_timezone)
+    return datetime.now(timezone).strftime("%Y%m%d")
+
+
+def _parse_video_ids(raw: Any) -> list[str]:
+    if raw is None:
+        return []
+    items: list[Any]
+    if isinstance(raw, str):
+        text = raw.strip()
+        if not text:
+            return []
+        try:
+            parsed = json.loads(text)
+            items = list(parsed) if isinstance(parsed, list) else [text]
+        except (ValueError, TypeError):
+            items = [part.strip() for part in text.split(",") if part.strip()]
+    elif isinstance(raw, (list, tuple)):
+        items = list(raw)
+    else:
+        return []
+    out: list[str] = []
+    seen: set[str] = set()
+    for item in items:
+        vid = str(item).strip() if item is not None else ""
+        if not vid or vid in seen:
+            continue
+        seen.add(vid)
+        out.append(vid)
+    return out
+
+
+def _to_float(value: Any) -> float | None:
+    if value is None:
+        return None
+    return float(value)
+
+
+def load_expand_contexts(
+    session,
+    biz_dt: str,
+    *,
+    run_id: str,
+    skip_finished: bool = True,
+) -> tuple[list[DemandExpandContext], dict[str, int]]:
+    """加载待拓展判断的需求上下文，并返回跳过统计。"""
+    stats = {
+        "total_sa": 0,
+        "skipped_no_video": 0,
+        "skipped_no_points": 0,
+        "skipped_already_done": 0,
+    }
+
+    grades = DemandGradeRepository(session).list_by_biz_dt_and_grades(biz_dt, ("S", "A"))
+    stats["total_sa"] = len(grades)
+
+    finished_ids: set[int] = set()
+    if skip_finished:
+        finished_ids = DemandVideoExpansionRunRepository(session).list_finished_grade_ids(
+            biz_dt
+        )
+
+    rows_with_video: list[tuple[DemandGrade, list[str]]] = []
+    all_video_ids: set[str] = set()
+
+    for row in grades:
+        if int(row.id) in finished_ids:
+            stats["skipped_already_done"] += 1
+            continue
+
+        video_ids = _parse_video_ids(row.video_list)
+        if not video_ids:
+            stats["skipped_no_video"] += 1
+            continue
+
+        rows_with_video.append((row, video_ids))
+        all_video_ids.update(video_ids)
+
+    points_by_vid = MultiDemandVideoPointRepository(session).list_by_video_ids(all_video_ids)
+
+    contexts: list[DemandExpandContext] = []
+    for row, video_ids in rows_with_video:
+        points: list[VideoPoint] = []
+        for vid in video_ids:
+            for point in points_by_vid.get(vid, []):
+                points.append(
+                    VideoPoint(
+                        video_id=vid,
+                        point_type=str(point.get("point_type") or ""),
+                        point_data=point.get("point_data"),
+                        point_desc=point.get("point_desc"),
+                    )
+                )
+        if not points:
+            stats["skipped_no_points"] += 1
+            continue
+
+        contexts.append(
+            DemandExpandContext(
+                biz_dt=biz_dt,
+                run_id=run_id,
+                demand_grade_id=int(row.id),
+                demand_name=str(row.demand_name),
+                grade=str(row.grade),
+                score=_to_float(row.score),
+                video_ids=video_ids,
+                points=points,
+            )
+        )
+
+    return contexts, stats
+
+
+def _record_run(
+    *,
+    biz_dt: str,
+    run_id: str,
+    source_demand_grade_id: int,
+    saved_count: int,
+    status: str,
+    error_message: str | None = None,
+) -> None:
+    with get_session() as session:
+        DemandVideoExpansionRunRepository(session).upsert_run(
+            biz_dt=biz_dt,
+            run_id=run_id,
+            source_demand_grade_id=source_demand_grade_id,
+            saved_count=saved_count,
+            status=status,
+            error_message=error_message,
+        )
+
+
+def _process_single_expand(
+    ctx: DemandExpandContext,
+    *,
+    biz_dt: str,
+    run_id: str,
+) -> dict[str, Any]:
+    """并发 worker：对单个需求执行拓展判断并落库执行记录。"""
+    try:
+        agent_result = judge_demand_expansion(ctx)
+        saved_count = extract_saved_count(agent_result)
+        _record_run(
+            biz_dt=biz_dt,
+            run_id=run_id,
+            source_demand_grade_id=ctx.demand_grade_id,
+            saved_count=saved_count,
+            status="finished",
+        )
+        logger.info(
+            "expand demand done: grade_id=%s demand=%s saved=%d iterations=%d",
+            ctx.demand_grade_id,
+            ctx.demand_name,
+            saved_count,
+            agent_result.iterations,
+        )
+        return {
+            "success": True,
+            "demand_grade_id": ctx.demand_grade_id,
+            "demand_name": ctx.demand_name,
+            "saved_count": saved_count,
+            "iterations": agent_result.iterations,
+        }
+    except Exception as exc:
+        error_text = str(exc)
+        _record_run(
+            biz_dt=biz_dt,
+            run_id=run_id,
+            source_demand_grade_id=ctx.demand_grade_id,
+            saved_count=0,
+            status="failed",
+            error_message=error_text,
+        )
+        logger.exception(
+            "expand demand failed: grade_id=%s demand=%s",
+            ctx.demand_grade_id,
+            ctx.demand_name,
+        )
+        return {
+            "success": False,
+            "demand_grade_id": ctx.demand_grade_id,
+            "demand_name": ctx.demand_name,
+            "error": error_text,
+        }
+
+
+def expand_demand_from_video_points(
+    biz_dt: str | None = None,
+    *,
+    skip_finished: bool = True,
+    workers: int = _DEFAULT_WORKERS,
+) -> dict[str, Any]:
+    """
+    对指定业务日的 S/A 需求执行视频点位拓展判断。
+
+    程序负责查需求与点位；无视频或无点位则跳过；有数据则并发调用 Agent。
+    """
+    started_at = datetime.now()
+    run_id = uuid.uuid4().hex
+
+    try:
+        resolved_biz_dt = _resolve_biz_dt(biz_dt)
+    except Exception as exc:
+        logger.exception("expand_demand_from_video_points preflight failed")
+        return {
+            "success": False,
+            "error": str(exc),
+            "started_at": started_at.isoformat(),
+            "finished_at": datetime.now().isoformat(),
+        }
+
+    with get_session() as session:
+        contexts, preload_stats = load_expand_contexts(
+            session,
+            resolved_biz_dt,
+            run_id=run_id,
+            skip_finished=skip_finished,
+        )
+
+    logger.info(
+        "expand_demand_from_video_points start: biz_dt=%s run_id=%s workers=%s pending=%s",
+        resolved_biz_dt,
+        run_id,
+        workers,
+        len(contexts),
+    )
+
+    result: dict[str, Any] = {
+        "success": True,
+        "run_id": run_id,
+        "biz_dt": resolved_biz_dt,
+        "started_at": started_at.isoformat(),
+        "workers": 0,
+        **preload_stats,
+        "processed": 0,
+        "saved_total": 0,
+        "failed": 0,
+        "errors": [],
+    }
+
+    if not contexts:
+        finished_at = datetime.now()
+        result["finished_at"] = finished_at.isoformat()
+        result["duration_seconds"] = round((finished_at - started_at).total_seconds(), 2)
+        logger.info("expand_demand_from_video_points finished: %s", result)
+        return result
+
+    worker_count = max(1, min(int(workers), len(contexts)))
+    result["workers"] = worker_count
+
+    with ThreadPoolExecutor(max_workers=worker_count) as executor:
+        futures = [
+            executor.submit(_process_single_expand, ctx, biz_dt=resolved_biz_dt, run_id=run_id)
+            for ctx in contexts
+        ]
+        for future in as_completed(futures):
+            try:
+                item_result = future.result()
+            except Exception as exc:
+                logger.exception("expand demand worker 出现未捕获错误: biz_dt=%s", resolved_biz_dt)
+                result["failed"] += 1
+                result["errors"].append({"error": str(exc)})
+                continue
+
+            result["processed"] += 1
+            if item_result.get("success"):
+                result["saved_total"] += int(item_result.get("saved_count") or 0)
+                continue
+
+            result["failed"] += 1
+            result["errors"].append(
+                {
+                    "demand_grade_id": item_result.get("demand_grade_id"),
+                    "demand_name": item_result.get("demand_name"),
+                    "error": item_result.get("error"),
+                }
+            )
+
+    finished_at = datetime.now()
+    result["finished_at"] = finished_at.isoformat()
+    result["duration_seconds"] = round((finished_at - started_at).total_seconds(), 2)
+    result["success"] = result["failed"] == 0
+
+    logger.info("expand_demand_from_video_points finished: %s", result)
+    return result

supply_infra/scheduler/jobs/grade_demand_pool.py

@@ -259,6 +259,104 @@ def _grade_demand_pool_impl(
     return result
 
 
+def retry_failed_plan_group_items(
+    biz_dt: str | None = None,
+    *,
+    workers: int = _DEFAULT_WORKERS,
+    max_demands_per_batch: int = MAX_DEMANDS_PER_BATCH,
+    group_ids: list[int] | None = None,
+    dry_run: bool = False,
+) -> dict[str, Any]:
+    """将 demand_grade_plan_group_item 中 failed 记录重置后重新执行分级。"""
+    resolved_biz_dt = _resolve_biz_dt(biz_dt)
+    try:
+        with get_session() as session:
+            failed_items = DemandGradePlanRepository(session).list_failed_group_items(
+                resolved_biz_dt,
+                group_ids=group_ids,
+            )
+
+        if not failed_items:
+            with get_session() as session:
+                snapshot = DemandGradePlanRepository(session).get_execution_snapshot(resolved_biz_dt)
+            return {
+                "success": True,
+                "biz_dt": resolved_biz_dt,
+                "dry_run": dry_run,
+                "failed_items": 0,
+                "reset": {"reset_items": 0, "reset_groups": 0, "group_ids": [], "items": []},
+                "execution": None,
+                "final_snapshot": snapshot,
+                "run_at": datetime.now().isoformat(),
+            }
+
+        if dry_run:
+            affected_group_ids = sorted({int(item["group_id"]) for item in failed_items})
+            return {
+                "success": True,
+                "biz_dt": resolved_biz_dt,
+                "dry_run": True,
+                "failed_items": len(failed_items),
+                "reset": {
+                    "reset_items": len(failed_items),
+                    "reset_groups": len(affected_group_ids),
+                    "group_ids": affected_group_ids,
+                    "items": failed_items,
+                },
+                "execution": None,
+                "run_at": datetime.now().isoformat(),
+            }
+
+        with get_session() as session:
+            reset_result = DemandGradePlanRepository(session).reset_failed_items_to_pending(
+                resolved_biz_dt,
+                group_ids=group_ids,
+            )
+
+        with get_session() as session:
+            graded_before = DemandGradeRepository(session).count_by_biz_dt(resolved_biz_dt)
+
+        plan_execution = execute_plan_tasks_until_complete(
+            resolved_biz_dt,
+            workers=max(1, int(workers)),
+            max_demands_per_batch=max(1, min(int(max_demands_per_batch), MAX_DEMANDS_PER_BATCH)),
+        )
+
+        with get_session() as session:
+            graded_after = DemandGradeRepository(session).count_by_biz_dt(resolved_biz_dt)
+            remaining_failed = DemandGradePlanRepository(session).list_failed_group_items(
+                resolved_biz_dt,
+                group_ids=group_ids,
+            )
+
+        final_snapshot = plan_execution["final_snapshot"]
+        return {
+            "success": len(remaining_failed) == 0,
+            "biz_dt": resolved_biz_dt,
+            "dry_run": False,
+            "failed_items": len(failed_items),
+            "reset": reset_result,
+            "graded_before": graded_before,
+            "graded_after": graded_after,
+            "remaining_failed": len(remaining_failed),
+            "execution": plan_execution,
+            "final_snapshot": final_snapshot,
+            "group_status": final_snapshot["group_status"],
+            "run_at": datetime.now().isoformat(),
+        }
+    except Exception as exc:
+        logger.exception(
+            "重试 failed plan group items 发生未捕获错误: biz_dt=%s",
+            resolved_biz_dt,
+        )
+        return {
+            "success": False,
+            "biz_dt": resolved_biz_dt,
+            "error": str(exc),
+            "run_at": datetime.now().isoformat(),
+        }
+
+
 def grade_demand_pool(
     biz_dt: str | None = None,
     *,

supply_infra/scheduler/jobs/run_supply_pipeline.py

@@ -4,6 +4,7 @@
 1. ODPS → MySQL 全局树同步（T-1 分区）
 2. ODPS → MySQL 策略需求池同步（当天 biz_dt）
 3. 需求池分级评估（同上 biz_dt）
+4. S/A 需求视频点位拓展判断
 
 各子步骤内部已做去重（INSERT IGNORE、diff 同步、跳过已分级词等）；
 本文件额外用进程内锁防止同一轮次并发重入，并隔离各步骤异常：前一步失败时记录
@@ -23,6 +24,9 @@ from supply_infra.scheduler.constants import (
     SUPPLY_PIPELINE_JOB_NAME,
 )
 from supply_infra.scheduler.job_execution import JobExecutionRecorder, record_skipped
+from supply_infra.scheduler.jobs.expand_demand_from_video_points import (
+    expand_demand_from_video_points,
+)
 from supply_infra.scheduler.jobs.grade_demand_pool import grade_demand_pool
 from supply_infra.scheduler.jobs.sync_global_tree_odps_to_mysql import sync_global_tree_odps_to_mysql
 from supply_infra.scheduler.jobs.sync_multi_demand_pool_odps_to_mysql import (
@@ -90,7 +94,7 @@ def _preflight_failure_result(biz_dt: str | None, exc: Exception) -> dict[str, A
 
 def run_supply_pipeline(biz_dt: str | None = None) -> dict[str, Any]:
     """
-    按顺序执行全局树同步 → 需求池同步 → 需求分级。
+    按顺序执行全局树同步 → 需求池同步 → 需求分级 → 视频点位拓展。
 
     Args:
         biz_dt: 业务日 YYYYMMDD；省略则取当天。
@@ -159,6 +163,13 @@ def run_supply_pipeline(biz_dt: str | None = None) -> dict[str, Any]:
                 "grade",
                 lambda: grade_demand_pool(biz_dt=resolved_biz_dt),
             ),
+            (
+                "expand_video_points",
+                lambda: expand_demand_from_video_points(
+                    biz_dt=resolved_biz_dt,
+                    workers=5,
+                ),
+            ),
         ]
         for step_name, action in steps:
             payload, step_success, step_error = _run_step(step_name, action)

supply_infra/scheduler/plan_group_batch.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 
 from typing import Any
 
-from agents.demand_grade_agent.tools.demand_priority import build_demand_priority_index
 from supply_infra.db.repositories.demand_belong_category_repo import DemandBelongCategoryRepository
 from supply_infra.db.repositories.demand_belong_pool_rel_repo import DemandBelongPoolRelRepository
 from supply_infra.db.repositories.multi_demand_pool_di_repo import MultiDemandPoolDiRepository
@@ -62,6 +61,8 @@ def resolve_demands_for_category_ids(
         pool_ids = sorted({pool_id for values in pool_ids_by_belong.values() for pool_id in values})
         pool_repo = MultiDemandPoolDiRepository(session)
         pool_rows = pool_repo.get_by_ids(pool_ids)
+        from agents.demand_grade_agent.tools.demand_priority import build_demand_priority_index
+
         priority_index = build_demand_priority_index(pool_repo.list_by_biz_dt(biz_dt))
         candidates: list[dict[str, Any]] = []
         for row in pool_rows:

web/src/components/DemandPathPanel.vue

@@ -1,7 +1,7 @@
 <script setup lang="ts">
 import { computed, ref, watch } from 'vue'
 import { fetchDemandGradeVideos } from '../api/demand'
-import type { DemandGradeItem, DemandVideoItem } from '../types/demand'
+import type { DemandExpansionPoint, DemandGradeItem, DemandVideoItem } from '../types/demand'
 import { gradeRank, parseStrategies } from '../types/demand'
 
 const props = defineProps<{
@@ -41,9 +41,12 @@ const selectedVideo = computed(
   () => videos.value.find((v) => v.vid === selectedVid.value) ?? null,
 )
 
+const hasVideoInstances = computed(() => videos.value.length > 0)
+
 interface TopicPoint {
   title: string
-  description: string
+  pointDesc: string
+  reason: string
 }
 
 interface TopicSection {
@@ -53,50 +56,49 @@ interface TopicSection {
   items: TopicPoint[]
 }
 
-function parsePointsJson(raw: string | null | undefined): TopicPoint[] {
-  if (!raw) return []
-  try {
-    const parsed = JSON.parse(raw)
-    const list = Array.isArray(parsed) ? parsed : [parsed]
-    return list
-      .filter((item): item is Record<string, unknown> => !!item && typeof item === 'object')
-      .map((item) => ({
-        title: String(item['点'] ?? '').trim(),
-        description: String(item['点描述'] ?? '').trim(),
-      }))
-      .filter((item) => item.title || item.description)
-  } catch {
-    return []
+const POINT_TYPE_LABELS: Record<string, string> = {
+  purpose: '目的点',
+  key: '关键点',
+  inspiration: '灵感点',
+}
+
+const POINT_TYPE_ORDER = ['purpose', 'key', 'inspiration'] as const
+
+function buildTopicSections(points: DemandExpansionPoint[]): TopicSection[] | null {
+  if (!points.length) return null
+
+  const grouped = new Map<string, TopicPoint[]>()
+  for (const point of points) {
+    const key = point.point_type || 'other'
+    const items = grouped.get(key) ?? []
+    items.push({
+      title: point.expanded_text?.trim() || '未命名',
+      pointDesc: point.point_desc?.trim() || '',
+      reason: point.reason?.trim() || '',
+    })
+    grouped.set(key, items)
   }
+
+  const sections: TopicSection[] = []
+  for (const key of POINT_TYPE_ORDER) {
+    const items = grouped.get(key)
+    if (!items?.length) continue
+    sections.push({
+      key,
+      label: POINT_TYPE_LABELS[key] ?? key,
+      accent: key === 'purpose' ? 'purpose' : key === 'key' ? 'key' : 'inspiration',
+      items,
+    })
+  }
+
+  if (!sections.length) return null
+  return sections
 }
 
 const topicSections = computed((): TopicSection[] | null => {
   const video = selectedVideo.value
   if (!video) return null
-
-  const sections: TopicSection[] = [
-    {
-      key: 'inspiration',
-      label: '灵感点',
-      accent: 'inspiration',
-      items: parsePointsJson(video.inspiration_points_json),
-    },
-    {
-      key: 'purpose',
-      label: '目的点',
-      accent: 'purpose',
-      items: parsePointsJson(video.purpose_points_json),
-    },
-    {
-      key: 'key',
-      label: '关键点',
-      accent: 'key',
-      items: parsePointsJson(video.key_points_json),
-    },
-  ]
-
-  if (sections.every((s) => s.items.length === 0)) return null
-  return sections
+  return buildTopicSections(video.expansion_points ?? [])
 })
 
 watch(
@@ -235,85 +237,86 @@ function selectVideo(video: DemandVideoItem) {
     </section>
 
     <!-- 需求词 → 视频 -->
-    <div class="arrow-col" aria-hidden="true">
-      <div v-if="selectedDemand" class="arrow arrow-green">
-        <span class="arrow-line" />
-        <span class="arrow-head">▶</span>
+    <template v-if="hasVideoInstances">
+      <div class="arrow-col" aria-hidden="true">
+        <div v-if="selectedDemand" class="arrow arrow-green">
+          <span class="arrow-line" />
+          <span class="arrow-head">▶</span>
+        </div>
       </div>
-    </div>
 
-    <section class="col col-video">
-      <h3 class="col-title">真实视频实例</h3>
-      <template v-if="selectedDemand">
-        <div class="detail-meta">
-          <span class="grade-badge" :class="gradeClass(selectedDemand.grade)">
-            {{ selectedDemand.grade || '—' }}
-          </span>
-          <div v-if="selectedStrategies.length" class="strategy-tags">
-            <span v-for="s in selectedStrategies" :key="s" class="strategy-tag">{{ s }}</span>
+      <section class="col col-video">
+        <h3 class="col-title">真实视频实例</h3>
+        <template v-if="selectedDemand">
+          <div class="detail-meta">
+            <span class="grade-badge" :class="gradeClass(selectedDemand.grade)">
+              {{ selectedDemand.grade || '—' }}
+            </span>
+            <div v-if="selectedStrategies.length" class="strategy-tags">
+              <span v-for="s in selectedStrategies" :key="s" class="strategy-tag">{{ s }}</span>
+            </div>
           </div>
-        </div>
-        <div v-if="videosLoading" class="empty">加载中…</div>
-        <div v-else-if="videosError" class="empty error">{{ videosError }}</div>
-        <div v-else-if="videos.length" class="card-list">
-          <button
-            v-for="video in videos"
-            :key="video.vid"
-            type="button"
-            class="card video-card"
-            :class="{ active: selectedVid === video.vid }"
-            @click="selectVideo(video)"
-          >
-            <span class="card-label">Video {{ video.vid }}</span>
-            <span class="card-value">{{ video.title || '（无标题）' }}</span>
-          </button>
-        </div>
-        <div v-else class="empty">暂无关联视频</div>
-      </template>
-      <div v-else class="empty hint">点击需求词展开</div>
-    </section>
+          <div v-if="videosLoading" class="empty">加载中…</div>
+          <div v-else-if="videosError" class="empty error">{{ videosError }}</div>
+          <div v-else class="card-list">
+            <button
+              v-for="video in videos"
+              :key="video.vid"
+              type="button"
+              class="card video-card"
+              :class="{ active: selectedVid === video.vid }"
+              @click="selectVideo(video)"
+            >
+              <span class="card-label">Video {{ video.vid }}</span>
+              <span class="card-value">{{ video.title || '（无标题）' }}</span>
+            </button>
+          </div>
+        </template>
+      </section>
 
-    <!-- 视频 → JSON -->
-    <div class="arrow-col" aria-hidden="true">
-      <div v-if="selectedVideo" class="arrow arrow-purple">
-        <span class="arrow-line" />
-        <span class="arrow-head">▶</span>
+      <!-- 视频 → 选题 -->
+      <div class="arrow-col" aria-hidden="true">
+        <div v-if="selectedVideo" class="arrow arrow-purple">
+          <span class="arrow-line" />
+          <span class="arrow-head">▶</span>
+        </div>
       </div>
-    </div>
 
-    <section class="col col-topic">
-      <h3 class="col-title">选题结果</h3>
-      <template v-if="selectedVideo">
-        <div v-if="topicSections" class="topic-card">
-          <div class="topic-meta">Video {{ selectedVideo.vid }}</div>
-          <div class="topic-body">
-            <section
-              v-for="section in topicSections"
-              :key="section.key"
-              class="topic-section"
-              :class="`accent-${section.accent}`"
-            >
-              <div class="section-head">
-                <span class="section-label">{{ section.label }}</span>
-                <span class="section-count">{{ section.items.length }}</span>
-              </div>
-              <div v-if="section.items.length" class="point-list">
-                <article v-for="(item, idx) in section.items" :key="idx" class="point-item">
-                  <div class="point-title-row">
-                    <span class="point-index">{{ idx + 1 }}</span>
-                    <h4 class="point-title">{{ item.title || '未命名' }}</h4>
-                  </div>
-                  <p v-if="item.description" class="point-desc">{{ item.description }}</p>
-                </article>
-              </div>
-              <div v-else class="section-empty">暂无</div>
-            </section>
+      <section class="col col-topic">
+        <h3 class="col-title">选题结果</h3>
+        <template v-if="selectedVideo">
+          <div v-if="topicSections" class="topic-card">
+            <div class="topic-meta">Video {{ selectedVideo.vid }}</div>
+            <div class="topic-body">
+              <section
+                v-for="section in topicSections"
+                :key="section.key"
+                class="topic-section"
+                :class="`accent-${section.accent}`"
+              >
+                <div class="section-head">
+                  <span class="section-label">{{ section.label }}</span>
+                  <span class="section-count">{{ section.items.length }}</span>
+                </div>
+                <div v-if="section.items.length" class="point-list">
+                  <article v-for="(item, idx) in section.items" :key="idx" class="point-item">
+                    <div class="point-title-row">
+                      <span class="point-index">{{ idx + 1 }}</span>
+                      <h4 class="point-title">{{ item.title || '未命名' }}</h4>
+                    </div>
+                    <p v-if="item.pointDesc" class="point-desc">{{ item.pointDesc }}</p>
+                    <p v-if="item.reason" class="point-reason">{{ item.reason }}</p>
+                  </article>
+                </div>
+                <div v-else class="section-empty">暂无</div>
+              </section>
+            </div>
           </div>
-        </div>
-        <div v-else class="empty">暂无选题结果</div>
-      </template>
-      <div v-else class="empty hint">点击视频展开</div>
-    </section>
+          <div v-else class="empty">暂无选题结果</div>
+        </template>
+        <div v-else class="empty hint">点击视频展开</div>
+      </section>
+    </template>
   </div>
 </template>
 
@@ -769,6 +772,16 @@ function selectVideo(video: DemandVideoItem) {
   word-break: break-word;
 }
 
+.point-reason {
+  margin: 4px 0 0;
+  padding-left: 26px;
+  font-size: 11px;
+  line-height: 1.5;
+  color: #64748b;
+  white-space: pre-wrap;
+  word-break: break-word;
+}
+
 .section-empty {
   padding: 8px 10px;
   font-size: 12px;

web/src/types/demand.ts

@@ -15,20 +15,28 @@ export interface DemandGradeResponse {
   items: DemandGradeItem[]
 }
 
+export interface DemandExpansionPoint {
+  expanded_text: string
+  point_type: 'inspiration' | 'purpose' | 'key'
+  point_desc: string | null
+  reason: string
+}
+
 export interface DemandVideoItem {
   vid: string
   title: string | null
-  inspiration_points_json: string | null
-  purpose_points_json: string | null
-  key_points_json: string | null
+  expansion_points: DemandExpansionPoint[]
 }
 
 export interface DemandGradeVideosResponse {
   demand_grade_id: number
   demand_name: string
+  biz_dt: string
   category_ids: number[]
   grade: string
   strategies: string[]
+  expansion_status: 'finished' | 'failed' | string | null
+  expansion_saved_count: number
   videos: DemandVideoItem[]
 }