Kafka Producers and Consumers

The Context Graph uses Kafka for streaming agent thinking events and decision events between services. The ContextGraphKafkaConsumer subscribes to event topics and persists them to Dgraph, while the producer streams events from the capture service for downstream analytics.

Overview

Kafka integration enables decoupled, asynchronous processing of context graph events. Thinking traces can be captured at the orchestrator level and consumed by multiple downstream services without blocking the agent execution path.

Source: data-plane/ai-service/src/context_graph/integration/kafka_consumer.py

Topics

Consumer Configuration

The consumer uses manual commit mode for at-least-once delivery:

consumer = ContextGraphKafkaConsumer(
    dgraph_store=store,
    config=kafka_config,
    group_id="context-graph-dgraph-consumer",
)
 
await consumer.start()

Event Processing

The consumer processes events by type:

1. Thinking Trace Events -- Upserts traces to Dgraph with all steps and metadata
2. Decision Events -- Stores decisions in the bi-temporal store with actor chain
3. Trace Events -- Updates trace status and completion metadata

Failed messages are logged with the error details and skipped (at-least-once semantics with manual retry).

Producer Integration

The producer is integrated into the AgentThinkingCaptureService and streams events as they are captured:

service = AgentThinkingCaptureService(
    dgraph_store=store,
    kafka_producer=producer,
)

Events are published asynchronously without blocking the agent execution path.

Feature Flag Control

Kafka integration is controlled by the context_graph_kafka feature flag:

Monitoring