Background Workers

By default, tracking runs as asyncio background tasks. For production durability, use Celery or RQ dispatchers.

Celery Dispatcher

pythonfrom 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

pythonfrom 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:

worker_startup.pyfrom beliefstate.dispatcher import register_global_tracker
from my_app import tracker


register_global_tracker(tracker)

⚠️ Important: ContextVar (session_context) does NOT propagate across process boundaries. The dispatchers explicitly serialize session_id into task payloads to ensure it reaches worker processes.

Resilience

BeliefState wraps internal adapter calls with production-grade resilience patterns:

Retry with Exponential Backoff

All adapter calls (generate, embeddings) automatically retry transient failures with jittered exponential backoff. Configure via TrackerConfig or per-adapter RetryConfig.

Circuit Breaker

When an LLM API goes down, the circuit breaker pattern prevents cascading failures:

pythonconfig = 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
)


from beliefstate import CircuitBreaker, CircuitBreakerOpenException

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

ℹ️ Circuit Breaker States: The circuit breaker operates as a state machine:

  • CLOSED: Normal state. All requests flow through. If consecutive failures exceed the threshold (default 5), the circuit trips to OPEN.
  • OPEN: Trips blockingly. All requests immediately raise CircuitBreakerOpenException without hitting the LLM provider, saving rate limits and latency.
  • HALF-OPEN: Initiated once the recovery cooldown timeout (default 30 seconds) has elapsed. The next request acts as a probe: if it succeeds, the breaker resets to CLOSED; if it fails, it returns to OPEN for another cooldown cycle.

Health Checks

python
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.

pythonfrom beliefstate.observability import setup_otel


setup_otel(
    enabled=True,
    service_name="my-app",
    otel_exporter_otlp_endpoint="http://localhost:4317"
)

ℹ️ Traces & Metrics Breakdown:

  • Distributed Tracing Spans: Captures hierarchical execution traces with performance metadata:
    • beliefstate.extract_beliefs: Measures fact extraction prompt latency and token usage.
    • beliefstate.detect_contradiction: Tracks time taken by semantic filters and the NLI logic check.
    • beliefstate.resolve_beliefs: Tracks store write times and resolution conflicts.
  • Exported Metrics:
    • beliefstate.beliefs_extracted: Counter for total facts parsed.
    • beliefstate.contradictions_detected: Counter for triggered logical conflicts.
    • beliefstate.beliefs_in_store: Gauge showing current active facts per session database.
    • beliefstate.*_latency_ms: Histograms tracking latency percentiles (p50, p90, p99) for database operations and provider calls.

Structured Logging

All components emit structured JSON log events via logging_utils.TrackerEvent for cloud collectors (Datadog, CloudWatch):

log output{"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"}

GDPR & Privacy

BeliefState supports full session deletion with audit receipts for GDPR compliance:

python
receipt = await tracker.clear_session("user_123")

print(receipt)

⚠️ Concurrency & Deletion Race Conditions: If a user requests account deletion while background belief extraction tasks are still running in the asyncio event loop, those tasks could complete and write new facts after the database is cleared. To prevent this, clear_session() calls the dispatcher's drain_session(session_id) method first, which cancels or waits for all active tracking tasks for that session ID before executing the SQL or Redis delete commands.

The DeletionReceipt confirms:

  • All stored beliefs were deleted from the store
  • All in-flight background tasks for the session were drained/completed
  • An auditable timestamp of when deletion occurred