From 61db0aa9b8728eea4fd6d5fd6659be532796407b Mon Sep 17 00:00:00 2001 From: Nelson <1475262689@qq.com> Date: Wed, 8 Jul 2026 17:49:52 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AF=B9=E8=AF=9D=E6=91=98=E8=A6=81+?= =?UTF-8?q?=E6=AE=B5=E8=90=BD=E6=8C=89=E9=9C=80=E8=AF=BB=E5=8F=96+think?= =?UTF-8?q?=E6=A0=87=E7=AD=BE=E8=A7=A3=E6=9E=90=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - chat.py: ChatSession 新增 summary/summary_up_to_message_id 字段 - rag_service.py: Prompt 支持历史摘要+最近5轮原文拼接 - model_service.py: 新增 summarize() 方法，复用已启用模型生成摘要 - chat_service.py: 集成对话历史传递，8轮后触发摘要生成 - feishu_service.py: 段落级按需读取，按标题匹配+关键词打分+top5文档 摘要策略： - 保留最近5轮原文保证追问连续性 - 更早历史由模型压缩成摘要（<=300字） - 摘要失败不阻断主流程 飞书读取优化： - 每个知识库最多读5个相关文档 - 每文档提取top10相关段落 - 按原文顺序拼接，不截断 Fix: _extract_openai_answer 重命名为 _extract_gemini_answer 避免同名函数覆盖导致MiniMax模型被错误解析为Gemini响应 --- .../apps/backend/app/models/chat.py | 2 + .../apps/backend/app/services/chat_service.py | 77 ++++++++- .../backend/app/services/feishu_service.py | 148 +++++++++++++++--- .../backend/app/services/model_service.py | 44 +++++- .../apps/backend/app/services/rag_service.py | 70 ++++++++- 5 files changed, 309 insertions(+), 32 deletions(-) diff --git a/ai_knowledge_base_v2/apps/backend/app/models/chat.py b/ai_knowledge_base_v2/apps/backend/app/models/chat.py index 997782e..16b8ea3 100644 --- a/ai_knowledge_base_v2/apps/backend/app/models/chat.py +++ b/ai_knowledge_base_v2/apps/backend/app/models/chat.py @@ -14,6 +14,8 @@ class ChatSession(Base, TimestampMixin): id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) user_id: Mapped[int] = mapped_column(ForeignKey("sys_user.id"), index=True, nullable=False) title: Mapped[str] = mapped_column(String(100), nullable=False) + summary: Mapped[str | None] = mapped_column(Text, nullable=True) + summary_up_to_message_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True) message_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False) last_message_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) is_deleted: Mapped[int] = mapped_column(default=0, nullable=False) 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 d7311f4..b4b9c15 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,7 +12,7 @@ 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.model_service import ModelClientService -from app.services.rag_service import RagService +from app.services.rag_service import RagService, SUMMARY_TRIGGER_ROUNDS class ChatService: @@ -88,7 +88,30 @@ class ChatService: started_at = perf_counter() try: - rag_result = RagService.build_result(db, user, normalized_question) + # 获取历史消息（不含刚插入的 user_message，它还没 flush id） + history = list( + db.scalars( + select(ChatMessage) + .where( + ChatMessage.session_id == session.id, + ChatMessage.user_id == user.id, + ChatMessage.id < user_message.id, + ) + .order_by(ChatMessage.created_at.asc(), ChatMessage.id.asc()) + ) + ) + + # 判断是否需要生成/更新摘要 + rounds = len(history) // 2 # 1 轮 = user + assistant + if rounds >= SUMMARY_TRIGGER_ROUNDS: + ChatService._maybe_update_summary(db, session, history) + + # 构建 prompt（传入历史 + 摘要） + rag_result = RagService.build_result( + db, user, normalized_question, + history=history, + session_summary=session.summary, + ) completion = ModelClientService.complete(db, rag_result) except ExternalServiceError as exc: cost_ms = int((perf_counter() - started_at) * 1000) @@ -160,6 +183,56 @@ class ChatService: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="会话不存在") return session + @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": "AI助手"} + 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 + @staticmethod def _ensure_quota(user: User) -> None: if user.daily_chat_used >= user.daily_chat_limit: diff --git a/ai_knowledge_base_v2/apps/backend/app/services/feishu_service.py b/ai_knowledge_base_v2/apps/backend/app/services/feishu_service.py index 9ae0bc6..b3f1c45 100644 --- a/ai_knowledge_base_v2/apps/backend/app/services/feishu_service.py +++ b/ai_knowledge_base_v2/apps/backend/app/services/feishu_service.py @@ -392,7 +392,7 @@ class FeishuKnowledgeService: for scope in scopes: try: - documents = cls._load_documents(scope, config) + documents = cls._load_documents(scope, question, config) except ExternalServiceError as exc: load_errors.append(f"{scope.name}：{exc}") continue @@ -432,9 +432,10 @@ class FeishuKnowledgeService: def _load_documents( cls, scope: KnowledgeScope, + question: str, config: FeishuRetrievalConfig, ) -> list[tuple[str, str, str | None]]: - """加载指定知识库 scope 下的所有文档""" + """加载指定知识库 scope 下的文档，按需读取（先按标题过滤，再读内容）""" node_id = scope.feishu_node_id base_url = f"https://zcn7hk6n047c.feishu.cn" @@ -461,52 +462,161 @@ class FeishuKnowledgeService: title = node_info.get("title", node_id) space_id = node_info.get("space_id", scope.feishu_space_id) - # 如果节点本身就是一个文档 + # 如果节点本身就是一个文档，直接读取后做段落筛选 if obj_type == "docx": content = cls._get_document_content_with_cache(obj_token, config) source_url = f"{base_url}/wiki/{node_id}" - return [(title, content, source_url)] + filtered = cls._extract_relevant_paragraphs(content, question) + return [(title, filtered, source_url)] if obj_type == "file": content = cls._get_file_content_with_cache(obj_token, config) source_url = f"{base_url}/wiki/{node_id}" - return [(title, content, source_url)] + filtered = cls._extract_relevant_paragraphs(content, question) + return [(title, filtered, source_url)] - # 如果节点是文件夹，获取子节点 + # 如果节点是文件夹，获取子节点列表（只读标题，不读内容） children = cls._get_wiki_children_with_cache(space_id, node_id, config) + # 先按标题和问题做关键词匹配，筛选可能相关的文档 + scored_children = cls._score_children_by_title(question, children) + + # 只读取 top 5 最可能相关的文档（避免全量读取） + MAX_DOCS_PER_SCOPE = 5 + documents = [] - for child in children: + for child, _score in scored_children[:MAX_DOCS_PER_SCOPE]: child_node_token = child.get("node_token", "") child_obj_type = child.get("obj_type", "") child_obj_token = child.get("obj_token", "") child_title = child.get("title", "") - if child_obj_type == "docx" and child_obj_token: - try: + if not child_obj_token: + continue + + try: + if child_obj_type == "docx": content = cls._get_document_content_with_cache(child_obj_token, config) - source_url = f"{base_url}/wiki/{child_node_token}" - documents.append((child_title, content, source_url)) - except ExternalServiceError: - continue - elif child_obj_type == "file" and child_obj_token: - try: + elif child_obj_type == "file": content = cls._get_file_content_with_cache(child_obj_token, config) - source_url = f"{base_url}/wiki/{child_node_token}" - documents.append((child_title, content, source_url)) - except ExternalServiceError: + else: continue + source_url = f"{base_url}/wiki/{child_node_token}" + filtered = cls._extract_relevant_paragraphs(content, question) + documents.append((child_title, filtered, source_url)) + except ExternalServiceError: + continue # 也读取节点本身的文档（如果有） if obj_type == "docx" and obj_token: try: content = cls._get_document_content_with_cache(obj_token, config) source_url = f"{base_url}/wiki/{node_id}" - documents.insert(0, (title, content, source_url)) + filtered = cls._extract_relevant_paragraphs(content, question) + documents.insert(0, (title, filtered, source_url)) except ExternalServiceError: pass return documents + @classmethod + def _score_children_by_title(cls, question: str, children: list[dict]) -> list[tuple[dict, float]]: + """按文档标题与问题的相关性打分排序""" + question_lower = question.lower() + question_chars = set(question_lower) + question_bigrams = set(question_lower[i : i + 2] for i in range(len(question_lower) - 1)) + + scored = [] + for child in children: + title = (child.get("title", "") or "").lower() + if not title: + continue + + score = 0.0 + # 标题包含问题中的字/词 + for char in question_chars: + if char in title: + score += 2.0 + for bigram in question_bigrams: + if bigram in title: + score += 5.0 + # 完全包含问题关键词 + if question_lower in title: + score += 10.0 + + scored.append((child, score)) + + scored.sort(key=lambda x: x[1], reverse=True) + return scored + + @classmethod + def _truncate_content(cls, content: str, max_len: int = 5000) -> str: + """截断文档内容，保留前 max_len 字符，优先保留完整段落""" + if len(content) <= max_len: + return content + # 找最后一个换行符，尽量保留完整段落 + truncated = content[:max_len] + last_newline = truncated.rfind("\n") + if last_newline > max_len * 0.8: + return truncated[:last_newline] + return truncated + + @classmethod + def _extract_relevant_paragraphs(cls, content: str, question: str, top_n: int = 10, min_len: int = 15) -> str: + """把文档拆成段落，只保留和问题最相关的 top_n 段""" + if not content or not question: + return content + + # 拆段落：按连续空行拆分，保留标题行 + raw_paragraphs = [] + for block in content.split("\n\n"): + block = block.strip() + if not block: + continue + # 一个 block 里可能有多行（比如列表），再按行拆 + lines = block.split("\n") + for line in lines: + line = line.strip() + if len(line) >= min_len: + raw_paragraphs.append(line) + + if not raw_paragraphs: + return content + + question_lower = question.lower() + question_chars = set(question_lower) + question_bigrams = set(question_lower[i : i + 2] for i in range(len(question_lower) - 1)) + + scored = [] + for para in raw_paragraphs: + para_lower = para.lower() + score = 0.0 + + # 完全包含问题关键词（最高权重） + if question_lower in para_lower: + score += 20.0 + + # 2-gram 匹配 + for bigram in question_bigrams: + if bigram in para_lower: + score += 3.0 + + # 单字匹配 + for char in question_chars: + if char in para_lower: + score += 1.0 + + scored.append((para, score)) + + # 按分数降序，取 top_n + scored.sort(key=lambda x: x[1], reverse=True) + selected = scored[:top_n] + + # 按原文顺序拼接（而不是按分数顺序），这样阅读更自然 + selected_set = set(p[0] for p in selected) + result_paras = [p for p in raw_paragraphs if p in selected_set] + + return "\n".join(result_paras) + @classmethod def _get_node_info_with_cache(cls, node_token: str, config: FeishuRetrievalConfig) -> dict: """带缓存获取节点信息""" 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 d18e608..46a2d8c 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 @@ -27,13 +27,8 @@ class ModelCompletion: class ModelClientService: @staticmethod def complete(db: Session, rag_result: RagResult) -> ModelCompletion: + model = ModelClientService._get_enabled_model(db) mock_model_enabled = _system_config_bool(db, "mock_model_enabled", get_settings().mock_model_enabled) - model = db.scalar( - select(ModelConfig) - .where(ModelConfig.enabled == 1) - .order_by(ModelConfig.id.desc()) - .limit(1) - ) if mock_model_enabled: model_name = model.model_name if model is not None else "mock-model" answer = _mock_answer(rag_result) @@ -42,7 +37,6 @@ class ModelClientService: raise ExternalServiceError("未启用可用模型，请先在模型管理中启用一个模型。", provider="model") model_name = model.model_name answer = _call_configured_model(model, rag_result) - return ModelCompletion( answer=answer, model_id=model.id if model is not None else None, @@ -51,6 +45,26 @@ class ModelClientService: output_token=_rough_token_count(answer), ) + @staticmethod + def summarize(db: Session, messages_text: str) -> str | None: + """调用模型生成对话摘要，失败时返回 None（不阻断主流程）""" + model = ModelClientService._get_enabled_model(db) + if model is None: + return None + try: + return _generate_summary(model, messages_text) + except Exception: + return None + + @staticmethod + def _get_enabled_model(db: Session) -> ModelConfig | None: + return db.scalar( + select(ModelConfig) + .where(ModelConfig.enabled == 1) + .order_by(ModelConfig.id.desc()) + .limit(1) + ) + @staticmethod def test_model(model: ModelConfig) -> dict[str, Any]: test_prompt = "请回复 OK，用于测试模型配置是否可用。" @@ -150,6 +164,22 @@ 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: + """用模型把历史对话压缩成摘要""" + summary_prompt = ( + "请将以下对话历史压缩成一段简洁的摘要，保留关键信息（用户问了什么、AI回答了什么要点、" + "是否有未解决的问题）。摘要不超过 300 字，用中文输出。\n\n" + f"对话历史：\n{messages_text}" + ) + rag_result = RagResult( + question=summary_prompt, + knowledge_scopes=[], + chunks=[], + prompt=summary_prompt, + ) + return _call_configured_model(model, rag_result, allow_no_hit=True) + + def _call_openai_compatible_model(model: ModelConfig, rag_result: RagResult) -> str: payload = { "model": model.model_name, 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 569eb1a..eac0856 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 @@ -6,12 +6,18 @@ from sqlalchemy import select 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.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: @@ -40,10 +46,16 @@ class RagResult: class RagService: @staticmethod - def build_result(db: Session, user: User, question: str) -> RagResult: + def build_result( + db: Session, + user: User, + question: str, + history: list[ChatMessage] | None = None, + session_summary: str | None = None, + ) -> RagResult: scopes = KnowledgeAccessService.get_allowed_knowledge(db, user) chunks = FeishuKnowledgeService.retrieve(question, scopes, db) - prompt = PromptService.build_prompt(db, question, chunks) + prompt = PromptService.build_prompt(db, question, chunks, history, session_summary) return RagResult(question=question, knowledge_scopes=scopes, chunks=chunks, prompt=prompt) @@ -54,15 +66,65 @@ class PromptService: ) @classmethod - def build_prompt(cls, db: Session, question: str, chunks: list[RetrievedChunk]) -> str: + def build_prompt( + cls, + db: Session, + question: str, + chunks: list[RetrievedChunk], + history: list[ChatMessage] | None = None, + session_summary: str | None = None, + ) -> str: prompt = cls._load_active_prompt(db) + + # 知识库上下文 context = "\n\n".join( f"[知识片段 {index}] 来源：{chunk.knowledge_name} / {chunk.title}\n{chunk.content}" for index, chunk in enumerate(chunks, start=1) ) if not context: context = "未检索到相关知识片段。" - return f"{prompt}\n\n{context}\n\n用户问题：{question.strip()}" + + parts = [prompt] + + # 对话历史：摘要 + 最近 N 轮原文 + history_part = cls._build_history_section(history, session_summary) + if history_part: + parts.append(history_part) + + 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": "AI助手"} + 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 "" @classmethod def _load_active_prompt(cls, db: Session) -> str: