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

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 immediately creates a unique task identifier using TaskManager.create_task (defined in 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 (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 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 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, 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.

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:

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:

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 handles the actual transmission of text chunks to the Zep service:

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/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:

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/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.
• 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 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 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, 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}