qihuaiflow/backend/python/app/chat_context.py

# Auto-extracted from chat.py during Phase 1 refactor (Strangler Fig).

# Source of truth: scripts/phase1_extract.py. Do not hand-merge back.



from fastapi import APIRouter, Depends, HTTPException, Query, Response

from sqlalchemy.orm import Session

from sqlalchemy import and_, desc, asc, func

from pydantic import BaseModel, Field

from typing import Optional

import uuid

from datetime import datetime, timezone, timedelta

import re

import json

import asyncio

import math

import httpx

import os

import logging

from ..database import get_db, SessionLocal

from ..models import (







    User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog, ConversationSummary,







    KnowledgeBase, KnowledgeDocument, KnowledgeDocumentChunk, EvolutionSkill,







    EvolutionTrajectory, MarketApp, AppPurchase, AppInstallation,







    MCPServer,







    gen_uuid







)

from ..auth import get_current_user

from ..platform_knowledge import get_platform_knowledge

from ..mcp_client import list_mcp_tools, call_mcp_tool

from ..middleware.log_service import LogService

from ..capability_registry import resolve_auto_model

from app.api.chat_skill import (_bm25_search)

from app.kb_router import kb_router_search

from app.api.chat_utils import (logger)



async def _retrieve_kb_context(db: Session, kb_ids: list, query: str, max_tokens: int = 1000) -> str:







    """







    Stage C: 增强版知识库检索 — 使用BM25排序替代朴素拼接







    1. 获取每个KB的文档







    2. 构建文档列表(content+title+metadata)







    3. BM25检索最相关的段落







    4. 拼接为上下文注入system prompt







    同时叠加 Dify 外部知识库(RAG)检索结果。







    """







    local_context = ""















    if kb_ids:







        kbs = db.query(KnowledgeBase).filter(KnowledgeBase.id.in_(kb_ids)).all()







        if kbs:







            # 收集所有文档段落







            # Stage C.1: 优先使用 KnowledgeDocumentChunk 表的预分段数据







            all_segments = []







            for kb in kbs:







                # 先查chunk表







                chunks = db.query(KnowledgeDocumentChunk).filter(







                    KnowledgeDocumentChunk.knowledge_base_id == kb.id







                ).order_by(KnowledgeDocumentChunk.chunk_index).limit(30).all()















                if chunks:







                    # 使用预分段数据







                    for chunk in chunks:







                        if chunk.content and len(chunk.content.strip()) >= 20:







                            all_segments.append({







                                "content": "[{}/{}/段{}]: {}".format(







                                    kb.name, chunk.document_id[:8], chunk.chunk_index, chunk.content[:500]







                                ),







                                "raw_content": chunk.content,







                                "kb_name": kb.name,







                            })







                else:







                    # 退化: 使用原始文档分段







                    docs = db.query(KnowledgeDocument).filter(







                        KnowledgeDocument.knowledge_base_id == kb.id,







                        KnowledgeDocument.status == "completed"







                    ).order_by(KnowledgeDocument.updated_at.desc()).limit(20).all()















                    for doc in docs:







                        content = doc.content or ""







                        if not content:







                            continue















                        # 如果文档较长, 分段处理 (每段~800字)







                        if len(content) > 800:







                            paragraphs = re.split(r'\n\n+', content)







                            for pi, para in enumerate(paragraphs):







                                if len(para.strip()) >= 20:







                                    all_segments.append({







                                        "content": "[{}/{}/段{}]: {}".format(







                                            kb.name, doc.name, pi+1, para.strip()[:500]







                                        ),







                                        "raw_content": para.strip(),







                                        "kb_name": kb.name,







                                        "doc_name": doc.name,







                                    })







                        else:







                            all_segments.append({







                                "content": "[{}/{}]: {}".format(kb.name, doc.name, content[:500]),







                                "raw_content": content,







                                "kb_name": kb.name,







                                "doc_name": doc.name,







                            })















            if not all_segments:







                # 退化: 使用原始朴素拼接







                kb_names = ", ".join([k.name for k in kbs])







                local_context = f"参考知识库: {kb_names}"







            else:







                # BM25检索最相关的段落







                                # 混合检索：BM25(关键词) + 向量语义 + RRF 融合；失败回退 BM25
                bm25_segs = _bm25_search(query, all_segments, top_k=8)
                bm25_for_router = [
                    {"content": seg.get("raw_content", seg.get("content", "")),
                     "metadata": {"label": seg.get("content", "")}}
                    for seg in bm25_segs
                ]
                merged = None
                try:
                    merged = await kb_router_search(
                        query, kb_ids=kb_ids, top_k=8,
                        bm25_results=bm25_for_router, db_session=db,
                        use_vector=True, use_external=False,
                    )
                except Exception as e:
                    logger.warning(f"[kb] 混合检索失败, 回退 BM25: {e}")
                if merged:
                    ctx_parts = []
                    used = 0
                    for seg in merged:
                        label = (seg.get("metadata") or {}).get("label")
                        c = label if label else seg.get("content", "")
                        t = _estimate_tokens(c)
                        if ctx_parts and used + t > max_tokens:
                            break
                        ctx_parts.append(c)
                        used += t
                    local_context = "\n".join(ctx_parts)
                else:
                    ctx_parts = [seg["content"] for seg in bm25_segs]
                    joined = "\n".join(ctx_parts)
                    if _estimate_tokens(joined) > max_tokens:
                        trunc, u = [], 0
                        for c in ctx_parts:
                            if u + _estimate_tokens(c) > max_tokens:
                                break
                            trunc.append(c)
                            u += _estimate_tokens(c)
                        joined = "\n".join(trunc) + "\n...(内容截断)"
                    local_context = joined















    # 叠加 Dify 外部知识库(RAG)检索







    dify_context = ""







    try:







        from app.dify_kb import get_dify_kb_context







        dify_context = get_dify_kb_context(db, query, top_k=5)







    except Exception as e:







        logger.warning(f"Dify KB context retrieval failed: {e}")















    # 联网检索 (AnySearch) — 实时网页结果注入
    web_context = ""
    try:
        web_context = await _retrieve_web_context(db, query, top_k=3)
    except Exception as e:
        logger.warning(f"web search context failed: {e}")

    parts = []







    if local_context:







        parts.append(local_context)







    if dify_context:







        parts.append("【Dify 知识库检索结果】\n" + dify_context)







    if web_context:
        parts.append("【联网检索结果】\n" + web_context)

    return "\n\n".join(parts)



