""" 广告调控 Agent — auto_put_ad_mini 入口 运行方式： cd /Users/liulidong/project/agent/Agent python examples/auto_put_ad_mini/run.py """ import asyncio import os import sys from pathlib import Path import logging # 添加项目根目录到 Python 路径 sys.path.insert(0, str(Path(__file__).parent.parent.parent)) from dotenv import load_dotenv load_dotenv() # 代理配置（从环境变量读取，海外部署时通过 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 agent.core.runner import AgentRunner from agent.trace import FileSystemTraceStore, Trace, Message 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 from examples.auto_put_ad_mini.tools.roi_calculator import calculate_roi_metrics 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 async def init_project_env(messages=None): """供 api_server 可视化调用：返回 (runner, messages, config)""" base_dir = Path(__file__).parent system_prompt = _load_system_prompt(base_dir) _load_presets(base_dir) 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 if not messages: messages = [{"role": "user", "content": "分析广告"}] if system_prompt: has_system = any(m.get("role") == "system" for m in messages) if not has_system: messages = [{"role": "system", "content": system_prompt}] + messages return runner, messages, config def _load_system_prompt(base_dir: Path) -> str: prompt_path = base_dir / "prompts" / "system.prompt" if prompt_path.exists(): return prompt_path.read_text(encoding="utf-8") return "" def _load_presets(base_dir: Path): 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)) async def main(): base_dir = Path(__file__).parent setup_logging(level=LOG_LEVEL, file=LOG_FILE) system_prompt = _load_system_prompt(base_dir) _load_presets(base_dir) 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("=" * 50) print(" 广告智能调控助手已启动") print("=" * 50) print("请输入指令（输入 'exit' 退出）：") print("指令示例：") print(" - 分析广告 → 全量分析") print(" - 广告 XXXXX 降价10% → 定向操作") print(" - 广告 XXXXX 不要暂停 → 修改决策") print() step_count = 0 session_trace_id = None # 会话级 trace_id，保持多轮对话记忆 while True: try: user_input = input("\n> ").strip() if not user_input: continue if user_input.lower() in ("exit", "quit", "q"): print("退出系统") break messages = [{"role": "user", "content": user_input}] config.trace_id = session_trace_id print(f"\n🚀 执行: {user_input}") print("=" * 70) print(" 流程：数据拉取 → ROI计算 → 人群包基线 → 候选筛选 → AI推理 → 保存决策 → 护栏验证 → 生成报告") print("=" * 70) print() step_count = 0 async for item in runner.run(messages=messages, config=config): if isinstance(item, Trace): if session_trace_id is None: session_trace_id = item.trace_id if item.status == "completed": print(f"\n✅ [Trace] 完成") elif item.status == "failed": print(f"\n❌ [Trace] 失败") session_trace_id = None # 失败后重置，下次开新会话 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))) # 识别关键步骤 if tool_name == "fetch_creative_data": step_count += 1 print(f"\n{'='*70}") print(f"📌 步骤 {step_count}: 数据拉取") print(f"{'='*70}") elif tool_name == "calculate_roi_metrics": step_count += 1 print(f"\n{'='*70}") print(f"📌 步骤 {step_count}: ROI 计算") print(f"{'='*70}") elif tool_name == "calculate_portfolio_summary": step_count += 1 print(f"\n{'='*70}") print(f"📌 步骤 {step_count}: 人群包基线计算") print(f"{'='*70}") elif tool_name == "get_ads_for_review": step_count += 1 print(f"\n{'='*70}") print(f"📌 步骤 {step_count}: 候选筛选（零消耗/待评估/正常运行）") print(f"{'='*70}") elif tool_name == "query_ad_detail": step_count += 1 print(f"\n{'='*70}") print(f"📌 步骤 {step_count}: 查询广告详情") print(f"{'='*70}") elif tool_name == "apply_decisions": step_count += 1 print(f"\n{'='*70}") print(f"📌 步骤 {step_count}: 保存智能引擎决策") print(f"{'='*70}") elif tool_name == "modify_decisions": step_count += 1 print(f"\n{'='*70}") print(f"📌 步骤 {step_count}: 修改已有决策") print(f"{'='*70}") elif tool_name == "validate_decisions": step_count += 1 print(f"\n{'='*70}") print(f"📌 步骤 {step_count}: 安全护栏验证") print(f"{'='*70}") elif tool_name == "execute_decisions": step_count += 1 print(f"\n{'='*70}") print(f"📌 步骤 {step_count}: 分级执行") print(f"{'='*70}") elif tool_name == "send_approval_request": step_count += 1 print(f"\n{'='*70}") print(f"📌 步骤 {step_count}: IM 审批请求") print(f"{'='*70}") elif tool_name == "generate_report": step_count += 1 print(f"\n{'='*70}") print(f"📌 步骤 {step_count}: 生成最终报告") print(f"{'='*70}") elif tool_name == "check_execution_feedback": step_count += 1 print(f"\n{'='*70}") print(f"📌 步骤 {step_count}: 执行效果检查") print(f"{'='*70}") # 打印简化结果 if isinstance(result, str): text = result else: text = str(result) if len(text) > 500: text = text[:500] + "..." print(f" {text}") else: text = str(content) if len(text) > 300: text = text[:300] + "..." print(f" [工具] {text}") print("\n" + "=" * 50) print("✅ 完成") print("=" * 50) except KeyboardInterrupt: print("\n用户中断，退出") break except Exception as e: print(f"\n❌ 失败: {e}") import traceback traceback.print_exc() if __name__ == "__main__": asyncio.run(main())