Initial commit: AI Novel Generation Tool with prologue support and progress tracking

This commit is contained in:
2025-07-16 00:45:41 +08:00
commit eab7f3379a
46 changed files with 6405 additions and 0 deletions

7
ai_novel/__init__.py Normal file
View File

@ -0,0 +1,7 @@
"""AI Novel Writer Agent using LangChain."""
from .writer import NovelWriter
from .config import Config
__version__ = "0.1.0"
__all__ = ["NovelWriter", "Config"]

160
ai_novel/config.py Normal file
View File

@ -0,0 +1,160 @@
"""Configuration management for AI Novel Writer."""
import os
import yaml
from pathlib import Path
from typing import Dict, Any, Optional
class Config:
"""Configuration manager for the AI Novel Writer."""
DEFAULT_CONFIG = {
"project_dir": ".", # 项目目录,包含梗概大纲.md和章节目录.yaml
"prompt_config": {
"previous_chapters_count": 2, # 提示词中包含前n章节的完整内容
},
"novelist_llm": {
"type": "openai",
"model": "gpt-3.5-turbo",
"temperature": 0.7,
"max_tokens": 3000,
},
"summarizer_llm": {
"type": "openai",
"model": "gpt-3.5-turbo",
"temperature": 0.3,
"max_tokens": 500,
},
}
def __init__(self, config_path: Optional[str] = None):
"""Initialize configuration.
Args:
config_path: Path to configuration file. If None, uses default config.
"""
self.config = self.DEFAULT_CONFIG.copy()
if config_path and Path(config_path).exists():
self.load_from_file(config_path)
# Override with environment variables
self._load_from_env()
def load_from_file(self, config_path: str):
"""Load configuration from YAML file.
Args:
config_path: Path to the configuration file
"""
with open(config_path, 'r', encoding='utf-8') as f:
file_config = yaml.safe_load(f)
# Deep merge with default config
self._deep_merge(self.config, file_config)
def _deep_merge(self, base: Dict[str, Any], override: Dict[str, Any]):
"""Deep merge two dictionaries."""
for key, value in override.items():
if key in base and isinstance(base[key], dict) and isinstance(value, dict):
self._deep_merge(base[key], value)
else:
base[key] = value
def _load_from_env(self):
"""Load configuration from environment variables."""
# OpenAI API Key
if os.getenv("OPENAI_API_KEY"):
self.config["novelist_llm"]["api_key"] = os.getenv("OPENAI_API_KEY")
self.config["summarizer_llm"]["api_key"] = os.getenv("OPENAI_API_KEY")
# OpenRouter API Key
if os.getenv("OPENROUTER_API_KEY"):
if self.config["novelist_llm"]["type"] == "openai_compatible":
self.config["novelist_llm"]["api_key"] = os.getenv("OPENROUTER_API_KEY")
if self.config["summarizer_llm"]["type"] == "openai_compatible":
self.config["summarizer_llm"]["api_key"] = os.getenv("OPENROUTER_API_KEY")
# Ollama base URL
if os.getenv("OLLAMA_BASE_URL"):
if self.config["novelist_llm"]["type"] == "ollama":
self.config["novelist_llm"]["base_url"] = os.getenv("OLLAMA_BASE_URL")
if self.config["summarizer_llm"]["type"] == "ollama":
self.config["summarizer_llm"]["base_url"] = os.getenv("OLLAMA_BASE_URL")
def get(self, key: str, default: Any = None) -> Any:
"""Get configuration value by key.
Args:
key: Configuration key (supports dot notation like 'novelist_llm.model')
default: Default value if key not found
Returns:
Configuration value
"""
keys = key.split('.')
value = self.config
for k in keys:
if isinstance(value, dict) and k in value:
value = value[k]
else:
return default
return value
def save_to_file(self, config_path: str):
"""Save current configuration to file.
Args:
config_path: Path to save the configuration file
"""
with open(config_path, 'w', encoding='utf-8') as f:
yaml.dump(self.config, f, allow_unicode=True, default_flow_style=False)
@classmethod
def create_example_config(cls, config_path: str):
"""Create an example configuration file.
Args:
config_path: Path where to create the example config
"""
example_config = {
"project_dir": "novel1", # 项目目录,包含梗概大纲.md和章节目录.yaml
"prompt_config": {
"previous_chapters_count": 2, # 提示词中包含前n章节的完整内容
},
"novelist_llm": {
"type": "openai", # or "openai_compatible", "ollama"
"model": "gpt-4",
"temperature": 0.7,
"max_tokens": 3000,
# "api_key": "your-api-key-here", # or set OPENAI_API_KEY env var
# "base_url": "https://api.openai.com/v1", # for openai_compatible
},
"summarizer_llm": {
"type": "openai",
"model": "gpt-3.5-turbo",
"temperature": 0.3,
"max_tokens": 500,
},
}
with open(config_path, 'w', encoding='utf-8') as f:
yaml.dump(example_config, f, allow_unicode=True, default_flow_style=False)
print(f"Example configuration created at: {config_path}")
print("Please edit the configuration file and set your API keys.")
def load_config(config_path: Optional[str] = None) -> Config:
"""Load configuration from file or create default.
Args:
config_path: Path to configuration file
Returns:
Config: Configuration instance
"""
return Config(config_path)

13
ai_novel/core/__init__.py Normal file
View File

@ -0,0 +1,13 @@
"""Core modules for AI Novel Writer."""
from .project import NovelProject
from .generator import ContentGenerator
from .summarizer import ContentSummarizer
from .compiler import NovelCompiler
__all__ = [
"NovelProject",
"ContentGenerator",
"ContentSummarizer",
"NovelCompiler"
]

144
ai_novel/core/compiler.py Normal file
View File

