소개
소셜 미디어는 빠르게 움직입니다. 단 하나의 게시물이 누구도 예측하지 못한 반응, 재구성, 반대 운동의 연쇄를 촉발할 수 있습니다. 현실에서 시나리오가 전개되기 전에 어떻게 진행될지 미리 볼 수 있다면 어떨까요?
MiroFish는 바로 그런 일을 합니다. MiroFish는 독자적인 성격, 기억, 행동 패턴을 가진 수천 개의 AI 에이전트가 자유롭게 상호 작용하는 디지털 병렬 세계를 생성하는 스웜 인텔리전스 엔진입니다. 뉴스 기사, 정책 초안, 심지어 소설과 같은 시드 자료를 업로드하면 MiroFish가 이벤트가 어떻게 전개될지 고성능 시뮬레이션을 구축합니다.
💡 팁:
MiroFish를 구축하려면 안정적인 API 테스트 기반이 필요했습니다. 팀은 시뮬레이션 로직을 작성하기 전에 모든 백엔드 API를 설계, 디버그 및 문서화하기 위해 Apidog를 사용했습니다. 이를 통해 초기 단계에서 엔드포인트 문제를 파악하고 개발 전반에 걸쳐 Python 백엔드와 Vue 프런트엔드를 동기화할 수 있었습니다.
이 게시물에서는 MiroFish의 기술 아키텍처를 구체적으로 다룹니다. 시스템이 원본 문서를 살아있는 시뮬레이션으로 변환하는 방법, 에이전트가 결정을 내리는 방법, 그리고 5단계 워크플로가 지식 그래프 구축부터 실시간 모니터링까지 모든 것을 어떻게 자동화하는지 살펴봅니다.
시스템 개요: 5단계 워크플로
MiroFish는 다음의 5단계로 시뮬레이션을 처리합니다.
┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Step 1 │ ──► │ Step 2 │ ──► │ Step 3 │ ──► │ Step 4 │ ──► │ Step 5 │
│ Ontology │ │ GraphRAG │ │ Env │ │ Simulation │ │ Report │
│ Generation │ │ Build │ │ Setup │ │ Run │ │ Generation │
└─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘
1단계: 온톨로지 생성
- 입력 문서와 요구 사항을 분석 후 LLM을 활용해 온톨로지 자동 생성
- 엔티티 유형 10개(예: 학생, 교수, 대학, 미디어, 정부)
- 관계 유형 10개(예: 재직, 댓글, 응답 등)
- 각 유형별 속성(예약어 제외)
온톨로지 생성기는 8가지 특정 유형 + 2가지 대체 유형(사람, 조직) 구조를 적용해 포괄성과 API 제한을 동시에 만족시킵니다.
2단계: GraphRAG 구축
- 문서를 500자(50자 중복) 단위로 청크 분할
- Zep Cloud API를 통해 일괄 전송
- 고유 ID 기반 그래프 생성
- 온톨로지 적용 및 엔티티/관계 추출
- Zep가 처리 완료 후, 최종 노드/엣지 그래프 반환
3단계: 환경 설정
- 지식 그래프를 분석해 상세 에이전트 매개변수 생성
- 시간 구성: 중국 시간대 패턴, 피크/비활동 시간
- 이벤트 구성: 초기 게시물, 인기 주제
- 에이전트 활동 구성: 시간당 게시물 수, 응답 지연, 영향력
- 플랫폼 구성: Twitter/Reddit별 바이럴 임계값 등
4단계: 시뮬레이션 실행
- 에이전트 활동 일정 기반 활성화
- 게시, 댓글, 반응 등 이벤트 실행
- Twitter/Reddit 두 플랫폼에서 병렬 시뮬레이션
- 모든 작업을 실시간 JSONL 파일로 로깅
5단계: 보고서 생성
- InsightForge: 복잡한 질의를 쿼리로 분해
- PanoramaSearch: 전체 이력 및 만료 데이터 검색
- InterviewAgents: IPC로 활성 에이전트 실시간 인터뷰
기술 심층 분석: 온톨로지 생성
온톨로지는 backend/app/services/ontology_generator.py에서 생성됩니다.
시스템 프롬프트에 유효/무효 엔티티 유형 구분 규칙을 명확히 정의하여, 실제 소셜 미디어 상에서 말하고 행동할 수 있는 에이전트만 생성하도록 설계합니다.
온톨로지 생성 후 _validate_and_process 메서드로 최종 제약 조건을 적용합니다:
def _validate_and_process(self, result: Dict[str, Any]) -> Dict[str, Any]:
# Zep API limits: max 10 entity types, max 10 edge types
MAX_ENTITY_TYPES = 10
MAX_EDGE_TYPES = 10
# Ensure fallback types exist
fallbacks_to_add = []
if "Person" not in entity_names:
fallbacks_to_add.append(person_fallback)
if "Organization" not in entity_names:
fallbacks_to_add.append(organization_fallback)
# Trim if adding fallbacks would exceed limit
if current_count + needed_slots > MAX_ENTITY_TYPES:
result["entity_types"] = result["entity_types"][:-to_remove]
result["entity_types"].extend(fallbacks_to_add)
return result
이 계층은 항상 Zep API 제한(10개) 내에서 필수 엔티티 타입을 보장합니다.
지식 그래프 구축: Zep 통합
그래프 빌더(backend/app/services/graph_builder.py)는 비동기 워크플로로 그래프를 구축합니다.
def _build_graph_worker(self, task_id: str, text: str, ontology: Dict, ...):
# 1. Create graph
graph_id = self.create_graph(graph_name)
# 2. Set ontology
self.set_ontology(graph_id, ontology)
# 3. Chunk text
chunks = TextProcessor.split_text(text, chunk_size, chunk_overlap)
# 4. Send batches
episode_uuids = self.add_text_batches(graph_id, chunks, batch_size)
# 5. Wait for Zep processing
self._wait_for_episodes(episode_uuids, progress_callback)
# 6. Retrieve final graph
graph_info = self._get_graph_info(graph_id)
동적 Pydantic 모델 생성
각 엔티티 유형별로 런타임에 Pydantic 모델이 동적으로 생성됩니다.
def set_ontology(self, graph_id: str, ontology: Dict[str, Any]):
RESERVED_NAMES = {'uuid', 'name', 'group_id', 'name_embedding', 'summary', 'created_at'}
def safe_attr_name(attr_name: str) -> str:
if attr_name.lower() in RESERVED_NAMES:
return f"entity_{attr_name}"
return attr_name
entity_types = {}
for entity_def in ontology.get("entity_types", []):
name = entity_def["name"]
attrs = {"__doc__": description}
annotations = {}
for attr_def in entity_def.get("attributes", []):
attr_name = safe_attr_name(attr_def["name"])
attrs[attr_name] = Field(description=attr_desc, default=None)
annotations[attr_name] = Optional[EntityText]
attrs["__annotations__"] = annotations
entity_class = type(name, (EntityModel,), attrs)
entity_types[name] = entity_class
사전 정의 없이도 동적 스키마 검증이 가능합니다.
대규모 그래프 페이지네이션
Zep이 반환하는 페이지 결과를 전부 가져오려면 다음과 같이 처리합니다.
def fetch_all_nodes(client: Zep, graph_id: str) -> List[Node]:
nodes = []
cursor = None
while True:
result = client.graph.get_nodes(graph_id=graph_id, cursor=cursor, limit=100)
nodes.extend(result.nodes)
if not result.next_cursor:
break
cursor = result.next_cursor
return nodes
시간 기반 에이전트 활동 시뮬레이션
backend/app/services/simulation_config_generator.py에서 시간대별 현실적인 활동 패턴을 정의합니다.
CHINA_TIMEZONE_CONFIG = {
"dead_hours": [0, 1, 2, 3, 4, 5],
"morning_hours": [6, 7, 8],
"work_hours": [9, 10, 11, 12, 13, 14, 15, 16, 17, 18],
"peak_hours": [19, 20, 21, 22],
"night_hours": [23],
"activity_multipliers": {
"dead": 0.05,
"morning": 0.4,
"work": 0.7,
"peak": 1.5,
"night": 0.5
}
}
각 에이전트 유형별로 활동 수준, 활동 시간, 응답 지연, 영향력 등 세부 파라미터를 설정합니다.
| 에이전트 유형 | 활동 수준 | 활동 시간 | 응답 지연 | 영향력 |
|---|---|---|---|---|
| 대학 | 0.2 | 9-17 | 60-240분 | 3.0 |
| 미디어 매체 | 0.5 | 7-23 | 5-30분 | 2.5 |
| 학생 | 0.8 | 8-12, 18-23 | 1-15분 | 0.8 |
| 교수 | 0.4 | 8-21 | 15-90분 | 2.0 |
LLM 호출이 실패하면 규칙 기반 기본값으로 자동 대체됩니다.
실시간 활동 추적
backend/app/services/simulation_runner.py는 에이전트 활동을 실시간으로 스트리밍하여 추적합니다.
def _read_action_log(self, log_path: str, position: int, state: SimulationRunState, platform: str):
with open(log_path, 'r', encoding='utf-8') as f:
f.seek(position)
for line in f:
action_data = json.loads(line)
# Handle events
if "event_type" in action_data:
if action_data["event_type"] == "simulation_end":
state.twitter_completed = True # or reddit
elif action_data["event_type"] == "round_end":
state.current_round = action_data["round"]
continue
# Parse agent actions
action = AgentAction(
round_num=action_data.get("round", 0),
platform=platform,
agent_id=action_data.get("agent_id", 0),
action_type=action_data.get("action_type", ""),
...
)
state.add_action(action)
return f.tell()
이 함수는 백그라운드 스레드에서 2초 간격으로 상태를 업데이트하며, 프런트엔드는 폴링을 통해 실시간 진행 상황을 표시합니다.
크로스 플랫폼 프로세스 관리
Windows/Unix 환경 모두에서 시뮬레이션 프로세스 종료를 안전하게 처리합니다.
def _terminate_process(cls, process: subprocess.Popen, simulation_id: str, timeout: int = 10):
if IS_WINDOWS:
# Windows: use taskkill to kill process tree
subprocess.run(['taskkill', '/PID', str(process.pid), '/T'], ...)
else:
# Unix: kill process group (created with start_new_session=True)
os.killpg(os.getpgid(process.pid), signal.SIGTERM)
또한 종료 신호(SIGINT, SIGTERM, SIGHUP)에 대한 정리 핸들러를 등록하여 서버 종료 시 정상적으로 시뮬레이션을 중지합니다.
def register_cleanup(cls):
def cleanup_handler(signum, frame):
cls.cleanup_all_simulations()
# Then call original handler
signal.signal(signal.SIGTERM, cleanup_handler)
signal.signal(signal.SIGINT, cleanup_handler)
if has_sighup:
signal.signal(signal.SIGHUP, cleanup_handler)
atexit.register(cls.cleanup_all_simulations)
보고서 생성: 3단계 검색
Zep 도구 서비스(backend/app/services/zep_tools.py)는 세 가지 자동화된 검색 기능을 제공합니다.
InsightForge (심층 분석)
질문을 LLM으로 하위 쿼리로 분해하여 각각을 검색, 집계합니다.
def insight_forge(self, graph_id: str, query: str, simulation_requirement: str):
# 1. Generate sub-queries using LLM
sub_queries = self._generate_sub_queries(query, simulation_requirement)
# 2. Search for each sub-query
for sub_query in sub_queries:
search_result = self.search_graph(graph_id, query=sub_query)
all_facts.extend(search_result.facts)
# 3. Extract entity UUIDs from edges
entity_uuids = set(edge['source_node_uuid'] for edge in all_edges)
# 4. Fetch detailed entity info
for uuid in entity_uuids:
node = self.get_node_detail(uuid)
entity_insights.append({...})
# 5. Build relationship chains
for edge in all_edges:
chain = f"{source_name} --[{relation_name}]--> {target_name}"
relationship_chains.append(chain)
PanoramaSearch (전체 범위)
만료/유효하지 않은 과거 사실까지 포함해 모든 이력을 검색합니다.
def panorama_search(self, graph_id: str, query: str, include_expired: bool = True):
all_nodes = self.get_all_nodes(graph_id)
all_edges = self.get_all_edges(graph_id, include_temporal=True)
for edge in all_edges:
is_historical = edge.is_expired or edge.is_invalid
if is_historical:
historical_facts.append(f"[{valid_at} - {invalid_at}] {edge.fact}")
else:
active_facts.append(edge.fact)
InterviewAgents (실시간)
실제 OASIS 인터뷰 API를 호출하여 활성 에이전트와 직접 대화합니다.
def interview_agents(self, simulation_id: str, interview_requirement: str):
# 1. Load agent profiles from CSV/JSON
profiles = self._load_agent_profiles(simulation_id)
# 2. Use LLM to select relevant agents
selected_agents, selected_indices, reasoning = self._select_agents_for_interview(...)
# 3. Generate interview questions
questions = self._generate_interview_questions(...)
# 4. Call real interview API (dual-platform)
api_result = SimulationRunner.interview_agents_batch(
simulation_id=simulation_id,
interviews=[{"agent_id": idx, "prompt": combined_prompt} for idx in selected_indices],
platform=None, # Interview both Twitter and Reddit
timeout=180.0
)
# 5. Parse and format results
for i, agent_idx in enumerate(selected_indices):
twitter_response = results_dict.get(f"twitter_{agent_idx}", {})
reddit_response = results_dict.get(f"reddit_{agent_idx}", {})
response_text = f"[Twitter]\n{twitter_response}\n\n[Reddit]\n{reddit_response}"
주요 엔지니어링 결정
1. 비동기 작업 관리
그래프 구축/시뮬레이션 실행 등 장기 작업은 비동기 스레드로 실행, 진행 상황을 추적합니다.
def build_graph_async(self, text: str, ontology: Dict, ...) -> str:
task_id = self.task_manager.create_task(task_type="graph_build", metadata={...})
thread = threading.Thread(
target=self._build_graph_worker,
args=(task_id, text, ontology, ...)
)
thread.daemon = True
thread.start()
return task_id
프런트엔드는 /api/graph/task/{task_id}로 상태를 폴링합니다.
2. 재시도 기능이 있는 배치 LLM 호출
에이전트 목록을 15개씩 배치 호출, JSON 잘림이 발생하면 복구 로직 적용
num_batches = math.ceil(len(entities) / self.AGENTS_PER_BATCH)
for batch_idx in range(num_batches):
batch_entities = entities[start_idx:end_idx]
batch_configs = self._generate_agent_configs_batch(context, batch_entities)
all_agent_configs.extend(batch_configs)
def _fix_truncated_json(self, content: str) -> str:
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
3. 듀얼 플랫폼 병렬 시뮬레이션
Twitter/Reddit 각각 별도 DB, 로그로 병렬 실행
uploads/simulations/{simulation_id}/
├── twitter/
│ ├── actions.jsonl
│ └── twitter_simulation.db
├── reddit/
│ ├── actions.jsonl
│ └── reddit_simulation.db
├── simulation_config.json
├── run_state.json
└── simulation.log
simulation_end 이벤트로 플랫폼별 완료 감지
성능 고려 사항
메모리 관리
- 대용량 문서는 50k자로 절단
- 엔티티 요약 300자 제한
- 최근 활동 50개만 메모리에 보관(전체 기록은 JSONL)
데이터베이스 격리
- 각 플랫폼별 별도 SQLite DB를 사용해 병렬 쓰기 시 경합 방지
점진적 성능 저하
Zep 검색 API 실패 시 로컬 키워드 매칭으로 자동 대체
try:
search_results = self.client.graph.search(...)
except Exception as e:
logger.warning(f"Zep Search API failed, falling back to local search: {e}")
return self._local_search(graph_id, query, limit, scope)
결론
MiroFish는 완전한 다중 에이전트 시뮬레이션 시스템을 처음부터 구축하는 구체적인 방법을 제공합니다.
5단계 워크플로는 원본 문서를 수천 명의 AI 에이전트가 현실적인 행동 패턴에 따라 상호작용하는 디지털 세계로 자동 변환합니다.
핵심 요약:
- 온톨로지 설계: 8+2 구조로 포괄성과 API 제한 동시 달성
- 비동기 워크플로: 장기 작업도 진행상황 추적 가능
- 시간 기반 활동: 신뢰할 수 있는 행동 패턴 생성
- 듀얼 플랫폼: Twitter/Reddit 병렬 실행으로 플랫폼 차이 분석
- 3단계 검색: 심층 분석, 전체 범위, 실시간 인터뷰 자동화
전체 소스코드는 github.com/666ghj/MiroFish에서 확인할 수 있습니다.
더 알아보고 싶다면 라이브 데모에서 직접 시뮬레이션을 체험해보세요.
Top comments (0)