SupplyAgent/agents/find_agent/run.py

#!/usr/bin/env python3
"""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)

    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()}")
    print()

    while True:
        try:
            run_id = input("run_id> ").strip()
            if not run_id or run_id.lower() in ("exit", "quit", "q"):
                break
            user_input = input("You> ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\nBye.")
            break
        if not user_input or user_input.lower() in ("exit", "quit", "q"):
            break
        result = run_find_agent(user_input, run_id=run_id)
        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(sys.argv[1:])