增加测试入口

agents/find_agent/LOCAL_TEST.md

@@ -0,0 +1,119 @@
+# find_agent 本地测试
+
+按 `demand_grade.id`（需求 id）在本地跑一条完整找视频链路，或只预览任务上下文。
+
+## 前置条件
+
+1. 项目根目录已配置 `.env`（MySQL、模型、抖音/TikHub 等）
+2. 目标 `demand_grade` 记录存在
+3. 该需求在对应 `biz_dt` 下已有有效拓展点位（`demand_video_expansion`，类型为 `inspiration` / `purpose` / `key`）
+
+## 快速开始
+
+在项目根目录执行：
+
+```bash
+# 按需求 id 执行完整 find_agent
+python -m agents.find_agent.run --demand-id 12345
+
+# 只看上下文，不跑 Agent
+python -m agents.find_agent.run --demand-id 12345 --dry-run
+
+# 强制重跑（忽略当天已完成/已尝试跳过）
+python -m agents.find_agent.run --demand-id 12345 --force
+
+# 指定业务日（一般不用；默认取该分级记录自己的 biz_dt）
+python -m agents.find_agent.run --demand-id 12345 --biz-dt 20260810
+
+# 输出 DEBUG 日志
+python -m agents.find_agent.run --demand-id 12345 -v
+```
+
+`--demand-grade-id` 与 `--demand-id` 等价。
+
+## 参数说明
+
+| 参数 | 含义 |
+|---|---|
+| `--demand-id` / `--demand-grade-id` | `demand_grade` 表主键，即需求 id |
+| `--biz-dt` | 业务日 `YYYYMMDD`；默认用该需求记录的 `biz_dt` |
+| `--force` | 强制重跑：复用已有 `run_id`，重置输入，不因已完成/已尝试而跳过 |
+| `--dry-run` | 只加载并打印上下文，不调用模型与外部搜索 |
+| `-v` / `--verbose` | DEBUG 日志 |
+
+不传 `--demand-id` 时进入交互模式：手动输入 `run_id` 和用户消息（适合调试 Prompt/工具）。
+
+## 执行流程
+
+1. 按 `demand_grade_id` 读取分级记录
+2. 组装参考视频与拓展点位，得到 `FindDemandContext`
+3. 预创建 / 复用 `video_discovery_run`
+4. 调用 `run_find_agent`
+5. 根据落库状态评估业务结果，并打印 JSON 摘要
+
+对应代码：
+
+- 加载：`agents/find_agent/demand_run.py` → `load_find_demand_context_by_id`
+- 执行：`agents/find_agent/demand_run.py` → `discover_videos_for_demand`
+- CLI：`agents/find_agent/run.py` → `run_by_demand_id`
+
+## 输出解读
+
+成功跑通时，终端会打印两段 JSON：
+
+1. **任务上下文**：`biz_dt`、`demand_grade_id`、`demand_name`、`grade`、参考视频与点位
+2. **执行结果**：
+   - `run_id`：本次找视频运行 id
+   - `skipped` / `skip_reason`：是否因已执行过而跳过
+   - `succeeded` / `goal_met` / `valid_primary_count`：业务是否达标
+   - `failure_reason`：失败原因
+   - `agent.content`：模型最终回复摘要
+
+可用 `run_id` 去库表或前端找视频记录页核对候选与证据。
+
+## 常见问题
+
+### 提示「未找到可执行上下文」
+
+依次检查：
+
+1. `demand_grade.id` 是否正确
+2. 对应 `biz_dt` 下是否有未删除的 `demand_video_expansion`
+3. 点位类型是否为 `inspiration` / `purpose` / `key`，且 `expanded_text` 非空
+
+可用 dry-run 快速验证：
+
+```bash
+python -m agents.find_agent.run --demand-id 12345 --dry-run
+```
+
+### 执行被跳过（`skipped: true`）
+
+当天该需求已有找视频运行且未加 `--force`。需要重跑时加上：
+
+```bash
+python -m agents.find_agent.run --demand-id 12345 --force
+```
+
+### 与定时任务的差异
+
+| 项 | 本地 `--demand-id` | 定时任务 `discover_videos_from_demands` |
+|---|---|---|
+| 选任务方式 | 指定单个需求 id | 按业务日批量拉 S/A |
+| 等级限制 | 不限制等级 | 默认仅 S/A |
+| 并发 | 单条串行 | 可多 worker |
+| 用途 | 联调 / 复现 / 单条验收 | 日常生产调度 |
+
+## 代码内调用
+
+```python
+from agents.find_agent.demand_run import (
+    discover_videos_for_demand,
+    load_find_demand_context_by_id,
+)
+
+ctx = load_find_demand_context_by_id(12345)
+assert ctx is not None
+result = discover_videos_for_demand(ctx, force=True)
+print(result.run_id, result.succeeded, result.valid_primary_count)
+```

agents/find_agent/demand_run.py

@@ -141,6 +141,66 @@ def _grade_priority_key(row: Any) -> tuple[float, int, str, int]:
     return (-score, grade_rank, str(row.demand_name), int(row.id))
 
 
