DEV Community

Cover image for Cara MiroFish Menciptakan Dunia Paralel Digital
Walse
Walse

Posted on • Originally published at apidog.com

Cara MiroFish Menciptakan Dunia Paralel Digital

Pendahuluan

Media sosial bergerak sangat cepat. Satu postingan bisa memicu reaksi berantai, pembentukan ulang opini, hingga gerakan balasan yang tidak terduga. Bagaimana jika Anda bisa mensimulasikan skenario sebelum benar-benar terjadi di dunia nyata?

Coba Apidog hari ini

MiroFish melakukan hal tersebut. Ini adalah mesin kecerdasan kelompok yang menciptakan dunia digital paralel: ribuan agen AI dengan kepribadian, memori, dan pola perilaku unik saling berinteraksi. Anda cukup mengunggah materi awal (artikel berita, draf kebijakan, atau novel), lalu MiroFish membangun simulasi akurat tentang bagaimana peristiwa bisa berkembang.

💡 Membangun MiroFish membutuhkan fondasi pengujian API yang andal. Tim menggunakan Apidog untuk merancang, debug, dan dokumentasi semua API backend sebelum menulis logika simulasi. Hal ini membantu mendeteksi masalah endpoint lebih awal dan menjaga sinkronisasi backend Python serta frontend Vue selama pengembangan.

Artikel ini membedah arsitektur teknis di balik MiroFish. Anda akan belajar bagaimana sistem mengubah dokumen mentah menjadi simulasi hidup, cara pengambilan keputusan agen, dan alur kerja lima langkah yang mengatur proses mulai dari pembuatan knowledge graph hingga pemantauan waktu nyata.

gambar-189

Ikhtisar Sistem: Alur Kerja Lima Langkah

MiroFish menjalankan simulasi melalui 5 fase utama:

┌─────────────┐     ┌─────────────┐     ┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│   Langkah 1   │ ──► │   Langkah 2   │ ──► │   Langkah 3   │ ──► │   Langkah 4   │ ──► │   Langkah 5   │
│  Pembuatan  │     │ Pembangunan │     │   Penyiapan │     │   Jalankan  │     │   Pembuatan │
│  Ontologi   │     │  GraphRAG   │     │   Lingkungan│     │   Simulasi  │     │   Laporan   │
└─────────────┘     └─────────────┘     └─────────────┘     └─────────────┘     └─────────────┘
Enter fullscreen mode Exit fullscreen mode

Langkah 1: Pembuatan Ontologi

Sistem menganalisis dokumen input dan kebutuhan simulasi, lalu menggunakan LLM untuk menghasilkan ontologi khusus yang terdiri dari:

  • 10 jenis entitas (mis. Mahasiswa, Profesor, Universitas, MediaOutlet, BadanPemerintah)
  • 10 jenis hubungan (mis. BEKERJA_UNTUK, MENGOMENTARI, MENANGGAPI)
  • Atribut untuk tiap jenis (hindari kata reserved seperti name, uuid, created_at)

Struktur dua tingkat diterapkan: 8 jenis spesifik sesuai konten + 2 fallback (Person, Organization).

Langkah 2: Pembangunan GraphRAG

  • Dokumen di-split (500 karakter, 50 overlap) lalu dikirim ke Zep Cloud secara batch.
  • Proses:
    1. Buat graph ID unik
    2. Set ontologi custom
    3. Kirim batch teks untuk ekstraksi entitas/hubungan
    4. Tunggu proses Zep per batch
    5. Ambil graph hasil ekstraksi

Langkah 3: Penyiapan Lingkungan

  • Generator konfigurasi membaca knowledge graph untuk membuat parameter detail:
    • Konfigurasi waktu (zona waktu Tiongkok: jam puncak 19-22, jam sepi 0-5)
    • Konfigurasi event: posting awal, topik panas
    • Aktivitas agen: posting/jam, delay, bobot pengaruh
    • Platform: Twitter dan Reddit, threshold viral berbeda

Langkah 4: Jalankan Simulasi

Agen aktif sesuai jadwal, melakukan posting, komentar, reaksi. Simulasi paralel untuk Twitter dan Reddit, semua aksi direkam real-time ke file JSONL.

Langkah 5: Pembuatan Laporan