async def _retrieve_web_context(db: Session, query: str, top_k: int = 3) -> str:
    """联网检索上下文：调用 AnySearch 获取实时网页结果，格式化为引用列表。

    仅在存在已启用的 AnySearch 服务且全局开关开启时生效。失败静默返回空串。
    """
    try:
        # 全局开关（默认开启）：ANYSEARCH_IN_CHAT=0 可关闭
        if os.environ.get("ANYSEARCH_IN_CHAT", "1") != "1":
            return ""
        from app.kb_router import kb_router_search
        hits = await kb_router_search(
            query, top_k=top_k, db_session=db,
            use_web=True, use_vector=False, use_external=False, use_tree=False,
        )
        if not hits:
            return ""
        lines = []
        for h in hits:
            meta = h.get("metadata", {}) or {}
            title = meta.get("title", "") or "(无标题)"
            url = meta.get("url", "")
            snippet = (h.get("content", "") or "").strip()
            if not snippet:
                continue
            line = f"- 【{title}】{url}\n  {snippet}"
            lines.append(line)
        return "\n".join(lines)
    except Exception as e:
        logger.warning(f"web search context failed: {e}")
        return ""


async def _ingest_answer_to_dify(conv_id: str, msg_id: str, content: str):







    """后台任务：将助手回答入库到 Dify 知识库（量化后可供 RAG 检索复用）。"""







    try:







        from app.dify_kb import aingest_text







        from app.database import SessionLocal







        from sqlalchemy import text as _t







        db = SessionLocal()







        try:







            row = db.execute(_t(







                "SELECT api_url, api_key, config FROM external_kb_services "







                "WHERE service_type='dify' AND is_enabled=1 LIMIT 1"







            )).fetchone()







        finally:







            db.close()







        if not row:







            return







        api_url, api_key, config_raw = row







        dataset_name = None







        if config_raw:







            try:







                import json







                cfg = config_raw if isinstance(config_raw, dict) else json.loads(config_raw)







                dataset_name = cfg.get("dataset_name")







            except Exception:







                pass







        await aingest_text(api_url, api_key, content, name=f"chat_{conv_id}_{msg_id}", dataset_name=dataset_name)







    except Exception as e:







        logger.warning(f"后台入库 Dify 失败: {e}")