+def _build_context_from_grade_row(
+    session,
+    grade_row: Any,
+    *,
+    biz_dt: str | None = None,
+) -> FindDemandContext | None:
+    """由单条 demand_grade 记录组装 FindDemandContext；无有效点位时返回 None。"""
+    resolved_biz_dt = str(biz_dt or grade_row.biz_dt).strip()
+    expansion_repo = DemandVideoExpansionRepository(session)
+    detail_repo = MultiDemandVideoDetailRepository(session)
+
+    expansions = expansion_repo.list_by_demand_grade(
+        resolved_biz_dt, int(grade_row.id)
+    )
+    if not expansions:
+        return None
+
+    by_video: dict[str, list[Any]] = {}
+    video_order: list[str] = []
+    for row in expansions:
+        video_id = str(row.video_id or "").strip()
+        if not video_id:
+            continue
+        if video_id not in by_video:
+            by_video[video_id] = []
+            video_order.append(video_id)
+        by_video[video_id].append(row)
+
+    if not video_order:
+        return None
+
+    details = detail_repo.list_by_vids(video_order)
+    videos: list[FindDemandVideo] = []
+    for video_id in video_order:
+        points = _build_video_points(by_video[video_id])
+        if not points:
+            continue
+
+        detail = details.get(video_id)
+        title = str(detail.title).strip() if detail and detail.title else ""
+        videos.append(
+            FindDemandVideo(
+                video_id=video_id,
+                title=title or f"(无标题|{video_id})",
+                points=points,
+            )
+        )
+
+    if not videos:
+        return None
+
+    return FindDemandContext(
+        biz_dt=resolved_biz_dt,
+        demand_grade_id=int(grade_row.id),
+        demand_name=str(grade_row.demand_name),
+        grade=str(grade_row.grade),
+        videos=videos,
+    )
+
+
 def load_find_demand_contexts(
     session,
     biz_dt: str,
@@ -157,60 +217,30 @@ def load_find_demand_contexts(
     if top_limit is not None and top_limit > 0:
         grades_rows = grades_rows[: int(top_limit)]
 
-    expansion_repo = DemandVideoExpansionRepository(session)
-    detail_repo = MultiDemandVideoDetailRepository(session)
-
     contexts: list[FindDemandContext] = []
     for grade_row in grades_rows:
-        expansions = expansion_repo.list_by_demand_grade(biz_dt, int(grade_row.id))
-        if not expansions:
-            continue
+        ctx = _build_context_from_grade_row(session, grade_row, biz_dt=biz_dt)
+        if ctx is not None:
+            contexts.append(ctx)
 
-        by_video: dict[str, list[Any]] = {}
-        video_order: list[str] = []
-        for row in expansions:
-            video_id = str(row.video_id or "").strip()
-            if not video_id:
-                continue
-            if video_id not in by_video:
-                by_video[video_id] = []
-                video_order.append(video_id)
-            by_video[video_id].append(row)
-
-        if not video_order:
-            continue
-
-        details = detail_repo.list_by_vids(video_order)
-        videos: list[FindDemandVideo] = []
-        for video_id in video_order:
-            points = _build_video_points(by_video[video_id])
-            if not points:
-                continue
-
-            detail = details.get(video_id)
-            title = str(detail.title).strip() if detail and detail.title else ""
-            videos.append(
-                FindDemandVideo(
-                    video_id=video_id,
-                    title=title or f"(无标题|{video_id})",
-                    points=points,
-                )
-            )
+    return contexts
 
-        if not videos:
-            continue
 
-        contexts.append(
-            FindDemandContext(
-                biz_dt=biz_dt,
-                demand_grade_id=int(grade_row.id),
-                demand_name=str(grade_row.demand_name),
-                grade=str(grade_row.grade),
-                videos=videos,
-            )
-        )
+def load_find_demand_context_by_id(
+    demand_grade_id: int,
+    *,
+    biz_dt: str | None = None,
+) -> FindDemandContext | None:
+    """按 demand_grade.id（需求 id）直接加载一条 find_agent 上下文。
 
-    return contexts
+    不限制 S/A 等级；若记录不存在、或该需求下无有效拓展点位，返回 None。
+    biz_dt 默认取该分级记录自身的业务日。
+    """
+    with get_session() as session:
+        grade_row = DemandGradeRepository(session).get_by_id(int(demand_grade_id))
+        if grade_row is None:
+            return None
+        return _build_context_from_grade_row(session, grade_row, biz_dt=biz_dt)
 
 
 def pick_find_demand_context(
@@ -221,6 +251,9 @@ def pick_find_demand_context(
     grades: Iterable[str] = ("S", "A"),
 ) -> FindDemandContext | None:
     """从数据库选取一条待执行的 find_agent 上下文。"""
+    if demand_grade_id is not None:
+        return load_find_demand_context_by_id(int(demand_grade_id), biz_dt=biz_dt)
+
     resolved_biz_dt = _resolve_biz_dt(biz_dt)
     with get_session() as session:
         contexts = load_find_demand_contexts(
@@ -229,12 +262,6 @@ def pick_find_demand_context(
             grades=grades,
         )
 
-    if demand_grade_id is not None:
-        for ctx in contexts:
-            if ctx.demand_grade_id == int(demand_grade_id):
-                return ctx
-        return None
-
     if 0 <= index < len(contexts):
         return contexts[index]
     return None

agents/find_agent/run.py

@@ -1,10 +1,94 @@
 #!/usr/bin/env python3
-"""Run find_agent interactively."""
+"""find_agent 本地入口。
+
+支持两种模式：
+1. 按需求 id（demand_grade.id）跑完整找视频链路；
+2. 无参数时进入交互式手动输入（调试 Prompt / 工具用）。
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import sys
+from typing import Any
 
 from agents.find_agent import create_find_agent, run_find_agent
+from agents.find_agent.demand_run import (
+    discover_videos_for_demand,
+    load_find_demand_context_by_id,
+    serialize_find_demand_context,
+)
+
+
+def _configure_logging(verbose: bool) -> None:
+    level = logging.DEBUG if verbose else logging.INFO
+    logging.basicConfig(
+        level=level,
+        format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
+    )
+
+
+def _print_json(payload: dict[str, Any]) -> None:
+    print(json.dumps(payload, ensure_ascii=False, indent=2))
+
+
+def run_by_demand_id(
+    demand_grade_id: int,
+    *,
+    biz_dt: str | None = None,
+    force: bool = False,
+    dry_run: bool = False,
+) -> dict[str, Any]:
+    """按 demand_grade.id 加载上下文并执行 find_agent。"""
+    ctx = load_find_demand_context_by_id(demand_grade_id, biz_dt=biz_dt)
+    if ctx is None:
+        raise SystemExit(
+            f"未找到可执行上下文：demand_grade_id={demand_grade_id}"
+            + (f", biz_dt={biz_dt}" if biz_dt else "")
+            + "（记录不存在，或无有效拓展点位）"
+        )
+
+    summary = serialize_find_demand_context(ctx)
+    print("=== find_agent 任务上下文 ===")
+    _print_json(summary)
+
+    if dry_run:
+        print("\n[dry-run] 已跳过 Agent 执行")
+        return {"dry_run": True, "context": summary}
 
+    print(
+        f"\n开始执行 find_agent："
+        f"demand_grade_id={ctx.demand_grade_id} "
+        f"demand={ctx.demand_name} grade={ctx.grade} force={force}"
+    )
+    execution = discover_videos_for_demand(ctx, force=force)
 
-def main() -> None:
+    result: dict[str, Any] = {
+        "context": summary,
+        "run_id": execution.run_id,
+        "skipped": execution.skipped,
+        "skip_reason": execution.skip_reason,
+        "succeeded": execution.succeeded,
+        "business_outcome": execution.business_outcome,
+        "goal_met": execution.goal_met,
+        "valid_primary_count": execution.valid_primary_count,
+        "failure_reason": execution.failure_reason,
+    }
+    if execution.agent_result is not None:
+        result["agent"] = {
+            "iterations": execution.agent_result.iterations,
+            "tool_calls_made": execution.agent_result.tool_calls_made,
+            "content": execution.agent_result.content,
+        }
+
+    print("\n=== 执行结果 ===")
+    _print_json(result)
+    return result
+
+
+def run_interactive() -> None:
+    """交互式手动输入 run_id + 用户消息。"""
     agent = create_find_agent()
     print(f"find_agent ready | model={agent.model}")
     print(f"tools: {agent.tools.list_tools()}")
@@ -25,5 +109,60 @@ def main() -> None:
         print(f"\nAgent> {result.content}\n")
 
 
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        description="find_agent 本地测试入口：可直接传需求 id（demand_grade.id）执行",
+    )
+    parser.add_argument(
+        "--demand-id",
+        "--demand-grade-id",
+        dest="demand_id",
+        type=int,
+        help="demand_grade 表主键 id（需求 id）",
+    )
+    parser.add_argument(
+        "--biz-dt",
+        default=None,
+        help="业务日 YYYYMMDD；默认使用该需求记录自身的 biz_dt",
+    )
+    parser.add_argument(
+        "--force",
+        action="store_true",
+        help="强制重跑：复用已有 run_id 并重置输入（忽略已完成/已尝试跳过）",
+    )
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="只加载并打印上下文，不调用 Agent",
+    )
+    parser.add_argument(
+        "-v",
+        "--verbose",
+        action="store_true",
+        help="输出 DEBUG 日志",
+    )
+    return parser
+
+
+def main(argv: list[str] | None = None) -> None:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    _configure_logging(args.verbose)
+
+    if args.demand_id is not None:
+        run_by_demand_id(
+            args.demand_id,
+            biz_dt=args.biz_dt,
+            force=args.force,
+            dry_run=args.dry_run,
+        )
+        return
+
+    if args.biz_dt or args.force or args.dry_run:
+        parser.error("未指定 --demand-id 时，不能单独使用 --biz-dt / --force / --dry-run")
+
+    run_interactive()
+
+
 if __name__ == "__main__":
-    main()
+    main(sys.argv[1:])