971 lines
31 KiB
Python
971 lines
31 KiB
Python
"""
|
||
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)}")
|
||
|
||
|
||
class OptimizeEpisodeRequest(BaseModel):
|
||
"""优化剧集请求"""
|
||
projectId: str
|
||
episodeNumber: int
|
||
content: str
|
||
skills: Optional[List[SkillInfo]] = None
|
||
customPrompt: Optional[str] = None
|
||
|
||
|
||
@router.post("/optimize-episode")
|
||
async def optimize_episode(request: OptimizeEpisodeRequest):
|
||
"""
|
||
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"
|
||
|
||
user_prompt = f"""请优化以下剧集内容：
|
||
|
||
项目 ID: {request.projectId}
|
||
集数: EP{request.episodeNumber}
|
||
{custom_requirements}
|
||
【剧集内容】
|
||
{request.content[:8000]}
|
||
|
||
【优化要求】
|
||
1. 保持原有的故事结构和情节走向
|
||
2. 改进对话，使其更符合人物性格
|
||
3. 增强场景描写的画面感
|
||
4. 优化叙事节奏
|
||
5. 严格遵守上面【应用技能指导】中的优化要求
|
||
|
||
请直接输出优化后的剧集内容，不要添加任何解释或说明。
|
||
"""
|
||
|
||
logger.info(f"优化剧集 EP{request.episodeNumber}，使用 {len(request.skills) if request.skills else 0} 个 Skills")
|
||
|
||
response = await glm_client.chat(
|
||
messages=[
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": user_prompt}
|
||
],
|
||
temperature=0.7
|
||
)
|
||
|
||
optimized_content = response["choices"][0]["message"]["content"]
|
||
|
||
return {
|
||
"success": True,
|
||
"optimizedContent": optimized_content,
|
||
"usage": response.get("usage")
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"优化剧集失败: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"优化失败: {str(e)}")
|