DEV Community

Cover image for 100+ Agenten Konfigurationen erstellen mit LLMs und Batch-Verarbeitung
Emre Demir
Emre Demir

Posted on • Originally published at apidog.com

100+ Agenten Konfigurationen erstellen mit LLMs und Batch-Verarbeitung

Einführung

Die Konfiguration von Hunderten KI-Agenten für eine Social-Media-Simulation ist eine Herausforderung. Jeder Agent benötigt individuelle Aktivitätspläne, Veröffentlichungsfrequenzen, Antwortverzögerungen, Einflussgewichte und Standpunkte. Manuelles Setup ist extrem zeitaufwändig.

Teste Apidog noch heute

MiroFish automatisiert diesen Prozess durch LLM-gestützte Konfigurationsgenerierung. Das System analysiert Dokumente, Wissensgraphen und Simulationsanforderungen und erzeugt detaillierte Konfigurationen für jeden Agenten.

Problem: LLMs sind unzuverlässig – Ausgaben können abgeschnitten, JSON-Strukturen gebrochen oder durch Token-Limits begrenzt werden.

Diese Schritt-für-Schritt-Anleitung zeigt eine vollständige, produktionsnahe Implementierung:

  • Generierung in Phasen: Zeit → Ereignisse → Agenten → Plattformen
  • Stapelweise Verarbeitung zur Vermeidung von Kontextbeschränkungen
  • Reparatur von fehlerhaften oder abgeschnittenen JSON-Ausgaben
  • Regelbasierte Fallbacks bei LLM-Fehlern
  • Typabhängige Aktivitätsmuster (Student, Offizieller, Medien, etc.)
  • Validierungs- und Korrekturlogik

💡 Die Konfigurationspipeline verarbeitet >100 Agenten durch mehrere API-Aufrufe. Apidog validiert Anfrage-/Antwortschemata, fängt JSON-Fehler frühzeitig ab und ermöglicht das Testen von Grenzfällen wie abgeschnittenen LLM-Ausgaben.

Der komplette Beispielcode stammt direkt aus dem Produktionseinsatz von MiroFish.

Architekturübersicht

Die Konfigurationsgenerierung erfolgt in einer 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     │
└─────────────────┘     └─────────────────┘     └─────────────────┘
Enter fullscreen mode Exit fullscreen mode

Dateistruktur

backend/app/services/
├── simulation_config_generator.py  # Hauptlogik der Konfigurationsgenerierung
├── ontology_generator.py           # Ontologie-Erstellung (geteilt)
└── zep_entity_reader.py            # Entitätenfilterung

backend/app/models/
├── task.py                         # Aufgabenverwaltung
└── project.py                      # Projektstatus
Enter fullscreen mode Exit fullscreen mode

Schritt-für-Schritt-Generierungsstrategie

Die Konfiguration wird nicht in einem Schritt, sondern in kontrollierten Etappen generiert:

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  # Zeit + Events + Agenten-Batches + Plattform
        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 = []

        # Schritt 1: Zeitkonfiguration
        report_progress(1, "Generating time configuration...")
        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', 'Success')}")

        # Schritt 2: Ereigniskonfiguration
        report_progress(2, "Generating event config and hot topics...")
        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', 'Success')}")

        # Schritt 3-N: Agenten-Konfigurationen in Batches
        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"Generating agent config ({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: Generated {len(all_agent_configs)} agents")

        # Initial-Post-Zuweisung
        event_config = self._assign_initial_post_agents(event_config, all_agent_configs)

        # Plattform-Konfiguration
        report_progress(total_steps, "Generating platform configuration...")
        twitter_config = PlatformConfig(platform="twitter", ...) if enable_twitter else None
        reddit_config = PlatformConfig(platform="reddit", ...) if enable_reddit else None

        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

Vorteile des gestuften Ansatzes:

  1. Jeder LLM-Aufruf bleibt übersichtlich und kontrollierbar im Kontext.
  2. Benutzer erhalten Fortschrittsupdates.
  3. Fehler in einer Phase ermöglichen eine partielle Wiederherstellung.

Kontextaufbau

Der Kontext-Builder sammelt alle notwendigen Informationen und berücksichtigt dabei die Token-Limits des LLM:

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

    entity_summary = self._summarize_entities(entities)
    context_parts = [
        f"## Simulation Requirement\n{simulation_requirement}",
        f"\n## Entity Information ({len(entities)} entities)\n{entity_summary}",
    ]

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

    if remaining_length > 0 and document_text:
        doc_text = document_text[:remaining_length]
        if len(document_text) > remaining_length:
            doc_text += "\n...(document truncated)"
        context_parts.append(f"\n## Original Document\n{doc_text}")

    return "\n".join(context_parts)
Enter fullscreen mode Exit fullscreen mode

Entitätszusammenfassung

Entitäten werden nach Typ gruppiert und gekürzt dargestellt:

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 "Unknown"
        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)} entities)")

        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"  ... and {len(type_entities) - display_count} more")

    return "\n".join(lines)
Enter fullscreen mode Exit fullscreen mode

Beispielausgabe:

### Student (45 entities)
- Zhang Wei: Active in student union, frequently posts about campus events and academic pressure...
- Li Ming: Graduate student researching AI ethics, often shares technology news...
... and 43 more

### University (3 entities)
- Wuhan University: Official account, posts announcements and news...
Enter fullscreen mode Exit fullscreen mode

Generierung der Zeitkonfiguration

Die Zeitkonfiguration steuert Simulationsdauer und Aktivitätsmuster:

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"""Based on the following simulation requirements, generate time configuration.

{context_truncated}

## Task
Generate time configuration JSON.

### Basic Principles (adjust based on event type and participant groups):
- User base is Chinese, must follow Beijing timezone habits
- 0-5 AM: Almost no activity (coefficient 0.05)
- 6-8 AM: Gradually waking up (coefficient 0.4)
- 9-18 PM: Work hours, moderate activity (coefficient 0.7)
- 19-22 PM: Evening peak, most active (coefficient 1.5)
- 23 PM: Activity declining (coefficient 0.5)

### Return JSON format (no markdown):

Example:
{{
    "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": "Time configuration explanation"
}}

Field descriptions:
- total_simulation_hours (int): 24-168 hours, shorter for breaking news, longer for ongoing topics
- minutes_per_round (int): 30-120 minutes, recommend 60
- agents_per_hour_min (int): Range 1-{max_agents_allowed}
- agents_per_hour_max (int): Range 1-{max_agents_allowed}
- peak_hours (int array): Adjust based on participant groups
- off_peak_hours (int array): Usually late night/early morning
- morning_hours (int array): Morning hours
- work_hours (int array): Work hours
- reasoning (string): Brief explanation"""

    system_prompt = "You are a social media simulation expert. Return pure JSON format."

    try:
        return self._call_llm_with_retry(prompt, system_prompt)
    except Exception as e:
        logger.warning(f"Time config LLM generation failed: {e}, using default")
        return self._get_default_time_config(num_entities)
Enter fullscreen mode Exit fullscreen mode

Parsen und Validieren

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

    if agents_per_hour_min > num_entities:
        logger.warning(f"agents_per_hour_min ({agents_per_hour_min}) exceeds total agents ({num_entities}), corrected")
        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}) exceeds total agents ({num_entities}), corrected")
        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)
        logger.warning(f"agents_per_hour_min >= max, corrected to {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
    )
Enter fullscreen mode Exit fullscreen mode

Standard-Zeitkonfiguration

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": "Using default Chinese timezone configuration"
    }
Enter fullscreen mode Exit fullscreen mode

Generierung der Ereigniskonfiguration

Die Ereigniskonfiguration legt Startposts, Hot Topics und die narrative Richtung fest:

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 "Unknown" for e in entities
    ))

    type_examples = {}
    for e in entities:
        etype = e.get_entity_type() or "Unknown"
        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"""Based on the following simulation requirements, generate event configuration.
...
    # Siehe Original oben für den vollständigen Prompt
...
    system_prompt = "You are an opinion analysis expert. Return pure JSON format."

    try:
        return self._call_llm_with_retry(prompt, system_prompt)
    except Exception as e:
        logger.warning(f"Event config LLM generation failed: {e}, using default")
        return {
            "hot_topics": [],
            "narrative_direction": "",
            "initial_posts": [],
            "reasoning": "Using default configuration"
        }
Enter fullscreen mode Exit fullscreen mode

Zuweisung der Herausgeber für Anfangsbeiträge

Die initialen Beiträge werden nach Typ auf konkrete Agenten gemappt:

def _assign_initial_post_agents(
    self,
    event_config: EventConfig,
    agent_configs: List[AgentActivityConfig]
) -> EventConfig:
    ...
    # Siehe Original oben für den vollständigen Code
    ...
    event_config.initial_posts = updated_posts
    return event_config
Enter fullscreen mode Exit fullscreen mode

Generierung der Batch-Agentenkonfiguration

Agentenkonfigurationen werden stapelweise (je 15 Agenten) generiert, um Token-Limits nicht zu überschreiten:

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 "Unknown",
            "summary": e.summary[:summary_len] if e.summary else ""
        })

    prompt = f"""Based on the following information, generate social media activity configuration for each entity.

Simulation Requirement: {simulation_requirement}

## Entity List
Enter fullscreen mode Exit fullscreen mode


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


### Aufgabe

Generieren Sie die Aktivitätskonfiguration für jede Entität. Hinweise:

- **Zeitmuster:** 0-5 Uhr fast keine Aktivität, 19-22 Uhr Peak
- **Offizielle Institutionen:** geringe Aktivität, Arbeitszeiten, langsame Antwort, hoher Einfluss
- **Medien:** moderate Aktivität, ganztägig, schnelle Antwort, hoher Einfluss
- **Einzelpersonen:** hohe Aktivität abends, schnelle Antwort, geringer Einfluss
- **Experten:** moderate Aktivität, mittlerer-höher Einfluss

system_prompt = "You are a social media behavior analysis expert. Return pure JSON format."

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"Agent config batch LLM generation failed: {e}, using rule-based generation")
    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 "Unknown",
        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

Regelbasierte Fallback-Konfigurationen

Wenn der LLM-Aufruf fehlschlägt, werden folgende Defaults genutzt:

def _generate_agent_config_by_rule(self, entity: EntityNode) -> Dict[str, Any]:
    entity_type = (entity.get_entity_type() or "Unknown").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

LLM-Aufruf mit Wiederholung und JSON-Reparatur

Robuste LLM-Aufrufe mit Reparaturmechanismen für JSON:

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"LLM output truncated (attempt {attempt+1})")
                content = self._fix_truncated_json(content)

            try:
                return json.loads(content)
            except json.JSONDecodeError as e:
                logger.warning(f"JSON parse failed (attempt {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"LLM call failed (attempt {attempt+1}): {str(e)[:80]}")
            last_error = e
            import time
            time.sleep(2 * (attempt + 1))

    raise last_error or Exception("LLM call failed")
Enter fullscreen mode Exit fullscreen mode

Beheben von abgeschnittenem JSON

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

Erweiterte JSON-Reparatur

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

Konfigurationsdatenstrukturen

Agenten-Aktivitätskonfiguration

@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

Zeitsimulationskonfiguration

@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

Vollständige Simulationsparameter

@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

Übersichtstabelle: Agententyp-Muster

Agententyp Aktivität Aktive Stunden Beiträge/Stunde Kommentare/Stunde Antwort (Min.) Einfluss
University 0.2 9-17 0.1 0.05 60-240 3.0
GovernmentAgency 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
Professor 0.4 8-21 0.3 0.5 15-90 2.0
Student 0.8 8-12, 18-23 0.6 1.5 1-15 0.8
Alumni 0.6 12-13, 19-23 0.4 0.8 5-30 1.0
Person (default) 0.7 9-13, 18-23 0.5 1.2 2-20 1.0

Fazit

Die LLM-gestützte Konfigurationsgenerierung für Social-Media-Simulationen ist in der Praxis zuverlässig umsetzbar, wenn du folgende Prinzipien beachtest:

  1. Schrittweise Generierung: Aufteilung in Zeit-, Ereignis-, Agenten- und Plattformphasen.
  2. Stapelverarbeitung: Je 15 Agenten pro Batch, um Kontext-Limits einzuhalten.
  3. JSON-Reparatur: Automatisches Beheben von Abschneidungen und Syntaxfehlern.
  4. Fallback-Logik: Regelbasierte Defaults sichern Stabilität bei LLM-Ausfällen.
  5. Typspezifische Muster: Unterschiedliche Agententypen erfordern spezifische Aktivitätsprofile.
  6. Validierung und Korrektur: Plausibilitätschecks und automatische Anpassungen verhindern Fehler.

So baust du eine robuste, automatisierte Simulationskonfiguration für KI-Agenten – direkt einsatzbereit für produktive Social-Media-Szenarien.

Top comments (0)