import gradio as gr import os import json import requests from datetime import datetime import time from typing import List, Dict, Any, Generator, Tuple, Optional import logging import re import tempfile from pathlib import Path import sqlite3 import hashlib import threading from contextlib import contextmanager from dataclasses import dataclass, field from collections import defaultdict # --- 로깅 설정 --- logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # --- Document export imports --- try: from docx import Document from docx.shared import Inches, Pt, RGBColor from docx.enum.text import WD_ALIGN_PARAGRAPH from docx.enum.style import WD_STYLE_TYPE DOCX_AVAILABLE = True except ImportError: DOCX_AVAILABLE = False logger.warning("python-docx not installed. DOCX export will be disabled.") # --- 환경 변수 및 상수 --- FRIENDLI_TOKEN = os.getenv("FRIENDLI_TOKEN", "") BRAVE_SEARCH_API_KEY = os.getenv("BRAVE_SEARCH_API_KEY", "") API_URL = "https://api.friendli.ai/dedicated/v1/chat/completions" MODEL_ID = "dep89a2fld32mcm" DB_PATH = "novel_sessions_v2.db" # --- 환경 변수 검증 --- if not FRIENDLI_TOKEN: logger.error("FRIENDLI_TOKEN not set. Application will not work properly.") # 실제 환경에서는 여기서 프로그램을 종료해야 하지만, 데모를 위해 더미 토큰을 사용합니다. FRIENDLI_TOKEN = "dummy_token_for_testing" if not BRAVE_SEARCH_API_KEY: logger.warning("BRAVE_SEARCH_API_KEY not set. Web search features will be disabled.") # --- 전역 변수 --- db_lock = threading.Lock() # 최적화된 단계 구성 (25단계로 압축 및 강화) OPTIMIZED_STAGES = [ ("director", "🎬 감독자: 초기 기획 (웹 검색 포함)"), ("critic", "📝 비평가: 기획 검토 (테마 및 일관성)"), ("director", "🎬 감독자: 수정된 마스터플랜"), ] + [ (f"writer{i}", f"✍️ 작가 {i}: 초안 (페이지 {(i-1)*3+1}-{i*3})") for i in range(1, 11) ] + [ ("critic", "📝 비평가: 중간 검토 (일관성 및 테마 유지)"), ] + [ (f"writer{i}", f"✍️ 작가 {i}: 수정본 (페이지 {(i-1)*3+1}-{i*3})") for i in range(1, 11) ] + [ ("critic", f"📝 비평가: 최종 검토 및 종합 보고서 작성"), ] # --- 데이터 클래스 --- @dataclass class CharacterState: """캐릭터의 현재 상태를 나타내는 데이터 클래스""" name: str alive: bool = True location: str = "" injuries: List[str] = field(default_factory=list) emotional_state: str = "" relationships: Dict[str, str] = field(default_factory=dict) last_seen_chapter: int = 0 description: str = "" role: str = "" @dataclass class PlotPoint: """플롯 포인트를 나타내는 데이터 클래스""" chapter: int event_type: str description: str characters_involved: List[str] impact_level: int timestamp: str = "" @dataclass class TimelineEvent: """시간선 이벤트를 나타내는 데이터 클래스""" chapter: int time_reference: str event_description: str duration: str = "" relative_time: str = "" # --- 핵심 로직 클래스 --- class ConsistencyTracker: """일관성 추적 시스템""" def __init__(self): self.character_states: Dict[str, CharacterState] = {} self.plot_points: List[PlotPoint] = [] self.timeline_events: List[TimelineEvent] = [] self.locations: Dict[str, str] = {} self.established_facts: List[str] = [] self.content_hashes: Dict[str, int] = {} # 해시와 해당 챕터 번호를 저장 def register_character(self, character: CharacterState): """새 캐릭터 등록""" self.character_states[character.name] = character logger.info(f"Character registered: {character.name}") def update_character_state(self, name: str, chapter: int, updates: Dict[str, Any]): """캐릭터 상태 업데이트""" if name not in self.character_states: self.register_character(CharacterState(name=name, last_seen_chapter=chapter)) char = self.character_states[name] for key, value in updates.items(): if hasattr(char, key): setattr(char, key, value) char.last_seen_chapter = chapter def add_plot_point(self, plot_point: PlotPoint): """플롯 포인트 추가""" plot_point.timestamp = datetime.now().isoformat() self.plot_points.append(plot_point) def check_repetition(self, content: str, current_chapter: int) -> Tuple[bool, str]: """향상된 반복 내용 검사""" sentences = re.split(r'[.!?]+', content) for sentence in sentences: sentence_strip = sentence.strip() if len(sentence_strip) > 20: # 너무 짧은 문장은 무시 sentence_hash = hashlib.md5(sentence_strip.encode('utf-8')).hexdigest() if sentence_hash in self.content_hashes: previous_chapter = self.content_hashes[sentence_hash] # 바로 이전 챕터와의 반복은 허용할 수 있으므로, 2챕터 이상 차이날 때만 오류로 간주 if current_chapter > previous_chapter + 1: return True, f"문장 반복 (챕터 {previous_chapter}과 유사): {sentence_strip[:50]}..." # 새 내용의 해시 추가 for sentence in sentences: sentence_strip = sentence.strip() if len(sentence_strip) > 20: sentence_hash = hashlib.md5(sentence_strip.encode('utf-8')).hexdigest() self.content_hashes[sentence_hash] = current_chapter return False, "" def validate_consistency(self, chapter: int, content: str) -> List[str]: """일관성 검증""" errors = [] # 사망한 캐릭터 등장 검사 for char_name, char_state in self.character_states.items(): if char_name.lower() in content.lower() and not char_state.alive: errors.append(f"⚠️ 사망한 캐릭터 '{char_name}'이(가) 등장했습니다.") # 내용 반복 검사 is_repetition, repeat_msg = self.check_repetition(content, chapter) if is_repetition: errors.append(f"🔄 {repeat_msg}") return errors def get_character_summary(self, chapter: int) -> str: """현재 챕터 기준 캐릭터 요약""" summary = "\n=== 캐릭터 현황 요약 (이전 2개 챕터 기준) ===\n" active_chars = [char for char in self.character_states.values() if char.last_seen_chapter >= chapter - 2] if not active_chars: return "\n(아직 주요 캐릭터 정보가 없습니다.)\n" for char in active_chars: status = "생존" if char.alive else "사망" summary += f"• {char.name}: {status}" if char.alive and char.location: summary += f" (위치: {char.location})" if char.injuries: summary += f" (부상: {', '.join(char.injuries[-1:])})" summary += "\n" return summary def get_plot_summary(self, chapter: int) -> str: """플롯 요약""" summary = "\n=== 최근 주요 사건 요약 ===\n" recent_events = [p for p in self.plot_points if p.chapter >= chapter - 2] if not recent_events: return "\n(아직 주요 사건이 없습니다.)\n" for event in recent_events[-3:]: # 최근 3개만 표시 summary += f"• [챕터 {event.chapter}] {event.description}\n" return summary class WebSearchIntegration: """웹 검색 기능 (감독자 단계에서만 사용)""" def __init__(self): self.brave_api_key = BRAVE_SEARCH_API_KEY self.search_url = "https://api.search.brave.com/res/v1/web/search" self.enabled = bool(self.brave_api_key) def search(self, query: str, count: int = 3, language: str = "en") -> List[Dict]: """웹 검색 수행""" if not self.enabled: return [] headers = { "Accept": "application/json", "X-Subscription-Token": self.brave_api_key } params = { "q": query, "count": count, "search_lang": "ko" if language == "Korean" else "en", "text_decorations": False, "safesearch": "moderate" } try: response = requests.get(self.search_url, headers=headers, params=params, timeout=10) response.raise_for_status() results = response.json().get("web", {}).get("results", []) logger.info(f"웹 검색 성공: '{query}'에 대해 {len(results)}개 결과 발견") return results except requests.exceptions.RequestException as e: logger.error(f"웹 검색 API 오류: {e}") return [] def extract_relevant_info(self, results: List[Dict], max_chars: int = 1500) -> str: """검색 결과에서 관련 정보 추출""" if not results: return "" extracted = [] total_chars = 0 for i, result in enumerate(results[:3], 1): title = result.get("title", "") description = result.get("description", "") url = result.get("url", "") info = f"[{i}] {title}\n{description}\nSource: {url}\n" if total_chars + len(info) < max_chars: extracted.append(info) total_chars += len(info) else: break return "\n---\n".join(extracted) class NovelDatabase: """소설 세션 관리 데이터베이스""" @staticmethod def init_db(): with sqlite3.connect(DB_PATH) as conn: conn.execute("PRAGMA journal_mode=WAL") cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS sessions ( session_id TEXT PRIMARY KEY, user_query TEXT NOT NULL, language TEXT NOT NULL, created_at TEXT DEFAULT (datetime('now')), updated_at TEXT DEFAULT (datetime('now')), status TEXT DEFAULT 'active', current_stage INTEGER DEFAULT 0, final_novel TEXT, consistency_report TEXT ) ''') cursor.execute(''' CREATE TABLE IF NOT EXISTS stages ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, stage_number INTEGER NOT NULL, stage_name TEXT NOT NULL, role TEXT NOT NULL, content TEXT, word_count INTEGER DEFAULT 0, status TEXT DEFAULT 'pending', consistency_score REAL DEFAULT 0.0, created_at TEXT DEFAULT (datetime('now')), updated_at TEXT DEFAULT (datetime('now')), FOREIGN KEY (session_id) REFERENCES sessions(session_id), UNIQUE(session_id, stage_number) ) ''') cursor.execute(''' CREATE TABLE IF NOT EXISTS character_states ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, character_name TEXT NOT NULL, chapter INTEGER NOT NULL, is_alive BOOLEAN DEFAULT TRUE, location TEXT, injuries TEXT, emotional_state TEXT, description TEXT, created_at TEXT DEFAULT (datetime('now')), FOREIGN KEY (session_id) REFERENCES sessions(session_id) ) ''') cursor.execute('CREATE INDEX IF NOT EXISTS idx_session_id ON stages(session_id)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_stage_number ON stages(stage_number)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_char_session ON character_states(session_id)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_session_status ON sessions(status)') conn.commit() @staticmethod @contextmanager def get_db(): with db_lock: conn = sqlite3.connect(DB_PATH, timeout=30.0) conn.row_factory = sqlite3.Row try: yield conn finally: conn.close() @staticmethod def create_session(user_query: str, language: str) -> str: session_id = hashlib.md5(f"{user_query}{datetime.now()}".encode()).hexdigest() with NovelDatabase.get_db() as conn: conn.cursor().execute( 'INSERT INTO sessions (session_id, user_query, language) VALUES (?, ?, ?)', (session_id, user_query, language) ) conn.commit() return session_id @staticmethod def save_stage(session_id: str, stage_number: int, stage_name: str, role: str, content: str, status: str = 'complete', consistency_score: float = 0.0): word_count = len(content.split()) if content else 0 with NovelDatabase.get_db() as conn: cursor = conn.cursor() cursor.execute(''' INSERT INTO stages (session_id, stage_number, stage_name, role, content, word_count, status, consistency_score) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(session_id, stage_number) DO UPDATE SET content=?, word_count=?, status=?, stage_name=?, consistency_score=?, updated_at=datetime('now') ''', (session_id, stage_number, stage_name, role, content, word_count, status, consistency_score, content, word_count, status, stage_name, consistency_score)) cursor.execute( "UPDATE sessions SET updated_at = datetime('now'), current_stage = ? WHERE session_id = ?", (stage_number, session_id) ) conn.commit() @staticmethod def get_session(session_id: str) -> Optional[Dict]: with NovelDatabase.get_db() as conn: row = conn.cursor().execute('SELECT * FROM sessions WHERE session_id = ?', (session_id,)).fetchone() return dict(row) if row else None @staticmethod def get_latest_active_session() -> Optional[Dict]: with NovelDatabase.get_db() as conn: row = conn.cursor().execute("SELECT * FROM sessions WHERE status = 'active' ORDER BY updated_at DESC LIMIT 1").fetchone() return dict(row) if row else None @staticmethod def get_stages(session_id: str) -> List[Dict]: with NovelDatabase.get_db() as conn: rows = conn.cursor().execute('SELECT * FROM stages WHERE session_id = ? ORDER BY stage_number', (session_id,)).fetchall() return [dict(row) for row in rows] @staticmethod def get_writer_content(session_id: str) -> str: with NovelDatabase.get_db() as conn: all_content = [] for writer_num in range(1, 11): row = conn.cursor().execute( "SELECT content FROM stages WHERE session_id = ? AND role = ? AND stage_name LIKE '%수정본%' ORDER BY stage_number DESC LIMIT 1", (session_id, f'writer{writer_num}') ).fetchone() if row and row['content']: all_content.append(row['content'].strip()) return '\n\n'.join(all_content) @staticmethod def update_final_novel(session_id: str, final_novel: str, consistency_report: str = ""): with NovelDatabase.get_db() as conn: conn.cursor().execute( "UPDATE sessions SET final_novel = ?, status = 'complete', updated_at = datetime('now'), consistency_report = ? WHERE session_id = ?", (final_novel, consistency_report, session_id) ) conn.commit() @staticmethod def get_active_sessions() -> List[Dict]: with NovelDatabase.get_db() as conn: rows = conn.cursor().execute( "SELECT session_id, user_query, language, created_at, current_stage FROM sessions WHERE status = 'active' ORDER BY updated_at DESC LIMIT 10" ).fetchall() return [dict(row) for row in rows] class NovelWritingSystem: """최적화된 소설 생성 시스템""" def __init__(self): self.token = FRIENDLI_TOKEN self.api_url = API_URL self.model_id = MODEL_ID self.consistency_tracker = ConsistencyTracker() self.web_search = WebSearchIntegration() self.current_session_id = None NovelDatabase.init_db() def create_headers(self): """API 헤더 생성""" return {"Authorization": f"Bearer {self.token}", "Content-Type": "application/json"} # --- 프롬프트 생성 함수들 (Thematic Guardian 개념 통합) --- def create_director_initial_prompt(self, user_query: str, language: str) -> str: """감독자 초기 기획 프롬프트 (웹 검색 및 테마 제약 조건 강화)""" search_results_str = "" if self.web_search.enabled: queries = [f"{user_query} novel setting", f"{user_query} background information"] search_results = self.web_search.search(queries[0], count=2, language=language) if search_results: search_results_str = self.web_search.extract_relevant_info(search_results) lang_prompts = { "Korean": { "title": "당신은 30페이지 분량의 중편 소설을 기획하는 문학 감독자입니다.", "user_theme": "사용자 주제", "plan_instruction": "다음 요소들을 포함한 상세한 소설 기획을 작성하세요:", "theme_section": "1. **주제와 장르 설정**\n - 핵심 주제와 메시지 (사용자 의도 깊이 반영)\n - 장르 및 분위기\n - 독자층 고려사항", "char_section": "2. **주요 등장인물** (3-5명)\n | 이름 | 역할 | 성격 | 배경 | 목표 | 갈등 |", "setting_section": "3. **배경 설정**\n - 시공간적 배경\n - 사회적/문화적 환경\n - 주요 장소들", "plot_section": "4. **플롯 구조** (10개 파트, 각 3페이지 분량)\n | 파트 | 페이지 | 주요 사건 | 긴장도 | 캐릭터 발전 |", "guideline_section": "5. **작가별 지침**\n - 일관성 유지를 위한 핵심 설정\n - 문체와 톤 가이드라인", "constraint_title": "⚠️매우 중요한 지시사항: 핵심 제약 조건⚠️", "constraint_body": "이 소설은 **AI로 인해 모든 것이 쉽게 해결되는 긍정적이고 단순한 이야기가 아닙니다.**\n반드시 사용자의 주제인 '{query}'에 담긴 **핵심 감정(예: 불안, 소외감, 상실감, 세대 갈등 등)을 중심으로 서사를 전개해야 합니다.**\nAI나 특정 기술은 편리한 도구가 아니라, 주인공에게 **갈등과 상실감을 안겨주는 핵심 원인**으로 작용해야 합니다.\n이 제약 조건을 절대 벗어나지 마십시오.", "final_instruction": "창의적이고 깊이 있는 소설이 될 수 있도록 상세하게 기획하세요." }, "English": { "title": "You are a literary director planning a 30-page novella.", "user_theme": "User Theme", "plan_instruction": "Create a detailed novel plan including:", "theme_section": "1. **Theme and Genre**\n - Core theme and message (Deeply reflect user's intent)\n - Genre and atmosphere", "char_section": "2. **Main Characters** (3-5)\n | Name | Role | Personality | Background | Goal | Conflict |", "setting_section": "3. **Setting**\n - Time and place\n - Social/cultural environment", "plot_section": "4. **Plot Structure** (10 parts, ~3 pages each)\n | Part | Pages | Main Events | Tension | Character Development |", "guideline_section": "5. **Writer Guidelines**\n - Key settings for consistency\n - Style and tone guidelines", "constraint_title": "⚠️CRITICAL INSTRUCTION: CORE CONSTRAINTS⚠️", "constraint_body": "This is **NOT a simple, positive story where AI solves everything.**\nYou must develop the narrative around the core emotions of the user's theme: '{query}' (e.g., anxiety, alienation, loss, generational conflict).\nAI or specific technology should be the **root cause of the protagonist's conflict and loss**, not a convenient tool.\nDo not deviate from this constraint.", "final_instruction": "Plan in detail for a creative and profound novel." } } p = lang_prompts[language] return f"{p['title']}\n\n{p['user_theme']}: {user_query}\n\n{search_results_str}\n\n{p['plan_instruction']}\n\n{p['theme_section']}\n\n{p['char_section']}\n\n{p['setting_section']}\n\n{p['plot_section']}\n\n{p['guideline_section']}\n\n---\n{p['constraint_title']}\n{p['constraint_body'].format(query=user_query)}\n---\n\n{p['final_instruction']}" def create_critic_director_prompt(self, director_plan: str, user_query: str, language: str) -> str: """비평가의 감독자 기획 검토 프롬프트 (테마 일관성 강화)""" lang_prompts = { "Korean": { "title": "당신은 문학 비평가입니다. 감독자의 소설 기획을 '주제 일관성'과 '기술적 일관성' 관점에서 검토하세요.", "theme_check": f"**1. 주제 일관성 (가장 중요)**\n - **원래 주제:** '{user_query}'\n - 기획안이 주제의 핵심 감정(불안, 상실감 등)에서 벗어나 긍정적이거나 단순한 방향으로 흐르지 않았습니까?\n - AI나 기술이 갈등의 원인이 아닌, 단순 해결사로 묘사되지 않았습니까?", "consistency_check": "**2. 기술적 일관성**\n - 캐릭터 설정의 모순, 플롯의 논리적 허점, 시간선/공간 설정의 문제점을 검토하세요.", "instruction": "위 항목들을 중심으로 구체적인 문제점과 개선안을 제시하세요." }, "English": { "title": "You are a literary critic. Review the director's plan from the perspectives of 'Thematic Consistency' and 'Technical Consistency'.", "theme_check": f"**1. Thematic Consistency (Most Important)**\n - **Original Theme:** '{user_query}'\n - Does the plan drift from the core emotions (e.g., anxiety, loss) towards an overly positive or simplistic narrative?\n - Is AI depicted as a simple problem-solver instead of the root of the conflict?", "consistency_check": "**2. Technical Consistency**\n - Review for character contradictions, plot holes, and timeline/setting issues.", "instruction": "Provide specific problems and suggestions for improvement based on the above." } } p = lang_prompts[language] return f"{p['title']}\n\n**감독자 기획:**\n{director_plan}\n\n---\n**검토 항목:**\n{p['theme_check']}\n\n{p['consistency_check']}\n\n{p['instruction']}" def create_director_revision_prompt(self, initial_plan: str, critic_feedback: str, user_query: str, language: str) -> str: """감독자 수정 프롬프트 (테마 제약 조건 재강조)""" return f"""감독자로서 비평가의 피드백을 반영하여 소설 기획을 수정합니다. **원래 주제:** {user_query} **초기 기획:**\n{initial_plan} **비평가 피드백:**\n{critic_feedback} **수정 지침:** - 비평가가 지적한 모든 일관성 문제와 주제 이탈 문제를 해결하세요. - **핵심 제약 조건**을 다시 한번 상기하고, 소설 전체가 '불안'과 '상실감'의 톤을 유지하도록 플롯을 구체화하세요. - 10명의 작가가 혼동 없이 작업할 수 있도록 명확하고 상세한 최종 마스터플랜을 작성하세요. """ def create_writer_prompt(self, writer_number: int, director_plan: str, previous_content_summary: str, user_query: str, language: str) -> str: """작가 프롬프트 (테마 리마인더 포함)""" pages_start = (writer_number - 1) * 3 + 1 pages_end = writer_number * 3 lang_prompts = { "Korean": { "title": f"당신은 작가 {writer_number}번입니다. 소설의 {pages_start}-{pages_end} 페이지를 작성하세요.", "plan": "감독자 마스터플랜", "prev_summary": "이전 내용 요약", "guidelines": "**작성 지침:**\n1. **분량**: 1,400-1,500 단어 내외\n2. **연결성**: 요약된 이전 내용과 자연스럽게 연결\n3. **일관성**: 캐릭터 설정과 상태, 플롯 구조를 반드시 따를 것", "reminder_title": "⭐ 잊지 마세요 (테마 리마인더)", "reminder_body": f"이 소설의 핵심은 '{user_query}'에 담긴 **불안, 소외, 상실감**입니다. 긍정적인 해결을 서두르지 말고, 주인공의 내면 갈등을 심도 있게 묘사하는 데 집중하세요.", "final_instruction": "창의적이면서도 주제와 일관성을 절대 잃지 마십시오." }, "English": { "title": f"You are Writer #{writer_number}. Write pages {pages_start}-{pages_end} of the novella.", "plan": "Director's Masterplan", "prev_summary": "Previous Content Summary", "guidelines": "**Writing Guidelines:**\n1. **Length**: Approx. 1,400-1,500 words\n2. **Connectivity**: Connect naturally with the summarized previous content.\n3. **Consistency**: Strictly follow character settings, states, and plot structure.", "reminder_title": "⭐ REMINDER (THEME)", "reminder_body": f"The core of this novel is the **anxiety, alienation, and loss** from the theme '{user_query}'. Do not rush to a positive resolution; focus on deeply describing the protagonist's internal conflict.", "final_instruction": "Be creative, but never lose consistency and the core theme." } } p = lang_prompts[language] consistency_info = self.consistency_tracker.get_character_summary(writer_number) + self.consistency_tracker.get_plot_summary(writer_number) return f"{p['title']}\n\n**{p['plan']}:**\n{director_plan}\n\n{consistency_info}\n\n**{p['prev_summary']}:**\n{previous_content_summary}\n\n---\n{p['guidelines']}\n\n**{p['reminder_title']}**\n{p['reminder_body']}\n---\n\n{p['final_instruction']}" def create_critic_consistency_prompt(self, all_content: str, user_query: str, language: str) -> str: """비평가 중간 검토 프롬프트 (테마 검토 강화)""" return f"""당신은 일관성 검토 전문 비평가입니다. 지금까지 작성된 내용을 검토하세요. **원래 주제:** {user_query} **현재까지 작성된 내용 (최근 3000자):**\n{all_content[-3000:]} **검토 항목:** 1. **주제 일관성 (가장 중요):** 내용이 원래 주제의 어두운 감정선에서 벗어나지 않았는지 확인하고, 벗어났다면 수정 방향을 제시하세요. 2. **기술적 일관성:** 캐릭터, 플롯, 설정의 연속성과 논리적 오류를 찾아내세요. 3. **반복 내용:** 의미적으로 중복되는 장면이나 표현이 없는지 확인하세요. **결과:** 발견된 문제점과 구체적인 수정 제안을 목록으로 제시하세요. """ def create_writer_revision_prompt(self, writer_number: int, initial_content: str, consistency_feedback: str, language: str) -> str: """작가 수정 프롬프트""" return f"""작가 {writer_number}번으로서 비평가의 피드백을 반영하여 내용을 수정하세요. **초기 작성 내용:**\n{initial_content} **비평가 피드백:**\n{consistency_feedback} **수정 지침:** - 지적된 모든 주제 이탈 및 일관성 문제를 해결하세요. - 분량(1,400-1,500 단어)을 유지하면서 내용의 질을 높이세요. - 수정된 최종 버전을 제시하세요. """ def create_critic_final_prompt(self, complete_novel: str, language: str) -> str: """최종 비평가 검토 및 보고서 작성 프롬프트""" return f"""완성된 소설의 최종 일관성 및 완성도에 대한 종합 보고서를 작성하세요. **완성된 소설 (마지막 2000자):**\n{complete_novel[-2000:]} **보고서 포함 항목:** 1. **전체 일관성 평가:** 캐릭터, 플롯, 설정, 주제 유지에 대한 점수(1-10)와 총평. 2. **최종 발견된 문제점:** 남아있는 사소한 문제점들. 3. **성공 요소:** 특히 잘 유지된 일관성 부분이나 주제 표현이 뛰어난 부분. 4. **최종 평가:** 소설의 전반적인 완성도와 독자에게 미칠 영향에 대한 평가. """ # --- LLM 호출 함수들 --- def call_llm_sync(self, messages: List[Dict[str, str]], role: str, language: str) -> str: """LLM 동기식 호출 (요약 등 내부용)""" full_content = "" for chunk in self.call_llm_streaming(messages, role, language): full_content += chunk if full_content.startswith("❌"): raise Exception(f"LLM Sync Call Failed: {full_content}") return full_content def call_llm_streaming(self, messages: List[Dict[str, str]], role: str, language: str) -> Generator[str, None, None]: """LLM 스트리밍 호출 (완전한 에러 처리 및 디버깅)""" try: system_prompts = self.get_system_prompts(language) full_messages = [{"role": "system", "content": system_prompts.get(role, "You are a helpful assistant.")}, *messages] payload = { "model": self.model_id, "messages": full_messages, "max_tokens": 10000, "temperature": 0.75, "top_p": 0.9, "presence_penalty": 0.3, "frequency_penalty": 0.2, "stream": True, "stream_options": {"include_usage": True} } logger.info(f"[{role}] API 스트리밍 시작") # API 호출 response = requests.post( self.api_url, headers=self.create_headers(), json=payload, stream=True, timeout=180 ) # 상태 코드 확인 if response.status_code != 200: logger.error(f"API 응답 오류: {response.status_code}") logger.error(f"응답 내용: {response.text[:500]}") yield f"❌ API 오류 (상태 코드: {response.status_code})" return response.raise_for_status() # 스트리밍 처리 buffer = "" total_content = "" chunk_count = 0 error_count = 0 for line in response.iter_lines(): if not line: continue try: line_str = line.decode('utf-8').strip() # SSE 형식 확인 if not line_str.startswith("data: "): continue data_str = line_str[6:] # "data: " 제거 # 스트림 종료 확인 if data_str == "[DONE]": logger.info(f"[{role}] 스트리밍 완료 - 총 {len(total_content)} 문자") break # JSON 파싱 try: data = json.loads(data_str) except json.JSONDecodeError: logger.warning(f"JSON 파싱 실패: {data_str[:100]}") continue # choices 배열 안전하게 확인 choices = data.get("choices", None) if not choices or not isinstance(choices, list) or len(choices) == 0: # 에러 응답 확인 if "error" in data: error_msg = data.get("error", {}).get("message", "Unknown error") logger.error(f"API 에러: {error_msg}") yield f"❌ API 에러: {error_msg}" return continue # delta에서 content 추출 delta = choices[0].get("delta", {}) content = delta.get("content", "") if content: buffer += content total_content += content chunk_count += 1 # 100자 또는 줄바꿈마다 yield if len(buffer) >= 100 or '\n' in buffer: yield buffer buffer = "" time.sleep(0.01) # UI 업데이트를 위한 짧은 대기 except Exception as e: error_count += 1 logger.error(f"청크 처리 오류 #{error_count}: {str(e)}") if error_count > 10: # 너무 많은 에러시 중단 yield f"❌ 스트리밍 중 과도한 오류 발생" return continue # 남은 버퍼 처리 if buffer: yield buffer # 결과 확인 if chunk_count == 0: logger.error(f"[{role}] 콘텐츠가 전혀 수신되지 않음") yield "❌ API로부터 응답을 받지 못했습니다." else: logger.info(f"[{role}] 성공적으로 {chunk_count}개 청크, 총 {len(total_content)}자 수신") except requests.exceptions.Timeout: logger.error("API 요청 시간 초과") yield "❌ API 요청 시간이 초과되었습니다." except requests.exceptions.ConnectionError: logger.error("API 연결 실패") yield "❌ API 서버에 연결할 수 없습니다." except Exception as e: logger.error(f"예기치 않은 오류: {type(e).__name__}: {str(e)}", exc_info=True) yield f"❌ 오류 발생: {str(e)}" def get_system_prompts(self, language: str) -> Dict[str, str]: """역할별 시스템 프롬프트 생성""" base_prompts = { "Korean": { "director": "당신은 창의적이고 체계적인 소설 기획 전문가입니다. 흥미롭고 일관성 있는 스토리를 설계하세요.", "critic": "당신은 일관성 검토 전문 비평가입니다. 캐릭터, 플롯, 설정의 일관성을 철저히 점검하고 개선방안을 제시하세요.", "writer_base": "당신은 전문 소설 작가입니다. 주어진 지침에 따라 몰입감 있고 일관성 있는 내용을 작성하세요." }, "English": { "director": "You are a creative and systematic novel planning expert. Design engaging and consistent stories.", "critic": "You are a consistency review specialist critic. Thoroughly check character, plot, and setting consistency and suggest improvements.", "writer_base": "You are a professional novel writer. Write immersive and consistent content according to the given guidelines." } } prompts = base_prompts[language].copy() # 작가별 특수 프롬프트 if language == "Korean": prompts["writer1"] = "당신은 소설의 매력적인 시작을 담당하는 작가입니다. 독자를 사로잡는 도입부를 만드세요." prompts["writer10"] = "당신은 완벽한 결말을 만드는 작가입니다. 독자에게 깊은 여운을 남기는 마무리를 하세요." else: prompts["writer1"] = "You are a writer responsible for the captivating beginning. Create an opening that hooks readers." prompts["writer10"] = "You are a writer who creates the perfect ending. Create a conclusion that leaves readers with deep resonance." # writer2-9는 기본 프롬프트 사용 for i in range(2, 10): prompts[f"writer{i}"] = prompts["writer_base"] return prompts # --- 메인 프로세스 --- def process_novel_stream(self, query: str, language: str, session_id: Optional[str] = None) -> Generator[Tuple[str, List[Dict[str, Any]], str], None, None]: """소설 생성 스트리밍 프로세스 (강화된 로직)""" try: resume_from_stage = 0 if session_id: self.current_session_id = session_id session = NovelDatabase.get_session(session_id) if session: query = session['user_query'] language = session['language'] resume_from_stage = session['current_stage'] + 1 logger.info(f"Resuming session {session_id} from stage {resume_from_stage}") else: self.current_session_id = NovelDatabase.create_session(query, language) logger.info(f"Created new session: {self.current_session_id}") stages = [] if resume_from_stage > 0: stages = [{ "name": s['stage_name'], "status": s['status'], "content": s.get('content', ''), "consistency_score": s.get('consistency_score', 0.0) } for s in NovelDatabase.get_stages(self.current_session_id)] for stage_idx in range(resume_from_stage, len(OPTIMIZED_STAGES)): role, stage_name = OPTIMIZED_STAGES[stage_idx] if stage_idx >= len(stages): stages.append({"name": stage_name, "status": "active", "content": "", "consistency_score": 0.0}) else: stages[stage_idx]["status"] = "active" yield "", stages, self.current_session_id prompt = self.get_stage_prompt(stage_idx, role, query, language, stages) stage_content = "" for chunk in self.call_llm_streaming([{"role": "user", "content": prompt}], role, language): stage_content += chunk stages[stage_idx]["content"] = stage_content yield "", stages, self.current_session_id consistency_score = 0.0 if role.startswith("writer"): writer_num = int(re.search(r'\d+', role).group()) all_previous = self.get_all_content(stages, stage_idx) errors = self.consistency_tracker.validate_consistency(writer_num, stage_content) consistency_score = max(0, 10 - len(errors) * 2) stages[stage_idx]["consistency_score"] = consistency_score stages[stage_idx]["status"] = "complete" NovelDatabase.save_stage( self.current_session_id, stage_idx, stage_name, role, stage_content, "complete", consistency_score ) yield "", stages, self.current_session_id final_novel = NovelDatabase.get_writer_content(self.current_session_id) final_report = self.generate_consistency_report(final_novel, language) NovelDatabase.update_final_novel(self.current_session_id, final_novel, final_report) yield f"✅ 소설 완성! 총 {len(final_novel.split())}단어", stages, self.current_session_id except Exception as e: logger.error(f"소설 생성 프로세스 오류: {e}", exc_info=True) yield f"❌ 오류 발생: {e}", stages if 'stages' in locals() else [], self.current_session_id def get_stage_prompt(self, stage_idx: int, role: str, query: str, language: str, stages: List[Dict]) -> str: """단계별 프롬프트 생성 (요약 기능 및 주제 전달 강화)""" if stage_idx == 0: return self.create_director_initial_prompt(query, language) if stage_idx == 1: return self.create_critic_director_prompt(stages[0]["content"], query, language) if stage_idx == 2: return self.create_director_revision_prompt(stages[0]["content"], stages[1]["content"], query, language) master_plan = stages[2]["content"] if 3 <= stage_idx <= 12: # 작가 초안 writer_num = stage_idx - 2 previous_content = self.get_all_content(stages, stage_idx) summary = self.create_summary(previous_content, language) return self.create_writer_prompt(writer_num, master_plan, summary, query, language) if stage_idx == 13: # 비평가 중간 검토 all_content = self.get_all_content(stages, stage_idx) return self.create_critic_consistency_prompt(all_content, query, language) if 14 <= stage_idx <= 23: # 작가 수정 writer_num = stage_idx - 13 initial_content = stages[2 + writer_num]["content"] feedback = stages[13]["content"] return self.create_writer_revision_prompt(writer_num, initial_content, feedback, language) if stage_idx == 24: # 최종 검토 complete_novel = self.get_all_writer_content(stages) return self.create_critic_final_prompt(complete_novel, language) return "" def create_summary(self, content: str, language: str) -> str: """LLM을 이용해 이전 내용을 요약""" if not content.strip(): return "이전 내용이 없습니다." if language == "Korean" else "No previous content." prompt_text = "다음 소설 내용을 3~5개의 핵심적인 문장으로 요약해줘. 다음 작가가 이야기를 이어가는 데 필요한 핵심 정보(등장인물의 현재 상황, 감정, 마지막 사건)를 포함해야 해." if language != "Korean": prompt_text = "Summarize the following novel content in 3-5 key sentences. Include crucial information for the next writer to continue the story (characters' current situation, emotions, and the last major event)." summary_prompt = f"{prompt_text}\n\n---\n{content[-2000:]}" try: summary = self.call_llm_sync([{"role": "user", "content": summary_prompt}], "critic", language) return summary except Exception as e: logger.error(f"요약 생성 실패: {e}") return content[-1000:] def get_all_content(self, stages: List[Dict], current_stage: int) -> str: """현재까지의 모든 내용 가져오기""" return "\n\n".join(s["content"] for i, s in enumerate(stages) if i < current_stage and s["content"]) def get_all_writer_content(self, stages: List[Dict]) -> str: """모든 작가 최종 수정본 내용 가져오기""" return "\n\n".join(s["content"] for i, s in enumerate(stages) if 14 <= i <= 23 and s["content"]) def generate_consistency_report(self, complete_novel: str, language: str) -> str: """최종 보고서 생성 (LLM 호출)""" prompt = self.create_critic_final_prompt(complete_novel, language) try: report = self.call_llm_sync([{"role": "user", "content": prompt}], "critic", language) return report except Exception as e: logger.error(f"최종 보고서 생성 실패: {e}") return "보고서 생성 중 오류 발생" # --- 유틸리티 함수들 --- def process_query(query: str, language: str, session_id: Optional[str] = None) -> Generator[Tuple[str, str, str, str], None, None]: """메인 쿼리 처리 함수""" if not query.strip(): yield "", "", "❌ 주제를 입력해주세요.", session_id return system = NovelWritingSystem() stages_markdown = "" novel_content = "" for status, stages, current_session_id in system.process_novel_stream(query, language, session_id): stages_markdown = format_stages_display(stages) # 최종 소설 내용 가져오기 if stages and all(s.get("status") == "complete" for s in stages[-10:]): novel_content = NovelDatabase.get_writer_content(current_session_id) novel_content = format_novel_display(novel_content) yield stages_markdown, novel_content, status or "🔄 처리 중...", current_session_id def get_active_sessions(language: str) -> List[str]: """활성 세션 목록 가져오기""" sessions = NovelDatabase.get_active_sessions() return [f"{s['session_id'][:8]}... - {s['user_query'][:50]}... ({s['created_at']})" for s in sessions] def auto_recover_session(language: str) -> Tuple[Optional[str], str]: """가장 최근 활성 세션 자동 복구""" latest_session = NovelDatabase.get_latest_active_session() if latest_session: return latest_session['session_id'], f"세션 {latest_session['session_id'][:8]}... 복구됨" return None, "복구할 세션이 없습니다." def resume_session(session_id: str, language: str) -> Generator[Tuple[str, str, str, str], None, None]: """세션 재개 함수""" if not session_id: yield "", "", "❌ 세션 ID가 없습니다.", session_id return # 드롭다운에서 세션 ID 추출 if "..." in session_id: session_id = session_id.split("...")[0] session = NovelDatabase.get_session(session_id) if not session: yield "", "", "❌ 세션을 찾을 수 없습니다.", None return # process_query를 통해 재개 yield from process_query(session['user_query'], session['language'], session_id) def download_novel(novel_text: str, format_type: str, language: str, session_id: str) -> Optional[str]: """소설 다운로드 파일 생성""" if not novel_text or not session_id: return None timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"novel_{session_id[:8]}_{timestamp}" try: if format_type == "DOCX" and DOCX_AVAILABLE: return export_to_docx(novel_text, filename, language) else: return export_to_txt(novel_text, filename) except Exception as e: logger.error(f"파일 생성 실패: {e}") return None def format_stages_display(stages: List[Dict]) -> str: """단계별 진행 상황 마크다운 포맷팅""" markdown = "## 🎬 진행 상황\n\n" for i, stage in enumerate(stages): status_icon = "✅" if stage['status'] == 'complete' else "🔄" if stage['status'] == 'active' else "⏳" markdown += f"{status_icon} **{stage['name']}**" if stage.get('consistency_score', 0) > 0: markdown += f" (일관성: {stage['consistency_score']:.1f}/10)" markdown += "\n" if stage['content']: preview = stage['content'][:200] + "..." if len(stage['content']) > 200 else stage['content'] markdown += f"> {preview}\n\n" return markdown def format_novel_display(novel_text: str) -> str: """소설 내용 마크다운 포맷팅""" if not novel_text: return "아직 완성된 내용이 없습니다." # 페이지 구분 추가 pages = novel_text.split('\n\n') formatted = "# 📖 완성된 소설\n\n" for i, page in enumerate(pages): if page.strip(): formatted += f"### 페이지 {i+1}\n\n{page}\n\n---\n\n" return formatted def export_to_docx(content: str, filename: str, language: str) -> str: """DOCX 파일로 내보내기""" doc = Document() # 제목 추가 title = doc.add_heading('AI 협업 소설', 0) title.alignment = WD_ALIGN_PARAGRAPH.CENTER # 메타데이터 doc.add_paragraph(f"생성일: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") doc.add_paragraph(f"언어: {language}") doc.add_page_break() # 본문 추가 paragraphs = content.split('\n\n') for para in paragraphs: if para.strip(): doc.add_paragraph(para.strip()) # 파일 저장 filepath = f"{filename}.docx" doc.save(filepath) return filepath def export_to_txt(content: str, filename: str) -> str: """TXT 파일로 내보내기""" filepath = f"{filename}.txt" with open(filepath, 'w', encoding='utf-8') as f: f.write(content) return filepath # CSS 스타일 custom_css = """ .gradio-container { background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); min-height: 100vh; } .main-header { background-color: rgba(255, 255, 255, 0.1); backdrop-filter: blur(10px); padding: 30px; border-radius: 12px; margin-bottom: 30px; text-align: center; color: white; } .input-section { background-color: rgba(255, 255, 255, 0.1); backdrop-filter: blur(10px); padding: 20px; border-radius: 12px; margin-bottom: 20px; } .session-section { background-color: rgba(255, 255, 255, 0.1); backdrop-filter: blur(10px); padding: 15px; border-radius: 8px; margin-top: 20px; color: white; } #stages-display { background-color: rgba(255, 255, 255, 0.95); padding: 20px; border-radius: 12px; max-height: 600px; overflow-y: auto; } #novel-output { background-color: rgba(255, 255, 255, 0.95); padding: 30px; border-radius: 12px; max-height: 400px; overflow-y: auto; } .download-section { background-color: rgba(255, 255, 255, 0.9); padding: 15px; border-radius: 8px; margin-top: 20px; } """ # Gradio 인터페이스 생성 def create_interface(): with gr.Blocks(css=custom_css, title="AI 협업 소설 생성 시스템") as interface: gr.HTML("""

📚 AI 협업 소설 생성 시스템

일관성 중심의 창의적 소설 생성

주제를 입력하면 AI 에이전트들이 협업하여 30페이지 분량의 완성된 소설을 생성합니다.
감독자 1명, 비평가 1명, 작가 10명이 함께 작업하며 일관성을 유지합니다.

""") # 상태 관리 current_session_id = gr.State(None) with gr.Row(): with gr.Column(scale=1): with gr.Group(elem_classes=["input-section"]): query_input = gr.Textbox( label="소설 주제 / Novel Theme", placeholder="소설의 주제나 초기 아이디어를 입력하세요...\nEnter your novel theme or initial idea...", lines=4 ) language_select = gr.Radio( choices=["English", "Korean"], value="English", label="언어 / Language" ) with gr.Row(): submit_btn = gr.Button("🚀 소설 생성 시작", variant="primary", scale=2) clear_btn = gr.Button("🗑️ 초기화", scale=1) status_text = gr.Textbox( label="상태", interactive=False, value="🔄 준비 완료" ) # 세션 관리 with gr.Group(elem_classes=["session-section"]): gr.Markdown("### 💾 이전 세션 재개") session_dropdown = gr.Dropdown( label="세션 선택", choices=[], interactive=True ) with gr.Row(): refresh_btn = gr.Button("🔄 목록 새로고침", scale=1) resume_btn = gr.Button("▶️ 선택 재개", variant="secondary", scale=1) auto_recover_btn = gr.Button("♻️ 자동 복구", scale=1) with gr.Column(scale=2): with gr.Tab("📝 작성 과정"): stages_display = gr.Markdown( value="작성 과정이 여기에 표시됩니다...", elem_id="stages-display" ) with gr.Tab("📖 완성된 소설"): novel_output = gr.Markdown( value="완성된 소설이 여기에 표시됩니다...", elem_id="novel-output" ) with gr.Group(elem_classes=["download-section"]): gr.Markdown("### 📥 소설 다운로드") with gr.Row(): format_select = gr.Radio( choices=["DOCX", "TXT"], value="DOCX" if DOCX_AVAILABLE else "TXT", label="형식" ) download_btn = gr.Button("⬇️ 다운로드", variant="secondary") download_file = gr.File( label="다운로드된 파일", visible=False ) # 숨겨진 상태 novel_text_state = gr.State("") # 예제 with gr.Row(): gr.Examples( examples=[ ["미래 도시에서 기억을 거래하는 상인의 이야기"], ["시간이 거꾸로 흐르는 마을의 미스터리"], ["A scientist discovers a portal to parallel universes"], ["In a world where dreams can be traded, a dream thief's story"], ["Two AI entities fall in love while preventing a cyber war"], ["책 속으로 들어갈 수 있는 능력을 가진 사서의 모험"] ], inputs=query_input, label="💡 예제 주제" ) # 이벤트 핸들러 def refresh_sessions(): try: sessions = get_active_sessions("English") return gr.update(choices=sessions) except Exception as e: logger.error(f"Error refreshing sessions: {str(e)}") return gr.update(choices=[]) def handle_auto_recover(language): session_id, message = auto_recover_session(language) return session_id # 이벤트 연결 submit_btn.click( fn=process_query, inputs=[query_input, language_select, current_session_id], outputs=[stages_display, novel_output, status_text, current_session_id] ) novel_output.change( fn=lambda x: x, inputs=[novel_output], outputs=[novel_text_state] ) resume_btn.click( fn=lambda x: x, inputs=[session_dropdown], outputs=[current_session_id] ).then( fn=resume_session, inputs=[current_session_id, language_select], outputs=[stages_display, novel_output, status_text, current_session_id] ) auto_recover_btn.click( fn=handle_auto_recover, inputs=[language_select], outputs=[current_session_id] ).then( fn=resume_session, inputs=[current_session_id, language_select], outputs=[stages_display, novel_output, status_text, current_session_id] ) refresh_btn.click( fn=refresh_sessions, outputs=[session_dropdown] ) clear_btn.click( fn=lambda: ("", "", "🔄 준비 완료", "", None), outputs=[stages_display, novel_output, status_text, novel_text_state, current_session_id] ) def handle_download(format_type, language, session_id, novel_text): if not session_id: return gr.update(visible=False) file_path = download_novel(novel_text, format_type, language, session_id) if file_path: return gr.update(value=file_path, visible=True) else: return gr.update(visible=False) download_btn.click( fn=handle_download, inputs=[format_select, language_select, current_session_id, novel_text_state], outputs=[download_file] ) # 시작 시 세션 로드 interface.load( fn=refresh_sessions, outputs=[session_dropdown] ) return interface # 메인 실행 if __name__ == "__main__": logger.info("AI 협업 소설 생성 시스템 시작...") logger.info("=" * 60) # 환경 확인 logger.info(f"API 엔드포인트: {API_URL}") if BRAVE_SEARCH_API_KEY: logger.info("웹 검색이 활성화되었습니다.") else: logger.warning("웹 검색이 비활성화되었습니다.") if DOCX_AVAILABLE: logger.info("DOCX 내보내기가 활성화되었습니다.") else: logger.warning("DOCX 내보내기가 비활성화되었습니다.") logger.info("=" * 60) # 데이터베이스 초기화 logger.info("데이터베이스 초기화 중...") NovelDatabase.init_db() logger.info("데이터베이스 초기화 완료.") # 인터페이스 생성 및 실행 interface = create_interface() interface.launch( server_name="0.0.0.0", server_port=7860, share=False, debug=True )