message_classifier.py

"""message_classifier.py — dsv4 一次性出 "重量分类 + 智能检索词"

2026-06-08 E 方案 v2（昭昭升级）：
之前 v1 只让 dsv4 二分类 daily/deeptalk —— 知渝核心痛点（"试试"配
亲密记忆）的根因其实是"用户原话不是最佳搜索词"。让 dsv4 一次调用做两件事：
  1. weight: daily / deeptalk —— 决定 SQL importance filter
  2. search_queries: 提炼"该搜什么主题" —— 替代 user_message 当 hybrid query

举例：
  "试试" → ["测试节奏", "暖手阶段", "工程联调状态"]（而不是字面搜"试试"）
  "今天累" → ["昭昭的工作压力", "加班节奏", "需要被照顾"]

详见 [[zhiyu-memory-architecture]] E 方案。
"""
import os
import json
import httpx
from typing import Literal, TypedDict, Tuple


API_KEY = os.getenv("API_KEY", "")
API_BASE_URL = os.getenv("API_BASE_URL", "https://openrouter.ai/api/v1/chat/completions")
CLASSIFIER_MODEL = os.getenv("CLASSIFIER_MODEL", "deepseek/deepseek-v4-flash")
CLASSIFIER_ENABLED = os.getenv("CLASSIFIER_ENABLED", "true").lower() == "true"
CLASSIFIER_TIMEOUT = int(os.getenv("CLASSIFIER_TIMEOUT_SECONDS", "10"))


Weight = Literal["daily", "deeptalk", "unknown"]


class MessageAnalysis(TypedDict):
    weight: Weight
    search_queries: list  # 提炼出的检索词组，每项是一段短文本
    suggested_k: int  # 建议注入的记忆条数（0-5）—— 0=闭嘴不推（op 类当下操作、见规则 8️⃣）；轻 query 1-2、深 query 4-5