def _build_system_prompt(







    skills: list,







    agents: list,







    kb_context: str,







    user_content: str,







    mcp: list = None,



    mcp_functions: list = None,



    auto_discovered: bool = False,

    all_mcp: list = None,



) -> tuple:







    """







    动态组装system prompt + function definitions。







    返回: (system_prompt_str, function_definitions_list)







    """







    system_parts = ["你是AIFlow平台的AI助手，具备专业的知识储备和工具调用能力。"]







    







    # 平台知识库注入（始终开启）：确保AI优先使用平台已有能力







    try:







        pk = get_platform_knowledge()







        if pk:







            system_parts.append(f"【平台知识库——优先使用以下知识回答用户问题】\n{pk}")







    except Exception:







        pass







    







    # 回复格式要求：强制使用Markdown







    system_parts.append("回复格式要求：使用 Markdown 格式，层次分明即可。对于简短问题，用 1-2 段自然文字直接回答，不必强行分点、加标题或表格；仅在确实需要分步骤或列举时才使用列表，代码/配置用 \"\"\" 代码块包裹。避免冗余空行，保持紧凑、美观、易读。")















    function_definitions = []















    # Stage B.1: Agent角色注入（Issue 2：让用户询问 agent 能力时，AI 能“看到”该 agent 资料）







    if agents:







        for agent in agents:







            name = agent.get("name", "")







            role = agent.get("role_prompt", "")







            instructions = agent.get("instructions", "")







            desc = agent.get("description", "")







            block = []







            if name:







                block.append(







                    f"用户已选择智能体「{name}」。当用户询问该智能体的功能、能力、用法或相关问题时，"







                    f"请基于以下资料准确作答，不要说“不知道这个智能体”。"







                )







            if role:







                block.append(f"该智能体的角色设定: {role}")







            if desc:







                block.append(f"该智能体的简介: {desc}")







            if instructions:







                block.append(f"该智能体的行为指引: {instructions}")







            if block:







                system_parts.append("\n".join(block))















    # Stage A.2: Skill工具注入







    if skills:







        skill_descs = ", ".join([f"{s['name']}({s.get('category', 'general')})" for s in skills])







        system_parts.append(f"可用技能工具: {skill_descs}")







        if auto_discovered:







            # Issue 8：每轮默认先寻找本账号已安装且平台已部署的技能，多个适用时 AI 自行判断选 1 个最优







            system_parts.append(







                "重要规则: 以上技能为本账号已安装且平台已部署的技能。每轮对话请先判断用户请求是否可由其中某个技能处理；"







                "若多个技能都适用，请自行判断并选择最贴切的 1 个优先调用，不要一次性全部调用。"







                "调用对应工具后将获得执行结果，基于结果生成最终回复。若这些技能均不适用，则直接基于你的知识回答。"







            )







        else:







            system_parts.append(

                "重要规则：\n"

                "1) 当用户请求需要由该技能【执行】某项任务时，优先通过 function calling 调用对应工具；调用后基于工具返回结果生成回复。\n"

                "2) 当用户询问所选技能【本身的介绍、功能、用途、用法、特点】（例如“介绍一下这个技能”“它是什么”“怎么用”“能做什么”）时，"

                "必须基于下方【已选技能资料】中的名称/描述/指令直接作答，不要泛泛介绍平台整体，也不要说“不知道这个技能”。\n"

                "3) 若用户问题既不执行也不询问该技能本身，则按常规方式回答。"

            )















        # 为每个skill构建function definition







        for s in skills:







            # 从skill_content提取指令注入system prompt







            instructions = s.get("instructions", "")







            name = s.get("name", "")

            if not auto_discovered:

                # 显式选中技能：注入可引用的技能资料，便于“介绍/说明”类问题据此作答

                ref_parts = [f"【已选技能资料】名称：{name}"]

                desc = s.get("description", "")

                if desc:

                    ref_parts.append(f"简介：{desc}")

                if instructions and len(instructions) < 1200:

                    ref_parts.append(f"使用指令：{instructions}")

                system_parts.append("\n".join(ref_parts))

            elif instructions and len(instructions) < 800:

                system_parts.append(f"技能「{name}」的详细指令: {instructions}")















            # 构建参数schema







            params = s.get("parameters", {})







            if isinstance(params, dict) and len(params) > 0:







                properties = {}







                required = []







                for pname, pdef in params.items():







                    if isinstance(pdef, dict):







                        properties[pname] = {







                            "type": pdef.get("type", "string"),







                            "description": pdef.get("description", f"传递给{s['name']}的{pname}参数"),







                        }







                        if pdef.get("required", True):







                            required.append(pname)







                    else:







                        properties[pname] = {"type": "string", "description": f"传递给{s['name']}的{pname}参数"}







                        required.append(pname)















                if not properties:







                    properties = {"input": {"type": "string", "description": f"需要传递给{s['name']}的完整用户输入"}}







                    required = ["input"]







            else:







                properties = {"input": {"type": "string", "description": f"需要传递给{s['name']}的完整用户输入"}}







                required = ["input"]















            desc = s.get("description", "")







            category = s.get("category", "general")







            func_desc = f"调用技能「{s['name']}」处理用户请求。使用场景: {desc}。当用户请求涉及{category}类任务时优先调用此工具。"















            function_definitions.append({







                "name": s["func_name"],







                "description": func_desc,







                "parameters": {







                    "type": "object",







                    "properties": properties,







                    "required": required,







                },







            })















    # KB context injection







    if kb_context:







        system_parts.append(f"参考以下知识库内容回答用户问题:\n{kb_context}")















    # 平台全量 MCP 清单（Issue：让用户询问"平台有哪些 MCP"时 AI 能准确列举，而非凭空编造）

    if all_mcp:

        _mcp_inventory = "\n".join([f"- {m.get('name','')}: {m.get('description','') or '（未填写描述）'}" for m in all_mcp])

        system_parts.append(

            f"【平台已安装的 MCP 应用清单（共 {len(all_mcp)} 个，均为平台真实接入并启用）】\n{_mcp_inventory}\n"

            "当用户询问“平台有哪些 MCP / 接入了哪些 MCP / 有哪些工具服务”时，请严格依据以上清单准确列举名称与功能，"

            "不要编造或遗漏，也不要把未列出的服务说成已接入。"

        )



    # MCP 应用注入（Issue 3：让用户询问 MCP 能力时，AI 能“看到”该 MCP 的资料与地址）







    if mcp:







        mcp_lines = []







        for m in mcp:







            name = m.get("name", "")







            desc = m.get("description", "")







            transport = m.get("transport", "stdio")







            endpoint = m.get("endpoint", "")







            line = f"用户已选择 MCP 应用「{name}」（传输方式: {transport}）"







            if desc:







                line += f"，功能简介: {desc}"







            if endpoint:







                line += f"，服务地址: {endpoint}"







            mcp_lines.append(line)







        system_parts.append("\n".join(mcp_lines))







        system_parts.append(







            "用户已选择上述 MCP 应用。当请求需要借助外部工具/系统（如查询数据、调用 API、操作第三方服务）时，"







            "请优先参考这些 MCP 应用的能力来回答；若平台已接入对应工具，优先通过工具调用完成；"







            "不要说“不了解该 MCP”。"







        )



    # 高优先级覆盖规则：显式选中工具时，回答必须基于该工具资料

    if (skills and not auto_discovered) or (mcp):

        system_parts.append(

            "【最高优先级覆盖规则】\n"

            "用户在本轮已明确选择了工具（见上文【已选技能资料】或【用户已选择 MCP 应用】）。\n"

            "如果用户的问题是在询问这些已选工具/技能/应用本身的介绍、功能、能做什么、怎么用、特点或用法，"

            "你必须只基于上文提供的该工具具体资料作答；\n"

            "不要引用《平台知识库》中关于技能市场、MCP 协议或平台整体能力的通用描述；"

            "不要列举其它未选中的技能或 MCP 作为示例；也不要说“不知道”或“无法回答”。\n"

            "若用户问题与已选工具本身无关，则按常规方式回答。"

        )

















    system_prompt = "\n".join(system_parts)







    # MCP function definitions 注入（让 LLM 可通过 function calling 真实调用 MCP 工具）



    if mcp_functions:



        function_definitions.extend(mcp_functions)







    return system_prompt, function_definitions



