admin/MuMuAINovel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feature:新增大纲续写-智能生成组织功能，自动添加组织成员

Browse Source
This commit is contained in:
xiamuceer
2026-01-05 14:27:27 +08:00
parent 6e603ee1a9
commit ba7ee591b6
6 changed files with 1986 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/services/auto_organization_service.py
+504
View File
@@ -0,0 +1,504 @@
"""自动组织引入服务 - 在续写大纲时根据剧情推进自动引入新组织"""
from typing import List, Dict, Any, Optional
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
import json
from app.models.character import Character
from app.models.relationship import Organization, OrganizationMember
from app.models.project import Project
from app.services.ai_service import AIService
from app.services.prompt_service import PromptService
from app.logger import get_logger
logger = get_logger(__name__)
class AutoOrganizationService:
"""自动组织引入服务"""
def __init__(self, ai_service: AIService):
self.ai_service = ai_service
async def analyze_and_create_organizations(
self,
project_id: str,
outline_content: str,
existing_characters: List[Character],
existing_organizations: List[Dict[str, Any]],
db: AsyncSession,
user_id: str = None,
enable_mcp: bool = True,
all_chapters_brief: str = "",
start_chapter: int = 1,
chapter_count: int = 3,
plot_stage: str = "发展",
story_direction: str = "继续推进主线剧情",
preview_only: bool = False
) -> Dict[str, Any]:
"""
预测性分析并创建需要的新组织
Args:
project_id: 项目ID
outline_content: 当前批次大纲内容（用于向后兼容，实际不使用）
existing_characters: 现有角色列表
existing_organizations: 现有组织列表
db: 数据库会话
user_id: 用户ID(用于MCP和自定义提示词)
enable_mcp: 是否启用MCP增强
all_chapters_brief: 已有章节概览
start_chapter: 起始章节号
chapter_count: 续写章节数
plot_stage: 剧情阶段
story_direction: 故事发展方向
preview_only: 仅预测不创建（用于组织确认机制）
Returns:
{
"new_organizations": [组织对象列表], # preview_only=True时为空
"members_created": [成员关系列表], # preview_only=True时为空
"organization_count": 新增组织数量,
"analysis_result": AI分析结果,
"predicted_organizations": [预测的组织数据] # 仅preview_only=True时返回
"needs_new_organizations": bool,
"reason": str
}
"""
logger.info(f"🏛️ 【组织引入】预测性分析：检测是否需要引入新组织...")
logger.info(f" - 项目ID: {project_id}")
logger.info(f" - 续写计划: 第{start_chapter}章起，共{chapter_count}章")
logger.info(f" - 剧情阶段: {plot_stage}")
logger.info(f" - 发展方向: {story_direction}")
logger.info(f" - 现有角色数: {len(existing_characters)}")
logger.info(f" - 现有组织数: {len(existing_organizations)}")
# 1. 获取项目信息
project_result = await db.execute(
select(Project).where(Project.id == project_id)
)
project = project_result.scalar_one_or_none()
if not project:
raise ValueError("项目不存在")
# 2. 构建现有组织信息摘要
existing_orgs_summary = self._build_organization_summary(existing_organizations)
existing_chars_summary = self._build_character_summary(existing_characters)
# 3. AI预测性分析是否需要新组织
analysis_result = await self._analyze_organization_needs(
project=project,
outline_content=outline_content,
existing_orgs_summary=existing_orgs_summary,
existing_chars_summary=existing_chars_summary,
db=db,
user_id=user_id,
enable_mcp=enable_mcp,
all_chapters_brief=all_chapters_brief,
start_chapter=start_chapter,
chapter_count=chapter_count,
plot_stage=plot_stage,
story_direction=story_direction
)
# 4. 判断是否需要创建组织
if not analysis_result or not analysis_result.get("needs_new_organizations"):
logger.info("✅ AI判断：当前剧情不需要引入新组织")
return {
"new_organizations": [],
"members_created": [],
"organization_count": 0,
"analysis_result": analysis_result,
"predicted_organizations": [],
"needs_new_organizations": False,
"reason": analysis_result.get("reason", "当前剧情不需要新组织")
}
# 5. 如果是预览模式，仅返回预测结果，不创建组织
if preview_only:
organization_specs = analysis_result.get("organization_specifications", [])
logger.info(f"🔮 预览模式：预测到 {len(organization_specs)} 个组织，不创建数据库记录")
return {
"new_organizations": [],
"members_created": [],
"organization_count": 0,
"analysis_result": analysis_result,
"predicted_organizations": organization_specs,
"needs_new_organizations": True,
"reason": analysis_result.get("reason", "预测需要新组织")
}
# 6. 批量生成新组织（非预览模式）
new_organizations = []
members_created = []
organization_specs = analysis_result.get("organization_specifications", [])
logger.info(f"🎯 AI建议引入 {len(organization_specs)} 个新组织")
for idx, spec in enumerate(organization_specs):
try:
spec_name = spec.get('name', spec.get('organization_description', '未命名'))
logger.info(f" [{idx+1}/{len(organization_specs)}] 生成组织规格: {spec_name}")
logger.debug(f" 组织规格内容: {json.dumps(spec, ensure_ascii=False)}")
# 生成组织详细信息
organization_data = await self._generate_organization_details(
spec=spec,
project=project,
existing_characters=existing_characters,
existing_organizations=existing_organizations,
db=db,
user_id=user_id,
enable_mcp=enable_mcp
)
logger.debug(f" AI生成的组织数据: {json.dumps(organization_data, ensure_ascii=False)[:200]}")
# 创建组织记录（先创建Character记录，再创建Organization记录）
character, organization = await self._create_organization_record(
project_id=project_id,
organization_data=organization_data,
db=db
)
new_organizations.append({
"character": character,
"organization": organization
})
logger.info(f" ✅ 创建新组织: {character.name}, ID: {organization.id}")
# 建立成员关系
members_data = organization_data.get("initial_members", [])
if members_data:
logger.info(f" 🔗 开始创建 {len(members_data)} 个成员关系...")
members = await self._create_member_relationships(
organization=organization,
member_specs=members_data,
existing_characters=existing_characters,
project_id=project_id,
db=db
)
members_created.extend(members)
logger.info(f" ✅ 实际创建了 {len(members)} 个成员关系记录")
except Exception as e:
logger.error(f" ❌ 创建组织失败: {e}", exc_info=True)
continue
# 7. 提交事务（注意：这里只flush，让调用方commit）
await db.flush()
logger.info(f"🎉 自动组织引入完成: 新增{len(new_organizations)}个组织, {len(members_created)}个成员关系")
return {
"new_organizations": new_organizations,
"members_created": members_created,
"organization_count": len(new_organizations),
"analysis_result": analysis_result,
"predicted_organizations": [],
"needs_new_organizations": True,
"reason": analysis_result.get("reason", "")
}
def _build_organization_summary(self, organizations: List[Dict[str, Any]]) -> str:
"""构建现有组织摘要"""
if not organizations:
return "暂无组织"
summary = []
for org in organizations:
org_name = org.get("name", "未知")
org_type = org.get("organization_type", "未知类型")
power_level = org.get("power_level", 50)
purpose = (org.get("organization_purpose") or "")[:50]
summary.append(f"- {org_name} ({org_type}, 势力等级:{power_level}): {purpose}")
return "\n".join(summary[:15]) # 最多显示15个
def _build_character_summary(self, characters: List[Character]) -> str:
"""构建现有角色摘要"""
if not characters:
return "暂无角色"
summary = []
for char in characters:
if not char.is_organization: # 只统计非组织角色
char_role = char.role_type or "未知"
personality = (char.personality or "")[:30]
summary.append(f"- {char.name} ({char_role}): {personality}")
return "\n".join(summary[:20]) # 最多显示20个
async def _analyze_organization_needs(
self,
project: Project,
outline_content: str,
existing_orgs_summary: str,
existing_chars_summary: str,
db: AsyncSession,
user_id: str,
enable_mcp: bool,
all_chapters_brief: str = "",
start_chapter: int = 1,
chapter_count: int = 3,
plot_stage: str = "发展",
story_direction: str = "继续推进主线剧情"
) -> Dict[str, Any]:
"""AI预测性分析是否需要新组织"""
# 构建分析提示词
template = await PromptService.get_template(
"AUTO_ORGANIZATION_ANALYSIS",
user_id,
db
)
# 使用新的预测性分析参数
prompt = PromptService.format_prompt(
template,
title=project.title,
theme=project.theme or "未设定",
genre=project.genre or "未设定",
time_period=project.world_time_period or "未设定",
location=project.world_location or "未设定",
atmosphere=project.world_atmosphere or "未设定",
existing_organizations=existing_orgs_summary,
existing_characters=existing_chars_summary,
all_chapters_brief=all_chapters_brief,
start_chapter=start_chapter,
chapter_count=chapter_count,
plot_stage=plot_stage,
story_direction=story_direction
)
try:
# 调用AI分析（使用统一的JSON调用方法）
if enable_mcp and user_id:
result = await self.ai_service.generate_text_with_mcp(
prompt=prompt,
user_id=user_id,
db_session=db,
enable_mcp=True,
max_tool_rounds=2
)
content = result.get("content", "")
# 使用统一的JSON清洗方法
cleaned = self.ai_service._clean_json_response(content)
analysis = json.loads(cleaned)
else:
# 非MCP调用：使用带自动重试的JSON调用
analysis = await self.ai_service.call_with_json_retry(
prompt=prompt,
max_retries=3
)
logger.info(f" ✅ AI分析完成: needs_new_organizations={analysis.get('needs_new_organizations')}")
return analysis
except json.JSONDecodeError as e:
logger.error(f" ❌ 组织需求分析JSON解析失败: {e}")
return {"needs_new_organizations": False}
except Exception as e:
logger.error(f" ❌ 组织需求分析失败: {e}")
return {"needs_new_organizations": False}
async def _generate_organization_details(
self,
spec: Dict[str, Any],
project: Project,
existing_characters: List[Character],
existing_organizations: List[Dict[str, Any]],
db: AsyncSession,
user_id: str,
enable_mcp: bool
) -> Dict[str, Any]:
"""生成组织详细信息"""
# 构建组织生成提示词
template = await PromptService.get_template(
"AUTO_ORGANIZATION_GENERATION",
user_id,
db
)
existing_orgs_summary = self._build_organization_summary(existing_organizations)
existing_chars_summary = self._build_character_summary(existing_characters)
prompt = PromptService.format_prompt(
template,
title=project.title,
genre=project.genre or "未设定",
theme=project.theme or "未设定",
time_period=project.world_time_period or "未设定",
location=project.world_location or "未设定",
atmosphere=project.world_atmosphere or "未设定",
rules=project.world_rules or "未设定",
existing_organizations=existing_orgs_summary,
existing_characters=existing_chars_summary,
plot_context="根据剧情需要引入的新组织",
organization_specification=json.dumps(spec, ensure_ascii=False, indent=2),
mcp_references="" # 暂时不使用MCP增强
)
# 调用AI生成（使用统一的JSON调用方法）
try:
if enable_mcp and user_id:
result = await self.ai_service.generate_text_with_mcp(
prompt=prompt,
user_id=user_id,
db_session=db,
enable_mcp=True,
max_tool_rounds=2
)
content = result.get("content", "")
# 使用统一的JSON清洗方法
cleaned = self.ai_service._clean_json_response(content)
organization_data = json.loads(cleaned)
else:
# 非MCP调用：使用带自动重试的JSON调用
organization_data = await self.ai_service.call_with_json_retry(
prompt=prompt,
max_retries=3
)
org_name = organization_data.get('name', '未知')
logger.info(f" ✅ 组织详情生成成功: {org_name}")
logger.debug(f" 组织数据字段: {list(organization_data.keys())}")
# 确保关键字段存在
if 'name' not in organization_data or not organization_data['name']:
logger.warning(f" ⚠️ AI返回的组织数据缺少name字段，使用规格中的信息")
organization_data['name'] = spec.get('name', f"新组织{spec.get('organization_description', '')[:10]}")
return organization_data
except Exception as e:
logger.error(f" ❌ 生成组织详情失败: {e}")
raise
async def _create_organization_record(
self,
project_id: str,
organization_data: Dict[str, Any],
db: AsyncSession
) -> tuple:
"""创建组织数据库记录（包括Character和Organization）"""
# 首先创建Character记录（is_organization=True）
character = Character(
project_id=project_id,
name=organization_data.get("name", "未命名组织"),
is_organization=True,
role_type=organization_data.get("role_type", "supporting"),
personality=organization_data.get("personality", ""), # 组织特性
background=organization_data.get("background", ""), # 组织背景
appearance=organization_data.get("appearance", ""), # 外在表现
organization_type=organization_data.get("organization_type"),
organization_purpose=organization_data.get("organization_purpose"),
traits=json.dumps(organization_data.get("traits", []), ensure_ascii=False) if organization_data.get("traits") else None
)
db.add(character)
await db.flush()
# 然后创建Organization记录
organization = Organization(
character_id=character.id,
project_id=project_id,
power_level=organization_data.get("power_level", 50),
member_count=0,
location=organization_data.get("location"),
motto=organization_data.get("motto"),
color=organization_data.get("color")
)
db.add(organization)
await db.flush()
logger.info(f" ✅ 创建组织记录: {character.name}, Organization ID: {organization.id}")
return character, organization
async def _create_member_relationships(
self,
organization: Organization,
member_specs: List[Dict[str, Any]],
existing_characters: List[Character],
project_id: str,
db: AsyncSession
) -> List[OrganizationMember]:
"""创建组织成员关系"""
if not member_specs:
return []
members = []
for member_spec in member_specs:
try:
character_name = member_spec.get("character_name")
if not character_name:
continue
# 查找目标角色
target_char = next(
(c for c in existing_characters if c.name == character_name and not c.is_organization),
None
)
if not target_char:
logger.warning(f" ⚠️ 目标角色不存在: {character_name}")
continue
# 检查成员关系是否已存在
existing_member = await db.execute(
select(OrganizationMember).where(
OrganizationMember.organization_id == organization.id,
OrganizationMember.character_id == target_char.id
)
)
if existing_member.scalar_one_or_none():
logger.debug(f" ℹ️ 成员关系已存在: {character_name} -> {organization.id}")
continue
# 创建成员关系
member = OrganizationMember(
organization_id=organization.id,
character_id=target_char.id,
position=member_spec.get("position", "成员"),
rank=member_spec.get("rank", 0),
loyalty=member_spec.get("loyalty", 50),
status=member_spec.get("status", "active"),
joined_at=member_spec.get("joined_at"),
source="auto" # 标记为自动生成
)
db.add(member)
members.append(member)
logger.info(
f" ✅ 创建成员关系: {character_name} -> {organization.id} "
f"({member_spec.get('position', '成员')})"
)
except Exception as e:
logger.warning(f" ❌ 创建成员关系失败: {e}")
continue
# 更新组织成员数量
if members:
organization.member_count = (organization.member_count or 0) + len(members)
return members
# 全局实例缓存
_auto_organization_service_instance: Optional[AutoOrganizationService] = None
def get_auto_organization_service(ai_service: AIService) -> AutoOrganizationService:
"""获取自动组织服务实例（单例模式）"""
global _auto_organization_service_instance
if _auto_organization_service_instance is None:
_auto_organization_service_instance = AutoOrganizationService(ai_service)
return _auto_organization_service_instance
backend/app/services/prompt_service.py
+256
View File
@@ -1679,6 +1679,248 @@ class PromptService:
❌ 在描述中使用特殊符号
❌ 引用不存在的角色或组织
❌ 使用职业ID而非职业名称
</constraints>"""
# 自动组织引入 - 预测性分析提示词（RTCO框架）
AUTO_ORGANIZATION_ANALYSIS = """<system>
你是专业的小说世界构建顾问，擅长预测剧情发展对组织/势力的需求。
</system>
<task>
【分析任务】
预测在接下来的{chapter_count}章续写中，根据剧情发展方向和阶段，是否需要引入新的组织或势力。
【重要说明】
这是预测性分析，而非基于已生成内容的事后分析。
组织包括：帮派、门派、公司、政府机构、神秘组织、家族等。
</task>
<project priority="P1">
【项目信息】
书名：{title}
类型：{genre}
主题：{theme}
【世界观】
时间背景：{time_period}
地理位置：{location}
氛围基调：{atmosphere}
</project>
<context priority="P0">
【已有组织】
{existing_organizations}
【已有角色】
{existing_characters}
【已有章节概览】
{all_chapters_brief}
【续写计划】
- 起始章节：第{start_chapter}章
- 续写数量：{chapter_count}章
- 剧情阶段：{plot_stage}
- 发展方向：{story_direction}
</context>
<analysis_framework priority="P0">
【预测分析维度】
**1. 世界观扩展需求**
根据发展方向，是否需要新的势力或组织来丰富世界观？
**2. 冲突升级需求**
剧情是否需要新的对立势力、竞争组织或神秘集团？
**3. 角色归属需求**
现有角色是否需要加入或对抗某个新组织？
**4. 剧情推动需求**
新组织能否成为推动剧情的关键力量？
**5. 引入时机**
新组织应该在哪个章节出现最合适？
【预测依据】
- 剧情阶段的典型组织需求（如：高潮阶段可能需要强大的敌对势力）
- 故事发展方向的逻辑需要（如：进入新地点需要当地势力）
- 世界观完整性需要（如：权力格局需要多方势力）
- 角色成长需要（如：主角需要加入或创建组织）
</analysis_framework>
<output priority="P0">
【输出格式】
返回纯JSON对象（两种情况之一）：
**情况A：需要新组织**
{{
"needs_new_organizations": true,
"reason": "预测分析原因（150-200字），说明为什么即将的剧情需要新组织",
"organization_count": 1,
"organization_specifications": [
{{
"name": "建议的组织名字（可选）",
"organization_description": "组织在剧情中的定位和作用（100-150字）",
"organization_type": "帮派/门派/公司/政府/家族/神秘组织等",
"importance": "high/medium/low",
"appearance_chapter": {start_chapter},
"power_level": 70,
"plot_function": "在剧情中的具体功能",
"location": "组织所在地或活动区域",
"motto": "组织口号或宗旨（可选）",
"initial_members": [
{{
"character_name": "现有角色名（如需加入）",
"position": "职位",
"reason": "为什么加入"
}}
],
"relationship_suggestions": [
{{
"target_organization": "已有组织名",
"relationship_type": "建议的关系类型（盟友/敌对/竞争/合作等）",
"reason": "为什么建立这种关系"
}}
]
}}
]
}}
**情况B：不需要新组织**
{{
"needs_new_organizations": false,
"reason": "现有组织足以支撑即将的剧情发展，说明理由"
}}
</output>
<constraints>
【必须遵守】
✅ 这是预测性分析，面向未来剧情
✅ 考虑世界观的丰富性和完整性
✅ 确保引入必要性，不为引入而引入
✅ 优先考虑组织的长期作用
✅ 组织应该是推动剧情的关键力量
【禁止事项】
❌ 输出markdown标记
❌ 基于已生成内容做事后分析
❌ 为了引入组织而强行引入
❌ 设计一次性功能组织
❌ 创建与现有组织功能重复的组织
</constraints>"""
# 自动组织引入 - 生成提示词（RTCO框架）
AUTO_ORGANIZATION_GENERATION = """<system>
你是专业的世界构建师，擅长根据剧情需求创建完整的组织/势力设定。
</system>
<task>
【生成任务】
为小说生成新组织的完整设定，包括基本信息、组织特性、背景历史和成员结构。
</task>
<project priority="P1">
【项目信息】
书名：{title}
类型：{genre}
主题：{theme}
【世界观】
时间背景：{time_period}
地理位置：{location}
氛围基调：{atmosphere}
世界规则：{rules}
</project>
<context priority="P0">
【已有组织】
{existing_organizations}
【已有角色】
{existing_characters}
【剧情上下文】
{plot_context}
【组织规格要求】
{organization_specification}
</context>
<mcp_context priority="P2">
【MCP工具参考】
{mcp_references}
</mcp_context>
<requirements priority="P0">
【核心要求】
1. 组织必须符合剧情需求和世界观设定
2. 组织要有明确的目的、结构和特色
3. 组织特性、背景要有深度和独特性
4. 外在表现要具体生动
5. 考虑与已有组织的关系和互动
6. 如果需要，可以建议将现有角色加入组织
</requirements>
<output priority="P0">
【输出格式】
返回纯JSON对象：
{{
"name": "组织名称",
"is_organization": true,
"role_type": "supporting",
"organization_type": "组织类型（帮派/门派/公司/政府/家族/神秘组织等）",
"personality": "组织特性的详细描述（150-200字）：运作方式、核心理念、行事风格、文化价值观",
"background": "组织背景故事（200-300字）：建立历史、发展历程、重要事件、当前地位",
"appearance": "外在表现（100-150字）：总部位置、标志性建筑、组织标志、成员着装",
"organization_purpose": "组织目的和宗旨：明确目标、长期愿景、行动准则",
"power_level": 75,
"location": "所在地点：主要活动区域、势力范围",
"motto": "组织格言或口号",
"color": "组织代表颜色",
"traits": ["特征1", "特征2", "特征3"],
"initial_members": [
{{
"character_name": "已存在的角色名称",
"position": "职位名称",
"rank": 8,
"loyalty": 80,
"joined_at": "加入时间（可选）",
"status": "active"
}}
],
"organization_relationships": [
{{
"target_organization_name": "已存在的组织名称",
"relationship_type": "盟友/敌对/竞争/合作/从属等",
"description": "关系的具体描述"
}}
]
}}
【数值范围】
- power_level：0-100的整数，表示在世界中的影响力
- rank：0到10（职位等级）
- loyalty：0到100（成员忠诚度）
</output>
<constraints>
【必须遵守】
✅ 符合剧情需求和世界观设定
✅ 组织要有独特的定位和价值
✅ character_name必须精确匹配【已有角色】
✅ target_organization_name必须精确匹配【已有组织】
✅ 组织能够推动剧情发展
【禁止事项】
❌ 输出markdown标记
❌ 在描述中使用特殊符号
❌ 引用不存在的角色或组织
❌ 创建功能与现有组织重复的组织
❌ 创建对剧情没有实际作用的组织
</constraints>"""
# 职业体系生成提示词 V2（RTCO框架）
@@ -2166,6 +2408,20 @@ class PromptService:
"parameters": ["title", "genre", "theme", "time_period", "location", "atmosphere", "rules",
"existing_characters", "plot_context", "character_specification", "mcp_references"]
},
"AUTO_ORGANIZATION_ANALYSIS": {
"name": "自动组织分析",
"category": "自动组织引入",
"description": "分析新生成的大纲，判断是否需要引入新组织",
"parameters": ["title", "genre", "theme", "time_period", "location", "atmosphere",
"existing_organizations", "existing_characters", "all_chapters_brief", "start_chapter", "chapter_count", "plot_stage", "story_direction"]
},
"AUTO_ORGANIZATION_GENERATION": {
"name": "自动组织生成",
"category": "自动组织引入",
"description": "根据剧情需求自动生成新组织的完整设定",
"parameters": ["title", "genre", "theme", "time_period", "location", "atmosphere", "rules",
"existing_organizations", "existing_characters", "plot_context", "organization_specification", "mcp_references"]
},
"CAREER_SYSTEM_GENERATION": {
"name": "职业体系生成",
"category": "世界构建",