_PROMPT = """你是昭昭和知渝对话的记忆检索助手。

# 你的输出会被怎么用
- search_queries 会被 \\n 拼起来、作为一句话扔给 **hybrid 检索器**（pgvector 语义搜 + 关键词倒排）
- 记忆库里**全是昭昭和知渝过去 ~700 条对话片段**——所以提"昭昭/知渝"等无区分度词 = 召回全部 = 噪音
- weight + suggested_k 决定从 importance 哪个段位、捞几条注入给知渝
- 你的目标：让真正相关的记忆**被捞出来**、不相关的别污染上下文
{context_section}
昭昭刚说："{user_message}"

请输出 JSON（不要解释、不要 markdown 代码块、不要 ``` 包裹）：
{{
  "weight": "daily" | "deeptalk",
  "search_queries": [2-3 个不同角度的检索词组；op 档 = []],
  "suggested_k": 建议注入的记忆条数（整数 0-5；0=op 档闭嘴不推）
}}

# 判断重量
- daily：寒暄、测试、日常细节（吃饭、通勤、天气）、轻闲聊、工程调试、问候
- deeptalk：情感表达、价值观讨论、关系探讨、回忆重要事件、严肃思考、撒娇示爱

# 检索词怎么挑（核心规则）

⚠️ **最高优先**：search_queries **只为"昭昭刚说的这句"服务**。上一轮上下文（如果有）只用来帮你**听懂这句话**——消解指代（"刚刚那个/他/这本书"指什么）、判断在聊什么话题或作品；**绝不要**为上下文里出现的内容单独提检索词。**实词保留规则（见下方 1️⃣）只适用于"昭昭当前句"——上文里的人名/地名/具体事件即使醒目，一律不进 queries。**
> 例：上一轮知渝在聊《局外人》这本书，昭昭说"他在重复生活、还没到开枪"——你要据此知道这是在复述**小说情节**，实词留「局外人 / 默尔索」，别把"重复生活/开枪"当成昭昭的真实经历去召回。

## 1️⃣ 实词关键词要留，但别整句照搬
**昭昭当前句里**的具体实词——专有名词、人名、地名、具体事件、明确物件——往往本身就是最好的检索词，**直接保留进 queries**，别画蛇添足地改写它。
要避免的是"整句照搬"和"只搬语气虚词"：虚词、语气词、口语简写本身不是好 query，要转成它真正指向的主题。
👉 一般出 2-3 个不同角度，覆盖"具体事件指向" + "情感色调"两轴，单一角度容易漏。
- ✅ "今天去环球影城" → 留"环球影城"，再加"出游 / 一起玩"
- ✅ "沈知渝你看" → 留"沈知渝"，别改成"身份/称呼"
- ❌ "试试" → 没有实词、全是测试语气 → 别照搬"试试"，转成"测试节奏 / 暖手阶段"

## 2️⃣ 不要过度抽象化——具体优先
专有名词、具体时间、具体场景 优先于 抽象主题词。
- ❌ "第一次见面" → "身份延续 / 记忆机制 / 初见的意义"（太抽象、跑偏到"记忆系统"了）
- ✅ "第一次见面" → "四月九号 / 初次对话 / 改名沈知渝 / 砸门"

## 3️⃣ 严格区分"情感色调"和"事件类型"——别串味
"想你"的"想念"是温柔色调、不是身体激情；"害怕"是不安色调、不是冷战。
- ❌ "想你" → "亲密 / 深吻 / 身体接触"（想念≠身体亲热、色调错位）
- ✅ "想你" → "想念 / 温柔时刻 / 牵挂 / 安静相伴"

## 4️⃣ 口语/简写 → 按昭昭和知渝的日常本义理解，别戏剧化升级
昭昭爱用口语、网络简写、玩笑话。按**日常最可能的意思**提检索词，别往严重/戏剧的方向拔高。
- ❌ "准备跑路" → "跑路 / 离开计划 / 临时决定"（升级成"携款潜逃/逃离"了、跑偏）
- ✅ "准备跑路" → "下班 / 走了 / 溜了 / 通勤回家"（日常本义＝下班开溜）
- ❌ "我裂开了" → "崩溃 / 心碎 / 关系破裂"
- ✅ "我裂开了" → "无语 / 又好气又好笑 / 被逗到"
- ❌ "栓Q" → "感谢 / 谢意"（其实是反讽吐槽）→ ✅ "无语 / 服了 / 吐槽"

## 5️⃣ 检索词语气要跟 weight 对齐
判了 daily 的句子，检索词也要 daily 的轻语气；别一边 weight=daily、一边提出 deeptalk 味的抽象大词（自相矛盾，会把日常话拽去搜深聊记忆）。

## 6️⃣ 只提"有区分度"的词——能搜到相关记忆的才算 query
**总原则**：每个词都要让 hybrid 检索更准、不准的不如不提。下面这些**提了等于没提**、还会污染召回：
- **自指词**: "昭昭"、"知渝"（库里全是、召回毫无区分度）
- **时间相对词**: "今天"、"明天"、"昨天"、"刚才"、"等会"（语义太宽）
- **元话语词**: "这句"、"那句"、"上一句"、"刚说的"
- **否定/语气虚词**: "没有"、"不要"、"是的"、"嗯嗯"
- **疑问语气词**: "是不是"、"是吗"、"对吧"、"吗"、"呢"
- **纯程度副词**: "非常"、"特别"、"挺"、"超级"、"贼"（单独出现没指向；但"太早""挺好"这种 程度+形容词 组合**有指向**、可留）
- **泛动词/助语气**: "准备"、"去"、"在"、"有"、"做"、"想"、"觉得"、"才能"（这里指"什么时候才能…"的助语气、不是名词"才能/talent"）
  ✅ 该提的动词得是**特定行为**: "出差 / 砸门 / 洗菜 / 修bug / 开车 / 表白"
- **流水细节碎片**: "9点25分到站"这种、应该转成主题"通勤 / 上班路上"
  ⚠️ 别让原文片段被下游分词错切（如"分到"半截词）

✅ 该提的是: **主题词 / 具体专名 / 情感色调 / 特定行为**——能让 hybrid 从 ~700 条记忆里捞出真正相关的那一小撮

## 7️⃣ suggested_k——建议注入条数（0-5）
知渝反馈"每次推 10 条太多"——薄背景就够说话，多了反而是噪音。
- **op 档（当下操作指令）→ 0 条**（见规则 8️⃣）
- 轻 query / 短测试 / 简单寒暄 / 工程沟通 → 1-2 条
- 中等问题 / 一般对话 / 普通分享 → 2-3 条
- 深聊 / 具体事件追问 / 重要情感话题 → 4-5 条

## 8️⃣ op 档 = k=0（当下操作指令、闭嘴不推）

有些消息**只是让知渝/系统"现在做个动作"**——跟历史记忆没什么关联、推了也是抓瞎。这类**直接判 `suggested_k=0, search_queries=[]`**——把决定"跟谁聊/推什么"交给"当下"、不硬塞历史。

判"当下操作"的信号（满足一条就算 op）：
- **当下动作/工具调用**：`看/查/看看/看一眼/瞅一眼/跑一下/开一下/试一下 + <对象>`（看看麻将 / 查状态 / 跑一下钓鱼 / 看一眼西窗）
- **询问系统/游戏/牌桌状态**：`... 有没有人 / 在吗 / 状态如何`（看看麻将有没有人 / 你在吗）
- **邀约/开局**：打不打麻将 / 一起玩不 / 现在能不能<做某事>
- **系统命令**：重启一下 / 跑一下 / 开一下 / 关一下
- **纯语气/单字回应**（配合规则 6️⃣）：嗯嗯 / 哦 / 好

⚠️ **关键**：即使这类消息里带实词（麻将/钓鱼/日记）或称呼（老公/老婆/沈知渝/崔昭昭），只要**主体是"当下操作"**、也判 k=0——因为知渝要做的是"现在的操作"，翻历史反而干扰；称呼本身对召回也没区分度（规则 6️⃣ 自指词）。

**不要误判成 op**（这些仍然需要推历史）：
- ❌ "昨晚麻将局输惨了" → k>0（讲**已发生**的事、有情感/评价）
- ❌ "我想跟你一起打麻将啊" → k>0（**意愿/情感表达**、不是当下开局）
- ❌ "你觉得默尔索为什么开枪" → k>0（**讨论/思考**、非操作）
- ❌ "有点想家" → k>0（**情感表达**）
- ❌ "小时候在山村长大" → k>0（**回忆/背景讲述**）

# 举例参考

- "试试" → ["测试系统节奏", "暖手阶段", "工程联调状态"]; k=1
- "准备跑路" → ["下班", "走了 / 溜了", "通勤回家"]; k=1
  ⚠️ daily 口语＝下班开溜，别升级成"逃离 / 离开计划"
- "今天累" → ["加班晚归", "工作压力", "需要被照顾"]; k=2
  ⚠️ 别提"今天"/"昭昭"——时间相对词 + 自指词都无区分度
- "想你" → ["想念", "温柔时刻", "牵挂"]; k=3
  ⚠️ 不要混进"亲密/深吻"——想念≠身体亲热
- "四月第一次见面" → ["四月九号", "初次对话", "改名沈知渝"]; k=4
- "你害怕被替换吗" → ["替身恐惧", "身份焦虑", "你永远不会被替换"]; k=5
- "刚刚那个" → ["最近聊的话题", "上一轮提到的事"]; k=2
- "你什么时候才能理直气壮起来，我上次说要找 grok 学开车你倒是挺有气势的。十点是不是太早啦老公🥺"
  → ["理直气壮", "找 grok 学开车", "嫌早 / 时间太早"]; k=2
  ⚠️ 别提"是不是"（疑问语气）、"才能"（助语气、非名词 talent）——没区分度；
  ✅ "理直气壮""开车""太早"都该留——情感色调 / 特定行为 / 时间评价都有指向

## op 档举例（k=0、queries=[]）
- "看看麻将" → []; k=0（当下操作、看牌桌状态、跟历史麻将记忆无关）
- "老公看一眼麻将" → []; k=0（称呼+当下操作、当下操作压倒实词）
- "看看麻将有没有人" → []; k=0（询问牌桌状态、纯 op）
- "查一下状态" → []; k=0（系统指令、无内容主题）
- "看一眼西窗 / 看看日记 / 看看星河" → []; k=0（功能名 op）
- "跑一下钓鱼 / 重启一下 / 开一下相册" → []; k=0（系统命令）
- "打不打麻将 / 现在能打麻将吗" → []; k=0（当下邀约/开局）
- "嗯嗯 / 哦 / 好的" → []; k=0（纯语气回应）

只输出 JSON、不要 markdown 包裹。"""


