Event Reference and Observers¶

This document serves as the technical reference catalog and developer guide for the Episteme event system. It details all domain event schemas, available built-in observers, and patterns for emitting and consuming events.

For the conceptual system design, contextvar dispatch mechanics, and architectural guarantees, see Event System Architecture.

Base Event Schema¶

All pipeline events are Pydantic v2 models defined in pipeline/events/models.py inheriting from BaseEvent:

Event Reference Catalog¶

Entity & Maturation Events¶

Events related to entity processing, textual envelope injection, entity linking, and centroid description synthesis.

Extraction & Relation Events¶

Events capturing candidate generation, cross-encoder reranking, and LLM relation decoding.

Graph & Fusion Events¶

Events recording persistence into Neo4j and entity fusion / resolution.

Pipeline Lifecycle & Projection Events¶

Events tracking phase execution, component start/stop boundaries, chunk generation, and projection.

LLM & Inference Telemetry Events¶

Events capturing detailed latency, prompt/completion tokens, and model parameters for cost and performance tracking.

Progress, Evaluation & Validation Events¶

Events tracking progress indicators, epistemic metrics, and schema violation warnings.

Built-in Observers¶

The event system includes several built-in observers located in pipeline/events/:

How-To Guides¶

Emitting Events in Pipeline Components¶

Components retrieve the active emitter via get_event_emitter() from task-local context and emit typed event instances:

from pipeline.events import get_event_emitter
from pipeline.events.models import DenseCandidatesGenerated, RerankerScoreAssigned

class DenseRetrievalService:
    def retrieve_candidates(self, entities: list[dict], run_id: str) -> list[dict]:
        # Perform retrieval...
        candidates = [...]

        # Emit domain event:
        get_event_emitter().emit(
            DenseCandidatesGenerated(
                run_id=run_id,
                phase="phase_2_extraction",
                candidate_count=len(candidates),
                candidates=[{"pair": c["pair"], "score": c["score"]} for c in candidates[:10]],
                entities_count=len(entities),
            )
        )
        return candidates

Configuring Observers for a Pipeline Run¶

Use CompositeObserver to fan out events to multiple sinks, and bind the emitter using use_event_emitter:

from pathlib import Path
from pipeline.events import (
    SimpleEventEmitter,
    CompositeObserver,
    LoggingObserver,
    JsonlRunObserver,
    use_event_emitter,
)
from pipeline.events.langfuse_observer import LangfuseObserver

# 1. Instantiate desired observers:
observers = [
    LoggingObserver(),
    JsonlRunObserver(filepath=Path("runs/run_001/events.jsonl")),
]

# Add Langfuse if configured:
if langfuse_enabled:
    observers.append(LangfuseObserver())

# 2. Wrap in CompositeObserver and register with emitter:
emitter = SimpleEventEmitter()
emitter.register_observer(CompositeObserver(observers=observers))

# 3. Bind to task-local context for pipeline execution:
with use_event_emitter(emitter):
    result = await pipeline.run(pipeline_input)

Implementing a Custom Observer¶

To implement a new observer, subclass EventObserver and define the on_event handler:

import logging
from pipeline.events.bus import EventObserver
from pipeline.events.models import PipelineEvent, TripleCommitted

logger = logging.getLogger(__name__)

class TripleCounterObserver(EventObserver):
    """Custom observer that counts committed triples."""

    def __init__(self) -> None:
        self.count: int = 0

    def on_event(self, event: PipelineEvent) -> None:
        try:
            if isinstance(event, TripleCommitted):
                self.count += 1
        except Exception as e:
            # Observers must isolate exceptions so the pipeline is not interrupted
            logger.warning(f"Error in TripleCounterObserver: {e}")

Best Practices¶

1. Keep Payloads Compact:
2. Do not embed entire documents, giant prompt templates, or full graph states inside events.
3. Use stable identifiers (entity_id, chunk_id, document_id) and counts.
4. Ensure Timezone Awareness:
5. All events default to datetime.now(timezone.utc). Never emit naive datetime objects.
6. Isolate Failures:
7. Observers must catch their own exceptions. An observer crash must never fail the pipeline.
8. Side-Effect Only:
9. Observers must never mutate the received event instance or attempt to alter pipeline execution flow.