find_agent 本地测试

按 demand_grade.id（需求 id）在本地跑一条完整找视频链路，或只预览任务上下文。

前置条件

项目根目录已配置 .env（MySQL、模型、抖音/TikHub 等）
目标 demand_grade 记录存在
该需求在对应 biz_dt 下已有有效拓展点位（demand_video_expansion，类型为 inspiration / purpose / key）

快速开始

在项目根目录执行：

# 按需求 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/工具）。

执行流程

按 demand_grade_id 读取分级记录
组装参考视频与拓展点位，得到 FindDemandContext
预创建 / 复用 video_discovery_run
调用 run_find_agent
根据落库状态评估业务结果，并打印 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：

任务上下文：biz_dt、demand_grade_id、demand_name、grade、参考视频与点位
执行结果：
- run_id：本次找视频运行 id
- skipped / skip_reason：是否因已执行过而跳过
- succeeded / goal_met / valid_primary_count：业务是否达标
- failure_reason：失败原因
- agent.content：模型最终回复摘要

可用 run_id 去库表或前端找视频记录页核对候选与证据。

常见问题

提示「未找到可执行上下文」

依次检查：

demand_grade.id 是否正确
对应 biz_dt 下是否有未删除的 demand_video_expansion
点位类型是否为 inspiration / purpose / key，且 expanded_text 非空

可用 dry-run 快速验证：

python -m agents.find_agent.run --demand-id 12345 --dry-run

执行被跳过（`skipped: true`）

当天该需求已有找视频运行且未加 --force。需要重跑时加上：

python -m agents.find_agent.run --demand-id 12345 --force

与定时任务的差异

项	本地 `--demand-id`	定时任务 `discover_videos_from_demands`
选任务方式	指定单个需求 id	按业务日拉齐 S/A，先入待执行队列
等级限制	不限制等级	默认仅 S/A
并发	单条串行	默认 5 worker 循环取任务；超时退出后自动补齐
落库时机	开始执行时创建 `video_discovery_run`	同左（入队时不预建记录）
用途	联调 / 复现 / 单条验收	日常生产调度

代码内调用

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)

LOCAL_TEST.md 4.0 KB Историја Датотека