修改寻找agent_v2版本测试入口

find_agent_v2/README.md

@@ -76,30 +76,12 @@ result = run_find_agent_v2(user_input, run_id=run_id)
 
 `run_id` 必须来自 `find_agent_v2_run`，不会隐式接受或写入旧 `video_discovery_run`。
 
-本地数据库准备好需求、视频和点位后，可直接创建一条测试运行：
+测试入口只接收一个需求词。程序会在 `demand_grade.demand_name` 中精确匹配，并按
+`biz_dt DESC, create_time DESC, id DESC` 取最新一条，再查询视频和点位生成完整输入，最后直接
+执行寻找 Agent：
 
 ```bash
-.venv/bin/python -m find_agent_v2.test_entry
-```
-
-该命令默认只准备 `find_agent_v2_run`，不会调用模型。真正执行时显式传入：
-
-```bash
-.venv/bin/python -m find_agent_v2.test_entry --execute
-```
-
-已准备的任务可按 `run_id` 恢复执行：
-
-```bash
-.venv/bin/python -m find_agent_v2.test_entry \
-  --existing-run-id local-test-... --execute
-```
-
-已经进入 terminal 状态的任务可保留历史轮次和候选、从下一轮恢复：
-
-```bash
-.venv/bin/python -m find_agent_v2.test_entry \
-  --existing-run-id local-test-... --resume --execute
+.venv/bin/python -m find_agent_v2.test_entry "照顾父母"
 ```
 
 ## 复用边界

find_agent_v2/__init__.py

@@ -9,8 +9,10 @@ from find_agent_v2.demand_context import (
     PreparedV2DemandRun,
     V2DemandContext,
     load_v2_demand_context,
+    load_latest_v2_demand_context_by_name,
     pick_latest_v2_demand_context,
     prepare_v2_demand_run,
+    prepare_latest_v2_demand_run_by_name,
 )
 from find_agent_v2.runner import (
     arun_find_agent_v2,
@@ -31,8 +33,10 @@ __all__ = [
     "create_find_agent_v2_run",
     "create_find_agent_v2",
     "load_v2_demand_context",
+    "load_latest_v2_demand_context_by_name",
     "pick_latest_v2_demand_context",
     "prepare_v2_demand_run",
+    "prepare_latest_v2_demand_run_by_name",
     "run_prepared_find_agent_v2",
     "run_find_agent_v2",
 ]

find_agent_v2/demand_context.py

@@ -135,7 +135,10 @@ def _load_context_in_session(session, grade: DemandGrade) -> V2DemandContext | N
             DemandVideoExpansion.biz_dt == grade.biz_dt,
             DemandVideoExpansion.source_demand_grade_id == grade.id,
             DemandVideoExpansion.is_delete == 0,
-        ).order_by(DemandVideoExpansion.id)
+        ).order_by(
+            DemandVideoExpansion.create_time.desc(),
+            DemandVideoExpansion.id.desc(),
+        )
     ))
     video_order, points_by_video = _points_from_expansions(expansions)
     if not points_by_video:
@@ -143,7 +146,11 @@ def _load_context_in_session(session, grade: DemandGrade) -> V2DemandContext | N
         source_points = list(session.scalars(
             select(MultiDemandVideoPoint).where(
                 MultiDemandVideoPoint.video_id.in_(source_ids)
-            ).order_by(MultiDemandVideoPoint.video_id, MultiDemandVideoPoint.id)
+            ).order_by(
+                MultiDemandVideoPoint.video_id,
+                MultiDemandVideoPoint.create_time.desc(),
+                MultiDemandVideoPoint.id.desc(),
+            )
         )) if source_ids else []
         video_order, points_by_video = _points_from_source_rows(source_ids, source_points)
     if not points_by_video:
@@ -187,6 +194,37 @@ def load_v2_demand_context(demand_grade_id: int) -> V2DemandContext:
         return context
 
 
