소개
소셜 미디어 시뮬레이션에서 수백 개의 AI 에이전트 구성을 자동화하려면 시간, 이벤트, 활동 패턴, 응답 지연 등 다양한 속성을 체계적으로 정의해야 합니다. 수동으로 처리하면 시간과 오류가 크게 증가합니다.
MiroFish는 LLM 기반 구성 자동 생성을 통해 이 과정을 단축합니다. 시스템은 시뮬레이션 요구사항, 문서, 지식 그래프를 입력받아 각 에이전트의 세부 속성을 생성합니다.
하지만 LLM의 한계(잘린 출력, JSON 형식 오류, 토큰 제한 등)로 인해 다음과 같은 구현 전략이 필요합니다:
- 단계별 구성 생성 (시간 → 이벤트 → 에이전트 → 플랫폼)
- 배치 처리로 컨텍스트 초과 방지
- JSON 복구 및 오류 처리
- LLM 실패 시 규칙 기반 기본값 적용
- 유형별 에이전트 활동 패턴 적용
- 생성값 유효성 검사 및 자동 수정
💡 구성 생성 파이프라인은 100개 이상의 에이전트에 대해 일련의 API 호출을 수행합니다. Apidog는 각 단계의 요청/응답 스키마를 검증하고, 프로덕션 전 JSON 오류를 포착하며, 잘린 LLM 응답 등 예외 케이스 테스트에 활용합니다.
모든 코드는 MiroFish 실제 사용 사례에서 발췌했습니다.
아키텍처 개요
구성 생성기는 파이프라인 구조로 동작합니다:
┌────────────┐ ┌───────────────┐ ┌───────────────┐
│ Context │ ──► │ Time Config │ ──► │ Event Config │
│ Builder │ │ Generator │ │ Generator │
│ │ │ │ │ │
└────────────┘ └───────────────┘ └───────────────┘
│ │
▼ ▼
┌─────────────────┐ ┌────────────────────┐
│ Agent Config │ ──► │ Platform Config │
│ (Batch) │ │ │
└─────────────────┘ └────────────────────┘
파일 구조
backend/app/services/
├── simulation_config_generator.py # 메인 구성 생성 로직
├── ontology_generator.py # 온톨로지 생성(공통)
└── zep_entity_reader.py # 개체 필터링
backend/app/models/
├── task.py # 작업 추적
└── project.py # 프로젝트 상태
단계별 생성 전략
한 번에 모든 에이전트 구성을 생성하면 LLM 토큰 한계를 초과합니다. 따라서 15개 단위로 배치 처리합니다.
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(...):
num_batches = math.ceil(len(entities) / self.AGENTS_PER_BATCH)
total_steps = 3 + num_batches
current_step = 0
def report_progress(step: int, message: str):
...
# 1. 컨텍스트 구축
context = self._build_context(...)
# 2. 시간 구성 생성
report_progress(1, "시간 구성 생성 중...")
time_config_result = self._generate_time_config(context, len(entities))
time_config = self._parse_time_config(time_config_result, len(entities))
# 3. 이벤트 구성 생성
report_progress(2, "이벤트/핫토픽 구성 생성 중...")
event_config_result = self._generate_event_config(context, simulation_requirement, entities)
event_config = self._parse_event_config(event_config_result)
# 4. 에이전트 배치 구성 생성
all_agent_configs = []
for batch_idx in range(num_batches):
...
batch_configs = self._generate_agent_configs_batch(...)
all_agent_configs.extend(batch_configs)
# 5. 초기 게시물 발행자 할당
event_config = self._assign_initial_post_agents(event_config, all_agent_configs)
# 6. 플랫폼 구성
report_progress(total_steps, "플랫폼 구성 생성 중...")
twitter_config = PlatformConfig(platform="twitter", ...) if enable_twitter else None
reddit_config = PlatformConfig(platform="reddit", ...) if enable_reddit else None
# 7. 최종 구성 반환
params = SimulationParameters(...)
return params
주요 효과
- 각 LLM 호출의 컨텍스트를 제한하여 실패율 감소
- 사용자 진행률 표시 지원
- 부분 실패 시 재시도 및 복구 가능
컨텍스트 구축
토큰 한도를 초과하지 않으면서 시뮬레이션 요구, 개체 요약, 문서 일부를 포함합니다.
def _build_context(self, simulation_requirement, document_text, entities) -> str:
entity_summary = self._summarize_entities(entities)
context_parts = [
f"## Simulation Requirement\n{simulation_requirement}",
f"\n## Entity Information ({len(entities)} entities)\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...(document truncated)"
context_parts.append(f"\n## Original Document\n{doc_text}")
return "\n".join(context_parts)
개체 요약
개체는 유형별로 최대 20개만 노출하고 내용을 요약합니다.
def _summarize_entities(self, entities):
lines = []
by_type = {}
for e in entities:
t = e.get_entity_type() or "Unknown"
by_type.setdefault(t, []).append(e)
for entity_type, type_entities in by_type.items():
lines.append(f"\n### {entity_type} ({len(type_entities)} entities)")
for e in type_entities[:self.ENTITIES_PER_TYPE_DISPLAY]:
summary_preview = (e.summary[:self.ENTITY_SUMMARY_LENGTH] + "...") if len(e.summary) > self.ENTITY_SUMMARY_LENGTH else e.summary
lines.append(f"- {e.name}: {summary_preview}")
if len(type_entities) > self.ENTITIES_PER_TYPE_DISPLAY:
lines.append(f" ... and {len(type_entities) - self.ENTITIES_PER_TYPE_DISPLAY} more")
return "\n".join(lines)
출력 예시:
### Student (45 entities)
- Zhang Wei: Active in student union, frequently posts about campus events...
... and 43 more
### University (3 entities)
- Wuhan University: Official account, posts announcements and news...
시간 구성 생성
중국 시간대 기반의 활동 패턴을 자동 생성합니다.
def _generate_time_config(self, context, num_entities):
context_truncated = context[:self.TIME_CONFIG_CONTEXT_LENGTH]
max_agents_allowed = max(1, int(num_entities * 0.9))
prompt = f"""... (중략) ..."""
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, num_entities):
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
)
기본 시간 구성
def _get_default_time_config(self, num_entities):
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": "Using default Chinese timezone configuration"
}
이벤트 구성 생성
이벤트 구성은 핫토픽, 내러티브 방향, 초기 게시물(게시자 유형 포함)을 정의합니다.
def _generate_event_config(self, context, simulation_requirement, entities):
# (중략) entity_types_available, type_examples 생성
prompt = f"""... (중략) ..."""
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, agent_configs):
if not event_config.initial_posts:
return event_config
agents_by_type = {}
for agent in agent_configs:
etype = agent.entity_type.lower()
agents_by_type.setdefault(etype, []).append(agent)
type_aliases = {
"official": ["official", "university", ...],
...
}
used_indices = {}
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. 직접 매칭
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 매칭
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: 영향력 높은 agent
if matched_agent_id is None:
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
})
event_config.initial_posts = updated_posts
return event_config
배치 에이전트 구성 생성
15개 단위로 에이전트 활동 패턴을 생성합니다. 각 유형별로 다른 활동/응답/영향력 패턴을 적용합니다.
def _generate_agent_configs_batch(self, context, entities, start_idx, simulation_requirement):
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"""... (중략) ..."""
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 = {}
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
규칙 기반 대체 구성
LLM 실패 시, 유형별 기본값을 자동 적용합니다.
def _generate_agent_config_by_rule(self, entity):
entity_type = (entity.get_entity_type() or "Unknown").lower()
if entity_type in ["university", "governmentagency", "ngo"]:
return {...}
elif entity_type in ["mediaoutlet"]:
return {...}
elif entity_type in ["professor", "expert", "official"]:
return {...}
elif entity_type in ["student"]:
return {...}
elif entity_type in ["alumni"]:
return {...}
else:
return {...}
LLM 호출: 재시도 및 JSON 복구
LLM 출력이 잘리거나 JSON 형식이 깨질 경우, 자동으로 재시도 및 복구를 수행합니다.
def _call_llm_with_retry(self, prompt, system_prompt):
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':
content = self._fix_truncated_json(content)
try:
return json.loads(content)
except json.JSONDecodeError:
fixed = self._try_fix_config_json(content)
if fixed:
return fixed
last_error = e
except Exception as e:
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):
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
def _try_fix_config_json(self, content):
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
구성 데이터 구조
에이전트 활동 구성
@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
시간 시뮬레이션 구성
@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
완전한 시뮬레이션 매개변수
@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,
}
요약표: 에이전트 유형별 패턴
| 에이전트 유형 | 활동 | 활동 시간 | 시간당 게시물 | 시간당 댓글 | 응답(분) | 영향력 |
|---|---|---|---|---|---|---|
| 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 실패 시 기본값으로 fallback
- 유형별 패턴: 에이전트 유형에 따라 활동/응답/영향력 자동 분기
- 유효성 검사 및 수정: 생성값 자동 검증 및 보정
Top comments (0)