"""Use case : chat conversationnel LoreMind avec Structural Context. Construit un system prompt riche à partir de 4 contextes possibles (Lore, Page focalisée, Campagne, entité narrative focalisée) puis délègue au port `LLMChatProvider` pour le streaming token par token. Ne charge PAS le contenu détaillé des pages — l'IA doit savoir ce qui existe, pas être noyée sous le texte. Pattern "Structural Context", plus simple que le RAG sémantique tant que les univers restent de taille humaine. Combinaisons supportées : - lore seul → chat Lore (page-edit / page-create) - lore + page_context → chat Lore focalisé page - campaign (+lore si liée) + optional narrative_entity → chat Campagne """ from typing import AsyncIterator from app.domain.models import ( ArcSummary, CampaignStructuralContext, ChatMessage, ChapterSummary, CharacterSummary, NpcSummary, GameSystemContext, JournalEntrySummary, LoreStructuralContext, NarrativeEntityContext, PageContext, PageSummary, QuestSummary, SessionContext, ) from app.application.prompts import chat as prompts from app.core.language import DEFAULT as _DEFAULT_LANG from app.domain.ports import LLMChatProvider # Température moyenne : chat conversationnel créatif mais cohérent. # Plus élevée que le one-shot (0.4) car on veut de la variété d'idées, # mais sans partir en délire halluciné (1.0+). _DEFAULT_TEMPERATURE = 0.7 class ChatUseCase: """Orchestre un tour de conversation avec le LLM + contextes structurels.""" def __init__(self, llm: LLMChatProvider) -> None: self._llm = llm async def stream( self, messages: list[ChatMessage], *, lore_context: LoreStructuralContext | None = None, page_context: PageContext | None = None, campaign_context: CampaignStructuralContext | None = None, narrative_entity: NarrativeEntityContext | None = None, game_system_context: GameSystemContext | None = None, session_context: SessionContext | None = None, language: str = _DEFAULT_LANG, ) -> AsyncIterator[str]: """Streame les tokens de la réponse assistant pour le dernier message user. Les contextes sont tous optionnels, mais au moins l'un des deux "niveaux haut" (lore_context ou campaign_context) doit être fourni pour que le prompt ait du sens. Le controller (main.py) applique cette règle à la frontière HTTP. `language` pilote la langue de réponse. """ system_prompt = self._build_system_prompt( lore_context, page_context, campaign_context, narrative_entity, game_system_context, session_context, language, ) async for token in self._llm.stream_chat( messages, system_prompt=system_prompt, temperature=_DEFAULT_TEMPERATURE, ): yield token def build_system_prompt( self, lore_context: LoreStructuralContext | None = None, page_context: PageContext | None = None, campaign_context: CampaignStructuralContext | None = None, narrative_entity: NarrativeEntityContext | None = None, game_system_context: GameSystemContext | None = None, session_context: SessionContext | None = None, language: str = _DEFAULT_LANG, ) -> str: """Version publique — utilisée par le controller HTTP pour compter les tokens du system prompt avant de streamer (jauge de contexte). """ return self._build_system_prompt( lore_context, page_context, campaign_context, narrative_entity, game_system_context, session_context, language, ) # --- Construction du system prompt -------------------------------------- def _build_system_prompt( self, lore: LoreStructuralContext | None, page: PageContext | None, campaign: CampaignStructuralContext | None, narrative: NarrativeEntityContext | None, game_system: GameSystemContext | None = None, session: SessionContext | None = None, language: str = _DEFAULT_LANG, ) -> str: sections = [prompts.base_system(language)] if lore is not None: sections.append(self._format_lore(lore)) if campaign is not None: sections.append(self._format_campaign(campaign, lore_present=lore is not None)) if game_system is not None: sections.append(self._format_game_system(game_system)) if page is not None: sections.append(self._format_page(page)) if narrative is not None: sections.append(self._format_narrative_entity(narrative)) if session is not None: sections.append(self._format_session(session)) return "\n\n".join(sections) # --- Blocs Lore --------------------------------------------------------- @staticmethod def _format_lore(ctx: LoreStructuralContext) -> str: desc = f"\nDescription : {ctx.lore_description}" if ctx.lore_description else "" folders_block = ChatUseCase._format_folders(ctx.folders) tags_line = ", ".join(ctx.tags) if ctx.tags else "(aucun)" return ( "--- UNIVERS (Lore) ---\n" f"Nom : {ctx.lore_name}{desc}\n\n" f"Organisation :\n{folders_block}\n\n" f"Tags déjà utilisés : {tags_line}" ) @staticmethod def _format_folders(folders: dict[str, list[PageSummary]]) -> str: """Rend chaque page avec son contenu exploitable par le LLM. Depuis b9 : affiche en plus des champs values/tags/pages liées sous forme d'une fiche indentée par page, et seulement si l'info existe (prompt compact quand une page est vierge). """ if not folders: return "(Lore vide pour l'instant)" lines: list[str] = [] for folder_name, pages in folders.items(): lines.append(f"- {folder_name} (dossier)") if not pages: lines.append(" (vide)") continue for ps in pages: lines.append(f" - {ps.title} [template: {ps.template_name}]") for field_name, value in ps.values.items(): lines.append(f" · {field_name} : {value}") if ps.tags: lines.append(f" · tags : {', '.join(ps.tags)}") if ps.related_page_titles: lines.append( " · liée à : " + ", ".join(ps.related_page_titles) ) return "\n".join(lines) @staticmethod def _format_page(pc: PageContext) -> str: """Bloc "PAGE EN COURS" — oriente l'IA vers la page précise éditée.""" if pc.template_fields: fields_block = "\n".join( f'- "{f}" : {pc.values.get(f) or "(vide)"}' for f in pc.template_fields ) else: fields_block = "(aucun champ défini dans ce template)" return ( "--- PAGE EN COURS D'ÉDITION ---\n" f"Titre : {pc.title}\n" f"Template : {pc.template_name}\n" f"Champs et valeurs actuelles :\n{fields_block}\n\n" "IMPORTANT : concentre-toi EXCLUSIVEMENT sur cette page. " "Si l'utilisateur te demande de proposer des idées, elles doivent " "concerner UNIQUEMENT les champs listés ci-dessus. Ne déborde pas " "vers d'autres pages ou d'autres templates du Lore, même si ça te " "semblerait pertinent." ) # --- Blocs Campagne ----------------------------------------------------- @staticmethod def _format_campaign(ctx: CampaignStructuralContext, *, lore_present: bool) -> str: desc = f"\nDescription : {ctx.campaign_description}" if ctx.campaign_description else "" arcs_block = ChatUseCase._format_arcs(ctx.arcs) lore_note = ( "\n(Cette campagne est liée à l'univers ci-dessus : tu peux t'appuyer dessus.)" if lore_present else "\n(Cette campagne n'est associée à aucun univers — tu peux proposer des éléments d'ambiance libres.)" ) characters_block = ChatUseCase._format_characters(ctx.characters) npcs_block = ChatUseCase._format_npcs(ctx.npcs) return ( "--- CAMPAGNE COURANTE ---\n" f"Nom : {ctx.campaign_name}{desc}{lore_note}\n" f"{characters_block}" f"{npcs_block}\n" "Structure narrative (les flèches → indiquent des transitions de scène " "déclenchées par un choix des joueurs) :\n" f"{arcs_block}" ) @staticmethod def _format_characters(characters: list[CharacterSummary]) -> str: """Bloc PJ — liste nom + snippet. Rappel anti-hallucination IA. Si la campagne n'a aucun PJ, on le signale explicitement : l'IA ne doit pas inventer "les héros" ou leurs noms dans ses suggestions. """ if not characters: return ( "\nPersonnages joueurs : aucune fiche pour l'instant. Ne suppose " "ni noms ni classes pour les PJ tant que le MJ ne les a pas créés.\n" ) lines = ["\nPersonnages joueurs (PJ) :"] for c in characters: if c.snippet: lines.append(f"- **{c.name}** — {c.snippet}") else: lines.append(f"- **{c.name}** (fiche vide)") lines.append( "Pour une fiche complète (stats, backstory), n'invente rien : " "demande au MJ d'ouvrir l'éditeur du PJ pour te donner les détails." ) return "\n".join(lines) + "\n" @staticmethod def _format_npcs(npcs: list[NpcSummary]) -> str: """Bloc PNJ — symétrique aux PJ avec sa propre instruction anti-halluci. Distinction importante : pour les PNJ, l'IA est ENCOURAGÉE à proposer de nouveaux PNJ (création créative = OK). En revanche, elle ne doit pas référencer comme existant un PNJ qui n'est pas dans la liste ci-dessous. """ if not npcs: return ( "\nPersonnages non-joueurs (PNJ) : aucun défini pour l'instant. " "Tu peux librement proposer de nouveaux PNJ au MJ, mais ne " "fais pas comme s'ils existaient déjà dans la campagne.\n" ) lines = ["\nPersonnages non-joueurs (PNJ) connus :"] for n in npcs: if n.snippet: lines.append(f"- **{n.name}** — {n.snippet}") else: lines.append(f"- **{n.name}** (fiche vide)") lines.append( "Pour une fiche complète d'un PNJ existant (apparence, motivations), " "n'invente rien : demande au MJ d'ouvrir l'éditeur du PNJ. Tu peux " "en revanche proposer librement de NOUVEAUX PNJ." ) return "\n".join(lines) + "\n" @staticmethod def _format_arcs(arcs: list[ArcSummary]) -> str: if not arcs: return "(Aucun arc créé pour l'instant.)" lines: list[str] = [] for arc in arcs: lines.append(f"- {arc.name} (arc){ChatUseCase._illustration_hint(arc.illustration_count)}") if arc.description: lines.append(f" Synopsis : {arc.description}") if not arc.chapters: lines.append(" (aucun chapitre)") continue for chapter in arc.chapters: lines.extend(ChatUseCase._format_chapter_block(chapter)) return "\n".join(lines) @staticmethod def _format_chapter_block(chapter: ChapterSummary) -> list[str]: hint = ChatUseCase._illustration_hint(chapter.illustration_count) block = [f" - {chapter.name} (chapitre){hint}"] if chapter.description: block.append(f" Synopsis : {chapter.description}") if not chapter.scenes: block.append(" (aucune scène)") else: for scene in chapter.scenes: sc_hint = ChatUseCase._illustration_hint(scene.illustration_count) scene_kind = " (lieu explorable)" if scene.rooms else " (scène)" block.append(f" - {scene.name}{scene_kind}{sc_hint}") if scene.description: block.append(f" Description : {scene.description}") for br in scene.branches: cond = f" (si : {br.condition})" if br.condition else "" block.append( f' → "{br.label}" vers {br.target_scene_name}{cond}' ) # Pièces du lieu explorable (mode donjon) for room in scene.rooms: floor = f" [étage {room.floor}]" if room.floor is not None else "" block.append(f" ◆ {room.name}{floor}") if room.description: block.append(f" {room.description}") if room.enemies: block.append(f" Ennemis : {room.enemies}") for rb in room.branches: cond = f" (si : {rb.condition})" if rb.condition else "" block.append( f' ↳ "{rb.label}" vers {rb.target_room_name}{cond}' ) return block @staticmethod def _illustration_hint(count: int) -> str: """Rend " [N illustrations]" si count > 0, sinon chaine vide. Informe l'IA que l'entite a deja un support visuel. Permet de prioriser les suggestions ecrites qui collent a l'existant visuel plutot que de diverger. """ if count <= 0: return "" noun = "illustration" if count == 1 else "illustrations" return f" [{count} {noun}]" # --- Bloc Système de JDR ------------------------------------------------ @staticmethod def _format_game_system(gs: GameSystemContext) -> str: """Bloc des règles du système de JDR de la campagne. Les sections ont été filtrées côté Core selon l'intent (combat, classes, lore...). Si aucune section n'a matché, on affiche juste le nom du système comme rappel de cadre. """ desc = f"\nDescription : {gs.system_description}" if gs.system_description else "" if not gs.sections: return ( "--- SYSTÈME DE JDR ---\n" f"Nom : {gs.system_name}{desc}\n" "(Aucune section de règles pertinente pour ce type de génération — " "reste cohérent avec l'univers et les conventions du système.)" ) sections_block = "\n\n".join( f"### {title}\n{content}" for title, content in gs.sections.items() ) return ( "--- SYSTÈME DE JDR ---\n" f"Nom : {gs.system_name}{desc}\n\n" "Respecte scrupuleusement les règles et conventions ci-dessous quand " "tu proposes des stats, classes, rencontres, mécaniques ou éléments " "d'ambiance. Les noms propres (classes, sorts, monstres) doivent " "venir de ces règles — n'en invente pas d'autres.\n\n" f"{sections_block}" ) # --- Bloc Session de jeu (Play Context) --------------------------------- _ENTRY_TYPE_LABELS = { "NOTE": "Note du MJ", "EVENT": "Évènement", "DICE_ROLL": "Jet de dés", "PLAYER_ACTION": "Action joueur", } @staticmethod def _format_session(sc: SessionContext) -> str: """Bloc journal de la session en cours + résumé des sessions précédentes. Fournit à l'IA le contexte temporel : ce qui s'est passé jusqu'ici, dans l'ordre chronologique. Permet de référencer un PNJ rencontré, rappeler un évènement antérieur, ou rebondir sur une action joueur. Pour les sessions PRÉCÉDENTES de la même campagne, on ne remonte que les EVENTs (les moments marquants) pour préserver le contexte LLM. """ status = "EN COURS" if sc.active else "TERMINÉE" started = f" — démarrée {sc.started_at}" if sc.started_at else "" previous_block = ChatUseCase._format_previous_events(sc.previous_events) hub_block = ChatUseCase._format_hub_status(sc) if not sc.entries: current_block = "(Aucune entrée dans le journal pour l'instant — la session vient de commencer.)" else: lines: list[str] = [] for e in sc.entries: label = ChatUseCase._ENTRY_TYPE_LABELS.get(e.type, e.type) ts = f" [{e.occurred_at}]" if e.occurred_at else "" content = e.content.replace("\n", "\n ") lines.append(f"- {label}{ts} : {content}") current_block = "\n".join(lines) return ( "--- SESSION DE JEU EN COURS ---\n" f"Nom : {sc.session_name}\n" f"Statut : {status}{started}\n" f"{hub_block}" f"{previous_block}" "\nJournal chronologique de la session courante (du plus ancien au plus récent) :\n" f"{current_block}\n\n" "IMPORTANT : tu es l'assistant du MJ PENDANT la partie. Tes réponses doivent :\n" "- Tenir compte des évènements déjà capturés (sessions précédentes + journal courant).\n" "- Être concrètes et utiles en temps réel : descriptions sensorielles, " "réactions de PNJ cohérentes avec leur fiche, suggestions de complications " "qui s'enchaînent à ce qui vient de se passer.\n" "- Éviter les longs développements : le MJ est en train d'animer une partie, " "il a besoin d'idées immédiatement actionnables." ) @staticmethod def _format_hub_status(sc: SessionContext) -> str: """Bloc Hub : quêtes ouvertes + flags actifs. Vide si la campagne n'a aucun Arc HUB (toutes les listes vides côté Core). Les quêtes LOCKED apparaissent par leur TITRE uniquement : l'IA sait qu'elles existent (utile pour les teasers, les rumeurs en jeu) mais ne peut pas spoiler leurs détails. """ if (not sc.available_quests and not sc.in_progress_quests and not sc.locked_quest_titles and not sc.active_flags): return "" lines = ["", "État du Hub (quêtes parallèles et faits narratifs) :"] if sc.in_progress_quests: lines.append(" Quêtes en cours :") lines.extend(ChatUseCase._format_quest_lines(sc.in_progress_quests)) if sc.available_quests: lines.append(" Quêtes disponibles (non démarrées, prêtes à être lancées) :") lines.extend(ChatUseCase._format_quest_lines(sc.available_quests)) if sc.locked_quest_titles: titles = ", ".join(f'"{t}"' for t in sc.locked_quest_titles) lines.append( " Quêtes encore verrouillées (existent mais non accessibles — " f"tu peux y faire allusion sous forme de rumeurs sans spoiler leur contenu) : {titles}" ) if sc.active_flags: lines.append( " Faits actifs : " + ", ".join(f"`{f}`" for f in sc.active_flags) ) lines.append( " Conseille des actions cohérentes avec ces quêtes ouvertes. Ne fais " "PAS comme si une quête verrouillée était déjà accessible aux PJ." ) lines.append("") # séparateur visuel avant le récap des sessions précédentes return "\n".join(lines) @staticmethod def _format_quest_lines(quests: list[QuestSummary]) -> list[str]: """Sérialise une liste de QuestSummary en lignes indentées.""" out: list[str] = [] for q in quests: arc = f" [arc : {q.arc_name}]" if q.arc_name else "" out.append(f" - {q.name}{arc}") if q.description: out.append(f" Synopsis : {q.description}") return out @staticmethod def _format_previous_events(events: list[JournalEntrySummary]) -> str: """Bloc "Story so far" : EVENTs marquants des sessions antérieures. Vide si la campagne en est à sa première session. On groupe par session source pour aider l'IA à situer chaque évènement temporellement. """ if not events: return "" # Groupement par session source en préservant l'ordre d'apparition. grouped: dict[str, list[JournalEntrySummary]] = {} for e in events: key = e.source_session_name or "(session inconnue)" grouped.setdefault(key, []).append(e) lines = ["\nRécapitulatif des sessions précédentes (évènements marquants uniquement) :"] for session_name, items in grouped.items(): lines.append(f" • {session_name} :") for e in items: ts = f" [{e.occurred_at}]" if e.occurred_at else "" content = e.content.replace("\n", "\n ") lines.append(f" - {content}{ts}") lines.append("") # ligne vide avant le bloc journal courant return "\n".join(lines) + "\n" @staticmethod def _format_narrative_entity(ne: NarrativeEntityContext) -> str: """Bloc équivalent à _format_page mais pour Arc/Chapter/Scene.""" type_label = { "arc": "ARC", "chapter": "CHAPITRE", "scene": "SCÈNE", "character": "FICHE DE PERSONNAGE (PJ)", "npc": "FICHE DE PNJ", }.get(ne.entity_type.lower(), ne.entity_type.upper()) if ne.fields: fields_block = "\n".join( f'- "{key}" : {value or "(vide)"}' for key, value in ne.fields.items() ) else: fields_block = "(aucun champ renseigné)" return ( f"--- {type_label} EN COURS D'ÉDITION ---\n" f"Titre : {ne.title}\n" f"Champs et valeurs actuelles :\n{fields_block}\n\n" "IMPORTANT : concentre-toi EXCLUSIVEMENT sur cette entité narrative. " "Tes suggestions doivent enrichir UNIQUEMENT les champs listés ci-dessus. " "Ne déborde pas vers d'autres arcs, chapitres ou scènes de la campagne, " "même si ça te semblerait pertinent." )