Introduction
Configurer des centaines d'agents IA pour une simulation de médias sociaux semble décourageant. Chaque agent a besoin de plannings d'activité, de fréquences de publication, de délais de réponse, de poids d'influence et de positions. Faire cela manuellement prendrait des heures.
MiroFish automatise cela avec la génération de configuration basée sur les LLM. Le système analyse vos documents, votre graphe de connaissances et les exigences de la simulation, puis génère des configurations détaillées pour chaque agent.
Le défi : les LLM peuvent échouer. Les sorties sont tronquées. Le JSON se brise. Les limites de jetons mordent.
Ce guide couvre l'implémentation complète :
- Génération pas à pas (temps → événements → agents → plateformes)
- Traitement par lots pour éviter les limites de contexte
- Stratégies de réparation JSON pour les sorties tronquées
- Configurations de secours basées sur des règles en cas d'échec du LLM
- Modèles d'activité des agents par type (Étudiant vs Officiel vs Média)
- Logique de validation et de correction
💡
Le pipeline de génération de configuration traite plus de 100 agents à travers une série d'appels API. Apidog a été utilisé pour valider les schémas de requête/réponse à chaque étape, détecter les erreurs de format JSON avant qu'elles n'atteignent la production, et générer des cas de test pour des scénarios limites comme les sorties LLM tronquées.
Tout le code provient d'une utilisation en production dans MiroFish.
Vue d'ensemble de l'architecture
Le générateur de configuration utilise une approche par pipeline :
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Context │ ──► │ Time Config │ ──► │ Event Config │
│ Builder │ │ Generator │ │ Generator │
│ │ │ │ │ │
│ - Simulation │ │ - Total hours │ │ - Initial posts │
│ requirement │ │ - Minutes/round │ │ - Hot topics │
│ - Entity summary│ │ - Peak hours │ │ - Narrative │
│ - Document text │ │ - Activity mult │ │ direction │
└─────────────────┘ └─────────────────┘ └─────────────────┘
│
▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Final Config │ ◄── │ Platform │ ◄── │ Agent Config │
│ Assembly │ │ Config │ │ Batches │
│ │ │ │ │ │
│ - Merge all │ │ - Twitter params│ │ - 15 agents │
│ - Validate │ │ - Reddit params │ │ per batch │
│ - Save JSON │ │ - Viral threshold│ │ - N batches │
└─────────────────┘ └─────────────────┘ └─────────────────┘
Structure des fichiers
backend/app/services/
├── simulation_config_generator.py # Logique principale de génération de configuration
├── ontology_generator.py # Génération d'ontologie (partagée)
└── zep_entity_reader.py # Filtrage d'entités
backend/app/models/
├── task.py # Suivi des tâches
└── project.py # État du projet
Stratégie de génération pas à pas
Générer toutes les configurations en une seule fois dépasserait les limites de jetons. Au lieu de cela, le système génère par étapes :
class SimulationConfigGenerator:
# Chaque lot génère des configurations pour 15 agents
AGENTS_PER_BATCH = 15
# Limites de contexte
MAX_CONTEXT_LENGTH = 50000
TIME_CONFIG_CONTEXT_LENGTH = 10000
EVENT_CONFIG_CONTEXT_LENGTH = 8000
ENTITY_SUMMARY_LENGTH = 300
AGENT_SUMMARY_LENGTH = 300
ENTITIES_PER_TYPE_DISPLAY = 20
def generate_config(
self,
simulation_id: str,
project_id: str,
graph_id: str,
simulation_requirement: str,
document_text: str,
entities: List[EntityNode],
enable_twitter: bool = True,
enable_reddit: bool = True,
progress_callback: Optional[Callable[[int, int, str], None]] = None,
) -> SimulationParameters:
# Calculer le nombre total d'étapes
num_batches = math.ceil(len(entities) / self.AGENTS_PER_BATCH)
total_steps = 3 + num_batches # Temps + Événements + N Lots d'Agents + Plateforme
current_step = 0
def report_progress(step: int, message: str):
nonlocal current_step
current_step = step
if progress_callback:
progress_callback(step, total_steps, message)
logger.info(f"[{step}/{total_steps}] {message}")
# Construire le contexte
context = self._build_context(
simulation_requirement=simulation_requirement,
document_text=document_text,
entities=entities
)
reasoning_parts = []
# Étape 1 : Générer la configuration temporelle
report_progress(1, "Génération de la configuration temporelle...")
time_config_result = self._generate_time_config(context, len(entities))
time_config = self._parse_time_config(time_config_result, len(entities))
reasoning_parts.append(f"Time config: {time_config_result.get('reasoning', 'Succès')}")
# Étape 2 : Générer la configuration des événements
report_progress(2, "Génération de la configuration des événements et des sujets brûlants...")
event_config_result = self._generate_event_config(context, simulation_requirement, entities)
event_config = self._parse_event_config(event_config_result)
reasoning_parts.append(f"Event config: {event_config_result.get('reasoning', 'Succès')}")
# Étapes 3-N : Générer les configurations d'agents par lots
all_agent_configs = []
for batch_idx in range(num_batches):
start_idx = batch_idx * self.AGENTS_PER_BATCH
end_idx = min(start_idx + self.AGENTS_PER_BATCH, len(entities))
batch_entities = entities[start_idx:end_idx]
report_progress(
3 + batch_idx,
f"Génération de la configuration des agents ({start_idx + 1}-{end_idx}/{len(entities)})..."
)
batch_configs = self._generate_agent_configs_batch(
context=context,
entities=batch_entities,
start_idx=start_idx,
simulation_requirement=simulation_requirement
)
all_agent_configs.extend(batch_configs)
reasoning_parts.append(f"Agent config: {len(all_agent_configs)} agents générés")
# Assigner les éditeurs de messages initiaux
event_config = self._assign_initial_post_agents(event_config, all_agent_configs)
# Dernière étape : Configuration de la plateforme
report_progress(total_steps, "Génération de la configuration de la plateforme...")
twitter_config = PlatformConfig(platform="twitter", ...) if enable_twitter else None
reddit_config = PlatformConfig(platform="reddit", ...) if enable_reddit else None
# Assembler la configuration finale
params = SimulationParameters(
simulation_id=simulation_id,
project_id=project_id,
graph_id=graph_id,
simulation_requirement=simulation_requirement,
time_config=time_config,
agent_configs=all_agent_configs,
event_config=event_config,
twitter_config=twitter_config,
reddit_config=reddit_config,
generation_reasoning=" | ".join(reasoning_parts)
)
return params
Cette approche par étapes :
- Maintient chaque appel LLM ciblé et gérable
- Fournit des mises à jour de progression à l'utilisateur
- Permet une récupération partielle si une étape échoue
Construction du contexte
Le constructeur de contexte rassemble les informations pertinentes tout en respectant les limites de jetons :
def _build_context(
self,
simulation_requirement: str,
document_text: str,
entities: List[EntityNode]
) -> str:
# Résumé de l'entité
entity_summary = self._summarize_entities(entities)
context_parts = [
f"## Exigence de simulation\n{simulation_requirement}",
f"\n## Informations sur l'entité ({len(entities)} entités)\n{entity_summary}",
]
# Ajouter le texte du document si l'espace le permet
current_length = sum(len(p) for p in context_parts)
remaining_length = self.MAX_CONTEXT_LENGTH - current_length - 500 # Tampon de 500 caractères
if remaining_length > 0 and document_text:
doc_text = document_text[:remaining_length]
if len(document_text) > remaining_length:
doc_text += "\n...(document tronqué)"
context_parts.append(f"\n## Document original\n{doc_text}")
return "\n".join(context_parts)
Synthèse des entités
Les entités sont résumées par type :
def _summarize_entities(self, entities: List[EntityNode]) -> str:
lines = []
# Grouper par type
by_type: Dict[str, List[EntityNode]] = {}
for e in entities:
t = e.get_entity_type() or "Inconnu"
if t not in by_type:
by_type[t] = []
by_type[t].append(e)
for entity_type, type_entities in by_type.items():
lines.append(f"\n### {entity_type} ({len(type_entities)} entités)")
# Afficher un nombre limité avec une longueur de résumé limitée
display_count = self.ENTITIES_PER_TYPE_DISPLAY
summary_len = self.ENTITY_SUMMARY_LENGTH
for e in type_entities[:display_count]:
summary_preview = (e.summary[:summary_len] + "...") if len(e.summary) > summary_len else e.summary
lines.append(f"- {e.name}: {summary_preview}")
if len(type_entities) > display_count:
lines.append(f" ... et {len(type_entities) - display_count} autres")
return "\n".join(lines)
Ceci produit une sortie comme :
### Étudiant (45 entités)
- Zhang Wei : Actif dans le syndicat étudiant, publie fréquemment sur les événements du campus et la pression académique...
- Li Ming : Étudiant diplômé recherchant l'éthique de l'IA, partage souvent des nouvelles technologiques...
... et 43 autres
### Université (3 entités)
- Université de Wuhan : Compte officiel, publie des annonces et des nouvelles...
Génération de la configuration temporelle
La configuration temporelle détermine la durée de la simulation et les modèles d'activité :
def _generate_time_config(self, context: str, num_entities: int) -> Dict[str, Any]:
# Tronquer le contexte pour cette étape spécifique
context_truncated = context[:self.TIME_CONFIG_CONTEXT_LENGTH]
# Calculer la valeur maximale autorisée (90% du nombre d'agents)
max_agents_allowed = max(1, int(num_entities * 0.9))
prompt = f"""Basé sur les exigences de simulation suivantes, générez la configuration temporelle.
{context_truncated}
## Tâche
Générer la configuration temporelle au format JSON.
### Principes de base (à ajuster en fonction du type d'événement et des groupes de participants) :
- La base d'utilisateurs est chinoise, doit suivre les habitudes du fuseau horaire de Pékin
- 0-5 AM : Presque aucune activité (coefficient 0.05)
- 6-8 AM : Réveil progressif (coefficient 0.4)
- 9-18 PM : Heures de travail, activité modérée (coefficient 0.7)
- 19-22 PM : Pic du soir, plus actif (coefficient 1.5)
- 23 PM : Activité en baisse (coefficient 0.5)
### Retourner le format JSON (pas de markdown) :
Exemple:
{{
"total_simulation_hours": 72,
"minutes_per_round": 60,
"agents_per_hour_min": 5,
"agents_per_hour_max": 50,
"peak_hours": [19, 20, 21, 22],
"off_peak_hours": [0, 1, 2, 3, 4, 5],
"morning_hours": [6, 7, 8],
"work_hours": [9, 10, 11, 12, 13, 14, 15, 16, 17, 18],
"reasoning": "Explication de la configuration temporelle"
}}
Descriptions des champs :
- total_simulation_hours (int) : 24-168 heures, plus court pour les nouvelles urgentes, plus long pour les sujets en cours
- minutes_per_round (int) : 30-120 minutes, recommandé 60
- agents_per_hour_min (int) : Plage 1-{max_agents_allowed}
- agents_per_hour_max (int) : Plage 1-{max_agents_allowed}
- peak_hours (tableau d'entiers) : Ajuster en fonction des groupes de participants
- off_peak_hours (tableau d'entiers) : Généralement tard le soir/tôt le matin
- morning_hours (tableau d'entiers) : Heures du matin
- work_hours (tableau d'entiers) : Heures de travail
- reasoning (chaîne de caractères) : Brève explication"""
system_prompt = "Vous êtes un expert en simulation de médias sociaux. Retournez un format JSON pur."
try:
return self._call_llm_with_retry(prompt, system_prompt)
except Exception as e:
logger.warning(f"La génération LLM de la configuration temporelle a échoué : {e}, utilisation par défaut")
return self._get_default_time_config(num_entities)
Analyse et validation de la configuration temporelle
def _parse_time_config(self, result: Dict[str, Any], num_entities: int) -> TimeSimulationConfig:
# Obtenir les valeurs brutes
agents_per_hour_min = result.get("agents_per_hour_min", max(1, num_entities // 15))
agents_per_hour_max = result.get("agents_per_hour_max", max(5, num_entities // 5))
# Valider et corriger : s'assurer de ne pas dépasser le nombre total d'agents
if agents_per_hour_min > num_entities:
logger.warning(f"agents_per_hour_min ({agents_per_hour_min}) dépasse le total des agents ({num_entities}), corrigé")
agents_per_hour_min = max(1, num_entities // 10)
if agents_per_hour_max > num_entities:
logger.warning(f"agents_per_hour_max ({agents_per_hour_max}) dépasse le total des agents ({num_entities}), corrigé")
agents_per_hour_max = max(agents_per_hour_min + 1, num_entities // 2)
# Assurer min < max
if agents_per_hour_min >= agents_per_hour_max:
agents_per_hour_min = max(1, agents_per_hour_max // 2)
logger.warning(f"agents_per_hour_min >= max, corrigé à {agents_per_hour_min}")
return TimeSimulationConfig(
total_simulation_hours=result.get("total_simulation_hours", 72),
minutes_per_round=result.get("minutes_per_round", 60),
agents_per_hour_min=agents_per_hour_min,
agents_per_hour_max=agents_per_hour_max,
peak_hours=result.get("peak_hours", [19, 20, 21, 22]),
off_peak_hours=result.get("off_peak_hours", [0, 1, 2, 3, 4, 5]),
off_peak_activity_multiplier=0.05,
morning_activity_multiplier=0.4,
work_activity_multiplier=0.7,
peak_activity_multiplier=1.5
)
Configuration temporelle par défaut (fuseau horaire chinois)
def _get_default_time_config(self, num_entities: int) -> Dict[str, Any]:
return {
"total_simulation_hours": 72,
"minutes_per_round": 60, # 1 heure par tour
"agents_per_hour_min": max(1, num_entities // 15),
"agents_per_hour_max": max(5, num_entities // 5),
"peak_hours": [19, 20, 21, 22],
"off_peak_hours": [0, 1, 2, 3, 4, 5],
"morning_hours": [6, 7, 8],
"work_hours": [9, 10, 11, 12, 13, 14, 15, 16, 17, 18],
"reasoning": "Utilisation de la configuration par défaut du fuseau horaire chinois"
}
Génération de la configuration des événements
La configuration des événements définit les messages initiaux, les sujets brûlants et la direction narrative :
def _generate_event_config(
self,
context: str,
simulation_requirement: str,
entities: List[EntityNode]
) -> Dict[str, Any]:
# Obtenir les types d'entités disponibles pour référence LLM
entity_types_available = list(set(
e.get_entity_type() or "Inconnu" for e in entities
))
# Afficher des exemples par type
type_examples = {}
for e in entities:
etype = e.get_entity_type() or "Inconnu"
if etype not in type_examples:
type_examples[etype] = []
if len(type_examples[etype]) < 3:
type_examples[etype].append(e.name)
type_info = "\n".join([
f"- {t}: {', '.join(examples)}"
for t, examples in type_examples.items()
])
context_truncated = context[:self.EVENT_CONFIG_CONTEXT_LENGTH]
prompt = f"""Basé sur les exigences de simulation suivantes, générez la configuration des événements.
Exigence de simulation : {simulation_requirement}
{context_truncated}
## Types d'entités disponibles et exemples
{type_info}
## Tâche
Générer la configuration des événements au format JSON :
- Extraire les mots-clés des sujets brûlants
- Décrire la direction narrative
- Concevoir les messages initiaux, **chaque message doit spécifier poster_type**
**Important** : poster_type doit être sélectionné parmi les "Types d'entités disponibles" ci-dessus, afin que les messages initiaux puissent être assignés aux agents appropriés.
Par exemple : Les déclarations officielles doivent être publiées par les types Officiel/Université, les nouvelles par MediaOutlet, les opinions d'étudiants par Étudiant.
Retourner le format JSON (pas de markdown) :
{{
"hot_topics": ["motclé1", "motclé2", ...],
"narrative_direction": "<description de la direction narrative>",
"initial_posts": [
{{"content": "Contenu du message", "poster_type": "Type d'entité (doit correspondre aux types disponibles)"}},
...
],
"reasoning": "<brève explication>"
}}"""
system_prompt = "Vous êtes un expert en analyse d'opinions. Retournez un format JSON pur."
try:
return self._call_llm_with_retry(prompt, system_prompt)
except Exception as e:
logger.warning(f"La génération LLM de la configuration des événements a échoué : {e}, utilisation par défaut")
return {
"hot_topics": [],
"narrative_direction": "",
"initial_posts": [],
"reasoning": "Utilisation de la configuration par défaut"
}
Assignation des éditeurs de messages initiaux
Après avoir généré les messages initiaux, associez-les aux agents réels :
def _assign_initial_post_agents(
self,
event_config: EventConfig,
agent_configs: List[AgentActivityConfig]
) -> EventConfig:
if not event_config.initial_posts:
return event_config
# Indexer les agents par type
agents_by_type: Dict[str, List[AgentActivityConfig]] = {}
for agent in agent_configs:
etype = agent.entity_type.lower()
if etype not in agents_by_type:
agents_by_type[etype] = []
agents_by_type[etype].append(agent)
# Mappage des alias de type (gère les variations LLM)
type_aliases = {
"official": ["official", "university", "governmentagency", "government"],
"university": ["university", "official"],
"mediaoutlet": ["mediaoutlet", "media"],
"student": ["student", "person"],
"professor": ["professor", "expert", "teacher"],
"alumni": ["alumni", "person"],
"organization": ["organization", "ngo", "company", "group"],
"person": ["person", "student", "alumni"],
}
# Suivre les indices utilisés pour éviter de réutiliser le même agent
used_indices: Dict[str, int] = {}
updated_posts = []
for post in event_config.initial_posts:
poster_type = post.get("poster_type", "").lower()
content = post.get("content", "")
matched_agent_id = None
# 1. Correspondance directe
if poster_type in agents_by_type:
agents = agents_by_type[poster_type]
idx = used_indices.get(poster_type, 0) % len(agents)
matched_agent_id = agents[idx].agent_id
used_indices[poster_type] = idx + 1
else:
# 2. Correspondance d'alias
for alias_key, aliases in type_aliases.items():
if poster_type in aliases or alias_key == poster_type:
for alias in aliases:
if alias in agents_by_type:
agents = agents_by_type[alias]
idx = used_indices.get(alias, 0) % len(agents)
matched_agent_id = agents[idx].agent_id
used_indices[alias] = idx + 1
break
if matched_agent_id is not None:
break
# 3. Repli : utiliser l'agent ayant la plus grande influence
if matched_agent_id is None:
logger.warning(f"Aucun agent correspondant pour le type '{poster_type}', utilisation de l'agent ayant la plus grande influence")
if agent_configs:
sorted_agents = sorted(agent_configs, key=lambda a: a.influence_weight, reverse=True)
matched_agent_id = sorted_agents[0].agent_id
else:
matched_agent_id = 0
updated_posts.append({
"content": content,
"poster_type": post.get("poster_type", "Inconnu"),
"poster_agent_id": matched_agent_id
})
logger.info(f"Assignation du message initial : poster_type='{poster_type}' -> agent_id={matched_agent_id}")
event_config.initial_posts = updated_posts
return event_config
Génération de la configuration des agents par lots
Générer des configurations pour des centaines d'agents en une seule fois dépasserait les limites de jetons. Le système traite par lots de 15 :
def _generate_agent_configs_batch(
self,
context: str,
entities: List[EntityNode],
start_idx: int,
simulation_requirement: str
) -> List[AgentActivityConfig]:
# Construire les informations d'entité avec une longueur de résumé limitée
entity_list = []
summary_len = self.AGENT_SUMMARY_LENGTH
for i, e in enumerate(entities):
entity_list.append({
"agent_id": start_idx + i,
"entity_name": e.name,
"entity_type": e.get_entity_type() or "Inconnu",
"summary": e.summary[:summary_len] if e.summary else ""
})
prompt = f"""Basé sur les informations suivantes, générez la configuration d'activité des médias sociaux pour chaque entité.
Exigence de simulation : {simulation_requirement}
## Liste des entités
```json
{json.dumps(entity_list, ensure_ascii=False, indent=2)}
Tâche
Générez la configuration d'activité pour chaque entité. Remarque :
- **Le temps doit suivre les habitudes chinoises** : 0-5 AM presque aucune activité, 19-22 PM plus actif
- **Institutions officielles** (Université/AgenceGouvernementale) : Faible activité (0.1-0.3), heures de travail (9-17), réponse lente (60-240 min), forte influence (2.5-3.0)
- **Médias** (MediaOutlet) : Activité modérée (0.4-0.6), activité toute la journée (8-23), réponse rapide (5-30 min), forte influence (2.0-2.5)
- **Individus** (Étudiant/Personne/Ancien) : Forte activité (0.6-0.9), principalement le soir (18-23), réponse rapide (1-15 min), faible influence (0.8-1.2)
- **Personnalités publiques/Experts** : Activité modérée (0.4-0.6), influence moyenne-élevée (1.5-2.0)
system_prompt = "Vous êtes un expert en analyse du comportement des médias sociaux. Retournez un format JSON pur."
try:
result = self._call_llm_with_retry(prompt, system_prompt)
llm_configs = {cfg["agent_id"]: cfg for cfg in result.get("agent_configs", [])}
except Exception as e:
logger.warning(f"La génération LLM par lot de la configuration de l'agent a échoué : {e}, utilisation d'une génération basée sur des règles")
llm_configs = {}
# Construire les objets AgentActivityConfig
configs = []
for i, entity in enumerate(entities):
agent_id = start_idx + i
cfg = llm_configs.get(agent_id, {})
# Utiliser un repli basé sur des règles si le LLM a échoué
if not cfg:
cfg = self._generate_agent_config_by_rule(entity)
config = AgentActivityConfig(
agent_id=agent_id,
entity_uuid=entity.uuid,
entity_name=entity.name,
entity_type=entity.get_entity_type() or "Inconnu",
activity_level=cfg.get("activity_level", 0.5),
posts_per_hour=cfg.get("posts_per_hour", 0.5),
comments_per_hour=cfg.get("comments_per_hour", 1.0),
active_hours=cfg.get("active_hours", list(range(9, 23))),
response_delay_min=cfg.get("response_delay_min", 5),
response_delay_max=cfg.get("response_delay_max", 60),
sentiment_bias=cfg.get("sentiment_bias", 0.0),
stance=cfg.get("stance", "neutral"),
influence_weight=cfg.get("influence_weight", 1.0)
)
configs.append(config)
return configs
### Configurations de secours basées sur des règles
En cas d'échec du LLM, utilisez des modèles prédéfinis :
```python
def _generate_agent_config_by_rule(self, entity: EntityNode) -> Dict[str, Any]:
entity_type = (entity.get_entity_type() or "Inconnu").lower()
if entity_type in ["university", "governmentagency", "ngo"]:
# Institution officielle : heures de travail, faible fréquence, forte influence
return {
"activity_level": 0.2,
"posts_per_hour": 0.1,
"comments_per_hour": 0.05,
"active_hours": list(range(9, 18)), # 9:00-17:59
"response_delay_min": 60,
"response_delay_max": 240,
"sentiment_bias": 0.0,
"stance": "neutral",
"influence_weight": 3.0
}
elif entity_type in ["mediaoutlet"]:
# Média : activité toute la journée, fréquence modérée, forte influence
return {
"activity_level": 0.5,
"posts_per_hour": 0.8,
"comments_per_hour": 0.3,
"active_hours": list(range(7, 24)), # 7:00-23:59
"response_delay_min": 5,
"response_delay_max": 30,
"sentiment_bias": 0.0,
"stance": "observer",
"influence_weight": 2.5
}
elif entity_type in ["professor", "expert", "official"]:
# Expert/Professeur : travail + soirée, fréquence modérée
return {
"activity_level": 0.4,
"posts_per_hour": 0.3,
"comments_per_hour": 0.5,
"active_hours": list(range(8, 22)), # 8:00-21:59
"response_delay_min": 15,
"response_delay_max": 90,
"sentiment_bias": 0.0,
"stance": "neutral",
"influence_weight": 2.0
}
elif entity_type in ["student"]:
# Étudiant : pic du soir, haute fréquence
return {
"activity_level": 0.8,
"posts_per_hour": 0.6,
"comments_per_hour": 1.5,
"active_hours": [8, 9, 10, 11, 12, 13, 18, 19, 20, 21, 22, 23],
"response_delay_min": 1,
"response_delay_max": 15,
"sentiment_bias": 0.0,
"stance": "neutral",
"influence_weight": 0.8
}
elif entity_type in ["alumni"]:
# Ancien : axé sur le soir
return {
"activity_level": 0.6,
"posts_per_hour": 0.4,
"comments_per_hour": 0.8,
"active_hours": [12, 13, 19, 20, 21, 22, 23], # Déjeuner + soir
"response_delay_min": 5,
"response_delay_max": 30,
"sentiment_bias": 0.0,
"stance": "neutral",
"influence_weight": 1.0
}
else:
# Personne par défaut : pic du soir
return {
"activity_level": 0.7,
"posts_per_hour": 0.5,
"comments_per_hour": 1.2,
"active_hours": [9, 10, 11, 12, 13, 18, 19, 20, 21, 22, 23],
"response_delay_min": 2,
"response_delay_max": 20,
"sentiment_bias": 0.0,
"stance": "neutral",
"influence_weight": 1.0
}
Appel LLM avec réessai et réparation JSON
Les appels LLM échouent. Les sorties sont tronquées. Le JSON se brise. Le système gère tout cela :
def _call_llm_with_retry(self, prompt: str, system_prompt: str) -> Dict[str, Any]:
import re
max_attempts = 3
last_error = None
for attempt in range(max_attempts):
try:
response = self.client.chat.completions.create(
model=self.model_name,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt}
],
response_format={"type": "json_object"},
temperature=0.7 - (attempt * 0.1) # Diminuer la température lors du réessai
)
content = response.choices[0].message.content
finish_reason = response.choices[0].finish_reason
# Vérifier si tronqué
if finish_reason == 'length':
logger.warning(f"Sortie LLM tronquée (tentative {attempt+1})")
content = self._fix_truncated_json(content)
# Essayer d'analyser le JSON
try:
return json.loads(content)
except json.JSONDecodeError as e:
logger.warning(f"Échec de l'analyse JSON (tentative {attempt+1}) : {str(e)[:80]}")
# Essayer de réparer le JSON
fixed = self._try_fix_config_json(content)
if fixed:
return fixed
last_error = e
except Exception as e:
logger.warning(f"L'appel LLM a échoué (tentative {attempt+1}) : {str(e)[:80]}")
last_error = e
import time
time.sleep(2 * (attempt + 1))
raise last_error or Exception("L'appel LLM a échoué")
Correction du JSON tronqué
def _fix_truncated_json(self, content: str) -> str:
content = content.strip()
# Compter les accolades non fermées
open_braces = content.count('{') - content.count('}')
open_brackets = content.count('[') - content.count(']')
# Vérifier les chaînes de caractères non fermées
if content and content[-1] not in '",}]':
content += '"'
# Fermer les crochets
content += ']' * open_brackets
content += '}' * open_braces
return content
Réparation JSON avancée
def _try_fix_config_json(self, content: str) -> Optional[Dict[str, Any]]:
import re
# Correction de la troncature
content = self._fix_truncated_json(content)
# Extraire la portion JSON
json_match = re.search(r'\{[\s\S]*\}', content)
if json_match:
json_str = json_match.group()
# Supprimer les retours à la ligne dans les chaînes
def fix_string(match):
s = match.group(0)
s = s.replace('\n', ' ').replace('\r', ' ')
s = re.sub(r'\s+', ' ', s)
return s
json_str = re.sub(r'"[^"\\]*(?:\\.[^"\\]*)*"', fix_string, json_str)
try:
return json.loads(json_str)
except:
# Essayer de supprimer les caractères de contrôle
json_str = re.sub(r'[\x00-\x1f\x7f-\x9f]', ' ', json_str)
json_str = re.sub(r'\s+', ' ', json_str)
try:
return json.loads(json_str)
except:
pass
return None
Structures de données de configuration
Configuration de l'activité de l'agent
@dataclass
class AgentActivityConfig:
"""Configuration de l'activité d'un seul agent"""
agent_id: int
entity_uuid: str
entity_name: str
entity_type: str
# Niveau d'activité (0.0-1.0)
activity_level: float = 0.5
# Fréquence de publication (par heure)
posts_per_hour: float = 1.0
comments_per_hour: float = 2.0
# Heures actives (format 24 heures, 0-23)
active_hours: List[int] = field(default_factory=lambda: list(range(8, 23)))
# Vitesse de réponse (délai de réaction en minutes simulées)
response_delay_min: int = 5
response_delay_max: int = 60
# Tendance de sentiment (-1.0 à 1.0, négatif à positif)
sentiment_bias: float = 0.0
# Position sur des sujets spécifiques
stance: str = "neutral" # favorable, opposé, neutre, observateur
# Poids d'influence (affecte la probabilité d'être vu)
influence_weight: float = 1.0
Configuration de la simulation temporelle
@dataclass
class TimeSimulationConfig:
"""Configuration de la simulation temporelle (fuseau horaire chinois)"""
total_simulation_hours: int = 72 # Par défaut 72 heures (3 jours)
minutes_per_round: int = 60 # 60 minutes par tour
# Agents activés par heure
agents_per_hour_min: int = 5
agents_per_hour_max: int = 20
# Heures de pointe (soir 19-22, les plus actives en Chine)
peak_hours: List[int] = field(default_factory=lambda: [19, 20, 21, 22])
peak_activity_multiplier: float = 1.5
# Heures creuses (tôt le matin 0-5, presque aucune activité)
off_peak_hours: List[int] = field(default_factory=lambda: [0, 1, 2, 3, 4, 5])
off_peak_activity_multiplier: float = 0.05
# Heures du matin
morning_hours: List[int] = field(default_factory=lambda: [6, 7, 8])
morning_activity_multiplier: float = 0.4
# Heures de travail
work_hours: List[int] = field(default_factory=lambda: [9, 10, 11, 12, 13, 14, 15, 16, 17, 18])
work_activity_multiplier: float = 0.7
Paramètres de simulation complets
@dataclass
class SimulationParameters:
"""Configuration complète des paramètres de simulation"""
simulation_id: str
project_id: str
graph_id: str
simulation_requirement: str
time_config: TimeSimulationConfig = field(default_factory=TimeSimulationConfig)
agent_configs: List[AgentActivityConfig] = field(default_factory=list)
event_config: EventConfig = field(default_factory=EventConfig)
twitter_config: Optional[PlatformConfig] = None
reddit_config: Optional[PlatformConfig] = None
llm_model: str = ""
llm_base_url: str = ""
generated_at: str = field(default_factory=lambda: datetime.now().isoformat())
generation_reasoning: str = ""
def to_dict(self) -> Dict[str, Any]:
time_dict = asdict(self.time_config)
return {
"simulation_id": self.simulation_id,
"project_id": self.project_id,
"graph_id": self.graph_id,
"simulation_requirement": self.simulation_requirement,
"time_config": time_dict,
"agent_configs": [asdict(a) for a in self.agent_configs],
"event_config": asdict(self.event_config),
"twitter_config": asdict(self.twitter_config) if self.twitter_config else None,
"reddit_config": asdict(self.reddit_config) if self.reddit_config else None,
"llm_model": self.llm_model,
"llm_base_url": self.llm_base_url,
"generated_at": self.generated_at,
"generation_reasoning": self.generation_reasoning,
}
Tableau récapitulatif : Modèles de types d'agents
| Type d'agent | Activité | Heures actives | Publications/heure | Commentaires/heure | Réponse (min) | Influence |
|---|---|---|---|---|---|---|
| Université | 0.2 | 9-17 | 0.1 | 0.05 | 60-240 | 3.0 |
| AgenceGouvernementale | 0.2 | 9-17 | 0.1 | 0.05 | 60-240 | 3.0 |
| MediaOutlet | 0.5 | 7-23 | 0.8 | 0.3 | 5-30 | 2.5 |
| Professeur | 0.4 | 8-21 | 0.3 | 0.5 | 15-90 | 2.0 |
| Étudiant | 0.8 | 8-12, 18-23 | 0.6 | 1.5 | 1-15 | 0.8 |
| Ancien | 0.6 | 12-13, 19-23 | 0.4 | 0.8 | 5-30 | 1.0 |
| Personne (par défaut) | 0.7 | 9-13, 18-23 | 0.5 | 1.2 | 2-20 | 1.0 |
Conclusion
La génération de configuration basée sur les LLM nécessite une gestion minutieuse de :
- La génération pas à pas : Décomposer en étapes gérables (temps → événements → agents → plateformes)
- Le traitement par lots : Traiter 15 agents par lot pour éviter les limites de contexte
- La réparation JSON : Gérer la troncature avec la correspondance des crochets et l'échappement des chaînes
- Les replis basés sur des règles : Fournir des valeurs par défaut sensées en cas d'échec du LLM
- Les modèles spécifiques au type : Différents types d'agents ont des modèles d'activité différents
- La validation et la correction : Vérifier les valeurs générées et corriger les problèmes (par exemple, agents_per_hour > total_agents)