""" Skill 管理 API 路由 提供: 1. Skill 的 CRUD 操作 2. Agent-Skill 生命周期端点 3. AI 辅助 Skill 创建(完整流程) 4. Skill 选择和路由(LLM 与 Skills 结合) 5. Agent 工作流配置 """ 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, ) 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""" try: 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