""" AI 辅助生成 API 提供 AI 辅助生成人物、大纲、解析剧本等功能 支持 Skills 融入:将选定的 Skills 行为指导融入 LLM 调用 支持长剧本分段分析:自动切分并合并结果 """ from fastapi import APIRouter, HTTPException from typing import Dict, Any, List, Optional from pydantic import BaseModel import asyncio from app.core.llm.glm_client import get_glm_client from app.core.skills.skill_manager import get_skill_manager from app.utils.logger import get_logger from app.utils.script_splitter import split_script, ScriptSplitter logger = get_logger(__name__) router = APIRouter(prefix="/ai-assistant", tags=["AI 辅助生成"]) class SkillInfo(BaseModel): """Skill 信息(由前端传递)""" id: str name: str behavior: str # behavior_guide 内容 class GenerateCharactersRequest(BaseModel): """生成人物设定请求""" idea: str # 用户的初步想法或故事框架 projectName: Optional[str] = None totalEpisodes: Optional[int] = None skills: Optional[List[SkillInfo]] = None # 新增:Skills 列表 customPrompt: Optional[str] = None # 自定义提示词 class GenerateOutlineRequest(BaseModel): """生成大纲请求""" idea: str totalEpisodes: int = 30 genre: str = "古风" projectName: Optional[str] = None skills: Optional[List[SkillInfo]] = None # 新增:Skills 列表 customPrompt: Optional[str] = None # 自定义提示词 class ParseScriptRequest(BaseModel): """解析剧本请求""" content: str extractCharacters: bool = True extractOutline: bool = True skills: Optional[List[SkillInfo]] = None # 新增:Skills 列表 use_llm: bool = True # 新增:是否使用 LLM 分析(默认 True) customPrompt: Optional[str] = None # 新增:自定义提示词 class GenerateWorldRequest(BaseModel): """生成世界观请求""" idea: str projectName: Optional[str] = None genre: Optional[str] = "古风" skills: Optional[List[SkillInfo]] = None # Skills 列表 customPrompt: Optional[str] = None # 自定义提示词 class GenerateWorldFromScriptRequest(BaseModel): """从剧本分析世界观请求""" script: str projectName: Optional[str] = None skills: Optional[List[SkillInfo]] = None # Skills 列表 customPrompt: Optional[str] = None # 自定义提示词 # ============================================================================ # Skills 融入辅助函数 # ============================================================================ async def build_enhanced_system_prompt( base_role: str, skills: Optional[List[SkillInfo]] = None, skill_manager=None ) -> str: """ 构建融入 Skills 的增强 System Prompt Args: base_role: 基础角色描述(如 "你是剧集创作专家") skills: Skills 列表 skill_manager: Skill Manager 实例(用于加载参考文件) Returns: 增强后的 System Prompt """ prompt_parts = [base_role] if skills and len(skills) > 0: prompt_parts.append("\n## 应用技能指导") # 添加每个 Skill 的行为指导 for skill in skills: prompt_parts.append(f"\n### {skill.name}") prompt_parts.append(skill.behavior) # 尝试加载参考文件(如果有 skill_manager) if skill_manager: prompt_parts.append("\n## 参考资料") for skill in skills: try: references = await skill_manager.load_skill_references(skill.id) if references: prompt_parts.append(f"\n### 来自 {skill.name} 的参考文件:") for filename, content in references.items(): # 截取过长内容 if len(content) > 2000: content = content[:2000] + "\n...(内容过长,已截断)" prompt_parts.append(f"\n#### {filename}\n{content}") except Exception as e: logger.warning(f"加载 Skill {skill.id} 参考文件失败: {str(e)}") return "\n".join(prompt_parts) @router.post("/generate/characters") async def generate_characters(request: GenerateCharactersRequest): """ AI 辅助生成人物设定 根据用户的初步想法,使用 AI 生成完整的人物设定 支持融入 Skills 的行为指导 """ try: glm_client = get_glm_client() skill_manager = get_skill_manager() # 构建增强的 System Prompt(融入 Skills) base_role = "你是专业的剧集创作专家,擅长创作丰富立体的人物角色。" system_prompt = await build_enhanced_system_prompt( base_role=base_role, skills=request.skills, skill_manager=skill_manager ) # 构建用户提示 extra_info = "" if request.projectName: extra_info += f"\n项目名称:{request.projectName}" if request.totalEpisodes: extra_info += f"\n总集数:{request.totalEpisodes}" # 自定义提示词 custom_requirements = "" if request.customPrompt: custom_requirements = f"\n【用户自定义要求】\n{request.customPrompt}\n" prompt = f"""请根据以下想法生成 3-5 个主要人物设定: 用户想法:{request.idea}{extra_info}{custom_requirements} 要求: 1. 每个人物包含:姓名、身份、性格、说话风格、背景故事 2. 人物之间要有关系冲突 3. 每个人物 50-100 字 4. 格式:姓名:身份 - 性格 - 说话风格 - 背景故事 5. 严格遵守上面【应用技能指导】中的要求 请按以下格式输出: 【人物1】 姓名:xxx 身份:xxx 性格:xxx 说话风格:xxx 背景故事:xxx 【人物2】 ... """ logger.info(f"生成人物设定,使用 {len(request.skills) if request.skills else 0} 个 Skills") response = await glm_client.chat( messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": prompt} ], temperature=0.7 ) content = response["choices"][0]["message"]["content"] return { "success": True, "characters": content, "usage": response.get("usage") } except Exception as e: logger.error(f"生成人物失败: {str(e)}") raise HTTPException(status_code=500, detail=f"生成失败: {str(e)}") @router.post("/generate/outline") async def generate_outline(request: GenerateOutlineRequest): """ AI 辅助生成大纲 根据用户想法生成完整的剧集大纲 支持融入 Skills 的行为指导 """ try: glm_client = get_glm_client() skill_manager = get_skill_manager() # 构建增强的 System Prompt(融入 Skills) base_role = "你是专业的剧集创作专家,擅长构建引人入胜的剧情结构和故事节奏。" system_prompt = await build_enhanced_system_prompt( base_role=base_role, skills=request.skills, skill_manager=skill_manager ) # 构建用户提示 # 自定义提示词 custom_requirements = "" if request.customPrompt: custom_requirements = f"\n【用户自定义要求】\n{request.customPrompt}\n" prompt = f"""请根据以下想法生成完整的剧集大纲: 用户想法:{request.idea} 总集数:{request.totalEpisodes} 类型:{request.genre} {f'项目名称:{request.projectName}' if request.projectName else ''}{custom_requirements} 要求: 1. 将故事分为 4-5 个阶段 2. 每个阶段包含具体的集数范围 3. 标注每个阶段的关键事件和转折点 4. 字数 200-400 字 5. 严格遵守上面【应用技能指导】中的要求 请按以下格式输出: 【阶段1】EPxx-EPxx:阶段名称 内容概要... 【阶段2】EPxx-EPxx:阶段名称 ... """ logger.info(f"生成大纲,使用 {len(request.skills) if request.skills else 0} 个 Skills") response = await glm_client.chat( messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": prompt} ], temperature=0.7 ) content = response["choices"][0]["message"]["content"] return { "success": True, "outline": content, "usage": response.get("usage") } except Exception as e: logger.error(f"生成大纲失败: {str(e)}") raise HTTPException(status_code=500, detail=f"生成失败: {str(e)}") @router.post("/generate/world") async def generate_world(request: GenerateWorldRequest): """ AI 辅助生成世界观设定 根据用户想法生成世界观设定 支持融入 Skills 的行为指导 支持自定义提示词 """ try: glm_client = get_glm_client() skill_manager = get_skill_manager() # 构建增强的 System Prompt(融入 Skills) base_role = "你是专业的世界观设定专家,擅长构建架空世界的背景设定。" system_prompt = await build_enhanced_system_prompt( base_role=base_role, skills=request.skills, skill_manager=skill_manager ) # 构建用户提示 # 自定义提示词 custom_requirements = "" if request.customPrompt: custom_requirements = f"\n【用户自定义要求】\n{request.customPrompt}\n" prompt = f"""请根据以下想法生成世界观设定: 用户想法:{request.idea} 类型:{request.genre} {f'项目名称:{request.projectName}' if request.projectName else ''}{custom_requirements} 要求: 1. 描述时代背景(朝代、架空世界等) 2. 描述地理环境和主要场景 3. 描述社会结构(权力体系、阶级关系) 4. 描述文化特色(习俗、服饰、语言等) 5. 字数 200-500 字 6. 严格遵守上面【应用技能指导】中的要求 请输出详细的世界观设定: """ logger.info(f"生成世界观设定,使用 {len(request.skills) if request.skills else 0} 个 Skills") response = await glm_client.chat( messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": prompt} ], temperature=0.7 ) content = response["choices"][0]["message"]["content"] return { "success": True, "worldSetting": content, "usage": response.get("usage") } except Exception as e: logger.error(f"生成世界观设定失败: {str(e)}") raise HTTPException(status_code=500, detail=f"生成失败: {str(e)}") @router.post("/generate/world-from-script") async def generate_world_from_script(request: GenerateWorldFromScriptRequest): """ AI 辅助从剧本分析世界观 从剧本中提取和分析世界观设定 支持融入 Skills 的行为指导 支持自定义提示词 """ try: glm_client = get_glm_client() skill_manager = get_skill_manager() # 构建增强的 System Prompt(融入 Skills) base_role = """你是专业的世界观分析专家,擅长从剧本中提取和分析世界观设定。 你能识别人物关系、时代背景、社会结构、地理环境等深层信息。""" system_prompt = await build_enhanced_system_prompt( base_role=base_role, skills=request.skills, skill_manager=skill_manager ) # 构建用户提示 # 自定义提示词 custom_requirements = "" if request.customPrompt: custom_requirements = f"\n【用户自定义要求】\n{request.customPrompt}\n" prompt = f"""请分析以下剧本,提取世界观设定: {f'项目名称:{request.projectName}' if request.projectName else ''}{custom_requirements} 剧本内容: {request.script[:5000]} 要求: 1. 识别时代背景(朝代、架空世界等) 2. 提取地理环境和主要场景信息 3. 分析社会结构(权力体系、阶级关系) 4. 识别文化特色(习俗、服饰、语言等) 5. 输出结构化的世界观分析 请输出世界观分析: """ logger.info(f"从剧本分析世界观,使用 {len(request.skills) if request.skills else 0} 个 Skills") response = await glm_client.chat( messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": prompt} ], temperature=0.3 ) content = response["choices"][0]["message"]["content"] return { "success": True, "worldSetting": content, "usage": response.get("usage") } except Exception as e: logger.error(f"从剧本分析世界观失败: {str(e)}") raise HTTPException(status_code=500, detail=f"分析失败: {str(e)}") @router.post("/parse/script") async def parse_script(request: ParseScriptRequest): """ 解析上传的剧本文件 提取人物、大纲等关键信息 支持两种模式: 1. LLM 智能分析模式(use_llm=True,默认):使用 LLM 深入分析,支持 Skills 融入 2. 快速正则模式(use_llm=False):使用正则表达式快速提取 """ try: # 如果用户选择使用 LLM 分析 if request.use_llm: return await _parse_script_with_llm(request) # 否则使用快速正则模式 return await _parse_script_with_regex(request) except Exception as e: logger.error(f"解析剧本失败: {str(e)}") import traceback traceback.print_exc() raise HTTPException(status_code=500, detail=f"解析失败: {str(e)}") async def _parse_script_with_llm(request: ParseScriptRequest) -> Dict[str, Any]: """ 使用 LLM 智能分析剧本,支持 Skills 融入 支持长剧本分段分析和结果合并 """ glm_client = get_glm_client() skill_manager = get_skill_manager() # 检查内容长度,决定是否需要分段 content = request.content content_length = len(content) # 阈值:超过8000字符时使用分段分析 SPLIT_THRESHOLD = 8000 if content_length <= SPLIT_THRESHOLD: # 内容较短,直接分析 return await _analyze_single_segment( content=content, request=request, glm_client=glm_client, skill_manager=skill_manager ) else: # 内容较长,使用分段分析 logger.info(f"剧本较长 ({content_length} 字符),启用分段分析模式") return await _analyze_with_splitting( content=content, request=request, glm_client=glm_client, skill_manager=skill_manager ) async def _analyze_single_segment( content: str, request: ParseScriptRequest, glm_client, skill_manager ) -> Dict[str, Any]: """分析单个片段(原有逻辑)""" # 构建增强的 System Prompt(融入 Skills) base_role = """你是专业的剧本分析专家,擅长从剧本中提取人物关系、剧情结构、对话风格等关键信息。 你能识别人物的出场频率、人物关系、情感走向、对话特点等深层次信息。""" system_prompt = await build_enhanced_system_prompt( base_role=base_role, skills=request.skills, skill_manager=skill_manager ) # 构建分析提示 analysis_requirements = [] if request.extractCharacters: analysis_requirements.append(""" 1. 人物分析: - 识别所有出场人物 - 统计每个人的出场次数/对话次数 - 分析人物关系(上下级、敌对、盟友等) - 提取人物性格特点和说话风格 - 按重要性排序输出""") if request.extractOutline: analysis_requirements.append(""" 2. 剧情大纲: - 识别主要剧情阶段 - 提取关键转折点 - 分析故事结构(起承转合) - 总结核心冲突""") # 自定义提示词 custom_requirements = "" if request.customPrompt: custom_requirements = f"\n【用户自定义要求】\n{request.customPrompt}\n" prompt = f"""请分析以下剧本内容,提取关键信息: {content} 要求: {''.join(analysis_requirements)} 3. 严格遵守上面【应用技能指导】中的分析要求{custom_requirements} 请以结构化的格式输出,便于后续处理。 """ logger.info(f"使用 LLM 分析剧本 ({len(content)} 字符),使用 {len(request.skills) if request.skills else 0} 个 Skills") response = await glm_client.chat( messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": prompt} ], temperature=0.3 # 使用较低温度确保分析准确性 ) analysis_result = response["choices"][0]["message"]["content"] return { "success": True, "method": "llm", "analysis": analysis_result, "characters": [], "outline": "", "summary": f"LLM 智能分析完成 ({len(content)} 字符),使用 {len(request.skills) if request.skills else 0} 个 Skills", "usage": response.get("usage") } async def _analyze_with_splitting( content: str, request: ParseScriptRequest, glm_client, skill_manager ) -> Dict[str, Any]: """分段分析长剧本并合并结果""" # 1. 切分剧本 split_result = split_script(content) segments = split_result["segments"] summary = split_result["summary"] logger.info(f"剧本已切分为 {len(segments)} 个片段: {summary}") # 2. 并行分析各片段 segment_analyses = await _analyze_segments_parallel( segments=segments, request=request, glm_client=glm_client, skill_manager=skill_manager ) # 3. 合并分析结果 merged_result = await _merge_analysis_results( segment_analyses=segment_analyses, request=request, glm_client=glm_client, skill_manager=skill_manager, split_summary=summary ) return merged_result async def _analyze_segments_parallel( segments: List[Dict[str, Any]], request: ParseScriptRequest, glm_client, skill_manager ) -> List[Dict[str, Any]]: """并行分析多个片段""" # 构建基础系统提示词 base_role = """你是专业的剧本分析专家,擅长从剧本中提取人物关系、剧情结构、对话风格等关键信息。 这是长剧本的分段分析,请专注于当前片段的内容。""" system_prompt = await build_enhanced_system_prompt( base_role=base_role, skills=request.skills, skill_manager=skill_manager ) # 构建分析要求 analysis_requirements = [] if request.extractCharacters: analysis_requirements.append(""" 1. 人物分析: - 识别当前片段中的所有出场人物 - 统计每个人物的出场次数/对话次数 - 分析人物关系(上下级、敌对、盟友等) - 提取人物性格特点和说话风格 - 按重要性排序输出""") if request.extractOutline: analysis_requirements.append(""" 2. 剧情大纲: - 识别当前片段的剧情阶段 - 提取关键转折点 - 分析情节进展""") # 为每个片段创建分析任务 # 自定义提示词 custom_requirements = "" if request.customPrompt: custom_requirements = f"\n【用户自定义要求】\n{request.customPrompt}\n" async def analyze_segment(segment: Dict[str, Any]) -> Dict[str, Any]: segment_index = segment["index"] segment_content = segment["content"] scene_marker = segment.get("scene_marker", "") prompt = f"""请分析以下剧本片段(片段 {segment_index + 1}/{len(segments)}),提取关键信息: {f"场景标记:{scene_marker}" if scene_marker else ""} 片段内容: {segment_content} 要求: {''.join(analysis_requirements)} 3. 严格遵守上面【应用技能指导】中的分析要求 4. 输出结构化的分析结果,便于后续合并{custom_requirements} 请以结构化的格式输出。 """ try: response = await glm_client.chat( messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": prompt} ], temperature=0.3 ) return { "segment_index": segment_index, "scene_marker": scene_marker, "analysis": response["choices"][0]["message"]["content"], "usage": response.get("usage"), "success": True } except Exception as e: logger.error(f"分析片段 {segment_index} 失败: {str(e)}") return { "segment_index": segment_index, "scene_marker": scene_marker, "analysis": f"分析失败: {str(e)}", "success": False } # 并行执行所有片段分析(限制并发数为3) semaphore = asyncio.Semaphore(3) async def analyze_with_semaphore(segment): async with semaphore: return await analyze_segment(segment) results = await asyncio.gather( *[analyze_with_semaphore(seg) for seg in segments], return_exceptions=True ) # 过滤异常结果 valid_results = [r for r in results if isinstance(r, dict)] logger.info(f"并行分析完成:{len(valid_results)}/{len(segments)} 个片段成功") return valid_results async def _merge_analysis_results( segment_analyses: List[Dict[str, Any]], request: ParseScriptRequest, glm_client, skill_manager, split_summary: Dict[str, Any] ) -> Dict[str, Any]: """合并多个片段的分析结果""" # 收集所有片段的分析 all_analyses = [] total_usage = {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0} for analysis in segment_analyses: if analysis.get("success"): all_analyses.append(f""" --- 片段 {analysis['segment_index'] + 1}{f"({analysis['scene_marker']})" if analysis['scene_marker'] else ""}: {analysis['analysis']} --- """) # 累计 token 使用量 if analysis.get("usage"): for key in ["prompt_tokens", "completion_tokens", "total_tokens"]: total_usage[key] += analysis["usage"].get(key, 0) if not all_analyses: return { "success": False, "method": "llm_batch", "error": "所有片段分析均失败", "summary": "分段分析失败" } # 使用 LLM 合并和去重 base_role = """你是专业的剧本分析专家,擅长整合和总结多源信息。 你的任务是将多个片段的分析结果合并成一个完整的、去重的、结构化的分析报告。""" system_prompt = await build_enhanced_system_prompt( base_role=base_role, skills=request.skills, skill_manager=skill_manager ) merge_requirements = [] if request.extractCharacters: merge_requirements.append(""" 1. 人物合并: - 将各片段中的人物列表合并去重 - 累加每个人物的出场次数 - 整合人物关系分析 - 输出完整的人物列表""") if request.extractOutline: merge_requirements.append(""" 2. 大纲合并: - 将各片段的剧情大纲按时间/逻辑顺序整合 - 形成完整的剧情结构 - 标注关键转折点""") # 自定义提示词 custom_requirements = "" if request.customPrompt: custom_requirements = f"\n【用户自定义要求】\n{request.customPrompt}\n" merge_prompt = f"""以下是将长剧本切分后各片段的分析结果: {''.join(all_analyses)} 请合并以上分析,输出一个完整的、去重的分析报告。 要求: {''.join(merge_requirements)} 3. 保持结构化输出格式 4. 严格遵守上面【应用技能指导】中的分析要求{custom_requirements} 请输出合并后的完整分析报告。 """ logger.info("开始合并各片段分析结果...") try: response = await glm_client.chat( messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": merge_prompt} ], temperature=0.3 ) merged_analysis = response["choices"][0]["message"]["content"] # 加上合并步骤的 token 使用量 if response.get("usage"): for key in ["prompt_tokens", "completion_tokens", "total_tokens"]: total_usage[key] += response["usage"].get(key, 0) return { "success": True, "method": "llm_batch", "analysis": merged_analysis, "characters": [], "outline": "", "summary": f"分段分析完成:共 {len(segment_analyses)} 个片段,{split_summary.get('split_methods', {})}", "split_info": split_summary, "segment_count": len(segment_analyses), "usage": total_usage } except Exception as e: logger.error(f"合并分析结果失败: {str(e)}") # 如果合并失败,返回原始的片段分析 return { "success": True, "method": "llm_batch", "analysis": "\n\n".join(all_analyses), "characters": [], "outline": "", "summary": f"分段分析完成(合并失败,返回原始分析):共 {len(segment_analyses)} 个片段", "split_info": split_summary, "segment_count": len(segment_analyses), "usage": total_usage, "merge_error": str(e) } async def _parse_script_with_regex(request: ParseScriptRequest) -> Dict[str, Any]: """ 使用正则表达式快速提取剧本信息(不使用 LLM,不消耗 token) """ result: Dict[str, Any] = { "success": True, "method": "regex", "characters": [], "outline": "", "scenes": [], "summary": "" } content = request.content # 提取人物(简单实现) if request.extractCharacters: import re # 修复正则表达式,使用更健壮的模式 # 支持多种格式:【人物名】、【人物名】:等 dialogue_pattern = r'【([^】\n]+)[:】?\s*' matches = re.findall(dialogue_pattern, content) # 统计每个人物出现的次数 character_counts = {} for match in matches: name = match.strip() if name: character_counts[name] = character_counts.get(name, 0) + 1 # 转换为结果格式 result["characters"] = [ {"name": name, "lines": count} for name, count in sorted(character_counts.items(), key=lambda x: x[1], reverse=True) ] result["summary"] = f"共识别 {len(result['characters'])} 个人物(正则模式)" # 提取大纲(简单实现) if request.extractOutline: # 按场景或集数分割 import re scene_patterns = [ r'第[一二三四五六七八九十百千0-9]+[集场幕]', r'【[场景场] [^\n]*】', r'Scene\s*\d+', ] scenes = [] for pattern in scene_patterns: found = re.split(pattern, content) scenes.extend(found[1:]) # 跳过第一个元素 if scenes: result["scenes"] = [s.strip()[:100] for s in scenes if s.strip()] if not result["summary"]: result["summary"] = f"共识别 {len(result['scenes'])} 个场景(正则模式)" result["contentLength"] = len(content) return result @router.get("/available-skills") async def get_available_skills(): """ 获取可用的 Skills 列表,按分类展示 """ try: skill_manager = get_skill_manager() skills = await skill_manager.list_skills() # 按分类组织 by_category: Dict[str, List[Dict]] = {} for skill in skills: if skill.category not in by_category: by_category[skill.category] = [] by_category[skill.category].append({ "id": skill.id, "name": skill.name, "type": skill.type, "description": skill.behavior_guide[:100] + "..." if len(skill.behavior_guide) > 100 else skill.behavior_guide }) return { "categories": list(by_category.keys()), "skills": by_category } except Exception as e: logger.error(f"获取 Skills 失败: {str(e)}") raise HTTPException(status_code=500, detail=f"获取失败: {str(e)}")