Introdução
Configurar centenas de agentes de IA para simulações de mídia social pode ser complexo: é preciso definir cronogramas de atividade, frequências de postagem, atrasos de resposta, pesos de influência e posturas para cada agente. Fazer isso manualmente é inviável em larga escala.
O MiroFish automatiza essa geração de configuração usando LLMs, analisando documentos, grafos de conhecimento e requisitos da simulação para criar perfis detalhados para cada agente.
Principais desafios:
- LLMs podem falhar ou truncar saídas
- JSON inválido é comum
- Limites de tokens impedem geração massiva de dados em uma só chamada
Este guia mostra como implementar um pipeline robusto:
- Geração em etapas (tempo → eventos → agentes → plataformas)
- Processamento em lotes para evitar limites de contexto
- Reparos automáticos para JSON quebrados
- Fallbacks baseados em regras
- Padrões de atividade por tipo de agente
- Validação e autocorreção dos dados
💡 O pipeline processa mais de 100 agentes via múltiplas chamadas de API. O Apidog valida esquemas de request/response em cada etapa, detecta erros de JSON antes que cheguem à produção e gera testes para casos extremos, como saídas truncadas de LLM.
Todo o código apresentado é usado em produção no MiroFish.
Visão Geral da Arquitetura
O pipeline de geração de configuração segue uma cadeia linear, permitindo modularidade, validação em cada etapa e fácil troubleshooting:
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ 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 │
└─────────────────┘ └─────────────────┘ └─────────────────┘
Estrutura de Arquivos
backend/app/services/
├── simulation_config_generator.py # Lógica principal
├── ontology_generator.py # Geração de ontologia
└── zep_entity_reader.py # Filtragem de entidades
backend/app/models/
├── task.py # Rastreamento de tarefas
└── project.py # Estado do projeto
Estratégia de Geração Passo a Passo
Para evitar limites de token, a geração ocorre em estágios. Cada lote trata até 15 agentes:
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 # Tempo, 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 = []
# 1. Configuração de tempo
report_progress(1, "Gerando configuração de tempo...")
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"Configuração de tempo: {time_config_result.get('reasoning', 'Sucesso')}")
# 2. Configuração de eventos
report_progress(2, "Gerando configuração de eventos e tópicos quentes...")
event_config_result = self._generate_event_config(context, simulation_requirement, entities)
event_config = self._parse_event_config(event_config_result)
reasoning_parts.append(f"Configuração de eventos: {event_config_result.get('reasoning', 'Sucesso')}")
# 3-N. Configuração de agentes em 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"Gerando configuração 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"Configuração de agente: Gerados {len(all_agent_configs)} agentes")
# Atribuir publicadores de postagens iniciais
event_config = self._assign_initial_post_agents(event_config, all_agent_configs)
# Etapa final: Configuração de plataforma
report_progress(total_steps, "Gerando configuração 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
Vantagens dessa abordagem:
- Cada chamada LLM é pequena e específica
- Permite progresso parcial e recuperação de falhas
- Usuário recebe feedback contínuo
Construindo Contexto
O contexto precisa ser detalhado e sucinto para caber nos limites do modelo:
def _build_context(
self,
simulation_requirement: str,
document_text: str,
entities: List[EntityNode]
) -> str:
# Sumário de entidades
entity_summary = self._summarize_entities(entities)
context_parts = [
f"## Requisito de Simulação\n{simulation_requirement}",
f"\n## Informações da Entidade ({len(entities)} entidades)\n{entity_summary}",
]
# Adicionar texto do documento se houver espaço
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)
Sumarização de Entidades
Entidades são agrupadas e resumidas por tipo, limitando o output:
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" ... e mais {len(type_entities) - display_count}")
return "\n".join(lines)
Exemplo de saída:
### Estudante (45 entidades)
- Zhang Wei: Ativo no grêmio estudantil, frequentemente posta sobre eventos no campus e pressão acadêmica...
- Li Ming: Estudante de pós-graduação pesquisando ética da IA, frequentemente compartilha notícias de tecnologia...
... e mais 43
### Universidade (3 entidades)
- Universidade de Wuhan: Conta oficial, posta anúncios e notícias...
Geração da Configuração de Tempo
Defina parâmetros de duração e padrões de atividade:
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"""Com base nos seguintes requisitos de simulação, gere a configuração de tempo.
{context_truncated}
## Tarefa
Gere o JSON da configuração de tempo.
### Princípios Básicos:
- Base de usuários chinesa, hábitos de fuso horário de Pequim
- 0-5h: Quase nenhuma atividade (0.05)
- 6-8h: Acordando (0.4)
- 9-18h: Trabalho (0.7)
- 19-22h: Pico noite (1.5)
- 23h: Diminuindo (0.5)
### Formato de retorno JSON:
{{
"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": "Explicação da configuração de tempo"
}}
"""
system_prompt = "Você é um especialista em simulação de mídia social. Retorne o formato JSON puro."
try:
return self._call_llm_with_retry(prompt, system_prompt)
except Exception as e:
logger.warning(f"Falha na geração de LLM da configuração de tempo: {e}, usando padrão")
return self._get_default_time_config(num_entities)
Análise e Validação
Corrija valores fora do range permitido:
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
)
Configuração de Tempo Padrão
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 configuração padrão de fuso horário chinês"
}
Geração da Configuração de Eventos
Define tópicos quentes, narrativa e postagens iniciais:
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"""Com base nos seguintes requisitos de simulação, gere a configuração de eventos.
Requisito de Simulação: {simulation_requirement}
{context_truncated}
## Tipos de Entidade Disponíveis e Exemplos
{type_info}
## Tarefa
Gere o JSON da configuração de eventos:
- Extraia palavras-chave de tópicos quentes
- Descreva a direção narrativa
- Projete postagens iniciais, **cada postagem deve especificar poster_type**
**Importante**: poster_type deve ser selecionado entre os "Tipos de Entidade Disponíveis".
Formato JSON:
{{
"hot_topics": ["palavra_chave1", "palavra_chave2", ...],
"narrative_direction": "<descrição>",
"initial_posts": [
{{"content": "Conteúdo", "poster_type": "Tipo de Entidade"}},
...
],
"reasoning": "<explicação>"
}}
"""
system_prompt = "Você é um especialista em análise de opinião. Retorne o formato JSON puro."
try:
return self._call_llm_with_retry(prompt, system_prompt)
except Exception as e:
logger.warning(f"Falha na geração de LLM da configuração de eventos: {e}, usando padrão")
return {
"hot_topics": [],
"narrative_direction": "",
"initial_posts": [],
"reasoning": "Usando configuração padrão"
}
Atribuindo Publicadores
Associa cada postagem inicial ao agente real mais apropriado:
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"Nenhum agente correspondente para o tipo '{poster_type}', usando agente de maior influência")
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", "Desconhecido"),
"poster_agent_id": matched_agent_id
})
logger.info(f"Atribuição de postagem inicial: poster_type='{poster_type}' -> agent_id={matched_agent_id}")
event_config.initial_posts = updated_posts
return event_config
Geração da Configuração de Agentes em Lote
Processa grupos de até 15 entidades por vez:
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"""Com base nas seguintes informações, gere a configuração de atividade de mídia social para cada entidade.
Requisito de Simulação: {simulation_requirement}
## Lista de Entidades
json
{json.dumps(entity_list, ensure_ascii=False, indent=2)}
## Tarefa
Gere a configuração de atividade para cada entidade.
- O horário deve seguir hábitos chineses: 0-5h quase nenhuma atividade, 19-22h mais ativo
- Instituições oficiais: baixa atividade, horário de trabalho, resposta lenta, alta influência
- Mídia: atividade moderada, todo o dia, resposta rápida, alta influência
- Indivíduos: alta atividade, principalmente à noite, resposta rápida, baixa influência
- Figuras públicas/Especialistas: atividade moderada, influência média-alta
Retorne JSON puro.
"""
system_prompt = "Você é um especialista em análise de comportamento de mídia social. Retorne o 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"Falha na geração de LLM do lote de configuração de agente: {e}, usando geração baseada em regras")
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
python
Configurações de Fallback Baseadas em Regras
Quando o LLM falha, use padrões pré-definidos:
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
}
Chamada LLM com Tentativa e Reparo de JSON
Gestão robusta de falhas na integração com modelos de linguagem:
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"Saída LLM truncada (tentativa {attempt+1})")
content = self._fix_truncated_json(content)
try:
return json.loads(content)
except json.JSONDecodeError as e:
logger.warning(f"Falha na análise de JSON (tentativa {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"Falha na chamada LLM (tentativa {attempt+1}): {str(e)[:80]}")
last_error = e
import time
time.sleep(2 * (attempt + 1))
raise last_error or Exception("Falha na chamada LLM")
Corrigindo 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
Reparo Avançado de JSON
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
Estruturas de Dados da Configuração
Configuração de Atividade do 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
Configuração de Simulação de Tempo
@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 da Simulação
@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,
}
Tabela Resumo: Padrões de Tipo de Agente
| Tipo de Agente | Atividade | Horas Ativas | Postagens/Hora | Comentários/Hora | Resposta (min) | Influência |
|---|---|---|---|---|---|---|
| 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 |
Conclusão
Para automatizar configurações complexas de simulação com LLMs de forma segura e escalável:
- Divida o pipeline em etapas (tempo, eventos, agentes, plataformas)
- Processe agentes em lotes de até 15 para cada chamada
- Implemente reparo e validação de JSON para lidar com saídas truncadas
- Utilize fallbacks baseados em regras quando a IA falhar
- Ajuste padrões por tipo de agente para realismo
- Valide e corrija parâmetros críticos antes do deploy
Com essas práticas, é possível gerar configurações ricas para centenas de agentes de IA de forma confiável e automatizada.
Top comments (0)