refactor(context): Phase2 上下文引擎升级 分层记忆+压缩+隔离 [v2026.07.19.4]

VERSION

@@ -1,43 +1 @@
-# 七虎AI PaaS 平台 — 版本清单 (VERSION)
+v2026.07.19.4
-# 版本规范见 docs/VERSIONING.md
-# 版本号格式: vYYYY.MM.DD.N   (年.月.日.当日递增序号，从 1 开始)
-# 环境标签:   prod = 正式版 | test = 测试版
-
-VERSION=v2026.07.19.3
-ENV=prod
-DATE=2026-07-19
-BASE_IMAGE=qihu-aiflow:with-deps
-BACKUP_IMAGE=qihu-aiflow:backup-20260719-003
-
-## 本版本变更 (v2026.07.19.3 / prod)
-- Phase2.2 成本归因：token_usage_logs 新增 trace_id 列（迁移 020_add_trace_id_token_usage_logs.sql，幂等 ALTER），打通「链路 -> 成本」归集
-- _write_media_token_log（媒体图片/视频）注入 get_trace_id()，每条媒体 TokenUsageLog 现可回溯到请求链路
-- LogService.log_token_usage 增加 trace_id 参数（为后续文本 LLM token 记账接通成本归因预留）
-- chat_service / chat_media 在 LLM 补全、MCP 工具拉取/调用、技能执行、AGNES 媒体生成处加 span 子链路（async with span），一次请求可下钻耗时分布
-- 注意：_generate_media_task 为后台 fire-and-forget，其 span 会因处于 asyncio.create_task 表达式内而非法，已移除；真实媒体耗时由 chat_media 内 media.agnes_call span 覆盖
-- 纯增量埋点，不改接口返回结构与业务行为；pytest 9 passed + vitest 4 passed 门禁绿
-- 部署前已按 SOP 备份 rollback 点 qihu-aiflow:backup-20260719-003
-
-## 本版本变更 (v2026.07.19.1 / prod)
-- Phase1 chat.py 巨石拆分收官：10070 行单体拆分为 chat_{utils,media,agents,skill,mcp,context,knowledge,service}.py 共 8 个子模块，controller 层 chat.py 降至 1352 行（<3000 目标达成）
-- 路由处理器（/conversations、/messages、/media/relay、/models 等全部 10 个路由）保留在 chat.py，业务/服务函数迁至子模块并通过 import 再导出
-- 关键修复：此前拆分误删了全部 @router 装饰器导致路由为 0（重启即全 404），本版本以 AST 比对原始提交恢复装饰器，并补回 chat_service._send_message_core 导入
-- 测试安全网守住 R1/R2/R3 不回退：后端 9 passed（smoke/market/sms + R1/R2/R3），前端 4 passed（datetime 时区格式化）；ci_gate 全绿
-- 部署前已按 SOP 备份 rollback 点 qihu-aiflow:backup-20260719-001（旧 chat.py 10070 行）
-
-## 本版本变更 (v2026.07.19.2 / prod)
-- Phase2.1 统一可观测基座：新增 app/core/observability.py（trace_id contextvar + TraceIdFilter + span 计时器 + 可选 JSON formatter），零侵入为全平台日志注入链路 ID
-- main.py APITrackingMiddleware 复用既有 request_id 作为 trace_id，并为每个响应注入 X-Trace-Id 响应头（前端可关联排查）
-- config.py setup_file_logging 接入 TraceIdFilter 并将日志格式升级为 [tid:%(trace_id)s]，所有 getLogger(__name__) 日志自动带链路 ID
-- chat.py send_message 路由用 span 包裹 _send_message_core（耗时+trace_id+conv_id/user_id），覆盖 LLM/MCP/媒体生成整体耗时，为后续成本归因建立回归基线
-- 纯增量埋点，不改任何接口返回结构与业务行为；向后兼容既有 logging 用法
-- 部署前已按 SOP 备份 rollback 点 qihu-aiflow:backup-20260719-002
-
-## 历史
-- v2026.07.18.5 (prod)：Phase0 收官 — 最小 CI 门禁（ci_gate.py + pre-push 钩子 + 休眠 Gitea workflow）。后端 9 passed，前端 4 passed。
-- v2026.07.18.4 (prod)：前端 vitest 基建落地（4 passed，datetime 时区格式化）。
-- v2026.07.18.3 (prod)：回归用例 R1/R2/R3（9 passed）— 微信Bot截断、技能/MCP覆盖规则、媒体生成超时回写。
-- v2026.07.18.2 (prod)：后端 pytest 基建落地（6 passed）— 冒烟/市场14类/短信宝MD5。
-- v2026.07.18.1 (prod)：Phase0 启动 — 部署前镜像备份 SOP + 版本号规范 + 引入测试基建。
-- 旧方案：镜像曾用 AIFlow v2.0.1 / v2026.07.14 / v2026.07.14-nav；git 曾用 v2026.07.12-recover / v2026.07.12-ui-recover / v2026.07.13-recover。自 v2026.07.18.1 起统一为 vYYYY.MM.DD.N + prod/test，便于正式版推广后与测试版做环境切割。
-- 历史功能：AI聊天(real API + tool-call function calling)、知识库文件存储 + AI 图谱、频道二维码、邮件/短信配置 + AI 生成、支付配置、平台驾驶舱(token 累计 + 审计/操作日志)、Agent 资源 API、evolution pipeline 自动恢复、全局默认模型、广告位图片上传 + AI 绘图、应用市场安装/卸载/搜索/去重、资产文件类型分类 + 在线预览。迁移 001-015。

