Spaces:
Runtime error
Runtime error
File size: 3,238 Bytes
3291500 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | """
陈策记忆系统：持久化每次分析的核心结论，供后续分析时注入上下文。
存储在 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)
|