Advanced Features

Background Workers

By default, belief tracking runs as asyncio background tasks in the same event loop as your application. For production deployments that require task durability (surviving process restarts, guaranteed delivery, retry queues), you can offload tracking to Celery or RQ distributed workers.

Dispatcher Comparison

Celery Dispatcher

Constructor Parameters

from celery import Celery
from beliefstate import BeliefTracker, TrackerConfig
from beliefstate.dispatcher import CeleryDispatcher

celery_app = Celery("tasks", broker="redis://localhost:6379/0")

tracker = BeliefTracker(
    config=TrackerConfig(),
    adapter=app_adapter,
    dispatcher=CeleryDispatcher(celery_app=celery_app)
)

RQ (Redis Queue) Dispatcher

Constructor Parameters

from redis import Redis
from rq import Queue
from beliefstate.dispatcher import RQDispatcher

queue = Queue("belief-tasks", connection=Redis())

tracker = BeliefTracker(
    config=TrackerConfig(),
    adapter=app_adapter,
    dispatcher=RQDispatcher(queue=queue)
)

Worker Setup

In your worker process, register the global tracker so tasks can execute:

from beliefstate.dispatcher import register_global_tracker
from my_app import tracker

# Register BEFORE processing any tasks
register_global_tracker(tracker)

Resilience

BeliefState wraps internal adapter calls with production-grade resilience patterns to handle API failures gracefully. These patterns ensure that transient provider outages don't silently drop beliefs or cascade failures into your application.

Retry with Exponential Backoff

All adapter calls (generate, embeddings) automatically retry transient failures with jittered exponential backoff. Configure via TrackerConfig (global) or per-adapter RetryConfig (adapter-level). Transient errors include rate limits (HTTP 429), server errors (5xx), timeouts, and connection errors. Permanent errors (invalid API key, malformed request) are raised immediately.

Circuit Breaker

When an LLM API goes down, the circuit breaker pattern prevents cascading failures by fast-failing requests instead of queuing retries indefinitely:

config = TrackerConfig(
    enable_circuit_breaker=True,
    circuit_breaker_failure_threshold=5,     # Trip after 5 consecutive failures
    circuit_breaker_recovery_timeout=30.0,  # Wait 30s before trying recovery
)

# Catch circuit breaker open state
from beliefstate import CircuitBreaker, CircuitBreakerOpenException

try:
    result = await tracker.track(...)
except CircuitBreakerOpenException:
    print("Circuit breaker is OPEN — failing fast")

Health Checks

Use health checks to verify store and adapter connectivity at application startup or in readiness probes (Kubernetes /healthz endpoint):

# Check full tracker health (store + adapter)
health = await tracker.health_check()
print(health)  # {"store": True, "adapter": True}

# Check individual adapter
is_healthy = await adapter.health_check()
if not is_healthy:
    logger.error("Provider is not responding — use fallback")

Observability

BeliefState includes built-in OpenTelemetry instrumentation for distributed tracing and metric collection, plus structured JSON logging for cloud log aggregators.

OpenTelemetry Setup

setup_otel() Parameters

from beliefstate.observability import setup_otel

# Initialize OpenTelemetry tracing and metrics
setup_otel(
    enabled=True,
    service_name="my-app",
    otel_exporter_otlp_endpoint="http://localhost:4317"
)

Distributed Tracing Spans

BeliefState creates hierarchical spans that trace the full lifecycle of each belief tracking task:

Exported Metrics

Structured Logging

All components emit structured JSON log events via the TrackerEvent dataclass. These events are designed for cloud log aggregators like Datadog, CloudWatch, and Loki.

TrackerEvent Fields

{"session_id": "user_123", "operation": "extract_beliefs", "turn": 5, "detail": "Extracted 3 beliefs", "latency_ms": 142.5}
{"session_id": "user_123", "operation": "detect", "turn": 5, "detail": "Found 1 contradiction"}
{"session_id": "user_123", "operation": "resolve", "turn": 5, "detail": "Overwrote 1 belief", "latency_ms": 8.2}

from beliefstate.logging_utils import TrackerEvent, log_event

# Emit a custom tracker event
event = TrackerEvent(
    session_id="user_123",
    operation="custom_check",
    turn=5,
    latency_ms=42.0,
    detail="Ran custom validation"
)
log_event(event)

