diff --git a/ai_knowledge_base_v2/apps/admin-web/src/config/systemSettings.ts b/ai_knowledge_base_v2/apps/admin-web/src/config/systemSettings.ts index f7d807d..45d22e3 100644 --- a/ai_knowledge_base_v2/apps/admin-web/src/config/systemSettings.ts +++ b/ai_knowledge_base_v2/apps/admin-web/src/config/systemSettings.ts @@ -150,7 +150,7 @@ export const systemSettingSections: SystemSettingSection[] = [ defaultValue: 10, min: 0, max: 100, - description: "组装 Prompt 时带入的历史消息数量。", + description: "当前会话带入模型的最近历史消息条数；更早内容会滚动摘要，设为 0 将关闭会话记忆。修改后下一次提问立即生效。", }, { key: "show_reference_sources", diff --git a/ai_knowledge_base_v2/apps/backend/app/api/admin_settings.py b/ai_knowledge_base_v2/apps/backend/app/api/admin_settings.py index b0a49f8..85f6ead 100644 --- a/ai_knowledge_base_v2/apps/backend/app/api/admin_settings.py +++ b/ai_knowledge_base_v2/apps/backend/app/api/admin_settings.py @@ -23,7 +23,7 @@ from app.services.feishu_service import FeishuKnowledgeService from app.services.knowledge_agent_service import KnowledgeAgentService from app.services.knowledge_service import KnowledgeScope from app.services.model_service import ModelClientService -from app.services.rag_service import RagResult +from app.services.rag_service import PromptService, RagResult from app.services.secret_service import MASKED_SECRET, SENSITIVE_CONFIG_KEYS, SecretService router = APIRouter() @@ -151,14 +151,19 @@ async def debug_agent( version_overrides=payload.knowledgeVersions, preview_knowledge_ids=payload.knowledgeIds, ) + debug_messages = [ + {"role": "system", "content": payload.promptContent.strip()}, + *(rag_result.messages or []), + ] rag_result = RagResult( question=rag_result.question, knowledge_scopes=rag_result.knowledge_scopes, chunks=rag_result.chunks, - prompt=f"{payload.promptContent.strip()}\n\n{rag_result.prompt}", + prompt=PromptService.render_messages(debug_messages), allow_general_knowledge=rag_result.allow_general_knowledge, retrieval_log_id=rag_result.retrieval_log_id, tool_trace=rag_result.tool_trace, + messages=debug_messages, ) result = await ModelClientService.debug_model_async( model, diff --git a/ai_knowledge_base_v2/apps/backend/app/services/chat_context_service.py b/ai_knowledge_base_v2/apps/backend/app/services/chat_context_service.py new file mode 100644 index 0000000..6f5e20a --- /dev/null +++ b/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"", + "", + 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, + } diff --git a/ai_knowledge_base_v2/apps/backend/app/services/chat_service.py b/ai_knowledge_base_v2/apps/backend/app/services/chat_service.py index e890f60..41cea4b 100644 --- a/ai_knowledge_base_v2/apps/backend/app/services/chat_service.py +++ b/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"", "", 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: diff --git a/ai_knowledge_base_v2/apps/backend/app/services/chat_stream_service.py b/ai_knowledge_base_v2/apps/backend/app/services/chat_stream_service.py index 4993b03..2e4c43c 100644 --- a/ai_knowledge_base_v2/apps/backend/app/services/chat_stream_service.py +++ b/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 diff --git a/ai_knowledge_base_v2/apps/backend/app/services/knowledge_agent_service.py b/ai_knowledge_base_v2/apps/backend/app/services/knowledge_agent_service.py index a9d9b53..5009020 100644 --- a/ai_knowledge_base_v2/apps/backend/app/services/knowledge_agent_service.py +++ b/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, diff --git a/ai_knowledge_base_v2/apps/backend/app/services/model_service.py b/ai_knowledge_base_v2/apps/backend/app/services/model_service.py index 5e88090..69fdffb 100644 --- a/ai_knowledge_base_v2/apps/backend/app/services/model_service.py +++ b/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 diff --git a/ai_knowledge_base_v2/apps/backend/app/services/model_stream_service.py b/ai_knowledge_base_v2/apps/backend/app/services/model_stream_service.py index 500d93b..fec4f1f 100644 --- a/ai_knowledge_base_v2/apps/backend/app/services/model_stream_service.py +++ b/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)) diff --git a/ai_knowledge_base_v2/apps/backend/app/services/rag_async_service.py b/ai_knowledge_base_v2/apps/backend/app/services/rag_async_service.py index 328ec50..8200729 100644 --- a/ai_knowledge_base_v2/apps/backend/app/services/rag_async_service.py +++ b/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, ) diff --git a/ai_knowledge_base_v2/apps/backend/app/services/rag_service.py b/ai_knowledge_base_v2/apps/backend/app/services/rag_service.py index 22b2efb..c734a5c 100644 --- a/ai_knowledge_base_v2/apps/backend/app/services/rag_service.py +++ b/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"", "", content, flags=re.IGNORECASE).strip() - if content: - lines.append(f"{label}：{content}") - - return "[对话历史]\n" + "\n".join(lines) if lines else "" + return re.sub(r"", "", content, flags=re.IGNORECASE).strip() @classmethod def _load_active_prompt(cls, db: Session) -> str: diff --git a/ai_knowledge_base_v2/apps/backend/tests/test_chat_context.py b/ai_knowledge_base_v2/apps/backend/tests/test_chat_context.py new file mode 100644 index 0000000..0e54718 --- /dev/null +++ b/ai_knowledge_base_v2/apps/backend/tests/test_chat_context.py @@ -0,0 +1,164 @@ +from __future__ import annotations + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session +from sqlalchemy.pool import StaticPool + +from app.models import Base +from app.models.ai_config import ModelConfig, SystemConfig +from app.models.chat import ChatMessage, ChatSession +from app.services.chat_context_service import ChatContextService +from app.services.model_service import ( + ModelClientService, + _fit_source_messages, + _max_output_tokens, + _openai_messages, + _rough_token_count, + _system_and_turn_messages, +) +from app.services.rag_service import PromptService, RagResult + + +def _database(context_count: str = "10") -> Session: + engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool) + Base.metadata.create_all(engine) + db = Session(engine) + db.add(SystemConfig(id=1, config_key="chat_context_message_count", config_value=context_count)) + db.commit() + return db + + +def _messages(count: int) -> list[ChatMessage]: + result = [] + for index in range(1, count + 1): + result.append( + ChatMessage( + id=index, + session_id=1, + user_id=1, + role="user" if index % 2 else "assistant", + content=f"消息{index}", + ) + ) + return result + + +def test_context_message_count_is_read_at_runtime_and_zero_disables_all_memory(): + with _database("2") as db: + history = _messages(4) + messages = PromptService.build_messages(db, "现在的问题", [], history, "旧摘要") + assert [item["content"] for item in messages if item["role"] in {"user", "assistant"}] == [ + "消息3", + "消息4", + "现在的问题", + ] + assert any("旧摘要" in item["content"] for item in messages) + + config = db.query(SystemConfig).filter_by(config_key="chat_context_message_count").one() + config.config_value = "0" + db.commit() + messages = PromptService.build_messages(db, "现在的问题", [], history, "旧摘要") + assert [item["content"] for item in messages if item["role"] in {"user", "assistant"}] == ["现在的问题"] + assert all("旧摘要" not in item["content"] for item in messages) + + +def test_messages_leaving_window_are_summarized_immediately_and_summary_is_rewritten(monkeypatch): + with _database("4") as db: + session = ChatSession(id=1, user_id=1, title="测试", summary="已有摘要", message_count=6) + captured = {} + + def summarize(_db, messages_text, previous_summary=None): + captured["messages"] = messages_text + captured["previous"] = previous_summary + return "覆盖后的完整摘要" + + monkeypatch.setattr(ModelClientService, "summarize_or_raise", summarize) + result = ChatContextService.update_summary(db, session, _messages(6)) + + assert captured == {"messages": "用户：消息1\n大本营答疑助手：消息2", "previous": "已有摘要"} + assert session.summary == "覆盖后的完整摘要" + assert session.summary_up_to_message_id == 2 + assert result.trace["status"] == "success" + + +def test_messages_already_covered_by_summary_are_not_sent_twice_after_limit_increases(): + with _database("4") as db: + messages = PromptService.build_messages( + db, + "现在的问题", + [], + _messages(4), + "已经覆盖消息1到消息3的摘要", + summary_up_to_message_id=3, + ) + turns = [item["content"] for item in messages if item["role"] in {"user", "assistant"}] + assert turns == ["消息4", "现在的问题"] + + +def test_summary_failure_is_visible_and_does_not_break_chat(monkeypatch, caplog): + with _database("2") as db: + session = ChatSession(id=9, user_id=1, title="测试", summary=None, message_count=4) + + def fail(*_args, **_kwargs): + raise RuntimeError("摘要服务不可用") + + monkeypatch.setattr(ModelClientService, "summarize_or_raise", fail) + result = ChatContextService.update_summary(db, session, _messages(4)) + + assert session.summary is None + assert result.trace["status"] == "fallback" + assert result.trace["error"] == "摘要服务不可用" + assert "Chat history summary failed" in caplog.text + + +def test_provider_payload_uses_native_roles_and_normalizes_leading_assistant(): + rag = RagResult( + question="当前问题", + knowledge_scopes=[], + chunks=[], + prompt="兼容文本", + messages=[ + {"role": "system", "content": "系统规则"}, + {"role": "assistant", "content": "窗口从助手消息开始"}, + {"role": "user", "content": "历史问题"}, + {"role": "assistant", "content": "历史回答"}, + {"role": "system", "content": "知识上下文"}, + {"role": "user", "content": "当前问题"}, + ], + ) + + openai_messages = _openai_messages(rag) + system, turns = _system_and_turn_messages(rag) + assert [item["role"] for item in openai_messages] == ["system", "user", "assistant", "user"] + assert "窗口从助手消息开始" in system + assert "知识上下文" in system + assert turns[-1] == {"role": "user", "content": "当前问题"} + + +def test_provider_messages_are_trimmed_to_configured_model_context_window(): + rag = RagResult( + question="当前问题", + knowledge_scopes=[], + chunks=[], + prompt="兼容文本", + messages=[ + {"role": "system", "content": "规则" * 500}, + {"role": "user", "content": "很早的问题" * 200}, + {"role": "assistant", "content": "很早的回答" * 200}, + {"role": "system", "content": "知识" * 500}, + {"role": "user", "content": "当前问题"}, + ], + ) + model = ModelConfig( + provider="test", + display_name="test", + api_type="openai_compatible", + model_name="test", + context_window=1000, + max_token=600, + ) + + fitted = _fit_source_messages(rag, model) + input_tokens = sum(_rough_token_count(item["content"]) for item in fitted) + assert input_tokens + _max_output_tokens(model) + 64 <= model.context_window + assert fitted[-1] == {"role": "user", "content": "当前问题"} diff --git a/ai_knowledge_base_v2/apps/backend/tests/test_knowledge_agent.py b/ai_knowledge_base_v2/apps/backend/tests/test_knowledge_agent.py index 3368992..aa193c6 100644 --- a/ai_knowledge_base_v2/apps/backend/tests/test_knowledge_agent.py +++ b/ai_knowledge_base_v2/apps/backend/tests/test_knowledge_agent.py @@ -15,6 +15,7 @@ from app.models.knowledge import ( KnowledgeSourceSnapshot, KnowledgeVersion, ) +from app.models.chat import ChatMessage from app.services.knowledge_agent_service import KnowledgeAgentService @@ -118,7 +119,8 @@ def test_common_question_skips_knowledge_tools(): result = asyncio.run(KnowledgeAgentService.build_result(db, question="今天北京天气怎么样？")) assert result.allow_general_knowledge is True assert result.chunks == [] - assert result.tool_trace[1]["needKnowledge"] is False + decision = next(item for item in result.tool_trace if item["tool"] == "agent_decision") + assert decision["needKnowledge"] is False def test_course_question_without_catalog_does_not_fall_back_to_general_knowledge(): @@ -136,3 +138,19 @@ def test_course_question_searches_chunk_and_reads_parent_section(): assert len(result.chunks) == 1 assert "先稳定自己的焦虑" in result.chunks[0].content assert any(item["tool"] == "read_knowledge_sections" for item in result.tool_trace) + + +def test_contextual_follow_up_is_rewritten_before_agent_decision(): + with _database() as db: + history = [ + ChatMessage(id=1, session_id=1, user_id=1, role="user", content="课程退款条件有哪些？"), + ChatMessage(id=2, session_id=1, user_id=1, role="assistant", content="第一种和第二种情况不同。"), + ] + result = asyncio.run( + KnowledgeAgentService.build_result(db, question="那第二种情况呢？", history=history) + ) + rewrite = next(item for item in result.tool_trace if item["tool"] == "rewrite_contextual_question") + decision = next(item for item in result.tool_trace if item["tool"] == "agent_decision") + + assert rewrite["rewrittenQuestion"] == "关于“课程退款条件有哪些？”的追问：那第二种情况呢？" + assert decision["request"]["question"] == rewrite["rewrittenQuestion"]