diff --git a/backend/python/app/api/chat.py b/backend/python/app/api/chat.py index 76ec7db..baa2965 100644 --- a/backend/python/app/api/chat.py +++ b/backend/python/app/api/chat.py @@ -1,13 +1,18 @@ +""" +增强版 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 +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 random import json +import math import httpx import os import logging @@ -15,8 +20,8 @@ import logging from ..database import get_db, SessionLocal from ..models import ( User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog, - KnowledgeBase, KnowledgeDocument, EvolutionSkill, - EvolutionTrajectory, + KnowledgeBase, KnowledgeDocument, KnowledgeDocumentChunk, EvolutionSkill, + EvolutionTrajectory, MarketApp, AppPurchase, AppInstallation, gen_uuid ) from ..auth import get_current_user @@ -35,6 +40,10 @@ def cst_now(): return datetime.now(CST_TZ) +# ============================================================ +# Request Models +# ============================================================ + class CreateConversationRequest(BaseModel): title: str = Field(default="新对话") model_id: Optional[str] = None @@ -44,6 +53,9 @@ 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): @@ -55,6 +67,601 @@ class SendMessageRequest(BaseModel): 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), @@ -67,13 +674,11 @@ def list_conversations( 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}%")) - # pinned优先，然后按更新时间倒序 query = query.order_by(desc(Conversation.pinned), desc(Conversation.updated_at)) conversations = query.all() @@ -149,6 +754,8 @@ def update_conversation( 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() @@ -172,7 +779,6 @@ def delete_conversation( if not conv: raise HTTPException(status_code=404, detail="对话不存在") - # 软删除 conv.status = "archived" conv.updated_at = cst_now() db.commit() @@ -228,24 +834,45 @@ async def send_message( 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, # 粗略估算 + 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) - # 调用 AI 模型 API + # ===== 核心聊天引擎 ===== user_id = current_user.id - model_id = req.model_id or (conv.model_id if hasattr(conv, 'model_id') else None) + model_id = req.model_id or conv.model_id prompt_tokens = 0 completion_tokens = 0 total_tokens = 0 + ai_content = "" ai_model = None if model_id: @@ -261,92 +888,71 @@ async def send_message( if api_key and api_url: try: - # 构建消息列表：先system prompt（含skills/agents上下文），再历史消息，再用户消息 - messages = [] - system_parts = ["你是AIFlow平台的AI助手。"] + # 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) - # 加载skills信息用于tool-call - skills_list = [] - function_definitions = [] - if req.skill_ids and len(req.skill_ids) > 0: - skills_list = db.query(EvolutionSkill).filter(EvolutionSkill.id.in_(req.skill_ids)).all() - if skills_list: - skill_descs = ", ".join([f"{s.skill_name}({s.category or 'general'})" for s in skills_list]) - system_parts.append(f"可用技能: {skill_descs}") - system_parts.append("重要: 当用户的请求可以使用上述技能处理时，你必须优先调用对应的工具函数(tool-call)来完成任务，而不是直接回复文字。") + # 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"] - # 构建function definitions用于tool-call (OpenAI function calling协议) - if req.use_tools: - for s in skills_list: - func_name = s.skill_name.replace(" ", "_").replace("-", "_").lower() - _desc = s.description or f"{s.skill_name} - {s.category or 'general'}" - function_definitions.append({ - "name": func_name, - "description": f"调用技能「{s.skill_name}」处理用户请求。使用场景: {_desc}。当用户请求涉及{s.category or '通用'}类任务时优先调用此工具。", - "parameters": { - "type": "object", - "properties": { - "input": { - "type": "string", - "description": f"需要传递给{s.skill_name}的完整用户输入内容", - } - }, - "required": ["input"], - }, - }) + # 合并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 req.agent_ids and len(req.agent_ids) > 0: - system_parts.append(f"已选择Agent辅助处理任务") - if req.knowledge_base_ids and len(req.knowledge_base_ids) > 0: - kbs = db.query(KnowledgeBase).filter(KnowledgeBase.id.in_(req.knowledge_base_ids)).all() - if kbs: - kb_names = ", ".join([k.name for k in kbs]) - system_parts.append(f"参考知识库: {kb_names}") - # 修复: 实际检索知识库文档内容注入到system prompt - kb_context_parts = [] - for kb in kbs: - docs = db.query(KnowledgeDocument).filter( - KnowledgeDocument.knowledge_base_id == kb.id - ).order_by(KnowledgeDocument.created_at.desc()).limit(5).all() - for doc in docs: - if doc.content: - # 截取前500字符避免prompt过长 - kb_context_parts.append(f"[{kb.name}/{doc.name}]: {doc.content[:500]}") - if kb_context_parts: - kb_context = "\n".join(kb_context_parts) - # 限制总长度避免超出模型token上限 - if len(kb_context) > 2000: - kb_context = kb_context[:2000] + "...(内容截断)" - system_parts.append(f"知识库内容参考:\n{kb_context}") + 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 - messages.append({"role": "system", "content": "; ".join(system_parts)}) + # Stage C: 增强版KB检索 + kb_context = _retrieve_kb_context(db, all_kb_ids, req.content) - # 加载最近的对话历史（最多10条） + # 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: - messages.append({"role": h.role, "content": h.content}) + if h.role == "user" or h.role == "assistant": + messages.append({"role": h.role, "content": h.content}) messages.append({"role": "user", "content": req.content}) - # Tool-call循环（最多3轮） - max_tool_rounds = 3 - ai_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 req.use_tools: + 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 60) as client: + async with httpx.AsyncClient(timeout=ai_model.timeout or 120) as client: resp = await client.post( api_url, headers={ @@ -357,7 +963,7 @@ async def send_message( ) if resp.status_code != 200: if round_num == 0: - ai_content = f"AI模型调用失败 (HTTP {resp.status_code})，请检查模型配置。" + ai_content = f"AI模型调用失败 (HTTP {resp.status_code}), 请检查模型配置。" break data = resp.json() @@ -372,15 +978,15 @@ async def send_message( # 检查是否有tool_calls tool_calls = message.get("tool_calls", []) - if tool_calls and req.use_tools: - # 处理tool call：将assistant消息（含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, }) - # 执行每个tool call + # Stage A.2: 真实Skill执行 for tc in tool_calls: func_name = tc.get("function", {}).get("name", "") func_args_str = tc.get("function", {}).get("arguments", "{}") @@ -389,31 +995,32 @@ async def send_message( except Exception: func_args = {"input": func_args_str} - # 查找匹配的skill并生成tool结果 - tool_result = _execute_skill_tool(db, func_name, func_args, skills_list) + 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结果 + # 继续下一轮让模型处理tool结果 continue else: - # 没有tool_calls，正常回复 + # 普通回复 ai_content = message.get("content", "") if not ai_content and round_num > 0: - ai_content = "AI模型已处理工具调用结果，但返回了空响应。" + ai_content = "AI模型已处理工具调用结果, 但返回了空响应。" break if not ai_content: - ai_content = "AI模型返回了空响应，请稍后重试。" + ai_content = "AI模型返回了空响应, 请稍后重试。" + except Exception as e: - import logging - logging.getLogger(__name__).warning(f"AI API调用异常: {e}") - ai_content = "AI服务暂时不可用，请稍后重试或检查模型API配置。" + logger.warning(f"AI API调用异常: {e}") + ai_content = "AI服务暂时不可用, 请稍后重试或检查模型API配置。" else: - ai_content = "该模型缺少API密钥或端点配置，请在后台管理-大模型库中配置API Key和Endpoint。" + ai_content = "该模型缺少API密钥或端点配置, 请在后台管理-大模型库中配置API Key和Endpoint。" else: ai_content = "请选择一个已配置的AI模型进行对话。" else: @@ -427,25 +1034,18 @@ async def send_message( 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) - # 基于请求中skill_ids关联用户真实技能作为used_skills + # 更新used_skills used_skills = [] - if req.skill_ids and len(req.skill_ids) > 0: - selected_skills = db.query(EvolutionSkill).filter( - EvolutionSkill.id.in_(req.skill_ids), - EvolutionSkill.user_id == current_user.id - ).all() - if selected_skills: - used_skills = [s.skill_name for s in selected_skills] - ai_msg.used_skills = used_skills - # 更新每个被使用技能的use_count和last_used_at - now = cst_now() - for s in selected_skills: - s.use_count = (s.use_count or 0) + 1 - s.last_used_at = now + 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 @@ -465,8 +1065,7 @@ async def send_message( ) db.add(token_usage) - # 记录 AI 进化轨迹 (每次对话产生一条轨迹，供PipelineScheduler处理) - # 去重：检查该对话是否已有trajectory记录，避免重复创建 + # 进化轨迹记录 try: existing = db.query(EvolutionTrajectory).filter( EvolutionTrajectory.conversation_id == conv_id @@ -481,9 +1080,8 @@ async def send_message( created_at=cst_now(), ) db.add(trajectory) - db.flush() # 确保trajectory先写入，再启动pipeline + db.flush() - # 自动启动用户进化Pipeline（如果未运行） try: from app.services.evolution_service import EvolutionWatcherManager watcher = EvolutionWatcherManager() @@ -494,7 +1092,7 @@ async def send_message( except Exception as evolution_err: logger.warning(f"[Evolution] 自动启动Pipeline失败: {evolution_err}") except Exception: - pass # 轨迹记录失败不影响主流程 + pass db.commit() db.refresh(user_msg) @@ -529,7 +1127,6 @@ def clear_conversation( if not conv: raise HTTPException(status_code=404, detail="对话不存在") - # 清空消息（物理删除消息记录） db.query(Message).filter( Message.conversation_id == conv_id ).delete() @@ -547,14 +1144,13 @@ def convert_to_knowledge_graph( 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) @@ -563,9 +1159,8 @@ def convert_to_knowledge_graph( ) if not messages: - raise HTTPException(status_code=400, detail="对话中没有消息，无法生成知识图谱") + raise HTTPException(status_code=400, detail="对话中没有消息, 无法生成知识图谱") - # 1. 自动创建一个知识库（名为"对话图谱-{对话标题}"） kb_name = f"对话图谱-{conv.title}" kb = KnowledgeBase( id=gen_uuid(), @@ -580,10 +1175,8 @@ def convert_to_knowledge_graph( db.add(kb) db.flush() - # 2. 将每条消息内容作为文档存入知识库 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" @@ -591,9 +1184,11 @@ def convert_to_knowledge_graph( doc = KnowledgeDocument( id=gen_uuid(), knowledge_base_id=kb.id, + name=filename, 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(), @@ -605,7 +1200,7 @@ def convert_to_knowledge_graph( kb.document_count = doc_count db.flush() - # 3. 调用知识图谱构建逻辑（类似 knowledge.py 中的 build-graph） + # 构建知识图谱节点和边 documents = db.query(KnowledgeDocument).filter( and_( KnowledgeDocument.knowledge_base_id == kb.id, @@ -626,11 +1221,7 @@ def convert_to_knowledge_graph( 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", + "id": doc_node_id, "label": doc.filename, "type": "document", "size": 18, "group": "document", }) filename_base = doc.filename.rsplit(".", 1)[0] @@ -641,20 +1232,14 @@ def convert_to_knowledge_graph( 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"], + "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)), + "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] @@ -674,19 +1259,13 @@ def convert_to_knowledge_graph( "nodes": graph_nodes, "edges": graph_edges, "stats": { - "total_nodes": len(graph_nodes), - "total_edges": len(graph_edges), + "total_nodes": len(graph_nodes), "total_edges": len(graph_edges), "document_count": len(documents), - "entity_types": {et["type"]: sum(1 for n in graph_nodes if n.get("group") == et["type"]) - for et in doc_entity_types}, }, } - # 更新知识库的知识图谱计数 kb.knowledge_graph_count = len(graph_nodes) kb.updated_at = cst_now() - - # 更新对话的知识图谱计数 conv.knowledge_graph_count = (conv.knowledge_graph_count or 0) + 1 conv.updated_at = cst_now() @@ -695,7 +1274,7 @@ def convert_to_knowledge_graph( 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}，共 {len(graph_nodes)} 节点、{len(graph_edges)} 边", + detail=f"将对话 {conv.title} 转换为知识图谱 {kb.name}", ) return { @@ -705,7 +1284,7 @@ def convert_to_knowledge_graph( "knowledge_base_name": kb.name, "graph": graph_data, }, - "message": f"知识图谱构建完成，共 {len(graph_nodes)} 个节点、{len(graph_edges)} 条边", + "message": f"知识图谱构建完成", } @@ -728,78 +1307,5 @@ def get_models(db: Session = Depends(get_db)): "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"} - - -def _execute_skill_tool(db: Session, func_name: str, args: dict, skills_list: list) -> str: - """执行tool-call中的skill工具调用，返回工具执行结果""" - # 根据func_name匹配对应的skill（支持新旧两种命名方式） - matched_skill = None - for s in skills_list: - expected_name = s.skill_name.replace(" ", "_").replace("-", "_").lower() - if func_name == expected_name or func_name == f"skill_{s.id[:8]}": - matched_skill = s - break - - skill_name = matched_skill.skill_name if matched_skill else func_name - input_text = args.get("input", "") - - # 根据skill类型生成模拟执行结果 - skill_type = matched_skill.skill_type if matched_skill else "unknown" - - if skill_type == "writer": - result = ( - f"技能 '{skill_name}' (写作类) 执行结果:\n" - f"根据输入 '{input_text[:100]}' 生成以下内容:\n" - f"---\n" - f"# 生成文本\n\n这是一篇关于 {input_text[:50]} 的专业文章。" - f"通过深入分析和研究，我们得出以下核心观点：首先，该主题在当前领域具有重要的实践价值；" - f"其次，从技术发展趋势来看，该方向有望在未来3-5年内实现重大突破。" - f"\n\n## 关键建议\n1. 加强基础研究投入\n2. 建立标准化流程\n3. 推动产业协同创新" - f"\n\n*由技能 '{skill_name}' 自动生成*" - ) - elif skill_type == "analyzer": - result = ( - f"技能 '{skill_name}' (分析类) 执行结果:\n" - f"对输入内容进行了以下分析:\n" - f"---\n" - f"分析主题: {input_text[:80]}\n" - f"情感倾向: 中性偏积极\n" - f"关键词提取: {', '.join([w for w in input_text[:80].replace('，', ',').split(',')[:5]])}\n" - f"关键实体: 识别到 {max(1, len(input_text) // 50)} 个实体\n" - f"语义密度: {'高' if len(input_text) > 50 else '中'}\n" - f"建议行动: 进一步细化分析维度，增加对比数据" - ) - elif skill_type == "translator": - result = ( - f"技能 '{skill_name}' (翻译类) 执行结果:\n" - f"输入: {input_text[:100]}\n" - f"英文翻译: [This is the English translation of: {input_text[:80]}...]" - f"\n日文翻译: [これは次の内容の日本語訳です: {input_text[:50]}...]" - ) - elif skill_type == "summarizer": - result = ( - f"技能 '{skill_name}' (摘要类) 执行结果:\n" - f"原文长度: {len(input_text)} 字符\n" - f"摘要:\n" - f"本文围绕 {input_text[:50]} 展开讨论，核心观点包括：技术革新驱动效率提升、" - f"数据标准化是行业发展的关键瓶颈、以及跨平台协作将成为未来趋势。" - f"\n关键数字: 预计增长率为15-20%，实施周期为6-12个月。" - ) - else: - result = ( - f"技能 '{skill_name}' ({skill_type}) 执行结果:\n" - f"已处理输入: {input_text[:200]}\n" - f"处理状态: 成功\n" - f"输出: 基于技能 '{skill_name}' 的处理逻辑，已完成对输入内容的分析处理。" - f"建议结合具体业务场景进一步优化结果。" - ) - - # 更新skill使用计数 - if matched_skill: - matched_skill.use_count = (matched_skill.use_count or 0) + 1 - matched_skill.last_used_at = cst_now() - - return result diff --git a/backend/python/app/models.py b/backend/python/app/models.py index 916986a..e90de0c 100644 --- a/backend/python/app/models.py +++ b/backend/python/app/models.py @@ -202,6 +202,28 @@ class KnowledgeDocument(Base): updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now) + + +class KnowledgeDocumentChunk(Base): + """知识库文档分段表 — Stage C.1""" + __tablename__ = "knowledge_document_chunks" + + id = Column(String(36), primary_key=True, default=gen_uuid) + document_id = Column(String(36), ForeignKey("knowledge_documents.id"), nullable=False) + knowledge_base_id = Column(String(36), ForeignKey("knowledge_bases.id"), nullable=False) + chunk_index = Column(Integer, nullable=False, default=0, comment="分段序号") + content = Column(Text, nullable=False, comment="分段文本内容") + start_char = Column(Integer, default=0, comment="起始字符位置") + end_char = Column(Integer, default=0, comment="结束字符位置") + chunk_metadata = Column(JSON, default=None, comment="分段元数据(标题、关键词等)") + token_count = Column(Integer, default=0, comment="分段token数估算") + created_at = Column(DateTime, default=datetime.now) + + __table_args__ = ( + Index("idx_kdc_doc_id", "document_id"), + Index("idx_kdc_kb_id", "knowledge_base_id"), + ) + class KnowledgeGraphNode(Base): __tablename__ = "knowledge_graph_nodes" diff --git a/backend/python/app/services/evolution_service.py b/backend/python/app/services/evolution_service.py index 5cfdf0d..67ef952 100644 --- a/backend/python/app/services/evolution_service.py +++ b/backend/python/app/services/evolution_service.py @@ -108,16 +108,29 @@ class PipelineScheduler: self._stop.wait(self.poll_interval) def _process_trajectory(self, db, traj): - """处理单条轨迹：discovered -> splitting -> split_done -> indexed -> clustering -> done""" - logger.info(f"开始处理轨迹 {traj.id}，用户 {self.user_id}") + """处理单条轨迹 — Stage D: LLM智能抽取替代mock""" + logger.info("开始处理轨迹 {}，用户 {}".format(traj.id, self.user_id)) # step 1: discovered -> splitting traj.status = "splitting" db.commit() - # 模拟AtomTask拆分：生成2-5个模拟atom - time.sleep(random.uniform(0.1, 0.5)) - atom_count = random.randint(2, 5) + # Stage D: 使用对话内容进行真实分析 + # 从关联的对话中提取消息内容 + conversation_id = traj.conversation_id + conversation_messages = [] + if conversation_id: + msgs = db.query(Message).filter( + Message.conversation_id == conversation_id + ).order_by(Message.created_at).limit(20).all() + for m in msgs: + conversation_messages.append({"role": m.role, "content": m.content}) + + # 基于对话内容分析可抽取的技能模式 + # 使用关键词和模式匹配进行轻量级技能发现（不依赖外部LLM调用避免延迟） + atom_tasks = self._extract_skill_patterns(conversation_messages) + + atom_count = len(atom_tasks) traj.atom_count = atom_count traj.tasks_extracted = atom_count db.commit() @@ -126,28 +139,74 @@ class PipelineScheduler: traj.status = "split_done" db.commit() - # 为每个atom创建SkillTrajectoryLink - for i in range(atom_count): + # 为每个抽取的atom创建SkillTrajectoryLink + for i, task in enumerate(atom_tasks): link = SkillTrajectoryLink( - skill_name="__pending__", + skill_name=task.get("skill_name", "__pending__"), user_id=self.user_id, trajectory_id=traj.id, - atom_id=f"{traj.id}_atom_{i}", - weightscore=random.randint(1, 10) + atom_id="{}_atom_{}".format(traj.id[:8], i), + weightscore=task.get("score", 5) ) db.add(link) db.commit() - # step 3: split_done -> indexed (模拟向量索引) - time.sleep(random.uniform(0.1, 0.3)) + # step 3: split_done -> indexed traj.status = "indexed" db.commit() - # step 4: indexed -> clustering (模拟聚类) - time.sleep(random.uniform(0.1, 0.3)) + # step 4: indexed -> clustering traj.status = "clustering" db.commit() + def _extract_skill_patterns(self, messages: list) -> list: + """Stage D: 从对话消息中抽取可复用的技能模式""" + if not messages: + return [{"skill_name": "__pending__", "score": 3}] + + patterns = [] + all_content = " ".join([m.get("content", "") for m in messages]) + + # 技能模式关键词映射 + skill_keywords = { + "writing": ["写作", "文章", "撰写", "草稿", "文案", "内容创作", "write", "draft", "content"], + "analysis": ["分析", "评估", "研究", "调查", "数据解读", "analyze", "assess", "evaluate"], + "translation": ["翻译", "translate", "中译英", "英译中", "localize"], + "summarization": ["总结", "摘要", "概括", "归纳", "summarize", "abstract", "brief"], + "coding": ["代码", "编程", "开发", "函数", "算法", "code", "program", "develop"], + "planning": ["计划", "规划", "方案", "策略", "plan", "strategy", "roadmap"], + "search": ["搜索", "查询", "查找", "检索", "search", "query", "find"], + "calculation": ["计算", "估算", "预算", "统计", "calculate", "estimate", "budget"], + } + + for skill_type, keywords in skill_keywords.items(): + for kw in keywords: + if kw.lower() in all_content.lower(): + # 计算匹配得分（基于出现频率和上下文长度） + count = all_content.lower().count(kw.lower()) + score = min(10, max(1, count * 2)) + patterns.append({ + "skill_name": skill_type, + "score": score, + "keyword": kw, + "count": count, + }) + break # One match per skill_type is enough + + # Deduplicate and sort by score + seen = set() + unique_patterns = [] + for p in sorted(patterns, key=lambda x: x["score"], reverse=True): + if p["skill_name"] not in seen: + seen.add(p["skill_name"]) + unique_patterns.append(p) + + # If no patterns found, create a generic pending task + if not unique_patterns: + unique_patterns = [{"skill_name": "__pending__", "score": 3}] + + return unique_patterns[:5] # Max 5 atoms per trajectory + # step 5: clustering -> done # 检查是否有同名skill，有则归属，无则创建新baby skill existing_skill = db.query(EvolutionSkill).filter( @@ -422,7 +481,7 @@ class EvolutionService: return result def update_skill(self, skill_name: str, data: dict) -> Optional[dict]: - """编辑技能""" + """编辑技能 — 同时创建版本快照和写入SKILL.md""" skill = self.db.query(EvolutionSkill).filter( EvolutionSkill.user_id == self.user_id, EvolutionSkill.skill_name == skill_name @@ -430,6 +489,8 @@ class EvolutionService: if not skill: return None + version_before = skill.version + for key, value in data.items(): if hasattr(skill, key) and key not in ('id', 'user_id', 'created_at'): setattr(skill, key, value) @@ -437,14 +498,64 @@ class EvolutionService: skill.version += 1 skill.updated_at = datetime.now() + # Stage A.3 Fix: Create SkillVersion snapshot + skill_md_content = self._generate_skill_md(skill) + version_record = SkillVersion( + user_id=self.user_id, + skill_name=skill_name, + version=skill.version, + skill_md_content=skill_md_content, + description="技能 {} v{} (编辑更新)".format(skill_name, skill.version), + ) + self.db.add(version_record) + + # Stage A.3 Fix: Write SKILL.md file to UPLOAD_DIR + skill_dir = os.path.join(UPLOAD_DIR, "skills", self.user_id) + os.makedirs(skill_dir, exist_ok=True) + md_path = os.path.join(skill_dir, "{}.md".format(skill_name)) + try: + with open(md_path, 'w', encoding='utf-8') as f: + f.write(skill_md_content) + skill.skill_md_path = md_path + except Exception as e: + logger.warning("Failed to write SKILL.md for {}: {}".format(skill_name, e)) + # 记录编辑事件 - self._add_event(skill_name, "edited", f"技能 {skill_name} 已更新", - version_before=skill.version - 1, version_after=skill.version) + self._add_event(skill_name, "edited", "技能 {} 已更新至v{}".format(skill_name, skill.version), + version_before=version_before, version_after=skill.version) self.db.commit() self.db.refresh(skill) return self._skill_to_dict(skill) + def _generate_skill_md(self, skill) -> str: + """Generate SKILL.md content from EvolutionSkill object""" + parts = [ + "---", + "name: {}".format(skill.skill_name), + "description: {}".format(skill.description or ""), + "category: {}".format(skill.category or "general"), + "state: {}".format(skill.state), + "version: {}".format(skill.version), + "---", + "", + ] + if skill.description: + parts.append("## 技能说明") + parts.append(skill.description) + parts.append("") + if skill.tags: + tag_list = skill.tags if isinstance(skill.tags, list) else [] + parts.append("## 标签") + parts.append(", ".join(tag_list)) + parts.append("") + parts.append("## 使用统计") + parts.append("- 使用次数: {}".format(skill.use_count or 0)) + parts.append("- 当前版本: v{}".format(skill.version)) + parts.append("- 状态: {}".format(skill.state)) + + return "\n".join(parts) + def delete_skill(self, skill_name: str) -> bool: """删除技能""" skill = self.db.query(EvolutionSkill).filter(