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

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

Constructor Parameters

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

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
)

# 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):

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

pythonfrom 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

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"}
{"session_id": "user_123", "operation": "resolve", "turn": 5, "detail": "Overwrote 1 belief", "latency_ms": 8.2}

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

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