From 774f1648dccf41bc6143e1bba85e346da3d515e1 Mon Sep 17 00:00:00 2001 From: qyyfw Date: Fri, 10 Jul 2026 10:50:06 +0800 Subject: [PATCH] fix(chat/wechat): messages order by created_at,id (sub-second DATETIME(3)); container+MySQL TZ->Asia/Shanghai; correct UTC timestamps to CST; rebuild WeXin Bot history --- backend/python/app/api/chat.py | 2690 ++++++++++++------------- backend/python/app/api/im_bot_user.py | 9 +- 2 files changed, 1351 insertions(+), 1348 deletions(-) diff --git a/backend/python/app/api/chat.py b/backend/python/app/api/chat.py index f5510d7..213746d 100644 --- a/backend/python/app/api/chat.py +++ b/backend/python/app/api/chat.py @@ -1,1345 +1,1345 @@ -""" -增强版 Chat API — Phase 2.5 -包含: Skill ID统一(MarketApp映射), 真实Skill执行, Agent工作流注入, - 多轮上下文保持, BM25 KB检索, Function Calling循环增强 -""" -from fastapi import APIRouter, Depends, HTTPException, Query -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 math -import httpx -import os -import logging - -from ..database import get_db, SessionLocal -from ..models import ( - User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog, - KnowledgeBase, KnowledgeDocument, KnowledgeDocumentChunk, EvolutionSkill, - EvolutionTrajectory, MarketApp, AppPurchase, AppInstallation, - gen_uuid -) -from ..auth import get_current_user -from ..middleware.log_service import LogService - -router = APIRouter() - -logger = logging.getLogger(__name__) - -# CST timezone (UTC+8) for consistent time display -CST_TZ = timezone(timedelta(hours=8)) - - -def cst_now(): - """Get current time in CST timezone""" - return datetime.now(CST_TZ) - - -# ============================================================ -# Request Models -# ============================================================ - -class CreateConversationRequest(BaseModel): - title: str = Field(default="新对话") - model_id: Optional[str] = None - - -class UpdateConversationRequest(BaseModel): - title: Optional[str] = None - status: Optional[str] = None # active, archived - pinned: Optional[bool] = None - agent_ids: Optional[list] = None - skill_ids: Optional[list] = None - knowledge_base_ids: Optional[list] = None - - -class SendMessageRequest(BaseModel): - content: str - model_id: Optional[str] = None - skill_ids: Optional[list] = None - agent_ids: Optional[list] = None - knowledge_base_ids: Optional[list] = None - use_tools: Optional[bool] = False - - -# ============================================================ -# Skill Resolution: MarketApp → Skill Definition -# ============================================================ - -def _resolve_market_skills(db: Session, skill_ids: list, user_id: str) -> list: - """ - Stage A.1: 统一Skill ID空间 - 前端传MarketApp.id, 需要映射为可用的技能定义。 - 返回: [{id, name, description, skill_content, category, func_name}] - """ - if not skill_ids: - return [] - - # 1. 从MarketApp查询 (前端传的是MarketApp.id) - market_skills = db.query(MarketApp).filter( - MarketApp.id.in_(skill_ids), - MarketApp.type == "skill" - ).all() - - results = [] - for ms in market_skills: - skill_content = ms.skill_content or "" - # 尝试从skill_content解析SKILL.md格式 - # SKILL.md格式: ---\nname: ...\ndescription: ...\n---\n内容... - parsed = _parse_skill_content(skill_content) - - func_name = ms.name.replace(" ", "_").replace("-", "_").lower() - results.append({ - "id": ms.id, - "market_app_id": ms.id, - "name": ms.name, - "description": ms.description or parsed.get("description", ""), - "skill_content": skill_content, - "category": parsed.get("category", "general"), - "skill_type": parsed.get("skill_type", "general"), - "func_name": func_name, - "parameters": parsed.get("parameters", {"input": {"type": "string", "description": "用户输入内容"}}), - "instructions": parsed.get("instructions", ""), - }) - - # 2. 如果有未匹配的ID, 尝试查EvolutionSkill (兼容旧的进化技能ID) - matched_ids = {r["id"] for r in results} - unmatched = [sid for sid in skill_ids if sid not in matched_ids] - - if unmatched: - evo_skills = db.query(EvolutionSkill).filter( - EvolutionSkill.id.in_(unmatched), - EvolutionSkill.user_id == user_id - ).all() - - for es in evo_skills: - # 读取SKILL.md内容 - skill_md = "" - if es.skill_md_path and os.path.exists(es.skill_md_path): - try: - with open(es.skill_md_path, "r", encoding="utf-8") as f: - skill_md = f.read() - except Exception: - skill_md = "" - elif es.description: - skill_md = es.description - - parsed = _parse_skill_content(skill_md) - func_name = es.skill_name.replace(" ", "_").replace("-", "_").lower() - - results.append({ - "id": es.id, - "market_app_id": None, - "name": es.skill_name, - "description": es.description or parsed.get("description", ""), - "skill_content": skill_md, - "category": es.category or parsed.get("category", "general"), - "skill_type": parsed.get("skill_type", es.category or "general"), - "func_name": func_name, - "parameters": parsed.get("parameters", {"input": {"type": "string", "description": "用户输入内容"}}), - "instructions": parsed.get("instructions", ""), - }) - - return results - - -def _resolve_market_agents(db: Session, agent_ids: list, user_id: str) -> list: - """ - Stage B.1: 加载Agent配置 - 前端传MarketApp.id(agent类型), 需要解析AgentConfig。 - 返回: [{id, name, config, skills, knowledge_base_ids}] - """ - if not agent_ids: - return [] - - agents = db.query(MarketApp).filter( - MarketApp.id.in_(agent_ids), - MarketApp.type == "agent" - ).all() - - results = [] - for ag in agents: - # AgentConfig在skill_content字段(JSON序列化) - agent_config = {} - try: - if ag.skill_content: - agent_config = json.loads(ag.skill_content) - except Exception: - agent_config = {"role": ag.description or "AI助手", "instructions": ag.description or ""} - - # 提取配置字段 - role_prompt = agent_config.get("role", ag.description or "AI助手") - instructions = agent_config.get("instructions", "") - agent_skill_ids = agent_config.get("skills", []) # 关联的skill MarketApp IDs - agent_kb_ids = agent_config.get("knowledge_base_ids", []) # 关联的知识库IDs - agent_model_id = agent_config.get("model_id", None) # 指定使用的模型 - tools = agent_config.get("tools", []) # MCP server IDs(暂不用) - - results.append({ - "id": ag.id, - "name": ag.name, - "role_prompt": role_prompt, - "instructions": instructions, - "skill_ids": agent_skill_ids, - "knowledge_base_ids": agent_kb_ids, - "model_id": agent_model_id, - "tools": tools, - "description": ag.description or "", - }) - - return results - - -def _parse_skill_content(content: str) -> dict: - """ - 解析SKILL.md格式内容, 提取结构化信息。 - SKILL.md格式: - --- - name: 技能名 - description: 描述 - category: 分类 - skill_type: 类型 - parameters: JSON或简单定义 - --- - 技能指令/规则内容 - """ - result = { - "name": "", - "description": "", - "category": "general", - "skill_type": "general", - "parameters": {}, - "instructions": "", - } - - if not content: - return result - - # 尝试解析YAML front matter - if content.strip().startswith("---"): - parts = content.split("---") - if len(parts) >= 3: - front_matter = parts[1].strip() - body = parts[2].strip() if len(parts) > 2 else "" - - # 简单YAML解析 (不用pyyaml库) - for line in front_matter.split("\n"): - line = line.strip() - if ":" in line: - key, val = line.split(":", 1) - key = key.strip().lower() - val = val.strip() - if key == "name": - result["name"] = val - elif key == "description": - result["description"] = val - elif key == "category": - result["category"] = val - elif key == "skill_type": - result["skill_type"] = val - elif key == "parameters": - try: - result["parameters"] = json.loads(val) - except Exception: - # 简单格式: input:string,query:string - params = {} - for p in val.split(","): - p = p.strip() - if ":" in p: - pname, ptype = p.split(":", 1) - params[pname.strip()] = {"type": ptype.strip(), "description": pname.strip()} - result["parameters"] = params - - result["instructions"] = body - else: - result["instructions"] = content - else: - # 没有front matter, 全部当作instructions - result["instructions"] = content - # 尝试从第一行提取name - first_line = content.split("\n")[0].strip() - if first_line and len(first_line) < 50: - result["name"] = first_line - - return result - - -# ============================================================ -# BM25 Text Search (Stage C.2) -# ============================================================ - -def _bm25_search(query: str, documents: list, top_k: int = 5) -> list: - """ - 简易BM25检索 — 替代朴素的文档拼接。 - 对文档列表进行关键词匹配+TF-IDF排序, 返回最相关的top_k段。 - """ - if not documents or not query: - return documents[:top_k] - - # 中文简单分词(按字符+标点) - def tokenize(text: str) -> list: - # 移除标点, 拆分为2-4字词组 - text = re.sub(r'[^\w\u4e00-\u9fff]', ' ', text.lower()) - tokens = [] - words = text.split() - for w in words: - if len(w) >= 2: - tokens.append(w) - # 也生成2-gram - if len(w) >= 4: - for i in range(len(w) - 2): - tokens.append(w[i:i+2]) - return tokens - - query_tokens = tokenize(query) - if not query_tokens: - return documents[:top_k] - - scored_docs = [] - avg_doc_len = sum(len(tokenize(d.get("content", ""))) for d in documents) / max(len(documents), 1) - - # BM25参数 - k1 = 1.5 - b = 0.75 - - # 计算每个token的IDF - df = {} - for token in query_tokens: - for doc in documents: - if token in tokenize(doc.get("content", "")): - df[token] = df.get(token, 0) + 1 - - N = len(documents) - idf = {} - for token in query_tokens: - idf[token] = math.log((N - df.get(token, 0) + 0.5) / (df.get(token, 0) + 0.5) + 1) - - # 计算每个文档的BM25得分 - for doc in documents: - doc_tokens = tokenize(doc.get("content", "")) - doc_len = len(doc_tokens) - score = 0 - - for token in query_tokens: - tf = doc_tokens.count(token) - numerator = tf * (k1 + 1) - denominator = tf + k1 * (1 - b + b * doc_len / max(avg_doc_len, 1)) - score += idf.get(token, 0) * numerator / denominator - - scored_docs.append((score, doc)) - - # 按得分排序 - scored_docs.sort(key=lambda x: x[0], reverse=True) - return [d for s, d in scored_docs[:top_k]] - - -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 - """ - if not kb_ids: - return "" - - kbs = db.query(KnowledgeBase).filter(KnowledgeBase.id.in_(kb_ids)).all() - if not kbs: - return "" - - # 收集所有文档段落 - # 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]) - return f"参考知识库: {kb_names}" - - # BM25检索最相关的段落 - top_segments = _bm25_search(query, all_segments, top_k=8) - - # 拼接上下文 - context_parts = [seg["content"] for seg in top_segments] - context = "\n".join(context_parts) - - # 限制总长度 - if len(context) > max_chars: - context = context[:max_chars] + "...(内容截断)" - - return context - - -# ============================================================ -# Dynamic System Prompt Builder (Stage B) -# ============================================================ - -def _build_system_prompt( - skills: list, - agents: list, - kb_context: str, - user_content: str, -) -> tuple: - """ - 动态组装system prompt + function definitions。 - 返回: (system_prompt_str, function_definitions_list) - """ - system_parts = ["你是AIFlow平台的AI助手，具备专业的知识储备和工具调用能力。"] - - function_definitions = [] - - # Stage B.1: Agent角色注入 - if agents: - for agent in agents: - role = agent.get("role_prompt", "") - instructions = agent.get("instructions", "") - if role: - system_parts.append(f"当前角色设定: {role}") - if instructions: - system_parts.append(f"角色行为指引: {instructions}") - - # 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}") - system_parts.append("重要规则: 当用户请求涉及上述技能工具的处理范围时, 你必须优先通过function calling调用对应工具, 而不是直接回复文字。调用后将获得工具执行结果, 基于结果生成最终回复。") - - # 为每个skill构建function definition - for s in skills: - # 从skill_content提取指令注入system prompt - instructions = s.get("instructions", "") - if instructions and len(instructions) < 800: - system_parts.append(f"技能「{s['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}") - - system_prompt = "\n".join(system_parts) - return system_prompt, function_definitions - - -# ============================================================ -# Skill Tool Execution (Stage A.2 rewritten) -# ============================================================ - -def _execute_skill_tool( - db: Session, - func_name: str, - args: dict, - skills: list, - user_content: str, -) -> str: - """ - Stage A.2: 真实Skill执行引擎 — 替代mock实现 - 1. 查找匹配的skill定义 - 2. 读取skill_content(SKILL.md)中的指令 - 3. 根据指令类型组装执行prompt - 4. 返回结构化的执行结果给LLM - """ - # 查找匹配的skill - matched_skill = None - for s in skills: - expected_name = s["func_name"] - if func_name == expected_name or func_name == f"skill_{s['id'][:8]}": - matched_skill = s - break - - if not matched_skill: - # 退化: 返回通用结果 - return f"工具 '{func_name}' 执行结果: 已处理输入内容, 但未找到对应的技能定义。请基于通用知识回复用户。" - - skill_name = matched_skill["name"] - instructions = matched_skill.get("instructions", "") - input_text = args.get("input", user_content) - - # 根据skill_content中的指令生成真实执行结果 - # 核心思路: 将skill的instructions作为"执行逻辑", - # 把input_text传入, 让结果包含指令要求的处理格式 - - if instructions: - # 有明确指令的技能 — 按指令结构化输出 - result_parts = [ - f"技能「{skill_name}」执行结果:", - f"---", - ] - - # 尝试从指令中提取输出格式要求 - # 如果指令包含"输出"或"格式"关键字, 遵循其要求 - output_format = "" - if "输出格式" in instructions or "output format" in instructions.lower(): - # 提取格式描述 - for line in instructions.split("\n"): - if "输出" in line or "格式" in line or "format" in line.lower(): - output_format += line + "\n" - - # 根据skill_type增强处理 - skill_type = matched_skill.get("skill_type", "general") - - if skill_type == "writer" or "写作" in skill_name or "写" in instructions: - result_parts.append(f"基于技能「{skill_name}」的写作指令处理用户输入:") - result_parts.append(f"主题: {input_text[:200]}") - result_parts.append(f"根据技能规则: {instructions[:300]}") - result_parts.append(f"处理完成 — 请基于上述技能指令和输入内容, 生成符合技能要求的写作输出。") - elif skill_type == "analyzer" or "分析" in skill_name or "分析" in instructions: - result_parts.append(f"基于技能「{skill_name}」的分析指令处理用户输入:") - result_parts.append(f"分析对象: {input_text[:200]}") - result_parts.append(f"分析维度(按技能规则): {instructions[:300]}") - result_parts.append(f"关键发现: 识别到{max(1, len(input_text)//50)}个分析要点") - result_parts.append(f"处理完成 — 请基于上述技能指令, 对输入内容进行深入分析并给出结构化的分析报告。") - elif skill_type == "translator" or "翻译" in skill_name or "翻译" in instructions: - result_parts.append(f"基于技能「{skill_name}」的翻译指令处理用户输入:") - result_parts.append(f"原文: {input_text[:200]}") - result_parts.append(f"翻译规则: {instructions[:300]}") - result_parts.append(f"处理完成 — 请基于技能翻译规则, 将输入内容翻译为目标语言。") - elif skill_type == "summarizer" or "总结" in skill_name or "摘要" in instructions: - result_parts.append(f"基于技能「{skill_name}」的摘要指令处理用户输入:") - result_parts.append(f"原文长度: {len(input_text)} 字符") - result_parts.append(f"摘要规则: {instructions[:300]}") - result_parts.append(f"处理完成 — 请基于技能摘要规则, 对输入内容进行精准摘要。") - else: - # 通用技能 — 传递指令让LLM执行 - result_parts.append(f"基于技能「{skill_name}」的指令处理用户请求:") - result_parts.append(f"用户输入: {input_text[:200]}") - result_parts.append(f"技能指令: {instructions[:500]}") - if output_format: - result_parts.append(f"期望输出格式: {output_format[:200]}") - result_parts.append(f"处理完成 — 请严格按照技能指令的规则和要求, 处理用户输入并生成最终回复。") - - result = "\n".join(result_parts) - else: - # 无明确指令的技能 — 提供基本信息让LLM自行处理 - desc = matched_skill.get("description", "") - result = ( - f"技能「{skill_name}」已激活。\n" - f"技能描述: {desc}\n" - f"用户输入: {input_text[:200]}\n" - f"请基于技能「{skill_name}」的专业能力处理上述输入。" - ) - - # 更新使用计数(如果关联了MarketApp) - market_app_id = matched_skill.get("market_app_id") - if market_app_id: - try: - app = db.query(MarketApp).filter(MarketApp.id == market_app_id).first() - if app: - app.install_count = (app.install_count or 0) + 1 - except Exception: - pass - - return result - - -# ============================================================ -# Conversation Context Persistence (Stage B.2) -# ============================================================ - -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 - - return context - - -# ============================================================ -# API Endpoints -# ============================================================ - -@router.get("/conversations") -def list_conversations( - status: Optional[str] = Query(None), - search: Optional[str] = Query(None), - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - query = db.query(Conversation).filter(Conversation.user_id == current_user.id) - - if status: - query = query.filter(Conversation.status == status) - else: - query = query.filter(Conversation.status != "archived") - - if search: - query = query.filter(Conversation.title.ilike(f"%{search}%")) - - query = query.order_by(desc(Conversation.pinned), desc(Conversation.updated_at)) - - conversations = query.all() - result = [] - for conv in conversations: - result.append({ - "id": conv.id, - "title": conv.title, - "model_id": conv.model_id, - "status": conv.status, - "pinned": conv.pinned, - "knowledge_graph_count": conv.knowledge_graph_count or 0, - "source": getattr(conv, "source", None), - "created_at": str(conv.created_at) if conv.created_at else None, - "updated_at": str(conv.updated_at) if conv.updated_at else None, - }) - - return {"code": 0, "data": result, "message": "ok"} - - -@router.post("/conversations") -def create_conversation( - req: CreateConversationRequest, - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - conv = Conversation( - id=gen_uuid(), - user_id=current_user.id, - title=req.title, - model_id=req.model_id, - status="active", - pinned=False, - knowledge_graph_count=0, - created_at=cst_now(), - updated_at=cst_now(), - ) - db.add(conv) - db.commit() - db.refresh(conv) - - return { - "code": 0, - "data": { - "id": conv.id, - "title": conv.title, - "model_id": conv.model_id, - "status": conv.status, - "pinned": conv.pinned, - "knowledge_graph_count": conv.knowledge_graph_count, - "created_at": str(conv.created_at) if conv.created_at else None, - "updated_at": str(conv.updated_at) if conv.updated_at else None, - }, - "message": "创建成功", - } - - -@router.put("/conversations/{conv_id}") -def update_conversation( - conv_id: str, - req: UpdateConversationRequest, - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - conv = db.query(Conversation).filter( - and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) - ).first() - if not conv: - raise HTTPException(status_code=404, detail="对话不存在") - - if req.title is not None: - conv.title = req.title - if req.status is not None: - conv.status = req.status - if req.pinned is not None: - conv.pinned = req.pinned - # Stage B.2: 保存agent/skill/kb选择到对话上下文 - # (存储在第一条system消息的knowledge_ids/agent_ids/skill_ids字段中) - conv.updated_at = cst_now() - - db.commit() - db.refresh(conv) - - return {"code": 0, "data": { - "id": conv.id, "title": conv.title, "status": conv.status, "pinned": conv.pinned, - "updated_at": str(conv.updated_at), - }, "message": "更新成功"} - - -@router.delete("/conversations/{conv_id}") -def delete_conversation( - conv_id: str, - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - conv = db.query(Conversation).filter( - and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) - ).first() - if not conv: - raise HTTPException(status_code=404, detail="对话不存在") - - conv.status = "archived" - conv.updated_at = cst_now() - db.commit() - - return {"code": 0, "message": "删除成功"} - - -@router.get("/conversations/{conv_id}/messages") -def get_messages( - conv_id: str, - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - conv = db.query(Conversation).filter( - and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) - ).first() - if not conv: - raise HTTPException(status_code=404, detail="对话不存在") - - messages = ( - db.query(Message) - .filter(Message.conversation_id == conv_id) - .order_by(asc(Message.created_at)) - .all() - ) - - result = [] - for msg in messages: - result.append({ - "id": msg.id, - "conversation_id": msg.conversation_id, - "role": msg.role, - "content": msg.content, - "token_used": msg.token_used or 0, - "model_id": msg.model_id, - "used_skills": msg.used_skills or [], - "created_at": str(msg.created_at) if msg.created_at else None, - }) - - return {"code": 0, "data": result, "message": "ok"} - - -@router.post("/conversations/{conv_id}/messages") -async def send_message( - conv_id: str, - req: SendMessageRequest, - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - conv = db.query(Conversation).filter( - and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) - ).first() - if not conv: - raise HTTPException(status_code=404, detail="对话不存在") - - # Stage B.2: 多轮上下文保持 — 如果当前请求没有skill/agent/kb, - # 从对话历史中继承上一次的选择 - effective_skill_ids = req.skill_ids or [] - effective_agent_ids = req.agent_ids or [] - effective_kb_ids = req.knowledge_base_ids or [] - effective_use_tools = req.use_tools or False - - if not effective_skill_ids and not effective_agent_ids and not effective_kb_ids: - # 从对话上下文继承 - prev_context = _get_conversation_context(db, conv_id) - if prev_context: - effective_skill_ids = prev_context.get("skill_ids", []) - effective_agent_ids = prev_context.get("agent_ids", []) - effective_kb_ids = prev_context.get("knowledge_base_ids", []) - if effective_skill_ids or effective_agent_ids: - effective_use_tools = True - - # 保存用户消息(含skill/agent/kb选择用于上下文保持) - user_msg = Message( - id=gen_uuid(), - conversation_id=conv_id, - role="user", - content=req.content, - token_used=len(req.content) // 2, - model_id=req.model_id or conv.model_id, - skill_ids=req.skill_ids if req.skill_ids else None, - agent_ids=req.agent_ids if req.agent_ids else None, - knowledge_ids=req.knowledge_base_ids if req.knowledge_base_ids else None, - created_at=cst_now(), - ) - db.add(user_msg) - - # ===== 核心聊天引擎 ===== - user_id = current_user.id - model_id = req.model_id or conv.model_id - prompt_tokens = 0 - completion_tokens = 0 - total_tokens = 0 - ai_content = "" - - ai_model = None - skills = [] # 初始化避免UnboundLocalError - agents = [] # 初始化避免UnboundLocalError - if model_id: - ai_model = db.query(AIModel).filter(AIModel.id == model_id).first() - - if ai_model and ai_model.endpoint: - provider = db.query(ModelProvider).filter(ModelProvider.id == ai_model.provider_id).first() - api_key = ai_model.api_key or (provider.api_key if provider else None) - api_url = ai_model.endpoint - - if not api_url and provider and provider.api_base_url: - api_url = f"{provider.api_base_url.rstrip('/')}/v1/chat/completions" - - if api_key and api_url: - try: - # Stage A+B: 解析Skills和Agents - skills = _resolve_market_skills(db, effective_skill_ids, user_id) - agents = _resolve_market_agents(db, effective_agent_ids, user_id) - - # Stage B.1: Agent可能自带skill_ids和kb_ids - agent_extra_skill_ids = [] - agent_extra_kb_ids = [] - agent_model_override = None - for agent in agents: - if agent.get("skill_ids"): - agent_extra_skill_ids.extend(agent["skill_ids"]) - if agent.get("knowledge_base_ids"): - agent_extra_kb_ids.extend(agent["knowledge_base_ids"]) - if agent.get("model_id"): - agent_model_override = agent["model_id"] - - # 合并Agent自带的skills和KBs - all_skill_ids = list(set(effective_skill_ids + agent_extra_skill_ids)) - all_kb_ids = list(set(effective_kb_ids + agent_extra_kb_ids)) - - if agent_extra_skill_ids or agent_extra_kb_ids: - # 重新解析合并后的skills - skills = _resolve_market_skills(db, all_skill_ids, user_id) - effective_use_tools = True - - # Stage C: 增强版KB检索 - kb_context = _retrieve_kb_context(db, all_kb_ids, req.content) - - # Stage B: 动态组装system prompt + function definitions - system_prompt, function_definitions = _build_system_prompt( - skills=skills, - agents=agents, - kb_context=kb_context, - user_content=req.content, - ) - - # 构建消息列表 - messages = [{"role": "system", "content": system_prompt}] - - # 加载对话历史(最多10条) - history = db.query(Message).filter( - Message.conversation_id == conv_id - ).order_by(asc(Message.created_at)).limit(10).all() - for h in history: - if h.role == "user" or h.role == "assistant": - messages.append({"role": h.role, "content": h.content}) - - messages.append({"role": "user", "content": req.content}) - - # Stage B.3: Function Calling循环(最多5轮) - max_tool_rounds = 5 - for round_num in range(max_tool_rounds): - request_body = { - "model": ai_model.name, - "messages": messages, - "max_tokens": 2048, - } - if function_definitions and effective_use_tools: - request_body["tools"] = [ - {"type": "function", "function": fd} - for fd in function_definitions - ] - request_body["tool_choice"] = "auto" - - async with httpx.AsyncClient(timeout=ai_model.timeout or 120) as client: - resp = await client.post( - api_url, - headers={ - "Authorization": f"Bearer {api_key}", - "Content-Type": "application/json" - }, - json=request_body - ) - if resp.status_code != 200: - if round_num == 0: - ai_content = f"AI模型调用失败 (HTTP {resp.status_code}), 请检查模型配置。" - break - - data = resp.json() - choice = data.get("choices", [{}])[0] - message = choice.get("message", {}) - - # 记录token使用量 - usage = data.get("usage", {}) - prompt_tokens += usage.get("prompt_tokens", 0) - completion_tokens += usage.get("completion_tokens", 0) - total_tokens += usage.get("total_tokens", 0) - - # 检查是否有tool_calls - tool_calls = message.get("tool_calls", []) - if tool_calls and effective_use_tools: - # 将assistant消息(含tool_calls)加入上下文 - messages.append({ - "role": "assistant", - "content": message.get("content") or "", - "tool_calls": tool_calls, - }) - - # Stage A.2: 真实Skill执行 - for tc in tool_calls: - func_name = tc.get("function", {}).get("name", "") - func_args_str = tc.get("function", {}).get("arguments", "{}") - try: - func_args = json.loads(func_args_str) - except Exception: - func_args = {"input": func_args_str} - - tool_result = _execute_skill_tool( - db, func_name, func_args, skills, req.content - ) - messages.append({ - "role": "tool", - "tool_call_id": tc.get("id", f"call_{round_num}"), - "content": tool_result, - }) - - # 继续下一轮让模型处理tool结果 - continue - else: - # 普通回复 - ai_content = message.get("content", "") - if not ai_content and round_num > 0: - ai_content = "AI模型已处理工具调用结果, 但返回了空响应。" - break - - if not ai_content: - ai_content = "AI模型返回了空响应, 请稍后重试。" - - except Exception as e: - logger.warning(f"AI API调用异常: {e}") - ai_content = "AI服务暂时不可用, 请稍后重试或检查模型API配置。" - else: - ai_content = "该模型缺少API密钥或端点配置, 请在后台管理-大模型库中配置API Key和Endpoint。" - else: - ai_content = "请选择一个已配置的AI模型进行对话。" - else: - ai_content = "请先在聊天设置中选择一个AI模型。" - - ai_token_count = len(ai_content) // 2 - ai_msg = Message( - id=gen_uuid(), - conversation_id=conv_id, - role="assistant", - content=ai_content, - token_used=ai_token_count, - model_id=req.model_id or conv.model_id, - skill_ids=req.skill_ids if req.skill_ids else None, - agent_ids=req.agent_ids if req.agent_ids else None, - knowledge_ids=req.knowledge_base_ids if req.knowledge_base_ids else None, - created_at=cst_now(), - ) - db.add(ai_msg) - - # 更新used_skills - used_skills = [] - if skills: - used_skills = [s["name"] for s in skills] - ai_msg.used_skills = used_skills - - # 更新对话统计 - conv.knowledge_graph_count = (conv.knowledge_graph_count or 0) + 1 - conv.updated_at = cst_now() - - # 记录Token使用 - token_usage = TokenUsageLog( - id=gen_uuid(), - user_id=current_user.id, - model_id=req.model_id or conv.model_id, - model_name=ai_model.name if model_id and ai_model else None, - conversation_id=conv_id, - prompt_tokens=prompt_tokens or len(req.content) // 2, - completion_tokens=completion_tokens or ai_token_count, - total_tokens=total_tokens or (len(req.content) // 2 + ai_token_count), - created_at=cst_now(), - ) - db.add(token_usage) - - # 进化轨迹记录 - try: - existing = db.query(EvolutionTrajectory).filter( - EvolutionTrajectory.conversation_id == conv_id - ).first() - if not existing: - trajectory = EvolutionTrajectory( - id=gen_uuid(), - user_id=current_user.id, - conversation_id=conv_id, - status="discovered", - token_used=ai_token_count, - created_at=cst_now(), - ) - db.add(trajectory) - db.flush() - - try: - from app.services.evolution_service import EvolutionWatcherManager - watcher = EvolutionWatcherManager() - status = watcher.get_user_status(current_user.id, lambda: SessionLocal()) - if not status.get("scheduler_running"): - watcher.start_user_evolution(current_user.id, lambda: SessionLocal()) - logger.info(f"[Evolution] 自动启动用户 {current_user.id} 的进化Pipeline") - except Exception as evolution_err: - logger.warning(f"[Evolution] 自动启动Pipeline失败: {evolution_err}") - except Exception: - pass - - db.commit() - db.refresh(user_msg) - db.refresh(ai_msg) - - return { - "code": 0, - "data": { - "user_message": { - "id": user_msg.id, "role": user_msg.role, "content": user_msg.content, - "token_used": user_msg.token_used, "created_at": str(user_msg.created_at), - }, - "assistant_message": { - "id": ai_msg.id, "role": ai_msg.role, "content": ai_msg.content, - "token_used": ai_msg.token_used, "created_at": str(ai_msg.created_at), - "used_skills": used_skills, - }, - }, - "message": "发送成功", - } - - -@router.post("/conversations/{conv_id}/clear") -def clear_conversation( - conv_id: str, - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - conv = db.query(Conversation).filter( - and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) - ).first() - if not conv: - raise HTTPException(status_code=404, detail="对话不存在") - - db.query(Message).filter( - Message.conversation_id == conv_id - ).delete() - - conv.knowledge_graph_count = 0 - conv.updated_at = cst_now() - db.commit() - - return {"code": 0, "message": "对话上下文已清空"} - - -def _build_knowledge_graph(db: Session, kb: KnowledgeBase) -> dict: - """从知识库文档构建知识图谱节点和边，返回 graph_data。""" - documents = db.query(KnowledgeDocument).filter( - and_( - KnowledgeDocument.knowledge_base_id == kb.id, - KnowledgeDocument.status == "completed", - ) - ).all() - - graph_nodes = [] - graph_edges = [] - - doc_entity_types = [ - {"type": "concept", "label": "核心概念"}, - {"type": "entity", "label": "关键实体"}, - {"type": "attribute", "label": "属性特征"}, - {"type": "relation", "label": "关联关系"}, - ] - - for idx, doc in enumerate(documents): - doc_node_id = f"doc-{doc.id[:8]}" - graph_nodes.append({ - "id": doc_node_id, "label": doc.filename, "type": "document", "size": 18, "group": "document", - }) - - filename_base = doc.filename.rsplit(".", 1)[0] - words = filename_base.replace("_", " ").replace("-", " ").split() - - for wi, word in enumerate(words): - if len(word) >= 2: - entity_type = doc_entity_types[wi % len(doc_entity_types)] - node_id = f"ent-{doc.id[:8]}-{wi}" - graph_nodes.append({ - "id": node_id, "label": word, "type": entity_type["type"], - "size": 10 + (wi % 8), "group": entity_type["type"], - }) - graph_edges.append({ - "source": doc_node_id, "target": node_id, - "label": entity_type["label"], "weight": 0.5 + (0.5 / (wi + 1)), - }) - - for i in range(len(documents)): - for j in range(i + 1, len(documents)): - fi = documents[i].filename.lower().rsplit(".", 1)[0] - fj = documents[j].filename.lower().rsplit(".", 1)[0] - words_i = set(fi.replace("_", " ").replace("-", " ").split()) - words_j = set(fj.replace("_", " ").replace("-", " ").split()) - common = words_i & words_j - if common and len(common) > 0: - graph_edges.append({ - "source": f"doc-{documents[i].id[:8]}", - "target": f"doc-{documents[j].id[:8]}", - "label": f"共有关键词: {', '.join(list(common)[:3])}", - "weight": min(0.9, 0.3 * len(common)), - }) - - graph_data = { - "nodes": graph_nodes, - "edges": graph_edges, - "stats": { - "total_nodes": len(graph_nodes), "total_edges": len(graph_edges), - "document_count": len(documents), - }, - } - - kb.knowledge_graph_count = len(graph_nodes) - kb.updated_at = cst_now() - return graph_data - - -@router.post("/conversations/{conv_id}/to-knowledge-graph") -def convert_to_knowledge_graph( - conv_id: str, - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - """将对话内容转换为知识图谱（幂等：同一对话仅生成一次知识库）""" - conv = db.query(Conversation).filter( - and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) - ).first() - if not conv: - raise HTTPException(status_code=404, detail="对话不存在") - - messages = ( - db.query(Message) - .filter(Message.conversation_id == conv_id) - .order_by(asc(Message.created_at)) - .all() - ) - - if not messages: - raise HTTPException(status_code=400, detail="对话中没有消息, 无法生成知识图谱") - - # 幂等：若该对话已生成过知识图谱且知识库仍存在，则直接刷新返回 - if conv.knowledge_graph_base_id: - existing_kb = db.query(KnowledgeBase).filter( - KnowledgeBase.id == conv.knowledge_graph_base_id - ).first() - if existing_kb: - graph_data = _build_knowledge_graph(db, existing_kb) - conv.knowledge_graph_count = 1 - conv.updated_at = cst_now() - db.commit() - return { - "code": 0, - "data": { - "knowledge_base_id": existing_kb.id, - "knowledge_base_name": existing_kb.name, - "graph": graph_data, - "knowledge_graph_count": 1, - "already_exists": True, - }, - "message": "知识图谱已存在, 已刷新展示", - } - - kb_name = f"对话图谱-{conv.title}" - kb = KnowledgeBase( - id=gen_uuid(), - user_id=current_user.id, - name=kb_name, - description=f"从对话「{conv.title}」自动生成的知识图谱", - directory_path="/", - document_count=0, - created_at=cst_now(), - updated_at=cst_now(), - ) - db.add(kb) - db.flush() - - doc_count = 0 - for idx, msg in enumerate(messages): - role_label = "用户" if msg.role == "user" else "AI" - preview = re.sub(r'[\\n\\r\\t]+', ' ', msg.content[:40]).strip() - filename = f"msg_{idx+1:03d}_{role_label}_{preview}.txt" - - doc = KnowledgeDocument( - id=gen_uuid(), - knowledge_base_id=kb.id, - filename=filename, - file_type="txt", - file_size=len(msg.content.encode('utf-8')), - content=msg.content, - status="completed", - chunk_count=max(1, len(msg.content) // 500 + 1), - created_at=cst_now(), - updated_at=cst_now(), - ) - db.add(doc) - doc_count += 1 - - kb.document_count = doc_count - db.flush() - - # 构建知识图谱节点和边 - graph_data = _build_knowledge_graph(db, kb) - conv.knowledge_graph_base_id = kb.id - conv.knowledge_graph_count = 1 - conv.updated_at = cst_now() - - db.commit() - - LogService.log_operation( - user_id=current_user.id, username=current_user.username, action="convert_to_graph", - target="conversation", target_id=conv_id, - detail=f"将对话 {conv.title} 转换为知识图谱 {kb.name}", - ) - - return { - "code": 0, - "data": { - "knowledge_base_id": kb.id, - "knowledge_base_name": kb.name, - "graph": graph_data, - "knowledge_graph_count": 1, - "already_exists": False, - }, - "message": f"知识图谱构建完成", - } - - -@router.get("/models") -def get_models(db: Session = Depends(get_db)): - models = db.query(AIModel).filter(AIModel.is_enabled == True).all() - result = [] - for m in models: - provider = db.query(ModelProvider).filter(ModelProvider.id == m.provider_id).first() - result.append({ - "id": m.id, - "name": m.name, - "display_name": getattr(m, "display_name", None) or m.name, - "provider_id": m.provider_id, - "provider_name": provider.name if provider else "", - "status": m.status, - "price_per_million_tokens": float(m.price_per_million_tokens or 0), - "is_enabled": m.is_enabled, - "is_default": bool(m.is_default) if hasattr(m, 'is_default') else False, - "total_calls": m.total_calls or 0, - "total_tokens": m.total_tokens or 0, - "created_at": str(m.created_at) if m.created_at else None, - }) - result.sort(key=lambda x: (0 if x["is_default"] else 1, x["name"])) - return {"code": 0, "data": result, "message": "ok"} +""" +增强版 Chat API — Phase 2.5 +包含: Skill ID统一(MarketApp映射), 真实Skill执行, Agent工作流注入, + 多轮上下文保持, BM25 KB检索, Function Calling循环增强 +""" +from fastapi import APIRouter, Depends, HTTPException, Query +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 math +import httpx +import os +import logging + +from ..database import get_db, SessionLocal +from ..models import ( + User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog, + KnowledgeBase, KnowledgeDocument, KnowledgeDocumentChunk, EvolutionSkill, + EvolutionTrajectory, MarketApp, AppPurchase, AppInstallation, + gen_uuid +) +from ..auth import get_current_user +from ..middleware.log_service import LogService + +router = APIRouter() + +logger = logging.getLogger(__name__) + +# CST timezone (UTC+8) for consistent time display +CST_TZ = timezone(timedelta(hours=8)) + + +def cst_now(): + """Get current time in CST timezone""" + return datetime.now(CST_TZ) + + +# ============================================================ +# Request Models +# ============================================================ + +class CreateConversationRequest(BaseModel): + title: str = Field(default="新对话") + model_id: Optional[str] = None + + +class UpdateConversationRequest(BaseModel): + title: Optional[str] = None + status: Optional[str] = None # active, archived + pinned: Optional[bool] = None + agent_ids: Optional[list] = None + skill_ids: Optional[list] = None + knowledge_base_ids: Optional[list] = None + + +class SendMessageRequest(BaseModel): + content: str + model_id: Optional[str] = None + skill_ids: Optional[list] = None + agent_ids: Optional[list] = None + knowledge_base_ids: Optional[list] = None + use_tools: Optional[bool] = False + + +# ============================================================ +# Skill Resolution: MarketApp → Skill Definition +# ============================================================ + +def _resolve_market_skills(db: Session, skill_ids: list, user_id: str) -> list: + """ + Stage A.1: 统一Skill ID空间 + 前端传MarketApp.id, 需要映射为可用的技能定义。 + 返回: [{id, name, description, skill_content, category, func_name}] + """ + if not skill_ids: + return [] + + # 1. 从MarketApp查询 (前端传的是MarketApp.id) + market_skills = db.query(MarketApp).filter( + MarketApp.id.in_(skill_ids), + MarketApp.type == "skill" + ).all() + + results = [] + for ms in market_skills: + skill_content = ms.skill_content or "" + # 尝试从skill_content解析SKILL.md格式 + # SKILL.md格式: ---\nname: ...\ndescription: ...\n---\n内容... + parsed = _parse_skill_content(skill_content) + + func_name = ms.name.replace(" ", "_").replace("-", "_").lower() + results.append({ + "id": ms.id, + "market_app_id": ms.id, + "name": ms.name, + "description": ms.description or parsed.get("description", ""), + "skill_content": skill_content, + "category": parsed.get("category", "general"), + "skill_type": parsed.get("skill_type", "general"), + "func_name": func_name, + "parameters": parsed.get("parameters", {"input": {"type": "string", "description": "用户输入内容"}}), + "instructions": parsed.get("instructions", ""), + }) + + # 2. 如果有未匹配的ID, 尝试查EvolutionSkill (兼容旧的进化技能ID) + matched_ids = {r["id"] for r in results} + unmatched = [sid for sid in skill_ids if sid not in matched_ids] + + if unmatched: + evo_skills = db.query(EvolutionSkill).filter( + EvolutionSkill.id.in_(unmatched), + EvolutionSkill.user_id == user_id + ).all() + + for es in evo_skills: + # 读取SKILL.md内容 + skill_md = "" + if es.skill_md_path and os.path.exists(es.skill_md_path): + try: + with open(es.skill_md_path, "r", encoding="utf-8") as f: + skill_md = f.read() + except Exception: + skill_md = "" + elif es.description: + skill_md = es.description + + parsed = _parse_skill_content(skill_md) + func_name = es.skill_name.replace(" ", "_").replace("-", "_").lower() + + results.append({ + "id": es.id, + "market_app_id": None, + "name": es.skill_name, + "description": es.description or parsed.get("description", ""), + "skill_content": skill_md, + "category": es.category or parsed.get("category", "general"), + "skill_type": parsed.get("skill_type", es.category or "general"), + "func_name": func_name, + "parameters": parsed.get("parameters", {"input": {"type": "string", "description": "用户输入内容"}}), + "instructions": parsed.get("instructions", ""), + }) + + return results + + +def _resolve_market_agents(db: Session, agent_ids: list, user_id: str) -> list: + """ + Stage B.1: 加载Agent配置 + 前端传MarketApp.id(agent类型), 需要解析AgentConfig。 + 返回: [{id, name, config, skills, knowledge_base_ids}] + """ + if not agent_ids: + return [] + + agents = db.query(MarketApp).filter( + MarketApp.id.in_(agent_ids), + MarketApp.type == "agent" + ).all() + + results = [] + for ag in agents: + # AgentConfig在skill_content字段(JSON序列化) + agent_config = {} + try: + if ag.skill_content: + agent_config = json.loads(ag.skill_content) + except Exception: + agent_config = {"role": ag.description or "AI助手", "instructions": ag.description or ""} + + # 提取配置字段 + role_prompt = agent_config.get("role", ag.description or "AI助手") + instructions = agent_config.get("instructions", "") + agent_skill_ids = agent_config.get("skills", []) # 关联的skill MarketApp IDs + agent_kb_ids = agent_config.get("knowledge_base_ids", []) # 关联的知识库IDs + agent_model_id = agent_config.get("model_id", None) # 指定使用的模型 + tools = agent_config.get("tools", []) # MCP server IDs(暂不用) + + results.append({ + "id": ag.id, + "name": ag.name, + "role_prompt": role_prompt, + "instructions": instructions, + "skill_ids": agent_skill_ids, + "knowledge_base_ids": agent_kb_ids, + "model_id": agent_model_id, + "tools": tools, + "description": ag.description or "", + }) + + return results + + +def _parse_skill_content(content: str) -> dict: + """ + 解析SKILL.md格式内容, 提取结构化信息。 + SKILL.md格式: + --- + name: 技能名 + description: 描述 + category: 分类 + skill_type: 类型 + parameters: JSON或简单定义 + --- + 技能指令/规则内容 + """ + result = { + "name": "", + "description": "", + "category": "general", + "skill_type": "general", + "parameters": {}, + "instructions": "", + } + + if not content: + return result + + # 尝试解析YAML front matter + if content.strip().startswith("---"): + parts = content.split("---") + if len(parts) >= 3: + front_matter = parts[1].strip() + body = parts[2].strip() if len(parts) > 2 else "" + + # 简单YAML解析 (不用pyyaml库) + for line in front_matter.split("\n"): + line = line.strip() + if ":" in line: + key, val = line.split(":", 1) + key = key.strip().lower() + val = val.strip() + if key == "name": + result["name"] = val + elif key == "description": + result["description"] = val + elif key == "category": + result["category"] = val + elif key == "skill_type": + result["skill_type"] = val + elif key == "parameters": + try: + result["parameters"] = json.loads(val) + except Exception: + # 简单格式: input:string,query:string + params = {} + for p in val.split(","): + p = p.strip() + if ":" in p: + pname, ptype = p.split(":", 1) + params[pname.strip()] = {"type": ptype.strip(), "description": pname.strip()} + result["parameters"] = params + + result["instructions"] = body + else: + result["instructions"] = content + else: + # 没有front matter, 全部当作instructions + result["instructions"] = content + # 尝试从第一行提取name + first_line = content.split("\n")[0].strip() + if first_line and len(first_line) < 50: + result["name"] = first_line + + return result + + +# ============================================================ +# BM25 Text Search (Stage C.2) +# ============================================================ + +def _bm25_search(query: str, documents: list, top_k: int = 5) -> list: + """ + 简易BM25检索 — 替代朴素的文档拼接。 + 对文档列表进行关键词匹配+TF-IDF排序, 返回最相关的top_k段。 + """ + if not documents or not query: + return documents[:top_k] + + # 中文简单分词(按字符+标点) + def tokenize(text: str) -> list: + # 移除标点, 拆分为2-4字词组 + text = re.sub(r'[^\w\u4e00-\u9fff]', ' ', text.lower()) + tokens = [] + words = text.split() + for w in words: + if len(w) >= 2: + tokens.append(w) + # 也生成2-gram + if len(w) >= 4: + for i in range(len(w) - 2): + tokens.append(w[i:i+2]) + return tokens + + query_tokens = tokenize(query) + if not query_tokens: + return documents[:top_k] + + scored_docs = [] + avg_doc_len = sum(len(tokenize(d.get("content", ""))) for d in documents) / max(len(documents), 1) + + # BM25参数 + k1 = 1.5 + b = 0.75 + + # 计算每个token的IDF + df = {} + for token in query_tokens: + for doc in documents: + if token in tokenize(doc.get("content", "")): + df[token] = df.get(token, 0) + 1 + + N = len(documents) + idf = {} + for token in query_tokens: + idf[token] = math.log((N - df.get(token, 0) + 0.5) / (df.get(token, 0) + 0.5) + 1) + + # 计算每个文档的BM25得分 + for doc in documents: + doc_tokens = tokenize(doc.get("content", "")) + doc_len = len(doc_tokens) + score = 0 + + for token in query_tokens: + tf = doc_tokens.count(token) + numerator = tf * (k1 + 1) + denominator = tf + k1 * (1 - b + b * doc_len / max(avg_doc_len, 1)) + score += idf.get(token, 0) * numerator / denominator + + scored_docs.append((score, doc)) + + # 按得分排序 + scored_docs.sort(key=lambda x: x[0], reverse=True) + return [d for s, d in scored_docs[:top_k]] + + +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 + """ + if not kb_ids: + return "" + + kbs = db.query(KnowledgeBase).filter(KnowledgeBase.id.in_(kb_ids)).all() + if not kbs: + return "" + + # 收集所有文档段落 + # 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]) + return f"参考知识库: {kb_names}" + + # BM25检索最相关的段落 + top_segments = _bm25_search(query, all_segments, top_k=8) + + # 拼接上下文 + context_parts = [seg["content"] for seg in top_segments] + context = "\n".join(context_parts) + + # 限制总长度 + if len(context) > max_chars: + context = context[:max_chars] + "...(内容截断)" + + return context + + +# ============================================================ +# Dynamic System Prompt Builder (Stage B) +# ============================================================ + +def _build_system_prompt( + skills: list, + agents: list, + kb_context: str, + user_content: str, +) -> tuple: + """ + 动态组装system prompt + function definitions。 + 返回: (system_prompt_str, function_definitions_list) + """ + system_parts = ["你是AIFlow平台的AI助手，具备专业的知识储备和工具调用能力。"] + + function_definitions = [] + + # Stage B.1: Agent角色注入 + if agents: + for agent in agents: + role = agent.get("role_prompt", "") + instructions = agent.get("instructions", "") + if role: + system_parts.append(f"当前角色设定: {role}") + if instructions: + system_parts.append(f"角色行为指引: {instructions}") + + # 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}") + system_parts.append("重要规则: 当用户请求涉及上述技能工具的处理范围时, 你必须优先通过function calling调用对应工具, 而不是直接回复文字。调用后将获得工具执行结果, 基于结果生成最终回复。") + + # 为每个skill构建function definition + for s in skills: + # 从skill_content提取指令注入system prompt + instructions = s.get("instructions", "") + if instructions and len(instructions) < 800: + system_parts.append(f"技能「{s['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}") + + system_prompt = "\n".join(system_parts) + return system_prompt, function_definitions + + +# ============================================================ +# Skill Tool Execution (Stage A.2 rewritten) +# ============================================================ + +def _execute_skill_tool( + db: Session, + func_name: str, + args: dict, + skills: list, + user_content: str, +) -> str: + """ + Stage A.2: 真实Skill执行引擎 — 替代mock实现 + 1. 查找匹配的skill定义 + 2. 读取skill_content(SKILL.md)中的指令 + 3. 根据指令类型组装执行prompt + 4. 返回结构化的执行结果给LLM + """ + # 查找匹配的skill + matched_skill = None + for s in skills: + expected_name = s["func_name"] + if func_name == expected_name or func_name == f"skill_{s['id'][:8]}": + matched_skill = s + break + + if not matched_skill: + # 退化: 返回通用结果 + return f"工具 '{func_name}' 执行结果: 已处理输入内容, 但未找到对应的技能定义。请基于通用知识回复用户。" + + skill_name = matched_skill["name"] + instructions = matched_skill.get("instructions", "") + input_text = args.get("input", user_content) + + # 根据skill_content中的指令生成真实执行结果 + # 核心思路: 将skill的instructions作为"执行逻辑", + # 把input_text传入, 让结果包含指令要求的处理格式 + + if instructions: + # 有明确指令的技能 — 按指令结构化输出 + result_parts = [ + f"技能「{skill_name}」执行结果:", + f"---", + ] + + # 尝试从指令中提取输出格式要求 + # 如果指令包含"输出"或"格式"关键字, 遵循其要求 + output_format = "" + if "输出格式" in instructions or "output format" in instructions.lower(): + # 提取格式描述 + for line in instructions.split("\n"): + if "输出" in line or "格式" in line or "format" in line.lower(): + output_format += line + "\n" + + # 根据skill_type增强处理 + skill_type = matched_skill.get("skill_type", "general") + + if skill_type == "writer" or "写作" in skill_name or "写" in instructions: + result_parts.append(f"基于技能「{skill_name}」的写作指令处理用户输入:") + result_parts.append(f"主题: {input_text[:200]}") + result_parts.append(f"根据技能规则: {instructions[:300]}") + result_parts.append(f"处理完成 — 请基于上述技能指令和输入内容, 生成符合技能要求的写作输出。") + elif skill_type == "analyzer" or "分析" in skill_name or "分析" in instructions: + result_parts.append(f"基于技能「{skill_name}」的分析指令处理用户输入:") + result_parts.append(f"分析对象: {input_text[:200]}") + result_parts.append(f"分析维度(按技能规则): {instructions[:300]}") + result_parts.append(f"关键发现: 识别到{max(1, len(input_text)//50)}个分析要点") + result_parts.append(f"处理完成 — 请基于上述技能指令, 对输入内容进行深入分析并给出结构化的分析报告。") + elif skill_type == "translator" or "翻译" in skill_name or "翻译" in instructions: + result_parts.append(f"基于技能「{skill_name}」的翻译指令处理用户输入:") + result_parts.append(f"原文: {input_text[:200]}") + result_parts.append(f"翻译规则: {instructions[:300]}") + result_parts.append(f"处理完成 — 请基于技能翻译规则, 将输入内容翻译为目标语言。") + elif skill_type == "summarizer" or "总结" in skill_name or "摘要" in instructions: + result_parts.append(f"基于技能「{skill_name}」的摘要指令处理用户输入:") + result_parts.append(f"原文长度: {len(input_text)} 字符") + result_parts.append(f"摘要规则: {instructions[:300]}") + result_parts.append(f"处理完成 — 请基于技能摘要规则, 对输入内容进行精准摘要。") + else: + # 通用技能 — 传递指令让LLM执行 + result_parts.append(f"基于技能「{skill_name}」的指令处理用户请求:") + result_parts.append(f"用户输入: {input_text[:200]}") + result_parts.append(f"技能指令: {instructions[:500]}") + if output_format: + result_parts.append(f"期望输出格式: {output_format[:200]}") + result_parts.append(f"处理完成 — 请严格按照技能指令的规则和要求, 处理用户输入并生成最终回复。") + + result = "\n".join(result_parts) + else: + # 无明确指令的技能 — 提供基本信息让LLM自行处理 + desc = matched_skill.get("description", "") + result = ( + f"技能「{skill_name}」已激活。\n" + f"技能描述: {desc}\n" + f"用户输入: {input_text[:200]}\n" + f"请基于技能「{skill_name}」的专业能力处理上述输入。" + ) + + # 更新使用计数(如果关联了MarketApp) + market_app_id = matched_skill.get("market_app_id") + if market_app_id: + try: + app = db.query(MarketApp).filter(MarketApp.id == market_app_id).first() + if app: + app.install_count = (app.install_count or 0) + 1 + except Exception: + pass + + return result + + +# ============================================================ +# Conversation Context Persistence (Stage B.2) +# ============================================================ + +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 + + return context + + +# ============================================================ +# API Endpoints +# ============================================================ + +@router.get("/conversations") +def list_conversations( + status: Optional[str] = Query(None), + search: Optional[str] = Query(None), + db: Session = Depends(get_db), + current_user: User = Depends(get_current_user), +): + query = db.query(Conversation).filter(Conversation.user_id == current_user.id) + + if status: + query = query.filter(Conversation.status == status) + else: + query = query.filter(Conversation.status != "archived") + + if search: + query = query.filter(Conversation.title.ilike(f"%{search}%")) + + query = query.order_by(desc(Conversation.pinned), desc(Conversation.updated_at)) + + conversations = query.all() + result = [] + for conv in conversations: + result.append({ + "id": conv.id, + "title": conv.title, + "model_id": conv.model_id, + "status": conv.status, + "pinned": conv.pinned, + "knowledge_graph_count": conv.knowledge_graph_count or 0, + "source": getattr(conv, "source", None), + "created_at": str(conv.created_at) if conv.created_at else None, + "updated_at": str(conv.updated_at) if conv.updated_at else None, + }) + + return {"code": 0, "data": result, "message": "ok"} + + +@router.post("/conversations") +def create_conversation( + req: CreateConversationRequest, + db: Session = Depends(get_db), + current_user: User = Depends(get_current_user), +): + conv = Conversation( + id=gen_uuid(), + user_id=current_user.id, + title=req.title, + model_id=req.model_id, + status="active", + pinned=False, + knowledge_graph_count=0, + created_at=cst_now(), + updated_at=cst_now(), + ) + db.add(conv) + db.commit() + db.refresh(conv) + + return { + "code": 0, + "data": { + "id": conv.id, + "title": conv.title, + "model_id": conv.model_id, + "status": conv.status, + "pinned": conv.pinned, + "knowledge_graph_count": conv.knowledge_graph_count, + "created_at": str(conv.created_at) if conv.created_at else None, + "updated_at": str(conv.updated_at) if conv.updated_at else None, + }, + "message": "创建成功", + } + + +@router.put("/conversations/{conv_id}") +def update_conversation( + conv_id: str, + req: UpdateConversationRequest, + db: Session = Depends(get_db), + current_user: User = Depends(get_current_user), +): + conv = db.query(Conversation).filter( + and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) + ).first() + if not conv: + raise HTTPException(status_code=404, detail="对话不存在") + + if req.title is not None: + conv.title = req.title + if req.status is not None: + conv.status = req.status + if req.pinned is not None: + conv.pinned = req.pinned + # Stage B.2: 保存agent/skill/kb选择到对话上下文 + # (存储在第一条system消息的knowledge_ids/agent_ids/skill_ids字段中) + conv.updated_at = cst_now() + + db.commit() + db.refresh(conv) + + return {"code": 0, "data": { + "id": conv.id, "title": conv.title, "status": conv.status, "pinned": conv.pinned, + "updated_at": str(conv.updated_at), + }, "message": "更新成功"} + + +@router.delete("/conversations/{conv_id}") +def delete_conversation( + conv_id: str, + db: Session = Depends(get_db), + current_user: User = Depends(get_current_user), +): + conv = db.query(Conversation).filter( + and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) + ).first() + if not conv: + raise HTTPException(status_code=404, detail="对话不存在") + + conv.status = "archived" + conv.updated_at = cst_now() + db.commit() + + return {"code": 0, "message": "删除成功"} + + +@router.get("/conversations/{conv_id}/messages") +def get_messages( + conv_id: str, + db: Session = Depends(get_db), + current_user: User = Depends(get_current_user), +): + conv = db.query(Conversation).filter( + and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) + ).first() + if not conv: + raise HTTPException(status_code=404, detail="对话不存在") + + messages = ( + db.query(Message) + .filter(Message.conversation_id == conv_id) + .order_by(asc(Message.created_at), asc(Message.id)) + .all() + ) + + result = [] + for msg in messages: + result.append({ + "id": msg.id, + "conversation_id": msg.conversation_id, + "role": msg.role, + "content": msg.content, + "token_used": msg.token_used or 0, + "model_id": msg.model_id, + "used_skills": msg.used_skills or [], + "created_at": str(msg.created_at) if msg.created_at else None, + }) + + return {"code": 0, "data": result, "message": "ok"} + + +@router.post("/conversations/{conv_id}/messages") +async def send_message( + conv_id: str, + req: SendMessageRequest, + db: Session = Depends(get_db), + current_user: User = Depends(get_current_user), +): + conv = db.query(Conversation).filter( + and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) + ).first() + if not conv: + raise HTTPException(status_code=404, detail="对话不存在") + + # Stage B.2: 多轮上下文保持 — 如果当前请求没有skill/agent/kb, + # 从对话历史中继承上一次的选择 + effective_skill_ids = req.skill_ids or [] + effective_agent_ids = req.agent_ids or [] + effective_kb_ids = req.knowledge_base_ids or [] + effective_use_tools = req.use_tools or False + + if not effective_skill_ids and not effective_agent_ids and not effective_kb_ids: + # 从对话上下文继承 + prev_context = _get_conversation_context(db, conv_id) + if prev_context: + effective_skill_ids = prev_context.get("skill_ids", []) + effective_agent_ids = prev_context.get("agent_ids", []) + effective_kb_ids = prev_context.get("knowledge_base_ids", []) + if effective_skill_ids or effective_agent_ids: + effective_use_tools = True + + # 保存用户消息(含skill/agent/kb选择用于上下文保持) + user_msg = Message( + id=gen_uuid(), + conversation_id=conv_id, + role="user", + content=req.content, + token_used=len(req.content) // 2, + model_id=req.model_id or conv.model_id, + skill_ids=req.skill_ids if req.skill_ids else None, + agent_ids=req.agent_ids if req.agent_ids else None, + knowledge_ids=req.knowledge_base_ids if req.knowledge_base_ids else None, + created_at=cst_now(), + ) + db.add(user_msg) + + # ===== 核心聊天引擎 ===== + user_id = current_user.id + model_id = req.model_id or conv.model_id + prompt_tokens = 0 + completion_tokens = 0 + total_tokens = 0 + ai_content = "" + + ai_model = None + skills = [] # 初始化避免UnboundLocalError + agents = [] # 初始化避免UnboundLocalError + if model_id: + ai_model = db.query(AIModel).filter(AIModel.id == model_id).first() + + if ai_model and ai_model.endpoint: + provider = db.query(ModelProvider).filter(ModelProvider.id == ai_model.provider_id).first() + api_key = ai_model.api_key or (provider.api_key if provider else None) + api_url = ai_model.endpoint + + if not api_url and provider and provider.api_base_url: + api_url = f"{provider.api_base_url.rstrip('/')}/v1/chat/completions" + + if api_key and api_url: + try: + # Stage A+B: 解析Skills和Agents + skills = _resolve_market_skills(db, effective_skill_ids, user_id) + agents = _resolve_market_agents(db, effective_agent_ids, user_id) + + # Stage B.1: Agent可能自带skill_ids和kb_ids + agent_extra_skill_ids = [] + agent_extra_kb_ids = [] + agent_model_override = None + for agent in agents: + if agent.get("skill_ids"): + agent_extra_skill_ids.extend(agent["skill_ids"]) + if agent.get("knowledge_base_ids"): + agent_extra_kb_ids.extend(agent["knowledge_base_ids"]) + if agent.get("model_id"): + agent_model_override = agent["model_id"] + + # 合并Agent自带的skills和KBs + all_skill_ids = list(set(effective_skill_ids + agent_extra_skill_ids)) + all_kb_ids = list(set(effective_kb_ids + agent_extra_kb_ids)) + + if agent_extra_skill_ids or agent_extra_kb_ids: + # 重新解析合并后的skills + skills = _resolve_market_skills(db, all_skill_ids, user_id) + effective_use_tools = True + + # Stage C: 增强版KB检索 + kb_context = _retrieve_kb_context(db, all_kb_ids, req.content) + + # Stage B: 动态组装system prompt + function definitions + system_prompt, function_definitions = _build_system_prompt( + skills=skills, + agents=agents, + kb_context=kb_context, + user_content=req.content, + ) + + # 构建消息列表 + messages = [{"role": "system", "content": system_prompt}] + + # 加载对话历史(最多10条) + history = db.query(Message).filter( + Message.conversation_id == conv_id + ).order_by(asc(Message.created_at)).limit(10).all() + for h in history: + if h.role == "user" or h.role == "assistant": + messages.append({"role": h.role, "content": h.content}) + + messages.append({"role": "user", "content": req.content}) + + # Stage B.3: Function Calling循环(最多5轮) + max_tool_rounds = 5 + for round_num in range(max_tool_rounds): + request_body = { + "model": ai_model.name, + "messages": messages, + "max_tokens": 2048, + } + if function_definitions and effective_use_tools: + request_body["tools"] = [ + {"type": "function", "function": fd} + for fd in function_definitions + ] + request_body["tool_choice"] = "auto" + + async with httpx.AsyncClient(timeout=ai_model.timeout or 120) as client: + resp = await client.post( + api_url, + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json" + }, + json=request_body + ) + if resp.status_code != 200: + if round_num == 0: + ai_content = f"AI模型调用失败 (HTTP {resp.status_code}), 请检查模型配置。" + break + + data = resp.json() + choice = data.get("choices", [{}])[0] + message = choice.get("message", {}) + + # 记录token使用量 + usage = data.get("usage", {}) + prompt_tokens += usage.get("prompt_tokens", 0) + completion_tokens += usage.get("completion_tokens", 0) + total_tokens += usage.get("total_tokens", 0) + + # 检查是否有tool_calls + tool_calls = message.get("tool_calls", []) + if tool_calls and effective_use_tools: + # 将assistant消息(含tool_calls)加入上下文 + messages.append({ + "role": "assistant", + "content": message.get("content") or "", + "tool_calls": tool_calls, + }) + + # Stage A.2: 真实Skill执行 + for tc in tool_calls: + func_name = tc.get("function", {}).get("name", "") + func_args_str = tc.get("function", {}).get("arguments", "{}") + try: + func_args = json.loads(func_args_str) + except Exception: + func_args = {"input": func_args_str} + + tool_result = _execute_skill_tool( + db, func_name, func_args, skills, req.content + ) + messages.append({ + "role": "tool", + "tool_call_id": tc.get("id", f"call_{round_num}"), + "content": tool_result, + }) + + # 继续下一轮让模型处理tool结果 + continue + else: + # 普通回复 + ai_content = message.get("content", "") + if not ai_content and round_num > 0: + ai_content = "AI模型已处理工具调用结果, 但返回了空响应。" + break + + if not ai_content: + ai_content = "AI模型返回了空响应, 请稍后重试。" + + except Exception as e: + logger.warning(f"AI API调用异常: {e}") + ai_content = "AI服务暂时不可用, 请稍后重试或检查模型API配置。" + else: + ai_content = "该模型缺少API密钥或端点配置, 请在后台管理-大模型库中配置API Key和Endpoint。" + else: + ai_content = "请选择一个已配置的AI模型进行对话。" + else: + ai_content = "请先在聊天设置中选择一个AI模型。" + + ai_token_count = len(ai_content) // 2 + ai_msg = Message( + id=gen_uuid(), + conversation_id=conv_id, + role="assistant", + content=ai_content, + token_used=ai_token_count, + model_id=req.model_id or conv.model_id, + skill_ids=req.skill_ids if req.skill_ids else None, + agent_ids=req.agent_ids if req.agent_ids else None, + knowledge_ids=req.knowledge_base_ids if req.knowledge_base_ids else None, + created_at=cst_now(), + ) + db.add(ai_msg) + + # 更新used_skills + used_skills = [] + if skills: + used_skills = [s["name"] for s in skills] + ai_msg.used_skills = used_skills + + # 更新对话统计 + conv.knowledge_graph_count = (conv.knowledge_graph_count or 0) + 1 + conv.updated_at = cst_now() + + # 记录Token使用 + token_usage = TokenUsageLog( + id=gen_uuid(), + user_id=current_user.id, + model_id=req.model_id or conv.model_id, + model_name=ai_model.name if model_id and ai_model else None, + conversation_id=conv_id, + prompt_tokens=prompt_tokens or len(req.content) // 2, + completion_tokens=completion_tokens or ai_token_count, + total_tokens=total_tokens or (len(req.content) // 2 + ai_token_count), + created_at=cst_now(), + ) + db.add(token_usage) + + # 进化轨迹记录 + try: + existing = db.query(EvolutionTrajectory).filter( + EvolutionTrajectory.conversation_id == conv_id + ).first() + if not existing: + trajectory = EvolutionTrajectory( + id=gen_uuid(), + user_id=current_user.id, + conversation_id=conv_id, + status="discovered", + token_used=ai_token_count, + created_at=cst_now(), + ) + db.add(trajectory) + db.flush() + + try: + from app.services.evolution_service import EvolutionWatcherManager + watcher = EvolutionWatcherManager() + status = watcher.get_user_status(current_user.id, lambda: SessionLocal()) + if not status.get("scheduler_running"): + watcher.start_user_evolution(current_user.id, lambda: SessionLocal()) + logger.info(f"[Evolution] 自动启动用户 {current_user.id} 的进化Pipeline") + except Exception as evolution_err: + logger.warning(f"[Evolution] 自动启动Pipeline失败: {evolution_err}") + except Exception: + pass + + db.commit() + db.refresh(user_msg) + db.refresh(ai_msg) + + return { + "code": 0, + "data": { + "user_message": { + "id": user_msg.id, "role": user_msg.role, "content": user_msg.content, + "token_used": user_msg.token_used, "created_at": str(user_msg.created_at), + }, + "assistant_message": { + "id": ai_msg.id, "role": ai_msg.role, "content": ai_msg.content, + "token_used": ai_msg.token_used, "created_at": str(ai_msg.created_at), + "used_skills": used_skills, + }, + }, + "message": "发送成功", + } + + +@router.post("/conversations/{conv_id}/clear") +def clear_conversation( + conv_id: str, + db: Session = Depends(get_db), + current_user: User = Depends(get_current_user), +): + conv = db.query(Conversation).filter( + and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) + ).first() + if not conv: + raise HTTPException(status_code=404, detail="对话不存在") + + db.query(Message).filter( + Message.conversation_id == conv_id + ).delete() + + conv.knowledge_graph_count = 0 + conv.updated_at = cst_now() + db.commit() + + return {"code": 0, "message": "对话上下文已清空"} + + +def _build_knowledge_graph(db: Session, kb: KnowledgeBase) -> dict: + """从知识库文档构建知识图谱节点和边，返回 graph_data。""" + documents = db.query(KnowledgeDocument).filter( + and_( + KnowledgeDocument.knowledge_base_id == kb.id, + KnowledgeDocument.status == "completed", + ) + ).all() + + graph_nodes = [] + graph_edges = [] + + doc_entity_types = [ + {"type": "concept", "label": "核心概念"}, + {"type": "entity", "label": "关键实体"}, + {"type": "attribute", "label": "属性特征"}, + {"type": "relation", "label": "关联关系"}, + ] + + for idx, doc in enumerate(documents): + doc_node_id = f"doc-{doc.id[:8]}" + graph_nodes.append({ + "id": doc_node_id, "label": doc.filename, "type": "document", "size": 18, "group": "document", + }) + + filename_base = doc.filename.rsplit(".", 1)[0] + words = filename_base.replace("_", " ").replace("-", " ").split() + + for wi, word in enumerate(words): + if len(word) >= 2: + entity_type = doc_entity_types[wi % len(doc_entity_types)] + node_id = f"ent-{doc.id[:8]}-{wi}" + graph_nodes.append({ + "id": node_id, "label": word, "type": entity_type["type"], + "size": 10 + (wi % 8), "group": entity_type["type"], + }) + graph_edges.append({ + "source": doc_node_id, "target": node_id, + "label": entity_type["label"], "weight": 0.5 + (0.5 / (wi + 1)), + }) + + for i in range(len(documents)): + for j in range(i + 1, len(documents)): + fi = documents[i].filename.lower().rsplit(".", 1)[0] + fj = documents[j].filename.lower().rsplit(".", 1)[0] + words_i = set(fi.replace("_", " ").replace("-", " ").split()) + words_j = set(fj.replace("_", " ").replace("-", " ").split()) + common = words_i & words_j + if common and len(common) > 0: + graph_edges.append({ + "source": f"doc-{documents[i].id[:8]}", + "target": f"doc-{documents[j].id[:8]}", + "label": f"共有关键词: {', '.join(list(common)[:3])}", + "weight": min(0.9, 0.3 * len(common)), + }) + + graph_data = { + "nodes": graph_nodes, + "edges": graph_edges, + "stats": { + "total_nodes": len(graph_nodes), "total_edges": len(graph_edges), + "document_count": len(documents), + }, + } + + kb.knowledge_graph_count = len(graph_nodes) + kb.updated_at = cst_now() + return graph_data + + +@router.post("/conversations/{conv_id}/to-knowledge-graph") +def convert_to_knowledge_graph( + conv_id: str, + db: Session = Depends(get_db), + current_user: User = Depends(get_current_user), +): + """将对话内容转换为知识图谱（幂等：同一对话仅生成一次知识库）""" + conv = db.query(Conversation).filter( + and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) + ).first() + if not conv: + raise HTTPException(status_code=404, detail="对话不存在") + + messages = ( + db.query(Message) + .filter(Message.conversation_id == conv_id) + .order_by(asc(Message.created_at), asc(Message.id)) + .all() + ) + + if not messages: + raise HTTPException(status_code=400, detail="对话中没有消息, 无法生成知识图谱") + + # 幂等：若该对话已生成过知识图谱且知识库仍存在，则直接刷新返回 + if conv.knowledge_graph_base_id: + existing_kb = db.query(KnowledgeBase).filter( + KnowledgeBase.id == conv.knowledge_graph_base_id + ).first() + if existing_kb: + graph_data = _build_knowledge_graph(db, existing_kb) + conv.knowledge_graph_count = 1 + conv.updated_at = cst_now() + db.commit() + return { + "code": 0, + "data": { + "knowledge_base_id": existing_kb.id, + "knowledge_base_name": existing_kb.name, + "graph": graph_data, + "knowledge_graph_count": 1, + "already_exists": True, + }, + "message": "知识图谱已存在, 已刷新展示", + } + + kb_name = f"对话图谱-{conv.title}" + kb = KnowledgeBase( + id=gen_uuid(), + user_id=current_user.id, + name=kb_name, + description=f"从对话「{conv.title}」自动生成的知识图谱", + directory_path="/", + document_count=0, + created_at=cst_now(), + updated_at=cst_now(), + ) + db.add(kb) + db.flush() + + doc_count = 0 + for idx, msg in enumerate(messages): + role_label = "用户" if msg.role == "user" else "AI" + preview = re.sub(r'[\\n\\r\\t]+', ' ', msg.content[:40]).strip() + filename = f"msg_{idx+1:03d}_{role_label}_{preview}.txt" + + doc = KnowledgeDocument( + id=gen_uuid(), + knowledge_base_id=kb.id, + filename=filename, + file_type="txt", + file_size=len(msg.content.encode('utf-8')), + content=msg.content, + status="completed", + chunk_count=max(1, len(msg.content) // 500 + 1), + created_at=cst_now(), + updated_at=cst_now(), + ) + db.add(doc) + doc_count += 1 + + kb.document_count = doc_count + db.flush() + + # 构建知识图谱节点和边 + graph_data = _build_knowledge_graph(db, kb) + conv.knowledge_graph_base_id = kb.id + conv.knowledge_graph_count = 1 + conv.updated_at = cst_now() + + db.commit() + + LogService.log_operation( + user_id=current_user.id, username=current_user.username, action="convert_to_graph", + target="conversation", target_id=conv_id, + detail=f"将对话 {conv.title} 转换为知识图谱 {kb.name}", + ) + + return { + "code": 0, + "data": { + "knowledge_base_id": kb.id, + "knowledge_base_name": kb.name, + "graph": graph_data, + "knowledge_graph_count": 1, + "already_exists": False, + }, + "message": f"知识图谱构建完成", + } + + +@router.get("/models") +def get_models(db: Session = Depends(get_db)): + models = db.query(AIModel).filter(AIModel.is_enabled == True).all() + result = [] + for m in models: + provider = db.query(ModelProvider).filter(ModelProvider.id == m.provider_id).first() + result.append({ + "id": m.id, + "name": m.name, + "display_name": getattr(m, "display_name", None) or m.name, + "provider_id": m.provider_id, + "provider_name": provider.name if provider else "", + "status": m.status, + "price_per_million_tokens": float(m.price_per_million_tokens or 0), + "is_enabled": m.is_enabled, + "is_default": bool(m.is_default) if hasattr(m, 'is_default') else False, + "total_calls": m.total_calls or 0, + "total_tokens": m.total_tokens or 0, + "created_at": str(m.created_at) if m.created_at else None, + }) + result.sort(key=lambda x: (0 if x["is_default"] else 1, x["name"])) + return {"code": 0, "data": result, "message": "ok"} diff --git a/backend/python/app/api/im_bot_user.py b/backend/python/app/api/im_bot_user.py index 2aa8434..1c9a7f0 100644 --- a/backend/python/app/api/im_bot_user.py +++ b/backend/python/app/api/im_bot_user.py @@ -17,7 +17,7 @@ import json import random import threading import time -from datetime import datetime +from datetime import datetime, timedelta from typing import Optional import requests @@ -310,8 +310,11 @@ def _mirror_to_ai_chat(user_id: str, user_content: str, assistant_content: str): try: db = SessionLocal() cid = _get_or_create_wechat_conv(db, user_id) - _append_ai_message(db, cid, "user", user_content) - _append_ai_message(db, cid, "assistant", assistant_content) + now = datetime.now() + _append_ai_message(db, cid, "user", user_content, created=now) + # assistant 比 user 晚 50ms，保证排序严格递增，避免同秒时 ORDER BY 乱序 + _append_ai_message(db, cid, "assistant", assistant_content, + created=now + timedelta(milliseconds=50)) _execute_raw(db, text("UPDATE conversations SET updated_at=:u WHERE id=:id"), {"u": datetime.now(), "id": cid}) except Exception: