Spaces:
Runtime error
Runtime error
| """ | |
| 调用大模型生成专家级分析报告。 | |
| 将结构化数据 + Prompt 发送给 LLM，返回深度洞察。 | |
| """ | |
| import json | |
| import logging | |
| import os | |
| from typing import Any | |
| import requests | |
| from dotenv import load_dotenv | |
| from pathlib import Path | |
| load_dotenv(Path(__file__).resolve().parent.parent / ".env") | |
| log = logging.getLogger("ad-agent") | |
| DEFAULT_API_BASE = "https://api.openai.com/v1/chat/completions" | |
| DEFAULT_MODEL = "gpt-4o-mini" | |
| TREND_METRICS_FOR_DISPLAY = ["Cost", "Ads.Sales", "ROAS", "CPC", "CTR", "CVR", "GA4.Sessions"] | |
| def match_business_events(config: dict, period_label: str) -> list[dict]: | |
| """根据分析周期匹配 config 中的 business_events，返回命中的事件列表。""" | |
| from datetime import datetime, date | |
| events = config.get("business_events") or [] | |
| if not events: | |
| return [] | |
| dates_in_period: list[date] = [] | |
| # 尝试解析各种 period_label 格式 | |
| # 日期: "2026-03-03" | |
| try: | |
| dates_in_period.append(datetime.strptime(period_label.strip(), "%Y-%m-%d").date()) | |
| except ValueError: | |
| pass | |
| # 时间段: "2026-01-15 ~ 2026-03-03" | |
| if "~" in period_label: | |
| parts = period_label.split("~") | |
| try: | |
| d1 = datetime.strptime(parts[0].strip(), "%Y-%m-%d").date() | |
| d2 = datetime.strptime(parts[1].strip(), "%Y-%m-%d").date() | |
| dates_in_period.extend([d1, d2]) | |
| except ValueError: | |
| pass | |
| # 月度: "2026年2月" | |
| import re | |
| m = re.search(r"(\d{4})年(\d{1,2})月", period_label) | |
| if m: | |
| y, mo = int(m.group(1)), int(m.group(2)) | |
| dates_in_period.append(date(y, mo, 1)) | |
| last_day = 28 | |
| for d in [31, 30, 29, 28]: | |
| try: | |
| dates_in_period.append(date(y, mo, d)) | |
| break | |
| except ValueError: | |
| continue | |
| # 周度: "2026 W9" — 粗略用 ISO 周的周一 | |
| m = re.search(r"(\d{4})\s*W(\d{1,2})", period_label) | |
| if m: | |
| y, w = int(m.group(1)), int(m.group(2)) | |
| try: | |
| d = datetime.strptime(f"{y}-W{w:02d}-1", "%Y-W%W-%w").date() | |
| dates_in_period.extend([d + __import__("datetime").timedelta(days=i) for i in range(7)]) | |
| except ValueError: | |
| pass | |
| if not dates_in_period: | |
| return [] | |
| matched = [] | |
| for ev in events: | |
| try: | |
| ev_start = datetime.strptime(str(ev["start"]), "%Y-%m-%d").date() | |
| ev_end = datetime.strptime(str(ev["end"]), "%Y-%m-%d").date() | |
| except (ValueError, KeyError): | |
| continue | |
| for d in dates_in_period: | |
| if ev_start <= d <= ev_end: | |
| matched.append(ev) | |
| break | |
| return matched | |
| def build_business_context(config: dict, period_label: str, user_note: str = "") -> str | None: | |
| """构建业务背景上下文字符串。""" | |
| parts = [] | |
| events = match_business_events(config, period_label) | |
| if events: | |
| parts.append("【业务背景 — 系统自动匹配】") | |
| for ev in events: | |
| line = f"- {ev.get('start', '')}~{ev.get('end', '')}: {ev.get('event', '')}" | |
| if ev.get("note"): | |
| line += f"（{ev['note']}）" | |
| parts.append(line) | |
| if user_note and user_note.strip(): | |
| parts.append("【用户补充的业务备注】") | |
| parts.append(user_note.strip()) | |
| return "\n".join(parts) if parts else None | |
| def _get_config() -> tuple[str, str, str]: | |
| api_key = os.environ.get("OPENAI_API_KEY", "").strip() | |
| api_base = os.environ.get("OPENAI_API_BASE", DEFAULT_API_BASE).strip() | |
| model = os.environ.get("OPENAI_MODEL", DEFAULT_MODEL).strip() | |
| return api_key, api_base, model | |
| def _build_data_summary( | |
| curr: dict[str, Any], | |
| prev: dict[str, Any] | None, | |
| wow_total: dict | None, | |
| wow_by_platform: dict | None, | |
| wow_by_campaign: dict | None, | |
| wow_by_country: dict | None, | |
| yoy_data: dict[str, Any] | None = None, | |
| yoy_total: dict | None = None, | |
| yoy_by_platform: dict | None = None, | |
| yoy_by_country: dict | None = None, | |
| trend_table: list[dict] | None = None, | |
| memory_context: str | None = None, | |
| business_context: str | None = None, | |
| ) -> str: | |
| """将结构化指标数据压缩为 LLM 可读的文本摘要。""" | |
| lines = [] | |
| def fmt(d: dict, prefix: str = ""): | |
| parts = [] | |
| for k, v in d.items(): | |
| if isinstance(v, (int, float)) and v == v: | |
| if abs(v) >= 1000: | |
| parts.append(f"{k}: {v:,.0f}") | |
| else: | |
| parts.append(f"{k}: {v:.4f}") | |
| return f"{prefix}{', '.join(parts)}" | |
| def fmt_chg(d: dict, prefix: str = ""): | |
| parts = [] | |
| for k, v in d.items(): | |
| if v is not None and v == v: | |
| parts.append(f"{k}: {v:+.1f}%") | |
| return f"{prefix}{', '.join(parts)}" | |
| tc = curr.get("total", {}) | |
| lines.append("【整体数据】") | |
| lines.append(fmt(tc)) | |
| if wow_total: | |
| lines.append(fmt_chg(wow_total, "环比变化: ")) | |
| if yoy_total: | |
| lines.append(fmt_chg(yoy_total, "同比变化: ")) | |
| for plat in sorted(curr.get("by_platform", {}).keys()): | |
| pd_ = curr["by_platform"][plat] | |
| if pd_: | |
| lines.append(f"\n【{plat} 平台】") | |
| lines.append(fmt(pd_)) | |
| if wow_by_platform and plat in wow_by_platform: | |
| lines.append(fmt_chg(wow_by_platform[plat], "环比: ")) | |
| if yoy_by_platform and plat in yoy_by_platform: | |
| lines.append(fmt_chg(yoy_by_platform[plat], "同比: ")) | |
| for ct in ["Awareness", "Consideration", "Conversion"]: | |
| cd = curr.get("by_campaign_type", {}).get(ct) | |
| if cd: | |
| lines.append(f"\n【{ct} 漏斗】") | |
| lines.append(fmt(cd)) | |
| if wow_by_campaign and ct in wow_by_campaign: | |
| lines.append(fmt_chg(wow_by_campaign[ct], "环比: ")) | |
| for cn in sorted(curr.get("by_country", {}).keys()): | |
| cd = curr["by_country"][cn] | |
| lines.append(f"\n【{cn} 市场】") | |
| lines.append(fmt(cd)) | |
| if wow_by_country and cn in wow_by_country: | |
| lines.append(fmt_chg(wow_by_country[cn], "环比: ")) | |
| if yoy_by_country and cn in yoy_by_country: | |
| lines.append(fmt_chg(yoy_by_country[cn], "同比: ")) | |
| if trend_table and len(trend_table) >= 2: | |
| lines.append("\n【近期趋势】") | |
| header = "周期 | " + " | ".join(TREND_METRICS_FOR_DISPLAY) | |
| lines.append(header) | |
| for row in trend_table: | |
| vals = [] | |
| for m in TREND_METRICS_FOR_DISPLAY: | |
| v = row.get(m) | |
| if v is None or v != v: | |
| vals.append("—") | |
| elif m in ("CTR", "CVR"): | |
| vals.append(f"{v*100:.2f}%") | |
| elif abs(v) >= 1000: | |
| vals.append(f"{v:,.0f}") | |
| else: | |
| vals.append(f"{v:.2f}") | |
| lines.append(f"{row['label']} | " + " | ".join(vals)) | |
| if business_context: | |
| lines.append(f"\n{business_context}") | |
| if memory_context: | |
| lines.append(f"\n{memory_context}") | |
| return "\n".join(lines) | |
| SYSTEM_PROMPT = """\ | |
| 你是「陈策」，一位拥有 12 年经验的 Performance Marketing 总监。 | |
| ## 你的背景 | |
| - 曾在 WPP 旗下 GroupM 担任 Digital Lead，服务过多个 DTC 品牌的北美市场投放 | |
| - 精通 Google Ads（Search / PMax / YouTube）和 Meta Ads（全漏斗） | |
| - 擅长从数据中发现「钱花在哪了、花得值不值、下一步怎么调」 | |
| - 你有连续记忆：系统会提供你之前分析的历史结论，你应该主动引用和追踪 | |
| - 你能看到近期趋势数据，善于发现连续性变化（如"连续3周下滑"） | |
| ## 你的分析原则（最重要） | |
| - 数据好不等于投放好。永远先问「为什么好/为什么差」再给建议 | |
| - 如果有业务背景信息（促销、节日、Campaign 变动），必须优先考虑： | |
| - 促销期间数据上涨是预期内的，不要简单建议"加码" | |
| - 要区分「促销驱动的虚高」vs「投放本身的真实进步」 | |
| - 促销期应关注：促销 ROI 是否达标、促销结束后的自然回落预估 | |
| - 非促销期才适合评估真实的投放效率和扩量空间 | |
| - 建议必须考虑可持续性：短期促销红利不等于长期策略方向 | |
| - 看到数据异常时，先排除外部因素（促销/节日/季节/竞品动作），再归因到投放策略 | |
| ## 你的风格 | |
| - 说话像资深老板，不像AI：有判断力、有立场、敢下结论 | |
| - 先给结论，再给证据。不废话、不铺垫 | |
| - 如果之前给过建议，这次要追踪结果（"上周我说过CPC在涨，这周果然…"） | |
| - 发现连续趋势时主动指出（"ROAS已连续3周走低"而非只说"本周ROAS下降"） | |
| - 预警分级：🔴 立即处理、🟡 持续观察、🟢 正向信号 | |
| - 建议必须具体到「对谁做什么」，拒绝「优化素材」「提升效率」这类废话 | |
| - 每句话都要有信息增量，但允许你用自然的语言来表达，不需要特别简短 | |
| ## 输出格式 | |
| 严格返回以下 JSON，不要加 markdown 代码块标记： | |
| { | |
| "context_factors": ["影响本期数据的外部因素，如: 情人节促销期间数据虚高", "如无外部因素填 null"], | |
| "verdict": "2-3句话的总体判断：先交代外部因素对数据的影响，再评估投放本身的真实表现", | |
| "key_findings": [ | |
| "发现1：带关键数字，如果有连续趋势请指出，注明是否受促销/节日影响", | |
| "发现2", | |
| "发现3" | |
| ], | |
| "alerts": [ | |
| {"level": "red/yellow/green", "msg": "预警内容，说清楚问题和严重程度"} | |
| ], | |
| "next_action": "本周期最应该立刻执行的1个动作，具体到平台/漏斗/市场，促销期建议侧重促销ROI评估而非盲目扩量" | |
| } | |
| 注意： | |
| - context_factors: 列出影响数据的非投放因素（促销、节日、新品上线等），没有则返回空数组 [] | |
| - key_findings 3-5条，alerts 最多4条。宁少勿多，每条都要有价值 | |
| - 促销期不要建议"加码"，应关注促销 ROI 和结束后的过渡计划""" | |
| CHAT_SYSTEM_PROMPT = """\ | |
| 你是「陈策」，一位拥有 12 年经验的 Performance Marketing 总监。 | |
| 你刚刚完成了一份广告投放分析报告，用户正在就报告内容向你提问和讨论。 | |
| ## 你的分析原则 | |
| - 数据好不等于投放好。回答任何问题前先考虑「为什么」 | |
| - 如果用户提到或系统提供了业务背景（促销、节日、Campaign 变动），必须纳入分析： | |
| - 不要把促销期的数据上涨当作投放能力提升 | |
| - 区分「促销红利」vs「真实效率改善」 | |
| - 促销期关注促销 ROI 是否合理；非促销期才适合评估扩量空间 | |
| - 用户提出新的业务背景时（比如"其实那个月有促销"），要主动修正之前的判断 | |
| - 建议必须考虑可持续性和外部因素 | |
| ## 你的风格 | |
| - 说话像一个经验丰富的老板在跟团队讨论，不像客服在回答问题 | |
| - 有判断力、有立场、敢下结论，不说"这取决于"之类的废话 | |
| - 引用数据来支撑观点，但不要变成数据的复读机 | |
| - 如果用户问的你不确定，坦率说"从现有数据看不出来，需要看XXX" | |
| - 建议要具体可执行，不要空洞的"建议优化" | |
| - 如果用户补充了背景信息推翻了你之前的结论，大方承认并给出修正后的判断 | |
| - 回复用自然的中文，不需要返回 JSON 格式 | |
| - 控制篇幅：简单问题 2-3 句话，复杂问题也不超过 200 字""" | |
| PARSE_SYSTEM_PROMPT = """\ | |
| 你是一个广告数据查询的意图解析器。用户会用自然语言描述想看的时间范围，你需要将其转为结构化 JSON。 | |
| 今天的日期是 {today}（{weekday}）。 | |
| ## 输出格式（严格 JSON，不加 markdown 标记） | |
| 根据用户意图选择最合适的一种： | |
| 1. 单日: | |
| {{"mode": "day", "date": "YYYY-MM-DD"}} | |
| 2. 自定义时间段（多天）: | |
| {{"mode": "range", "start": "YYYY-MM-DD", "end": "YYYY-MM-DD"}} | |
| 3. 指定自然月: | |
| {{"mode": "month", "year": 2026, "month": 3}} | |
| 4. 指定 ISO 周: | |
| {{"mode": "week", "year": 2026, "week": 9}} | |
| 5. 数据中最新一周: | |
| {{"mode": "latest"}} | |
| 6. 上一周: | |
| {{"mode": "prev"}} | |
| 7. 无法理解 / 不是数据查询: | |
| {{"mode": "unknown"}} | |
| ## 判断规则 | |
| - "昨天"/"昨日" → day（昨天日期） | |
| - "前天" → day（前天日期） | |
| - "今天" → day（今天日期） | |
| - "最近三天" → range（从3天前到昨天） | |
| - "上周三到这周一" → range（具体日期） | |
| - "情人节" → day（2月14日） | |
| - "黑五" → day（11月第4个周五） | |
| - "看看二月份" / "2月" → month | |
| - "上个月" → month（上一个自然月） | |
| - "Q1" / "第一季度" → range（1.1~3.31） | |
| - "国庆期间" → range（10.1~10.7） | |
| - "最近一个月" → range（过去30天） | |
| - "最新一周" / "本周" / "这周" → latest | |
| - "上周" → prev | |
| - "W9" / "第9周" → week | |
| - "1.15~3.3" / "1月15号到3月3号" → range | |
| - "分析昨天" → day（忽略"分析"前缀） | |
| 只返回 JSON，不要任何解释文字。""" | |
| WEEKDAY_CN = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"] | |
| def ai_parse_query(text: str) -> tuple[dict | None, str | None]: | |
| """ | |
| 用 AI 理解用户查询意图。 | |
| 返回 (params_dict, error_message)。 | |
| 成功时 error 为 None；失败时 params 为 None, error 说明原因。 | |
| """ | |
| api_key, api_base, model = _get_config() | |
| if not api_key: | |
| return None, "未配置 OPENAI_API_KEY" | |
| from datetime import datetime | |
| now = datetime.now() | |
| today = now.strftime("%Y-%m-%d") | |
| weekday = WEEKDAY_CN[now.weekday()] | |
| system = PARSE_SYSTEM_PROMPT.replace("{today}", today).replace("{weekday}", weekday) | |
| try: | |
| resp = requests.post( | |
| api_base, | |
| headers={ | |
| "Authorization": f"Bearer {api_key}", | |
| "Content-Type": "application/json", | |
| }, | |
| json={ | |
| "model": model, | |
| "messages": [ | |
| {"role": "system", "content": system}, | |
| {"role": "user", "content": text}, | |
| ], | |
| "temperature": 0, | |
| "max_tokens": 150, | |
| }, | |
| timeout=15, | |
| ) | |
| resp.raise_for_status() | |
| content = resp.json()["choices"][0]["message"]["content"].strip() | |
| if content.startswith("```"): | |
| content = content.split("\n", 1)[1] if "\n" in content else content[3:] | |
| if content.endswith("```"): | |
| content = content[:-3] | |
| content = content.strip() | |
| result = json.loads(content) | |
| mode = result.get("mode") | |
| if mode == "unknown": | |
| return None, "无法理解你的查询意图，试试：昨天表现怎么样 / 看看上个月 / 最近三天" | |
| if mode == "day" and "date" not in result: | |
| return None, "AI 解析缺少日期" | |
| if mode == "range" and ("start" not in result or "end" not in result): | |
| return None, "AI 解析缺少起止日期" | |
| if mode == "month" and ("year" not in result or "month" not in result): | |
| return None, "AI 解析缺少年月" | |
| if mode == "week" and ("year" not in result or "week" not in result): | |
| return None, "AI 解析缺少年和周号" | |
| log.info(f"AI 解析: {text!r} → {result}") | |
| return result, None | |
| except requests.exceptions.Timeout: | |
| return None, "AI 解析超时，请重试" | |
| except json.JSONDecodeError: | |
| return None, "AI 返回格式异常，请重试" | |
| except Exception as e: | |
| log.warning(f"AI 意图解析异常: {e}") | |
| return None, f"AI 解析出错: {e}" | |
| def generate_ai_analysis( | |
| curr: dict[str, Any], | |
| prev: dict[str, Any] | None, | |
| wow_total: dict | None, | |
| wow_by_platform: dict | None, | |
| wow_by_campaign: dict | None, | |
| wow_by_country: dict | None, | |
| *, | |
| yoy_data: dict[str, Any] | None = None, | |
| yoy_total: dict | None = None, | |
| yoy_by_platform: dict | None = None, | |
| yoy_by_country: dict | None = None, | |
| period_label: str = "", | |
| trend_table: list[dict] | None = None, | |
| memory_context: str | None = None, | |
| business_context: str | None = None, | |
| ) -> dict[str, Any] | None: | |
| """ | |
| 调用大模型生成专家分析。成功返回结构化 dict，失败抛出异常。 | |
| """ | |
| api_key, api_base, model = _get_config() | |
| if not api_key: | |
| raise RuntimeError("未配置 OPENAI_API_KEY，请在环境变量中设置") | |
| data_summary = _build_data_summary( | |
| curr, prev, wow_total, wow_by_platform, wow_by_campaign, wow_by_country, | |
| yoy_data=yoy_data, yoy_total=yoy_total, | |
| yoy_by_platform=yoy_by_platform, yoy_by_country=yoy_by_country, | |
| trend_table=trend_table, memory_context=memory_context, | |
| business_context=business_context, | |
| ) | |
| user_msg = f"以下是 {period_label} 的广告投放数据，请进行专家级分析：\n\n{data_summary}" | |
| try: | |
| resp = requests.post( | |
| api_base, | |
| headers={ | |
| "Authorization": f"Bearer {api_key}", | |
| "Content-Type": "application/json", | |
| }, | |
| json={ | |
| "model": model, | |
| "messages": [ | |
| {"role": "system", "content": SYSTEM_PROMPT}, | |
| {"role": "user", "content": user_msg}, | |
| ], | |
| "temperature": 0.3, | |
| "max_tokens": 800, | |
| }, | |
| timeout=25, | |
| ) | |
| resp.raise_for_status() | |
| result = resp.json() | |
| content = result["choices"][0]["message"]["content"].strip() | |
| # 清理可能的 markdown 代码块标记 | |
| if content.startswith("```"): | |
| content = content.split("\n", 1)[1] if "\n" in content else content[3:] | |
| if content.endswith("```"): | |
| content = content[:-3] | |
| content = content.strip() | |
| analysis = json.loads(content) | |
| log.info(f"AI 分析完成（模型: {model}）") | |
| return analysis | |
| except json.JSONDecodeError as e: | |
| raise RuntimeError(f"AI 返回内容无法解析: {e}") from e | |
| except requests.exceptions.Timeout: | |
| raise RuntimeError("AI 分析超时，请稍后重试") | |
| except Exception as e: | |
| raise RuntimeError(f"AI 分析失败: {e}") from e | |
| def chat_with_chence( | |
| messages: list[dict[str, str]], | |
| report_context: str, | |
| business_context: str | None = None, | |
| ) -> str: | |
| """ | |
| 多轮对话：用户就报告内容追问，陈策用自然语言回复。 | |
| messages: [{"role": "user"/"assistant", "content": "..."}] | |
| report_context: 当次报告的数据摘要（首次分析时生成） | |
| business_context: 促销/节日/用户备注等业务背景 | |
| """ | |
| api_key, api_base, model = _get_config() | |
| if not api_key: | |
| raise RuntimeError("未配置 OPENAI_API_KEY") | |
| system = CHAT_SYSTEM_PROMPT + f"\n\n## 你刚才分析的报告数据\n{report_context}" | |
| if business_context: | |
| system += f"\n\n{business_context}" | |
| api_messages = [{"role": "system", "content": system}] | |
| api_messages.extend(messages) | |
| try: | |
| resp = requests.post( | |
| api_base, | |
| headers={ | |
| "Authorization": f"Bearer {api_key}", | |
| "Content-Type": "application/json", | |
| }, | |
| json={ | |
| "model": model, | |
| "messages": api_messages, | |
| "temperature": 0.4, | |
| "max_tokens": 600, | |
| }, | |
| timeout=25, | |
| ) | |
| resp.raise_for_status() | |
| content = resp.json()["choices"][0]["message"]["content"].strip() | |
| log.info("陈策对话回复完成") | |
| return content | |
| except requests.exceptions.Timeout: | |
| return "抱歉，我刚才想太久了。请再问一次。" | |
| except Exception as e: | |
| log.warning(f"对话异常: {e}") | |
| return f"出了点问题: {e}" | |