""" 一键执行智能引擎 - 临时脚本 """ import asyncio import os import sys import time from pathlib import Path from datetime import datetime, timedelta import logging # 添加项目根目录到 Python 路径 sys.path.insert(0, str(Path(__file__).parent.parent.parent)) # 代理配置（从环境变量读取，海外部署时通过 Docker 环境变量注入） logger = logging.getLogger(__name__) http_proxy = os.getenv("HTTP_PROXY") https_proxy = os.getenv("HTTPS_PROXY") if http_proxy or https_proxy: logger.info(f"使用代理：HTTP={http_proxy}, HTTPS={https_proxy}") from dotenv import load_dotenv load_dotenv() from agent.core.runner import AgentRunner from agent.trace import FileSystemTraceStore, Trace, Message from agent.llm import create_openrouter_llm_call from examples.auto_put_ad_mini.logging_setup import setup_logging, get_trace_id from examples.auto_put_ad_mini.config import ( MAIN_CONFIG, SKILLS_DIR, TRACE_STORE_PATH, LOG_LEVEL, ) # 导入自定义工具 from examples.auto_put_ad_mini.tools.data_query import fetch_creative_data, merge_creative_data from examples.auto_put_ad_mini.tools.roi_calculator import calculate_roi_metrics from examples.auto_put_ad_mini.tools.creative_roi_calculator import calculate_creative_roi from examples.auto_put_ad_mini.tools.portfolio_metrics import calculate_portfolio_summary from examples.auto_put_ad_mini.tools.ad_decision import get_ads_for_review, apply_decisions, query_ad_detail, modify_decisions from examples.auto_put_ad_mini.tools.report_generator import generate_report from examples.auto_put_ad_mini.tools.guardrails import validate_decisions from examples.auto_put_ad_mini.tools.execution_engine import execute_decisions, check_execution_feedback from examples.auto_put_ad_mini.tools.im_approval import send_approval_request, check_approval_status, send_feishu_text_message # 尝试导入飞书文档工具（如果存在） try: from examples.auto_put_ad_mini.tools.feishu_doc import import_to_feishu except ImportError: pass # 工具不存在，忽略 async def main(): """广告智能调控主流程：数据拉取 → ROI计算 → 候选筛选 → AI推理 → 护栏验证 → 报告生成。""" base_dir = Path(__file__).parent setup_logging(level=LOG_LEVEL) pipeline_logger = logging.getLogger("auto_put_ad_mini.pipeline") trace_id = get_trace_id() or "-" started_ms = int(time.monotonic() * 1000) pipeline_logger.info( "event=pipeline stage=run result=started trace_id=%s date=%s", trace_id, target_date if 'target_date' in dir() else "auto", ) # 加载 system prompt prompt_path = base_dir / "prompts" / "system.prompt" system_prompt = "" if prompt_path.exists(): system_prompt = prompt_path.read_text(encoding="utf-8") # 加载 presets presets_path = base_dir / "presets.json" if presets_path.exists(): from agent.core.presets import load_presets_from_json load_presets_from_json(str(presets_path)) 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", ) config = MAIN_CONFIG if system_prompt: config.system_prompt = system_prompt print("=" * 70) print(" 广告智能调控助手 — 智能引擎执行") print("=" * 70) print() print("🚀 自动执行：分析广告") print() print("=" * 70) print(" 流程：数据拉取 → ROI计算 → 人群包基线 → 候选筛选 → AI推理 → 保存决策 → 护栏验证 → 生成报告") print("=" * 70) print() # 自动取 T-1（昨天）作为数据截止日期，避免硬编码 target_date = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d") target_date_display = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d") messages = [{"role": "user", "content": f"分析广告，执行完整的ROI计算和决策流程。请使用 {target_date_display}（end_date={target_date}）作为数据截止日期，因为当天数据尚未回流。"}] config.trace_id = None step_count = 0 try: async for item in runner.run(messages=messages, config=config): if isinstance(item, Trace): if item.status == "completed": print(f"\n✅ [Trace] 完成") elif item.status == "failed": print(f"\n❌ [Trace] 失败") elif isinstance(item, Message): if item.role == "assistant" and item.content: content = item.content text = content.get("text", "") if isinstance(content, dict) else content if text and text.strip(): print(f"\n💭 {text}\n") elif item.role == "tool" and item.content: content = item.content if isinstance(content, dict): tool_name = content.get("tool_name", "unknown") result = content.get("result", content.get("text", str(content))) # 识别关键步骤 _step_label = None if tool_name == "fetch_creative_data": step_count += 1 _step_label = "数据拉取" elif tool_name == "calculate_roi_metrics": step_count += 1 _step_label = "ROI 计算" elif tool_name == "calculate_portfolio_summary": step_count += 1 _step_label = "人群包基线计算" elif tool_name == "get_ads_for_review": step_count += 1 _step_label = "候选筛选" elif tool_name == "apply_decisions": step_count += 1 _step_label = "保存智能引擎决策" elif tool_name == "validate_decisions": step_count += 1 _step_label = "安全护栏验证" elif tool_name == "execute_decisions": step_count += 1 _step_label = "分级执行" elif tool_name == "send_approval_request": step_count += 1 _step_label = "IM 审批请求" elif tool_name == "generate_report": step_count += 1 _step_label = "生成最终报告" elif tool_name == "check_execution_feedback": step_count += 1 _step_label = "执行效果检查" if _step_label: step_slug = _step_label.replace(" ", "_") pipeline_logger.info( "event=step stage=%s step=%d result=started trace_id=%s", step_slug, step_count, trace_id, ) print(f"\n{'='*70}") print(f"📌 步骤 {step_count}: {_step_label}") print(f"{'='*70}") # 打印简化结果 if isinstance(result, str): text = result else: text = str(result) if len(text) > 500: text = text[:500] + "..." print(f" {text}") print("\n" + "=" * 70) print("✅ 执行完成") print("=" * 70) print() print("📁 输出文件：") print(f" - 智能引擎决策：examples/auto_put_ad_mini/outputs/reports/llm_decisions_*.csv") print(f" - 最终报告（带格式）：examples/auto_put_ad_mini/outputs/reports/decision_*.xlsx") print() pipeline_logger.info( "event=pipeline stage=run result=completed duration_ms=%d steps=%d trace_id=%s", int(time.monotonic() * 1000) - started_ms, step_count, trace_id, ) except Exception as e: print(f"\n❌ 执行失败: {e}") import traceback traceback.print_exc() pipeline_logger.error( "event=pipeline stage=run result=failed duration_ms=%d error=%s trace_id=%s", int(time.monotonic() * 1000) - started_ms, str(e)[:200], trace_id, ) if __name__ == "__main__": asyncio.run(main())