Agen laporan menganalisis hasil lewat tiga alat inti:

  • InsightForge: pencarian mendalam, sub-kueri otomatis
  • PanoramaSearch: cakupan penuh, termasuk fakta historis/expired
  • InterviewAgents: wawancara real-time dengan agen aktif via IPC

Pembahasan Teknis Mendalam: Pembuatan Ontologi

Kode generator ontologi ada di backend/app/services/ontology_generator.py, menggunakan system prompt ketat untuk memastikan entitas valid (orang, organisasi, media), bukan konsep abstrak.

Validasi output dilakukan dengan:

def _validate_and_process(self, result: Dict[str, Any]) -> Dict[str, Any]:
    MAX_ENTITY_TYPES = 10
    MAX_EDGE_TYPES = 10

    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)

    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
Enter fullscreen mode Exit fullscreen mode

Ini memastikan jumlah entitas/hubungan sesuai batas API Zep, tetap dengan struktur dua tingkat.

Pembangunan Grafik Pengetahuan: Integrasi Zep

Layanan pembangun graph (backend/app/services/graph_builder.py) menangani pipeline asinkron:

def _build_graph_worker(self, task_id: str, text: str, ontology: Dict, ...):
    graph_id = self.create_graph(graph_name)
    self.set_ontology(graph_id, ontology)
    chunks = TextProcessor.split_text(text, chunk_size, chunk_overlap)
    episode_uuids = self.add_text_batches(graph_id, chunks, batch_size)
    self._wait_for_episodes(episode_uuids, progress_callback)
    graph_info = self._get_graph_info(graph_id)
Enter fullscreen mode Exit fullscreen mode

Pembuatan Model Pydantic Dinamis

Model Pydantic dibuat dinamis per tipe entitas:

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
Enter fullscreen mode Exit fullscreen mode

Paginasi pada Grafik Besar

Untuk mengambil seluruh node secara paginasi:

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
Enter fullscreen mode Exit fullscreen mode

Simulasi Aktivitas Agen Berbasis Waktu

Generator konfigurasi simulasi (backend/app/services/simulation_config_generator.py) membangun pola aktivitas berdasarkan zona waktu Tiongkok:

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
    }
}
Enter fullscreen mode Exit fullscreen mode

Setiap tipe agen punya pola berbeda sesuai peran dan pengaruh:

Jenis Agen Tingkat Aktivitas Jam Aktif Penundaan Respons Pengaruh
University 0.2 9-17 60-240 min 3.0
MediaOutlet 0.5 7-23 5-30 min 2.5
Student 0.8 8-12, 18-23 1-15 min 0.8
Professor 0.4 8-21 15-90 min 2.0

Nilai-nilai ini bisa dikustomisasi via LLM, fallback ke default jika gagal.

Pelacakan Aksi Waktu Nyata

Runner simulasi (backend/app/services/simulation_runner.py) memantau aktivitas dengan streaming file JSONL:

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)

            if "event_type" in action_data:
                if action_data["event_type"] == "simulation_end":
                    state.twitter_completed = True  # atau reddit
                elif action_data["event_type"] == "round_end":
                    state.current_round = action_data["round"]
                continue

            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()
Enter fullscreen mode Exit fullscreen mode

Thread background ini memperbarui status simulasi tiap 2 detik. Frontend polling status untuk tampilan real-time.

Manajemen Proses Lintas-Platform

Menghentikan simulasi harus aman di Windows & Unix:

def _terminate_process(cls, process: subprocess.Popen, simulation_id: str, timeout: int = 10):
    if IS_WINDOWS:
        subprocess.run(['taskkill', '/PID', str(process.pid), '/T'], ...)
    else:
        os.killpg(os.getpgid(process.pid), signal.SIGTERM)
Enter fullscreen mode Exit fullscreen mode

Pembersihan otomatis dengan handler sinyal:

def register_cleanup(cls):
    def cleanup_handler(signum, frame):
        cls.cleanup_all_simulations()

    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)
Enter fullscreen mode Exit fullscreen mode

Pembuatan Laporan: Pengambilan Tiga Tingkat

Layanan alat Zep (backend/app/services/zep_tools.py) menyediakan tiga mode pengambilan inti:

InsightForge (Penelusuran Mendalam)

Memecah pertanyaan rumit menjadi sub-kueri lalu agregasi hasil:

def insight_forge(self, graph_id: str, query: str, simulation_requirement: str):
    sub_queries = self._generate_sub_queries(query, simulation_requirement)
    for sub_query in sub_queries:
        search_result = self.search_graph(graph_id, query=sub_query)
        all_facts.extend(search_result.facts)

    entity_uuids = set(edge['source_node_uuid'] for edge in all_edges)
    for uuid in entity_uuids:
        node = self.get_node_detail(uuid)
        entity_insights.append({...})

    for edge in all_edges:
        chain = f"{source_name} --[{relation_name}]--> {target_name}"
        relationship_chains.append(chain)
Enter fullscreen mode Exit fullscreen mode

PanoramaSearch (Cakupan Penuh)

Mengambil fakta aktif dan historis sekaligus:

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)
Enter fullscreen mode Exit fullscreen mode

InterviewAgents (Waktu Nyata)

Wawancara langsung ke agen aktif via API OASIS:

def interview_agents(self, simulation_id: str, interview_requirement: str):
    profiles = self._load_agent_profiles(simulation_id)
    selected_agents, selected_indices, reasoning = self._select_agents_for_interview(...)
    questions = self._generate_interview_questions(...)
    api_result = SimulationRunner.interview_agents_batch(
        simulation_id=simulation_id,
        interviews=[{"agent_id": idx, "prompt": combined_prompt} for idx in selected_indices],
        platform=None,
        timeout=180.0
    )

    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}"
Enter fullscreen mode Exit fullscreen mode

Keputusan Rekayasa Utama

1. Manajemen Tugas Asinkron

Operasi berat (graph build, simulasi) dijalankan asinkron dengan progress tracking:

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
Enter fullscreen mode Exit fullscreen mode

Frontend polling status via endpoint /api/graph/task/{task_id}.

2. Panggilan LLM Batch dengan Retry

Konfigurasi agen dibagi per batch 15, termasuk logika perbaikan JSON jika output terpotong:

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
Enter fullscreen mode Exit fullscreen mode

3. Simulasi Paralel Dua-Platform

Struktur file simulasi:

uploads/simulations/{simulation_id}/
├── twitter/
│   ├── actions.jsonl
│   └── twitter_simulation.db
├── reddit/
│   ├── actions.jsonl
│   └── reddit_simulation.db
├── simulation_config.json
├── run_state.json
└── simulation.log
Enter fullscreen mode Exit fullscreen mode

Runner mendeteksi selesai per-platform via event simulation_end.

Pertimbangan Kinerja

Manajemen Memori

  • Dokumen besar di-split maksimal 50 ribu karakter untuk input LLM
  • Ringkasan entitas dibatasi 300 karakter
  • Riwayat aksi terbaru di memori: 50 aksi (full history di JSONL)

Isolasi Basis Data

Setiap platform pakai SQLite sendiri untuk menghindari konflik penulisan paralel.

Graceful Degradation

Jika API Zep gagal, fallback ke pencarian lokal:

try:
    search_results = self.client.graph.search(...)
except Exception as e:
    logger.warning(f"API Pencarian Zep gagal, kembali ke pencarian lokal: {e}")
    return self._local_search(graph_id, query, limit, scope)
Enter fullscreen mode Exit fullscreen mode

Kesimpulan

MiroFish mendemonstrasikan cara membangun sistem simulasi multi-agen end-to-end. Alur kerja lima langkah mengubah dokumen mentah menjadi simulasi dunia digital realistis dengan ribuan agen interaktif.

Highlight utama:

  1. Desain ontologi krusial: Struktur dua tingkat (8 spesifik + 2 fallback) memastikan coverage tanpa melebihi batas API.
  2. Alur kerja asinkron: Task tracking & progress update menjaga pengalaman pengguna selama proses panjang.
  3. Aktivitas berbasis waktu: Pola zona waktu dan jadwal tiap jenis agen menciptakan realisme.
  4. Simulasi dua-platform: Twitter dan Reddit berjalan paralel, memperlihatkan perbedaan dinamika.
  5. Tiga mode pengambilan laporan: InsightForge untuk kedalaman, PanoramaSearch untuk luas, InterviewAgents untuk perspektif live.

Kode sumber lengkap tersedia di github.com/666ghj/MiroFish.

Ingin mencoba MiroFish? Kunjungi demo langsung untuk simulasi event hotspot secara real-time.

Top comments (0)