backend/python/app/api/chat_context.py

@@ -20,7 +20,7 @@ from ..models import (
 
 
 
-    User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog,
+    User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog, ConversationSummary,
 
 
 
@@ -1230,3 +1230,163 @@ def _get_conversation_context(db: Session, conv_id: str) -> dict:
 
 
     return context
+
+
+
+# ===== Phase 2 上下文引擎升级：分层记忆 + 压缩 + 隔离 =====
+logger = logging.getLogger(__name__)
+
+
+def _estimate_tokens(text):
+    """粗略 token 估算：中英文混排约 2 字符/token。"""
+    if not text:
+        return 0
+    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 存在摘要，以 system 角色「对话摘要」前置（压缩层）。
+    - 按 token 估算从新到旧累加，超预算即截断（隔离：不溢出上下文窗口）。
+    - exclude_latest_user=True 时丢弃最新一条 user 消息（当前轮次另行发送）。
+    """
+    summary_block = None
+    if include_summary:
+        try:
+            summ = db.query(ConversationSummary).filter(
+                ConversationSummary.conversation_id == conv_id
+            ).first()
+            if 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
+
+
+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
+
+
+async def _maybe_compress_history(conv_id):
+    """后台异步压缩：将受保护窗口外的旧消息摘要落库 conversation_summaries。
+
+    失败静默回滚，绝不阻塞用户本次响应（fire-and-forget）。
+    """
+    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
+        chunk = compressible[:30]
+        text = "\n".join(f"{m.role}: {m.content}" for m in chunk)
+        existing = sdb.query(ConversationSummary).filter(
+            ConversationSummary.conversation_id == conv_id
+        ).first()
+        prior = existing.summary_text if existing else ""
+        prompt = (
+            "请把以下对话片段浓缩为简明摘要，保留用户核心诉求、已确认结论、关键事实与待办。"
+            "若提供『已有摘要』，请合并而非重复。\n\n"
+            f"已有摘要：\n{prior}\n\n待摘要片段：\n{text}"
+        )
+        summary = await _summarize_via_llm(prompt)
+        if not summary:
+            return
+        if existing:
+            existing.summary_text = summary
+            existing.token_count = _estimate_tokens(summary)
+            existing.updated_at = datetime.now()
+        else:
+            sdb.add(ConversationSummary(
+                conversation_id=conv_id,
+                summary_text=summary,
+                token_count=_estimate_tokens(summary),
+                updated_at=datetime.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

backend/python/tests/test_context_engine.py

@@ -0,0 +1,114 @@
+"""Phase 2 context-engine regression: _build_chat_history budget/layering/summary.
+
+Pure unit test with a fake DB session — no network, no real model. Guards the
+layered-memory + compression + isolation behaviour against regressions.
+"""
+from app.api.chat_context import _estimate_tokens, _build_chat_history
+
+
+class _Msg:
+    def __init__(self, role, content, cid="c1"):
+        self.role = role
+        self.content = content
+        self.conversation_id = cid
+
+
+class _Summary:
+    def __init__(self, text):
+        self.summary_text = text
+
+
+class _Query:
+    def __init__(self, rows):
+        self._rows = list(rows)
+
+    def filter(self, *a, **k):
+        return self
+
+    def order_by(self, *a, **k):
+        # test pre-sorts newest-first to match desc(created_at)
+        return self
+
+    def limit(self, n=None):
+        return self
+
+    def all(self):
+        return self._rows
+
+    def first(self):
+        return self._rows[0] if self._rows else None
+
+
+class _DB:
+    def __init__(self, messages, summary=None):
+        self._messages = messages
+        self._summary = summary
+
+    def query(self, model):
+        if getattr(model, "__name__", "") == "ConversationSummary":
+            return _Query([self._summary] if self._summary else [])
+        return _Query(self._messages)
+
+
+def _newest_first(msgs):
+    # _build_chat_history expects desc(created_at) -> newest first.
+    # Test authors write the list already newest-first, so return as-is.
+    return list(msgs)
+
+
+def test_estimate_tokens():
+    assert _estimate_tokens("") == 0
+    assert _estimate_tokens(None) == 0
+    assert _estimate_tokens("abcd") == 2  # 4//2
+    assert _estimate_tokens("a" * 5) == 2  # max(1, 5//2)=2
+    # just ensure positive monotonic
+    assert _estimate_tokens("x" * 100) == 50
+
+
+def test_history_excludes_latest_user():
+    # newest-first: [current user, prev assistant, older user]
+    msgs = _newest_first([
+        _Msg("user", "current question"),
+        _Msg("assistant", "previous answer"),
+        _Msg("user", "older question"),
+    ])
+    db = _DB(msgs)
+    out = _build_chat_history(db, "c1", max_tokens=10000, exclude_latest_user=True)
+    roles = [m["role"] for m in out]
+    # current user dropped; chronological order = [older user, previous assistant]
+    assert roles == ["user", "assistant"]
+    assert out[0]["content"] == "older question"
+
+
+def test_summary_injected_as_system_first():
+    msgs = _newest_first([
+        _Msg("user", "hi"),
+        _Msg("assistant", "hello"),
+    ])
+    db = _DB(msgs, summary=_Summary("用户想做图片生成"))
+    out = _build_chat_history(db, "c1", max_tokens=10000, exclude_latest_user=True)
+    assert out[0]["role"] == "system"
+    assert out[0]["content"].startswith("【对话历史摘要")
+    assert "用户想做图片生成" in out[0]["content"]
+    # after summary: chronological history (assistant then user 'hi' excluded? exclude_latest_user drops 'hi')
+    assert out[1]["role"] == "assistant"
+
+
+def test_budget_truncation():
+    # 5 assistant messages each ~200 tokens; budget small -> truncated
+    msgs = _newest_first([_Msg("assistant", "x" * 400) for _ in range(5)])
+    db = _DB(msgs)
+    out = _build_chat_history(db, "c1", max_tokens=500, exclude_latest_user=False)
+    # each msg ~200 tokens; budget 500 -> keep newest ~2
+    assert len(out) <= 3
+    total = sum(_estimate_tokens(m["content"]) for m in out)
+    assert total <= 500 + 200  # within budget (+1 msg slack)
+
+
+def test_no_summary_when_absent():
+    msgs = _newest_first([_Msg("user", "hi"), _Msg("assistant", "hello")])
+    db = _DB(msgs)
+    out = _build_chat_history(db, "c1", max_tokens=10000, exclude_latest_user=True)
+    # 'hi' excluded; only assistant remains, no system summary
+    assert all(m["role"] != "system" for m in out)
+    assert [m["role"] for m in out] == ["assistant"]

migrations/021_create_conversation_summaries.sql

@@ -0,0 +1,7 @@
+CREATE TABLE IF NOT EXISTS conversation_summaries (
+  conversation_id VARCHAR(64) NOT NULL,
+  summary_text TEXT NULL,
+  token_count INT NULL DEFAULT 0,
+  updated_at DATETIME(3) NULL,
+  PRIMARY KEY (conversation_id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;