# How the MiroFish Graph Builder Service Integrates with Zep for Knowledge Graph Construction

> Learn how the MiroFish GraphBuilderService integrates with Zep to automate knowledge graph construction by managing initialization, ontology definition, and text chunking.

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

---

**The GraphBuilderService acts as an orchestration bridge between MiroFish’s Flask backend and Zep Cloud, managing graph initialization, ontology definition, text chunking, and asynchronous episode processing to automate knowledge graph construction.**

The open-source **MiroFish** repository implements a sophisticated pipeline for automated knowledge graph construction by integrating with the **Zep Cloud** platform. This deep dive examines how the `GraphBuilderService` in [`backend/app/services/graph_builder.py`](https://github.com/666ghj/mirofish/blob/main/backend/app/services/graph_builder.py) wraps Zep’s SDK to handle the complete lifecycle of graph creation—from API authentication and ontology modeling to batch data ingestion and status polling—while maintaining a clean asynchronous interface for the REST API layer.

## Architecture of the Zep Integration

The integration centers on the `GraphBuilderService` class, which encapsulates all Zep Cloud interactions behind a Pythonic API. This design decouples MiroFish’s business logic from Zep-specific implementation details, allowing the backend to treat knowledge graph construction as a managed background task.

### Service Initialization and Client Configuration

The service initializes a **Zep client** using the `ZEP_API_KEY` environment variable, falling back to the `Config` object if no explicit key is provided at instantiation. According to the source code at lines 44–50 of [`backend/app/services/graph_builder.py`](https://github.com/666ghj/mirofish/blob/main/backend/app/services/graph_builder.py), the constructor establishes this authenticated session, which all subsequent methods reuse for API calls.

### The Graph Construction Workflow

The complete integration follows a nine-step orchestration pattern:

1. **Graph Allocation** – `create_graph()` invokes `self.client.graph.create()` to provision a new Zep graph and returns its UUID (lines 86–96).
2. **Ontology Modeling** – `set_ontology()` dynamically generates Pydantic entity and edge classes from a dictionary schema, then pushes them to Zep via `self.client.graph.set_ontology()` (lines 98–86).
3. **Text Segmentation** – The static `TextProcessor.split_text()` utility segments raw documents into 500-character chunks with 50-character overlaps.
4. **Batch Ingestion** – `add_text_batches()` converts chunks into `EpisodeData` objects and streams them to Zep using `self.client.graph.add_batch()`, returning episode UUIDs for tracking (lines 87–138).
5. **Processing Synchronization** – `_wait_for_episodes()` polls `self.client.graph.episode.get()` until every episode’s `processed` flag indicates completion or a timeout occurs (lines 40–94).
6. **Metadata Extraction** – `_get_graph_info()` queries node and edge counts, summarizes distinct entity types, and returns a structured `GraphInfo` dataclass (lines 96–117).
7. **Data Export** – `get_graph_data()` retrieves complete node and edge records with properties and timestamps, assembling a JSON-serializable payload (lines 119–194).
8. **Resource Cleanup** – `delete_graph()` forwards deletion requests directly to `self.client.graph.delete()` (lines 96–99).

## Core Methods in the Graph Builder Service

The service exposes granular control over Zep’s graph management capabilities through specific method implementations that handle both data transformation and API communication.

### Creating Graphs and Defining Ontologies

When constructing a new knowledge graph, **MiroFish** first allocates the graph resource, then establishes its schema. The `create_graph()` method handles the initial provisioning, while `set_ontology()` translates Python dictionaries into Zep-compatible Pydantic models. This dynamic class generation allows runtime ontology modification without code changes, as implemented around lines 98–116 of the service file.

### Text Processing and Batch Uploads

Before ingestion, documents undergo preprocessing to meet Zep’s episode format requirements. The `TextProcessor.split_text()` function creates overlapping windows to preserve context across chunk boundaries. The `add_text_batches()` method then wraps these chunks into `EpisodeData` instances and manages the bulk upload process, accepting an optional `progress_callback` parameter for real-time status reporting during long-running uploads.

### Polling and Data Retrieval

Since Zep processes episodes asynchronously, the service implements active polling via `_wait_for_episodes()`. This method iterates through the returned UUID list, checking each episode’s processing status until completion. For data extraction, `_get_graph_info()` provides lightweight statistical summaries, while `get_graph_data()` performs a full export suitable for frontend visualization or downstream analysis pipelines.

## API Integration and Asynchronous Execution

The **Flask** backend in [`backend/app/api/graph.py`](https://github.com/666ghj/mirofish/blob/main/backend/app/api/graph.py) leverages the service within a background task framework to prevent HTTP request timeouts during graph construction.

### The /build Endpoint Orchestration

The `/build` endpoint (lines 84–120) instantiates `GraphBuilderService` with the configured API key, then delegates the construction workflow to a background thread. This pattern uses Python’s threading capabilities alongside the shared `TaskManager` to track progress across the multi-stage pipeline:

```python

# backend/app/api/graph.py – excerpt from build_graph() view

builder = GraphBuilderService(api_key=Config.ZEP_API_KEY)

def build_task():
    graph_id = builder.create_graph(name=graph_name)
    builder.set_ontology(graph_id, ontology)
    episode_uuids = builder.add_text_batches(
        graph_id, chunks, batch_size=3,
        progress_callback=add_progress_callback
    )
    builder._wait_for_episodes(episode_uuids, wait_progress_callback)
    graph_data = builder.get_graph_data(graph_id)
    # Store results and update TaskManager...

```

The endpoint immediately returns a `task_id` to the client, which can query `/task/<task_id>` for completion status while the background thread continues polling Zep.

## Implementation Examples

### Standalone Service Usage

For direct programmatic access without the REST API layer:

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

# Initialize with explicit or configured Zep key

builder = GraphBuilderService(api_key="your-zep-key")

# 1. Create graph and define schema

graph_id = builder.create_graph(name="Research Graph")
builder.set_ontology(graph_id, {
    "entity_types": [{"name": "Concept", "attributes": [{"name": "label"}]}],
    "edge_types": [{"name": "relates_to", "source_targets": [{"source": "Concept", "target": "Concept"}]}]
})

# 2. Process document

raw_text = FileParser.extract_text("document.txt")
chunks = TextProcessor.split_text(raw_text, chunk_size=500, overlap=50)

# 3. Ingest and wait

episode_uuids = builder.add_text_batches(graph_id, chunks)
builder._wait_for_episodes(episode_uuids)

# 4. Export results

data = builder.get_graph_data(graph_id)
print(f"Constructed graph with {data['node_count']} nodes")

```

### REST API Integration Pattern

When extending the Flask application, inject the service into route handlers and defer execution to background workers:

```python
from backend.app.services.graph_builder import GraphBuilderService
from backend.app.config import Config

def handle_graph_request(file_path, ontology_def):
    builder = GraphBuilderService(api_key=Config.ZEP_API_KEY)
    
    def background_build():
        graph_id = builder.create_graph(name="API Graph")
        builder.set_ontology(graph_id, ontology_def)
        # ... ingestion logic ...

        return builder.get_graph_data(graph_id)
    
    # Delegate to TaskManager/background thread

    task_id = TaskManager.submit(background_build)
    return {"task_id": task_id}

```

## Summary

- The **GraphBuilderService** in [`backend/app/services/graph_builder.py`](https://github.com/666ghj/mirofish/blob/main/backend/app/services/graph_builder.py) encapsulates all Zep Cloud SDK interactions, providing a Pythonic interface for graph lifecycle management.
- The integration handles **ontology definition** through dynamic Pydantic class generation, ensuring type-safe schema enforcement in Zep.
- **Text chunking** uses a 500-character window with 50-character overlap to maintain semantic continuity across episode boundaries.
- The `/build` endpoint in [`backend/app/api/graph.py`](https://github.com/666ghj/mirofish/blob/main/backend/app/api/graph.py) orchestrates the service asynchronously using background threads and the `TaskManager` to prevent HTTP timeouts.
- Comprehensive **polling mechanisms** via `_wait_for_episodes()` ensure reliable synchronization with Zep’s asynchronous processing pipeline before data retrieval.

## Frequently Asked Questions

### What is the role of the GraphBuilderService in MiroFish?

The **GraphBuilderService** functions as the primary integration layer between MiroFish’s Python backend and Zep Cloud. It translates high-level graph construction requests into specific Zep SDK method calls, handles authentication, manages text preprocessing, and implements polling logic to track asynchronous processing status. This service allows the rest of the application to interact with knowledge graph construction through simple method calls rather than direct HTTP API management.

### How does the service handle long-running Zep operations?

Long-running operations utilize **background thread execution** combined with active polling. The `_wait_for_episodes()` method repeatedly queries Zep’s episode status endpoint until all uploaded chunks report `processed=True` or a timeout threshold is reached. The Flask API layer (specifically in [`backend/app/api/graph.py`](https://github.com/666ghj/mirofish/blob/main/backend/app/api/graph.py)) immediately returns a task identifier to the client, allowing the HTTP connection to close while the background thread continues monitoring Zep’s progress through the service layer.

### What text processing pipeline feeds into the graph construction?

The pipeline uses the **`TextProcessor.split_text()`** static method to segment raw documents into manageable chunks. This implementation creates 500-character segments with a 50-character overlap between consecutive chunks, preserving context that might otherwise be lost at boundaries. These chunks are then wrapped in `EpisodeData` objects by `add_text_batches()` and uploaded to Zep in configurable batch sizes, optimizing network utilization while respecting API rate limits.

### How is the Zep API key configured for the service?

The service accepts the API key through two mechanisms: explicitly via the `api_key` constructor parameter, or implicitly through the **`Config.ZEP_API_KEY`** environment variable. As shown in lines 44–50 of [`backend/app/services/graph_builder.py`](https://github.com/666ghj/mirofish/blob/main/backend/app/services/graph_builder.py), the initialization logic checks for an explicit key first, then falls back to the configuration object. This pattern supports both development scenarios (explicit keys) and production deployments (environment-based configuration) without code modifications.