GDPR & Privacy

BeliefState supports full session deletion with auditable receipts for GDPR compliance (right to erasure, Article 17). The clear_session() method ensures complete data removal including draining any in-flight background tasks.

# Delete all beliefs for a user (GDPR right to erasure)
receipt = await tracker.clear_session("user_123")

print(receipt)
# DeletionReceipt(
#   session_id="user_123",
#   beliefs_deleted=42,
#   in_flight_tasks_drained=2,
#   deleted_at=datetime(...)
# )

DeletionReceipt Fields

Deletion Workflow

1. Drain in-flight tasks: The dispatcher cancels or awaits all active tracking tasks for the session.
2. Delete from store: All beliefs for the session are permanently removed from the store (SQLite DELETE, Redis DEL).
3. Clean up locks: Session-specific async locks, turn counters, and provider tracking entries are removed from memory.
4. Generate receipt: A DeletionReceipt is returned with the count of deleted records and timestamp.

Best Practices

• Store receipts: Log or persist DeletionReceipt objects as audit evidence for GDPR compliance reviews.
• Redis TTL: For Redis stores, combine clear_session() with set_session_ttl() to auto-expire sessions that aren't explicitly deleted.
• Celery/RQ workers: The drain mechanism works across distributed workers — it signals the task queue to cancel pending tasks for the session.

Graceful Shutdown

BeliefTracker tracks all in-flight background tasks and provides a shutdown() method to drain them before process exit. This ensures no beliefs are lost during deployment restarts.

# Drain all pending background tasks before exit
await tracker.shutdown(grace_seconds=10.0)

# Or use as a context manager
async with tracker:
    await tracker.track(...)

Shutdown Workflow

1. Stop accepting new tasks: The tracker stops dispatching new background tasks.
2. Drain in-flight tasks: Waits up to grace_seconds for all pending tasks to complete.
3. Cancel remaining: Any tasks still running after the grace period are cancelled.
4. Close store: The store connection is closed (SQLite file handle, PostgreSQL pool, etc.).

Concurrency Model

BeliefState uses a multi-layered concurrency model to coordinate background writes safely.

Per-Session Async Locks

Each session gets a dedicated asyncio.Lock that serializes background belief writes within that session. This prevents two concurrent extraction tasks from racing on the same belief triple.

# Internal: per-session lock ensures ordered writes
# Task A and Task B for the same session_id will execute sequentially
# Task C for a different session_id runs concurrently

Turn-Based Optimistic Concurrency

Every upsert() call includes the current turn number. The store compares it against the stored belief's turn and only overwrites if the incoming turn is newer or equal. This is a lightweight guard against stale writes from background tasks that complete out of order.

Task Dispatchers

Background tasks are dispatched via a pluggable dispatcher. The default AsyncioDispatcher wraps each task in an asyncio.Task and tracks it for shutdown draining. The tracker's _dispatch() method adds each task to a tracked set and removes it on completion.

Hedging & Deduplication

BeliefState applies two layers of filtering before storing new beliefs: hedging-based confidence calibration and exact duplicate detection.

Hedging Pattern Calibration

After extraction, calibrate_confidence() scans the source quote for hedging language and caps the confidence score:

Hypothetical beliefs (is_hypothetical=True) are excluded from prompt injection to avoid polluting the LLM context with unconfirmed claims.

Exact Duplicate Check

Before running the expensive embedding similarity pipeline, the detector performs an O(1) exact duplicate check via store.get_by_key(). If a belief with the same subject, predicate, and session already exists and the normalized values match, the new belief is silently skipped — no embedding, no NLI, no LLM call.

# Step 0: O(1) exact check — zero LLM/embedding cost
existing = await store.get_by_key(subject, predicate, session_id)
if existing and normalize_value(existing.value) == normalize_value(new_belief.value):
    return  # skip — exact duplicate

Deduplication Pipeline

The full detect_with_deduplication() pipeline runs these steps for each new belief:

1. Exact duplicate check — O(1) key lookup, NFKC value comparison.
2. Embedding dimension guard — Skips vector comparison if old and new embeddings have different dimensions (model upgrade scenario).
3. Cosine similarity gate — Fast approximate check against stored embeddings.
4. NLI judgment — Full natural language inference on cosine-similar candidates to classify as entailment, contradiction, or neutral.