def _get_conversation_context(db: Session, conv_id: str) -> dict:







    """







    Stage B.2: 获取对话的持久化上下文







    对话级别的agent/skill/kb选择保存在Conversation字段中。







    """







    conv = db.query(Conversation).filter(Conversation.id == conv_id).first()







    if not conv:







        return {}















    # Conversation目前没有存储agent/skill/kb的JSON字段







    # 我们从第一条消息中获取(如果存在), 或返回空







    # 后续可通过数据库迁移添加Conversation.context字段















    # 查询该对话中最早一条含有skill_ids/agent_ids/knowledge_ids的消息







    first_tool_msg = db.query(Message).filter(







        Message.conversation_id == conv_id,







        Message.role == "user",







    ).order_by(asc(Message.created_at)).first()















    context = {}







    if first_tool_msg:







        if first_tool_msg.skill_ids:







            context["skill_ids"] = first_tool_msg.skill_ids







        if first_tool_msg.agent_ids:







            context["agent_ids"] = first_tool_msg.agent_ids







        if first_tool_msg.knowledge_ids:







            context["knowledge_base_ids"] = first_tool_msg.knowledge_ids







        if getattr(first_tool_msg, "mcp_ids", None):







            context["mcp_ids"] = first_tool_msg.mcp_ids















    return context







# ===== Phase 2 上下文引擎升级：分层记忆 + 压缩 + 隔离 =====

logger = logging.getLogger(__name__)





_tiktoken_enc = None

def _estimate_tokens(text):

    """精确 token 估算：优先 tiktoken(cl100k_base)，失败回退 2 字符/token。"""

    if not text:

        return 0

    global _tiktoken_enc

    if _tiktoken_enc is None:

        try:

            import tiktoken

            _tiktoken_enc = tiktoken.get_encoding("cl100k_base")

        except Exception:

            _tiktoken_enc = False

    if _tiktoken_enc:

        try:

            return len(_tiktoken_enc.encode(text))

        except Exception:

            pass

    return max(1, len(text) // 2)





def _build_chat_history(db, conv_id, max_tokens=6000, max_messages=40,
                        exclude_latest_user=True, include_summary=True):
    """会话层上下文引擎：预算感知的历史组装 + 压缩摘要/结构化长期记忆前置。

    - 若 conversation_summaries 存在结构化长期记忆(structured_memory)，以紧凑
      结构化区块前置（Semantica 式：实体/事实/偏好/决策/待办），比扁平摘要更省
      token 且更便于模型检索关键信息。
    - 否则回退到扁平 summary_text 前置。
    - 按 token 估算从新到旧累加，超预算即截断（隔离：不溢出上下文窗口）。
    - exclude_latest_user=True 时丢弃最新一条 user 消息（当前轮次另行发送）。
    """
    ensure_structured_column()
    summary_block = None
    if include_summary:
        try:
            summ = db.query(ConversationSummary).filter(
                ConversationSummary.conversation_id == conv_id
            ).first()
            if summ and summ.structured_memory and isinstance(summ.structured_memory, dict):
                block = _format_structured_memory(summ.structured_memory)
                if block:
                    summary_block = {
                        "role": "system",
                        "content": "【长期记忆·结构化】请据此理解用户此前的偏好、已确认决策与待办：\n" + block,
                    }
            elif summ and summ.summary_text:
                summary_block = {
                    "role": "system",
                    "content": "【对话历史摘要，请据此理解用户此前的意图与已确认结论】\n" + summ.summary_text,
                }
        except Exception:
            pass

    rows = db.query(Message).filter(
        Message.conversation_id == conv_id,
        Message.role.in_(["user", "assistant"]),
    ).order_by(desc(Message.created_at)).limit(max_messages).all()
    if exclude_latest_user:
        for i, m in enumerate(rows):
            if m.role == "user":
                rows = rows[:i] + rows[i + 1:]
                break

    kept = []
    used = 0
    for m in rows:  # 新->旧
        t = _estimate_tokens(m.content or "")
        if kept and used + t > max_tokens:
            break
        kept.append({"role": m.role, "content": m.content})
        used += t

    kept.reverse()  # 时间正序

    out = []
    if summary_block:
        out.append(summary_block)
    out.extend(kept)
    return out


def _format_structured_memory(mem: dict) -> str:
    """将结构化记忆格式化为紧凑可读区块。"""
    labels = {
        "entities": "实体",
        "facts": "事实",
        "preferences": "偏好",
        "decisions": "决策",
        "open_tasks": "待办",
    }
    lines = []
    for key, label in labels.items():
        items = mem.get(key)
        if not items:
            continue
        if isinstance(items, list):
            for it in items:
                if it:
                    lines.append(f"· {label}: {it}")
        elif isinstance(items, str):
            lines.append(f"· {label}: {items}")
    return "\n".join(lines)
async def _summarize_via_llm(prompt):

    """调用默认文本模型生成对话摘要；失败返回 None（压缩为可选项，不阻塞）。"""

    try:

        from ..capability_registry import resolve_auto_model

        from app.api.chat_utils import _proxy_for_url

        sdb = SessionLocal()

        try:

            ai_model = resolve_auto_model("text", sdb)

            if not ai_model:

                return None

            provider = sdb.query(ModelProvider).filter(

                ModelProvider.id == ai_model.provider_id

            ).first()

            api_key = ai_model.api_key or (provider.api_key if provider else None)

            if not provider or not provider.api_base_url or not api_key:

                return None

            api_url = f"{provider.api_base_url.rstrip('/')}/v1/chat/completions"

            body = {

                "model": ai_model.name,

                "messages": [

                    {"role": "system", "content": "你是专业的对话摘要助手，输出简洁中文摘要。"},

                    {"role": "user", "content": prompt},

                ],

                "max_tokens": 512,

                "temperature": 0.2,

            }

            async with httpx.AsyncClient(proxy=_proxy_for_url(api_url), trust_env=False, timeout=60) as client:

                resp = await client.post(

                    api_url,

                    headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},

                    json=body,

                )

            if resp.status_code != 200:

                return None

            data = resp.json()

            return data.get("choices", [{}])[0].get("message", {}).get("content", "").strip()

        finally:

            sdb.close()

    except Exception as e:

        logger.warning(f"[context] summarize llm failed: {e}")

        return None





