En tant qu'ingénieur qui a passé trois ans à intégrer des modèles de langage dans des moteurs de jeux AAA, je peux vous dire que faire parler un PNJ avec une vraie intelligence n'est pas un simple appel à une API. C'est une architecture complète de systèmes qui doivent gérer la cohérence narrative, la latence réseau, le budget de tokens, et la personnalité des personnages. Aujourd'hui, je vais partager mon expérience terrain et vous montrer comment construire un système de PNJ conversationnels production-ready.
Architecture Système : Le Cerveau des PNJ Intelligents
Avant d'écrire une seule ligne de code, comprenons l'architecture. Un PNJ intelligent ne se résume pas à un chatbot avec uneskin 3D. C'est un système multicouche où le LLM joue le rôle de moteur de raisonnement, mais où d'autres composants gèrent la mémoire, le contexte de jeu, et la cohérence narrative.
Schéma d'Architecture Global
- Couche 1 - Interface Jeu : Le moteur de jeu (Unity/Unreal) envoie les événements de jeu au système PNJ
- Couche 2 - Gestionnaire de Contexte : Aggrege le contexte narratif, l'inventaire du joueur, l'historique des interactions
- Couche 3 - Moteur LLM : Le modèle génère les réponses en langage naturel
- Couche 4 - Contrôleur de Personnalité : Applique les contraintes de personnage, le ton, les limites narratives
- Couche 5 - Cache et Optimisation : Gestion du cache de réponses, Rate Limiting, debouncing
Implémentation du Client de Communication
La première brique technique est le client qui communique avec l'API LLM. Voici mon implémentation complète en Python, testée en production sur un MMORPG avec 50 000 joueurs simultanés.
#!/usr/bin/env python3
"""
PNJ Intelligence Engine - HolySheep AI Integration
Version: 2.1.0
Auteur: Équipe HolySheep AI
"""
import asyncio
import hashlib
import json
import logging
import time
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from typing import Any, Optional
import aiohttp
from aiohttp import ClientTimeout
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class NPCTone(Enum):
"""Tonalité du PNJ selon le contexte narratif."""
FRIENDLY = "amical"
HOSTILE = "hostile"
MYSTERIOUS = "mystérieux"
WISE = "sage"
COMBAT = "combat"
@dataclass
class NPCMemory:
"""Mémoire contextuelle du PNJ pour maintenir la cohérence."""
short_term: list[dict] = field(default_factory=list)
long_term: dict[str, Any] = field(default_factory=dict)
emotional_state: str = "neutre"
relationship_level: int = 50
def add_interaction(self, player_id: str, content: str, npc_response: str):
"""Ajoute une interaction à la mémoire court terme."""
self.short_term.append({
"timestamp": datetime.now().isoformat(),
"player_id": player_id,
"player_message": content,
"npc_response": npc_response,
"tokens_used": len(content.split()) + len(npc_response.split())
})
# Garder seulement les 20 dernières interactions
if len(self.short_term) > 20:
self.short_term = self.short_term[-20:]
@dataclass
class NPCContext:
"""Contexte complet pour la génération de réponse."""
npc_id: str
npc_name: str
npc_role: str # "marchand", "gardien", "quête", etc.
tone: NPCTone
game_state: dict
player_inventory: list[str] = field(default_factory=list)
quest_progress: dict = field(default_factory=dict)
memory: NPCMemory = field(default_factory=NPCMemory)
def build_system_prompt(self) -> str:
"""Construit le prompt système pour le LLM."""
return f"""Tu es {self.npc_name}, {self.npc_role} dans un monde de fantasy.
PERSONNALITÉ:
- Tonalité actuelle: {self.tone.value}
- État émotionnel: {self.memory.emotional_state}
- Niveau de relation avec le joueur: {self.memory.relationship_level}/100
CONTEXTE DE JEU:
- Quête actuelle: {self.quest_progress.get('current_quest', 'Aucune')}
- Progression: {self.quest_progress.get('progress', 0)}%
- Récompenses disponibles: {self.quest_progress.get('rewards', [])}
INVENTAIRE DU JOUEUR:
{', '.join(self.player_inventory) if self.player_inventory else 'Vide'}
RÈGLES ABSOLUES:
1. Ne jamais révéler les secrets majeurs de l'histoire
2. Adapter les réponses selon le niveau de relation
3. Mentionner les indices de quête de manière naturelle
4. Maximum 150 mots par réponse
5. Répondre en français uniquement
MÉMOIRE RÉCENTE (5 dernières interactions):
{self._format_memory()}""" def _format_memory(self) -> str:
"""Formate la mémoire court terme pour le prompt."""
if not self.memory.short_term:
return "Première rencontre avec ce joueur."
recent = self.memory.short_term[-5:]
formatted = []
for i, interaction in enumerate(recent):
formatted.append(
f"- [{interaction['timestamp'][-8:]}] "
f"Joueur: {interaction['player_message'][:50]}... "
f"→ Toi: {interaction['npc_response'][:50]}..."
)
return "\n".join(formatted)
class HolySheepLLMClient:
"""Client optimisé pour HolySheep AI avec cache intelligent."""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
model: str = "deepseek-v3.2",
max_tokens: int = 150,
temperature: float = 0.7,
cache_ttl: int = 300
):
self.api_key = api_key
self.base_url = base_url
self.model = model
self.max_tokens = max_tokens
self.temperature = temperature
self.cache_ttl = cache_ttl
self._cache: dict[str, tuple[str, float]] = {}
self._request_count = 0
self._tokens_used = 0
self._session: Optional[aiohttp.ClientSession] = None
# Latence measurements
self._latencies: list[float] = []
# Prix HolySheep 2026 (USD par million de tokens)
self._pricing = {
"deepseek-v3.2": 0.42,
"gpt-4.1": 8.00,
"claude-sonnet-4.5": 15.00,
"gemini-2.5-flash": 2.50
}
async def __aenter__(self):
"""Context manager pour la session aiohttp."""
timeout = ClientTimeout(total=10, connect=5)
self._session = aiohttp.ClientSession(timeout=timeout)
return self
async def __aexit__(self, *args):
"""Fermeture propre de la session."""
if self._session:
await self._session.close()
def _generate_cache_key(self, messages: list[dict]) -> str:
"""Génère une clé de cache stable pour les requêtes similaires."""
content = json.dumps(messages, sort_keys=True)
return hashlib.sha256(content.encode()).hexdigest()[:32]
def _get_cached_response(self, cache_key: str) -> Optional[str]:
"""Récupère une réponse en cache si valide."""
if cache_key in self._cache:
response, timestamp = self._cache[cache_key]
if time.time() - timestamp < self.cache_ttl:
logger.info(f"✅ Cache HIT: {cache_key[:8]}...")
return response
else:
del self._cache[cache_key]
return None
def _cache_response(self, cache_key: str, response: str):
"""Stocke une réponse en cache."""
self._cache[cache_key] = (response, time.time())
# Limite la taille du cache à 1000 entrées
if len(self._cache) > 1000:
oldest = min(self._cache.items(), key=lambda x: x[1][1])
del self._cache[oldest[0]]
async def generate_response(
self,
context: NPCContext,
player_message: str
) -> tuple[str, dict]:
"""
Génère une réponse de PNJ avec optimisations.
Returns:
tuple: (réponse_text, métadonnées)
"""
start_time = time.time()
# Construction des messages
messages = [
{"role": "system", "content": context.build_system_prompt()},
{"role": "user", "content": player_message}
]
# Vérification du cache
cache_key = self._generate_cache_key(messages)
cached = self._get_cached_response(cache_key)
if cached:
latency = (time.time() - start_time) * 1000
self._latencies.append(latency)
return cached, {
"source": "cache",
"latency_ms": latency,
"cached": True
}
# Préparation de la requête
payload = {
"model": self.model,
"messages": messages,
"max_tokens": self.max_tokens,
"temperature": self.temperature,
"stream": False
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
try:
async with self._session.post(
f"{self.base_url}/chat/completions",
json=payload,
headers=headers
) as response:
if response.status != 200:
error_text = await response.text()
logger.error(f"❌ API Error {response.status}: {error_text}")
raise Exception(f"API Error: {response.status}")
data = await response.json()
latency = (time.time() - start_time) * 1000
self._latencies.append(latency)
self._request_count += 1
# Extraction de la réponse
assistant_message = data["choices"][0]["message"]["content"]
usage = data.get("usage", {})
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
total_tokens = usage.get("total_tokens", input_tokens + output_tokens)
self._tokens_used += total_tokens
# Mise en cache
self._cache_response(cache_key, assistant_message)
# Mise à jour de la mémoire du PNJ
context.memory.add_interaction(
player_id="current_player",
content=player_message,
npc_response=assistant_message
)
logger.info(
f"📤 Response: {latency:.1f}ms, "
f"{total_tokens} tokens, "
f"Coût: ${self._calculate_cost(total_tokens):.4f}"
)
return assistant_message, {
"source": "api",
"latency_ms": latency,
"tokens": total_tokens,
"cost_usd": self._calculate_cost(total_tokens),
"model": self.model
}
except asyncio.TimeoutError:
logger.error("❌ Timeout lors de l'appel API")
return "Je réfléchis... Peux-tu répéter ta question ?", {
"source": "fallback",
"error": "timeout"
}
except Exception as e:
logger.error(f"❌ Erreur: {e}")
raise
def _calculate_cost(self, tokens: int) -> float:
"""Calcule le coût en USD pour les tokens utilisés."""
price_per_million = self._pricing.get(self.model, 0.42)
return (tokens / 1_000_000) * price_per_million
def get_stats(self) -> dict:
"""Retourne les statistiques d'utilisation."""
avg_latency = sum(self._latencies) / len(self._latencies) if self._latencies else 0
return {
"requests": self._request_count,
"total_tokens": self._tokens_used,
"total_cost_usd": self._calculate_cost(self._tokens_used),
"avg_latency_ms": round(avg_latency, 2),
"cache_size": len(self._cache),
"model": self.model
}
Démonstration d'utilisation
async def demo():
"""Exemple d'utilisation du système de PNJ."""
async with HolySheepLLMClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
model="deepseek-v3.2"
) as client:
# Création du contexte PNJ
context = NPCContext(
npc_id="npc_merchant_001",
npc_name="Maître Aldric",
npc_role="Marchand神秘的古董商",
tone=NPCTone.FRIENDLY,
game_state={
"location": "Cité des Artisans",
"time_of_day": "soir",
"weather": "pluie légère"
},
player_inventory=["Épée de Bronze", "Potion de Soin x3"],
quest_progress={
"current_quest": "Le Mystère de l'Échine d'Or",
"progress": 35,
"rewards": ["500 pièces d'or", "Amulette de Chance"]
},
memory=NPCMemory(
emotional_state="curieux",
relationship_level=65
)
)
# Dialogue de démonstration
player_inputs = [
"Bonjour, j'ai entendu dire que vous avez des objets rares à vendre.",
"Pouvez-vous me parler de l'Échine d'Or ? J'ai besoin d'informations pour ma quête.",
"Combien demanderiez-vous pour cette amulette ?"
]
print("=" * 60)
print("🎮 DÉMONSTRATION SYSTÈME PNJ INTELLIGENT")
print("=" * 60)
for player_msg in player_inputs:
print(f"\n👤 Joueur: {player_msg}")
response, metadata = await client.generate_response(
context, player_msg
)
print(f"🏪 {context.npc_name}: {response}")
print(f" ⚡ Latence: {metadata['latency_ms']:.1f}ms | "
f"Source: {metadata['source']}")
# Statistiques finales
print("\n" + "=" * 60)
print("📊 STATISTIQUES DE SESSION")
stats = client.get_stats()
print(f" Requêtes API: {stats['requests']}")
print(f" Tokens totaux: {stats['total_tokens']:,}")
print(f" Coût total: ${stats['total_cost_usd']:.4f}")
print(f" Latence moyenne: {stats['avg_latency_ms']:.1f}ms")
print("=" * 60)
if __name__ == "__main__":
asyncio.run(demo())Gestion Avancée de la Concurrence et Rate Limiting
En production, vous n'allez pas gérer un seul PNJ mais des centaines simultanément. Sans une gestion correcte de la concurrence, vous allez saturer votre quota API ou pire, créer des réponses incohérentes car deux requêtes modifient le même état simultanément.
Pattern du Semaphore pour limiter la Concurrence
#!/usr/bin/env python3
"""
NPC Concurrent Manager - Gestion avancée de la concurrence
avec rate limiting intelligent et fallback multi-fournisseur
"""
import asyncio
import logging
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Optional
import random
logger = logging.getLogger(__name__)
@dataclass
class RateLimitConfig:
"""Configuration des limites de taux par fournisseur."""
requests_per_minute: int = 60
tokens_per_minute: int = 100_000
burst_size: int = 10
@property
def requests_per_second(self) -> float:
return self.requests_per_minute / 60
@dataclass
class TokenBucket:
"""Implémentation du pattern Token Bucket pour le rate limiting."""
capacity: int
refill_rate: float # tokens par seconde
current: float = field(init=False)
last_refill: float = field(init=False)
def __post_init__(self):
self.current = float(self.capacity)
self.last_refill = asyncio.get_event_loop().time()
async def acquire(self, tokens_needed: int) -> bool:
"""Tente d'acquérir les tokens nécessaires."""
loop = asyncio.get_event_loop()
now = loop.time()
# Refill automatique
elapsed = now - self.last_refill
self.current = min(
self.capacity,
self.current + elapsed * self.refill_rate
)
self.last_refill = now
if self.current >= tokens_needed:
self.current -= tokens_needed
return True
return False
def waiting_time(self, tokens_needed: int) -> float:
"""Calcule le temps d'attente estimé en secondes."""
if self.current >= tokens_needed:
return 0.0
return (tokens_needed - self.current) / self.refill_rate
class ConcurrentNPCManager:
"""
Gestionnaire de PNJ concurrents avec:
- Rate Limiting par fournisseur
- Failover automatique entre modèles
- Queue prioritaire avec backpressure
- Circuit Breaker pour éviter les pannes en cascade
"""
def __init__(self):
# Limites par défaut pour HolySheep AI
self.holysheep_limits = RateLimitConfig(
requests_per_minute=120, # Dépend du plan
tokens_per_minute=200_000,
burst_size=15
)
# Buckets de rate limiting
self.request_bucket = TokenBucket(
capacity=self.holysheep_limits.burst_size,
refill_rate=self.holysheep_limits.requests_per_second
)
self.token_bucket = TokenBucket(
capacity=self.holysheep_limits.tokens_per_minute,
refill_rate=self.holysheep_limits.tokens_per_minute / 60
)
# Queue de requêtes avec priorité
self._request_queue: asyncio.PriorityQueue = asyncio.PriorityQueue(
maxsize=1000
)
# État des services (pour le circuit breaker)
self._service_states: dict[str, str] = defaultdict(
lambda: "closed" # closed, open, half-open
)
self._failure_counts: dict[str, int] = defaultdict(int)
self._circuit_threshold = 5
self._circuit_timeout = 30 # secondes
# Métriques
self._metrics = {
"requests_total": 0,
"requests_success": 0,
"requests_failed": 0,
"requests_queued": 0,
"tokens_used": 0
}
# Worker pour traiter la queue
self._worker_task: Optional[asyncio.Task] = None
self._running = False
async def start(self):
"""Démarre le gestionnaire de queue."""
self._running = True
self._worker_task = asyncio.create_task(self._process_queue())
logger.info("🚀 ConcurrentNPCManager démarré")
async def stop(self):
"""Arrête proprement le gestionnaire."""
self._running = False
if self._worker_task:
self._worker_task.cancel()
try:
await self._worker_task
except asyncio.CancelledError:
pass
logger.info("🛑 ConcurrentNPCManager arrêté")
async def submit_request(
self,
priority: int, # 1 = haute priorité, 5 = basse
llm_client,
context,
player_message: str,
timeout: float = 30.0
) -> tuple[str, dict]:
"""
Soumet une requête de PNJ avec