fix: 修复会话上下文记忆链路

ai_knowledge_base_v2/apps/backend/app/services/chat_context_service.py

@@ -0,0 +1,239 @@
+from __future__ import annotations
+
+import logging
+import re
+from dataclasses import dataclass
+from time import perf_counter
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.models.ai_config import SystemConfig
+from app.models.chat import ChatMessage, ChatSession
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_CONTEXT_MESSAGE_COUNT = 10
+MAX_CONTEXT_MESSAGE_COUNT = 100
+
+
+@dataclass(frozen=True)
+class SummaryUpdateResult:
+    trace: dict | None = None
+
+
+@dataclass(frozen=True)
+class _SummaryWork:
+    messages: list[ChatMessage]
+    messages_text: str
+    request: dict
+
+
+class ChatContextService:
+    """Owns the runtime policy for a session's conversational memory."""
+
+    @staticmethod
+    def message_limit(db: Session) -> int:
+        config = db.scalar(
+            select(SystemConfig).where(SystemConfig.config_key == "chat_context_message_count")
+        )
+        if config is None or not config.config_value.strip():
+            return DEFAULT_CONTEXT_MESSAGE_COUNT
+        try:
+            value = int(float(config.config_value.strip()))
+        except ValueError:
+            logger.warning(
+                "Invalid chat_context_message_count=%r; falling back to %s",
+                config.config_value,
+                DEFAULT_CONTEXT_MESSAGE_COUNT,
+            )
+            return DEFAULT_CONTEXT_MESSAGE_COUNT
+        return max(0, min(MAX_CONTEXT_MESSAGE_COUNT, value))
+
+    @classmethod
+    def select_memory(
+        cls,
+        db: Session,
+        history: list[ChatMessage] | None,
+        summary: str | None,
+        summary_up_to_message_id: int | None = None,
+    ) -> tuple[list[ChatMessage], str | None]:
+        limit = cls.message_limit(db)
+        if limit == 0:
+            return [], None
+        recent = list((history or [])[-limit:])
+        if summary and summary_up_to_message_id is not None:
+            recent = [message for message in recent if message.id > summary_up_to_message_id]
+        return recent, summary
+
+    @classmethod
+    def update_summary(
+        cls,
+        db: Session,
+        session: ChatSession,
+        history: list[ChatMessage],
+    ) -> SummaryUpdateResult:
+        """Roll older messages into one bounded summary as soon as they leave the raw window."""
+        work = cls._prepare_summary(db, session, history)
+        if work is None:
+            return SummaryUpdateResult()
+        started = perf_counter()
+        try:
+            # Local import avoids a rag_service -> context_service -> model_service cycle.
+            from app.services.model_service import ModelClientService
+
+            new_summary = ModelClientService.summarize_or_raise(
+                db,
+                work.messages_text,
+                previous_summary=session.summary,
+            )
+            return cls._apply_summary(session, work, new_summary, started)
+        except Exception as exc:
+            return cls._summary_failure(session, work, exc, started)
+
+    @classmethod
+    async def update_summary_async(
+        cls,
+        db: Session,
+        session: ChatSession,
+        history: list[ChatMessage],
+    ) -> SummaryUpdateResult:
+        work = cls._prepare_summary(db, session, history)
+        if work is None:
+            return SummaryUpdateResult()
+        started = perf_counter()
+        try:
+            from app.services.model_service import ModelClientService
+
+            new_summary = await ModelClientService.summarize_or_raise_async(
+                db,
+                work.messages_text,
+                previous_summary=session.summary,
+            )
+            return cls._apply_summary(session, work, new_summary, started)
+        except Exception as exc:
+            return cls._summary_failure(session, work, exc, started)
+
+    @classmethod
+    def _prepare_summary(
+        cls,
+        db: Session,
+        session: ChatSession,
+        history: list[ChatMessage],
+    ) -> _SummaryWork | None:
+        limit = cls.message_limit(db)
+        if limit == 0 or len(history) <= limit:
+            return None
+        cutoff = len(history) - limit
+        start_idx = cls._covered_history_index(session, history)
+        if cutoff <= start_idx:
+            return None
+        messages = history[start_idx:cutoff]
+        messages_text = cls.messages_text(messages)
+        if not messages_text:
+            return None
+        return _SummaryWork(
+            messages=messages,
+            messages_text=messages_text,
+            request={
+                "previousSummary": session.summary or "",
+                "messages": messages_text,
+                "messageCount": len(messages),
+                "contextMessageLimit": limit,
+            },
+        )
+
+    @classmethod
+    def _apply_summary(
+        cls,
+        session: ChatSession,
+        work: _SummaryWork,
+        summary: str,
+        started: float,
+    ) -> SummaryUpdateResult:
+        summary = summary.strip()
+        if not summary:
+            raise ValueError("模型返回了空的历史摘要")
+        session.summary = summary
+        session.summary_up_to_message_id = work.messages[-1].id
+        duration_ms = int((perf_counter() - started) * 1000)
+        return SummaryUpdateResult(
+            trace=cls._trace(
+                work.request,
+                {"summary": summary, "count": len(work.messages)},
+                duration_ms=duration_ms,
+            )
+        )
+
+    @classmethod
+    def _summary_failure(
+        cls,
+        session: ChatSession,
+        work: _SummaryWork,
+        exc: Exception,
+        started: float,
+    ) -> SummaryUpdateResult:
+        duration_ms = int((perf_counter() - started) * 1000)
+        logger.exception(
+            "Chat history summary failed for session_id=%s message_count=%s",
+            session.id,
+            len(work.messages),
+            exc_info=exc,
+        )
+        return SummaryUpdateResult(
+            trace=cls._trace(
+                work.request,
+                {"count": 0, "fallback": "recent_messages_only"},
+                duration_ms=duration_ms,
+                status="fallback",
+                error=str(exc)[:500],
+            )
+        )
+
+    @staticmethod
+    def _covered_history_index(session: ChatSession, history: list[ChatMessage]) -> int:
+        if session.summary_up_to_message_id is None:
+            return 0
+        for index, message in enumerate(history):
+            if message.id > session.summary_up_to_message_id:
+                return index
+        return len(history)
+
+    @staticmethod
+    def messages_text(messages: list[ChatMessage]) -> str:
+        role_labels = {"user": "用户", "assistant": "大本营答疑助手"}
+        parts: list[str] = []
+        for message in messages:
+            content = re.sub(
+                r"<think[\s\S]*?</think\s*>",
+                "",
+                message.content,
+                flags=re.IGNORECASE,
+            ).strip()
+            if content:
+                parts.append(f"{role_labels.get(message.role, message.role)}：{content}")
+        return "\n".join(parts)
+
+    @staticmethod
+    def _trace(
+        request: dict,
+        response: dict,
+        *,
+        duration_ms: int,
+        status: str = "success",
+        error: str | None = None,
+    ) -> dict:
+        return {
+            "tool": "chat_memory_summary",
+            "order": 0,
+            "request": request,
+            "response": response,
+            **response,
+            "status": status,
+            "startedAtMs": 0,
+            "endedAtMs": duration_ms,
+            "durationMs": duration_ms,
+            "responseCount": response.get("count"),
+            "error": error,
+            "retries": 0,
+        }