@ -0,0 +1,144 @@
"""Novel compilation and file management."""
from pathlib import Path
from typing import List
from .project import NovelProject
class NovelCompiler:
"""Compiles novel sections into complete chapters and novels."""
def __init__(self, project: NovelProject):
"""Initialize novel compiler.
Args:
project: Novel project instance
"""
self.project = project
def save_section(self, part_number: int, chapter_number: int, section_number: int,
content: str, summary: str):
"""Save section content and update chapter structure with summary.
Args:
part_number: Part number
chapter_number: Chapter number
section_number: Section number
content: Section content
summary: Section summary
"""
# Get section info
section_info = self.project.get_section_info(part_number, chapter_number, section_number)
if not section_info:
raise ValueError(f"Section {part_number}.{chapter_number}.{section_number} not found")
# Create part directory
part_dir = self.project.get_part_dir(part_number)
part_dir.mkdir(exist_ok=True)
# Save section content
section_file = part_dir / f"section_{chapter_number}_{section_number}.md"
with open(section_file, 'w', encoding='utf-8') as f:
f.write(f"# {section_number}. {section_info['title']}\n\n")
f.write(content)
# Update chapter structure with summary
self.project.update_section_summary(part_number, chapter_number, section_number, summary)
def compile_chapter(self, part_number: int, chapter_number: int) -> str:
"""Compile a complete chapter from its sections.
Args:
part_number: Part number
chapter_number: Chapter number
Returns:
str: Complete chapter content
"""
# Get chapter info
chapter_info = self.project.get_chapter_info(part_number, chapter_number)
if not chapter_info:
raise ValueError(f"Chapter {chapter_number} in part {part_number} not found")
chapter_content = [f"# 第{chapter_number}{chapter_info['title']}\n"]
# Load each section
part_dir = self.project.get_part_dir(part_number)
for section in chapter_info.get("sections", []):
section_file = part_dir / f"section_{chapter_number}_{section['number']}.md"
if section_file.exists():
with open(section_file, 'r', encoding='utf-8') as f:
section_content = f.read()
chapter_content.append(section_content)
chapter_content.append("") # Add spacing between sections
complete_chapter = "\n".join(chapter_content)
# Save complete chapter
chapter_file = part_dir / f"chapter_{chapter_number}_{chapter_info['title']}.md"
with open(chapter_file, 'w', encoding='utf-8') as f:
f.write(complete_chapter)
return complete_chapter
def compile_complete_novel(self):
"""Compile all chapters into a single novel file."""
novel_content = []
# Add title and metadata
title = self.project.get_novel_title()
subtitle = self.project.chapter_structure.get("subtitle", "")
novel_content.extend([
f"# {title}",
f"## {subtitle}" if subtitle else "",
"",
f"**时间跨度**: {self.project.chapter_structure.get('time_span', '')}",
f"**核心主题**: {self.project.chapter_structure.get('core_theme', '')}",
"",
"---",
"",
])
chapters_dir = self.project.get_chapters_dir()
# Add prologue
if "prologue" in self.project.chapter_structure.get("structure", {}):
novel_content.append("# 序章\n")
# Add prologue content (simplified)
# Add all parts
for part in self.project.chapter_structure["structure"]["parts"]:
novel_content.extend([
f"# {part['title']}",
f"*{part['time_period']}*",
"",
f"**文明标志**: {part['civilization_marker']}",
f"**物理限制**: {part['physics_constraint']}",
f"**社会形态**: {part['social_form']}",
"",
])
part_dir = chapters_dir / f"part_{part['part_number']}_{part['title']}"
for chapter in part["chapters"]:
chapter_file = part_dir / f"chapter_{chapter['number']}_{chapter['title']}.md"
if chapter_file.exists():
with open(chapter_file, 'r', encoding='utf-8') as f:
chapter_content = f.read()
novel_content.append(chapter_content)
novel_content.append("\n---\n")
# Add finale
if "finale" in self.project.chapter_structure.get("structure", {}):
novel_content.append(f"# {self.project.chapter_structure['structure']['finale']['title']}\n")
# Add finale content (simplified)
# Save complete novel
novel_file = self.project.project_dir / f"{title}.md"
with open(novel_file, 'w', encoding='utf-8') as f:
f.write("\n".join(novel_content))
return novel_file

203
ai_novel/core/generator.py Normal file
View File

