소개
소셜 미디어 시뮬레이션을 위해 수백 개의 AI 에이전트를 구성하는 것은 어려운 작업처럼 들립니다. 각 에이전트에는 활동 일정, 게시 빈도, 응답 지연, 영향력 가중치 및 입장 위치가 필요합니다. 이 작업을 수동으로 수행하려면 몇 시간이 걸릴 것입니다.
MiroFish는 LLM 기반 구성 생성을 통해 이 과정을 자동화합니다. 시스템은 문서, 지식 그래프 및 시뮬레이션 요구 사항을 분석한 다음, 모든 에이전트에 대한 상세 구성을 생성합니다.
문제는 LLM이 실패할 수 있다는 것입니다. 출력이 잘리거나, JSON 형식이 깨지거나, 토큰 제한에 걸릴 수 있습니다.
이 가이드는 완전한 구현에 대해 다룹니다:
- 단계별 생성 (시간 → 이벤트 → 에이전트 → 플랫폼)
- 컨텍스트 제한을 피하기 위한 배치 처리
- 잘린 출력에 대한 JSON 복구 전략
- LLM 실패 시 규칙 기반 대체 구성
- 유형별 에이전트 활동 패턴 (학생 vs 공식 vs 미디어)
- 유효성 검사 및 수정 로직
💡
구성 생성 파이프라인은 일련의 API 호출을 통해 100개 이상의 에이전트를 처리합니다. Apidog는 각 단계에서 요청/응답 스키마를 검증하고, 프로덕션에 도달하기 전에 JSON 형식 오류를 포착하며, 잘린 LLM 출력과 같은 예외 상황에 대한 테스트 케이스를 생성하는 데 사용되었습니다.
모든 코드는 MiroFish의 실제 프로덕션 사용 사례에서 가져왔습니다.
아키텍처 개요
구성 생성기는 파이프라인 접근 방식을 사용합니다:
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ 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 │
└─────────────────┘ └─────────────────┘ └─────────────────┘
파일 구조
backend/app/services/
├── simulation_config_generator.py # Main config generation logic
├── ontology_generator.py # Ontology generation (shared)
└── zep_entity_reader.py # Entity filtering
backend/app/models/
├── task.py # Task tracking
└── project.py # Project state
단계별 생성 전략
모든 구성을 한 번에 생성하면 토큰 제한을 초과합니다. 대신 시스템은 단계별로 생성합니다:
class SimulationConfigGenerator:
# Each batch generates configs for 15 agents
AGENTS_PER_BATCH = 15
# Context limits
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:
# Calculate total steps
num_batches = math.ceil(len(entities) / self.AGENTS_PER_BATCH)
total_steps = 3 + num_batches # Time + Events + N Agent Batches + Platform
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}")
# Build context
context = self._build_context(
simulation_requirement=simulation_requirement,
document_text=document_text,
entities=entities
)
reasoning_parts = []
# Step 1: Generate time config
report_progress(1, "Generating time configuration...")
time_config_result = self._generate_time_config(context, len(entities))
time_config = self._parse_time_config(time_config_result, len(entities))
reasoning_parts.append(f"Time config: {time_config_result.get('reasoning', 'Success')}")
# Step 2: Generate event config
report_progress(2, "Generating event config and hot topics...")
event_config_result = self._generate_event_config(context, simulation_requirement, entities)
event_config = self._parse_event_config(event_config_result)
reasoning_parts.append(f"Event config: {event_config_result.get('reasoning', 'Success')}")
# Steps 3-N: Generate agent configs in batches
all_agent_configs = []
for batch_idx in range(num_batches):
start_idx = batch_idx * self.AGENTS_PER_BATCH
end_idx = min(start_idx + self.AGENTS_PER_BATCH, len(entities))
batch_entities = entities[start_idx:end_idx]
report_progress(
3 + batch_idx,
f"Generating agent config ({start_idx + 1}-{end_idx}/{len(entities)})..."
)
batch_configs = self._generate_agent_configs_batch(
context=context,
entities=batch_entities,
start_idx=start_idx,
simulation_requirement=simulation_requirement
)
all_agent_configs.extend(batch_configs)
reasoning_parts.append(f"Agent config: Generated {len(all_agent_configs)} agents")
# Assign initial post publishers
event_config = self._assign_initial_post_agents(event_config, all_agent_configs)
# Final step: Platform config
report_progress(total_steps, "Generating platform configuration...")
twitter_config = PlatformConfig(platform="twitter", ...) if enable_twitter else None
reddit_config = PlatformConfig(platform="reddit", ...) if enable_reddit else None
# Assemble final config
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
이러한 단계별 접근 방식은 다음과 같습니다:
- 각 LLM 호출을 집중적이고 관리하기 쉽게 유지합니다.
- 사용자에게 진행 상황 업데이트를 제공합니다.
- 한 단계가 실패하더라도 부분 복구를 허용합니다.
컨텍스트 구축
컨텍스트 빌더는 토큰 제한을 준수하면서 관련 정보를 취합합니다:
def _build_context(
self,
simulation_requirement: str,
document_text: str,
entities: List[EntityNode]
) -> str:
# Entity summary
entity_summary = self._summarize_entities(entities)
context_parts = [
f"## Simulation Requirement\n{simulation_requirement}",
f"\n## Entity Information ({len(entities)} entities)\n{entity_summary}",
]
# Add document text if space allows
current_length = sum(len(p) for p in context_parts)
remaining_length = self.MAX_CONTEXT_LENGTH - current_length - 500 # 500 char buffer
if remaining_length > 0 and document_text:
doc_text = document_text[:remaining_length]
if len(document_text) > remaining_length:
doc_text += "\n...(document truncated)"
context_parts.append(f"\n## Original Document\n{doc_text}")
return "\n".join(context_parts)
개체 요약
개체는 유형별로 요약됩니다:
def _summarize_entities(self, entities: List[EntityNode]) -> str:
lines = []
# Group by type
by_type: Dict[str, List[EntityNode]] = {}
for e in entities:
t = e.get_entity_type() or "Unknown"
if t not in by_type:
by_type[t] = []
by_type[t].append(e)
for entity_type, type_entities in by_type.items():
lines.append(f"\n### {entity_type} ({len(type_entities)} entities)")
# Display limited number with limited summary length
display_count = self.ENTITIES_PER_TYPE_DISPLAY
summary_len = self.ENTITY_SUMMARY_LENGTH
for e in type_entities[:display_count]:
summary_preview = (e.summary[:summary_len] + "...") if len(e.summary) > summary_len else e.summary
lines.append(f"- {e.name}: {summary_preview}")
if len(type_entities) > display_count:
lines.append(f" ... and {len(type_entities) - display_count} more")
return "\n".join(lines)
이는 다음과 같은 출력을 생성합니다:
### Student (45 entities)
- Zhang Wei: Active in student union, frequently posts about campus events and academic pressure...
- Li Ming: Graduate student researching AI ethics, often shares technology news...
... and 43 more
### University (3 entities)
- Wuhan University: Official account, posts announcements and news...
시간 구성 생성
시간 구성은 시뮬레이션 기간과 활동 패턴을 결정합니다:
def _generate_time_config(self, context: str, num_entities: int) -> Dict[str, Any]:
# Truncate context for this specific step
context_truncated = context[:self.TIME_CONFIG_CONTEXT_LENGTH]
# Calculate max allowed value (90% of agent count)
max_agents_allowed = max(1, int(num_entities * 0.9))
prompt = f"""Based on the following simulation requirements, generate time configuration.
{context_truncated}
## Task
Generate time configuration JSON.
### Basic Principles (adjust based on event type and participant groups):
- User base is Chinese, must follow Beijing timezone habits
- 0-5 AM: Almost no activity (coefficient 0.05)
- 6-8 AM: Gradually waking up (coefficient 0.4)
- 9-18 PM: Work hours, moderate activity (coefficient 0.7)
- 19-22 PM: Evening peak, most active (coefficient 1.5)
- 23 PM: Activity declining (coefficient 0.5)
### Return JSON format (no markdown):
Example:
{{
"total_simulation_hours": 72,
"minutes_per_round": 60,
"agents_per_hour_min": 5,
"agents_per_hour_max": 50,
"peak_hours": [19, 20, 21, 22],
"off_peak_hours": [0, 1, 2, 3, 4, 5],
"morning_hours": [6, 7, 8],
"work_hours": [9, 10, 11, 12, 13, 14, 15, 16, 17, 18],
"reasoning": "Time configuration explanation"
}}
Field descriptions:
- total_simulation_hours (int): 24-168 hours, shorter for breaking news, longer for ongoing topics
- minutes_per_round (int): 30-120 minutes, recommend 60
- agents_per_hour_min (int): Range 1-{max_agents_allowed}
- agents_per_hour_max (int): Range 1-{max_agents_allowed}
- peak_hours (int array): Adjust based on participant groups
- off_peak_hours (int array): Usually late night/early morning
- morning_hours (int array): Morning hours
- work_hours (int array): Work hours
- reasoning (string): Brief explanation"""
system_prompt = "You are a social media simulation expert. Return pure JSON format."
try:
return self._call_llm_with_retry(prompt, system_prompt)
except Exception as e:
logger.warning(f"Time config LLM generation failed: {e}, using default")
return self._get_default_time_config(num_entities)
시간 구성 파싱 및 유효성 검사
def _parse_time_config(self, result: Dict[str, Any], num_entities: int) -> TimeSimulationConfig:
# Get raw values
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))
# Validate and correct: ensure not exceeding total agent count
if agents_per_hour_min > num_entities:
logger.warning(f"agents_per_hour_min ({agents_per_hour_min}) exceeds total agents ({num_entities}), corrected")
agents_per_hour_min = max(1, num_entities // 10)
if agents_per_hour_max > num_entities:
logger.warning(f"agents_per_hour_max ({agents_per_hour_max}) exceeds total agents ({num_entities}), corrected")
agents_per_hour_max = max(agents_per_hour_min + 1, num_entities // 2)
# Ensure 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, corrected to {agents_per_hour_min}")
return TimeSimulationConfig(
total_simulation_hours=result.get("total_simulation_hours", 72),
minutes_per_round=result.get("minutes_per_round", 60),
agents_per_hour_min=agents_per_hour_min,
agents_per_hour_max=agents_per_hour_max,
peak_hours=result.get("peak_hours", [19, 20, 21, 22]),
off_peak_hours=result.get("off_peak_hours", [0, 1, 2, 3, 4, 5]),
off_peak_activity_multiplier=0.05,
morning_activity_multiplier=0.4,
work_activity_multiplier=0.7,
peak_activity_multiplier=1.5
)
기본 시간 구성 (중국 시간대)
def _get_default_time_config(self, num_entities: int) -> Dict[str, Any]:
return {
"total_simulation_hours": 72,
"minutes_per_round": 60, # 1 hour per round
"agents_per_hour_min": max(1, num_entities // 15),
"agents_per_hour_max": max(5, num_entities // 5),
"peak_hours": [19, 20, 21, 22],
"off_peak_hours": [0, 1, 2, 3, 4, 5],
"morning_hours": [6, 7, 8],
"work_hours": [9, 10, 11, 12, 13, 14, 15, 16, 17, 18],
"reasoning": "Using default Chinese timezone configuration"
}
이벤트 구성 생성
이벤트 구성은 초기 게시물, 인기 주제 및 내러티브 방향을 정의합니다:
def _generate_event_config(
self,
context: str,
simulation_requirement: str,
entities: List[EntityNode]
) -> Dict[str, Any]:
# Get available entity types for LLM reference
entity_types_available = list(set(
e.get_entity_type() or "Unknown" for e in entities
))
# Show examples per type
type_examples = {}
for e in entities:
etype = e.get_entity_type() or "Unknown"
if etype not in type_examples:
type_examples[etype] = []
if len(type_examples[etype]) < 3:
type_examples[etype].append(e.name)
type_info = "\n".join([
f"- {t}: {', '.join(examples)}"
for t, examples in type_examples.items()
])
context_truncated = context[:self.EVENT_CONFIG_CONTEXT_LENGTH]
prompt = f"""Based on the following simulation requirements, generate event configuration.
Simulation Requirement: {simulation_requirement}
{context_truncated}
## Available Entity Types and Examples
{type_info}
## Task
Generate event configuration JSON:
- Extract hot topic keywords
- Describe narrative direction
- Design initial posts, **each post must specify poster_type**
**Important**: poster_type must be selected from "Available Entity Types" above, so initial posts can be assigned to appropriate agents.
For example: Official statements should be posted by Official/University types, news by MediaOutlet, student opinions by Student.
Return JSON format (no markdown):
{{
"hot_topics": ["keyword1", "keyword2", ...],
"narrative_direction": "<narrative direction description>",
"initial_posts": [
{{"content": "Post content", "poster_type": "Entity Type (must match available types)"}},
...
],
"reasoning": "<brief explanation>"
}}"""
system_prompt = "You are an opinion analysis expert. Return pure JSON format."
try:
return self._call_llm_with_retry(prompt, system_prompt)
except Exception as e:
logger.warning(f"Event config LLM generation failed: {e}, using default")
return {
"hot_topics": [],
"narrative_direction": "",
"initial_posts": [],
"reasoning": "Using default configuration"
}
초기 게시물 발행자 할당
초기 게시물을 생성한 후, 실제 에이전트에 게시물을 매칭합니다:
def _assign_initial_post_agents(
self,
event_config: EventConfig,
agent_configs: List[AgentActivityConfig]
) -> EventConfig:
if not event_config.initial_posts:
return event_config
# Index agents by type
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 alias mapping (handles LLM variations)
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"],
}
# Track used indices to avoid reusing same agent
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. Direct match
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. Alias match
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: use highest influence agent
if matched_agent_id is None:
logger.warning(f"No matching agent for type '{poster_type}', using highest influence agent")
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"Initial post assignment: poster_type='{poster_type}' -> agent_id={matched_agent_id}")
event_config.initial_posts = updated_posts
return event_config
배치 에이전트 구성 생성
수백 개의 에이전트에 대한 구성을 한 번에 생성하면 토큰 제한을 초과합니다. 시스템은 15개 단위로 배치 처리합니다:
def _generate_agent_configs_batch(
self,
context: str,
entities: List[EntityNode],
start_idx: int,
simulation_requirement: str
) -> List[AgentActivityConfig]:
# Build entity info with limited summary length
entity_list = []
summary_len = self.AGENT_SUMMARY_LENGTH
for i, e in enumerate(entities):
entity_list.append({
"agent_id": start_idx + i,
"entity_name": e.name,
"entity_type": e.get_entity_type() or "Unknown",
"summary": e.summary[:summary_len] if e.summary else ""
})
prompt = f"""Based on the following information, generate social media activity configuration for each entity.
Simulation Requirement: {simulation_requirement}
## Entity List
```json
{json.dumps(entity_list, ensure_ascii=False, indent=2)}
작업
각 개체에 대한 활동 구성을 생성합니다. 참고:
- 시간은 중국 습관을 따라야 합니다: 오전 0-5시에는 거의 활동이 없으며, 오후 19-22시가 가장 활동적입니다.
- 공식 기관 (대학/정부 기관): 낮은 활동 수준 (0.1-0.3), 근무 시간 (9-17시), 느린 응답 (60-240분), 높은 영향력 (2.5-3.0)
- 미디어 (언론 매체): 중간 활동 수준 (0.4-0.6), 하루 종일 활동 (8-23시), 빠른 응답 (5-30분), 높은 영향력 (2.0-2.5)
- 개인 (학생/개인/동문): 높은 활동 수준 (0.6-0.9), 주로 저녁 (18-23시), 빠른 응답 (1-15분), 낮은 영향력 (0.8-1.2)
- 공인/전문가: 중간 활동 수준 (0.4-0.6), 중간-높은 영향력 (1.5-2.0)
system_prompt = "You are a social media behavior analysis expert. Return pure JSON format."
try:
result = self._call_llm_with_retry(prompt, system_prompt)
llm_configs = {cfg["agent_id"]: cfg for cfg in result.get("agent_configs", [])}
except Exception as e:
logger.warning(f"Agent config batch LLM generation failed: {e}, using rule-based generation")
llm_configs = {}
# Build AgentActivityConfig objects
configs = []
for i, entity in enumerate(entities):
agent_id = start_idx + i
cfg = llm_configs.get(agent_id, {})
# Use rule-based fallback if LLM failed
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
규칙 기반 대체 구성
LLM이 실패할 경우, 미리 정의된 패턴을 사용합니다:
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"]:
# Official institution: work hours, low frequency, high influence
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"]:
# Media: all-day activity, moderate frequency, high influence
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"]:
# Expert/Professor: work + evening, moderate frequency
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"]:
# Student: evening peak, high frequency
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"]:
# Alumni: evening focused
return {
"activity_level": 0.6,
"posts_per_hour": 0.4,
"comments_per_hour": 0.8,
"active_hours": [12, 13, 19, 20, 21, 22, 23], # Lunch + evening
"response_delay_min": 5,
"response_delay_max": 30,
"sentiment_bias": 0.0,
"stance": "neutral",
"influence_weight": 1.0
}
else:
# Default person: evening peak
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
}
재시도 및 JSON 복구 기능이 있는 LLM 호출
LLM 호출은 실패할 수 있습니다. 출력이 잘리거나, JSON 형식이 깨질 수 있습니다. 시스템은 이 모든 것을 처리합니다:
def _call_llm_with_retry(self, prompt: str, system_prompt: str) -> Dict[str, Any]:
import re
max_attempts = 3
last_error = None
for attempt in range(max_attempts):
try:
response = self.client.chat.completions.create(
model=self.model_name,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt}
],
response_format={"type": "json_object"},
temperature=0.7 - (attempt * 0.1) # Lower temp on retry
)
content = response.choices[0].message.content
finish_reason = response.choices[0].finish_reason
# Check if truncated
if finish_reason == 'length':
logger.warning(f"LLM output truncated (attempt {attempt+1})")
content = self._fix_truncated_json(content)
# Try parsing JSON
try:
return json.loads(content)
except json.JSONDecodeError as e:
logger.warning(f"JSON parse failed (attempt {attempt+1}): {str(e)[:80]}")
# Try repairing JSON
fixed = self._try_fix_config_json(content)
if fixed:
return fixed
last_error = e
except Exception as e:
logger.warning(f"LLM call failed (attempt {attempt+1}): {str(e)[:80]}")
last_error = e
import time
time.sleep(2 * (attempt + 1))
raise last_error or Exception("LLM call failed")
잘린 JSON 수정
def _fix_truncated_json(self, content: str) -> str:
content = content.strip()
# Count unclosed brackets
open_braces = content.count('{') - content.count('}')
open_brackets = content.count('[') - content.count(']')
# Check for unclosed string
if content and content[-1] not in '",}]':
content += '"'
# Close brackets
content += ']' * open_brackets
content += '}' * open_braces
return content
고급 JSON 복구
def _try_fix_config_json(self, content: str) -> Optional[Dict[str, Any]]:
import re
# Fix truncation
content = self._fix_truncated_json(content)
# Extract JSON portion
json_match = re.search(r'\{[\s\S]*\}', content)
if json_match:
json_str = json_match.group()
# Remove newlines in 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:
# Try removing control characters
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
구성 데이터 구조
에이전트 활동 구성
@dataclass
class AgentActivityConfig:
"""Single agent activity configuration"""
agent_id: int
entity_uuid: str
entity_name: str
entity_type: str
# Activity level (0.0-1.0)
activity_level: float = 0.5
# Posting frequency (per hour)
posts_per_hour: float = 1.0
comments_per_hour: float = 2.0
# Active hours (24-hour format, 0-23)
active_hours: List[int] = field(default_factory=lambda: list(range(8, 23)))
# Response speed (reaction delay in simulated minutes)
response_delay_min: int = 5
response_delay_max: int = 60
# Sentiment tendency (-1.0 to 1.0, negative to positive)
sentiment_bias: float = 0.0
# Stance on specific topics
stance: str = "neutral" # supportive, opposing, neutral, observer
# Influence weight (affects probability of being seen)
influence_weight: float = 1.0
시간 시뮬레이션 구성
@dataclass
class TimeSimulationConfig:
"""Time simulation configuration (Chinese timezone)"""
total_simulation_hours: int = 72 # Default 72 hours (3 days)
minutes_per_round: int = 60 # 60 minutes per round
# Agents activated per hour
agents_per_hour_min: int = 5
agents_per_hour_max: int = 20
# Peak hours (evening 19-22, Chinese most active)
peak_hours: List[int] = field(default_factory=lambda: [19, 20, 21, 22])
peak_activity_multiplier: float = 1.5
# Off-peak hours (early morning 0-5, almost no activity)
off_peak_hours: List[int] = field(default_factory=lambda: [0, 1, 2, 3, 4, 5])
off_peak_activity_multiplier: float = 0.05
# Morning hours
morning_hours: List[int] = field(default_factory=lambda: [6, 7, 8])
morning_activity_multiplier: float = 0.4
# Work hours
work_hours: List[int] = field(default_factory=lambda: [9, 10, 11, 12, 13, 14, 15, 16, 17, 18])
work_activity_multiplier: float = 0.7
완전한 시뮬레이션 매개변수
@dataclass
class SimulationParameters:
"""Complete simulation parameter configuration"""
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,
}
요약표: 에이전트 유형 패턴
| 에이전트 유형 | 활동 | 활동 시간 | 시간당 게시물 | 시간당 댓글 | 응답 (분) | 영향력 |
|---|---|---|---|---|---|---|
| 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 |
결론
LLM 기반 구성 생성은 다음을 신중하게 처리해야 합니다:
- 단계별 생성: 관리 가능한 단계로 분할 (시간 → 이벤트 → 에이전트 → 플랫폼)
- 배치 처리: 컨텍스트 제한을 피하기 위해 배치당 15개의 에이전트 처리
- JSON 복구: 괄호 매칭 및 문자열 이스케이프를 통해 잘림 처리
- 규칙 기반 대체: LLM 실패 시 합리적인 기본값 제공
- 유형별 패턴: 에이전트 유형마다 다른 활동 패턴 가짐
- 유효성 검사 및 수정: 생성된 값 확인 및 문제 해결 (예: 시간당 에이전트 수 > 총 에이전트 수)