DEV Community

Cover image for Générer +100 Configurations d'Agents avec LLMs et Traitement par Lots
Antoine Laurent
Antoine Laurent

Posted on • Originally published at apidog.com

Générer +100 Configurations d'Agents avec LLMs et Traitement par Lots

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.

Essayez Apidog dès aujourd'hui

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 détaille 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 s'appuie sur un pipeline structuré :

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│   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     │
└─────────────────┘     └─────────────────┘     └─────────────────┘
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

Stratégie de génération pas à pas

Pour générer des configurations sans dépasser les limites des LLM, adoptez une génération par étapes :

class SimulationConfigGenerator:
    AGENTS_PER_BATCH = 15
    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:

        num_batches = math.ceil(len(entities) / self.AGENTS_PER_BATCH)
        total_steps = 3 + num_batches  # Temps + Événements + Agents (N lots) + 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}")

        context = self._build_context(
            simulation_requirement=simulation_requirement,
            document_text=document_text,
            entities=entities
        )

        reasoning_parts = []

        # Étape 1 : Génération de 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ération de 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ération des 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")

        # Assignation des é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

        # Assemblage final
        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
Enter fullscreen mode Exit fullscreen mode

Avantages de cette méthode :

  1. Chaque appel LLM reste ciblé et gérable en taille/context.
  2. La progression peut être suivie et affichée à l'utilisateur.
  3. La récupération partielle est possible en cas d'échec d'une étape.

Construction du contexte

La fonction _build_context assemble les données en respectant la taille maximale du prompt :

def _build_context(
    self,
    simulation_requirement: str,
    document_text: str,
    entities: List[EntityNode]
) -> str:

    # Résumé des entités
    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}",
    ]

    current_length = sum(len(p) for p in context_parts)
    remaining_length = self.MAX_CONTEXT_LENGTH - current_length - 500  # Tampon

    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)
Enter fullscreen mode Exit fullscreen mode

Synthèse des entités

Les entités sont groupées et résumées par type :

def _summarize_entities(self, entities: List[EntityNode]) -> str:
    lines = []
    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)")
        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)
Enter fullscreen mode Exit fullscreen mode

Exemple de sortie :

### É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...
Enter fullscreen mode Exit fullscreen mode

Génération de la configuration temporelle

Déterminez 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]:
    context_truncated = context[:self.TIME_CONFIG_CONTEXT_LENGTH]
    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) :
...
"""

    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)
Enter fullscreen mode Exit fullscreen mode

Validation de la configuration temporelle

Corrigez les valeurs incohérentes :

def _parse_time_config(self, result: Dict[str, Any], num_entities: int) -> TimeSimulationConfig:
    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))

    # Correction des incohérences
    if agents_per_hour_min > num_entities:
        agents_per_hour_min = max(1, num_entities // 10)
    if agents_per_hour_max > num_entities:
        agents_per_hour_max = max(agents_per_hour_min + 1, num_entities // 2)
    if agents_per_hour_min >= agents_per_hour_max:
        agents_per_hour_min = max(1, agents_per_hour_max // 2)

    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
    )
Enter fullscreen mode Exit fullscreen mode

Configuration temporelle par défaut

def _get_default_time_config(self, num_entities: int) -> Dict[str, Any]:
    return {
        "total_simulation_hours": 72,
        "minutes_per_round": 60,
        "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"
    }
Enter fullscreen mode Exit fullscreen mode

Génération de la configuration des événements

Générez les sujets brûlants, la narration et les messages initiaux :

def _generate_event_config(
    self,
    context: str,
    simulation_requirement: str,
    entities: List[EntityNode]
) -> Dict[str, Any]:

    entity_types_available = list(set(
        e.get_entity_type() or "Inconnu" for e in entities
    ))

    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.

...
"""

    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"
        }
Enter fullscreen mode Exit fullscreen mode

Assignation des éditeurs de messages initiaux

Mappez chaque message initial sur un agent réel :

def _assign_initial_post_agents(
    self,
    event_config: EventConfig,
    agent_configs: List[AgentActivityConfig]
) -> EventConfig:

    if not event_config.initial_posts:
        return event_config

    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)

    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"],
    }

    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

        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:
            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

        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
Enter fullscreen mode Exit fullscreen mode

Génération de la configuration des agents par lots

Pour traiter un grand nombre d'agents, générez par lots de 15 :

def _generate_agent_configs_batch(
    self,
    context: str,
    entities: List[EntityNode],
    start_idx: int,
    simulation_requirement: str
) -> List[AgentActivityConfig]:

    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
Enter fullscreen mode Exit fullscreen mode


json
{json.dumps(entity_list, ensure_ascii=False, indent=2)}

...
"""

    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 = {}

    configs = []
    for i, entity in enumerate(entities):
        agent_id = start_idx + i
        cfg = llm_configs.get(agent_id, {})
        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
Enter fullscreen mode Exit fullscreen mode


python

Configurations de secours basées sur des règles

Utilisez des valeurs par défaut adaptées au type d'entité :

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"]:
        return {
            "activity_level": 0.2,
            "posts_per_hour": 0.1,
            "comments_per_hour": 0.05,
            "active_hours": list(range(9, 18)),
            "response_delay_min": 60,
            "response_delay_max": 240,
            "sentiment_bias": 0.0,
            "stance": "neutral",
            "influence_weight": 3.0
        }

    elif entity_type in ["mediaoutlet"]:
        return {
            "activity_level": 0.5,
            "posts_per_hour": 0.8,
            "comments_per_hour": 0.3,
            "active_hours": list(range(7, 24)),
            "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"]:
        return {
            "activity_level": 0.4,
            "posts_per_hour": 0.3,
            "comments_per_hour": 0.5,
            "active_hours": list(range(8, 22)),
            "response_delay_min": 15,
            "response_delay_max": 90,
            "sentiment_bias": 0.0,
            "stance": "neutral",
            "influence_weight": 2.0
        }

    elif entity_type in ["student"]:
        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"]:
        return {
            "activity_level": 0.6,
            "posts_per_hour": 0.4,
            "comments_per_hour": 0.8,
            "active_hours": [12, 13, 19, 20, 21, 22, 23],
            "response_delay_min": 5,
            "response_delay_max": 30,
            "sentiment_bias": 0.0,
            "stance": "neutral",
            "influence_weight": 1.0
        }

    else:
        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
        }
Enter fullscreen mode Exit fullscreen mode

Appel LLM avec réessai et réparation JSON

Pour renforcer la robustesse, gérez les erreurs LLM et réparez le JSON automatiquement :

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)
            )

            content = response.choices[0].message.content
            finish_reason = response.choices[0].finish_reason

            if finish_reason == 'length':
                logger.warning(f"Sortie LLM tronquée (tentative {attempt+1})")
                content = self._fix_truncated_json(content)

            try:
                return json.loads(content)
            except json.JSONDecodeError as e:
                logger.warning(f"Échec de l'analyse JSON (tentative {attempt+1}) : {str(e)[:80]}")
                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é")
Enter fullscreen mode Exit fullscreen mode

Correction du JSON tronqué

def _fix_truncated_json(self, content: str) -> str:
    content = content.strip()
    open_braces = content.count('{') - content.count('}')
    open_brackets = content.count('[') - content.count(']')
    if content and content[-1] not in '",}]':
        content += '"'
    content += ']' * open_brackets
    content += '}' * open_braces
    return content
Enter fullscreen mode Exit fullscreen mode

Réparation JSON avancée

def _try_fix_config_json(self, content: str) -> Optional[Dict[str, Any]]:
    import re

    content = self._fix_truncated_json(content)
    json_match = re.search(r'\{[\s\S]*\}', content)
    if json_match:
        json_str = json_match.group()

        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:
            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
Enter fullscreen mode Exit fullscreen mode

Structures de données de configuration

Configuration de l'activité de l'agent

@dataclass
class AgentActivityConfig:
    agent_id: int
    entity_uuid: str
    entity_name: str
    entity_type: str

    activity_level: float = 0.5
    posts_per_hour: float = 1.0
    comments_per_hour: float = 2.0
    active_hours: List[int] = field(default_factory=lambda: list(range(8, 23)))
    response_delay_min: int = 5
    response_delay_max: int = 60
    sentiment_bias: float = 0.0
    stance: str = "neutral"
    influence_weight: float = 1.0
Enter fullscreen mode Exit fullscreen mode

Configuration de la simulation temporelle

@dataclass
class TimeSimulationConfig:
    total_simulation_hours: int = 72
    minutes_per_round: int = 60
    agents_per_hour_min: int = 5
    agents_per_hour_max: int = 20
    peak_hours: List[int] = field(default_factory=lambda: [19, 20, 21, 22])
    peak_activity_multiplier: float = 1.5
    off_peak_hours: List[int] = field(default_factory=lambda: [0, 1, 2, 3, 4, 5])
    off_peak_activity_multiplier: float = 0.05
    morning_hours: List[int] = field(default_factory=lambda: [6, 7, 8])
    morning_activity_multiplier: float = 0.4
    work_hours: List[int] = field(default_factory=lambda: [9, 10, 11, 12, 13, 14, 15, 16, 17, 18])
    work_activity_multiplier: float = 0.7
Enter fullscreen mode Exit fullscreen mode

Paramètres de simulation complets

@dataclass
class SimulationParameters:
    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,
        }
Enter fullscreen mode Exit fullscreen mode

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 :

  1. Génération pas à pas : Séquencez en étapes (temps → événements → agents → plateformes)
  2. Traitement par lots : Traitez 15 agents par lot pour éviter les limites de contexte
  3. Réparation JSON : Gérez la troncature et les erreurs de structure automatiquement
  4. Replis basés sur des règles : Fournissez des valeurs par défaut robustes
  5. Modèles spécifiques au type : Adaptez l'activité selon le type d'agent
  6. Validation et correction : Vérifiez et corrigez les incohérences (ex : agents_per_hour > total_agents)

Avec cette approche, vous pouvez automatiser la génération de configurations d'agents IA pour des simulations sociales à grande échelle, de façon fiable et maintenable.

Top comments (0)