How to Ingest Data from Kafka Streams Using Semantica: A Complete Guide

To ingest data from Kafka streams using Semantica, instantiate the StreamIngestor class and call ingest_kafka(), or use the ingest_stream convenience method to create a KafkaProcessor, register a message handler with set_message_handler(), and invoke start_consuming() to begin polling with built-in health monitoring.

Semantica provides a flexible, extensible streaming ingestion framework for building real-time AI applications. This guide demonstrates the complete workflow to ingest data from Kafka streams using Semantica's Python API, referencing the actual implementation in the semantica-agi/semantica repository.

Core Architecture Components

StreamIngestor

The StreamIngestor class defined in semantica/ingest/stream_ingestor.py (lines 19-38) serves as the primary orchestrator for stream processing. It creates and registers stream processors and attaches a health-monitoring layer that tracks runtime metrics across all active connections.

KafkaProcessor

Defined in semantica/ingest/stream_ingestor.py (lines 25-48), KafkaProcessor is a concrete subclass of StreamProcessor that wraps a kafka.KafkaConsumer. It polls records from subscribed topics, converts them to the internal StreamMessage representation, applies optional transformations and validations, and forwards processed payloads to a user-provided handler function.

StreamMonitor

The StreamMonitor component maintains runtime metrics including processed message count, error rate, and health status for every active processor (lines 29-41). It enables integration with external observability tools by exposing structured health data through the check_health() method.

Ingestion Methods

Semantica offers two primary approaches to initialize Kafka ingestion. Choose based on your control requirements:

High-Level: ingest_stream Convenience Method

ingest_stream – Use this function from semantica.ingest.methods for rapid prototyping and simple configurations. This wrapper internally creates a StreamIngestor and instantiates the appropriate processor based on the method argument.

from semantica.ingest.methods import ingest_stream

# Define the Kafka source configuration

kafka_source = {
    "topic": "my_topic",
    "bootstrap_servers": ["localhost:9092"]
}

# Create a processor (KafkaProcessor under the hood)

processor = ingest_stream(kafka_source, method="kafka")

# Register a handler that receives the normalized message payload

def handle_message(msg: dict) -> None:
    print("Received:", msg["content"])

processor.set_message_handler(handle_message)

# Start the consumer loop

processor.start_consuming()

According to semantica/ingest/methods.py (lines 64-76), this wrapper handles the boilerplate of processor selection while exposing the same configuration options as the direct API.

Direct API: StreamIngestor Class

StreamIngestor – Use this class directly when managing multiple processors simultaneously or when you need to inject global configuration at the ingestor level. The ingest_kafka() method returns a configured KafkaProcessor instance that you can customize before starting consumption.

from semantica.ingest.stream_ingestor import StreamIngestor

# Initialise the ingestor (can accept global config here)

ingestor = StreamIngestor()

# Obtain a Kafka processor for the desired topic

kafka_proc = ingestor.ingest_kafka(
    topic="my_topic",
    bootstrap_servers=["localhost:9092"],
    # Optional: custom validation or transformation functions

    transform=lambda d: {"value": d["value"] * 2},
    validate=lambda d: isinstance(d.get("value"), int)
)

# Define a handler to process each message after validation/transformation

def process(msg: dict) -> None:
    print("Processed:", msg)

kafka_proc.set_message_handler(process)

# Start the ingestion loop (multiple processors can be started together)

ingestor.start_streaming()

The StreamIngestor.__init__ method in semantica/ingest/stream_ingestor.py (lines 41-55) accepts global configuration, while ingest_kafka (lines 64-78) supports optional transform and validate callables that execute sequentially on each message before reaching your handler. The start_streaming() method (lines 38-49) initiates consumption for all registered processors in the ingestor's pool.

Monitoring Stream Health

After initiating ingestion, query the StreamMonitor to retrieve real-time metrics and health status. This is essential for production deployments requiring observability integration.

# After starting the ingestion as in the Direct API example

import time

while True:
    health = ingestor.monitor.check_health()
    print("Overall status:", health["overall"])
    for name, stats in health["processors"].items():
        print(f"{name}: running={stats['running']}, errors={stats['errors']}")
    time.sleep(30)   # poll every 30 seconds

The check_health() method in semantica/ingest/stream_ingestor.py (lines 73-86) returns a dictionary containing an overall status string and a processors dictionary with per-processor statistics including running boolean flags and cumulative errors counts.

Summary

• The StreamIngestor class in semantica/ingest/stream_ingestor.py serves as the central orchestrator for Kafka ingestion workflows.
• Use ingest_stream from semantica.ingest.methods for quick setup, or instantiate StreamIngestor directly for fine-grained control over multiple processors and global configuration.
• The KafkaProcessor wraps the underlying Kafka client, converts records to StreamMessage objects, and supports optional transform and validate functions.
• Access runtime health metrics and error rates through StreamMonitor.check_health() to integrate with external monitoring systems.

Frequently Asked Questions

How do I start consuming messages from a Kafka topic in Semantica?

Create a processor using either ingest_stream() or StreamIngestor().ingest_kafka(), register your processing logic via set_message_handler(), and call start_consuming() on the processor or start_streaming() on the ingestor. The framework handles polling, message conversion, and error handling automatically.

What is the difference between ingest_stream and StreamIngestor?

ingest_stream is a convenience function in semantica/ingest/methods.py that instantiates a StreamIngestor internally and returns a ready-to-use processor. StreamIngestor is the underlying class that manages multiple processors and global state; use it directly when you need to coordinate several Kafka topics or share configuration across processors.

Can I apply custom transformations to Kafka messages before processing?

Yes. Both the ingest_kafka() method and the high-level ingest_stream() function accept optional transform and validate parameters. The transform function receives the raw message dictionary and must return a modified dictionary, while validate returns a boolean determining whether the message should proceed to your handler.

How does Semantica handle health monitoring for Kafka streams?

Semantica attaches a StreamMonitor instance to every StreamIngestor that tracks processed counts, error rates, and running status for each processor. Call ingestor.monitor.check_health() to retrieve a structured health report suitable for integration with Prometheus, Datadog, or other observability platforms.

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