@ -0,0 +1,203 @@
"""Content generation for novel sections."""
from typing import Dict, Any, List
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_core.language_models.chat_models import BaseChatModel
from .project import NovelProject
class ContentGenerator:
"""Generates novel content using LLM."""
def __init__(self, llm: BaseChatModel, project: NovelProject):
"""Initialize content generator.
Args:
llm: Language model for content generation
project: Novel project instance
"""
self.llm = llm
self.project = project
def generate_prologue_chapter(self, chapter_number: int) -> str:
"""Generate content for a prologue chapter.
Args:
chapter_number: Chapter number in prologue
Returns:
str: Generated chapter content
"""
# Get prologue chapter information
prologue_info = self.project.get_prologue_info()
chapter_info = self.project.get_prologue_chapter_info(chapter_number)
if not all([prologue_info, chapter_info]):
raise ValueError(f"Could not find prologue chapter {chapter_number}")
# Build prompt for prologue
prompt = self._build_prologue_prompt(prologue_info, chapter_info)
# Generate content
messages = [
SystemMessage(content=self._get_system_prompt()),
HumanMessage(content=prompt)
]
response = self.llm.invoke(messages)
return response.content
def generate_section(self, part_number: int, chapter_number: int, section_number: int) -> str:
"""Generate content for a specific section.
Args:
part_number: Part number
chapter_number: Chapter number within the part
section_number: Section number within the chapter
Returns:
str: Generated section content
"""
# Get section information
part_info = self.project.get_part_info(part_number)
chapter_info = self.project.get_chapter_info(part_number, chapter_number)
section_info = self.project.get_section_info(part_number, chapter_number, section_number)
if not all([part_info, chapter_info, section_info]):
raise ValueError(f"Could not find part {part_number}, chapter {chapter_number}, section {section_number}")
# Load previous chapters
previous_chapters = self.project.load_previous_chapters(part_number, chapter_number)
# Build prompt
prompt = self._build_section_prompt(part_info, chapter_info, section_info, previous_chapters)
# Generate content
messages = [
SystemMessage(content=self._get_system_prompt()),
HumanMessage(content=prompt)
]
response = self.llm.invoke(messages)
return response.content
def _get_system_prompt(self) -> str:
"""Get the system prompt for content generation."""
return """你是一位专业的科幻小说作家,擅长创作硬科幻作品。你的任务是根据提供的故事梗概、章节大纲和前文内容,创作一个具体的小说章节。
要求:
1. 严格按照提供的故事设定和世界观
2. 保持与前文的连贯性和一致性
3. 文笔优美,情节引人入胜
4. 体现硬科幻的特点,注重科学细节
5. 每个小节约2000-3000字
6. 突出物理法则对文明发展的限制这一核心主题
重要:直接输出小说正文内容,不要添加任何解释性文字、分析说明或格式标记。不要包含"好的,我将...""以下是...""希望..."等开头或结尾的话语。只输出纯粹的小说正文。"""
def _build_section_prompt(self, part_info: Dict, chapter_info: Dict, section_info: Dict,
previous_chapters: List[str]) -> str:
"""Build the prompt for generating a specific section.
Args:
part_info: Information about the current part
chapter_info: Information about the current chapter
section_info: Information about the current section
previous_chapters: List of previous chapter contents
Returns:
str: Complete prompt for the LLM
"""
# Build context
context_parts = [
"# 小说梗概和大纲",
self.project.synopsis,
"",
"# 章节目录结构",
self._format_chapter_structure_for_prompt(),
"",
f"# 当前创作任务",
f"纪元:{part_info['title']} ({part_info['time_period']})",
f"章节:第{chapter_info['number']}{chapter_info['title']}",
f"小节:{section_info['number']}.{section_info['title']}",
"",
]
# Add previous chapters if available
if previous_chapters:
context_parts.extend([
"# 前面章节内容",
*previous_chapters[-2:], # Only include last 2 chapters
"",
])
context_parts.extend([
f"# 请创作小节:{section_info['title']}",
f"请根据以上信息创作这个小节的内容。直接输出小说正文,不要添加任何解释、分析或格式说明。",
])
return "\n".join(context_parts)
def _build_prologue_prompt(self, prologue_info: Dict, chapter_info: Dict) -> str:
"""Build the prompt for generating a prologue chapter.
Args:
prologue_info: Information about the prologue
chapter_info: Information about the current chapter
Returns:
str: Complete prompt for the LLM
"""
# Build context
context_parts = [
"# 小说梗概和大纲",
self.project.synopsis,
"",
"# 章节目录结构",
self._format_chapter_structure_for_prompt(),
"",
f"# 当前创作任务",
f"序章:{prologue_info['title']}",
f"章节:第{chapter_info['number']}{chapter_info['title']}",
f"描述:{chapter_info.get('description', '')}",
"",
f"# 请创作序章章节:{chapter_info['title']}",
f"请根据以上信息创作这个序章章节的内容。直接输出小说正文,不要添加任何解释、分析或格式说明。",
]
return "\n".join(context_parts)
def _format_chapter_structure_for_prompt(self) -> str:
"""Format the chapter structure for inclusion in prompts."""
lines = []
structure = self.project.chapter_structure.get("structure", {})
# Add prologue if exists
if "prologue" in structure:
lines.append("## 序章")
for chapter in structure["prologue"]["chapters"]:
lines.append(f"- 第{chapter['number']}章:{chapter['title']}")
# Add main parts
if "parts" in structure:
for part in structure["parts"]:
lines.append(f"\n## {part['title']} ({part['time_period']})")
lines.append(f"文明标志:{part['civilization_marker']}")
lines.append(f"物理限制:{part['physics_constraint']}")
for chapter in part["chapters"]:
lines.append(f"\n### 第{chapter['number']}章:{chapter['title']}")
if "sections" in chapter:
for section in chapter["sections"]:
summary = getattr(section, 'summary', '') or section.get('summary', '')
summary_text = f" - {summary}" if summary else ""
lines.append(f" - {section['number']}.{section['title']}{summary_text}")
# Add finale if exists
if "finale" in structure:
lines.append(f"\n## {structure['finale']['title']}")
for chapter in structure["finale"]["chapters"]:
lines.append(f"- 第{chapter['number']}章:{chapter['title']}")
return "\n".join(lines)

193
ai_novel/core/progress.py Normal file
View File

@ -0,0 +1,193 @@
"""Progress tracking for novel generation."""
import json
from pathlib import Path
from typing import Dict, Any, List, Optional
from datetime import datetime
class GenerationProgress:
"""Tracks the progress of novel generation for resume functionality."""
def __init__(self, project_dir: str):
"""Initialize progress tracker.
Args:
project_dir: Path to the project directory
"""
self.project_dir = Path(project_dir)
self.progress_file = self.project_dir / ".generation_progress.json"
self.progress_data = self._load_progress()
def _load_progress(self) -> Dict[str, Any]:
"""Load progress data from file."""
if self.progress_file.exists():
try:
with open(self.progress_file, 'r', encoding='utf-8') as f:
return json.load(f)
except (json.JSONDecodeError, IOError):
pass
# Return default structure
return {
"version": "1.0",
"created_at": datetime.now().isoformat(),
"last_updated": datetime.now().isoformat(),
"prologue": {
"completed": False,
"chapters": {}
},
"parts": {}
}
def _save_progress(self):
"""Save progress data to file."""
self.progress_data["last_updated"] = datetime.now().isoformat()
with open(self.progress_file, 'w', encoding='utf-8') as f:
json.dump(self.progress_data, f, indent=2, ensure_ascii=False)
def mark_prologue_chapter_complete(self, chapter_number: int):
"""Mark a prologue chapter as completed.
Args:
chapter_number: Chapter number in prologue
"""
self.progress_data["prologue"]["chapters"][str(chapter_number)] = {
"completed": True,
"completed_at": datetime.now().isoformat()
}
self._save_progress()
def mark_section_complete(self, part_number: int, chapter_number: int, section_number: int):
"""Mark a section as completed.
Args:
part_number: Part number
chapter_number: Chapter number
section_number: Section number
"""
part_key = str(part_number)
chapter_key = str(chapter_number)
section_key = str(section_number)
if part_key not in self.progress_data["parts"]:
self.progress_data["parts"][part_key] = {"chapters": {}}
if chapter_key not in self.progress_data["parts"][part_key]["chapters"]:
self.progress_data["parts"][part_key]["chapters"][chapter_key] = {"sections": {}}
self.progress_data["parts"][part_key]["chapters"][chapter_key]["sections"][section_key] = {
"completed": True,
"completed_at": datetime.now().isoformat()
}
# Check if chapter is complete
self._update_chapter_completion(part_number, chapter_number)
self._save_progress()
def _update_chapter_completion(self, part_number: int, chapter_number: int):
"""Update chapter completion status based on sections."""
# This would need access to the project structure to determine
# if all sections in a chapter are complete
pass
def is_prologue_chapter_complete(self, chapter_number: int) -> bool:
"""Check if a prologue chapter is completed.
Args:
chapter_number: Chapter number in prologue
Returns:
bool: True if chapter is completed
"""
return self.progress_data["prologue"]["chapters"].get(str(chapter_number), {}).get("completed", False)
def is_section_complete(self, part_number: int, chapter_number: int, section_number: int) -> bool:
"""Check if a section is completed.
Args:
part_number: Part number
chapter_number: Chapter number
section_number: Section number
Returns:
bool: True if section is completed
"""
part_key = str(part_number)
chapter_key = str(chapter_number)
section_key = str(section_number)
return (self.progress_data["parts"]
.get(part_key, {})
.get("chapters", {})
.get(chapter_key, {})
.get("sections", {})
.get(section_key, {})
.get("completed", False))
def get_incomplete_sections(self, part_number: Optional[int] = None,
chapter_number: Optional[int] = None) -> List[Dict[str, int]]:
"""Get list of incomplete sections.
Args:
part_number: Optional part number to filter by
chapter_number: Optional chapter number to filter by
Returns:
List of incomplete sections with part, chapter, section numbers
"""
incomplete = []
# This would need to be implemented with access to the full project structure
# to know what sections should exist
return incomplete
def get_progress_summary(self) -> Dict[str, Any]:
"""Get a summary of generation progress.
Returns:
Dictionary with progress statistics
"""
summary = {
"prologue": {
"chapters_completed": len([c for c in self.progress_data["prologue"]["chapters"].values()
if c.get("completed", False)])
},
"parts": {}
}
for part_num, part_data in self.progress_data["parts"].items():
chapters_completed = 0
sections_completed = 0
for chapter_data in part_data.get("chapters", {}).values():
sections_in_chapter = len([s for s in chapter_data.get("sections", {}).values()
if s.get("completed", False)])
sections_completed += sections_in_chapter
# A chapter is considered complete if it has at least one completed section
if sections_in_chapter > 0:
chapters_completed += 1
summary["parts"][part_num] = {
"chapters_completed": chapters_completed,
"sections_completed": sections_completed
}
return summary
def reset_progress(self):
"""Reset all progress data."""
self.progress_data = {
"version": "1.0",
"created_at": datetime.now().isoformat(),
"last_updated": datetime.now().isoformat(),
"prologue": {
"completed": False,
"chapters": {}
},
"parts": {}
}
self._save_progress()

205
ai_novel/core/project.py Normal file
View File

@ -0,0 +1,205 @@
"""Novel project management."""
import yaml
from pathlib import Path
from typing import Dict, Any, List, Optional
from collections import OrderedDict
from .progress import GenerationProgress
class NovelProject:
"""Manages novel project files and metadata."""
def __init__(self, project_dir: str, config: Optional[Dict[str, Any]] = None):
"""Initialize novel project.
Args:
project_dir: Path to the project directory containing 梗概大纲.md and 章节目录.yaml
config: Configuration dictionary containing prompt_config and other settings
"""
self.project_dir = Path(project_dir)
self.config = config or {}
# Validate project structure
self._validate_project_structure()
# Load project files
self.synopsis = self._load_synopsis()
self.chapter_structure = self._load_chapter_structure()
# Initialize progress tracking
self.progress = GenerationProgress(project_dir)
# Create output directories
self._create_output_directories()
def _validate_project_structure(self):
"""Validate that required project files exist."""
if not self.project_dir.exists():
raise FileNotFoundError(f"Project directory not found: {self.project_dir}")
synopsis_path = self.project_dir / "梗概大纲.md"
if not synopsis_path.exists():
raise FileNotFoundError(f"Synopsis file not found: {synopsis_path}")
structure_path = self.project_dir / "章节目录.yaml"
if not structure_path.exists():
raise FileNotFoundError(f"Chapter structure file not found: {structure_path}")
def _load_synopsis(self) -> str:
"""Load the novel synopsis from 梗概大纲.md."""
synopsis_path = self.project_dir / "梗概大纲.md"
with open(synopsis_path, 'r', encoding='utf-8') as f:
return f.read()
def _load_chapter_structure(self) -> Dict[str, Any]:
"""Load the chapter structure from 章节目录.yaml."""
structure_path = self.project_dir / "章节目录.yaml"
with open(structure_path, 'r', encoding='utf-8') as f:
return yaml.safe_load(f)
def _create_output_directories(self):
"""Create output directories for chapters and final novel."""
chapters_dir = self.project_dir / "chapters"
chapters_dir.mkdir(exist_ok=True)
# Create directory for prologue if exists
if "prologue" in self.chapter_structure.get("structure", {}):
prologue_dir = chapters_dir / "prologue"
prologue_dir.mkdir(exist_ok=True)
# Create directories for each part
if "parts" in self.chapter_structure.get("structure", {}):
for part in self.chapter_structure["structure"]["parts"]:
part_dir = chapters_dir / f"part_{part['part_number']}_{part['title']}"
part_dir.mkdir(exist_ok=True)
def get_prologue_info(self) -> Optional[Dict[str, Any]]:
"""Get information about the prologue."""
return self.chapter_structure.get("structure", {}).get("prologue")
def get_prologue_chapter_info(self, chapter_number: int) -> Optional[Dict[str, Any]]:
"""Get information about a specific prologue chapter."""
prologue = self.get_prologue_info()
if not prologue:
return None
for chapter in prologue.get("chapters", []):
if chapter["number"] == chapter_number:
return chapter
return None
def get_part_info(self, part_number: int) -> Optional[Dict[str, Any]]:
"""Get information about a specific part."""
for part in self.chapter_structure["structure"]["parts"]:
if part["part_number"] == part_number:
return part
return None
def get_chapter_info(self, part_number: int, chapter_number: int) -> Optional[Dict[str, Any]]:
"""Get information about a specific chapter."""
part_info = self.get_part_info(part_number)
if not part_info:
return None
for chapter in part_info["chapters"]:
if chapter["number"] == chapter_number:
return chapter
return None
def get_section_info(self, part_number: int, chapter_number: int, section_number: int) -> Optional[Dict[str, Any]]:
"""Get information about a specific section."""
chapter_info = self.get_chapter_info(part_number, chapter_number)
if not chapter_info:
return None
for section in chapter_info.get("sections", []):
if section["number"] == section_number:
return section
return None
def update_section_summary(self, part_number: int, chapter_number: int, section_number: int, summary: str):
"""Update section summary in the chapter structure."""
for part in self.chapter_structure["structure"]["parts"]:
if part["part_number"] == part_number:
for chapter in part["chapters"]:
if chapter["number"] == chapter_number:
for section in chapter.get("sections", []):
if section["number"] == section_number:
section["summary"] = summary
break
break
break
# Save updated chapter structure
self._save_chapter_structure()
def _save_chapter_structure(self):
"""Save the updated chapter structure to file."""
structure_path = self.project_dir / "章节目录.yaml"
with open(structure_path, 'w', encoding='utf-8') as f:
yaml.dump(self.chapter_structure, f, allow_unicode=True, default_flow_style=False, sort_keys=False)
def get_chapters_dir(self) -> Path:
"""Get the chapters output directory."""
return self.project_dir / "chapters"
def get_part_dir(self, part_number: int) -> Path:
"""Get the directory for a specific part."""
part_info = self.get_part_info(part_number)
if not part_info:
raise ValueError(f"Part {part_number} not found")
return self.get_chapters_dir() / f"part_{part_number}_{part_info['title']}"
def get_novel_title(self) -> str:
"""Get the novel title."""
return self.chapter_structure.get("title", "Unknown Novel")
def load_previous_chapters(self, current_part: int, current_chapter: int) -> List[str]:
"""Load content from previous chapters.
Args:
current_part: Current part number
current_chapter: Current chapter number
Returns:
List of previous chapter contents, limited by configuration
"""
previous_chapters = []
chapters_dir = self.get_chapters_dir()
# Get the maximum number of previous chapters to include from config
max_chapters = self.config.get("prompt_config", {}).get("previous_chapters_count", 2)
# Collect all previous chapters first
all_previous_chapters = []
# Load from previous parts and current part
for part in self.chapter_structure["structure"]["parts"]:
if part["part_number"] > current_part:
break
part_dir = chapters_dir / f"part_{part['part_number']}_{part['title']}"
if not part_dir.exists():
continue
for chapter in part["chapters"]:
# Skip future chapters in current part
if part["part_number"] == current_part and chapter["number"] >= current_chapter:
break
chapter_file = part_dir / f"chapter_{chapter['number']}_{chapter['title']}.md"
if chapter_file.exists():
with open(chapter_file, 'r', encoding='utf-8') as f:
all_previous_chapters.append(f.read())
# Return only the last N chapters as specified in config
if max_chapters > 0:
previous_chapters = all_previous_chapters[-max_chapters:]
elif max_chapters == 0:
previous_chapters = [] # Return no chapters if set to 0
else:
previous_chapters = all_previous_chapters # Return all if negative
return previous_chapters

View File

@ -0,0 +1,63 @@
"""Content summarization for novel sections."""
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_core.language_models.chat_models import BaseChatModel
class ContentSummarizer:
"""Summarizes novel content using LLM."""
def __init__(self, llm: BaseChatModel):
"""Initialize content summarizer.
Args:
llm: Language model for summarization
"""
self.llm = llm
def summarize_section(self, section_content: str, section_title: str) -> str:
"""Generate a summary for a section.
Args:
section_content: The content of the section
section_title: The title of the section
Returns:
str: Summary of the section
"""
prompt = self._build_summary_prompt(section_content, section_title)
messages = [
SystemMessage(content=self._get_system_prompt()),
HumanMessage(content=prompt)
]
response = self.llm.invoke(messages)
return response.content.strip()
def _get_system_prompt(self) -> str:
"""Get the system prompt for summarization."""
return "你是一位专业的编辑,擅长总结小说章节内容。"
def _build_summary_prompt(self, section_content: str, section_title: str) -> str:
"""Build the prompt for summarizing a section.
Args:
section_content: The content to summarize
section_title: The title of the section
Returns:
str: Complete prompt for summarization
"""
return f"""请为以下小说章节内容生成一个简洁的总结100-200字
章节标题:{section_title}
章节内容:
{section_content}
要求:
1. 总结要简洁明了,突出关键情节
2. 保留重要的科学设定和世界观元素
3. 便于后续章节创作时参考
4. 不超过200字"""

90
ai_novel/llm_providers.py Normal file
View File

@ -0,0 +1,90 @@
"""LLM Provider configurations and factory."""
from typing import Dict, Any, Optional
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_openai import ChatOpenAI
from langchain_ollama import ChatOllama
from langchain_community.chat_models import ChatOpenAI as CommunityOpenAI
class LLMProviderFactory:
"""Factory for creating LLM instances based on provider configuration."""
@staticmethod
def create_llm(provider_config: Dict[str, Any]) -> BaseChatModel:
"""Create an LLM instance based on provider configuration.
Args:
provider_config: Configuration dictionary containing provider type and settings
Returns:
BaseChatModel: Configured LLM instance
Raises:
ValueError: If provider type is not supported
"""
provider_type = provider_config.get("type", "").lower()
if provider_type == "openai":
return LLMProviderFactory._create_openai_llm(provider_config)
elif provider_type == "openai_compatible":
return LLMProviderFactory._create_openai_compatible_llm(provider_config)
elif provider_type == "ollama":
return LLMProviderFactory._create_ollama_llm(provider_config)
else:
raise ValueError(f"Unsupported provider type: {provider_type}")
@staticmethod
def _create_openai_llm(config: Dict[str, Any]) -> ChatOpenAI:
"""Create OpenAI LLM instance."""
return ChatOpenAI(
model=config.get("model", "gpt-3.5-turbo"),
temperature=config.get("temperature", 0.7),
max_tokens=config.get("max_tokens", 2000),
openai_api_key=config.get("api_key"),
openai_api_base=config.get("base_url"),
)
@staticmethod
def _create_openai_compatible_llm(config: Dict[str, Any]) -> ChatOpenAI:
"""Create OpenAI-compatible LLM instance (e.g., OpenRouter)."""
return ChatOpenAI(
model=config.get("model", "gpt-3.5-turbo"),
temperature=config.get("temperature", 0.7),
max_tokens=config.get("max_tokens", 2000),
openai_api_key=config.get("api_key"),
openai_api_base=config.get("base_url"),
)
@staticmethod
def _create_ollama_llm(config: Dict[str, Any]) -> ChatOllama:
"""Create Ollama LLM instance."""
return ChatOllama(
model=config.get("model", "llama3.1"),
temperature=config.get("temperature", 0.7),
base_url=config.get("base_url", "http://localhost:11434"),
)
# Default configurations for different providers
DEFAULT_CONFIGS = {
"openai": {
"type": "openai",
"model": "gpt-3.5-turbo",
"temperature": 0.7,
"max_tokens": 2000,
},
"openrouter": {
"type": "openai_compatible",
"model": "anthropic/claude-3-haiku",
"temperature": 0.7,
"max_tokens": 2000,
"base_url": "https://openrouter.ai/api/v1",
},
"ollama": {
"type": "ollama",
"model": "llama3.1",
"temperature": 0.7,
"base_url": "http://localhost:11434",
},
}

252
ai_novel/main.py Normal file
View File

@ -0,0 +1,252 @@
"""Main CLI interface for AI Novel Writer."""
import click
import os
from pathlib import Path
from rich.console import Console
from rich.progress import Progress, SpinnerColumn, TextColumn
from rich.panel import Panel
from rich.text import Text
from .config import Config
from .writer import NovelWriter
console = Console()
@click.group()
@click.version_option()
def cli():
"""AI Novel Writer Agent using LangChain.
Generate novels using LLM agents based on synopsis and chapter structure.
"""
pass
@cli.command()
@click.option('--config', '-c', type=click.Path(exists=True), required=True, help='Configuration file path')
@click.option('--prologue', type=int, help='Generate specific prologue chapter number')
@click.option('--part', type=int, help='Generate specific part only')
@click.option('--chapter', type=int, help='Generate specific chapter only (requires --part)')
@click.option('--section', type=int, help='Generate specific section only (requires --part and --chapter)')
@click.option('--resume/--no-resume', default=True, help='Resume from previous progress (default: True)')
def generate(config, prologue, part, chapter, section, resume):
"""Generate novel content based on configuration.
The configuration file should specify the project_dir containing 梗概大纲.md and 章节目录.yaml files.
"""
try:
# Load configuration
config_obj = Config(config)
project_dir = config_obj.get("project_dir", ".")
# Initialize novel writer
console.print(Panel.fit("🤖 Initializing AI Novel Writer", style="blue"))
console.print(f"[blue]Project directory: {project_dir}[/blue]")
writer = NovelWriter(project_dir, config_obj.config)
# Generate content based on options
if prologue:
# Generate specific prologue chapter
prologue_info = writer.project.get_prologue_info()
chapter_info = writer.project.get_prologue_chapter_info(prologue)
if not prologue_info or not chapter_info:
raise click.ClickException(f"Prologue chapter {prologue} not found")
console.print(Panel.fit(f"📖 Generating Prologue Chapter", style="green"))
console.print(f"[cyan]序章:[/cyan] {prologue_info['title']}")
console.print(f"[cyan]章节:[/cyan] 第{prologue}{chapter_info['title']}")
console.print(f"[cyan]描述:[/cyan] {chapter_info.get('description', '')}")
console.print()
with console.status("[bold green]正在生成内容..."):
content = writer.generate_prologue_chapter(prologue)
console.print(f"[green]✓ 内容生成完成[/green] ({len(content)} 字符)")
with console.status("[bold blue]正在生成总结..."):
summary = writer.summarize_section(content, chapter_info['title'])
console.print(f"[blue]✓ 总结生成完成[/blue]")
console.print(f"[yellow]总结:[/yellow] {summary[:100]}..." if len(summary) > 100 else f"[yellow]总结:[/yellow] {summary}")
writer.save_prologue_chapter(prologue, content, summary)
console.print(f"[green]✓ 文件已保存[/green]")
console.print()
elif section and chapter and part:
# Generate specific section
# Get section info for detailed output
part_info = writer.project.get_part_info(part)
chapter_info = writer.project.get_chapter_info(part, chapter)
section_info = writer.project.get_section_info(part, chapter, section)
console.print(Panel.fit(f"📖 Generating Section", style="green"))
console.print(f"[cyan]纪元:[/cyan] {part_info['title']} ({part_info['time_period']})")
console.print(f"[cyan]章节:[/cyan] 第{chapter}{chapter_info['title']}")
console.print(f"[cyan]小节:[/cyan] {section}.{section_info['title']}")
console.print()
with console.status("[bold green]正在生成内容..."):
content = writer.generate_section(part, chapter, section)
console.print(f"[green]✓ 内容生成完成[/green] ({len(content)} 字符)")
with console.status("[bold blue]正在生成总结..."):
summary = writer.summarize_section(content, section_info['title'])
console.print(f"[blue]✓ 总结生成完成[/blue]")
console.print(f"[yellow]总结:[/yellow] {summary[:100]}..." if len(summary) > 100 else f"[yellow]总结:[/yellow] {summary}")
writer.save_section(part, chapter, section, content, summary)
console.print(f"[green]✓ 文件已保存[/green]")
console.print()
elif chapter and part:
# Generate specific chapter
part_info = writer.project.get_part_info(part)
chapter_info = writer.project.get_chapter_info(part, chapter)
console.print(Panel.fit(f"📚 Generating Chapter", style="green"))
console.print(f"[cyan]纪元:[/cyan] {part_info['title']} ({part_info['time_period']})")
console.print(f"[cyan]章节:[/cyan] 第{chapter}{chapter_info['title']}")
console.print(f"[cyan]小节数量:[/cyan] {len(chapter_info.get('sections', []))}")
console.print()
writer.generate_complete_chapter(part, chapter)
console.print("[green]✓ Chapter generated successfully[/green]")
elif part:
# Generate specific part
console.print(f"[green]Generating Part {part}[/green]")
writer.generate_part(part)
console.print("[green]✓ Part generated successfully[/green]")
else:
# Generate complete novel
resume_text = "with resume" if resume else "from scratch"
console.print(f"[green]Generating complete novel ({resume_text})...[/green]")
writer.generate_complete_novel(resume=resume)
except Exception as e:
console.print(f"[red]Error: {str(e)}[/red]")
raise click.ClickException(str(e))
@cli.command()
@click.argument('config_path', type=click.Path())
def init_config(config_path):
"""Create an example configuration file."""
try:
Config.create_example_config(config_path)
console.print(f"[green]✓ Example configuration created at: {config_path}[/green]")
console.print("[yellow]Please edit the configuration file and set your API keys.[/yellow]")
except Exception as e:
console.print(f"[red]Error creating config: {str(e)}[/red]")
raise click.ClickException(str(e))
@cli.command()
@click.option('--config', '-c', type=click.Path(exists=True), required=True, help='Configuration file path')
def progress(config):
"""Show generation progress for the novel project."""
try:
# Load configuration
config_obj = Config(config)
project_dir = config_obj.get("project_dir", ".")
# Initialize project to access progress
from .core.project import NovelProject
project = NovelProject(project_dir, config_obj.config)
console.print(Panel.fit("📊 Generation Progress", style="blue"))
# Get progress summary
summary = project.progress.get_progress_summary()
# Show prologue progress
prologue_info = project.get_prologue_info()
if prologue_info:
prologue_chapters = len(prologue_info.get("chapters", []))
prologue_completed = summary["prologue"]["chapters_completed"]
console.print(f"[cyan]序章进度:[/cyan] {prologue_completed}/{prologue_chapters} 章节完成")
# Show parts progress
for part in project.chapter_structure["structure"]["parts"]:
part_num = str(part["part_number"])
if part_num in summary["parts"]:
part_summary = summary["parts"][part_num]
total_chapters = len(part["chapters"])
console.print(f"[cyan]{part['title']}:[/cyan] {part_summary['chapters_completed']}/{total_chapters} 章节, {part_summary['sections_completed']} 小节完成")
else:
console.print(f"[yellow]{part['title']}:[/yellow] 未开始")
console.print(f"\n[green]进度文件位置:[/green] {project.progress.progress_file}")
except Exception as e:
console.print(f"[red]Error: {str(e)}[/red]")
raise click.ClickException(str(e))
@cli.command()
@click.argument('project_path', type=click.Path())
def init_project(project_path):
"""Initialize a new novel project directory."""
try:
project_path = Path(project_path)
project_path.mkdir(exist_ok=True)
# Create example files
synopsis_content = """# 小说梗概
请在此处编写您的小说梗概和大纲...
## 核心设定
## 主要角色
## 故事大纲
"""
structure_content = """---
title: "您的小说标题"
subtitle: "章节目录"
time_span: "故事时间跨度"
core_theme: "核心主题"
structure:
parts:
- part_number: 1
title: "第一部分"
time_period: "时间段"
civilization_marker: "文明标志"
physics_constraint: "物理限制"
social_form: "社会形态"
chapters:
- number: 1
title: "第一章"
sections:
- number: 1
title: "第一节"
- number: 2
title: "第二节"
"""
with open(project_path / "梗概大纲.md", 'w', encoding='utf-8') as f:
f.write(synopsis_content)
with open(project_path / "章节目录.yaml", 'w', encoding='utf-8') as f:
f.write(structure_content)
console.print(f"[green]✓ Project initialized at: {project_path}[/green]")
console.print("[yellow]Please edit 梗概大纲.md and 章节目录.yaml before generating content.[/yellow]")
except Exception as e:
console.print(f"[red]Error initializing project: {str(e)}[/red]")
raise click.ClickException(str(e))
if __name__ == "__main__":
cli()

252
ai_novel/writer.py Normal file
View File

@ -0,0 +1,252 @@
"""Main Novel Writer orchestrator."""
from typing import Dict, Any
from .llm_providers import LLMProviderFactory
from .core import NovelProject, ContentGenerator, ContentSummarizer, NovelCompiler
from rich.console import Console
from rich.progress import Progress, SpinnerColumn, TextColumn
class NovelWriter:
"""Main orchestrator for novel writing process."""
def __init__(self, project_dir: str, config: Dict[str, Any]):
"""Initialize the Novel Writer.
Args:
project_dir: Path to the novel project directory
config: Configuration dictionary containing LLM settings
"""
# Initialize project
self.project = NovelProject(project_dir, config)
# Initialize LLMs
self.novelist_llm = LLMProviderFactory.create_llm(config["novelist_llm"])
self.summarizer_llm = LLMProviderFactory.create_llm(config["summarizer_llm"])
# Initialize components
self.generator = ContentGenerator(self.novelist_llm, self.project)
self.summarizer = ContentSummarizer(self.summarizer_llm)
self.compiler = NovelCompiler(self.project)
def generate_section(self, part_number: int, chapter_number: int, section_number: int) -> str:
"""Generate content for a specific section.
Args:
part_number: Part number
chapter_number: Chapter number within the part
section_number: Section number within the chapter
Returns:
str: Generated section content
"""
return self.generator.generate_section(part_number, chapter_number, section_number)
def summarize_section(self, section_content: str, section_title: str) -> str:
"""Generate a summary for a section.
Args:
section_content: The content of the section
section_title: The title of the section
Returns:
str: Summary of the section
"""
return self.summarizer.summarize_section(section_content, section_title)
def generate_prologue_chapter(self, chapter_number: int) -> str:
"""Generate content for a prologue chapter.
Args:
chapter_number: Chapter number in prologue
Returns:
str: Generated chapter content
"""
return self.generator.generate_prologue_chapter(chapter_number)
def save_prologue_chapter(self, chapter_number: int, content: str, summary: str):
"""Save a generated prologue chapter to file.
Args:
chapter_number: Chapter number in prologue
content: Generated content
summary: Generated summary
"""
# Get chapter info for validation
chapter_info = self.project.get_prologue_chapter_info(chapter_number)
if not chapter_info:
raise ValueError(f"Prologue chapter {chapter_number} not found")
# Create prologue directory
prologue_dir = self.project.get_chapters_dir() / "prologue"
prologue_dir.mkdir(exist_ok=True)
# Save chapter content
chapter_file = prologue_dir / f"chapter_{chapter_number}_{chapter_info['title']}.md"
with open(chapter_file, 'w', encoding='utf-8') as f:
f.write(f"# 第{chapter_number}{chapter_info['title']}\n\n")
f.write(content)
# Mark as complete in progress
self.project.progress.mark_prologue_chapter_complete(chapter_number)
def save_section(self, part_number: int, chapter_number: int, section_number: int,
content: str, summary: str):
"""Save section content and update chapter structure with summary.
Args:
part_number: Part number
chapter_number: Chapter number
section_number: Section number
content: Section content
summary: Section summary
"""
self.compiler.save_section(part_number, chapter_number, section_number, content, summary)
# Mark as complete in progress
self.project.progress.mark_section_complete(part_number, chapter_number, section_number)
def generate_complete_chapter(self, part_number: int, chapter_number: int) -> str:
"""Generate a complete chapter by creating all its sections.
Args:
part_number: Part number
chapter_number: Chapter number
Returns:
str: Complete chapter content
"""
# Get chapter info
chapter_info = self.project.get_chapter_info(part_number, chapter_number)
if not chapter_info:
raise ValueError(f"Chapter {chapter_number} in part {part_number} not found")
console = Console()
sections = chapter_info.get("sections", [])
# Generate each section
for i, section in enumerate(sections, 1):
# Create a separate progress for each section
with Progress(
SpinnerColumn(),
TextColumn("[progress.description]{task.description}"),
console=console,
) as progress:
task = progress.add_task(f"生成小节 {section['number']}: {section['title']} ({i}/{len(sections)})", total=None)
# Generate section content
section_content = self.generate_section(part_number, chapter_number, section["number"])
# Generate summary
summary = self.summarize_section(section_content, section["title"])
# Save section
self.save_section(part_number, chapter_number, section["number"], section_content, summary)
# Progress will automatically stop when exiting the context
# Print completion message after progress stops
console.print(f"[green]✓[/green] 小节 {section['number']}: {section['title']} 完成 ({len(section_content)} 字符)")
# Compile complete chapter
console.print("[blue]正在编译完整章节...[/blue]")
complete_chapter = self.compiler.compile_chapter(part_number, chapter_number)
console.print(f"[green]✓ 章节编译完成[/green] (总计 {len(complete_chapter)} 字符)")
return complete_chapter
def generate_complete_novel(self, resume: bool = True):
"""Generate the complete novel by processing all parts and chapters.
Args:
resume: Whether to resume from previous progress (default: True)
"""
from rich.console import Console
from rich.progress import Progress, SpinnerColumn, TextColumn
console = Console()
console.print(f"[green]Starting novel generation for: {self.project.get_novel_title()}[/green]")
if not resume:
self.project.progress.reset_progress()
console.print("[yellow]Progress reset. Starting from beginning.[/yellow]")
# Generate prologue if exists
prologue_info = self.project.get_prologue_info()
if prologue_info:
console.print("[blue]Processing prologue...[/blue]")
for chapter in prologue_info.get("chapters", []):
chapter_num = chapter["number"]
if not self.project.progress.is_prologue_chapter_complete(chapter_num):
with Progress(
SpinnerColumn(),
TextColumn("[progress.description]{task.description}"),
console=console,
) as progress:
task = progress.add_task(f"生成序章 第{chapter_num}章: {chapter['title']}", total=None)
# Generate chapter content
content = self.generate_prologue_chapter(chapter_num)
# Generate summary
summary = self.summarize_section(content, chapter["title"])
# Save chapter
self.save_prologue_chapter(chapter_num, content, summary)
console.print(f"[green]✓[/green] 序章第{chapter_num}章: {chapter['title']} 完成 ({len(content)} 字符)")
else:
console.print(f"[yellow]⏭[/yellow] 序章第{chapter_num}章: {chapter['title']} 已完成,跳过")
# Generate all parts
for part in self.project.chapter_structure["structure"]["parts"]:
console.print(f"\n[blue]Processing {part['title']} ({part['time_period']})[/blue]")
for chapter in part["chapters"]:
console.print(f" [cyan]Chapter {chapter['number']}: {chapter['title']}[/cyan]")
# Check if any sections in this chapter are incomplete
sections = chapter.get("sections", [])
incomplete_sections = [
s for s in sections
if not self.project.progress.is_section_complete(part["part_number"], chapter["number"], s["number"])
]
if incomplete_sections:
self.generate_complete_chapter(part["part_number"], chapter["number"])
else:
console.print(f" [yellow]⏭ 章节已完成,跳过[/yellow]")
# Compile complete novel
console.print("\n[blue]正在编译完整小说...[/blue]")
novel_file = self.compiler.compile_complete_novel()
console.print(f"[green]✓ 小说生成完成!保存至: {novel_file}[/green]")
return novel_file
def generate_part(self, part_number: int):
"""Generate all chapters in a specific part.
Args:
part_number: Part number to generate
"""
part_info = self.project.get_part_info(part_number)
if not part_info:
raise ValueError(f"Part {part_number} not found")
print(f"Generating {part_info['title']} ({part_info['time_period']})")
for chapter in part_info["chapters"]:
print(f" Generating Chapter {chapter['number']}: {chapter['title']}")
self.generate_complete_chapter(part_number, chapter["number"])
print(f"Part {part_number} generation completed!")
@property
def chapter_structure(self):
"""Get the chapter structure for backward compatibility."""
return self.project.chapter_structure