Pendahuluan
Media sosial bergerak cepat. Satu postingan dapat memicu serangkaian reaksi, pembentukan ulang, dan gerakan balasan yang tidak terduga. Bagaimana jika Anda dapat melihat bagaimana sebuah skenario terungkap sebelum terjadi di dunia nyata?
MiroFish melakukan hal itu. Ini adalah mesin kecerdasan kelompok yang menciptakan dunia paralel digital di mana ribuan agen AI dengan kepribadian, memori, dan pola perilaku yang berbeda berinteraksi secara bebas. Anda mengunggah materi awal—artikel berita, draf kebijakan, bahkan novel—dan MiroFish membangun simulasi akurasi tinggi tentang bagaimana peristiwa dapat terungkap.
Postingan ini menguraikan arsitektur teknis di balik MiroFish. Anda akan mempelajari bagaimana sistem mengubah dokumen mentah menjadi simulasi yang hidup, bagaimana agen membuat keputusan, dan bagaimana alur kerja lima langkah mengatur segalanya mulai dari pembangunan grafik pengetahuan hingga pemantauan waktu nyata.
Ikhtisar Sistem: Alur Kerja Lima Langkah
MiroFish memproses simulasi melalui lima fase yang berbeda:
┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Langkah 1 │ ──► │ Langkah 2 │ ──► │ Langkah 3 │ ──► │ Langkah 4 │ ──► │ Langkah 5 │
│ Pembuatan │ │ Pembangunan │ │ Penyiapan │ │ Jalankan │ │ Pembuatan │
│ Ontologi │ │ GraphRAG │ │ Lingkungan│ │ Simulasi │ │ Laporan │
└─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘
Langkah 1: Pembuatan Ontologi
Sistem menganalisis dokumen masukan dan persyaratan simulasi Anda, kemudian menggunakan LLM untuk menghasilkan ontologi khusus. Ini mendefinisikan:
- 10 jenis entitas (misalnya, Mahasiswa, Profesor, Universitas, MediaOutlet, BadanPemerintah)
- 10 jenis hubungan (misalnya, BEKERJA_UNTUK, MENGOMENTARI, MENANGGAPI)
- Atribut untuk setiap jenis (menghindari kata-kata yang dipesan seperti
name,uuid,created_at)
Generator ontologi memberlakukan struktur dua tingkat: 8 jenis spesifik berdasarkan konten Anda, ditambah 2 jenis cadangan (Orang dan Organisasi) untuk menangkap apa pun yang tidak sesuai di tempat lain.
Langkah 2: Pembangunan GraphRAG
Dokumen dipotong-potong (500 karakter, 50 tumpang tindih) dan dikirim ke Zep Cloud dalam batch. Sistem:
- Membuat grafik mandiri dengan ID unik
- Mengatur ontologi khusus
- Mengirim batch teks untuk ekstraksi entitas dan hubungan
- Menunggu Zep memproses setiap episode
- Mengambil grafik akhir dengan node dan edge
Langkah 3: Penyiapan Lingkungan
Generator konfigurasi simulasi menganalisis grafik pengetahuan dan membuat parameter agen yang terperinci:
- Konfigurasi waktu berdasarkan pola zona waktu Tiongkok (jam sibuk 19-22, jam sepi 0-5)
- Konfigurasi acara dengan postingan awal dan topik hangat
- Konfigurasi aktivitas agen (postingan per jam, penundaan respons, bobot pengaruh)
- Konfigurasi platform untuk Twitter dan Reddit dengan ambang batas viral yang berbeda
Langkah 4: Jalankan Simulasi
Agen bangun sesuai jadwal aktivitas mereka dan mulai memposting, berkomentar, dan bereaksi. Sistem menjalankan simulasi paralel di Twitter dan Reddit, mencatat setiap tindakan ke file JSONL secara waktu nyata.
Langkah 5: Pembuatan Laporan
Agen Laporan menggunakan tiga alat pengambilan inti untuk menganalisis apa yang terjadi:
- InsightForge: Pencarian mendalam yang menguraikan pertanyaan menjadi sub-kueri
- PanoramaSearch: Tampilan cakupan penuh termasuk fakta historis yang kedaluwarsa/tidak valid
- InterviewAgents: Wawancara waktu nyata dengan agen aktif melalui IPC
Pembahasan Teknis Mendalam: Pembuatan Ontologi
Generator ontologi berada di backend/app/services/ontology_generator.py. Ini menggunakan perintah sistem yang dibuat dengan cermat yang memberlakukan aturan ketat.
Perintah sistem mencakup panduan ekstensif tentang apa yang dianggap sebagai entitas yang valid (orang, organisasi, media) versus apa yang tidak (konsep abstrak, tema, sudut pandang). Perbedaan ini penting karena simulasi membutuhkan agen yang benar-benar dapat berbicara dan bertindak di media sosial.
Setelah LLM menghasilkan ontologi, metode _validate_and_process memberlakukan batasan:
def _validate_and_process(self, result: Dict[str, Any]) -> Dict[str, Any]:
# Batasan API Zep: maksimal 10 jenis entitas, maksimal 10 jenis edge
MAX_ENTITY_TYPES = 10
MAX_EDGE_TYPES = 10
# Pastikan jenis cadangan ada
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)
# Pangkas jika penambahan cadangan akan melebihi batas
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
Lapisan validasi ini memastikan output selalu berfungsi dengan batasan API Zep sambil mempertahankan struktur dua tingkat.
Pembangunan Grafik Pengetahuan: Integrasi Zep
Layanan pembangun grafik (backend/app/services/graph_builder.py) menangani alur kerja asinkron:
def _build_graph_worker(self, task_id: str, text: str, ontology: Dict, ...):
# 1. Buat grafik
graph_id = self.create_graph(graph_name)
# 2. Atur ontologi
self.set_ontology(graph_id, ontology)
# 3. Potong teks
chunks = TextProcessor.split_text(text, chunk_size, chunk_overlap)
# 4. Kirim batch
episode_uuids = self.add_text_batches(graph_id, chunks, batch_size)
# 5. Tunggu pemrosesan Zep
self._wait_for_episodes(episode_uuids, progress_callback)
# 6. Ambil grafik akhir
graph_info = self._get_graph_info(graph_id)
Pembuatan Model Pydantic Dinamis
Satu bagian cerdik: sistem secara dinamis membuat model Pydantic untuk setiap jenis entitas pada saat runtime:
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
Ini memungkinkan Zep memvalidasi atribut entitas terhadap skema khusus tanpa memerlukan model yang telah ditentukan sebelumnya.
Paginasi Melalui Grafik Besar
Zep mengembalikan hasil yang dipaginasi. Utilitas zep_paging.py mengambil semuanya:
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
Simulasi Aktivitas Agen Berbasis Waktu
Generator konfigurasi simulasi (backend/app/services/simulation_config_generator.py) membuat pola aktivitas realistis berdasarkan perilaku zona waktu Tiongkok:
CHINA_TIMEZONE_CONFIG = {
"dead_hours": [0, 1, 2, 3, 4, 5], # Hampir tidak ada orang di dini hari
"morning_hours": [6, 7, 8], # Secara bertahap aktif di pagi hari
"work_hours": [9, 10, 11, 12, 13, 14, 15, 16, 17, 18],
"peak_hours": [19, 20, 21, 22], # Puncak malam hari
"night_hours": [23],
"activity_multipliers": {
"dead": 0.05,
"morning": 0.4,
"work": 0.7,
"peak": 1.5,
"night": 0.5
}
}
Jenis agen yang berbeda mendapatkan pola yang berbeda:
| 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 |
Generator konfigurasi menggunakan panggilan LLM untuk menyesuaikan nilai-nilai ini berdasarkan skenario spesifik Anda, kemudian kembali ke nilai default berbasis aturan jika LLM gagal.
Pelacakan Aksi Waktu Nyata
Runner simulasi (backend/app/services/simulation_runner.py) memantau aktivitas agen dengan mengalirkan log 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)
# Tangani peristiwa
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
# Uraikan tindakan agen
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()
Ini berjalan dalam thread latar belakang, memperbarui status simulasi setiap 2 detik. Frontend melakukan polling status ini untuk menampilkan kemajuan waktu nyata.
Manajemen Proses Lintas-Platform
Menghentikan simulasi memerlukan manajemen proses yang cermat di Windows dan Unix:
def _terminate_process(cls, process: subprocess.Popen, simulation_id: str, timeout: int = 10):
if IS_WINDOWS:
# Windows: gunakan taskkill untuk mematikan pohon proses
subprocess.run(['taskkill', '/PID', str(process.pid), '/T'], ...)
else:
# Unix: matikan grup proses (dibuat dengan start_new_session=True)
os.killpg(os.getpgid(process.pid), signal.SIGTERM)
Penangan pembersihan mendaftarkan penangan sinyal untuk SIGINT, SIGTERM, dan SIGHUP:
def register_cleanup(cls):
def cleanup_handler(signum, frame):
cls.cleanup_all_simulations()
# Kemudian panggil penangan asli
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)
Ini memastikan simulasi berhenti dengan baik saat server dimatikan.
Pembuatan Laporan: Pengambilan Tiga Tingkat
Layanan alat Zep (backend/app/services/zep_tools.py) menyediakan tiga fungsi pengambilan:
InsightForge (Penelusuran Mendalam)
Menguraikan pertanyaan kompleks menjadi sub-kueri, mencari masing-masing, lalu mengagregasi:
def insight_forge(self, graph_id: str, query: str, simulation_requirement: str):
# 1. Hasilkan sub-kueri menggunakan LLM
sub_queries = self._generate_sub_queries(query, simulation_requirement)
# 2. Cari setiap sub-kueri
for sub_query in sub_queries:
search_result = self.search_graph(graph_id, query=sub_query)
all_facts.extend(search_result.facts)
# 3. Ekstrak UUID entitas dari edge
entity_uuids = set(edge['source_node_uuid'] for edge in all_edges)
# 4. Ambil info entitas terperinci
for uuid in entity_uuids:
node = self.get_node_detail(uuid)
entity_insights.append({...})
# 5. Bangun rantai hubungan
for edge in all_edges:
chain = f"{source_name} --[{relation_name}]--> {target_name}"
relationship_chains.append(chain)
PanoramaSearch (Cakupan Penuh)
Mengambil semuanya termasuk fakta historis yang kedaluwarsa/tidak valid:
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 (Waktu Nyata)
Memanggil API wawancara OASIS yang sebenarnya untuk berbicara dengan agen aktif:
def interview_agents(self, simulation_id: str, interview_requirement: str):
# 1. Muat profil agen dari CSV/JSON
profiles = self._load_agent_profiles(simulation_id)
# 2. Gunakan LLM untuk memilih agen yang relevan
selected_agents, selected_indices, reasoning = self._select_agents_for_interview(...)
# 3. Hasilkan pertanyaan wawancara
questions = self._generate_interview_questions(...)
# 4. Panggil API wawancara nyata (dua-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. Uraikan dan format hasil
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}"
Keputusan Rekayasa Utama
1. Manajemen Tugas Asinkron
Operasi yang berjalan lama (pembangunan grafik, jalankan simulasi) menggunakan tugas asinkron dengan pelacakan kemajuan:
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
Frontend melakukan polling status tugas melalui /api/graph/task/{task_id}.
2. Panggilan LLM Batch dengan Coba Lagi
Pembuatan konfigurasi membagi daftar agen besar menjadi batch 15:
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)
Setiap batch mencakup logika perbaikan JSON untuk output yang terpotong:
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. Simulasi Paralel Dua-Platform
Twitter dan Reddit berjalan secara paralel dengan database terpisah dan log tindakan:
uploads/simulations/{simulation_id}/
├── twitter/
│ ├── actions.jsonl
│ └── twitter_simulation.db
├── reddit/
│ ├── actions.jsonl
│ └── reddit_simulation.db
├── simulation_config.json
├── run_state.json
└── simulation.log
Runner mendeteksi penyelesaian per-platform melalui peristiwa simulation_end.
Pertimbangan Kinerja
Manajemen Memori
- Dokumen besar dipotong menjadi 50 ribu karakter untuk konteks LLM
- Ringkasan entitas dibatasi hingga 300 karakter masing-masing
- Tindakan terbaru dibatasi 50 dalam memori (riwayat lengkap dalam file JSONL)
Isolasi Basis Data
Setiap platform menggunakan database SQLite-nya sendiri untuk menghindari pertentangan kunci selama penulisan paralel.
Penurunan Kinerja yang Elegan (Graceful Degradation)
Ketika API Pencarian Zep gagal, sistem kembali menggunakan pencocokan kata kunci 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)
Kesimpulan
MiroFish menunjukkan bagaimana membangun sistem simulasi multi-agen lengkap dari awal. Alur kerja lima langkah mengubah dokumen mentah menjadi dunia digital yang hidup di mana ribuan agen berinteraksi sesuai dengan pola perilaku realistis.
Poin-poin penting:
- Desain ontologi itu penting: Struktur dua tingkat (8 jenis spesifik + 2 jenis cadangan) memastikan cakupan tanpa melebihi batasan API
- Alur kerja asinkron memungkinkan operasi yang panjang: Pelacakan tugas dengan pembaruan kemajuan membuat pengguna tetap terinformasi selama operasi yang berlangsung beberapa menit
- Aktivitas berbasis waktu menciptakan realisme: Pola zona waktu Tiongkok dan jadwal khusus jenis agen menghasilkan perilaku yang meyakinkan
- Simulasi dua-platform menyediakan perbandingan: Menjalankan Twitter dan Reddit secara paralel menunjukkan bagaimana dinamika platform memengaruhi hasil
- Pengambilan tiga tingkat melayani kebutuhan yang berbeda: InsightForge untuk kedalaman, PanoramaSearch untuk luas, InterviewAgents untuk perspektif langsung
Kode sumber lengkap tersedia di github.com/666ghj/MiroFish.
Ingin mencoba MiroFish? Kunjungi demo langsung untuk melihat simulasi acara hotspot beraksi.