Skip to content

Run Workers and Sync Jobs

Raw to Knowledge offloads long-running operations — answer generation, transcript-source synchronization, and graph synchronization — to Celery workers. The API service enqueues tasks and returns 202 Accepted immediately. Workers process tasks asynchronously from Redis queues. If no workers are running, work items will remain in GENERATING state indefinitely, transcript-source sync state will not advance, and graph projections will not update.

Task reference

Task Queue Trigger What it does
sync_due_connectors_task raw_to_knowledge_connectors Celery beat schedule Loads enabled connector sync states, checks cadence due-ness, and pulls only the sources whose watermark windows are due.
generate_answer_task raw_to_knowledge_generation POST /v1/generation/work-items/{id}/generate Assembles RAG context, calls LLM, stores GeneratedCandidate, transitions work item to PENDING_VALIDATION.
sync_graph_task raw_to_knowledge_graph POST /v1/graph/answers/{id}/sync Projects an ApprovedAnswer into Neo4j as nodes and relationships.
detect_contradictions_task raw_to_knowledge_graph POST /v1/graph/answers/{id}/detect-contradictions Queries Neo4j for conflicting subject+predicate+scope combinations and writes ConflictFlag records to PostgreSQL.

Start the connector sync worker

The connector sync worker executes scheduled transcript pulls. It requires connector credentials, PostgreSQL, and Redis in its environment.

celery -A raw_to_knowledge.infrastructure.celery_app worker \
  --queues raw_to_knowledge_connectors \
  --concurrency 2 \
  --loglevel info

Run Celery beat alongside it so the sync cycle is scheduled:

celery -A raw_to_knowledge.infrastructure.celery_app beat --loglevel info

CONNECTOR_SYNC_POLL_INTERVAL_MINUTES controls how often beat enqueues the due-check cycle. Each source system's own persisted cadence (hourly, daily, weekly) determines whether that source is actually pulled during a given cycle.

Start the generation worker

The generation worker handles LLM calls and RAG context assembly. These are I/O-bound and typically run well at 4 concurrency with gevent or the default prefork pool.

celery -A raw_to_knowledge.infrastructure.celery_app worker \
  --queues raw_to_knowledge_generation \
  --concurrency 4 \
  --loglevel info

The worker must have OPENAI_API_KEY, WEAVIATE_URL, and DATABASE_URL set in its environment. It does not need NEO4J_* variables.

Start the graph worker

The graph worker handles Neo4j projection and contradiction detection. These tasks require both PostgreSQL (source of truth) and Neo4j (projection target).

celery -A raw_to_knowledge.infrastructure.celery_app worker \
  --queues raw_to_knowledge_graph \
  --concurrency 2 \
  --loglevel info

The worker must have DATABASE_URL, NEO4J_URI, NEO4J_USERNAME, and NEO4J_PASSWORD set in its environment.

Note: You can run all queues on a single worker process during development by specifying --queues raw_to_knowledge_connectors,raw_to_knowledge_generation,raw_to_knowledge_graph. In production, keep them separate to prevent LLM latency from blocking connector sync or graph work.

Check worker health

List all currently executing tasks across all workers:

celery -A raw_to_knowledge.infrastructure.celery_app inspect active

List tasks reserved (queued but not yet started) by each worker:

celery -A raw_to_knowledge.infrastructure.celery_app inspect reserved

Ping all connected workers to confirm they are alive:

celery -A raw_to_knowledge.infrastructure.celery_app inspect ping

Trigger a manual graph sync

To force a graph sync for a specific approved answer, POST to the sync endpoint. The API enqueues sync_graph_task and returns 202 Accepted.

curl -X POST http://localhost:8000/v1/graph/answers/{answer_id}/sync

Replace {answer_id} with the UUID of the approved answer. The task will run on the next available raw_to_knowledge_graph worker.

Trigger contradiction detection

To run contradiction detection for a specific approved answer:

curl -X POST http://localhost:8000/v1/graph/answers/{answer_id}/detect-contradictions

This enqueues detect_contradictions_task. The task queries Neo4j for answers with the same subject+predicate+scope but different object_value and writes ConflictFlag records to PostgreSQL.

Circuit breaker behavior

The LLM adapter includes a CircuitBreaker that trips after 5 consecutive failures within a 60-second window. When tripped, the circuit breaker causes all subsequent LLM calls to fail immediately with CircuitOpenError rather than waiting for a timeout.

Recovery is automatic: the breaker resets 60 seconds after it opened.

Note: The circuit breaker is in-process and not backed by Redis. In a deployment with multiple generation workers, each worker maintains its own circuit breaker state independently. A circuit that has opened on one worker does not affect other workers, and vice versa. For production deployments with strict availability requirements, consider implementing a Redis-backed circuit breaker shared across workers.

To immediately recover from an open circuit, restart the affected worker. The breaker initializes in the closed state on startup.

Deployment notes

  • Workers require the same Python environment and installed package as the API service. Use the same Docker image or virtual environment.
  • Workers do not auto-discover new task modules. Task imports are resolved at worker startup from raw_to_knowledge.infrastructure.celery_app.
  • Set LOG_LEVEL=DEBUG on workers for verbose task lifecycle logging during troubleshooting.
  • In containerized deployments, set CELERY_WORKER_HIJACK_ROOT_LOGGER=false to preserve structlog formatting.