Introducción
Configurar cientos de agentes de IA para una simulación de redes sociales suena abrumador. Cada agente necesita horarios de actividad, frecuencias de publicación, retrasos de respuesta, pesos de influencia y posiciones de postura. Hacer esto manualmente llevaría horas.
MiroFish automatiza esto con la generación de configuración impulsada por LLM. El sistema analiza sus documentos, grafo de conocimiento y requisitos de simulación, luego genera configuraciones detalladas para cada agente.
El desafío: los LLM pueden fallar. Las salidas se truncan. El JSON se rompe. Los límites de tokens son un problema.
Esta guía cubre la implementación completa:
- Generación paso a paso (tiempo → eventos → agentes → plataformas)
- Procesamiento por lotes para evitar límites de contexto
- Estrategias de reparación de JSON para salidas truncadas
- Configuraciones de respaldo basadas en reglas cuando el LLM falla
- Patrones de actividad del agente por tipo (Estudiante vs Oficial vs Medios)
- Lógica de validación y corrección
💡
El pipeline de generación de configuración procesa más de 100 agentes a través de una serie de llamadas a la API. Apidog se utilizó para validar los esquemas de solicitud/respuesta en cada etapa, detectar errores de formato JSON antes de que llegaran a producción y generar casos de prueba para escenarios extremos como salidas de LLM truncadas.
Todo el código proviene del uso en producción en MiroFish.
Descripción General de la Arquitectura
El generador de configuración utiliza un enfoque de pipeline:
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Contexto │ ──► │ Config de │ ──► │ Config de │
│ Constructor │ │ Tiempo │ │ Eventos │
│ │ │ Generador │ │ Generador │
│ │ │ │ │ │
│ - Requisito de │ │ - Horas totales │ │ - Publicaciones │
│ Simulación │ │ - Minutos/ronda │ │ iniciales │
│ - Resumen de │ │ - Horas pico │ │ - Temas candentes│
│ entidad │ │ - Multiplicador │ │ - Dirección de │
│ - Texto de │ │ de actividad │ │ narrativa │
│ documento │ │ │ │ │
└─────────────────┘ └─────────────────┘ └─────────────────┘
│
▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Ensamblaje │ ◄── │ Config de │ ◄── │ Lotes de │
│ Final de │ │ Plataforma │ │ Agentes │
│ Config │ │ │ │ Config │
│ │ │ │ │ │
│ - Unir todo │ │ - Parámetros de │ │ - 15 agentes │
│ - Validar │ │ Twitter │ │ por lote │
│ - Guardar JSON │ │ - Parámetros de │ │ - N lotes │
│ │ │ Reddit │ │ │
│ │ │ - Umbral viral │ │ │
└─────────────────┘ └─────────────────┘ └─────────────────┘
Estructura de Archivos
backend/app/services/
├── simulation_config_generator.py # Lógica principal de generación de configuración
├── ontology_generator.py # Generación de ontología (compartida)
└── zep_entity_reader.py # Filtrado de entidades
backend/app/models/
├── task.py # Seguimiento de tareas
└── project.py # Estado del proyecto
Estrategia de Generación Paso a Paso
Generar todas las configuraciones a la vez excedería los límites de tokens. En cambio, el sistema genera en etapas:
class SimulationConfigGenerator:
# Cada lote genera configuraciones para 15 agentes
AGENTS_PER_BATCH = 15
# Límites de contexto
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:
# Calcular pasos totales
num_batches = math.ceil(len(entities) / self.AGENTS_PER_BATCH)
total_steps = 3 + num_batches # Tiempo + Eventos + N Lotes de Agentes + Plataforma
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}")
# Construir contexto
context = self._build_context(
simulation_requirement=simulation_requirement,
document_text=document_text,
entities=entities
)
reasoning_parts = []
# Paso 1: Generar configuración de tiempo
report_progress(1, "Generando configuración de tiempo...")
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')}")
# Paso 2: Generar configuración de eventos
report_progress(2, "Generando configuración de eventos y temas candentes...")
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')}")
# Pasos 3-N: Generar configuraciones de agente en lotes
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"Generando configuración de agente ({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")
# Asignar publicadores de publicaciones iniciales
event_config = self._assign_initial_post_agents(event_config, all_agent_configs)
# Paso final: Configuración de plataforma
report_progress(total_steps, "Generando configuración de plataforma...")
twitter_config = PlatformConfig(platform="twitter", ...) if enable_twitter else None
reddit_config = PlatformConfig(platform="reddit", ...) if enable_reddit else None
# Ensamblar configuración 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
Este enfoque por etapas:
- Mantiene cada llamada a LLM enfocada y manejable
- Proporciona actualizaciones de progreso al usuario
- Permite la recuperación parcial si una etapa falla
Construyendo el Contexto
El constructor de contexto ensambla información relevante respetando los límites de tokens:
def _build_context(
self,
simulation_requirement: str,
document_text: str,
entities: List[EntityNode]
) -> str:
# Resumen de entidades
entity_summary = self._summarize_entities(entities)
context_parts = [
f"## Requisito de Simulación\n{simulation_requirement}",
f"\n## Información de Entidad ({len(entities)} entidades)\n{entity_summary}",
]
# Añadir texto del documento si el espacio lo permite
current_length = sum(len(p) for p in context_parts)
remaining_length = self.MAX_CONTEXT_LENGTH - current_length - 500 # búfer de 500 caracteres
if remaining_length > 0 and document_text:
doc_text = document_text[:remaining_length]
if len(document_text) > remaining_length:
doc_text += "\n...(documento truncado)"
context_parts.append(f"\n## Documento Original\n{doc_text}")
return "\n".join(context_parts)
Resumen de Entidades
Las entidades se resumen por tipo:
def _summarize_entities(self, entities: List[EntityNode]) -> str:
lines = []
# Agrupar por tipo
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)} entidades)")
# Mostrar número limitado con longitud de resumen limitada
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" ... y {len(type_entities) - display_count} más")
return "\n".join(lines)
Esto produce una salida como:
### Estudiante (45 entidades)
- Zhang Wei: Activo en la unión estudiantil, publica frecuentemente sobre eventos del campus y presión académica...
- Li Ming: Estudiante de posgrado investigando la ética de la IA, a menudo comparte noticias de tecnología...
... y 43 más
### Universidad (3 entidades)
- Universidad de Wuhan: Cuenta oficial, publica anuncios y noticias...
Generación de Configuración de Tiempo
La configuración de tiempo determina la duración de la simulación y los patrones de actividad:
def _generate_time_config(self, context: str, num_entities: int) -> Dict[str, Any]:
# Truncar contexto para este paso específico
context_truncated = context[:self.TIME_CONFIG_CONTEXT_LENGTH]
# Calcular valor máximo permitido (90% del recuento de agentes)
max_agents_allowed = max(1, int(num_entities * 0.9))
prompt = f"""Basándose en los siguientes requisitos de simulación, genere la configuración de tiempo.
{context_truncated}
## Tarea
Generar el JSON de configuración de tiempo.
### Principios Básicos (ajuste según el tipo de evento y los grupos de participantes):
- La base de usuarios es china, debe seguir los hábitos de la zona horaria de Pekín
- 0-5 AM: Casi ninguna actividad (coeficiente 0.05)
- 6-8 AM: Despertando gradualmente (coeficiente 0.4)
- 9-18 PM: Horas de trabajo, actividad moderada (coeficiente 0.7)
- 19-22 PM: Pico vespertino, más activo (coeficiente 1.5)
- 23 PM: Actividad disminuyendo (coeficiente 0.5)
### Formato de retorno JSON (sin markdown):
Ejemplo:
{{
"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": "Explicación de la configuración de tiempo"
}}
Descripción de los campos:
- total_simulation_hours (int): 24-168 horas, más corto para noticias de última hora, más largo para temas en curso
- minutes_per_round (int): 30-120 minutos, se recomienda 60
- agents_per_hour_min (int): Rango 1-{max_agents_allowed}
- agents_per_hour_max (int): Rango 1-{max_agents_allowed}
- peak_hours (array de int): Ajustar según los grupos de participantes
- off_peak_hours (array de int): Normalmente de madrugada/temprano por la mañana
- morning_hours (array de int): Horas de la mañana
- work_hours (array de int): Horas de trabajo
- reasoning (string): Breve explicación"""
system_prompt = "Eres un experto en simulación de redes sociales. Devuelve el formato JSON puro."
try:
return self._call_llm_with_retry(prompt, system_prompt)
except Exception as e:
logger.warning(f"La generación de LLM de configuración de tiempo falló: {e}, usando valores predeterminados")
return self._get_default_time_config(num_entities)
Análisis y Validación de la Configuración de Tiempo
def _parse_time_config(self, result: Dict[str, Any], num_entities: int) -> TimeSimulationConfig:
# Obtener valores brutos
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))
# Validar y corregir: asegurar que no exceda el número total de agentes
if agents_per_hour_min > num_entities:
logger.warning(f"agents_per_hour_min ({agents_per_hour_min}) excede el total de agentes ({num_entities}), corregido")
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}) excede el total de agentes ({num_entities}), corregido")
agents_per_hour_max = max(agents_per_hour_min + 1, num_entities // 2)
# Asegurar que 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, corregido a {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
)
Configuración de Tiempo Predeterminada (Zona Horaria China)
def _get_default_time_config(self, num_entities: int) -> Dict[str, Any]:
return {
"total_simulation_hours": 72,
"minutes_per_round": 60, # 1 hora por ronda
"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": "Usando la configuración predeterminada de la zona horaria china"
}
Generación de Configuración de Eventos
La configuración de eventos define las publicaciones iniciales, los temas candentes y la dirección narrativa:
def _generate_event_config(
self,
context: str,
simulation_requirement: str,
entities: List[EntityNode]
) -> Dict[str, Any]:
# Obtener tipos de entidad disponibles para referencia del LLM
entity_types_available = list(set(
e.get_entity_type() or "Unknown" for e in entities
))
# Mostrar ejemplos por tipo
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"""Basándose en los siguientes requisitos de simulación, genere la configuración de eventos.
Requisito de Simulación: {simulation_requirement}
{context_truncated}
## Tipos de Entidades Disponibles y Ejemplos
{type_info}
## Tarea
Generar el JSON de configuración de eventos:
- Extraer palabras clave de temas candentes
- Describir la dirección narrativa
- Diseñar publicaciones iniciales, **cada publicación debe especificar poster_type**
**Importante**: poster_type debe seleccionarse de los "Tipos de Entidades Disponibles" anteriores, para que las publicaciones iniciales puedan asignarse a los agentes apropiados.
Por ejemplo: las declaraciones oficiales deben ser publicadas por tipos Oficial/Universidad, las noticias por MediaOutlet, las opiniones estudiantiles por Estudiante.
Devolver el formato JSON (sin markdown):
{{
"hot_topics": ["palabraclave1", "palabraclave2", ...],
"narrative_direction": "<descripción de la dirección narrativa>",
"initial_posts": [
{{"content": "Contenido de la publicación", "poster_type": "Tipo de Entidad (debe coincidir con los tipos disponibles)"}},
...
],
"reasoning": "<breve explicación>"
}}"""
system_prompt = "Eres un experto en análisis de opinión. Devuelve el formato JSON puro."
try:
return self._call_llm_with_retry(prompt, system_prompt)
except Exception as e:
logger.warning(f"La generación de LLM de configuración de eventos falló: {e}, usando valores predeterminados")
return {
"hot_topics": [],
"narrative_direction": "",
"initial_posts": [],
"reasoning": "Usando la configuración predeterminada"
}
Asignación de Publicadores de Publicaciones Iniciales
Después de generar las publicaciones iniciales, asigne a los agentes reales:
def _assign_initial_post_agents(
self,
event_config: EventConfig,
agent_configs: List[AgentActivityConfig]
) -> EventConfig:
if not event_config.initial_posts:
return event_config
# Indexar agentes por tipo
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)
# Mapeo de alias de tipo (maneja variaciones de 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"],
}
# Rastrear índices utilizados para evitar reutilizar el mismo agente
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. Coincidencia directa
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. Coincidencia por 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. Respaldo: usar el agente de mayor influencia
if matched_agent_id is None:
logger.warning(f"No hay agente coincidente para el tipo '{poster_type}', usando el agente de mayor influencia")
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", "Unknown"),
"poster_agent_id": matched_agent_id
})
logger.info(f"Asignación de publicación inicial: poster_type='{poster_type}' -> agent_id={matched_agent_id}")
event_config.initial_posts = updated_posts
return event_config
Generación por Lotes de la Configuración de Agentes
Generar configuraciones para cientos de agentes a la vez excedería los límites de tokens. El sistema procesa en lotes de 15:
def _generate_agent_configs_batch(
self,
context: str,
entities: List[EntityNode],
start_idx: int,
simulation_requirement: str
) -> List[AgentActivityConfig]:
# Construir información de entidad con longitud de resumen limitada
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"""Basándose en la siguiente información, genere la configuración de actividad de redes sociales para cada entidad.
Requisito de Simulación: {simulation_requirement}
## Lista de Entidades
```json
{json.dumps(entity_list, ensure_ascii=False, indent=2)}
Tarea
Generar la configuración de actividad para cada entidad. Nota:
- El horario debe seguir los hábitos chinos: 0-5 AM casi sin actividad, 19-22 PM la más activa
- Instituciones oficiales (Universidad/AgenciaGubernamental): Actividad baja (0.1-0.3), horario laboral (9-17), respuesta lenta (60-240 min), alta influencia (2.5-3.0)
- Medios de comunicación (MediaOutlet): Actividad moderada (0.4-0.6), actividad durante todo el día (8-23), respuesta rápida (5-30 min), alta influencia (2.0-2.5)
- Individuos (Estudiante/Persona/Exalumno): Actividad alta (0.6-0.9), principalmente por la noche (18-23), respuesta rápida (1-15 min), baja influencia (0.8-1.2)
- Figuras públicas/Expertos: Actividad moderada (0.4-0.6), influencia media-alta (1.5-2.0)
system_prompt = "Eres un experto en análisis de comportamiento en redes sociales. Devuelve el formato JSON puro."
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 generación por lotes de LLM de configuración de agente falló: {e}, usando generación basada en reglas")
llm_configs = {}
# Construir objetos AgentActivityConfig
configs = []
for i, entity in enumerate(entities):
agent_id = start_idx + i
cfg = llm_configs.get(agent_id, {})
# Usar configuración de respaldo basada en reglas si el LLM falló
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
Configuraciones de Respaldo Basadas en Reglas
Cuando el LLM falla, use patrones predefinidos:
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"]:
# Institución oficial: horario laboral, baja frecuencia, alta influencia
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"]:
# Medios de comunicación: actividad todo el día, frecuencia moderada, alta influencia
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"]:
# Experto/Profesor: trabajo + noche, frecuencia moderada
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"]:
# Estudiante: pico nocturno, alta frecuencia
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"]:
# Exalumno: enfocado en la noche
return {
"activity_level": 0.6,
"posts_per_hour": 0.4,
"comments_per_hour": 0.8,
"active_hours": [12, 13, 19, 20, 21, 22, 23], # Almuerzo + noche
"response_delay_min": 5,
"response_delay_max": 30,
"sentiment_bias": 0.0,
"stance": "neutral",
"influence_weight": 1.0
}
else:
# Persona predeterminada: pico nocturno
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
}
Llamada a LLM con Reintento y Reparación de JSON
Las llamadas a LLM fallan. Las salidas se truncan. El JSON se rompe. El sistema maneja todo esto:
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) # Bajar temperatura al reintentar
)
content = response.choices[0].message.content
finish_reason = response.choices[0].finish_reason
# Comprobar si está truncado
if finish_reason == 'length':
logger.warning(f"Salida del LLM truncada (intento {attempt+1})")
content = self._fix_truncated_json(content)
# Intentar analizar JSON
try:
return json.loads(content)
except json.JSONDecodeError as e:
logger.warning(f"Fallo al analizar JSON (intento {attempt+1}): {str(e)[:80]}")
# Intentar reparar JSON
fixed = self._try_fix_config_json(content)
if fixed:
return fixed
last_error = e
except Exception as e:
logger.warning(f"La llamada al LLM falló (intento {attempt+1}): {str(e)[:80]}")
last_error = e
import time
time.sleep(2 * (attempt + 1))
raise last_error or Exception("La llamada al LLM falló")
Reparando JSON Truncado
def _fix_truncated_json(self, content: str) -> str:
content = content.strip()
# Contar corchetes no cerrados
open_braces = content.count('{') - content.count('}')
open_brackets = content.count('[') - content.count(']')
# Comprobar si la cadena no está cerrada
if content and content[-1] not in '",}]':
content += '"'
# Cerrar corchetes
content += ']' * open_brackets
content += '}' * open_braces
return content
Reparación Avanzada de JSON
def _try_fix_config_json(self, content: str) -> Optional[Dict[str, Any]]:
import re
# Corregir truncamiento
content = self._fix_truncated_json(content)
# Extraer la parte JSON
json_match = re.search(r'\{[\s\S]*\}', content)
if json_match:
json_str = json_match.group()
# Eliminar saltos de línea en cadenas
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:
# Intentar eliminar caracteres de control
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
Estructuras de Datos de Configuración
Configuración de Actividad de Agente
@dataclass
class AgentActivityConfig:
"""Configuración de actividad de agente individual"""
agent_id: int
entity_uuid: str
entity_name: str
entity_type: str
# Nivel de actividad (0.0-1.0)
activity_level: float = 0.5
# Frecuencia de publicación (por hora)
posts_per_hour: float = 1.0
comments_per_hour: float = 2.0
# Horas activas (formato de 24 horas, 0-23)
active_hours: List[int] = field(default_factory=lambda: list(range(8, 23)))
# Velocidad de respuesta (retraso de reacción en minutos simulados)
response_delay_min: int = 5
response_delay_max: int = 60
# Tendencia de sentimiento (-1.0 a 1.0, negativo a positivo)
sentiment_bias: float = 0.0
# Postura sobre temas específicos
stance: str = "neutral" # supportive, opposing, neutral, observer
# Peso de influencia (afecta la probabilidad de ser visto)
influence_weight: float = 1.0
Configuración de Simulación de Tiempo
@dataclass
class TimeSimulationConfig:
"""Configuración de simulación de tiempo (zona horaria china)"""
total_simulation_hours: int = 72 # Por defecto 72 horas (3 días)
minutes_per_round: int = 60 # 60 minutos por ronda
# Agentes activados por hora
agents_per_hour_min: int = 5
agents_per_hour_max: int = 20
# Horas pico (tarde 19-22, más activo en China)
peak_hours: List[int] = field(default_factory=lambda: [19, 20, 21, 22])
peak_activity_multiplier: float = 1.5
# Horas valle (madrugada 0-5, casi sin actividad)
off_peak_hours: List[int] = field(default_factory=lambda: [0, 1, 2, 3, 4, 5])
off_peak_activity_multiplier: float = 0.05
# Horas de la mañana
morning_hours: List[int] = field(default_factory=lambda: [6, 7, 8])
morning_activity_multiplier: float = 0.4
# Horas de trabajo
work_hours: List[int] = field(default_factory=lambda: [9, 10, 11, 12, 13, 14, 15, 16, 17, 18])
work_activity_multiplier: float = 0.7
Parámetros Completos de Simulación
@dataclass
class SimulationParameters:
"""Configuración completa de parámetros de simulación"""
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,
}
Tabla Resumen: Patrones de Tipo de Agente
| Tipo de Agente | Actividad | Horas Activas | Publicaciones/Hora | Comentarios/Hora | Respuesta (min) | Influencia |
|---|---|---|---|---|---|---|
| Universidad | 0.2 | 9-17 | 0.1 | 0.05 | 60-240 | 3.0 |
| AgenciaGubernamental | 0.2 | 9-17 | 0.1 | 0.05 | 60-240 | 3.0 |
| MedioComunicación | 0.5 | 7-23 | 0.8 | 0.3 | 5-30 | 2.5 |
| Profesor | 0.4 | 8-21 | 0.3 | 0.5 | 15-90 | 2.0 |
| Estudiante | 0.8 | 8-12, 18-23 | 0.6 | 1.5 | 1-15 | 0.8 |
| Exalumno | 0.6 | 12-13, 19-23 | 0.4 | 0.8 | 5-30 | 1.0 |
| Persona (predeterminado) | 0.7 | 9-13, 18-23 | 0.5 | 1.2 | 2-20 | 1.0 |
Conclusión
La generación de configuración impulsada por LLM requiere un manejo cuidadoso de:
- Generación paso a paso: Dividir en etapas manejables (tiempo → eventos → agentes → plataformas)
- Procesamiento por lotes: Procesar 15 agentes por lote para evitar límites de contexto
- Reparación de JSON: Manejar el truncamiento con coincidencia de corchetes y escape de cadenas
- Respaldos basados en reglas: Proporcionar valores predeterminados sensatos cuando el LLM falla
- Patrones específicos de tipo: Diferentes tipos de agentes tienen diferentes patrones de actividad
- Validación y corrección: Verificar los valores generados y corregir problemas (ej., agents_per_hour > total_agents)