diff --git a/backend/python/app/api/chat_context.py b/backend/python/app/api/chat_context.py index fed8117..30fbc5d 100644 --- a/backend/python/app/api/chat_context.py +++ b/backend/python/app/api/chat_context.py @@ -1,1392 +1,2819 @@ -# Auto-extracted from chat.py during Phase 1 refactor (Strangler Fig). -# Source of truth: scripts/phase1_extract.py. Do not hand-merge back. - -from fastapi import APIRouter, Depends, HTTPException, Query, Response -from sqlalchemy.orm import Session -from sqlalchemy import and_, desc, asc, func -from pydantic import BaseModel, Field -from typing import Optional -import uuid -from datetime import datetime, timezone, timedelta -import re -import json -import asyncio -import math -import httpx -import os -import logging -from ..database import get_db, SessionLocal -from ..models import ( - - - - User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog, ConversationSummary, - - - - KnowledgeBase, KnowledgeDocument, KnowledgeDocumentChunk, EvolutionSkill, - - - - EvolutionTrajectory, MarketApp, AppPurchase, AppInstallation, - - - - MCPServer, - - - - gen_uuid - - - -) -from ..auth import get_current_user -from ..platform_knowledge import get_platform_knowledge -from ..mcp_client import list_mcp_tools, call_mcp_tool -from ..middleware.log_service import LogService -from ..capability_registry import resolve_auto_model -from app.api.chat_skill import (_bm25_search) -from app.api.chat_utils import (logger) - -def _retrieve_kb_context(db: Session, kb_ids: list, query: str, max_chars: int = 3000) -> str: - - - - """ - - - - Stage C: 增强版知识库检索 — 使用BM25排序替代朴素拼接 - - - - 1. 获取每个KB的文档 - - - - 2. 构建文档列表(content+title+metadata) - - - - 3. BM25检索最相关的段落 - - - - 4. 拼接为上下文注入system prompt - - - - 同时叠加 Dify 外部知识库(RAG)检索结果。 - - - - """ - - - - local_context = "" - - - - - - - - if kb_ids: - - - - kbs = db.query(KnowledgeBase).filter(KnowledgeBase.id.in_(kb_ids)).all() - - - - if kbs: - - - - # 收集所有文档段落 - - - - # Stage C.1: 优先使用 KnowledgeDocumentChunk 表的预分段数据 - - - - all_segments = [] - - - - for kb in kbs: - - - - # 先查chunk表 - - - - chunks = db.query(KnowledgeDocumentChunk).filter( - - - - KnowledgeDocumentChunk.knowledge_base_id == kb.id - - - - ).order_by(KnowledgeDocumentChunk.chunk_index).limit(30).all() - - - - - - - - if chunks: - - - - # 使用预分段数据 - - - - for chunk in chunks: - - - - if chunk.content and len(chunk.content.strip()) >= 20: - - - - all_segments.append({ - - - - "content": "[{}/{}/段{}]: {}".format( - - - - kb.name, chunk.document_id[:8], chunk.chunk_index, chunk.content[:500] - - - - ), - - - - "raw_content": chunk.content, - - - - "kb_name": kb.name, - - - - }) - - - - else: - - - - # 退化: 使用原始文档分段 - - - - docs = db.query(KnowledgeDocument).filter( - - - - KnowledgeDocument.knowledge_base_id == kb.id, - - - - KnowledgeDocument.status == "completed" - - - - ).order_by(KnowledgeDocument.updated_at.desc()).limit(20).all() - - - - - - - - for doc in docs: - - - - content = doc.content or "" - - - - if not content: - - - - continue - - - - - - - - # 如果文档较长, 分段处理 (每段~800字) - - - - if len(content) > 800: - - - - paragraphs = re.split(r'\n\n+', content) - - - - for pi, para in enumerate(paragraphs): - - - - if len(para.strip()) >= 20: - - - - all_segments.append({ - - - - "content": "[{}/{}/段{}]: {}".format( - - - - kb.name, doc.name, pi+1, para.strip()[:500] - - - - ), - - - - "raw_content": para.strip(), - - - - "kb_name": kb.name, - - - - "doc_name": doc.name, - - - - }) - - - - else: - - - - all_segments.append({ - - - - "content": "[{}/{}]: {}".format(kb.name, doc.name, content[:500]), - - - - "raw_content": content, - - - - "kb_name": kb.name, - - - - "doc_name": doc.name, - - - - }) - - - - - - - - if not all_segments: - - - - # 退化: 使用原始朴素拼接 - - - - kb_names = ", ".join([k.name for k in kbs]) - - - - local_context = f"参考知识库: {kb_names}" - - - - else: - - - - # BM25检索最相关的段落 - - - - top_segments = _bm25_search(query, all_segments, top_k=8) - - - - context_parts = [seg["content"] for seg in top_segments] - - - - local_context = "\n".join(context_parts) - - - - if len(local_context) > max_chars: - - - - local_context = local_context[:max_chars] + "...(内容截断)" - - - - - - - - # 叠加 Dify 外部知识库(RAG)检索 - - - - dify_context = "" - - - - try: - - - - from app.dify_kb import get_dify_kb_context - - - - dify_context = get_dify_kb_context(db, query, top_k=5) - - - - except Exception as e: - - - - logger.warning(f"Dify KB context retrieval failed: {e}") - - - - - - - - parts = [] - - - - if local_context: - - - - parts.append(local_context) - - - - if dify_context: - - - - parts.append("【Dify 知识库检索结果】\n" + dify_context) - - - - return "\n\n".join(parts) - -async def _ingest_answer_to_dify(conv_id: str, msg_id: str, content: str): - - - - """后台任务：将助手回答入库到 Dify 知识库（量化后可供 RAG 检索复用）。""" - - - - try: - - - - from app.dify_kb import aingest_text - - - - from app.database import SessionLocal - - - - from sqlalchemy import text as _t - - - - db = SessionLocal() - - - - try: - - - - row = db.execute(_t( - - - - "SELECT api_url, api_key, config FROM external_kb_services " - - - - "WHERE service_type='dify' AND is_enabled=1 LIMIT 1" - - - - )).fetchone() - - - - finally: - - - - db.close() - - - - if not row: - - - - return - - - - api_url, api_key, config_raw = row - - - - dataset_name = None - - - - if config_raw: - - - - try: - - - - import json - - - - cfg = config_raw if isinstance(config_raw, dict) else json.loads(config_raw) - - - - dataset_name = cfg.get("dataset_name") - - - - except Exception: - - - - pass - - - - await aingest_text(api_url, api_key, content, name=f"chat_{conv_id}_{msg_id}", dataset_name=dataset_name) - - - - except Exception as e: - - - - logger.warning(f"后台入库 Dify 失败: {e}") - -def _build_system_prompt( - - - - skills: list, - - - - agents: list, - - - - kb_context: str, - - - - user_content: str, - - - - mcp: list = None, - - mcp_functions: list = None, - - auto_discovered: bool = False, - all_mcp: list = None, - -) -> tuple: - - - - """ - - - - 动态组装system prompt + function definitions。 - - - - 返回: (system_prompt_str, function_definitions_list) - - - - """ - - - - system_parts = ["你是AIFlow平台的AI助手，具备专业的知识储备和工具调用能力。"] - - - - - - - - # 平台知识库注入（始终开启）：确保AI优先使用平台已有能力 - - - - try: - - - - pk = get_platform_knowledge() - - - - if pk: - - - - system_parts.append(f"【平台知识库——优先使用以下知识回答用户问题】\n{pk}") - - - - except Exception: - - - - pass - - - - - - - - # 回复格式要求：强制使用Markdown - - - - system_parts.append("回复格式要求：使用 Markdown 格式，层次分明即可。对于简短问题，用 1-2 段自然文字直接回答，不必强行分点、加标题或表格；仅在确实需要分步骤或列举时才使用列表，代码/配置用 ``` 代码块包裹。避免冗余空行，保持紧凑、美观、易读。") - - - - - - - - function_definitions = [] - - - - - - - - # Stage B.1: Agent角色注入（Issue 2：让用户询问 agent 能力时，AI 能“看到”该 agent 资料） - - - - if agents: - - - - for agent in agents: - - - - name = agent.get("name", "") - - - - role = agent.get("role_prompt", "") - - - - instructions = agent.get("instructions", "") - - - - desc = agent.get("description", "") - - - - block = [] - - - - if name: - - - - block.append( - - - - f"用户已选择智能体「{name}」。当用户询问该智能体的功能、能力、用法或相关问题时，" - - - - f"请基于以下资料准确作答，不要说“不知道这个智能体”。" - - - - ) - - - - if role: - - - - block.append(f"该智能体的角色设定: {role}") - - - - if desc: - - - - block.append(f"该智能体的简介: {desc}") - - - - if instructions: - - - - block.append(f"该智能体的行为指引: {instructions}") - - - - if block: - - - - system_parts.append("\n".join(block)) - - - - - - - - # Stage A.2: Skill工具注入 - - - - if skills: - - - - skill_descs = ", ".join([f"{s['name']}({s.get('category', 'general')})" for s in skills]) - - - - system_parts.append(f"可用技能工具: {skill_descs}") - - - - if auto_discovered: - - - - # Issue 8：每轮默认先寻找本账号已安装且平台已部署的技能，多个适用时 AI 自行判断选 1 个最优 - - - - system_parts.append( - - - - "重要规则: 以上技能为本账号已安装且平台已部署的技能。每轮对话请先判断用户请求是否可由其中某个技能处理；" - - - - "若多个技能都适用，请自行判断并选择最贴切的 1 个优先调用，不要一次性全部调用。" - - - - "调用对应工具后将获得执行结果，基于结果生成最终回复。若这些技能均不适用，则直接基于你的知识回答。" - - - - ) - - - - else: - - - - system_parts.append( - "重要规则：\n" - "1) 当用户请求需要由该技能【执行】某项任务时，优先通过 function calling 调用对应工具；调用后基于工具返回结果生成回复。\n" - "2) 当用户询问所选技能【本身的介绍、功能、用途、用法、特点】（例如“介绍一下这个技能”“它是什么”“怎么用”“能做什么”）时，" - "必须基于下方【已选技能资料】中的名称/描述/指令直接作答，不要泛泛介绍平台整体，也不要说“不知道这个技能”。\n" - "3) 若用户问题既不执行也不询问该技能本身，则按常规方式回答。" - ) - - - - - - - - # 为每个skill构建function definition - - - - for s in skills: - - - - # 从skill_content提取指令注入system prompt - - - - instructions = s.get("instructions", "") - - - - name = s.get("name", "") - if not auto_discovered: - # 显式选中技能：注入可引用的技能资料，便于“介绍/说明”类问题据此作答 - ref_parts = [f"【已选技能资料】名称：{name}"] - desc = s.get("description", "") - if desc: - ref_parts.append(f"简介：{desc}") - if instructions and len(instructions) < 1200: - ref_parts.append(f"使用指令：{instructions}") - system_parts.append("\n".join(ref_parts)) - elif instructions and len(instructions) < 800: - system_parts.append(f"技能「{name}」的详细指令: {instructions}") - - - - - - - - # 构建参数schema - - - - params = s.get("parameters", {}) - - - - if isinstance(params, dict) and len(params) > 0: - - - - properties = {} - - - - required = [] - - - - for pname, pdef in params.items(): - - - - if isinstance(pdef, dict): - - - - properties[pname] = { - - - - "type": pdef.get("type", "string"), - - - - "description": pdef.get("description", f"传递给{s['name']}的{pname}参数"), - - - - } - - - - if pdef.get("required", True): - - - - required.append(pname) - - - - else: - - - - properties[pname] = {"type": "string", "description": f"传递给{s['name']}的{pname}参数"} - - - - required.append(pname) - - - - - - - - if not properties: - - - - properties = {"input": {"type": "string", "description": f"需要传递给{s['name']}的完整用户输入"}} - - - - required = ["input"] - - - - else: - - - - properties = {"input": {"type": "string", "description": f"需要传递给{s['name']}的完整用户输入"}} - - - - required = ["input"] - - - - - - - - desc = s.get("description", "") - - - - category = s.get("category", "general") - - - - func_desc = f"调用技能「{s['name']}」处理用户请求。使用场景: {desc}。当用户请求涉及{category}类任务时优先调用此工具。" - - - - - - - - function_definitions.append({ - - - - "name": s["func_name"], - - - - "description": func_desc, - - - - "parameters": { - - - - "type": "object", - - - - "properties": properties, - - - - "required": required, - - - - }, - - - - }) - - - - - - - - # KB context injection - - - - if kb_context: - - - - system_parts.append(f"参考以下知识库内容回答用户问题:\n{kb_context}") - - - - - - - - # 平台全量 MCP 清单（Issue：让用户询问"平台有哪些 MCP"时 AI 能准确列举，而非凭空编造） - if all_mcp: - _mcp_inventory = "\n".join([f"- {m.get('name','')}: {m.get('description','') or '（未填写描述）'}" for m in all_mcp]) - system_parts.append( - f"【平台已安装的 MCP 应用清单（共 {len(all_mcp)} 个，均为平台真实接入并启用）】\n{_mcp_inventory}\n" - "当用户询问“平台有哪些 MCP / 接入了哪些 MCP / 有哪些工具服务”时，请严格依据以上清单准确列举名称与功能，" - "不要编造或遗漏，也不要把未列出的服务说成已接入。" - ) - - # MCP 应用注入（Issue 3：让用户询问 MCP 能力时，AI 能“看到”该 MCP 的资料与地址） - - - - if mcp: - - - - mcp_lines = [] - - - - for m in mcp: - - - - name = m.get("name", "") - - - - desc = m.get("description", "") - - - - transport = m.get("transport", "stdio") - - - - endpoint = m.get("endpoint", "") - - - - line = f"用户已选择 MCP 应用「{name}」（传输方式: {transport}）" - - - - if desc: - - - - line += f"，功能简介: {desc}" - - - - if endpoint: - - - - line += f"，服务地址: {endpoint}" - - - - mcp_lines.append(line) - - - - system_parts.append("\n".join(mcp_lines)) - - - - system_parts.append( - - - - "用户已选择上述 MCP 应用。当请求需要借助外部工具/系统（如查询数据、调用 API、操作第三方服务）时，" - - - - "请优先参考这些 MCP 应用的能力来回答；若平台已接入对应工具，优先通过工具调用完成；" - - - - "不要说“不了解该 MCP”。" - - - - ) - - # 高优先级覆盖规则：显式选中工具时，回答必须基于该工具资料 - if (skills and not auto_discovered) or (mcp): - system_parts.append( - "【最高优先级覆盖规则】\n" - "用户在本轮已明确选择了工具（见上文【已选技能资料】或【用户已选择 MCP 应用】）。\n" - "如果用户的问题是在询问这些已选工具/技能/应用本身的介绍、功能、能做什么、怎么用、特点或用法，" - "你必须只基于上文提供的该工具具体资料作答；\n" - "不要引用《平台知识库》中关于技能市场、MCP 协议或平台整体能力的通用描述；" - "不要列举其它未选中的技能或 MCP 作为示例；也不要说“不知道”或“无法回答”。\n" - "若用户问题与已选工具本身无关，则按常规方式回答。" - ) - - - - - - - - - system_prompt = "\n".join(system_parts) - - - - # MCP function definitions 注入（让 LLM 可通过 function calling 真实调用 MCP 工具） - - if mcp_functions: - - function_definitions.extend(mcp_functions) - - - - return system_prompt, function_definitions - -def _get_conversation_context(db: Session, conv_id: str) -> dict: - - - - """ - - - - Stage B.2: 获取对话的持久化上下文 - - - - 对话级别的agent/skill/kb选择保存在Conversation字段中。 - - - - """ - - - - conv = db.query(Conversation).filter(Conversation.id == conv_id).first() - - - - if not conv: - - - - return {} - - - - - - - - # Conversation目前没有存储agent/skill/kb的JSON字段 - - - - # 我们从第一条消息中获取(如果存在), 或返回空 - - - - # 后续可通过数据库迁移添加Conversation.context字段 - - - - - - - - # 查询该对话中最早一条含有skill_ids/agent_ids/knowledge_ids的消息 - - - - first_tool_msg = db.query(Message).filter( - - - - Message.conversation_id == conv_id, - - - - Message.role == "user", - - - - ).order_by(asc(Message.created_at)).first() - - - - - - - - context = {} - - - - if first_tool_msg: - - - - if first_tool_msg.skill_ids: - - - - context["skill_ids"] = first_tool_msg.skill_ids - - - - if first_tool_msg.agent_ids: - - - - context["agent_ids"] = first_tool_msg.agent_ids - - - - if first_tool_msg.knowledge_ids: - - - - context["knowledge_base_ids"] = first_tool_msg.knowledge_ids - - - - if getattr(first_tool_msg, "mcp_ids", None): - - - - context["mcp_ids"] = first_tool_msg.mcp_ids - - - - - - - - return context - - - -# ===== Phase 2 上下文引擎升级：分层记忆 + 压缩 + 隔离 ===== -logger = logging.getLogger(__name__) - - -def _estimate_tokens(text): - """粗略 token 估算：中英文混排约 2 字符/token。""" - if not text: - return 0 - return max(1, len(text) // 2) - - -def _build_chat_history(db, conv_id, max_tokens=6000, max_messages=40, - exclude_latest_user=True, include_summary=True): - """会话层上下文引擎：预算感知的历史组装 + 可选压缩摘要前置。 - - - 若 conversation_summaries 存在摘要，以 system 角色「对话摘要」前置（压缩层）。 - - 按 token 估算从新到旧累加，超预算即截断（隔离：不溢出上下文窗口）。 - - exclude_latest_user=True 时丢弃最新一条 user 消息（当前轮次另行发送）。 - """ - summary_block = None - if include_summary: - try: - summ = db.query(ConversationSummary).filter( - ConversationSummary.conversation_id == conv_id - ).first() - if summ and summ.summary_text: - summary_block = { - "role": "system", - "content": "【对话历史摘要，请据此理解用户此前的意图与已确认结论】\n" + summ.summary_text, - } - except Exception: - pass - rows = db.query(Message).filter( - Message.conversation_id == conv_id, - Message.role.in_(["user", "assistant"]), - ).order_by(desc(Message.created_at)).limit(max_messages).all() - if exclude_latest_user: - for i, m in enumerate(rows): - if m.role == "user": - rows = rows[:i] + rows[i + 1:] - break - kept = [] - used = 0 - for m in rows: # 新->旧 - t = _estimate_tokens(m.content or "") - if kept and used + t > max_tokens: - break - kept.append({"role": m.role, "content": m.content}) - used += t - kept.reverse() # 时间正序 - out = [] - if summary_block: - out.append(summary_block) - out.extend(kept) - return out - - -async def _summarize_via_llm(prompt): - """调用默认文本模型生成对话摘要；失败返回 None（压缩为可选项，不阻塞）。""" - try: - from ..capability_registry import resolve_auto_model - from app.api.chat_utils import _proxy_for_url - sdb = SessionLocal() - try: - ai_model = resolve_auto_model("text", sdb) - if not ai_model: - return None - provider = sdb.query(ModelProvider).filter( - ModelProvider.id == ai_model.provider_id - ).first() - api_key = ai_model.api_key or (provider.api_key if provider else None) - if not provider or not provider.api_base_url or not api_key: - return None - api_url = f"{provider.api_base_url.rstrip('/')}/v1/chat/completions" - body = { - "model": ai_model.name, - "messages": [ - {"role": "system", "content": "你是专业的对话摘要助手，输出简洁中文摘要。"}, - {"role": "user", "content": prompt}, - ], - "max_tokens": 512, - "temperature": 0.2, - } - async with httpx.AsyncClient(proxy=_proxy_for_url(api_url), trust_env=False, timeout=60) as client: - resp = await client.post( - api_url, - headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}, - json=body, - ) - if resp.status_code != 200: - return None - data = resp.json() - return data.get("choices", [{}])[0].get("message", {}).get("content", "").strip() - finally: - sdb.close() - except Exception as e: - logger.warning(f"[context] summarize llm failed: {e}") - return None - - -async def _maybe_compress_history(conv_id): - """后台异步压缩：将受保护窗口外的旧消息摘要落库 conversation_summaries。 - - 失败静默回滚，绝不阻塞用户本次响应（fire-and-forget）。 - """ - sdb = None - try: - sdb = SessionLocal() - all_msgs = sdb.query(Message).filter( - Message.conversation_id == conv_id, - Message.role.in_(["user", "assistant"]), - ).order_by(asc(Message.created_at)).all() - PROTECTED = 20 - if len(all_msgs) <= PROTECTED: - return - compressible = all_msgs[: len(all_msgs) - PROTECTED] - if len(compressible) < 8: - return - chunk = compressible[:30] - text = "\n".join(f"{m.role}: {m.content}" for m in chunk) - existing = sdb.query(ConversationSummary).filter( - ConversationSummary.conversation_id == conv_id - ).first() - prior = existing.summary_text if existing else "" - prompt = ( - "请把以下对话片段浓缩为简明摘要，保留用户核心诉求、已确认结论、关键事实与待办。" - "若提供『已有摘要』，请合并而非重复。\n\n" - f"已有摘要：\n{prior}\n\n待摘要片段：\n{text}" - ) - summary = await _summarize_via_llm(prompt) - if not summary: - return - if existing: - existing.summary_text = summary - existing.token_count = _estimate_tokens(summary) - existing.updated_at = datetime.now() - else: - sdb.add(ConversationSummary( - conversation_id=conv_id, - summary_text=summary, - token_count=_estimate_tokens(summary), - updated_at=datetime.now(), - )) - sdb.commit() - except Exception as e: - logger.warning(f"[context] compress history failed conv={conv_id}: {e}") - if sdb is not None: - try: - sdb.rollback() - except Exception: - pass - finally: - if sdb is not None: - try: - sdb.close() - except Exception: - pass +# Auto-extracted from chat.py during Phase 1 refactor (Strangler Fig). + +# Source of truth: scripts/phase1_extract.py. Do not hand-merge back. + + + +from fastapi import APIRouter, Depends, HTTPException, Query, Response + +from sqlalchemy.orm import Session + +from sqlalchemy import and_, desc, asc, func + +from pydantic import BaseModel, Field + +from typing import Optional + +import uuid + +from datetime import datetime, timezone, timedelta + +import re + +import json + +import asyncio + +import math + +import httpx + +import os + +import logging + +from ..database import get_db, SessionLocal + +from ..models import ( + + + + + + + + User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog, ConversationSummary, + + + + + + + + KnowledgeBase, KnowledgeDocument, KnowledgeDocumentChunk, EvolutionSkill, + + + + + + + + EvolutionTrajectory, MarketApp, AppPurchase, AppInstallation, + + + + + + + + MCPServer, + + + + + + + + gen_uuid + + + + + + + +) + +from ..auth import get_current_user + +from ..platform_knowledge import get_platform_knowledge + +from ..mcp_client import list_mcp_tools, call_mcp_tool + +from ..middleware.log_service import LogService + +from ..capability_registry import resolve_auto_model + +from app.api.chat_skill import (_bm25_search) + +from app.kb_router import kb_router_search + +from app.api.chat_utils import (logger) + + + +async def _retrieve_kb_context(db: Session, kb_ids: list, query: str, max_tokens: int = 1000) -> str: + + + + + + + + """ + + + + + + + + Stage C: 增强版知识库检索 — 使用BM25排序替代朴素拼接 + + + + + + + + 1. 获取每个KB的文档 + + + + + + + + 2. 构建文档列表(content+title+metadata) + + + + + + + + 3. BM25检索最相关的段落 + + + + + + + + 4. 拼接为上下文注入system prompt + + + + + + + + 同时叠加 Dify 外部知识库(RAG)检索结果。 + + + + + + + + """ + + + + + + + + local_context = "" + + + + + + + + + + + + + + + + if kb_ids: + + + + + + + + kbs = db.query(KnowledgeBase).filter(KnowledgeBase.id.in_(kb_ids)).all() + + + + + + + + if kbs: + + + + + + + + # 收集所有文档段落 + + + + + + + + # Stage C.1: 优先使用 KnowledgeDocumentChunk 表的预分段数据 + + + + + + + + all_segments = [] + + + + + + + + for kb in kbs: + + + + + + + + # 先查chunk表 + + + + + + + + chunks = db.query(KnowledgeDocumentChunk).filter( + + + + + + + + KnowledgeDocumentChunk.knowledge_base_id == kb.id + + + + + + + + ).order_by(KnowledgeDocumentChunk.chunk_index).limit(30).all() + + + + + + + + + + + + + + + + if chunks: + + + + + + + + # 使用预分段数据 + + + + + + + + for chunk in chunks: + + + + + + + + if chunk.content and len(chunk.content.strip()) >= 20: + + + + + + + + all_segments.append({ + + + + + + + + "content": "[{}/{}/段{}]: {}".format( + + + + + + + + kb.name, chunk.document_id[:8], chunk.chunk_index, chunk.content[:500] + + + + + + + + ), + + + + + + + + "raw_content": chunk.content, + + + + + + + + "kb_name": kb.name, + + + + + + + + }) + + + + + + + + else: + + + + + + + + # 退化: 使用原始文档分段 + + + + + + + + docs = db.query(KnowledgeDocument).filter( + + + + + + + + KnowledgeDocument.knowledge_base_id == kb.id, + + + + + + + + KnowledgeDocument.status == "completed" + + + + + + + + ).order_by(KnowledgeDocument.updated_at.desc()).limit(20).all() + + + + + + + + + + + + + + + + for doc in docs: + + + + + + + + content = doc.content or "" + + + + + + + + if not content: + + + + + + + + continue + + + + + + + + + + + + + + + + # 如果文档较长, 分段处理 (每段~800字) + + + + + + + + if len(content) > 800: + + + + + + + + paragraphs = re.split(r'\n\n+', content) + + + + + + + + for pi, para in enumerate(paragraphs): + + + + + + + + if len(para.strip()) >= 20: + + + + + + + + all_segments.append({ + + + + + + + + "content": "[{}/{}/段{}]: {}".format( + + + + + + + + kb.name, doc.name, pi+1, para.strip()[:500] + + + + + + + + ), + + + + + + + + "raw_content": para.strip(), + + + + + + + + "kb_name": kb.name, + + + + + + + + "doc_name": doc.name, + + + + + + + + }) + + + + + + + + else: + + + + + + + + all_segments.append({ + + + + + + + + "content": "[{}/{}]: {}".format(kb.name, doc.name, content[:500]), + + + + + + + + "raw_content": content, + + + + + + + + "kb_name": kb.name, + + + + + + + + "doc_name": doc.name, + + + + + + + + }) + + + + + + + + + + + + + + + + if not all_segments: + + + + + + + + # 退化: 使用原始朴素拼接 + + + + + + + + kb_names = ", ".join([k.name for k in kbs]) + + + + + + + + local_context = f"参考知识库: {kb_names}" + + + + + + + + else: + + + + + + + + # BM25检索最相关的段落 + + + + + + + + # 混合检索：BM25(关键词) + 向量语义 + RRF 融合；失败回退 BM25 + bm25_segs = _bm25_search(query, all_segments, top_k=8) + bm25_for_router = [ + {"content": seg.get("raw_content", seg.get("content", "")), + "metadata": {"label": seg.get("content", "")}} + for seg in bm25_segs + ] + merged = None + try: + merged = await kb_router_search( + query, kb_ids=kb_ids, top_k=8, + bm25_results=bm25_for_router, db_session=db, + use_vector=True, use_external=False, + ) + except Exception as e: + logger.warning(f"[kb] 混合检索失败, 回退 BM25: {e}") + if merged: + ctx_parts = [] + used = 0 + for seg in merged: + label = (seg.get("metadata") or {}).get("label") + c = label if label else seg.get("content", "") + t = _estimate_tokens(c) + if ctx_parts and used + t > max_tokens: + break + ctx_parts.append(c) + used += t + local_context = "\n".join(ctx_parts) + else: + ctx_parts = [seg["content"] for seg in bm25_segs] + joined = "\n".join(ctx_parts) + if _estimate_tokens(joined) > max_tokens: + trunc, u = [], 0 + for c in ctx_parts: + if u + _estimate_tokens(c) > max_tokens: + break + trunc.append(c) + u += _estimate_tokens(c) + joined = "\n".join(trunc) + "\n...(内容截断)" + local_context = joined + + + + + + + + + + + + + + + + # 叠加 Dify 外部知识库(RAG)检索 + + + + + + + + dify_context = "" + + + + + + + + try: + + + + + + + + from app.dify_kb import get_dify_kb_context + + + + + + + + dify_context = get_dify_kb_context(db, query, top_k=5) + + + + + + + + except Exception as e: + + + + + + + + logger.warning(f"Dify KB context retrieval failed: {e}") + + + + + + + + + + + + + + + + parts = [] + + + + + + + + if local_context: + + + + + + + + parts.append(local_context) + + + + + + + + if dify_context: + + + + + + + + parts.append("【Dify 知识库检索结果】\n" + dify_context) + + + + + + + + return "\n\n".join(parts) + + + +async def _ingest_answer_to_dify(conv_id: str, msg_id: str, content: str): + + + + + + + + """后台任务：将助手回答入库到 Dify 知识库（量化后可供 RAG 检索复用）。""" + + + + + + + + try: + + + + + + + + from app.dify_kb import aingest_text + + + + + + + + from app.database import SessionLocal + + + + + + + + from sqlalchemy import text as _t + + + + + + + + db = SessionLocal() + + + + + + + + try: + + + + + + + + row = db.execute(_t( + + + + + + + + "SELECT api_url, api_key, config FROM external_kb_services " + + + + + + + + "WHERE service_type='dify' AND is_enabled=1 LIMIT 1" + + + + + + + + )).fetchone() + + + + + + + + finally: + + + + + + + + db.close() + + + + + + + + if not row: + + + + + + + + return + + + + + + + + api_url, api_key, config_raw = row + + + + + + + + dataset_name = None + + + + + + + + if config_raw: + + + + + + + + try: + + + + + + + + import json + + + + + + + + cfg = config_raw if isinstance(config_raw, dict) else json.loads(config_raw) + + + + + + + + dataset_name = cfg.get("dataset_name") + + + + + + + + except Exception: + + + + + + + + pass + + + + + + + + await aingest_text(api_url, api_key, content, name=f"chat_{conv_id}_{msg_id}", dataset_name=dataset_name) + + + + + + + + except Exception as e: + + + + + + + + logger.warning(f"后台入库 Dify 失败: {e}") + + + +def _build_system_prompt( + + + + + + + + skills: list, + + + + + + + + agents: list, + + + + + + + + kb_context: str, + + + + + + + + user_content: str, + + + + + + + + mcp: list = None, + + + + mcp_functions: list = None, + + + + auto_discovered: bool = False, + + all_mcp: list = None, + + + +) -> tuple: + + + + + + + + """ + + + + + + + + 动态组装system prompt + function definitions。 + + + + + + + + 返回: (system_prompt_str, function_definitions_list) + + + + + + + + """ + + + + + + + + system_parts = ["你是AIFlow平台的AI助手，具备专业的知识储备和工具调用能力。"] + + + + + + + + + + + + + + + + # 平台知识库注入（始终开启）：确保AI优先使用平台已有能力 + + + + + + + + try: + + + + + + + + pk = get_platform_knowledge() + + + + + + + + if pk: + + + + + + + + system_parts.append(f"【平台知识库——优先使用以下知识回答用户问题】\n{pk}") + + + + + + + + except Exception: + + + + + + + + pass + + + + + + + + + + + + + + + + # 回复格式要求：强制使用Markdown + + + + + + + + system_parts.append("回复格式要求：使用 Markdown 格式，层次分明即可。对于简短问题，用 1-2 段自然文字直接回答，不必强行分点、加标题或表格；仅在确实需要分步骤或列举时才使用列表，代码/配置用 ``` 代码块包裹。避免冗余空行，保持紧凑、美观、易读。") + + + + + + + + + + + + + + + + function_definitions = [] + + + + + + + + + + + + + + + + # Stage B.1: Agent角色注入（Issue 2：让用户询问 agent 能力时，AI 能“看到”该 agent 资料） + + + + + + + + if agents: + + + + + + + + for agent in agents: + + + + + + + + name = agent.get("name", "") + + + + + + + + role = agent.get("role_prompt", "") + + + + + + + + instructions = agent.get("instructions", "") + + + + + + + + desc = agent.get("description", "") + + + + + + + + block = [] + + + + + + + + if name: + + + + + + + + block.append( + + + + + + + + f"用户已选择智能体「{name}」。当用户询问该智能体的功能、能力、用法或相关问题时，" + + + + + + + + f"请基于以下资料准确作答，不要说“不知道这个智能体”。" + + + + + + + + ) + + + + + + + + if role: + + + + + + + + block.append(f"该智能体的角色设定: {role}") + + + + + + + + if desc: + + + + + + + + block.append(f"该智能体的简介: {desc}") + + + + + + + + if instructions: + + + + + + + + block.append(f"该智能体的行为指引: {instructions}") + + + + + + + + if block: + + + + + + + + system_parts.append("\n".join(block)) + + + + + + + + + + + + + + + + # Stage A.2: Skill工具注入 + + + + + + + + if skills: + + + + + + + + skill_descs = ", ".join([f"{s['name']}({s.get('category', 'general')})" for s in skills]) + + + + + + + + system_parts.append(f"可用技能工具: {skill_descs}") + + + + + + + + if auto_discovered: + + + + + + + + # Issue 8：每轮默认先寻找本账号已安装且平台已部署的技能，多个适用时 AI 自行判断选 1 个最优 + + + + + + + + system_parts.append( + + + + + + + + "重要规则: 以上技能为本账号已安装且平台已部署的技能。每轮对话请先判断用户请求是否可由其中某个技能处理；" + + + + + + + + "若多个技能都适用，请自行判断并选择最贴切的 1 个优先调用，不要一次性全部调用。" + + + + + + + + "调用对应工具后将获得执行结果，基于结果生成最终回复。若这些技能均不适用，则直接基于你的知识回答。" + + + + + + + + ) + + + + + + + + else: + + + + + + + + system_parts.append( + + "重要规则：\n" + + "1) 当用户请求需要由该技能【执行】某项任务时，优先通过 function calling 调用对应工具；调用后基于工具返回结果生成回复。\n" + + "2) 当用户询问所选技能【本身的介绍、功能、用途、用法、特点】（例如“介绍一下这个技能”“它是什么”“怎么用”“能做什么”）时，" + + "必须基于下方【已选技能资料】中的名称/描述/指令直接作答，不要泛泛介绍平台整体，也不要说“不知道这个技能”。\n" + + "3) 若用户问题既不执行也不询问该技能本身，则按常规方式回答。" + + ) + + + + + + + + + + + + + + + + # 为每个skill构建function definition + + + + + + + + for s in skills: + + + + + + + + # 从skill_content提取指令注入system prompt + + + + + + + + instructions = s.get("instructions", "") + + + + + + + + name = s.get("name", "") + + if not auto_discovered: + + # 显式选中技能：注入可引用的技能资料，便于“介绍/说明”类问题据此作答 + + ref_parts = [f"【已选技能资料】名称：{name}"] + + desc = s.get("description", "") + + if desc: + + ref_parts.append(f"简介：{desc}") + + if instructions and len(instructions) < 1200: + + ref_parts.append(f"使用指令：{instructions}") + + system_parts.append("\n".join(ref_parts)) + + elif instructions and len(instructions) < 800: + + system_parts.append(f"技能「{name}」的详细指令: {instructions}") + + + + + + + + + + + + + + + + # 构建参数schema + + + + + + + + params = s.get("parameters", {}) + + + + + + + + if isinstance(params, dict) and len(params) > 0: + + + + + + + + properties = {} + + + + + + + + required = [] + + + + + + + + for pname, pdef in params.items(): + + + + + + + + if isinstance(pdef, dict): + + + + + + + + properties[pname] = { + + + + + + + + "type": pdef.get("type", "string"), + + + + + + + + "description": pdef.get("description", f"传递给{s['name']}的{pname}参数"), + + + + + + + + } + + + + + + + + if pdef.get("required", True): + + + + + + + + required.append(pname) + + + + + + + + else: + + + + + + + + properties[pname] = {"type": "string", "description": f"传递给{s['name']}的{pname}参数"} + + + + + + + + required.append(pname) + + + + + + + + + + + + + + + + if not properties: + + + + + + + + properties = {"input": {"type": "string", "description": f"需要传递给{s['name']}的完整用户输入"}} + + + + + + + + required = ["input"] + + + + + + + + else: + + + + + + + + properties = {"input": {"type": "string", "description": f"需要传递给{s['name']}的完整用户输入"}} + + + + + + + + required = ["input"] + + + + + + + + + + + + + + + + desc = s.get("description", "") + + + + + + + + category = s.get("category", "general") + + + + + + + + func_desc = f"调用技能「{s['name']}」处理用户请求。使用场景: {desc}。当用户请求涉及{category}类任务时优先调用此工具。" + + + + + + + + + + + + + + + + function_definitions.append({ + + + + + + + + "name": s["func_name"], + + + + + + + + "description": func_desc, + + + + + + + + "parameters": { + + + + + + + + "type": "object", + + + + + + + + "properties": properties, + + + + + + + + "required": required, + + + + + + + + }, + + + + + + + + }) + + + + + + + + + + + + + + + + # KB context injection + + + + + + + + if kb_context: + + + + + + + + system_parts.append(f"参考以下知识库内容回答用户问题:\n{kb_context}") + + + + + + + + + + + + + + + + # 平台全量 MCP 清单（Issue：让用户询问"平台有哪些 MCP"时 AI 能准确列举，而非凭空编造） + + if all_mcp: + + _mcp_inventory = "\n".join([f"- {m.get('name','')}: {m.get('description','') or '（未填写描述）'}" for m in all_mcp]) + + system_parts.append( + + f"【平台已安装的 MCP 应用清单（共 {len(all_mcp)} 个，均为平台真实接入并启用）】\n{_mcp_inventory}\n" + + "当用户询问“平台有哪些 MCP / 接入了哪些 MCP / 有哪些工具服务”时，请严格依据以上清单准确列举名称与功能，" + + "不要编造或遗漏，也不要把未列出的服务说成已接入。" + + ) + + + + # MCP 应用注入（Issue 3：让用户询问 MCP 能力时，AI 能“看到”该 MCP 的资料与地址） + + + + + + + + if mcp: + + + + + + + + mcp_lines = [] + + + + + + + + for m in mcp: + + + + + + + + name = m.get("name", "") + + + + + + + + desc = m.get("description", "") + + + + + + + + transport = m.get("transport", "stdio") + + + + + + + + endpoint = m.get("endpoint", "") + + + + + + + + line = f"用户已选择 MCP 应用「{name}」（传输方式: {transport}）" + + + + + + + + if desc: + + + + + + + + line += f"，功能简介: {desc}" + + + + + + + + if endpoint: + + + + + + + + line += f"，服务地址: {endpoint}" + + + + + + + + mcp_lines.append(line) + + + + + + + + system_parts.append("\n".join(mcp_lines)) + + + + + + + + system_parts.append( + + + + + + + + "用户已选择上述 MCP 应用。当请求需要借助外部工具/系统（如查询数据、调用 API、操作第三方服务）时，" + + + + + + + + "请优先参考这些 MCP 应用的能力来回答；若平台已接入对应工具，优先通过工具调用完成；" + + + + + + + + "不要说“不了解该 MCP”。" + + + + + + + + ) + + + + # 高优先级覆盖规则：显式选中工具时，回答必须基于该工具资料 + + if (skills and not auto_discovered) or (mcp): + + system_parts.append( + + "【最高优先级覆盖规则】\n" + + "用户在本轮已明确选择了工具（见上文【已选技能资料】或【用户已选择 MCP 应用】）。\n" + + "如果用户的问题是在询问这些已选工具/技能/应用本身的介绍、功能、能做什么、怎么用、特点或用法，" + + "你必须只基于上文提供的该工具具体资料作答；\n" + + "不要引用《平台知识库》中关于技能市场、MCP 协议或平台整体能力的通用描述；" + + "不要列举其它未选中的技能或 MCP 作为示例；也不要说“不知道”或“无法回答”。\n" + + "若用户问题与已选工具本身无关，则按常规方式回答。" + + ) + + + + + + + + + + + + + + + + + + system_prompt = "\n".join(system_parts) + + + + + + + + # MCP function definitions 注入（让 LLM 可通过 function calling 真实调用 MCP 工具） + + + + if mcp_functions: + + + + function_definitions.extend(mcp_functions) + + + + + + + + return system_prompt, function_definitions + + + +def _get_conversation_context(db: Session, conv_id: str) -> dict: + + + + + + + + """ + + + + + + + + Stage B.2: 获取对话的持久化上下文 + + + + + + + + 对话级别的agent/skill/kb选择保存在Conversation字段中。 + + + + + + + + """ + + + + + + + + conv = db.query(Conversation).filter(Conversation.id == conv_id).first() + + + + + + + + if not conv: + + + + + + + + return {} + + + + + + + + + + + + + + + + # Conversation目前没有存储agent/skill/kb的JSON字段 + + + + + + + + # 我们从第一条消息中获取(如果存在), 或返回空 + + + + + + + + # 后续可通过数据库迁移添加Conversation.context字段 + + + + + + + + + + + + + + + + # 查询该对话中最早一条含有skill_ids/agent_ids/knowledge_ids的消息 + + + + + + + + first_tool_msg = db.query(Message).filter( + + + + + + + + Message.conversation_id == conv_id, + + + + + + + + Message.role == "user", + + + + + + + + ).order_by(asc(Message.created_at)).first() + + + + + + + + + + + + + + + + context = {} + + + + + + + + if first_tool_msg: + + + + + + + + if first_tool_msg.skill_ids: + + + + + + + + context["skill_ids"] = first_tool_msg.skill_ids + + + + + + + + if first_tool_msg.agent_ids: + + + + + + + + context["agent_ids"] = first_tool_msg.agent_ids + + + + + + + + if first_tool_msg.knowledge_ids: + + + + + + + + context["knowledge_base_ids"] = first_tool_msg.knowledge_ids + + + + + + + + if getattr(first_tool_msg, "mcp_ids", None): + + + + + + + + context["mcp_ids"] = first_tool_msg.mcp_ids + + + + + + + + + + + + + + + + return context + + + + + + + +# ===== Phase 2 上下文引擎升级：分层记忆 + 压缩 + 隔离 ===== + +logger = logging.getLogger(__name__) + + + + + +_tiktoken_enc = None + +def _estimate_tokens(text): + + """精确 token 估算：优先 tiktoken(cl100k_base)，失败回退 2 字符/token。""" + + if not text: + + return 0 + + global _tiktoken_enc + + if _tiktoken_enc is None: + + try: + + import tiktoken + + _tiktoken_enc = tiktoken.get_encoding("cl100k_base") + + except Exception: + + _tiktoken_enc = False + + if _tiktoken_enc: + + try: + + return len(_tiktoken_enc.encode(text)) + + except Exception: + + pass + + return max(1, len(text) // 2) + + + + + +def _build_chat_history(db, conv_id, max_tokens=6000, max_messages=40, + + exclude_latest_user=True, include_summary=True): + + """会话层上下文引擎：预算感知的历史组装 + 可选压缩摘要前置。 + + + + - 若 conversation_summaries 存在摘要，以 system 角色「对话摘要」前置（压缩层）。 + + - 按 token 估算从新到旧累加，超预算即截断（隔离：不溢出上下文窗口）。 + + - exclude_latest_user=True 时丢弃最新一条 user 消息（当前轮次另行发送）。 + + """ + + summary_block = None + + if include_summary: + + try: + + summ = db.query(ConversationSummary).filter( + + ConversationSummary.conversation_id == conv_id + + ).first() + + if summ and summ.summary_text: + + summary_block = { + + "role": "system", + + "content": "【对话历史摘要，请据此理解用户此前的意图与已确认结论】\n" + summ.summary_text, + + } + + except Exception: + + pass + + rows = db.query(Message).filter( + + Message.conversation_id == conv_id, + + Message.role.in_(["user", "assistant"]), + + ).order_by(desc(Message.created_at)).limit(max_messages).all() + + if exclude_latest_user: + + for i, m in enumerate(rows): + + if m.role == "user": + + rows = rows[:i] + rows[i + 1:] + + break + + kept = [] + + used = 0 + + for m in rows: # 新->旧 + + t = _estimate_tokens(m.content or "") + + if kept and used + t > max_tokens: + + break + + kept.append({"role": m.role, "content": m.content}) + + used += t + + kept.reverse() # 时间正序 + + out = [] + + if summary_block: + + out.append(summary_block) + + out.extend(kept) + + return out + + + + + +async def _summarize_via_llm(prompt): + + """调用默认文本模型生成对话摘要；失败返回 None（压缩为可选项，不阻塞）。""" + + try: + + from ..capability_registry import resolve_auto_model + + from app.api.chat_utils import _proxy_for_url + + sdb = SessionLocal() + + try: + + ai_model = resolve_auto_model("text", sdb) + + if not ai_model: + + return None + + provider = sdb.query(ModelProvider).filter( + + ModelProvider.id == ai_model.provider_id + + ).first() + + api_key = ai_model.api_key or (provider.api_key if provider else None) + + if not provider or not provider.api_base_url or not api_key: + + return None + + api_url = f"{provider.api_base_url.rstrip('/')}/v1/chat/completions" + + body = { + + "model": ai_model.name, + + "messages": [ + + {"role": "system", "content": "你是专业的对话摘要助手，输出简洁中文摘要。"}, + + {"role": "user", "content": prompt}, + + ], + + "max_tokens": 512, + + "temperature": 0.2, + + } + + async with httpx.AsyncClient(proxy=_proxy_for_url(api_url), trust_env=False, timeout=60) as client: + + resp = await client.post( + + api_url, + + headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}, + + json=body, + + ) + + if resp.status_code != 200: + + return None + + data = resp.json() + + return data.get("choices", [{}])[0].get("message", {}).get("content", "").strip() + + finally: + + sdb.close() + + except Exception as e: + + logger.warning(f"[context] summarize llm failed: {e}") + + return None + + + + + +async def _maybe_compress_history(conv_id): + + """后台异步压缩：将受保护窗口外的旧消息摘要落库 conversation_summaries。 + + + + 失败静默回滚，绝不阻塞用户本次响应（fire-and-forget）。 + + """ + + sdb = None + + try: + + sdb = SessionLocal() + + all_msgs = sdb.query(Message).filter( + + Message.conversation_id == conv_id, + + Message.role.in_(["user", "assistant"]), + + ).order_by(asc(Message.created_at)).all() + + PROTECTED = 20 + + if len(all_msgs) <= PROTECTED: + + return + + compressible = all_msgs[: len(all_msgs) - PROTECTED] + + if len(compressible) < 8: + + return + + chunk = compressible[:30] + + text = "\n".join(f"{m.role}: {m.content}" for m in chunk) + + existing = sdb.query(ConversationSummary).filter( + + ConversationSummary.conversation_id == conv_id + + ).first() + + prior = existing.summary_text if existing else "" + + prompt = ( + + "请把以下对话片段浓缩为简明摘要，保留用户核心诉求、已确认结论、关键事实与待办。" + + "若提供『已有摘要』，请合并而非重复。\n\n" + + f"已有摘要：\n{prior}\n\n待摘要片段：\n{text}" + + ) + + summary = await _summarize_via_llm(prompt) + + if not summary: + + return + + if existing: + + existing.summary_text = summary + + existing.token_count = _estimate_tokens(summary) + + existing.updated_at = datetime.now() + + else: + + sdb.add(ConversationSummary( + + conversation_id=conv_id, + + summary_text=summary, + + token_count=_estimate_tokens(summary), + + updated_at=datetime.now(), + + )) + + sdb.commit() + + except Exception as e: + + logger.warning(f"[context] compress history failed conv={conv_id}: {e}") + + if sdb is not None: + + try: + + sdb.rollback() + + except Exception: + + pass + + finally: + + if sdb is not None: + + try: + + sdb.close() + + except Exception: + + pass +