File size: 8,084 Bytes
a61ba58 a402ff7 a61ba58 a402ff7 4c8df65 a61ba58 4c8df65 a61ba58 1e8f4c6 a61ba58 1e8f4c6 a61ba58 1e8f4c6 78b81a5 a61ba58 1e8f4c6 eec194a 1e8f4c6 a61ba58 1e8f4c6 a61ba58 4c8df65 a61ba58 1e8f4c6 a61ba58 ddf672c b239fb1 ddf672c 78b81a5 ddf672c 78b81a5 ddf672c 78b81a5 ddf672c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 |
from typing import List, Dict
from core.constants import GameConfig
from services.mistral_client import MistralClient
from api.models import StoryResponse, Choice
from core.generators.story_segment_generator import StorySegmentGenerator
from core.generators.image_prompt_generator import ImagePromptGenerator
from core.generators.metadata_generator import MetadataGenerator
from core.game_state import GameState
import random
from core.constants import GameConfig
class StoryGenerator:
_instance = None
def __new__(cls, *args, **kwargs):
if cls._instance is None:
print("Creating new StoryGenerator instance")
cls._instance = super().__new__(cls)
cls._instance._initialized = False
return cls._instance
def __init__(self, api_key: str, model_name: str = "mistral-small"):
if not self._initialized:
print("Initializing StoryGenerator singleton")
self.api_key = api_key
self.model_name = model_name
self.turn_before_end = random.randint(GameConfig.MIN_SEGMENTS_BEFORE_END, GameConfig.MAX_SEGMENTS_BEFORE_END)
self.is_winning_story = random.random() < GameConfig.WINNING_STORY_CHANCE
# Client principal avec limite standard
self.mistral_client = MistralClient(api_key=api_key, model_name=model_name)
# Client spécifique pour les segments d'histoire avec limite plus basse
self.story_segment_client = MistralClient(api_key=api_key, model_name=model_name, max_tokens=50)
self.image_prompt_generator = None # Will be initialized with the first universe style
self.metadata_generator = None # Will be initialized with hero description
self.segment_generators: Dict[str, StorySegmentGenerator] = {}
self._initialized = True
def create_segment_generator(self, session_id: str, style: dict, genre: str, epoch: str, base_story: str, macguffin: str, hero_name: str, hero_desc: str):
"""Create a new StorySegmentGenerator adapted to the specified universe for a given session."""
try:
# Use selected_artist if available, otherwise get the first artist from references
if "selected_artist" in style:
artist = style["selected_artist"]
else:
artist = style["references"][0]["artist"]
# Create a detailed artist style string
artist_style = f"{style['name']}, {genre} in {epoch}"
# Always create a new ImagePromptGenerator for each session with the correct artist and hero
self.image_prompt_generator = ImagePromptGenerator(
self.mistral_client,
artist_style=artist_style,
hero_name=hero_name,
hero_desc=hero_desc,
universe_style=style["name"],
universe_genre=genre,
universe_epoch=epoch
)
# Create a new MetadataGenerator with hero description
self.metadata_generator = MetadataGenerator(
self.mistral_client,
hero_name=hero_name,
hero_desc=hero_desc
)
# Create a new StorySegmentGenerator with all universe parameters
self.segment_generators[session_id] = StorySegmentGenerator(
self.story_segment_client,
universe_style=style["name"],
universe_genre=genre,
universe_epoch=epoch,
universe_story=base_story,
universe_macguffin=macguffin,
hero_name=hero_name,
hero_desc=hero_desc
)
# print(f"Current StorySegmentGenerators in StoryGenerator: {list(self.segment_generators.keys())}")
except KeyError as e:
print(f"Error accessing style data: {e}")
print(f"Style object received: {style}")
raise ValueError(f"Invalid style format: {str(e)}")
except Exception as e:
print(f"Unexpected error in create_segment_generator: {str(e)}")
raise
def get_segment_generator(self, session_id: str) -> StorySegmentGenerator:
"""Get the StorySegmentGenerator associated with a session."""
# print(f"Getting StorySegmentGenerator for session {session_id} from StoryGenerator singleton")
# print(f"Current StorySegmentGenerators in StoryGenerator: {list(self.segment_generators.keys())}")
if session_id not in self.segment_generators:
raise RuntimeError(f"No story segment generator found for session {session_id}. Generate a universe first.")
return self.segment_generators[session_id]
async def generate_story_segment(self, session_id: str, game_state: GameState, previous_choice: str) -> StoryResponse:
try:
# On utilise toujours le générateur de segments, même pour un choix personnalisé
segment_generator = self.get_segment_generator(session_id)
if not segment_generator:
raise ValueError("No story segment generator found for this session")
if(game_state.story_beat == GameConfig.STORY_BEAT_INTRO):
story_text = game_state.universe_story
else:
segment_response = await segment_generator.generate(
story_beat=game_state.story_beat,
current_time=game_state.current_time,
current_location=game_state.current_location,
previous_choice=previous_choice,
story_history=game_state.format_history(),
turn_before_end=self.turn_before_end,
is_winning_story=self.is_winning_story
)
story_text = segment_response.story_text
# Then get metadata using the new story text
metadata_response = await self.metadata_generator.generate(
story_text=story_text,
current_time=game_state.current_time,
current_location=game_state.current_location,
story_beat=game_state.story_beat,
turn_before_end=self.turn_before_end,
is_winning_story=self.is_winning_story,
story_history=game_state.format_history()
)
# Generate image prompts
prompts_response = await self.image_prompt_generator.generate(
story_text=story_text,
time=metadata_response.time,
location=metadata_response.location,
is_death=metadata_response.is_death,
is_victory=metadata_response.is_victory,
turn_before_end=self.turn_before_end,
is_winning_story=self.is_winning_story
)
# Create choices
choices = [
Choice(id=i, text=choice_text)
for i, choice_text in enumerate(metadata_response.choices, 1)
]
response = StoryResponse(
story_text=story_text,
choices=choices,
raw_choices=metadata_response.choices,
time=metadata_response.time,
location=metadata_response.location,
image_prompts=prompts_response.image_prompts,
is_first_step=(game_state.story_beat == GameConfig.STORY_BEAT_INTRO),
is_death=metadata_response.is_death,
is_victory=metadata_response.is_victory,
previous_choice=previous_choice
)
# Add the response to game state history
game_state.add_to_history(response)
return response
except Exception as e:
print(f"Unexpected error in generate_story_segment: {str(e)}")
raise |