Spaces:
Runtime error
Runtime error
| """ | |
| 陈策记忆系统：持久化每次分析的核心结论，供后续分析时注入上下文。 | |
| 存储在 data/memory.json，轻量且无外部依赖。 | |
| """ | |
| import json | |
| import logging | |
| from datetime import datetime | |
| from pathlib import Path | |
| from typing import Any | |
| log = logging.getLogger("ad-agent") | |
| _root = Path(__file__).resolve().parent.parent | |
| MEMORY_FILE = _root / "data" / "memory.json" | |
| MAX_RECORDS = 50 | |
| def _ensure_dir(): | |
| MEMORY_FILE.parent.mkdir(parents=True, exist_ok=True) | |
| def _load_all() -> list[dict[str, Any]]: | |
| if not MEMORY_FILE.exists(): | |
| return [] | |
| try: | |
| with open(MEMORY_FILE, "r", encoding="utf-8") as f: | |
| data = json.load(f) | |
| return data if isinstance(data, list) else [] | |
| except Exception as e: | |
| log.warning(f"读取记忆文件失败: {e}") | |
| return [] | |
| def _save_all(records: list[dict[str, Any]]): | |
| _ensure_dir() | |
| records = records[-MAX_RECORDS:] | |
| with open(MEMORY_FILE, "w", encoding="utf-8") as f: | |
| json.dump(records, f, ensure_ascii=False, indent=2) | |
| def save_analysis( | |
| period_label: str, | |
| expert: dict[str, Any] | None, | |
| metrics_snapshot: dict[str, float] | None = None, | |
| ): | |
| """保存一次分析的核心结论到记忆。""" | |
| if not expert: | |
| return | |
| record = { | |
| "timestamp": datetime.now().isoformat(timespec="minutes"), | |
| "period": period_label, | |
| "verdict": expert.get("verdict", ""), | |
| "key_findings": expert.get("key_findings", []), | |
| "alerts": expert.get("alerts", []), | |
| "next_action": expert.get("next_action", ""), | |
| } | |
| if metrics_snapshot: | |
| compact = {} | |
| for k in ["Cost", "Ads.Sales", "ROAS", "CPC", "CTR", "CVR"]: | |
| v = metrics_snapshot.get(k) | |
| if v is not None and v == v: | |
| compact[k] = round(v, 4) if abs(v) < 10 else round(v, 0) | |
| record["metrics"] = compact | |
| records = _load_all() | |
| records.append(record) | |
| _save_all(records) | |
| log.info(f"记忆已保存: {period_label}") | |
| def load_recent(n: int = 5) -> list[dict[str, Any]]: | |
| """加载最近 n 条分析记录。""" | |
| records = _load_all() | |
| return records[-n:] | |
| def build_memory_context(n: int = 5) -> str | None: | |
| """ | |
| 构建记忆上下文字符串，供注入 Prompt。 | |
| 返回 None 表示无历史记录。 | |
| """ | |
| records = load_recent(n) | |
| if not records: | |
| return None | |
| lines = ["【陈策历史记录】（你之前的分析结论，可引用、追踪、自我修正）"] | |
| for r in records: | |
| period = r.get("period", "?") | |
| verdict = r.get("verdict", "") | |
| action = r.get("next_action", "") | |
| alerts = r.get("alerts", []) | |
| parts = [f"- {period}: {verdict}"] | |
| if alerts: | |
| alert_strs = [] | |
| for a in alerts[:2]: | |
| msg = a.get("msg", "") | |
| level = a.get("level", "") | |
| icon = {"red": "🔴", "green": "🟢"}.get(level, "🟡") | |
| alert_strs.append(f"{icon}{msg}") | |
| parts.append(f" 预警: {'; '.join(alert_strs)}") | |
| if action: | |
| parts.append(f" 建议: {action}") | |
| lines.append("\n".join(parts)) | |
| return "\n".join(lines) | |