hello-agents/Co-creation-projects/DeLunnLi-PaperGraph/backend/app/services/feedback/negative_feedback_memory.py

"""负反馈记忆 —— 记录用户不感兴趣的论文/主题，优化后续推荐."""

from __future__ import annotations

import json
import sqlite3
import time
from typing import Any

from ..llm.agent_runtime import run_json_task
from ..llm.llm_service import get_llm
from ...utils.common import exec_sql

def ensure_tables(db_path: str) -> None:
    exec_sql(db_path,
        """CREATE TABLE IF NOT EXISTS negative_pref_memory (
          id INTEGER PRIMARY KEY AUTOINCREMENT,
          created_at INTEGER NOT NULL,
          expires_at INTEGER NOT NULL,
          identity_key TEXT,
          title TEXT,
          payload_json TEXT NOT NULL
        )""",
        "CREATE INDEX IF NOT EXISTS idx_negpref_exp ON negative_pref_memory(expires_at)",
        """CREATE TABLE IF NOT EXISTS negative_pref_longterm (
          id INTEGER PRIMARY KEY AUTOINCREMENT,
          kind TEXT NOT NULL,
          value TEXT NOT NULL,
          weight REAL DEFAULT -0.2,
          created_at INTEGER NOT NULL,
          last_triggered_at INTEGER NOT NULL,
          trigger_count INTEGER DEFAULT 0,
          disabled INTEGER DEFAULT 0,
          evidence_json TEXT
        )""",
        "CREATE UNIQUE INDEX IF NOT EXISTS ux_negpref_longterm_kind_val ON negative_pref_longterm(kind, value)",
        "CREATE INDEX IF NOT EXISTS idx_negpref_longterm_disabled ON negative_pref_longterm(disabled, last_triggered_at)",
    )

def _extract_pref_dims(payload: dict[str, Any]) -> dict[str, list[str]]:
    return {
        "topic": [str(x).strip().lower() for x in (payload.get("topics_to_downrank") or []) if str(x).strip()],
        "subdomain": [str(x).strip().lower() for x in (payload.get("subdomains_to_downrank") or []) if str(x).strip()],
        "style": [str(x).strip().lower() for x in (payload.get("styles_to_downrank") or []) if str(x).strip()],
        "venue": [str(x).strip().lower() for x in (payload.get("venues_to_downrank") or []) if str(x).strip()],
        "source": [str(x).strip().lower() for x in (payload.get("sources_to_downrank") or []) if str(x).strip()],
    }