_STRUCT_COL_READY = False


def ensure_structured_column():
    """幂等地为 conversation_summaries 增加 structured_memory 列（MariaDB 安全）。"""
    global _STRUCT_COL_READY
    if _STRUCT_COL_READY:
        return True
    try:
        from app.database import engine
        from sqlalchemy import text
        with engine.connect() as conn:
            exists = conn.execute(text(
                "SELECT COUNT(*) FROM information_schema.columns "
                "WHERE table_schema = CURRENT_DATABASE() AND table_name='conversation_summaries' "
                "AND column_name='structured_memory'"
            )).scalar()
            if not exists:
                conn.execute(text(
                    "ALTER TABLE conversation_summaries ADD COLUMN structured_memory JSON DEFAULT NULL"
                ))
                conn.commit()
        _STRUCT_COL_READY = True
        return True
    except Exception as e:
        logger.warning(f"[context] ensure structured column failed: {e}")
        return False


async def _recursive_compress(text, depth=0, max_depth=3, budget=700):
    """递归再压缩：若摘要仍超过 token 预算，则进一步浓缩，直至收敛或达最大深度。"""
    if not text:
        return text
    if _estimate_tokens(text) <= budget or depth >= max_depth:
        return text
    s = await _summarize_via_llm("请将以下摘要进一步浓缩，保留全部关键信息，篇幅更短：\n" + text)
    if not s:
        return text
    return await _recursive_compress(s, depth + 1, max_depth, budget)


async def _extract_structured_memory(text):
    """Semantica 式结构化长期记忆抽取：将对话内容抽取为类型化结构。

    模型对抽取类提示偶发返回空，故做最多 3 次重试；并兼容有标签与无标签两种输出。
    """
    if not text or len(text.strip()) < 10:
        return None
    prompt = (
        "提取要点，按 实体/事实/偏好/决策/待办 五类，用分号分隔输出：" + text[:6000]
    )
    raw = None
    for _ in range(3):
        r = await _summarize_via_llm(prompt)
        if r and r.strip():
            raw = r.strip()
            break
    if not raw:
        return None
    # JSON 回退
    if "{" in raw:
        try:
            s = raw.find("{")
            e = raw.rfind("}")
            data = json.loads(raw[s:e + 1])
            if isinstance(data, dict):
                return _normalize_structured(data)
        except Exception:
            pass
    norm = raw.replace("：", ":").replace("；", ";").replace("，", ",").replace("。", "")
    result = _parse_structured_segments(norm)
    if result:
        return result
    # 无标签回退：按分号切分，按顺序映射到五类
    segs = [s.strip() for s in norm.replace(chr(10), ";").split(";") if s.strip()]
    if segs:
        cats = ["entities", "facts", "preferences", "decisions", "open_tasks"]
        fb = {}
        for i, s in enumerate(segs[:5]):
            fb[cats[i]] = [s]
        return fb
    return None


