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

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

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

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:

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

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