SupplyAgent/agents/demand_grade_agent/run.py

#!/usr/bin/env python3
"""Run demand_grade_agent on one batch of demand names.

批次选择（哪些词、多少个）由调用方（通常是调度任务）负责，本入口只处理
传入的这一批，不做分页/自行遍历需求池。
"""
from __future__ import annotations

from typing import Any

from agents.demand_grade_agent import create_demand_grade_agent


def build_grade_user_input(demands: list[dict[str, Any]], biz_dt: str) -> str:
    """构建传给分级 Agent 的用户消息。"""
    demand_lines = []
    for item in demands:
        pool_id = item.get("pool_id", item.get("id"))
        if pool_id is None:
            raise ValueError(f"demand 缺少 pool_id: {item!r}")
        demand_lines.append(f"[{int(pool_id)}] {str(item['demand_name']).strip()}")

    lines_text = "\n".join(demand_lines)
    return f"""请对以下 {len(demand_lines)} 个需求词逐一评级（S/A/B/C/D）。
只处理这一批，不要尝试查找或列举更多需求词。判级完成后调用 batch_save_demand_grades 落库。
落库时 related_pool_ids 使用列表中对应行的 pool_id；分类、热度、后验等证据请通过工具从数据库查询。

biz_dt={biz_dt}
需求词列表 [pool_id] demand_name
{lines_text}
"""


def main(
    demands: list[dict[str, Any]],
    biz_dt: str | None = None,
) -> None:
    agent = create_demand_grade_agent()
    print(f"demand_grade_agent ready | model={agent.model}")
    print(f"tools: {agent.tools.list_tools()}")
    print()

    if not demands:
        raise ValueError("demands 不能为空")

    batch_biz_dt = (biz_dt or "").strip()
    if not batch_biz_dt:
        raise ValueError("biz_dt 不能为空")

    user_input = build_grade_user_input(demands, batch_biz_dt)
    result = agent.run(user_input)
    print(result.content)
    print(f"\n[iterations={result.iterations}, tool_calls={result.tool_calls_made}]")


if __name__ == "__main__":
    main(
        [
            {"pool_id": 101, "demand_name": "减脂期加餐"},
            {"pool_id": 102, "demand_name": "减脂期"},
        ],
        biz_dt="20260721",
    )