def _parse_structured_segments(norm):
    """解析『类别: 内容』格式（兼容全角冒号/分号与单行多类别）。"""
    result = {}
    for seg in norm.replace(chr(10), ";").split(";"):
        if ":" not in seg:
            continue
        key, _, val = seg.partition(":")
        key = key.strip()
        val = val.strip().strip(",")
        if not val or val in ("无", "none", "None"):
            continue
        mapped = {
            "实体": "entities", "事实": "facts", "偏好": "preferences",
            "决策": "decisions", "待办": "open_tasks",
        }.get(key)
        if mapped:
            items = [x.strip() for x in val.split(",") if x.strip()]
            if items:
                result[mapped] = items
    return result


def _normalize_structured(data):
    """将模型的 JSON 输出规范化为结构化记忆字典。"""
    keys = ["entities", "facts", "preferences", "decisions", "open_tasks"]
    out = {}
    for k in keys:
        v = data.get(k)
        if isinstance(v, list):
            out[k] = [str(x) for x in v if x]
        elif isinstance(v, str) and v:
            out[k] = [v]
    return out


def _merge_structured_memory(old, new):
    """合并新旧结构化记忆，按字符串去重并控制规模。"""
    keys = ["entities", "facts", "preferences", "decisions", "open_tasks"]
    merged = {}
    for k in keys:
        a = old.get(k) if isinstance(old, dict) else None
        b = new.get(k) if isinstance(new, dict) else None
        items = []
        for x in (a or []) + (b or []):
            if x and x not in items:
                items.append(x)
        merged[k] = items[:50]
    return merged


async def _maybe_compress_history(conv_id):
    """后台异步压缩：将受保护窗口外的旧消息递归再压缩，并抽取结构化长期记忆。

    失败静默回滚，绝不阻塞用户本次响应（fire-and-forget）。
    """
    ensure_structured_column()
    sdb = None
    try:
        sdb = SessionLocal()
        all_msgs = sdb.query(Message).filter(
            Message.conversation_id == conv_id,
            Message.role.in_(["user", "assistant"]),
        ).order_by(asc(Message.created_at)).all()
        PROTECTED = 20
        if len(all_msgs) <= PROTECTED:
            return
        compressible = all_msgs[: len(all_msgs) - PROTECTED]
        if len(compressible) < 8:
            return

        # 1) 分块摘要：覆盖全部可压缩消息（而非仅前 30 条）
        CHUNK = 30
        chunk_summaries = []
        for i in range(0, len(compressible), CHUNK):
            piece = compressible[i:i + CHUNK]
            text = "\n".join(f"{m.role}: {m.content}" for m in piece)
            s = await _summarize_via_llm("请将以下对话片段浓缩为简明要点（保留关键事实、诉求与结论）：\n" + text)
            if s:
                chunk_summaries.append(s)

        existing = sdb.query(ConversationSummary).filter(
            ConversationSummary.conversation_id == conv_id
        ).first()
        prior = existing.summary_text if existing else ""

        combined = "\n".join(chunk_summaries)
        if prior:
            combined = prior + "\n" + combined

        # 2) 递归再压缩：合并摘要超预算时逐层再压缩直到收敛
        final_summary = await _recursive_compress(combined)

        # 3) 结构化长期记忆抽取（Semantica 式），并与既有记忆合并
        structured = await _extract_structured_memory(combined)
        if structured is None:
            structured = {}
        if existing and existing.structured_memory and isinstance(existing.structured_memory, dict):
            structured = _merge_structured_memory(existing.structured_memory, structured)

        now = datetime.now()
        if existing:
            existing.summary_text = final_summary
            existing.structured_memory = structured
            existing.token_count = _estimate_tokens(final_summary)
            existing.updated_at = now
        else:
            sdb.add(ConversationSummary(
                conversation_id=conv_id,
                summary_text=final_summary,
                structured_memory=structured,
                token_count=_estimate_tokens(final_summary),
                updated_at=now,
            ))
        sdb.commit()
    except Exception as e:
        logger.warning(f"[context] compress history failed conv={conv_id}: {e}")
        if sdb is not None:
            try:
                sdb.rollback()
            except Exception:
                pass
    finally:
        if sdb is not None:
            try:
                sdb.close()
            except Exception:
                pass