# How the Mirofish Graph Builder Processes Large Batches of Text and Tracks Simulation Episodes

> Learn how the Mirofish Graph Builder efficiently processes large text batches by chunking and uploading to Zep, while tracking simulation episodes via a background task manager.

- Repository: [BaiFu/mirofish](https://github.com/666ghj/mirofish)
- Tags: internals
- Published: 2026-02-23

---

**The `GraphBuilderService` in the 666ghj/mirofish repository handles arbitrarily large text sources by chunking them into overlapping segments, uploading them in configurable batches to the Zep graph service, and tracking each simulation episode through a background task manager with real-time progress polling.**

The 666ghj/mirofish project implements a robust pipeline for converting massive text corpora into structured knowledge graphs. Understanding how the graph builder processes large batches of text and tracks simulation episodes is essential for optimizing ingestion performance and monitoring long-running extraction jobs.

## Core Architecture: Asynchronous Graph Building

The system decouples the HTTP request lifecycle from the heavy processing work by leveraging background tasks and daemon threads.

### Background Task Management

When a client initiates a graph build, the `build_graph_async` method in [`backend/app/services/graph_builder.py`](https://github.com/666ghj/mirofish/blob/main/backend/app/services/graph_builder.py) immediately creates a unique task identifier using `TaskManager.create_task` (defined in [`backend/app/models/task.py`](https://github.com/666ghj/mirofish/blob/main/backend/app/models/task.py)). This stores a new `Task` object with status `PENDING` and returns the task ID to the caller so the client can poll for updates while the heavy lifting continues in the background.

The service then spawns a daemon thread via `threading.Thread(target=self._build_graph_worker, …).start()`, allowing the HTTP request to return immediately while the worker processes the text asynchronously.

### Text Chunking Strategy

Before uploading, the raw text must be divided into manageable pieces. The `TextProcessor.split_text` method in [`backend/app/services/text_processor.py`](https://github.com/666ghj/mirofish/blob/main/backend/app/services/text_processor.py) (which internally calls `split_text_into_chunks`) slices the entire document into overlapping segments. By default, it produces **500-character chunks** with a **50-character overlap** to preserve context across boundaries (configurable via `chunk_size` and `chunk_overlap` parameters).

This chunking ensures that even documents containing hundreds of thousands of characters are processed incrementally, keeping memory usage bounded and preventing payload size errors from the Zep API.

## Processing Large Batches: The Upload Pipeline

Once chunked, the system uploads the text to Zep in groups rather than individual API calls, optimizing network throughput and respecting rate limits.

### Configurable Batch Sizes

The `add_text_batches` method in [`backend/app/services/graph_builder.py`](https://github.com/666ghj/mirofish/blob/main/backend/app/services/graph_builder.py) groups chunks into batches according to the `batch_size` parameter (defaulting to **3**). For each batch, it constructs `EpisodeData` objects and sends them via `self.client.graph.add_batch`.

The method iterates over the chunk list in steps of `batch_size`, creating episodes with `type="text"` and the chunk content as `data`. After each successful upload, it extracts the episode UUID using `getattr(ep, "uuid_", getattr(ep, "uuid", None))` and appends it to the tracking list. A 1-second throttle (`time.sleep(1)`) between batches prevents overwhelming the downstream service.

### Episode UUID Collection

Each batch upload returns a list of created episode objects. The service immediately collects their unique identifiers into the `episode_uuids` list, which serves as the authoritative registry for subsequent polling. This collection happens inside `add_text_batches` before the method returns control to the worker thread, ensuring no episodes are lost between the upload and tracking phases.

## Tracking Simulation Episodes with Real-Time Polling

After upload, the system must wait for Zep to asynchronously process each episode into graph nodes and edges. The service implements a robust polling mechanism with granular progress reporting.

### The Polling Mechanism

The `_wait_for_episodes` method in [`backend/app/services/graph_builder.py`](https://github.com/666ghj/mirofish/blob/main/backend/app/services/graph_builder.py) manages the synchronization loop. It maintains a `pending` set of episode UUIDs and repeatedly queries the Zep endpoint via `self.client.graph.episode.get(uuid_=ep_uuid)` until each episode reports `processed=True`.

The loop implements a fixed **3-second polling interval** (`time.sleep(3)`) and respects a configurable **timeout** parameter defaulting to **600 seconds**. If the polling loop exceeds this duration before all episodes report completion, the method exits gracefully with a partial progress report rather than hanging indefinitely. The task status remains updated with the current completion percentage, allowing clients to decide whether to retry the operation or investigate the Zep service status.

### Granular Progress Reporting

Throughout the pipeline, the service reports status via a callback function with the signature `Callable[[str, float], None]`, which receives a human-readable message and a fractional progress value between 0 and 1. The `_build_graph_worker` passes this closure to sub-methods like `add_text_batches` and `_wait_for_episodes`, which invoke it with stage-specific updates.

The worker thread maps each pipeline phase to specific percentage ranges:

- **0–10%**: Graph creation via `create_graph` (`graph_id = f"mirofish_{uuid.uuid4().hex[:16]}"`)
- **10–15%**: Ontology upload via `set_ontology` (building dynamic Pydantic models)
- **15–20%**: Text chunking via `TextProcessor.split_text`
- **20–60%**: Batch uploads (progress scaled by `batch_progress * 0.4`)
- **60–90%**: Episode processing wait (scaled by `episode_progress * 0.3`)
- **90–100%**: Final aggregation via `_get_graph_info` (querying nodes, edges, and entity types)

These updates are forwarded to `TaskManager.update_task` in [`backend/app/models/task.py`](https://github.com/666ghj/mirofish/blob/main/backend/app/models/task.py), persisting the state to the task record. Clients retrieve real-time status by polling the `GET /api/graph/tasks/{task_id}` endpoint defined in [`backend/app/api/graph.py`](https://github.com/666ghj/mirofish/blob/main/backend/app/api/graph.py).

## Code Implementation Examples

### Using the Graph Builder (High-Level)

To start processing a large document, instantiate the `GraphBuilderService` and invoke `build_graph_async` with your text and ontology configuration:

```python
from backend.app.services.graph_builder import GraphBuilderService
from backend.app.services.text_processor import TextProcessor

# Prepare your large text corpus

raw_text = "..."  # Your document content

processed_text = TextProcessor.preprocess_text(raw_text)

# Define your ontology structure

ontology = {
    "entities": ["Person", "Organization", "Location"],
    "relations": ["works_for", "located_in"]
}

# Initialize the builder and start async processing

builder = GraphBuilderService()
task_id = builder.build_graph_async(
    text=processed_text,
    ontology=ontology,
    graph_name="Corporate Knowledge Graph",
    chunk_size=800,        # Larger chunks for fewer API calls

    chunk_overlap=100,     # Increased context overlap

    batch_size=5           # Upload 5 chunks per request

)

print(f"Task started: {task_id}")

```

The function returns immediately; the caller can poll `/api/graph/tasks/{task_id}` for live updates.

### Polling Task Progress

Retrieve real-time status updates as the graph builder processes large batches and tracks simulation episodes:

```python
import requests
import time

API_BASE = "http://localhost:8000/api/graph"

def get_task(task_id: str) -> dict:
    resp = requests.get(f"{API_BASE}/tasks/{task_id}")
    resp.raise_for_status()
    return resp.json()

while True:
    info = get_task(task_id)
    print(f"{info['progress']}% – {info['message']}")
    if info["status"] in ("completed", "failed"):
        break
    time.sleep(2)

```

### Internals – Adding Batches

The `add_text_batches` method in [`backend/app/services/graph_builder.py`](https://github.com/666ghj/mirofish/blob/main/backend/app/services/graph_builder.py) handles the actual transmission of text chunks to the Zep service:

```python
def add_text_batches(self, graph_id, chunks, batch_size=3, progress_callback=None):
    episode_uuids = []
    total_chunks = len(chunks)

    for i in range(0, total_chunks, batch_size):
        batch_chunks = chunks[i:i + batch_size]
        
        # Report progress before API call

        if progress_callback:
            progress = (i + len(batch_chunks)) / total_chunks
            progress_callback(f"Sending batch {i // batch_size + 1}...", progress)

        # Create episode data objects

        episodes = [EpisodeData(data=c, type="text") for c in batch_chunks]
        
        # Send to Zep

        batch_result = self.client.graph.add_batch(
            graph_id=graph_id, 
            episodes=episodes
        )

        # Collect episode UUIDs for tracking

        for ep in batch_result:
            uuid = getattr(ep, "uuid_", getattr(ep, "uuid", None))
            episode_uuids.append(uuid)

        time.sleep(1)  # Rate limiting

    return episode_uuids

```

See the full implementation in [[`graph_builder.py`](https://github.com/666ghj/mirofish/blob/main/graph_builder.py)](https://github.com/666ghj/mirofish/blob/main/backend/app/services/graph_builder.py#L94-L138).

### Episode Polling Implementation

The `_wait_for_episodes` method ensures all simulation episodes are fully processed before finalizing the graph:

```python
def _wait_for_episodes(self, episode_uuids, progress_callback=None, timeout=600):
    pending = set(episode_uuids)
    start_time = time.time()
    
    while pending:
        # Check timeout

        if time.time() - start_time > timeout:
            break
            
        # Poll each pending episode

        for uuid_ in list(pending):
            episode = self.client.graph.episode.get(uuid_=uuid_)
            if getattr(episode, "processed", False):
                pending.remove(uuid_)
        
        # Report progress

        if progress_callback:
            completed = len(episode_uuids) - len(pending)
            progress = completed / len(episode_uuids)
            progress_callback(
                f"Processing episodes {completed}/{len(episode_uuids)}", 
                progress
            )
            
        time.sleep(3)  # Polling interval

```

Full code lives in [[`graph_builder.py`](https://github.com/666ghj/mirofish/blob/main/graph_builder.py)](https://github.com/666ghj/mirofish/blob/main/backend/app/services/graph_builder.py#L140-L165).

## Summary

- **The `GraphBuilderService`** orchestrates the entire pipeline from raw text to knowledge graph via asynchronous background tasks managed by `TaskManager` in [`backend/app/models/task.py`](https://github.com/666ghj/mirofish/blob/main/backend/app/models/task.py).
- **Text chunking** splits large documents into 500-character overlapping segments by default, ensuring memory efficiency while preserving context across chunk boundaries.
- **Configurable batching** groups chunks into batches (default size 3) for transmission to the Zep API via `add_text_batches`, collecting episode UUIDs for downstream tracking.
- **Episode polling** via `_wait_for_episodes` monitors the `processed` status of each simulation episode with 3-second intervals and a 600-second timeout, guaranteeing graph completeness before finalization.
- **Real-time progress reporting** maps each pipeline stage to specific percentage ranges (0-10% creation, 10-15% ontology, 15-20% chunking, 20-60% upload, 60-90% waiting, 90-100% completion), enabling clients to track status via the `GET /api/graph/tasks/{task_id}` endpoint.

## Frequently Asked Questions

### How does the graph builder handle text files larger than available memory?

The `TextProcessor.split_text` method in [`backend/app/services/text_processor.py`](https://github.com/666ghj/mirofish/blob/main/backend/app/services/text_processor.py) implements streaming chunking that processes text incrementally rather than loading the entire document into memory. By splitting the source into configurable chunks (defaulting to 500 characters with 50-character overlaps), the `GraphBuilderService` maintains bounded memory usage even when processing hundreds of thousands of characters, uploading each batch to Zep before processing the next segment.

### What happens if the Zep API fails to process an episode within the timeout window?

The `_wait_for_episodes` method in [`backend/app/services/graph_builder.py`](https://github.com/666ghj/mirofish/blob/main/backend/app/services/graph_builder.py) enforces a configurable `timeout` parameter (defaulting to 600 seconds). If the polling loop exceeds this duration before all episodes report `processed=True`, the method exits gracefully with a partial progress report rather than hanging indefinitely. The task status remains updated with the current completion percentage, allowing clients to decide whether to retry the operation or investigate the Zep service status before resubmitting.

### Can I adjust the batch size to optimize upload throughput for my specific network conditions?

Yes, the `add_text_batches` method exposes a `batch_size` parameter (defaulting to 3) that controls how many text chunks are grouped into a single `self.client.graph.add_batch` API call. Increasing this value reduces the number of HTTP requests and network overhead for large documents, though you should balance this against the Zep API's payload limits and rate constraints. The method also implements a 1-second throttle (`time.sleep(1)`) between batches to prevent overwhelming the downstream service.

### How does the progress callback system provide real-time visibility into the graph building process?

The `GraphBuilderService` implements a stage-based progress mapping where each pipeline phase occupies a specific percentage range of the 0-100% scale. The `_build_graph_worker` passes a closure callback to sub-methods like `add_text_batches` and `_wait_for_episodes`, which invoke it with human-readable messages and fractional progress values. These updates are forwarded to `TaskManager.update_task` in [`backend/app/models/task.py`](https://github.com/666ghj/mirofish/blob/main/backend/app/models/task.py), persisting the state to the task record. Clients retrieve real-time status by polling the `GET /api/graph/tasks/{task_id}` endpoint defined in [`backend/app/api/graph.py`](https://github.com/666ghj/mirofish/blob/main/backend/app/api/graph.py).