ai_knowledge_base_v2/apps/backend/app/services/chat_service.py

@@ -12,8 +12,9 @@ from app.models.chat import ChatMessage, ChatSession
 from app.models.user import User
 from app.services.ai_request_log_service import AiRequestLogService
 from app.services.external_errors import ExternalServiceError
+from app.services.chat_context_service import ChatContextService
 from app.services.model_service import ModelClientService
-from app.services.rag_service import RagService, SUMMARY_TRIGGER_ROUNDS
+from app.services.rag_service import RagService
 
 
 class ChatService:
@@ -104,16 +105,16 @@ class ChatService:
                 )
             )
 
-            # 判断是否需要生成/更新摘要
-            rounds = len(history) // 2  # 1 轮 = user + assistant
-            if rounds >= SUMMARY_TRIGGER_ROUNDS:
-                ChatService._maybe_update_summary(db, session, history)
+            summary_result = ChatContextService.update_summary(db, session, history)
+            context_trace = [summary_result.trace] if summary_result.trace else None
 
             # 构建 prompt（传入历史 + 摘要）
             rag_result = RagService.build_result(
                 db, user, normalized_question,
                 history=history,
                 session_summary=session.summary,
+                summary_up_to_message_id=session.summary_up_to_message_id,
+                context_trace=context_trace,
             )
             completion = ModelClientService.complete(db, rag_result)
         except ExternalServiceError as exc:
@@ -190,53 +191,7 @@ class ChatService:
 
     @staticmethod
     def _maybe_update_summary(db: Session, session: ChatSession, history: list[ChatMessage]) -> None:
-        """当历史超过阈值时，生成/更新摘要。摘要失败不阻断主流程。"""
-        if not history:
-            return
-
-        # 计算需要摘要的部分：已有摘要覆盖到哪条消息之后的部分
-        start_idx = 0
-        if session.summary_up_to_message_id is not None:
-            for i, msg in enumerate(history):
-                if msg.id <= session.summary_up_to_message_id:
-                    start_idx = i + 1
-                else:
-                    break
-
-        # 需要摘要的消息 = 已有摘要之后、到最近 KEEP_RECENT*2 条之前的部分
-        from app.services.rag_service import KEEP_RECENT
-        end_idx = max(start_idx, len(history) - KEEP_RECENT * 2)
-
-        if end_idx <= start_idx:
-            return  # 没有新的需要摘要的内容
-
-        to_summarize = history[start_idx:end_idx]
-        if not to_summarize:
-            return
-
-        # 拼接成文本
-        import re
-        role_labels = {"user": "用户", "assistant": "大本营答疑助手"}
-        parts = []
-        for msg in to_summarize:
-            content = re.sub(r"<think[\s\S]*?</think\s*>", "", msg.content, flags=re.IGNORECASE).strip()
-            if content:
-                parts.append(f"{role_labels.get(msg.role, msg.role)}：{content}")
-        messages_text = "\n".join(parts)
-
-        if not messages_text:
-            return
-
-        # 调用模型生成摘要
-        new_summary = ModelClientService.summarize(db, messages_text)
-        if new_summary:
-            # 追加到已有摘要后面
-            combined = session.summary or ""
-            if combined:
-                combined += "\n"
-            combined += new_summary
-            session.summary = combined
-            session.summary_up_to_message_id = to_summarize[-1].id
+        ChatContextService.update_summary(db, session, history)
 
     @staticmethod
     def _ensure_quota(user: User) -> None:

ai_knowledge_base_v2/apps/backend/app/services/chat_stream_service.py

@@ -15,6 +15,7 @@ from app.models.knowledge import KnowledgeRetrievalLog
 from app.models.user import User
 from app.services.ai_request_log_service import AiRequestLogService
 from app.services.chat_service import ChatService, _title_from_question
