Agent/examples/auto_put_ad_mini/test_approval_replay.py

"""
审批回放测试 — 复用 validated_decisions_20260507.csv 反复发审批。

每次启动跑一轮：发审批表 → 阻塞等您在飞书回复 → Agent 根据回复选择后续工具链 → 发回执。
您可在飞书群里换不同回复（通过/拒绝/通过广告 X/降幅改小一点/为什么 pause 这条 X）测试 Agent 的动作和回执文案。

用法：
    cd /Users/liulidong/project/agent/Agent
    .venv/bin/python3 examples/auto_put_ad_mini/test_approval_replay.py [validated_csv_path]
"""

import asyncio
import os
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent.parent))

from dotenv import load_dotenv
load_dotenv()

from agent.core.runner import AgentRunner
from agent.trace import FileSystemTraceStore
from agent.llm import create_openrouter_llm_call
from agent.utils import setup_logging

from examples.auto_put_ad_mini.config import (
    MAIN_CONFIG, SKILLS_DIR, TRACE_STORE_PATH, LOG_LEVEL, LOG_FILE,
)

# 触发 @tool 注册
from examples.auto_put_ad_mini.tools.data_query import fetch_creative_data, merge_creative_data  # noqa
from examples.auto_put_ad_mini.tools.roi_calculator import calculate_roi_metrics  # noqa
from examples.auto_put_ad_mini.tools.portfolio_metrics import calculate_portfolio_summary  # noqa
from examples.auto_put_ad_mini.tools.ad_decision import (  # noqa
    get_ads_for_review, apply_decisions, query_ad_detail, modify_decisions,
)
from examples.auto_put_ad_mini.tools.report_generator import generate_report  # noqa
from examples.auto_put_ad_mini.tools.guardrails import validate_decisions  # noqa
from examples.auto_put_ad_mini.tools.execution_engine import execute_decisions, check_execution_feedback  # noqa
from examples.auto_put_ad_mini.tools.im_approval import (  # noqa
    send_approval_request, check_approval_status, send_feishu_text_message,
)


def _load_system_prompt(base_dir: Path) -> str:
    p = base_dir / "prompts" / "system.prompt"
    return p.read_text(encoding="utf-8") if p.exists() else ""


async def main():
    setup_logging(level=LOG_LEVEL, file=LOG_FILE)
    base_dir = Path(__file__).parent

    validated_csv = sys.argv[1] if len(sys.argv) > 1 else str(
        base_dir / "outputs" / "reports" / "validated_decisions_20260507.csv"
    )
    if not Path(validated_csv).exists():
        print(f"❌ 找不到 validated CSV: {validated_csv}")
        sys.exit(1)

    print(f"▶ 使用决策 CSV: {validated_csv}")
    print("▶ 启动 Agent，将在飞书发审批表并阻塞等待回复...")
    print("▶ 在飞书群里 @机器人 + 回复任意以下内容测试:")
    print("    - 通过 / 同意 / 批准")
    print("    - 拒绝 / 驳回")
    print("    - 通过广告 12345678901")
    print("    - 通过 12345678901, 拒绝 22345678901")
    print("    - 降幅改小一点")
    print("    - 为什么 pause 这条 12345678901？")
    print("─" * 60)

    store = FileSystemTraceStore(base_path=TRACE_STORE_PATH)
    runner = AgentRunner(
        trace_store=store,
        llm_call=create_openrouter_llm_call(model=MAIN_CONFIG.model),
        skills_dir=SKILLS_DIR if Path(SKILLS_DIR).exists() else None,
        logger_name="agents.auto_put_ad_mini.replay",
    )

    config = MAIN_CONFIG
    system_prompt = _load_system_prompt(base_dir)
    if system_prompt:
        config.system_prompt = system_prompt

    user_msg = (
        f"复用已验证决策 CSV `{validated_csv}` 直接走审批协商流程：\n"
        "1. 调用 `send_approval_request(validated_csv='{path}', wait_for_reply=True)` 发审批表并阻塞等回复\n"
        "2. 收到回复后,严格按 system prompt 的反馈类型识别表选择后续工具链:\n"
        "   - 全部通过 → execute_decisions + send_feishu_text_message 摘要\n"
        "   - 全部拒绝 → 仅 send_feishu_text_message 简短回执,不执行\n"
        "   - 部分批准型 → execute_decisions(filter_ad_ids=approved_ids) + send_feishu_text_message 回执,**严禁重审**\n"
        "   - 策略型/方向型 → modify_decisions → validate_decisions → send_approval_request 重审\n"
        "   - 质疑型 → query_ad_detail + send_feishu_text_message 解释\n"
        "3. 全程使用第二人称「您」,禁用【】公文头\n"
    ).format(path=validated_csv)

    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_msg},
    ]

    async for item in runner.run(messages=messages, config=config):
        # 流式打印 Agent 每一步动作
        cls_name = type(item).__name__
        if hasattr(item, "tool_name") and getattr(item, "tool_name", None):
            print(f"  → tool: {item.tool_name}")
        elif hasattr(item, "role") and getattr(item, "role", None) == "assistant":
            content = getattr(item, "content", "")
            if isinstance(content, str) and content.strip():
                snippet = content[:500]
                print(f"\n[assistant] {snippet}")
            elif isinstance(content, list):
                for blk in content:
                    if isinstance(blk, dict) and blk.get("type") == "text":
                        print(f"\n[assistant text] {str(blk.get('text', ''))[:500]}")
        else:
            print(f"  · {cls_name}")
    print("─" * 60)
    print("▶ Agent 终止")


if __name__ == "__main__":
    asyncio.run(main())