Introducción
Configurar cientos de agentes de IA para una simulación de redes sociales puede ser una tarea abrumadora. Cada agente necesita horarios de actividad, frecuencias de publicación, retrasos de respuesta, pesos de influencia y posturas. Hacer esto manualmente es inviable y consume muchas horas de trabajo.
MiroFish automatiza este proceso utilizando generación de configuración impulsada por LLM. El sistema analiza tus documentos, el grafo de conocimiento y los requisitos de simulación, y genera automáticamente configuraciones detalladas para cada agente.
Sin embargo, los LLM pueden fallar: salidas truncadas, JSON roto y límites de tokens son comunes.
Esta guía cubre cómo implementar un pipeline robusto y automatizado:
- 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 según tipo (Estudiante, Oficial, Medios)
- Lógica de validación y corrección
💡 El pipeline procesa más de 100 agentes mediante múltiples llamadas a la API. Apidog se usó para validar esquemas de solicitud/respuesta, detectar errores de formato JSON antes de producción y generar casos de prueba para escenarios como salidas truncadas del LLM.
Todo el código mostrado está basado en la implementación real en MiroFish.
Descripción General de la Arquitectura
La generación de configuración se realiza en un pipeline, separando el proceso en etapas:
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Contexto │ ──► │ Config de │ ──► │ Config de │
│ Constructor │ │ Tiempo │ │ Eventos │
└─────────────────┘ └─────────────────┘ └─────────────────┘
│
▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Ensamblaje │ ◄── │ Config de │ ◄── │ Lotes de │
│ Final de │ │ Plataforma │ │ Agentes │
│ Config │ │ │ │ Config │
└─────────────────┘ └─────────────────┘ └─────────────────┘
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
No intentes generar toda la configuración de una sola vez; excederás los límites de tokens de los modelos LLM. Divide el proceso en etapas y usa procesamiento por lotes.
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 # 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}")
context = self._build_context(
simulation_requirement=simulation_requirement,
document_text=document_text,
entities=entities
)
reasoning_parts = []
# Paso 1: 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: 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: Configuración de agentes 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
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
Ventajas de este enfoque:
- Cada llamada a LLM es manejable.
- El usuario recibe feedback de progreso.
- Permite recuperación parcial si una etapa falla.
Construyendo el Contexto
El contexto debe contener solo la información relevante para cada etapa, respetando los límites de tokens.
def _build_context(
self,
simulation_requirement: str,
document_text: str,
entities: List[EntityNode]
) -> str:
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}",
]
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...(documento truncado)"
context_parts.append(f"\n## Documento Original\n{doc_text}")
return "\n".join(context_parts)
Resumen de Entidades
Agrupa y resume entidades por tipo para que el contexto sea compacto y útil.
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)} entidades)")
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)
Ejemplo de salida:
### 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
Generación de Configuración de Tiempo
Define la duración de la simulación y los ciclos de actividad.
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á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:
- Zona horaria Pekín
- 0-5 AM: Casi sin actividad (coef. 0.05)
- 6-8 AM: Gradual (coef. 0.4)
- 9-18 PM: Moderada (coef. 0.7)
- 19-22 PM: Pico (coef. 1.5)
- 23 PM: Baja (coef. 0.5)
### Formato de retorno JSON (sin markdown):
{{
"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"
}}
"""
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)
Validación y Corrección
Asegura que los valores generados sean válidos y estén dentro de los rangos permitidos.
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:
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
)
Configuración de Tiempo Predeterminada
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": "Usando la configuración predeterminada de la zona horaria china"
}
Generación de Configuración de Eventos
Define las publicaciones iniciales, temas candentes y narrativa.
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"""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.
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"}},
...
],
"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
Asigna cada publicación inicial a un agente real, manejando alias y coincidencias.
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"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
Divide la generación de agentes en lotes de 15 para evitar saturar el LLM. Usa patrones por tipo de agente para valores de respaldo.
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"""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)}
python
Tarea
- El horario debe seguir los hábitos chinos: 0-5 AM casi sin actividad, 19-22 PM la más activa.
- Instituciones oficiales: actividad baja, horario laboral, respuesta lenta, alta influencia.
- Medios: actividad moderada, todo el día, respuesta rápida, alta influencia.
- Individuos: actividad alta, principalmente nocturna, respuesta rápida, baja influencia.
- Expertos: actividad moderada, influencia media-alta.
El system prompt debe pedir formato JSON puro.
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 = {}
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
Configuraciones de Respaldo Basadas en Reglas
Utiliza este patrón para generar configuraciones razonables si el LLM falla.
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
}
Llamada a LLM con Reintento y Reparación de JSON
Implementa reintentos automáticos y reparación de JSON para manejar salidas truncadas o mal formateadas.
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"Salida del LLM truncada (intento {attempt+1})")
content = self._fix_truncated_json(content)
try:
return json.loads(content)
except json.JSONDecodeError as e:
logger.warning(f"Fallo al analizar JSON (intento {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"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()
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
Reparación Avanzada
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
Estructuras de Datos de Configuración
Configuración de Actividad de Agente
@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
Configuración de Simulación de Tiempo
@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
Parámetros Completos de Simulación
@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,
}
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 automática de configuraciones para simulaciones sociales impulsadas por LLM requiere:
- Generación paso a paso: Divide en etapas manejables (tiempo, eventos, agentes, plataformas).
- Procesamiento por lotes: Limita a 15 agentes por lote para evitar problemas de contexto.
- Reparación de JSON: Repara salidas truncadas y formatos inválidos automáticamente.
- Respaldos basados en reglas: Aplica defaults sensatos cuando el LLM falla.
- Patrones específicos por tipo: Diferentes tipos de agentes requieren configuraciones únicas.
- Validación y corrección: Verifica y corrige valores generados automáticamente.
Con este enfoque, puedes escalar la generación de configuraciones de IA para simulaciones complejas de redes sociales de forma robusta, rápida y confiable.
Top comments (0)