+from app.services.chat_context_service import ChatContextService
 from app.services.external_errors import ExternalServiceError
 from app.services.human_attention_service import HumanAttentionService
 from app.services.model_stream_service import ModelStreamService
@@ -53,7 +54,8 @@ class ChatStreamService:
                 .order_by(ChatMessage.created_at.asc(), ChatMessage.id.asc())
             )
         )
-        _maybe_update_summary(db, session, history)
+        summary_result = ChatContextService.update_summary(db, session, history)
+        context_trace = [summary_result.trace] if summary_result.trace else None
 
         started_at = perf_counter()
         rag_result = None
@@ -61,7 +63,15 @@ class ChatStreamService:
         answer_parts: list[str] = []
 
         try:
-            rag_result = _build_rag_result(db, user, normalized_question, history, getattr(session, "summary", None))
+            rag_result = _build_rag_result(
+                db,
+                user,
+                normalized_question,
+                history,
+                getattr(session, "summary", None),
+                getattr(session, "summary_up_to_message_id", None),
+                context_trace,
+            )
             model_response = ModelStreamService.stream(db, rag_result)
             for chunk in model_response.chunks:
                 if chunk:
@@ -178,7 +188,8 @@ class ChatStreamService:
                 .order_by(ChatMessage.created_at.asc(), ChatMessage.id.asc())
             )
         )
-        _maybe_update_summary(db, session, history)
+        summary_result = await ChatContextService.update_summary_async(db, session, history)
+        context_trace = [summary_result.trace] if summary_result.trace else None
 
         started_at = perf_counter()
         rag_result = None
@@ -192,7 +203,9 @@ class ChatStreamService:
                 normalized_question,
                 history=history,
                 session_summary=getattr(session, "summary", None),
+                summary_up_to_message_id=getattr(session, "summary_up_to_message_id", None),
                 session_id=session.id,
+                context_trace=context_trace,
             )
             model_response = ModelStreamService.stream_async(db, rag_result)
             async for chunk in model_response.chunks:
@@ -355,24 +368,24 @@ def _mark_retrieval_failed(db: Session, rag_result, error_message: str, cost_ms:
         db.add(retrieval_log)
 
 
-def _build_rag_result(db: Session, user: User, question: str, history: list[ChatMessage], summary: str | None):
+def _build_rag_result(
+    db: Session,
+    user: User,
+    question: str,
+    history: list[ChatMessage],
+    summary: str | None,
+    summary_up_to_message_id: int | None,
+    context_trace: list[dict] | None = None,
+):
     parameters = signature(RagService.build_result).parameters
     if "history" in parameters:
-        return RagService.build_result(db, user, question, history=history, session_summary=summary)
+        return RagService.build_result(
+            db,
+            user,
+            question,
+            history=history,
+            session_summary=summary,
+            summary_up_to_message_id=summary_up_to_message_id,
+            context_trace=context_trace,
+        )
     return RagService.build_result(db, user, question)
-
-
-def _maybe_update_summary(db: Session, session, history: list[ChatMessage]) -> None:
-    summary_handler = getattr(ChatService, "_maybe_update_summary", None)
-    if not callable(summary_handler) or not hasattr(session, "summary_up_to_message_id"):
-        return
-    try:
-        from app.services import rag_service
-
-        trigger_rounds = getattr(rag_service, "SUMMARY_TRIGGER_ROUNDS", None)
-        if trigger_rounds is None:
-            return
-        if len(history) // 2 >= trigger_rounds:
-            summary_handler(db, session, history)
-    except Exception:
-        return

ai_knowledge_base_v2/apps/backend/app/services/knowledge_agent_service.py

@@ -13,6 +13,7 @@ from sqlalchemy.orm import Session
 
 from app.core.config import get_settings
 from app.models.ai_config import ModelConfig
+from app.services.chat_context_service import ChatContextService
 from app.models.knowledge import (
     Knowledge,
     KnowledgeChunk,
@@ -55,10 +56,12 @@ class KnowledgeAgentService:
         question: str,
         history=None,
         session_summary: str | None = None,
+        summary_up_to_message_id: int | None = None,
         session_id: int | None = None,
         user_id: int | None = None,
         version_overrides: dict[int, int] | None = None,
         preview_knowledge_ids: list[int] | None = None,
+        context_trace: list[dict] | None = None,
     ) -> RagResult:
         started = perf_counter()
         catalog = cls.get_knowledge_catalog(
@@ -77,16 +80,33 @@ class KnowledgeAgentService:
         db.add(log)
         db.flush()
         scopes = [KnowledgeScope(id=item["knowledgeId"], name=item["name"], feishu_space_id="", feishu_node_id="") for item in catalog]
-        trace: list[dict] = [cls._trace("get_knowledge_catalog", 1, {}, {"items": catalog, "count": len(catalog)}, started)]
-        need_knowledge, selected_ids, reason = cls._decide(question, catalog)
-        trace.append(cls._trace("agent_decision", 2, {"question": question}, {"needKnowledge": need_knowledge, "selectedKnowledgeIds": selected_ids, "reason": reason}, started))
+        trace: list[dict] = list(context_trace or [])
+        trace.append(cls._trace("get_knowledge_catalog", len(trace) + 1, {}, {"items": catalog, "count": len(catalog)}, started))
+        retrieval_question, rewrite_trace = await cls._rewrite_question(
+            db,
+            question,
+            history=history,
+            session_summary=session_summary,
+            summary_up_to_message_id=summary_up_to_message_id,
+            started=started,
+            order=len(trace) + 1,
+        )
+        trace.append(rewrite_trace)
+        need_knowledge, selected_ids, reason = cls._decide(retrieval_question, catalog)
+        trace.append(cls._trace(
+            "agent_decision",
+            len(trace) + 1,
+            {"question": retrieval_question, "originalQuestion": question},
+            {"needKnowledge": need_knowledge, "selectedKnowledgeIds": selected_ids, "reason": reason},
+            started,
+        ))
         chunks: list[RetrievedChunk] = []
         try:
             if need_knowledge and selected_ids:
-                terms = cls._query_terms(question)
+                terms = cls._query_terms(retrieval_question)
                 candidates = cls.search_knowledge(db, terms, selected_ids, catalog)
                 trace.append(cls._trace("search_knowledge", len(trace) + 1, {"queryTerms": terms, "knowledgeIds": selected_ids}, {"candidateCount": len(candidates), "candidates": [cls._candidate_trace(x) for x in candidates]}, started))
-                await cls._rerank(db, question, candidates, trace, started)
+                await cls._rerank(db, retrieval_question, candidates, trace, started)
                 selected = cls._select_sections(candidates)
                 chunks = [
                     RetrievedChunk(
@@ -111,14 +131,23 @@ class KnowledgeAgentService:
             log.tool_cost_ms = int((perf_counter() - started) * 1000)
             log.status = "prepared"
             db.add(log)
+            messages = PromptService.build_messages(
+                db,
+                question,
+                chunks,
+                history,
+                session_summary,
+                summary_up_to_message_id,
+            )
             return RagResult(
                 question=question,
                 knowledge_scopes=scopes,
                 chunks=chunks,
-                prompt=PromptService.build_prompt(db, question, chunks, history, session_summary),
+                prompt=PromptService.render_messages(messages),
                 allow_general_knowledge=not need_knowledge,
                 retrieval_log_id=log.id,
                 tool_trace=trace,
+                messages=messages,
             )
         except Exception as exc:
             log.status = "failed"
@@ -128,6 +157,101 @@ class KnowledgeAgentService:
             db.add(log)
             raise
 
+    @classmethod
+    async def _rewrite_question(
+        cls,
+        db: Session,
+        question: str,
+        *,
+        history,
+        session_summary: str | None,
+        summary_up_to_message_id: int | None,
+        started: float,
+        order: int,
+    ) -> tuple[str, dict]:
+        recent, summary = ChatContextService.select_memory(
+            db,
+            history,
+            session_summary,
+            summary_up_to_message_id,
+        )
+        if not recent or not cls._needs_contextual_rewrite(question):
+            return question, cls._trace(
+                "rewrite_contextual_question",
+                order,
+                {"question": question, "historyMessageCount": len(recent)},
+                {"rewrittenQuestion": question, "mode": "not_needed", "count": 1},
+                started,
+            )
+
+        history_text = ChatContextService.messages_text(recent)
+        request = {
+            "question": question,
+            "summary": summary or "",
+            "recentHistory": history_text,
+        }
+        fallback = cls._fallback_rewrite(question, recent)
+        model = db.scalar(
+            select(ModelConfig).where(ModelConfig.enabled == 1).order_by(ModelConfig.id.desc()).limit(1)
+        )
+        if model is None or _system_config_bool(db, "mock_model_enabled", get_settings().mock_model_enabled):
+            return fallback, cls._trace(
+                "rewrite_contextual_question",
+                order,
+                request,
+                {"rewrittenQuestion": fallback, "mode": "deterministic_fallback", "count": 1},
+                started,
+                status="fallback",
+            )
+
+        prompt = (
+            "你是对话问题改写器。结合历史，把当前追问改写成可独立理解、适合知识检索的完整问题。"
+            "不得回答问题，不得添加历史中没有的事实，只输出改写后的一个问题。\n\n"
+            f"历史摘要：\n{summary or '无'}\n\n最近对话：\n{history_text}\n\n当前追问：\n{question}"
+        )
+        rag = RagResult(question=question, knowledge_scopes=[], chunks=[], prompt=prompt, allow_general_knowledge=True)
+        try:
+            rewritten = (await asyncio.to_thread(
+                _call_configured_model,
+                model,
+                rag,
+                allow_no_hit=True,
+            )).strip().strip('"“”')
+            if not rewritten or len(rewritten) > 500:
+                raise ValueError("问题改写结果为空或过长")
+            return rewritten, cls._trace(
+                "rewrite_contextual_question",
+                order,
+                request,
+                {"rewrittenQuestion": rewritten, "mode": "model", "count": 1},
+                started,
+            )
+        except Exception as exc:
+            return fallback, cls._trace(
+                "rewrite_contextual_question",
+                order,
+                request,
+                {"rewrittenQuestion": fallback, "mode": "deterministic_fallback", "count": 1},
+                started,
+                status="fallback",
+                error=str(exc)[:300],
+            )
+
+    @staticmethod
+    def _needs_contextual_rewrite(question: str) -> bool:
+        text = question.strip()
+        if len(text) <= 16:
+            return True
+        return bool(re.search(r"(这个|那个|上述|上面|前面|其中|第二|继续|然后呢|怎么办|怎么做|什么意思|为什么)$", text))
+
+    @staticmethod
+    def _fallback_rewrite(question: str, history) -> str:
+        previous_user = next(
+            (message.content.strip() for message in reversed(history) if message.role == "user" and message.content.strip()),
+            "",
+        )
+        return f"关于“{previous_user}”的追问：{question.strip()}" if previous_user else question.strip()
+
     @staticmethod
     def get_knowledge_catalog(
         db: Session,

ai_knowledge_base_v2/apps/backend/app/services/model_service.py

@@ -48,16 +48,45 @@ class ModelClientService:
         )
 
     @staticmethod
-    def summarize(db: Session, messages_text: str) -> str | None:
+    def summarize(
+        db: Session,
+        messages_text: str,
+        previous_summary: str | None = None,
+    ) -> str | None:
         """调用模型生成对话摘要，失败时返回 None（不阻断主流程）"""
-        model = ModelClientService._get_enabled_model(db)
-        if model is None:
-            return None
         try:
-            return _generate_summary(model, messages_text)
+            return ModelClientService.summarize_or_raise(db, messages_text, previous_summary)
         except Exception:
             return None
 
+    @staticmethod
+    def summarize_or_raise(
+        db: Session,
+        messages_text: str,
+        previous_summary: str | None = None,
+    ) -> str:
+        model = ModelClientService._get_enabled_model(db)
+        if _system_config_bool(db, "mock_model_enabled", get_settings().mock_model_enabled):
+            combined = "\n".join(part for part in [previous_summary, messages_text] if part).strip()
+            return combined[-1000:]
+        if model is None:
+            raise ExternalServiceError("未启用可用于生成历史摘要的模型", provider="model")
+        return _generate_summary(model, messages_text, previous_summary)
+
+    @staticmethod
+    async def summarize_or_raise_async(
+        db: Session,
+        messages_text: str,
+        previous_summary: str | None = None,
+    ) -> str:
+        if _system_config_bool(db, "mock_model_enabled", get_settings().mock_model_enabled):
+            combined = "\n".join(part for part in [previous_summary, messages_text] if part).strip()
+            return combined[-1000:]
+        model = ModelClientService._get_enabled_model(db)
+        if model is None:
+            raise ExternalServiceError("未启用可用于生成历史摘要的模型", provider="model")
+        return await asyncio.to_thread(_generate_summary, model, messages_text, previous_summary)
+
     @staticmethod
     def _get_enabled_model(db: Session) -> ModelConfig | None:
         return db.scalar(
@@ -171,12 +200,19 @@ def _call_configured_model(model: ModelConfig, rag_result: RagResult, *, allow_n
     return _call_openai_compatible_model(model, rag_result)
 
 
-def _generate_summary(model: ModelConfig, messages_text: str) -> str:
+def _generate_summary(
+    model: ModelConfig,
+    messages_text: str,
+    previous_summary: str | None = None,
+) -> str:
     """用模型把历史对话压缩成摘要"""
     summary_prompt = (
-        "请将以下对话历史压缩成一段简洁的摘要，保留关键信息（用户问了什么、AI回答了什么要点、"
-        "是否有未解决的问题）。摘要不超过 300 字，用中文输出。\n\n"
-        f"对话历史：\n{messages_text}"
+        "请把已有摘要和新增历史滚动重写成一份完整、无重复的会话记忆。"
+        "必须覆盖：用户背景与明确事实、用户当前目标、已确认结论、用户偏好和限制、"
+        "尚未解决的问题、最近对话进展。后出现的修正覆盖旧信息。"
+        "只输出中文摘要，不要解释，控制在1000字以内。\n\n"
+        f"已有摘要：\n{previous_summary or '无'}\n\n"
+        f"新增历史：\n{messages_text}"
     )
     rag_result = RagResult(
         question=summary_prompt,
@@ -190,12 +226,9 @@ def _generate_summary(model: ModelConfig, messages_text: str) -> str:
 def _call_openai_compatible_model(model: ModelConfig, rag_result: RagResult) -> str:
     payload = {
         "model": model.model_name,
-        "messages": [
-            {"role": "system", "content": _agent_system_prompt()},
-            {"role": "user", "content": rag_result.prompt},
-        ],
+        "messages": _openai_messages(rag_result, model),
         "stream": False,
-        "max_tokens": model.max_token or 1024,
+        "max_tokens": _max_output_tokens(model),
     }
     _put_if_not_none(payload, "temperature", _decimal_to_float(model.temperature))
     _put_if_not_none(payload, "top_p", _decimal_to_float(model.top_p))
@@ -221,11 +254,12 @@ def _call_openai_compatible_model(model: ModelConfig, rag_result: RagResult) ->
 
 
 def _call_anthropic_messages(model: ModelConfig, rag_result: RagResult) -> str:
+    system, messages = _system_and_turn_messages(rag_result, model)
     payload = {
         "model": model.model_name,
-        "max_tokens": model.max_token or 1024,
-        "system": _agent_system_prompt(),
-        "messages": [{"role": "user", "content": rag_result.prompt}],
+        "max_tokens": _max_output_tokens(model),
+        "system": system,
+        "messages": messages,
         "stream": False,
     }
     _put_if_not_none(payload, "temperature", _decimal_to_float(model.temperature))
@@ -252,15 +286,20 @@ def _call_gemini_generate_content(model: ModelConfig, rag_result: RagResult) ->
     _put_if_not_none(generation_config, "temperature", _decimal_to_float(model.temperature))
     _put_if_not_none(generation_config, "topP", _decimal_to_float(model.top_p))
     _put_if_not_none(generation_config, "topK", model.top_k)
-    _put_if_not_none(generation_config, "maxOutputTokens", model.max_token)
+    generation_config["maxOutputTokens"] = _max_output_tokens(model)
     if model.response_format == "json_object":
         generation_config["responseMimeType"] = "application/json"
 
+    system, messages = _system_and_turn_messages(rag_result, model)
     payload = {
-        "systemInstruction": {
-            "parts": [{"text": _agent_system_prompt()}]
-        },
-        "contents": [{"role": "user", "parts": [{"text": rag_result.prompt}]}],
+        "systemInstruction": {"parts": [{"text": system}]},
+        "contents": [
+            {
+                "role": "model" if message["role"] == "assistant" else "user",
+                "parts": [{"text": message["content"]}],
+            }
+            for message in messages
+        ],
     }
     if generation_config:
         payload["generationConfig"] = generation_config
@@ -333,17 +372,23 @@ def _call_minimax(model: ModelConfig, rag_result: RagResult) -> str:
     elif "GroupId" not in base_url:
         base_url = f"{base_url}&GroupId={urllib.parse.quote(group_id)}"
 
+    system, messages = _system_and_turn_messages(rag_result, model)
     payload: dict[str, Any] = {
         "model": model.model_name,
-        "tokens_to_generate": model.max_token or 1024,
+        "tokens_to_generate": _max_output_tokens(model),
         "reply_constraints": {"sender_type": "BOT", "sender_name": "大本营答疑助手"},
         "messages": [
-            {"sender_type": "USER", "sender_name": "用户", "text": rag_result.prompt},
+            {
+                "sender_type": "BOT" if message["role"] == "assistant" else "USER",
+                "sender_name": "大本营答疑助手" if message["role"] == "assistant" else "用户",
+                "text": message["content"],
+            }
+            for message in messages
         ],
         "bot_setting": [
             {
                 "bot_name": "大本营答疑助手",
-                "content": _agent_system_prompt(),
+                "content": system,
             }
         ],
     }
@@ -448,6 +493,88 @@ def _agent_system_prompt() -> str:
     )
 
 
+def _max_output_tokens(model: ModelConfig) -> int:
+    configured = max(1, model.max_token or 1024)
+    if not model.context_window:
+        return configured
+    return min(configured, max(1, model.context_window // 3))
+
+
+def _openai_messages(
+    rag_result: RagResult,
+    model: ModelConfig | None = None,
+) -> list[dict[str, str]]:
+    system, messages = _system_and_turn_messages(rag_result, model)
+    return [{"role": "system", "content": system}, *messages]
+
+
+def _system_and_turn_messages(
+    rag_result: RagResult,
+    model: ModelConfig | None = None,
+) -> tuple[str, list[dict[str, str]]]:
+    source = _fit_source_messages(rag_result, model)
+    system_parts = [message["content"] for message in source if message.get("role") == "system"]
+    system = "\n\n".join(system_parts).strip() or _agent_system_prompt()
+    turns: list[dict[str, str]] = []
+    for message in source:
+        role = message.get("role")
+        content = str(message.get("content", "")).strip()
+        if role not in {"user", "assistant"} or not content:
+            continue
+        if turns and turns[-1]["role"] == role:
+            turns[-1]["content"] += "\n\n" + content
+        else:
+            turns.append({"role": role, "content": content})
+    if turns and turns[0]["role"] == "assistant":
+        system += "\n\n[最近一条助手历史回复]\n" + turns.pop(0)["content"]
+    if not turns:
+        turns.append({"role": "user", "content": rag_result.prompt})
+    return system, turns
+
+
+def _fit_source_messages(rag_result: RagResult, model: ModelConfig | None) -> list[dict[str, str]]:
+    fallback_messages = [
+        {"role": "system", "content": _agent_system_prompt()},
+        {"role": "user", "content": rag_result.prompt},
+    ]
+    source = [dict(message) for message in (rag_result.messages or fallback_messages)]
+    if model is None or not model.context_window:
+        return source
+
+    output_budget = _max_output_tokens(model)
+    input_budget = max(1, model.context_window - output_budget - 64)
+
+    def token_count() -> int:
+        return sum(_rough_token_count(str(message.get("content", ""))) for message in source)
+
+    while token_count() > input_budget:
+        removable = next(
+            (
+                index
+                for index, message in enumerate(source[:-1])
+                if message.get("role") in {"user", "assistant"}
+            ),
+            None,
+        )
+        if removable is None:
+            break
+        source.pop(removable)
+
+    while token_count() > input_budget:
+        system_indexes = [index for index, message in enumerate(source) if message.get("role") == "system"]
+        if not system_indexes:
+            break
+        longest = max(system_indexes, key=lambda index: len(str(source[index].get("content", ""))))
+        content = str(source[longest].get("content", ""))
+        excess_chars = max(100, (token_count() - input_budget) * 2)
+        target_length = max(200, len(content) - excess_chars)
+        if target_length >= len(content):
+            break
+        half = target_length // 2
+        source[longest]["content"] = content[:half] + "\n...[上下文窗口自动截断]...\n" + content[-half:]
+    return source
+
+
 def _put_if_not_none(target: dict[str, Any], key: str, value: Any) -> None:
     if value is not None:
         target[key] = value

ai_knowledge_base_v2/apps/backend/app/services/model_stream_service.py

@@ -15,16 +15,18 @@ from app.models.ai_config import ModelConfig
 from app.services.external_errors import ExternalServiceError
 from app.services.model_service import (
     _anthropic_headers,
-    _agent_system_prompt,
     _auth_headers,
     _call_configured_model,
     _decimal_to_float,
     _load_extra_params,
     _mock_answer,
+    _max_output_tokens,
+    _openai_messages,
     _put_if_not_none,
     _resolve_anthropic_endpoint,
     _resolve_openai_endpoint,
     _rough_token_count,
+    _system_and_turn_messages,
     _system_config_bool,
 )
 from app.services.rag_service import NO_HIT_ANSWER, RagResult
@@ -166,11 +168,8 @@ async def _stream_configured_model_async(model: ModelConfig, rag_result: RagResu
 def _stream_openai_compatible_model(model: ModelConfig, rag_result: RagResult) -> Iterator[str]:
     payload: dict[str, Any] = {
         "model": model.model_name,
-        "messages": [
-            {"role": "system", "content": _agent_system_prompt()},
-            {"role": "user", "content": rag_result.prompt},
-        ],
-        "max_tokens": model.max_token or 1024,
+        "messages": _openai_messages(rag_result, model),
+        "max_tokens": _max_output_tokens(model),
     }
     _put_if_not_none(payload, "temperature", _decimal_to_float(model.temperature))
     _put_if_not_none(payload, "top_p", _decimal_to_float(model.top_p))
@@ -215,11 +214,8 @@ async def _stream_openai_compatible_model_async(model: ModelConfig, rag_result:
 def _openai_stream_payload(model: ModelConfig, rag_result: RagResult) -> dict[str, Any]:
     payload: dict[str, Any] = {
         "model": model.model_name,
-        "messages": [
-            {"role": "system", "content": _agent_system_prompt()},
-            {"role": "user", "content": rag_result.prompt},
-        ],
-        "max_tokens": model.max_token or 1024,
+        "messages": _openai_messages(rag_result, model),
+        "max_tokens": _max_output_tokens(model),
     }
     _put_if_not_none(payload, "temperature", _decimal_to_float(model.temperature))
     _put_if_not_none(payload, "top_p", _decimal_to_float(model.top_p))
@@ -319,11 +315,12 @@ async def _iter_openai_stream_async(response: httpx.Response) -> AsyncIterator[s
 
 
 def _stream_anthropic_messages(model: ModelConfig, rag_result: RagResult) -> Iterator[str]:
+    system, messages = _system_and_turn_messages(rag_result, model)
     payload: dict[str, Any] = {
         "model": model.model_name,
-        "max_tokens": model.max_token or 1024,
-        "system": _agent_system_prompt(),
-        "messages": [{"role": "user", "content": rag_result.prompt}],
+        "max_tokens": _max_output_tokens(model),
+        "system": system,
+        "messages": messages,
     }
     _put_if_not_none(payload, "temperature", _decimal_to_float(model.temperature))
     _put_if_not_none(payload, "top_p", _decimal_to_float(model.top_p))
@@ -363,11 +360,12 @@ async def _stream_anthropic_messages_async(model: ModelConfig, rag_result: RagRe
 
 
 def _anthropic_stream_payload(model: ModelConfig, rag_result: RagResult) -> dict[str, Any]:
+    system, messages = _system_and_turn_messages(rag_result, model)
     payload: dict[str, Any] = {
         "model": model.model_name,
-        "max_tokens": model.max_token or 1024,
-        "system": _agent_system_prompt(),
-        "messages": [{"role": "user", "content": rag_result.prompt}],
+        "max_tokens": _max_output_tokens(model),
+        "system": system,
+        "messages": messages,
     }
     _put_if_not_none(payload, "temperature", _decimal_to_float(model.temperature))
     _put_if_not_none(payload, "top_p", _decimal_to_float(model.top_p))

ai_knowledge_base_v2/apps/backend/app/services/rag_async_service.py

@@ -16,13 +16,17 @@ class AsyncRagService:
         question: str,
         history: list[ChatMessage] | None = None,
         session_summary: str | None = None,
+        summary_up_to_message_id: int | None = None,
         session_id: int | None = None,
+        context_trace: list[dict] | None = None,
     ) -> RagResult:
         return await KnowledgeAgentService.build_result(
             db,
             question=question,
             history=history,
             session_summary=session_summary,
+            summary_up_to_message_id=summary_up_to_message_id,
             session_id=session_id,
             user_id=user.id,
+            context_trace=context_trace,
         )

ai_knowledge_base_v2/apps/backend/app/services/rag_service.py

@@ -8,16 +8,12 @@ from sqlalchemy.orm import Session
 from app.models.ai_config import Prompt
 from app.models.chat import ChatMessage
 from app.models.user import User
+from app.services.chat_context_service import ChatContextService
 from app.services.feishu_service import FeishuKnowledgeService
 from app.services.knowledge_service import KnowledgeAccessService, KnowledgeScope
 
 NO_HIT_ANSWER = "当前知识库中未检索到相关内容，请联系管理员补充相关知识。"
 
-# 摘要配置：超过 KEEP_RECENT 轮的历史会被压缩成摘要
-KEEP_RECENT = 5
-# 触发摘要的轮数阈值（当历史总轮数 > 这个值时才触发）
-SUMMARY_TRIGGER_ROUNDS = 8
-
 
 @dataclass(frozen=True)
 class RetrievedChunk:
@@ -41,6 +37,7 @@ class RagResult:
     allow_general_knowledge: bool = False
     retrieval_log_id: int | None = None
     tool_trace: list[dict] | None = None
+    messages: list[dict[str, str]] | None = None
 
     @property
     def is_hit(self) -> bool:
@@ -60,11 +57,28 @@ class RagService:
         question: str,
         history: list[ChatMessage] | None = None,
         session_summary: str | None = None,
+        summary_up_to_message_id: int | None = None,
+        context_trace: list[dict] | None = None,
     ) -> RagResult:
         scopes = KnowledgeAccessService.get_allowed_knowledge(db, user)
         chunks = FeishuKnowledgeService.retrieve(question, scopes, db)
-        prompt = PromptService.build_prompt(db, question, chunks, history, session_summary)
-        return RagResult(question=question, knowledge_scopes=scopes, chunks=chunks, prompt=prompt)
+        messages = PromptService.build_messages(
+            db,
+            question,
+            chunks,
+            history,
+            session_summary,
+            summary_up_to_message_id,
+        )
+        prompt = PromptService.render_messages(messages)
+        return RagResult(
+            question=question,
+            knowledge_scopes=scopes,
+            chunks=chunks,
+            prompt=prompt,
+            tool_trace=context_trace,
+            messages=messages,
+        )
 
 
 class PromptService:
@@ -80,7 +94,29 @@ class PromptService:
         chunks: list[RetrievedChunk],
         history: list[ChatMessage] | None = None,
         session_summary: str | None = None,
+        summary_up_to_message_id: int | None = None,
     ) -> str:
+        return cls.render_messages(
+            cls.build_messages(
+                db,
+                question,
+                chunks,
+                history,
+                session_summary,
+                summary_up_to_message_id,
+            )
+        )
+
+    @classmethod
+    def build_messages(
+        cls,
+        db: Session,
+        question: str,
+        chunks: list[RetrievedChunk],
+        history: list[ChatMessage] | None = None,
+        session_summary: str | None = None,
+        summary_up_to_message_id: int | None = None,
+    ) -> list[dict[str, str]]:
         prompt = cls._load_active_prompt(db)
 
         # 知识库上下文
@@ -91,52 +127,48 @@ class PromptService:
         if not context:
             context = "本轮没有可靠的正式知识章节。对于一般常识可以谨慎回答；涉及课程、老师观点或公司业务时必须说明依据不足，不得编造。"
 
-        parts = [prompt]
-        parts.append(
-            "[不可关闭的最低安全规则 v1]\n"
-            "现实危险、自伤伤人风险应优先建议立即寻求线下专业帮助；医疗、法律、财务问题不得给出替代专业意见的结论；"
-            "不得伪造老师观点或课程内容；不得输出整篇课程文章、大段连续原文，也不得通过多轮拼接还原完整资料。"
+        messages: list[dict[str, str]] = [
+            {
+                "role": "system",
+                "content": prompt
+                + "\n\n"
+                + "[不可关闭的最低安全规则 v1]\n"
+                + "现实危险、自伤伤人风险应优先建议立即寻求线下专业帮助；医疗、法律、财务问题不得给出替代专业意见的结论；"
+                + "不得伪造老师观点或课程内容；不得输出整篇课程文章、大段连续原文，也不得通过多轮拼接还原完整资料。",
+            }
+        ]
+
+        recent_history, visible_summary = ChatContextService.select_memory(
+            db,
+            history,
+            session_summary,
+            summary_up_to_message_id,
+        )
+        if visible_summary:
+            messages.append({"role": "system", "content": f"[历史对话摘要]\n{visible_summary}"})
+
+        for message in recent_history:
+            content = cls._clean_history_content(message.content)
+            if content and message.role in {"user", "assistant"}:
+                messages.append({"role": message.role, "content": content})
+
+        messages.append({"role": "system", "content": f"[本轮可靠知识上下文]\n{context}"})
+        messages.append({"role": "user", "content": question.strip()})
+        return messages
+
+    @staticmethod
+    def render_messages(messages: list[dict[str, str]]) -> str:
+        role_labels = {"system": "系统", "user": "用户", "assistant": "大本营答疑助手"}
+        return "\n\n".join(
+            f"[{role_labels.get(message['role'], message['role'])}]\n{message['content']}"
+            for message in messages
         )
 
-        # 对话历史：摘要 + 最近 N 轮原文
-        history_part = cls._build_history_section(history, session_summary)
-        if history_part:
-            parts.append(history_part)
+    @staticmethod
+    def _clean_history_content(content: str) -> str:
+        import re
 
-        parts.append(context)
-        parts.append(f"用户问题：{question.strip()}")
-
-        return "\n\n".join(parts)
-
-    @classmethod
-    def _build_history_section(
-        cls,
-        history: list[ChatMessage] | None,
-        session_summary: str | None,
-    ) -> str:
-        if not history:
-            return ""
-
-        lines: list[str] = []
-
-        # 如果有历史摘要，先放摘要
-        if session_summary:
-            lines.append(f"[历史对话摘要]\n{session_summary}")
-
-        # 只保留最近 KEEP_RECENT 轮的原文（1 轮 = 1 个 user + 1 个 assistant）
-        recent = history[-(KEEP_RECENT * 2):]
-        if recent:
-            role_labels = {"user": "用户", "assistant": "大本营答疑助手"}
-            for msg in recent:
-                label = role_labels.get(msg.role, msg.role)
-                content = msg.content.strip()
-                # 去掉 think 标签，不需要在历史中重复传递
-                import re
-                content = re.sub(r"<think[\s\S]*?</think\s*>", "", content, flags=re.IGNORECASE).strip()
-                if content:
-                    lines.append(f"{label}：{content}")
-
-        return "[对话历史]\n" + "\n".join(lines) if lines else ""
+        return re.sub(r"<think[\s\S]*?</think\s*>", "", content, flags=re.IGNORECASE).strip()
 
     @classmethod
     def _load_active_prompt(cls, db: Session) -> str: