How MiroFish Creates Digital Parallel Worlds ?

Introduction

Social media moves fast. A single post can trigger cascades of reactions, reshapes, and counter-movements that nobody predicted. What if you could see how a scenario plays out before it happens in the real world?

Try Apidog today

MiroFish is a swarm-intelligence engine that creates digital parallel worlds where thousands of AI agents with distinct personalities, memories, and behavioral patterns interact freely. Upload seed material—a news article, policy draft, or novel—and MiroFish builds a simulation of how events might unfold.

💡 MiroFish used Apidog to design, debug, and document backend APIs before implementing simulation logic. This helped catch endpoint issues early and keep the Python backend and Vue frontend aligned.

This post breaks down the technical architecture behind MiroFish. You’ll see how the system turns raw documents into simulations, how agents make decisions, and how the five-step workflow handles knowledge graph construction, simulation execution, and real-time monitoring.

System Overview: The Five-Step Workflow

MiroFish processes a simulation through five phases:

┌─────────────┐     ┌─────────────┐     ┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│   Step 1    │ ──► │   Step 2    │ ──► │   Step 3    │ ──► │   Step 4    │ ──► │   Step 5    │
│  Ontology   │     │  GraphRAG   │     │   Env       │     │ Simulation  │     │   Report    │
│  Generation │     │   Build     │     │   Setup     │     │   Run       │     │ Generation  │
└─────────────┘     └─────────────┘     └─────────────┘     └─────────────┘     └─────────────┘

Step 1: Generate an Ontology

Start by analyzing the source documents and simulation requirements. An LLM generates a custom ontology that defines:

• Up to 10 entity types, such as Student, Professor, University, MediaOutlet, and GovernmentAgency
• Up to 10 relationship types, such as WORKS_FOR, COMMENTS_ON, and RESPONDS_TO
• Attributes for every entity type, excluding reserved names such as name, uuid, and created_at

Use a two-tier entity strategy:

1. Generate eight content-specific entity types.
2. Always reserve Person and Organization as fallback types.

This prevents unmatched entities from being dropped while staying within Zep API limits.

Step 2: Build the GraphRAG Knowledge Graph

Next, split documents into chunks of 500 characters with a 50-character overlap. Send those chunks to Zep Cloud in batches:

1. Create a graph with a unique ID.
2. Apply the generated ontology.
3. Submit text batches for entity and relationship extraction.
4. Wait for each episode to finish processing.
5. Fetch the completed graph, including nodes and edges.

Step 3: Generate the Simulation Environment

Use the knowledge graph to generate simulation configuration:

• Time windows based on Chinese timezone behavior
• Initial posts and hot topics
• Agent activity rates, response delays, and influence weights
• Separate platform configuration for Twitter and Reddit

Step 4: Run the Simulation

Agents wake up based on their schedules, then post, comment, and react. Twitter and Reddit simulations run in parallel, with every action written to JSONL logs in real time.

Step 5: Generate a Report

The Report Agent analyzes the completed simulation through three retrieval tools:

• InsightForge for deep, multi-query investigation
• PanoramaSearch for full graph context, including historical facts
• InterviewAgents for real-time interviews with active agents over IPC

---

Implement Ontology Generation

The ontology generator is implemented in:

backend/app/services/ontology_generator.py

The LLM prompt must clearly distinguish valid entities from abstract ideas. For example:

• Valid entities: people, organizations, media outlets
• Invalid entities: themes, viewpoints, and abstract concepts

This matters because the simulation needs entities that can act, post, respond, and influence others.

After the LLM returns an ontology, validate it before sending it to Zep.

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

The key implementation detail is adding fallback types after validating the generated types. If the generated ontology already consumes all available slots, trim it first so Person and Organization are guaranteed to fit.

---

Build the Knowledge Graph with Zep

The graph builder service lives in:

backend/app/services/graph_builder.py

Run graph construction in a worker instead of blocking the request handler:

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)

This workflow gives the frontend a place to report progress while Zep processes uploaded content.

Generate Pydantic Models Dynamically

MiroFish generates a Pydantic entity model at runtime for each ontology type. This lets Zep validate entity attributes against the generated schema without requiring pre-defined Python classes.

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

When implementing dynamic schemas, protect framework-reserved fields. Prefixing a conflicting field with entity_ preserves the source meaning without breaking the generated model.

Fetch All Pages from Large Graphs

Zep graph results are paginated. Do not assume a single request returns every node.

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

Use the same cursor-based pattern for edges and any other paginated graph resource.

---

Model Time-Based Agent Activity

The simulation config generator is located at:

backend/app/services/simulation_config_generator.py

MiroFish uses Chinese timezone activity patterns to make agent behavior less uniform:

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,
    },
}

Assign different baseline behavior by agent type:

Agent type	Activity level	Active hours	Response delay	Influence
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

The config generator can use LLM calls to adapt these values to a scenario. If an LLM call fails, use rule-based defaults so simulation setup can continue.

---

Stream and Track Agent Actions in Real Time

The simulation runner is implemented in:

backend/app/services/simulation_runner.py

MiroFish streams action logs from JSONL files. Store the current file position so each polling cycle only processes newly appended actions.

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 simulation 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()

Run this reader in a background thread and update simulation state every two seconds. The frontend can then poll the state endpoint to render live progress.

---

Stop Cross-Platform Simulations Safely

Twitter and Reddit run as separate processes, so shutdown logic must terminate process trees correctly on each operating system.

def _terminate_process(
    cls,
    process: subprocess.Popen,
    simulation_id: str,
    timeout: int = 10,
):
    if IS_WINDOWS:
        # Windows: kill the process tree
        subprocess.run(["taskkill", "/PID", str(process.pid), "/T"], ...)
    else:
        # Unix: kill the process group created with start_new_session=True
        os.killpg(os.getpgid(process.pid), signal.SIGTERM)

Register cleanup handlers for server shutdown and process termination:

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)

This ensures simulations are cleaned up when the server receives SIGINT, SIGTERM, or—where available—SIGHUP.

---

Implement Three-Tier Report Retrieval

The Zep tools service is located at:

backend/app/services/zep_tools.py

Use different retrieval methods for different reporting needs.

InsightForge: Deep-Dive Analysis

InsightForge decomposes a broad question into smaller searches, then aggregates entities, facts, and relationship chains.

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 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)

Use this path when the report needs explanation and connected reasoning rather than a simple lookup.

PanoramaSearch: Full Graph Scope

PanoramaSearch includes both current and historical graph facts.

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)

Use this retrieval mode when a report needs to distinguish active facts from expired or invalid historical facts.

InterviewAgents: Query Active Agents

InterviewAgents calls the OASIS interview API to get responses from active agents on both platforms.

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 the interview API for both platforms
    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,
    )

    # 5. Format dual-platform responses
    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"
            f"[Reddit]\n{reddit_response}"
        )

Set platform=None to interview the selected agents on both Twitter and Reddit in one workflow.

---

Key Engineering Decisions

1. Run Long Operations as Async Tasks

Graph construction and simulation runs can take multiple minutes. Start them in a background thread and return a task ID immediately.

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

The frontend polls task state through:

/api/graph/task/{task_id}

2. Batch LLM Calls and Repair Truncated JSON

When generating many agent configurations, split entities into batches of 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)

If a model response contains truncated JSON, attempt structural repair before parsing:

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. Isolate Dual-Platform Simulation Data

Run Twitter and Reddit in parallel, but give each platform its own database and action log.

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

The runner detects completion independently for each platform by watching for simulation_end events.

---

Performance and Reliability Checklist

Manage Memory Explicitly

MiroFish limits in-memory and LLM-context data:

• Truncate large documents to 50,000 characters for LLM context.
• Limit entity summaries to 300 characters each.
• Keep only the 50 most recent actions in memory.
• Store full action history in JSONL files.

Use Database Isolation

Keep a separate SQLite database per platform. This avoids lock contention when Twitter and Reddit simulations write in parallel.

Add a Search Fallback

If Zep Search is unavailable, fall back to local keyword matching.

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)

This fallback keeps report generation functional even when the remote search API fails.

Conclusion

MiroFish demonstrates a complete multi-agent simulation workflow: transform documents into a knowledge graph, generate agent behavior, run parallel platform simulations, and analyze the resulting interactions.

Key implementation takeaways:

• Design ontologies defensively: use eight specific types plus Person and Organization fallbacks while respecting API limits.
• Move long-running work into tracked background tasks so clients can receive progress updates.
• Use time-based schedules and agent-specific behavior to create more realistic activity patterns.
• Isolate Twitter and Reddit data to support parallel simulation without database lock contention.
• Match retrieval strategy to the report: use InsightForge for depth, PanoramaSearch for breadth, and InterviewAgents for direct agent perspectives.

The full source code is available at github.com/666ghj/MiroFish.

Want to try MiroFish? Visit the live demo to see a hotspot event simulation in action.