The Worker Has Been Loaded… Now What?
In the previous article we followed a job from the Queue Manager into a newly created worker process. By this point our worker has:
Started a new Python process
Created its own asyncio event loop
Connected to MongoDB
Dynamically loaded the configured analysis pipeline
At this stage the worker is finally ready to execute the actual business logic. It might be tempting to simply instantiate the pipeline and call:
await pipeline.run()
That would certainly work. But an execution platform needs much more than simply running code.
The worker needs a way to:
These concerns are just as important as the analysis itself.
In this article we'll explore how those responsibilities are handled.
Dependency Injection Instead of Global State
One design decision that often goes unnoticed is how the analysis pipeline receives its dependencies.
Rather than importing global objects, everything required by the pipeline is injected during construction.
constructor_args = {
"survey_id": payload["survey_id"],
"question_id": payload["questionId"],
"logger": logger,
"register_thread": register_thread,
"register_process": register_process,
}
The worker knows how to construct the pipeline, but it doesn't dictate how the pipeline should behave.
This approach has several advantages.
Loose Coupling
The pipeline doesn't need to know where the logger came from. It simply receives one. The same applies to thread registration, subprocess registration, configuration, or database services.
Easier Testing
Because dependencies are injected rather than imported globally, unit tests can substitute lightweight mock implementations. Instead of starting real worker processes, tests can inject fake loggers and in-memory services.
Reusable Pipelines
Nothing inside the pipeline depends on FastAPI, Queue Manager or multiprocessing.
The pipeline could be executed from:
without modification.
This separation keeps the orchestration layer independent from the business logic.
Reporting Progress
AI workloads can easily take several minutes. Some may process thousands of comments. Others may call hundreds of external APIs. Without progress reporting the client has no visibility into what is happening.
Instead of waiting silently, the worker continuously publishes milestones back to MongoDB.
async def progress_cb(message, percentage):
await db_client.post_milestone(message, percentage)
The pipeline receives this callback during execution.
Whenever meaningful progress occurs, it simply calls
await progress_cb("Generating embeddings", 35)
or
await progress_cb("Clustering comments", 70)
The pipeline never needs to know where those updates are stored. It simply reports progress. The worker takes responsibility for persistence. This is another example of separating execution from business logic.
Why Callbacks Instead of Database Calls?
The answer is abstraction.
Imagine replacing MongoDB tomorrow with Redis, PostgreSQL or an event bus.
If every pipeline contained database code, every implementation would need to change.
Instead, the pipeline exposes intent:
"Progress has reached 60%."
The worker decides how that information should be persisted.
Keeping responsibilities separated makes the pipeline significantly easier to evolve.
Graceful Shutdown
Long-running AI jobs must also be interruptible: A deployment may require workers to stop , a watchdog may detect a stalled process or an administrator may terminate a runaway job.
If we simply kill the process immediately, that can leave:
Partially written data
Orphaned subprocesses
Unfinished threads
Inconsistent state
Instead the worker listens for operating system signals.
signal.signal(SIGTERM, handle_sigterm)
When the operating system sends SIGTERM, the handler doesn't terminate immediately.
Instead it requests a graceful shutdown.
stop_event.set()
The pipeline periodically checks this event. If shutdown has been requested it can complete its current operation, release resources and exit safely.
This gives the worker an opportunity to leave the system in a consistent state before terminating.
Cleanup Matters
One of the easiest ways to destabilise production systems is forgetting about resources created during execution.
Analysis pipelines often create:
Background threads
Subprocesses
External tools
Those resources don't automatically disappear when your analysis completes. To avoid leaks the worker keeps track of everything it creates. Whenever the pipeline starts a thread it registers it. Whenever it launches a subprocess it registers that as well. When execution finishes the worker performs one final cleanup pass.
cleanup_subtasks()
Internally this function:
Terminates child processes
Waits for graceful shutdown
Joins background threads
Clears internal registries
Only after cleanup completes does the worker exit.
This guarantees that every job leaves the system exactly as it found it.
Putting it all together
By now our worker has become much more than a background process.
It is a complete execution environment.
Worker Starts
│
▼
Load Pipeline
│
▼
Inject Dependencies
│
▼
Execute Analysis
│
▼
Report Progress
│
▼
Handle Shutdown Requests
│
▼
Cleanup Resources
│
▼
Exit
Every worker is isolated.
Every worker owns its lifecycle.
Every worker can be terminated independently.
Most importantly, the orchestration platform remains completely unaware of what the pipeline actually does.
It simply provides a safe environment in which arbitrary analysis workloads can execute reliably.
Conclusion
At this point we've completed the execution platform.
A client submits work.
FastAPI persists the request.
The Queue Manager schedules execution.
An isolated worker loads the configured pipeline, executes it safely, reports progress, and cleans up every resource before exiting.
What we haven't explored yet is the pipeline itself.
How do we process tens of thousands of comments without exceeding LLM token limits?
How do we execute embedding generation concurrently while respecting API rate limits?
How do we recover from failures without repeating expensive computation?
In the next—and final—article we'll step inside the AI pipeline itself and explore the engineering patterns behind scalable LLM-powered thematic analysis.
