How to Extend Semantica with Custom Ingestors: A Developer's Guide

Extending Semantica with custom ingestors requires subclassing the abstract Ingestor base class (or the concrete DBIngestor for relational sources), implementing the ingest_table() and ingest_query() methods, and registering the class in semantica/ingest/registry.py to enable discovery via the factory function.

The open-source Semantica framework provides a modular ingestion layer for building knowledge graphs from diverse data sources. While the platform ships with built-in connectors for common databases and APIs, its plugin-based architecture makes it straightforward to extend Semantica with custom ingestors for proprietary or specialized systems. This guide demonstrates the implementation pattern using the actual source code structure from the semantica-agi/semantica repository.

Understanding the Ingestor Architecture

Semantica’s data ingestion layer is built around a small set of core abstractions that enforce consistent interfaces across all data sources.

The Base Ingestor Class

All ingestors inherit from the abstract base class defined in semantica/ingest/api_ingestor.py. This class establishes the public API contract that the rest of the platform expects, including methods like ingest_table(), ingest_query(), and ingest_directory(). These methods must return either a list of Document objects or a pandas DataFrame for tabular sources.

The Database Ingestor Abstraction

For relational data sources, Semantica provides DBIngestor in semantica/ingest/db_ingestor.py. This concrete class implements generic helper methods—most notably _rows_to_documents()—that handle connection management, schema introspection, and the conversion of raw row dictionaries into the unified Document model. Subclassing DBIngestor allows you to reuse this boilerplate while only implementing source-specific connection logic.

Step-by-Step Guide to Creating a Custom Ingestor

Follow this four-step workflow to add a new data source to the framework.

1. Subclass the Base Ingestor

Create a new module under semantica/ingest/ and inherit from the appropriate base class. For relational databases, extend DBIngestor; for other sources, extend the generic Ingestor.

from semantica.ingest.db_ingestor import DBIngestor
from semantica.ingest.api_ingestor import Document

class MySQLIngestor(DBIngestor):
    """Ingests data from a MySQL database."""
    def __init__(self, host: str, user: str, password: str, database: str, **kwargs):
        super().__init__(**kwargs)
        self._conn_cfg = {
            "host": host,
            "user": user,
            "password": password,
            "database": database,
        }

2. Implement Required Methods

You must implement at minimum ingest_table() and/or ingest_query(). These methods handle data retrieval and must return standardized Document objects.

    def _get_connection(self):
        """Create a new MySQL connection on demand."""
        import mysql.connector
        return mysql.connector.connect(**self._conn_cfg)

    def ingest_table(self, table_name: str, limit: int | None = None) -> list[Document]:
        """Pull an entire table and return a list of Document objects."""
        conn = self._get_connection()
        cursor = conn.cursor(dictionary=True)
        sql = f"SELECT * FROM `{table_name}`"
        if limit:
            sql += f" LIMIT {limit}"
        cursor.execute(sql)
        rows = cursor.fetchall()
        conn.close()
        # Convert rows → Document using inherited helper

        return self._rows_to_documents(
            rows, 
            source=f"{self._conn_cfg['database']}.{table_name}"
        )

    def ingest_query(self, query: str, params: dict | None = None) -> list[Document]:
        """Execute an arbitrary SELECT query and return Documents."""
        conn = self._get_connection()
        cursor = conn.cursor(dictionary=True)
        cursor.execute(query, params or {})
        rows = cursor.fetchall()
        conn.close()
        return self._rows_to_documents(rows, source="custom_query")

3. Register in the Ingestor Registry

Add your class to the INGESTOR_REGISTRY dictionary in semantica/ingest/registry.py. This maps a short name (e.g., "mysql") to your class, enabling the factory function from_name() (also exposed as get_ingestor()) to instantiate your ingestor dynamically.

# semantica/ingest/registry.py

INGESTOR_REGISTRY = {
    # built-in ingestors …

    "mysql": MySQLIngestor,   # ← newly added entry

}

4. Expose in Package Init

Import the class in semantica/ingest/__init__.py so users can access it directly from the package root.

# semantica/ingest/__init__.py

from .mysql_ingestor import MySQLIngestor

__all__ = [
    # …other ingestors…

    "MySQLIngestor",
]

How the Factory Pattern Resolves Ingestors

When a user calls get_ingestor("mysql", host="...", ...), Semantica consults the INGESTOR_REGISTRY in semantica/ingest/registry.py to resolve the name, constructs the class with the provided keyword arguments, and returns the instance. This factory approach ensures that custom ingestors behave exactly like native ones—returning standardized Document lists that the downstream pipeline (splitting, semantic extraction, and knowledge graph construction) can process without modification.

Summary

• Extend the correct base class: Use DBIngestor from semantica/ingest/db_ingestor.py for relational sources, or the abstract Ingestor from semantica/ingest/api_ingestor.py for other types.
• Return standardized types: ingest_table() and ingest_query() must return lists of Document objects (or pandas DataFrames).
• Register statically: Add your class to INGESTOR_REGISTRY in semantica/ingest/registry.py to enable factory-based discovery.
• Expose publicly: Import the class in semantica/ingest/__init__.py for clean public APIs.
• Leverage helpers: Reuse _rows_to_documents() and connection pooling from DBIngestor to minimize boilerplate.

Frequently Asked Questions

What base class should I use when extending Semantica with custom ingestors for databases?

Subclass DBIngestor from semantica/ingest/db_ingestor.py. This class provides _rows_to_documents() and other helpers for connection handling and schema introspection, allowing you to focus only on source-specific connection logic rather than reimplementing the row-to-Document conversion.

Can I register a custom ingestor without modifying registry.py?

Yes, because INGESTOR_REGISTRY is a runtime dictionary, you can dynamically add entries by importing the registry in your application code and adding your class directly. However, static registration in semantica/ingest/registry.py is the recommended pattern for packaged extensions that you intend to share or maintain long-term.

What return type is required for ingest_table() and ingest_query()?

Both methods must return a list of Document objects defined in semantica/ingest/api_ingestor.py, or a pandas DataFrame for tabular sources. The Document model standardizes text content, metadata, and provenance information so that downstream components like the semantic extractor and graph builder can process data uniformly regardless of the original source.

How does the factory function locate my custom ingestor?

The from_name() function (exposed as get_ingestor()) looks up the provided string key in the INGESTOR_REGISTRY dictionary located at semantica/ingest/registry.py. If your class is properly registered under that key, the factory instantiates it with any additional keyword arguments and returns the configured ingestor ready for use.

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