def maybe_promote_longterm_from_recent_skips(
    db_path: str,
    *,
    window_days: int = 30,
    min_count: int = 5,
    min_confidence: float = 0.6,
    max_new_rules: int = 2,
) -> list[tuple[str, str]]:
    ensure_tables(db_path)
    now = int(time.time())
    win = max(7, min(90, int(window_days))) * 86400
    since = now - win

    conn = sqlite3.connect(db_path)
    try:
        cur = conn.cursor()
        cur.execute(
            """SELECT created_at, title, payload_json
            FROM negative_pref_memory
            WHERE created_at >= ? ORDER BY created_at DESC LIMIT 1000""",
            (since,),
        )
        counts: dict[tuple[str, str], int] = {}
        evidences: dict[tuple[str, str], dict[str, Any]] = {}

        for created_at, title, payload_json in cur.fetchall():
            try:
                payload = json.loads(payload_json or "{}")
            except Exception:
                continue
            if not isinstance(payload, dict):
                continue
            conf = float(payload.get("confidence") or 0.0)
            if conf < float(min_confidence):
                continue
            dims = _extract_pref_dims(payload)
            for kind, vals in dims.items():
                for v in vals[:8]:
                    vv = (v or "").strip().lower()[:64]
                    if len(vv) < 2:
                        continue
                    key = (kind, vv)
                    counts[key] = counts.get(key, 0) + 1
                    if key not in evidences:
                        evidences[key] = {
                            "window_days": int(window_days),
                            "min_confidence": float(min_confidence),
                            "example_titles": [],
                            "last_seen_at": int(created_at or 0),
                        }
                    if title and len(evidences[key]["example_titles"]) < 3:
                        evidences[key]["example_titles"].append(str(title)[:160])
                    evidences[key]["last_seen_at"] = max(int(evidences[key]["last_seen_at"]), int(created_at or 0))

        promoted: list[tuple[str, str]] = []
        items = sorted(counts.items(), key=lambda x: x[1], reverse=True)
        for (kind, value), cnt in items:
            if cnt < int(min_count):
                break
            if len(promoted) >= int(max_new_rules):
                break
            weight = -0.2
            ev = evidences.get((kind, value), {})
            ev["count"] = int(cnt)
            cur.execute(
                """INSERT INTO negative_pref_longterm(kind, value, weight, created_at, last_triggered_at, trigger_count, disabled, evidence_json)
                VALUES (?, ?, ?, ?, ?, ?, 0, ?)
                ON CONFLICT(kind, value) DO UPDATE SET
                  last_triggered_at = excluded.last_triggered_at,
                  trigger_count = COALESCE(negative_pref_longterm.trigger_count, 0) + 1,
                  evidence_json = excluded.evidence_json""",
                (kind, value, float(weight), now, int(ev.get("last_seen_at") or now), 1, json.dumps(ev, ensure_ascii=False)),
            )
            promoted.append((kind, value))
        conn.commit()
        return promoted
    finally:
        conn.close()

def record_skip_negative_pref(
    db_path: str,
    *,
    identity_key: str,
    title: str,
    abstract: str | None = None,
    journal: str | None = None,
    source: str | None = None,
    keywords: list[str | None] = None,
    category: str | None = None,
    ttl_days: int = 14,
) -> bool:
    ensure_tables(db_path)
    ttl = max(1, min(60, int(ttl_days or 14)))
    now = int(time.time())
    exp = now + ttl * 86400

    system_prompt = (
        "你是推荐系统的反馈分析器。用户点了「不感兴趣(skip)」。"
        "请把这一次 skip 总结成短期负偏好，用于未来 7-14 天轻微降权（不是硬过滤）。"
        "输出必须是 JSON，字段如下：\n"
        "- topics_to_downrank: string[]（主题关键词，2-8 个）\n"
        "- subdomains_to_downrank: string[]（子领域标签，0-5 个）\n"
        "- venues_to_downrank: string[]（会议/期刊关键词，0-3 个）\n"
        "- sources_to_downrank: string[]（arxiv/openalex/dblp，0-2 个）\n"
        "- styles_to_downrank: string[]（survey/tutorial/benchmark/...，0-3 个）\n"
        "- confidence: number（0-1）\n"
        "规则：宁可少写，避免误伤；不要输出解释文字，只输出 JSON。"
    )
    prompt = json.dumps({
        "title": title, "abstract": (abstract or "")[:1200],
        "journal": journal or "", "source": source or "",
        "keywords": (keywords or [])[:20], "category": category or "",
    }, ensure_ascii=False)
    payload = run_json_task(
        task_name="negative_pref_summarizer", agent_name="neg_pref_summarizer",
        llm=get_llm(), system_prompt=system_prompt, user_prompt=prompt,
        timeout_sec=10.0, retries=1, default={},
    )
    payload_json = json.dumps(payload or {}, ensure_ascii=False)

    conn = sqlite3.connect(db_path)
    try:
        cur = conn.cursor()
        cur.execute(
            """INSERT INTO negative_pref_memory(created_at, expires_at, identity_key, title, payload_json)
            VALUES (?, ?, ?, ?, ?)""",
            (now, exp, (identity_key or "")[:160], (title or "")[:400], payload_json),
        )
        conn.commit()
        return True
    except Exception:
        return False
    finally:
        conn.close()