""" Skill 管理 API 路由 提供： 1. Skill 的 CRUD 操作 2. Agent-Skill 生命周期端点 3. AI 辅助 Skill 创建（完整流程） 4. Skill 选择和路由（LLM 与 Skills 结合） 5. Agent 工作流配置 6. 文档驱动的 Skill 生成（新增） """ from datetime import datetime from fastapi import APIRouter, Depends, HTTPException, status from typing import List, Optional, Dict, Any from pydantic import BaseModel from app.models.skill import ( Skill, SkillCreate, SkillUpdate, SkillConfigUpdate, SkillTestRequest, SkillTestResponse, SkillGenerateRequest, SkillGenerateResponse, # 新增模型 DocFetchRequest, GitHubDocFetchRequest, SkillGenerateFromDocsRequest, SkillGenerateFromDocsResponse, SkillPreviewRequest, SkillPreviewResponse, SkillSaveFromPreviewRequest, SkillRefineRequest, SkillRefineResponse, ) from app.models.skill_integration import ( SkillMetadata, SkillTriggerType, SkillStructure, SkillGenerationRequest, SkillGenerationResponse, SkillAnalysis, SkillPlanning, SkillRefinementRequest, SkillRefinementResponse, SkillSelectionCriteria, SkillSelectionResult, SkillExecutionContext, SkillExecutionResponse, AgentWorkflowConfig, AgentWorkflowStep, ) from app.core.skills.skill_manager import get_skill_manager, SkillManager from app.core.llm.glm_client import get_glm_client, GLMClient from app.utils.logger import get_logger logger = get_logger(__name__) router = APIRouter(prefix="/skills", tags=["Skill管理"]) # ============================================================================ # 基础 CRUD 端点 # ============================================================================ @router.get("/", response_model=List[Skill]) async def list_skills( skill_type: Optional[str] = None, category: Optional[str] = None, skill_manager: SkillManager = Depends(get_skill_manager) ): """ 列出所有 Skills - **skill_type**: 筛选类型 (builtin/user) - **category**: 筛选分类 """ skills = await skill_manager.list_skills(skill_type=skill_type, category=category) return skills @router.get("/{skill_id}", response_model=Skill) async def get_skill( skill_id: str, skill_manager: SkillManager = Depends(get_skill_manager) ): """获取 Skill 详情""" skill = await skill_manager.load_skill(skill_id) if not skill: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Skill 不存在: {skill_id}" ) return skill @router.post("/", response_model=Skill, status_code=status.HTTP_201_CREATED) async def create_skill( skill_data: SkillCreate, skill_manager: SkillManager = Depends(get_skill_manager) ): """ 创建新的用户 Skill 如果不提供 ID，系统将自动生成： - 使用 UUID 作为唯一标识 - 基于名称生成可读 ID（kebab-case） """ try: # 如果没有提供 ID，自动生成 if not skill_data.id: import uuid import re # 生成基于名称的 kebab-case ID base_id = skill_data.name.lower() base_id = re.sub(r'[^\w\s-]', '', base_id) # 移除特殊字符 base_id = re.sub(r'\s+', '-', base_id) # 空格替换为连字符 base_id = re.sub(r'-{2,}', '-', base_id) # 多个连字符合并为一个 # 如果生成结果为空，使用名称的拼音或默认前缀 if not base_id or len(base_id) < 3: base_id = f"skill-{uuid.uuid4().hex[:8]}" # 确保 ID 唯一（添加随机后缀） unique_id = f"{base_id}-{uuid.uuid4().hex[:6]}" # 创建新的 SkillCreate 对象，使用自动生成的 ID skill_data_with_id = skill_data.copy() skill_data_with_id.id = unique_id skill = await skill_manager.create_user_skill(skill_data_with_id) else: skill = await skill_manager.create_user_skill(skill_data) return skill except Exception as e: logger.error(f"创建 Skill 失败: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"创建 Skill 失败: {str(e)}" ) @router.put("/{skill_id}", response_model=Skill) async def update_skill( skill_id: str, skill_data: SkillUpdate, skill_manager: SkillManager = Depends(get_skill_manager) ): """更新 Skill 内容""" skill = await skill_manager.update_user_skill(skill_id, skill_data) if not skill: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Skill 不存在或不允许更新: {skill_id}" ) return skill @router.put("/{skill_id}/config", response_model=Skill) async def update_skill_config( skill_id: str, config_update: SkillConfigUpdate, skill_manager: SkillManager = Depends(get_skill_manager) ): """更新 Skill 配置""" skill = await skill_manager.load_skill(skill_id) if not skill: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Skill 不存在: {skill_id}" ) # 更新配置 if config_update.preset is not None: skill.config.preset = config_update.preset if config_update.parameters is not None: skill.config.parameters.update(config_update.parameters) if config_update.weights is not None: skill.config.weights = config_update.weights return skill @router.post("/{skill_id}/test", response_model=SkillTestResponse) async def test_skill( skill_id: str, test_request: SkillTestRequest, skill_manager: SkillManager = Depends(get_skill_manager), glm_client: GLMClient = Depends(get_glm_client) ): """ 测试 Skill 使用 Skill 的行为指导来处理测试输入，返回 LLM 的响应 """ skill = await skill_manager.load_skill(skill_id) if not skill: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Skill 不存在: {skill_id}" ) try: # 使用 GLM 客户端的 chat_with_skill 方法 response = await glm_client.chat_with_skill( skill_behavior=skill.behavior_guide, user_input=test_request.test_input, context=test_request.context or {}, temperature=test_request.temperature ) return SkillTestResponse( skill_id=skill_id, skill_name=skill.name, response=response ) except Exception as e: logger.error(f"测试 Skill 失败: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"测试 Skill 失败: {str(e)}" ) @router.delete("/{skill_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_skill( skill_id: str, skill_manager: SkillManager = Depends(get_skill_manager) ): """删除用户 Skill""" success = await skill_manager.delete_user_skill(skill_id) if not success: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Skill 不存在或不允许删除: {skill_id}" ) return None @router.post("/{skill_id}/reload", response_model=Skill) async def reload_skill( skill_id: str, skill_manager: SkillManager = Depends(get_skill_manager) ): """重新加载 Skill（清除缓存）""" skill = await skill_manager.reload_skill(skill_id) if not skill: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Skill 不存在: {skill_id}" ) return skill @router.get("/categories/list", response_model=List[str]) async def list_categories( skill_manager: SkillManager = Depends(get_skill_manager) ): """列出所有 Skill 分类""" skills = await skill_manager.list_skills() categories = set(skill.category for skill in skills) return sorted(list(categories)) @router.post("/generate", response_model=SkillGenerateResponse) async def generate_skill_with_ai( request: SkillGenerateRequest, skill_manager: SkillManager = Depends(get_skill_manager), glm_client: GLMClient = Depends(get_glm_client) ): """ 使用 AI 生成 Skill（类似 Claude Code 的 skill-creator） 流程： 1. 加载 skill-creator 的行为指导 2. 将用户需求和 skill-creator 标准一起发送给 GLM-4.7 3. AI 生成符合标准的 Skill 内容 Args: request: 包含用户描述、分类、标签的请求 Returns: 生成的 Skill 内容和建议信息 """ try: # 1. 加载 skill-creator 的行为指导 skill_creator = await skill_manager.load_skill("skill-creator") if not skill_creator: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="skill-creator 未找到，请确保内置 Skills 正确安装" ) # 2. 构建提示词 user_requirements = f"""用户想要创建的 Skill 描述： {request.description} """ if request.category: user_requirements += f"\n指定分类：{request.category}" if request.tags: user_requirements += f"\n指定标签：{', '.join(request.tags)}" system_prompt = f"""你是一个专业的 Skill 创建专家。 以下是 skill-creator 的行为指导（关于如何创建有效 Skill 的指南）： {'━' * 60} {skill_creator.behavior_guide} {'━' * 60} 你的任务是根据用户的需求，创建一个符合上述标准的 Skill。 **重要要求**： 1. SKILL.md 必须以 YAML frontmatter 开始，包含 name 和 description 字段 2. description 应该清晰说明此 Skill 的用途和使用场景 3. 行为指导部分应该简洁、具体，避免冗余的解释 4. 使用 markdown 格式 5. 返回完整的 SKILL.md 内容 请以 JSON 格式返回结果，包含以下字段： - suggested_id: 建议 Skill ID（kebab-case，如 dialogue-writer-ancient） - suggested_name: 建议 Skill 名称（简短中文） - skill_content: 完整的 SKILL.md 内容 - category: 分类（如"编剧"、"审核"、"通用"等） - suggested_tags: 建议标签数组 - explanation: 对生成的 Skill 的简要说明（中文） """ # 3. 调用 GLM-4.7 生成 response = await glm_client.chat( messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_requirements} ], temperature=request.temperature ) # 4. 解析响应 import json import re response_text = response["choices"][0]["message"]["content"] # 尝试提取 JSON json_match = re.search(r'\{[\s\S]*\}', response_text) if json_match: result = json.loads(json_match.group()) else: # 如果没有找到 JSON，创建一个默认响应 result = { "suggested_id": "custom-skill", "suggested_name": "自定义 Skill", "skill_content": response_text, "category": request.category or "通用", "suggested_tags": request.tags or ["自定义"], "explanation": "AI 生成的 Skill 内容" } return SkillGenerateResponse(**result) except HTTPException: raise except Exception as e: logger.error(f"AI 生成 Skill 失败: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"AI 生成 Skill 失败: {str(e)}" ) @router.get("/{skill_id}/with-references", response_model=dict) async def get_skill_with_references( skill_id: str, skill_manager: SkillManager = Depends(get_skill_manager), include_references: bool = True ): """ 获取 Skill 及其参考文件 这是实现参考文件融入 LLM 的关键端点： - 返回 Skill 对象 - 返回所有参考文件的内容 Args: skill_id: Skill ID include_references: 是否包含参考文件内容 """ try: skill = await skill_manager.load_skill(skill_id) if not skill: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Skill 不存在: {skill_id}" ) result = {"skill": skill.model_dump()} # 加载参考文件（如果需要） if include_references: references = await skill_manager.load_skill_references(skill.id) result["references"] = references return result except HTTPException: raise except Exception as e: logger.error(f"获取 Skill 失败: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"获取 Skill 失败: {str(e)}" ) @router.post("/analyze", response_model=SkillAnalysis) async def analyze_skill_requirements( request: SkillGenerationRequest, skill_manager: SkillManager = Depends(get_skill_manager), glm_client: GLMClient = Depends(get_glm_client) ): """ 分析 Skill 创建需求（完整流程 Step 1: Understanding） Args: request: 包含用户意图、描述、用例的请求 Returns: SkillAnalysis: 需求分析结果 """ try: # Step 1: 加载 skill-creator skill_creator = await skill_manager.load_skill("skill-creator") if not skill_creator: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="skill-creator 未找到，请确保内置 Skills 正确安装" ) # Step 2: 需求分析 user_intent = request.user_intent or "创建新的 Skill" system_prompt = f"""你是一个专业的 Skill 创建分析师。 以下是 skill-creator 的行为指导（关于如何创建有效 Skill 的指南）： {'━' * 60} {skill_creator.behavior_guide} {'━' * 60} 你的任务是分析用户的需求：{user_intent} 请从以下几个方面分析： 1. **核心功能**：这个 Skill 主要用于什么？ 2. **目标用户**：谁会使用这个 Skill？ 3. **技术要求**：前端、后端、数据分析等 4. **特殊要求**：多语言支持、高精度输出、实时处理 分析用户需求，并提出 Skill 创作建议。 请以 JSON 格式返回分析结果，包含以下字段： - intent_analysis: 用户的意图分析 - suggested_category: 建议的分类（如 "dialogue", "analysis", "writing"） - suggested_features: 建议的核心功能列表 - complexity_assessment: 复杂度评估（simple/medium/complex） - recommended_approach: 推荐的实现方式 """ # 构建分析请求 analysis_request = { "user_intent": user_intent, "user_description": request.user_intent, "use_cases": request.use_cases or [] } # 调用 LLM 进行需求分析 analysis_response = await glm_client.chat( messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": f"分析以下需求：{json.dumps(analysis_request, ensure_ascii=False)}"} ], temperature=0.5 ) # 解析分析结果 try: import json analysis_result = json.loads(analysis_response["choices"][0]["message"]["content"]) # 提取结构化数据 intent_analysis = analysis_result.get("intent_analysis", {}) suggested_category = analysis_result.get("suggested_category", "general") suggested_features = analysis_result.get("suggested_features", []) complexity_assessment = analysis_result.get("complexity_assessment", "medium") recommended_approach = analysis_result.get("recommended_approach", "standard") except json.JSONDecodeError: # 如果无法解析，使用默认分析 intent_analysis = {"primary": "创建新的 Skill", "details": "用户想要创建自定义 Skill"} suggested_category = "general" suggested_features = ["standard_dialogue_creation", "basic_content_editing"] complexity_assessment = "medium" recommended_approach = "standard" return SkillAnalysis( intent_analysis=intent_analysis, suggested_category=suggested_category, suggested_features=suggested_features, complexity_assessment=complexity_assessment, recommended_approach=recommended_approach ) except HTTPException: raise except Exception as e: logger.error(f"需求分析失败: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"需求分析失败: {str(e)}" ) # ============================================================================ # Skill 选择和路由（LLM 自动选择合适的 Skills） # ============================================================================ @router.post("/select-skills", response_model=SkillSelectionResult) async def select_skills_for_context( criteria: SkillSelectionCriteria, skill_manager: SkillManager = Depends(get_skill_manager), glm_client: GLMClient = Depends(get_glm_client) ): """ LLM 根据描述自动选择最合适的 Skills 流程： 1. 分析用户输入 2. 匹配所有 Skills 的 metadata 3. 评分和排序 4. 返回选择结果 Args: criteria: 选择条件（用户意图、上下文、约束） """ try: # 加载所有可用 Skills all_skills = await skill_manager.list_skills(skill_type="user") # 提取 Skills 的 metadata skills_metadata = [] for skill in all_skills: metadata = { "id": skill.id, "name": skill.name, "description": skill.description, "category": skill.category, "tags": skill.tags, "keywords": skill.description.split(), "capabilities": skill.behavior_guide[:200] if skill.behavior_guide else "" } skills_metadata.append(metadata) # 构建提示词 context_info = "" if criteria.context: context_info = f"\n上下文信息：{json.dumps(criteria.context, ensure_ascii=False)}\n" if criteria.exclude_skills: exclude_info = f"\n排除的 Skills：{', '.join(criteria.exclude_skills)}\n" system_prompt = f"""你是一个智能 Skill 选择助手。 以下是可以使用的 Skills 及其元数据： {'━' * 60} """ for i, skill_meta in enumerate(skills_metadata, 1): system_prompt += f"\n{i+1}. {skill_meta['name']}: {skill_meta['description']}\n" if skill_meta['capabilities']: system_prompt += f"\n 能力：{skill_meta['capabilities']}\n" system_prompt += f"\n" system_prompt += f"""你的任务是根据用户的需求和偏好，从以下 Skills 中选择最合适的。 用户需求： - {criteria.user_intent} {criteria.exclude_info if criteria.exclude_skills else ""} {criteria.context_info if criteria.context else ""} {criteria.required_category if criteria.required_category else ""} {criteria.required_tags if criteria.required_tags else ""} 选择标准： 1. 相关性（description/keywords 匹配度） 2. 功能覆盖度（capabilities 是否满足需求） 3. 难度适配（complexity_assessment 是否匹配） 请返回 JSON 数组，包含选中的 Skill ID 列表和选择理由。 每个元素应包含： - skill_id: Skill ID - score: 匹配分数（0-100） - reason: 选择理由 """ # 调用 LLM 进行选择 selection_response = await glm_client.chat( messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": f"请分析以下 Skills，并返回最合适的 3-5 个：{json.dumps(skills_metadata, ensure_ascii=False)}"} ], temperature=0.3 ) # 解析选择结果 try: import json selected_ids = json.loads(selection_response["choices"][0]["message"]["content"]) except: selected_ids = [] # 如果是复杂场景，需要 LLM 进行智能选择 if len(skills_metadata) > 5 or criteria.user_intent == "复杂需求分析": # 使用结构化输出确保可解析性 structured_prompt = """ 请以以下 JSON 格式返回选择，必须是可以解析的 JSON 数组： [ {{"skill_id": "...", "score": 85, "reason": "..."}}, ... ] 确保输出是有效的 JSON 格式。 """ selection_response = await glm_client.chat( messages=[ {"role": "system", "content": structured_prompt}, {"role": "user", "content": f"用户需求：{criteria.user_intent or '标准 Skill 创建'}\n\n\nAvailable Skills: {json.dumps(skills_metadata, ensure_ascii=False)}"} ], temperature=0.2 ) try: selected_ids = json.loads(selection_response["choices"][0]["message"]["content"]) except: selected_ids = [] return SkillSelectionResult( selected_skills=[s for s in all_skills if s.id in selected_ids], selection_reason=f"基于需求分析选择了最相关的 Skills", confidence_scores={} ) except HTTPException: logger.error(f"Skill 选择失败: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Skill 选择失败: {str(e)}" ) # ============================================================================ # Agent 工作流配置 API # ============================================================================ @router.post("/agent-workflow/configure", response_model=AgentWorkflowConfig) async def configure_agent_workflow( config: AgentWorkflowConfig, skill_manager: SkillManager = Depends(get_skill_manager) ): """ 配置 Agent 工作流的 Skill 使用方式 这是实现 Agent 与 Skill 深度结合的关键配置： - 定义哪些工作流步骤使用哪些 Skills - 设置默认温度 - 启用/禁用 Skill 替代 Args: config: 包含工作流步骤配置 """ try: # 加载 skill-creator 获取指导 skill_creator = await skill_manager.load_skill("skill-creator") # 保存配置 await skill_manager.save_workflow_config(config) return { "success": True, "message": "Agent 工作流配置已更新", "config": config.model_dump() } except Exception as e: logger.error(f"配置失败: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"配置失败: {str(e)}" ) # ============================================================================ # 基础 CRUD 端点继续 # ============================================================================ # ============================================================================ # Agent-Skill 生命周期端点 - 使用 skill-creator 脚本 # ============================================================================ class SkillInitRequest(BaseModel): """初始化 Skill 模板请求""" skill_name: str # Skill 名称（hyphen-case） output_path: Optional[str] = None # 输出路径 class SkillValidateScriptRequest(BaseModel): """脚本验证 Skill 请求""" skill_id: str # 要验证的 Skill ID class SkillPackageRequest(BaseModel): """打包 Skill 请求""" skill_id: str # 要打包的 Skill ID output_dir: Optional[str] = None # 输出目录 @router.post("/init", response_model=dict) async def init_skill_template( request: SkillInitRequest, skill_manager: SkillManager = Depends(get_skill_manager) ): """ 初始化 Skill 模板（使用 skill-creator 的 init_skill.py 脚本） 这是 Agent-Skill 生命周期的正确实现： - 调用 skill-creator 的 init_skill.py 脚本 - 创建标准的 Skill 目录结构（SKILL.md、scripts/、references/、assets/） """ try: result = await skill_manager.init_skill_template( skill_name=request.skill_name, output_path=request.output_path ) if result["success"]: logger.info(f"成功初始化 Skill 模板: {request.skill_name}") return { "success": True, "message": f"Skill 模板 '{request.skill_name}' 初始化成功", "output": result["output"] } else: logger.error(f"初始化 Skill 失败: {result.get('error')}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=result.get("error", "初始化失败") ) except Exception as e: logger.error(f"初始化 Skill 异常: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"初始化 Skill 异常: {str(e)}" ) @router.post("/validate-script", response_model=dict) async def validate_skill_script( request: SkillValidateScriptRequest, skill_manager: SkillManager = Depends(get_skill_manager) ): """ 验证 Skill 结构（使用 skill-creator 的 quick_validate.py 脚本） 这是 Agent-Skill 生命周期的正确实现： - 调用 skill-creator 的 quick_validate.py 脚本 - 检查 YAML frontmatter、命名规范等 """ try: result = await skill_manager.validate_skill_structure(request.skill_id) if result["success"]: logger.info(f"Skill 验证通过: {request.skill_id}") return { "success": True, "valid": True, "message": result["output"] } else: logger.warning(f"Skill 验证失败: {request.skill_id}") return { "success": True, "valid": False, "error": result.get("error") } except Exception as e: logger.error(f"验证 Skill 异常: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"验证 Skill 异常: {str(e)}" ) @router.post("/package", response_model=dict) async def package_skill( request: SkillPackageRequest, skill_manager: SkillManager = Depends(get_skill_manager) ): """ 打包 Skill（使用 skill-creator 的 package_skill.py 脚本） 这是 Agent-Skill 生命周期的正确实现： - 调用 skill-creator 的 package_skill.py 脚本 - 将 Skill 打包成 .skill 文件用于分发 """ try: result = await skill_manager.package_skill( skill_id=request.skill_id, output_dir=request.output_dir ) if result["success"]: logger.info(f"成功打包 Skill: {request.skill_id}") return { "success": True, "message": f"Skill '{request.skill_id}' 打包成功", "output": result["output"] } else: logger.error(f"打包 Skill 失败: {request.skill_id}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=result.get("error", "打包失败") ) except Exception as e: logger.error(f"打包 Skill 异常: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"打包 Skill 异常: {str(e)}" ) @router.get("/tools", response_model=List[dict]) async def list_available_tools( skill_manager: SkillManager = Depends(get_skill_manager) ): """ 列出所有已注册的工具（Agent-Skill 生命周期：注册阶段） 返回所有已注册工具的 JSON Schema，用于： - 注入到 LLM System Prompt - 或作为 API 的 tools 参数 """ return skill_manager.get_available_tools_for_llm() @router.post("/execute-tool", response_model=dict) async def execute_tool( tool_name: str, parameters: dict, skill_manager: SkillManager = Depends(get_skill_manager) ): """ 执行工具（Agent-Skill 生命周期：路由 → 执行 → 反馈） 实现： - 路由：根据 tool_name 找到对应的 handler - 执行：调用 handler 执行实际操作 - 反馈：返回结果给调用者（通常是 LLM） 这是 Agent-Skill 生命周期的核心实现 """ result = await skill_manager.route_and_execute_tool(tool_name, parameters) return result # ============================================================================ # 增强的 Skill 生成工作流（文档驱动） # ============================================================================ import uuid from collections import defaultdict from app.core.documentation_fetcher import get_documentation_fetcher, DocumentationFetcher # 预览存储（生产环境应使用 Redis 或数据库） _preview_storage: Dict[str, Dict[str, Any]] = defaultdict(dict) _preview_references: Dict[str, Dict[str, str]] = defaultdict(dict) @router.post("/fetch-doc", response_model=dict) async def fetch_documentation( request: DocFetchRequest, doc_fetcher: DocumentationFetcher = Depends(get_documentation_fetcher) ): """ 获取文档内容（用于预览或生成 Skill） Args: request: 包含文档 URL 和可选的 CSS 选择器 Returns: 获取的文档内容 """ try: result = await doc_fetcher.fetch_from_url(request.url, request.selector) if result.get("success"): return { "success": True, "title": result.get("title"), "content": result.get("content"), "word_count": result.get("word_count"), "char_count": result.get("char_count"), "url": result.get("url") } else: return { "success": False, "error": result.get("error") } except Exception as e: logger.error(f"获取文档失败: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"获取文档失败: {str(e)}" ) @router.post("/fetch-github-doc", response_model=dict) async def fetch_github_documentation( request: GitHubDocFetchRequest, doc_fetcher: DocumentationFetcher = Depends(get_documentation_fetcher) ): """ 获取 GitHub 文档内容 Args: request: GitHub 仓库和文档路径 Returns: 获取的文档内容 """ try: result = await doc_fetcher.fetch_from_github( repo_url=request.repo_url, docs_path=request.docs_path ) if result.get("success"): return { "success": True, "repo_url": result.get("repo_url"), "docs_path": result.get("docs_path"), "content": result.get("content"), "word_count": result.get("word_count"), "char_count": result.get("char_count") } else: return { "success": False, "error": result.get("error") } except Exception as e: logger.error(f"获取 GitHub 文档失败: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"获取 GitHub 文档失败: {str(e)}" ) @router.post("/generate-from-docs", response_model=SkillGenerateFromDocsResponse) async def generate_skill_from_documentation( request: SkillGenerateFromDocsRequest, skill_manager: SkillManager = Depends(get_skill_manager), glm_client: GLMClient = Depends(get_glm_client), doc_fetcher: DocumentationFetcher = Depends(get_documentation_fetcher) ): """ 基于文档生成 Skill（新工作流） 完整流程： 1. 获取文档内容（URL/GitHub/上传） 2. 提取关键信息 3. 使用 LLM 生成 SKILL.md 4. 生成预览 ID 5. 返回可预览的内容 用户可以： - 预览生成的内容 - 手动编辑 - 用 AI 调整 - 确认后保存到系统 """ try: logger.info(f"开始基于文档生成 Skill: {request.skill_name}") # 1. 收集所有文档内容 all_docs = [] doc_summary_parts = [] # 从 URL 获取 if request.doc_urls: for url in request.doc_urls: result = await doc_fetcher.fetch_from_url(url) if result.get("success"): all_docs.append({ "source": f"URL: {url}", "title": result.get("title"), "content": result.get("content") }) doc_summary_parts.append(f"- {result.get('title')} ({url})") # 从 GitHub 获取 if request.github_repos: for github_req in request.github_repos: result = await doc_fetcher.fetch_from_github( repo_url=github_req.repo_url, docs_path=github_req.docs_path ) if result.get("success"): all_docs.append({ "source": f"GitHub: {github_req.repo_url}/{github_req.docs_path}", "title": github_req.docs_path, "content": result.get("content") }) doc_summary_parts.append(f"- {github_req.repo_url}/{github_req.docs_path}") # 用户上传的 references if request.uploaded_references: for filename, content in request.uploaded_references.items(): all_docs.append({ "source": f"上传: {filename}", "title": filename, "content": content }) doc_summary_parts.append(f"- {filename} (用户上传)") # 2. 构建 LLM 提示词 doc_context = "" if all_docs: # 取文档摘要（前 500 字） doc_summary = "\n".join(doc_summary_parts) # 构建文档上下文（避免超过上下文窗口） doc_context_parts = [] for doc in all_docs[:3]: # 最多 3 个文档 content_preview = doc["content"][:500].replace("\n", " ") doc_context_parts.append(f"#### {doc['title']}\n{content_preview}...") doc_context = f""" **文档来源**： {doc_summary} **文档内容摘要**： {''.join(doc_context_parts)} """ # 3. 加载 skill-creator 标准 skill_creator = await skill_manager.load_skill("skill-creator") creator_guide = skill_creator.behavior_guide if skill_creator else "" # 4. 构建 LLM 提示词 system_prompt = f"""你是一个专业的 Skill 创建专家。 {creator_guide if creator_guide else ''} 你的任务是基于提供的文档内容，创建一个高质量的 Skill。 **重要要求**： 1. SKILL.md 必须以 YAML frontmatter 开始 2. YAML 必须包含：name, description, category, tags 3. description 应该清晰说明此 Skill 的用途和使用场景 4. 行为指导部分应该简洁、具体，从文档中提取关键信息 5. 如果文档包含代码示例、API、配置等，应该在行为指导中体现 6. 使用 markdown 格式 请以 JSON 格式返回结果，包含以下字段： - skill_content: 完整的 SKILL.md 内容（以 --- 开头的 YAML frontmatter + markdown 内容） - explanation: 对生成的 Skill 的简要说明（中文） - suggested_category: 建议的分类 - suggested_tags: 建议的标签数组""" user_prompt = f"""请基于以下文档创建一个 Skill： **Skill 名称**：{request.skill_name} **描述**：{request.description or f'基于文档生成的 Skill：{request.skill_name}'} **指定分类**：{request.category or '自动推断'} **指定标签**：{', '.join(request.tags) if request.tags else '自动推断'} {doc_context if request.include_doc_summary else ''} 请生成完整的 SKILL.md：""" # 5. 调用 LLM 生成 response = await glm_client.chat( messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ], temperature=request.temperature ) response_text = response["choices"][0]["message"]["content"] # 6. 解析 JSON 响应 import json import re json_match = re.search(r'\{[\s\S]*\}', response_text) if json_match: try: result = json.loads(json_match.group()) skill_content = result.get("skill_content", response_text) explanation = result.get("explanation", "AI 生成的 Skill") suggested_category = result.get("suggested_category", request.category or "通用") suggested_tags = result.get("suggested_tags", request.tags or ["自定义"]) except json.JSONDecodeError: skill_content = response_text explanation = "AI 生成的 Skill" suggested_category = request.category or "通用" suggested_tags = request.tags or ["自定义"] else: skill_content = response_text explanation = "AI 生成的 Skill" suggested_category = request.category or "通用" suggested_tags = request.tags or ["自定义"] # 7. 生成预览 ID preview_id = str(uuid.uuid4()) # 8. 存储预览数据 _preview_storage[preview_id] = { "skill_name": request.skill_name, "skill_content": skill_content, "category": suggested_category, "tags": suggested_tags, "doc_urls": request.doc_urls, "github_repos": [r.model_dump() for r in request.github_repos], "uploaded_references": request.uploaded_references, "explanation": explanation, "created_at": str(datetime.now()) } # 9. 存储 references（后续保存时使用） references = {} for i, doc in enumerate(all_docs): filename = f"{doc['title'].replace('/', '-').replace(' ', '_')}.md" # 清理文件名 filename = ''.join(c for c in filename if c.isalnum() or c in '._-') references[filename] = doc["content"] _preview_references[preview_id] = references # 10. 生成文档摘要 doc_summary = "\n".join(doc_summary_parts) if doc_summary_parts else "无文档来源" return SkillGenerateFromDocsResponse( success=True, preview_id=preview_id, skill_content=skill_content, skill_name=request.skill_name, suggested_id=request.skill_name.lower().replace(" ", "-").replace("_", "-"), category=suggested_category, tags=suggested_tags, doc_summary=doc_summary, references_count=len(references), explanation=explanation ) except HTTPException: raise except Exception as e: logger.error(f"基于文档生成 Skill 失败: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"生成失败: {str(e)}" ) @router.post("/preview", response_model=SkillPreviewResponse) async def preview_skill( request: SkillPreviewRequest ): """ 预览 Skill（验证和解析） 返回： - 解析后的元数据 - 验证警告 - 字数统计 - 预估 token 数 """ try: # 解析 YAML frontmatter metadata = {} content = request.skill_content import yaml yaml_match = re.match(r'^---\n(.*?)\n---', content, re.DOTALL) if yaml_match: try: metadata = yaml.safe_load(yaml_match.group(1)) or {} except: pass # 验证警告 warnings = [] if not metadata.get("name"): warnings.append("缺少 name 字段") if not metadata.get("description"): warnings.append("缺少 description 字段") if len(content) < 100: warnings.append("内容过短，可能不够详细") if len(content) > 50000: warnings.append("内容过长，可能占用过多上下文") # 统计 word_count = len(content.split()) estimated_tokens = int(len(content) * 1.3) # 粗略估计 # 生成预览 ID preview_id = str(uuid.uuid4()) _preview_storage[preview_id] = { "skill_name": request.skill_name, "skill_content": content, "category": request.category, "tags": request.tags, "metadata": metadata, "warnings": warnings, "created_at": str(datetime.now()) } return SkillPreviewResponse( preview_id=preview_id, skill_content=content, parsed_metadata=metadata, validation_warnings=warnings, word_count=word_count, estimated_tokens=estimated_tokens ) except Exception as e: logger.error(f"预览 Skill 失败: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"预览失败: {str(e)}" ) @router.post("/refine", response_model=SkillRefineResponse) async def refine_skill_with_ai( request: SkillRefineRequest, glm_client: GLMClient = Depends(get_glm_client) ): """ 使用 AI 调整/优化 Skill 用户可以输入自然语言提示词，让 AI 修改 Skill 内容。 示例提示词： - "把行为指导改得更简洁" - "增加一个快速开始部分" - "添加更多代码示例" - "优化描述，让它更清晰" """ try: logger.info(f"AI 调整 Skill，提示词: {request.refinement_prompt[:50]}...") prompt = f"""请根据以下要求调整 SKILL.md 内容： **调整要求**： {request.refinement_prompt} **原始 SKILL.md**： {request.skill_content} 请返回完整的调整后的 SKILL.md 内容（不要包含其他解释）：""" response = await glm_client.chat( messages=[{"role": "user", "content": prompt}], temperature=request.temperature ) refined_content = response["choices"][0]["message"]["content"].strip() # 清理可能的 markdown 代码块标记 if refined_content.startswith("```"): lines = refined_content.split("\n") refined_content = "\n".join(lines[1:]) if refined_content.endswith("```"): refined_content = "\n".join(refined_content.split("\n")[:-1]) # 生成变更摘要 changes_summary = f"根据提示词「{request.refinement_prompt}」进行了调整" return SkillRefineResponse( success=True, refined_content=refined_content.strip(), changes_summary=changes_summary, original_length=len(request.skill_content), new_length=len(refined_content.strip()) ) except Exception as e: logger.error(f"AI 调整失败: {str(e)}") return SkillRefineResponse( success=False, refined_content=request.skill_content, changes_summary=f"调整失败: {str(e)}", original_length=len(request.skill_content), new_length=len(request.skill_content) ) @router.post("/save-from-preview", response_model=Skill) async def save_skill_from_preview( request: SkillSaveFromPreviewRequest, skill_manager: SkillManager = Depends(get_skill_manager) ): """ 从预览保存 Skill 到系统 保存流程： 1. 创建 Skill 目录结构 2. 保存 SKILL.md 3. 保存 references（如果有） 4. 注册到 SkillManager """ try: logger.info(f"从预览保存 Skill: {request.skill_id}") # 1. 创建 Skill skill_data = SkillCreate( id=request.skill_id, name=request.skill_id.replace("-", " ").title(), content=request.skill_content, category="通用", # 从 YAML frontmatter 解析 tags=[] # 从 YAML frontmatter 解析 ) skill = await skill_manager.create_user_skill(skill_data) # 2. 保存 references（如果有） if request.references: skill_path = skill_manager._find_skill_path(request.skill_id) if skill_path: refs_dir = skill_path.parent / "references" refs_dir.mkdir(exist_ok=True) for filename, content in request.references.items(): # 确保文件名安全 safe_filename = filename.replace("..", "").replace("/", "").replace("\\", "") if not safe_filename.endswith(".md"): safe_filename += ".md" ref_file = refs_dir / safe_filename ref_file.write_text(content, encoding="utf-8") logger.info(f"保存 reference: {request.skill_id}/references/{safe_filename}") # 3. 清理预览数据 if request.preview_id in _preview_storage: del _preview_storage[request.preview_id] if request.preview_id in _preview_references: del _preview_references[request.preview_id] return skill except Exception as e: logger.error(f"保存 Skill 失败: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"保存失败: {str(e)}" ) @router.get("/preview/{preview_id}", response_model=dict) async def get_preview(preview_id: str): """ 获取预览数据 """ if preview_id not in _preview_storage: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"预览不存在: {preview_id}" ) preview_data = _preview_storage[preview_id] references = _preview_references.get(preview_id, {}) return { "preview": preview_data, "references": references, "references_count": len(references) } @router.delete("/preview/{preview_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_preview(preview_id: str): """ 删除预览数据 """ if preview_id in _preview_storage: del _preview_storage[preview_id] if preview_id in _preview_references: del _preview_references[preview_id] return None @router.put("/{skill_id}/with-references", response_model=Skill) async def update_skill_with_references( skill_id: str, request: SkillSaveFromPreviewRequest, skill_manager: SkillManager = Depends(get_skill_manager) ): """ 更新 Skill 及其 References 用于编辑模式，同时更新 SKILL.md 和 references/ 目录中的文件 """ try: logger.info(f"更新 Skill 及 references: {skill_id}") # 1. 更新 SKILL.md skill_update = SkillUpdate(content=request.skill_content) skill = await skill_manager.update_user_skill(skill_id, skill_update) if not skill: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Skill 不存在: {skill_id}" ) # 2. 更新 references（如果有） if request.references is not None: # 允许传入空字典清空references skill_path = skill_manager._find_skill_path(skill_id) if skill_path: refs_dir = skill_path.parent / "references" # 如果传入空字典，删除整个references目录 if len(request.references) == 0: if refs_dir.exists(): import shutil shutil.rmtree(refs_dir) logger.info(f"清空 references: {skill_id}/references/") else: # 确保目录存在 refs_dir.mkdir(exist_ok=True) # 清空现有文件 for existing_file in refs_dir.iterdir(): if existing_file.is_file(): existing_file.unlink() # 写入新文件 for filename, content in request.references.items(): safe_filename = filename.replace("..", "").replace("/", "").replace("\\", "") if not safe_filename.endswith(".md"): safe_filename += ".md" ref_file = refs_dir / safe_filename ref_file.write_text(content, encoding="utf-8") logger.info(f"更新 reference: {skill_id}/references/{safe_filename}") # 清除缓存 if skill_id in skill_manager._cache: del skill_manager._cache[skill_id] # 重新加载并返回 return await skill_manager.load_skill(skill_id) except HTTPException: raise except Exception as e: logger.error(f"更新 Skill 失败: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"更新失败: {str(e)}" )