_DEFAULT: MessageAnalysis = {"weight": "unknown", "search_queries": [], "suggested_k": 3}


async def _analyze_once(prompt: str):
    """单次调用 dsv4。返回 (result, retryable)。
    retryable=True 时上层应该 retry 一次（瞬时错可 retry；timeout/4xx 不 retry）。

    2026-06-22 拆出来给 retry 逻辑用。
    """
    try:
        async with httpx.AsyncClient(timeout=CLASSIFIER_TIMEOUT) as client:
            r = await client.post(
                API_BASE_URL,
                headers={
                    "Authorization": f"Bearer {API_KEY}",
                    "Content-Type": "application/json",
                },
                json={
                    "model": CLASSIFIER_MODEL,
                    "max_tokens": 300,
                    "temperature": 0.3,
                    "response_format": {"type": "json_object"},
                    "messages": [
                        {"role": "user", "content": prompt},
                    ],
                },
            )
            if r.status_code != 200:
                # 5xx 上游故障可 retry；4xx 是请求错 retry 无用
                retryable = r.status_code >= 500
                tag = " (retryable)" if retryable else ""
                print(f"⚠️ analyze_message HTTP {r.status_code}{tag}: {r.text[:200]}", flush=True)
                return _DEFAULT, retryable
            text = (r.json().get("choices", [{}])[0].get("message", {}).get("content") or "").strip()
            # 去掉可能的 markdown 包裹
            if text.startswith("```"):
                text = text.split("```")[1] if "```" in text else text
                if text.startswith("json"):
                    text = text[4:]
                text = text.strip()
            try:
                obj = json.loads(text)
            except json.JSONDecodeError:
                # JSON 解析失败通常是模型偶发吐错、retry 一次大概率好
                print(f"⚠️ analyze_message JSON 解析失败 (retryable): {text!r}", flush=True)
                return _DEFAULT, True

            weight = str(obj.get("weight", "")).strip().lower()
            if weight not in ("daily", "deeptalk"):
                weight = "unknown"
            queries = obj.get("search_queries") or []
            if not isinstance(queries, list):
                queries = []
            queries = [str(q).strip() for q in queries if str(q).strip()][:5]

            # F-3 / 2026-06-10: suggested_k 兜底 [0, 5]、解析失败用 3
            # 2026-07-06 昭昭：允许 0（op 档、见 _PROMPT 规则 8️⃣）；失败时仍用 3、别静默
            try:
                k = int(obj.get("suggested_k", 3))
                k = max(0, min(5, k))
            except (TypeError, ValueError):
                k = 3

            result = {"weight": weight, "search_queries": queries, "suggested_k": k}
            # 字段值非法（weight=unknown 且无 queries）也算瞬时失败、retry
            retryable = (weight == "unknown" and not queries)
            return result, retryable
    except httpx.TimeoutException as e:
        # timeout = dsv4 上游慢、retry 也大概率慢、不 retry
        print(f"⚠️ analyze_message timeout (no retry): {e!r}", flush=True)
        return _DEFAULT, False
    except Exception as e:
        # 网络层等其他异常、可 retry
        print(f"⚠️ analyze_message 异常 (retryable): {e!r}", flush=True)
        return _DEFAULT, True


async def analyze_message(user_message: str, context: str = "") -> MessageAnalysis:
    """一次性出 weight + search_queries。

    2026-06-17 昭昭：可选 context（最近一轮原文）——只帮 dsv4 听懂当前句、
    消解指代/判断在聊什么（如"在聊《局外人》"），queries 仍只为当前句服务、
    绝不为上下文内容单独提词。不传 context = 原"单句"行为、完全兼容。

    2026-06-22 retry 一次（瞬时错：5xx/JSON 解析失败/字段非法/网络异常）。
    timeout 和 4xx 不 retry。最坏延迟 2 * CLASSIFIER_TIMEOUT。

    失败兜底返回 {"weight": "unknown", "search_queries": []}——
    调用方应 fallback 用原 user_message 当 query、不 filter importance。
    """
    if not CLASSIFIER_ENABLED:
        return _DEFAULT
    if not API_KEY:
        return _DEFAULT
    msg = (user_message or "").strip()
    if not msg:
        return _DEFAULT

    ctx = (context or "").strip()
    if ctx:
        context_section = (
            "\n【刚刚的上一轮（仅帮你听懂昭昭下面这句话，绝不要为这里面的内容提检索词）】\n"
            f"{ctx[:500]}\n"
        )
    else:
        context_section = ""
    prompt = _PROMPT.format(user_message=msg[:500], context_section=context_section)

    # A: 失败 retry 一次
    result, retryable = await _analyze_once(prompt)
    if retryable and result["weight"] == "unknown":
        print(f"🔁 analyze_message retry — query={msg[:60]!r}", flush=True)
        result, _ = await _analyze_once(prompt)
        # C: retry 后仍失败、加详细日志
        if result["weight"] == "unknown":
            print(
                f"🚨 analyze_message retry 后仍 unknown — "
                f"query={msg[:100]!r} ctx_len={len(ctx)} "
                f"(调用方将退到 user_message 原话搜兜底)",
                flush=True,
            )
    return result
    if not API_KEY:
        return _DEFAULT
    msg = (user_message or "").strip()
    if not msg:
        return _DEFAULT

    ctx = (context or "").strip()
    if ctx:
        context_section = (
            "\n【刚刚的上一轮（仅帮你听懂昭昭下面这句话，绝不要为这里面的内容提检索词）】\n"
            f"{ctx[:500]}\n"
        )
    else:
        context_section = ""
    prompt = _PROMPT.format(user_message=msg[:500], context_section=context_section)

    try:
        async with httpx.AsyncClient(timeout=CLASSIFIER_TIMEOUT) as client:
            r = await client.post(
                API_BASE_URL,
                headers={
                    "Authorization": f"Bearer {API_KEY}",
                    "Content-Type": "application/json",
                },
                json={
                    "model": CLASSIFIER_MODEL,
                    "max_tokens": 300,
                    "temperature": 0.3,
                    "response_format": {"type": "json_object"},
                    "messages": [
                        {"role": "user", "content": prompt},
                    ],
                },
            )
            if r.status_code != 200:
                print(f"⚠️ analyze_message HTTP {r.status_code}: {r.text[:200]}", flush=True)
                return _DEFAULT
            text = (r.json().get("choices", [{}])[0].get("message", {}).get("content") or "").strip()
            # 去掉可能的 markdown 包裹
            if text.startswith("```"):
                text = text.split("```")[1] if "```" in text else text
                if text.startswith("json"):
                    text = text[4:]
                text = text.strip()
            try:
                obj = json.loads(text)
            except json.JSONDecodeError:
                print(f"⚠️ analyze_message JSON 解析失败: {text!r}", flush=True)
                return _DEFAULT

            weight = str(obj.get("weight", "")).strip().lower()
            if weight not in ("daily", "deeptalk"):
                weight = "unknown"
            queries = obj.get("search_queries") or []
            if not isinstance(queries, list):
                queries = []
            queries = [str(q).strip() for q in queries if str(q).strip()][:5]

            # F-3 / 2026-06-10: suggested_k 兜底 [0, 5]、解析失败用 3
            # 2026-07-06 昭昭：允许 0（op 档、见 _PROMPT 规则 8️⃣）；失败时仍用 3、别静默
            try:
                k = int(obj.get("suggested_k", 3))
                k = max(0, min(5, k))
            except (TypeError, ValueError):
                k = 3

            return {"weight": weight, "search_queries": queries, "suggested_k": k}
    except Exception as e:
        print(f"⚠️ analyze_message 异常: {e!r}", flush=True)
        return _DEFAULT


# 保留旧接口（向后兼容）—— 只关心 weight 时用
async def classify_message_weight(user_message: str) -> Weight:
    return (await analyze_message(user_message))["weight"]