+def _latest_demand_grade_query(demand_word: str):
+    return (
+        select(DemandGrade)
+        .where(DemandGrade.demand_name == demand_word)
+        .order_by(
+            DemandGrade.biz_dt.desc(),
+            DemandGrade.create_time.desc(),
+            DemandGrade.id.desc(),
+        )
+    )
+
+
+def load_latest_v2_demand_context_by_name(demand_word: str) -> V2DemandContext:
+    """Load the newest exact ``demand_name`` match from demand_grade."""
+    normalized = str(demand_word or "").strip()
+    if not normalized:
+        raise ValueError("需求词不能为空")
+    with get_session() as session:
+        grades = list(session.scalars(_latest_demand_grade_query(normalized)))
+        if not grades:
+            raise LookupError(f"demand_grade 未找到需求词: {normalized}")
+        newest = grades[0]
+        context = _load_context_in_session(session, newest)
+        if context is not None:
+            return context
+        raise ValueError(
+            "需求的最新记录没有可用参考视频和点位: "
+            f"demand_name={normalized}, demand_grade_id={newest.id}, biz_dt={newest.biz_dt}"
+        )
+
+
 def pick_latest_v2_demand_context(*, index: int = 0) -> V2DemandContext:
     with get_session() as session:
         latest = session.scalar(select(DemandGrade.biz_dt).where(
@@ -254,3 +292,27 @@ def prepare_v2_demand_run(
         reference_video_count=len(context.videos),
         point_count=context.point_count,
     )
+
+
+def prepare_latest_v2_demand_run_by_name(demand_word: str) -> PreparedV2DemandRun:
+    """Resolve the newest demand record, build its full context, and create a V2 run."""
+    context = load_latest_v2_demand_context_by_name(demand_word)
+    run_key = f"demand-test-{uuid.uuid4().hex}"[:64]
+    rules = build_rule_snapshot()
+    user_input = build_v2_user_input(context, run_id=run_key, rules=rules)
+    get_find_agent_v2_service().create_run(
+        run_id=run_key,
+        user_input=user_input,
+        demand_word=context.demand_name,
+        demand_grade_id=context.demand_grade_id,
+        rule_config=rules,
+    )
+    return PreparedV2DemandRun(
+        run_id=run_key,
+        demand_grade_id=context.demand_grade_id,
+        demand_name=context.demand_name,
+        biz_dt=context.biz_dt,
+        user_input=user_input,
+        reference_video_count=len(context.videos),
+        point_count=context.point_count,
+    )

find_agent_v2/test_entry.py

@@ -1,61 +1,34 @@
-"""Local CLI for preparing or executing one database-backed v2 test run."""
+"""Execute one database-backed V2 test using only a demand word."""
 
 from __future__ import annotations
 
 import argparse
 import json
 
-from find_agent_v2.demand_context import prepare_v2_demand_run
-from find_agent_v2.runner import run_find_agent_v2, run_prepared_find_agent_v2
-from find_agent_v2.service import get_find_agent_v2_service
+from find_agent_v2.demand_context import prepare_latest_v2_demand_run_by_name
+from find_agent_v2.runner import run_find_agent_v2
 
 
 def main() -> None:
-    parser = argparse.ArgumentParser(description="准备或执行一条寻找 Agent v2 本地测试任务")
-    parser.add_argument("--demand-grade-id", type=int, default=None)
-    parser.add_argument("--index", type=int, default=0, help="未指定需求 ID 时选择最新 S 级上下文序号")
-    parser.add_argument("--run-id", default=None)
-    parser.add_argument("--existing-run-id", default=None, help="读取并执行已准备的 v2 run")
-    parser.add_argument("--execute", action="store_true", help="准备后立即调用模型和外部搜索")
-    parser.add_argument("--model", default=None)
-    parser.add_argument("--resume", action="store_true", help="从 terminal run 的下一轮恢复")
+    parser = argparse.ArgumentParser(
+        description="按需求词查询线上最新 demand_grade 上下文并执行寻找 Agent v2",
+    )
+    parser.add_argument("demand_word", help="与 demand_grade.demand_name 精确匹配的需求词")
     args = parser.parse_args()
 
-    if args.existing_run_id:
-        run = get_find_agent_v2_service().require_run(args.existing_run_id)
-        prepared = None
-        output: dict = {"prepared": {
-            "run_id": run["run_id"],
-            "demand_grade_id": run["demand_grade_id"],
-            "demand_name": run["demand_word"],
-            "status": run["status"],
-        }, "executed": False}
-    else:
-        prepared = prepare_v2_demand_run(
-            demand_grade_id=args.demand_grade_id,
-            index=args.index,
-            run_id=args.run_id,
-        )
-        output = {"prepared": prepared.summary(), "executed": False}
-    if args.execute:
-        result = (
-            run_prepared_find_agent_v2(
-                args.existing_run_id, model=args.model, resume=args.resume,
-            )
-            if args.existing_run_id
-            else run_find_agent_v2(
-                prepared.user_input,
-                run_id=prepared.run_id,
-                model=args.model,
-            )
-        )
-        output.update({"executed": True, "result": {
+    prepared = prepare_latest_v2_demand_run_by_name(args.demand_word)
+    result = run_find_agent_v2(prepared.user_input, run_id=prepared.run_id)
+    output = {
+        "prepared": prepared.summary(),
+        "executed": True,
+        "result": {
             "status": result.status,
             "succeeded": result.succeeded,
             "valid_primary_count": result.valid_primary_count,
             "stop_reason": result.stop_reason,
             "final_output": result.final_output,
-        }})
+        },
+    }
     print(json.dumps(output, ensure_ascii=False, default=str, indent=2))
 
 

tests/supply_agent/test_find_agent_v2.py

@@ -15,6 +15,7 @@ from find_agent_v2.demand_context import (
     V2DemandContext,
     V2ReferencePoint,
     V2ReferenceVideo,
+    _latest_demand_grade_query,
     _points_from_expansions,
     build_v2_user_input,
 )
@@ -187,6 +188,16 @@ def test_v2_context_builds_self_contained_user_input() -> None:
     assert '"关键内容"' in raw
 
 
+def test_demand_word_lookup_uses_exact_name_and_newest_record_order() -> None:
+    sql = str(
+        _latest_demand_grade_query("照顾父母").compile(
+            compile_kwargs={"literal_binds": True},
+        )
+    ).lower()
+    assert "demand_grade.demand_name = '照顾父母'" in sql
+    assert "order by demand_grade.biz_dt desc, demand_grade.create_time desc, demand_grade.id desc" in sql
+
+
 def test_obagent_identity_and_round_structure_are_stable() -> None:
     assert OBAGENT_PROJECT == "find_agent_v2"
     assert OBAGENT_AGENT == "find_agent_v2"