Introdução
Configurar centenas de agentes de IA para uma simulação de mídia social parece assustador. Cada agente precisa de cronogramas de atividade, frequências de postagem, atrasos de resposta, pesos de influência e posições de postura. Fazer isso manualmente levaria horas.
O MiroFish automatiza isso com a geração de configuração alimentada por LLM. O sistema analisa seus documentos, grafo de conhecimento e requisitos de simulação, então gera configurações detalhadas para cada agente.
O desafio: LLMs podem falhar. Saídas são truncadas. JSON quebra. Limites de token são um problema.
Este guia aborda a implementação completa:
- Geração passo a passo (tempo → eventos → agentes → plataformas)
- Processamento em lote para evitar limites de contexto
- Estratégias de reparo de JSON para saídas truncadas
- Configurações de fallback baseadas em regras quando o LLM falha
- Padrões de atividade do agente por tipo (Estudante vs Oficial vs Mídia)
- Lógica de validação e correção
💡
O pipeline de geração de configuração processa mais de 100 agentes através de uma série de chamadas de API. O Apidog foi usado para validar esquemas de solicitação/resposta em cada estágio, capturar erros de formato JSON antes que chegassem à produção e gerar casos de teste para cenários de ponta, como saídas LLM truncadas.
Todo o código vem do uso em produção no MiroFish.
Visão Geral da Arquitetura
O gerador de configuração utiliza uma abordagem em 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 │
└─────────────────┘ └─────────────────┘ └─────────────────┘
Estrutura de Arquivos
backend/app/services/
├── simulation_config_generator.py # Lógica principal de geração de configuração
├── ontology_generator.py # Geração de ontologia (compartilhada)
└── 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
Gerar todas as configurações de uma só vez excederia os limites de token. Em vez disso, o sistema gera em estágios:
class SimulationConfigGenerator:
# Cada lote gera configurações para 15 agentes
AGENTS_PER_BATCH = 15
# Limites 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 total de etapas
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}")
# Construir contexto
context = self._build_context(
simulation_requirement=simulation_requirement,
document_text=document_text,
entities=entities
)
reasoning_parts = []
# Etapa 1: Gerar 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')}")
# Etapa 2: Gerar 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')}")
# Etapas 3-N: Gerar configurações de agente 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
# Montar configuração 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
Esta abordagem em etapas:
- Mantém cada chamada LLM focada e gerenciável
- Fornece atualizações de progresso ao usuário
- Permite recuperação parcial se uma etapa falhar
Construindo Contexto
O construtor de contexto reúne informações relevantes enquanto respeita os limites de token:
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 # Buffer 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)
Sumarização de Entidades
Entidades são sumarizadas 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)")
# Exibir número limitado com comprimento de sumário limitado
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)
Isso produz uma saída como:
### 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
A configuração de tempo determina a duração da simulação e os padrões de atividade:
def _generate_time_config(self, context: str, num_entities: int) -> Dict[str, Any]:
# Truncar contexto para esta etapa específica
context_truncated = context[:self.TIME_CONFIG_CONTEXT_LENGTH]
# Calcular valor máximo permitido (90% da contagem de agentes)
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 (ajuste com base no tipo de evento e grupos de participantes):
- A base de usuários é chinesa, deve seguir os hábitos de fuso horário de Pequim
- 0-5h: Quase nenhuma atividade (coeficiente 0.05)
- 6-8h: Acordando gradualmente (coeficiente 0.4)
- 9-18h: Horário de trabalho, atividade moderada (coeficiente 0.7)
- 19-22h: Pico noturno, mais ativo (coeficiente 1.5)
- 23h: Atividade diminuindo (coeficiente 0.5)
### Formato de retorno JSON (sem markdown):
Exemplo:
{{
"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"
}}
Descrições dos campos:
- total_simulation_hours (int): 24-168 horas, menor para notícias de última hora, maior para tópicos em andamento
- minutes_per_round (int): 30-120 minutos, recomenda-se 60
- agents_per_hour_min (int): Faixa 1-{max_agents_allowed}
- agents_per_hour_max (int): Faixa 1-{max_agents_allowed}
- peak_hours (array de int): Ajustar com base nos grupos de participantes
- off_peak_hours (array de int): Geralmente tarde da noite/madrugada
- morning_hours (array de int): Horas da manhã
- work_hours (array de int): Horário de trabalho
- reasoning (string): Breve explicação"""
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 da Configuração de Tempo
def _parse_time_config(self, result: Dict[str, Any], num_entities: int) -> TimeSimulationConfig:
# Obter 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 e corrigir: garantir que não exceda a contagem total de agentes
if agents_per_hour_min > num_entities:
logger.warning(f"agents_per_hour_min ({agents_per_hour_min}) excede o total de agentes ({num_entities}), corrigido")
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 o total de agentes ({num_entities}), corrigido")
agents_per_hour_max = max(agents_per_hour_min + 1, num_entities // 2)
# Garantir 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, corrigido para {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
)
Configuração de Tempo Padrão (Fuso Horário Chinês)
def _get_default_time_config(self, num_entities: int) -> Dict[str, Any]:
return {
"total_simulation_hours": 72,
"minutes_per_round": 60, # 1 hora por rodada
"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
A configuração de eventos define postagens iniciais, tópicos quentes e direção narrativa:
def _generate_event_config(
self,
context: str,
simulation_requirement: str,
entities: List[EntityNode]
) -> Dict[str, Any]:
# Obter tipos de entidade disponíveis para referência do LLM
entity_types_available = list(set(
e.get_entity_type() or "Unknown" for e in entities
))
# Mostrar exemplos 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"""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" acima, para que as postagens iniciais possam ser atribuídas aos agentes apropriados.
Por exemplo: Declarações oficiais devem ser postadas por tipos Oficial/Universidade, notícias por Veículos de Mídia, opiniões de estudantes por Estudantes.
Retorne o formato JSON (sem markdown):
{{
"hot_topics": ["palavra_chave1", "palavra_chave2", ...],
"narrative_direction": "<descrição da direção narrativa>",
"initial_posts": [
{{"content": "Conteúdo da postagem", "poster_type": "Tipo de Entidade (deve corresponder aos tipos disponíveis)"}},
...
],
"reasoning": "<breve 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 de Postagens Iniciais
Após gerar as postagens iniciais, associe-as aos agentes reais:
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)
# Mapeamento de apelidos de tipo (lida com variações 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 usados para evitar reutilizar o mesmo 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. Correspondência direta
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. Correspondência por apelido
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. Fallback: usar agente de maior influência
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
Gerar configurações para centenas de agentes de uma vez excederia os limites de token. O sistema processa em lotes de 15:
def _generate_agent_configs_batch(
self,
context: str,
entities: List[EntityNode],
start_idx: int,
simulation_requirement: str
) -> List[AgentActivityConfig]:
# Construir informações da entidade com comprimento de sumário limitado
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. Observação:
- O horário deve seguir os hábitos chineses: 0-5h quase nenhuma atividade, 19-22h mais ativo
- Instituições oficiais (Universidade/Agência Governamental): Baixa atividade (0.1-0.3), horário de trabalho (9-17h), resposta lenta (60-240 min), alta influência (2.5-3.0)
- Mídia (Veículo de Mídia): Atividade moderada (0.4-0.6), atividade durante todo o dia (8-23h), resposta rápida (5-30 min), alta influência (2.0-2.5)
- Indivíduos (Estudante/Pessoa/Ex-aluno): Alta atividade (0.6-0.9), principalmente à noite (18-23h), resposta rápida (1-15 min), baixa influência (0.8-1.2)
- Figuras públicas/Especialistas: Atividade moderada (0.4-0.6), influência média-alta (1.5-2.0)
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 = {}
# Construir objetos AgentActivityConfig
configs = []
for i, entity in enumerate(entities):
agent_id = start_idx + i
cfg = llm_configs.get(agent_id, {})
# Usar fallback baseado em regras se o LLM falhar
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
### Configurações de Fallback Baseadas em Regras
Quando o LLM falha, use padrões predefinidos:
```python
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"]:
# Instituição oficial: horário de trabalho, baixa frequência, alta influência
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"]:
# Mídia: atividade durante todo o dia, frequência moderada, alta influência
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"]:
# Especialista/Professor: trabalho + noite, frequência 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"]:
# Estudante: pico noturno, alta frequência
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"]:
# Ex-aluno: foco na noite
return {
"activity_level": 0.6,
"posts_per_hour": 0.4,
"comments_per_hour": 0.8,
"active_hours": [12, 13, 19, 20, 21, 22, 23], # Almoço + noite
"response_delay_min": 5,
"response_delay_max": 30,
"sentiment_bias": 0.0,
"stance": "neutral",
"influence_weight": 1.0
}
else:
# Pessoa padrão: pico noturno
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
Chamadas LLM falham. Saídas são truncadas. JSON quebra. O sistema lida com tudo isso:
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) # Diminuir temperatura na tentativa
)
content = response.choices[0].message.content
finish_reason = response.choices[0].finish_reason
# Verificar se foi truncado
if finish_reason == 'length':
logger.warning(f"Saída LLM truncada (tentativa {attempt+1})")
content = self._fix_truncated_json(content)
# Tentar analisar JSON
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]}")
# Tentar reparar JSON
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()
# Contar chaves não fechadas
open_braces = content.count('{') - content.count('}')
open_brackets = content.count('[') - content.count(']')
# Verificar string não fechada
if content and content[-1] not in '",}]':
content += '"'
# Fechar chaves
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
# Corrigir truncamento
content = self._fix_truncated_json(content)
# Extrair porção JSON
json_match = re.search(r'\{[\s\S]*\}', content)
if json_match:
json_str = json_match.group()
# Remover quebras de linha em strings
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:
# Tentar remover caracteres de controle
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:
"""Configuração de atividade de agente único"""
agent_id: int
entity_uuid: str
entity_name: str
entity_type: str
# Nível de atividade (0.0-1.0)
activity_level: float = 0.5
# Frequência de postagem (por hora)
posts_per_hour: float = 1.0
comments_per_hour: float = 2.0
# Horas ativas (formato 24 horas, 0-23)
active_hours: List[int] = field(default_factory=lambda: list(range(8, 23)))
# Velocidade de resposta (atraso de reação em minutos simulados)
response_delay_min: int = 5
response_delay_max: int = 60
# Tendência de sentimento (-1.0 a 1.0, negativo a positivo)
sentiment_bias: float = 0.0
# Postura sobre tópicos específicos
stance: str = "neutral" # solidário, opositor, neutro, observador
# Peso de influência (afeta a probabilidade de ser visto)
influence_weight: float = 1.0
Configuração de Simulação de Tempo
@dataclass
class TimeSimulationConfig:
"""Configuração de simulação de tempo (fuso horário chinês)"""
total_simulation_hours: int = 72 # Padrão 72 horas (3 dias)
minutes_per_round: int = 60 # 60 minutos por rodada
# Agentes ativados por hora
agents_per_hour_min: int = 5
agents_per_hour_max: int = 20
# Horas de pico (noite 19-22, mais ativo na China)
peak_hours: List[int] = field(default_factory=lambda: [19, 20, 21, 22])
peak_activity_multiplier: float = 1.5
# Horas fora do pico (madrugada 0-5, quase nenhuma atividade)
off_peak_hours: List[int] = field(default_factory=lambda: [0, 1, 2, 3, 4, 5])
off_peak_activity_multiplier: float = 0.05
# Horas da manhã
morning_hours: List[int] = field(default_factory=lambda: [6, 7, 8])
morning_activity_multiplier: float = 0.4
# Horário de trabalho
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:
"""Configuração completa de parâmetros de simulação"""
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
A geração de configuração alimentada por LLM requer um tratamento cuidadoso de:
- Geração passo a passo: Divida em estágios gerenciáveis (tempo → eventos → agentes → plataformas)
- Processamento em lote: Processe 15 agentes por lote para evitar limites de contexto
- Reparo de JSON: Lide com o truncamento com correspondência de colchetes e escape de strings
- Fallbacks baseados em regras: Forneça padrões sensatos quando o LLM falha
- Padrões específicos por tipo: Diferentes tipos de agente têm diferentes padrões de atividade
- Validação e correção: Verifique os valores gerados e corrija problemas (por exemplo, agentes_por_hora > total_agentes)