In Part 6 we closed the loop on the execution platform: a worker process boots in isolation, reports progress through a callback, listens for SIGTERM, and cleans up after itself no matter how it exits. At that point we had a fully production-ready way to run arbitrary long-lived work but we never looked at what that work actually was.
It's time to open that box. Everything from here on happens inside a single method call the worker makes once it has loaded the pipeline class:
analysis: PipeLineInterface = _load_analysis(analysis_path, constructor_args)
await analysis.run(progress_cb, stop_event=stop_event)
That one call ThematicAnalysisPipeline.run() is ten stages deep. It takes a batch of raw, free-text responses and turns them into clustered, summarized, mapped themes. This post covers the first six stages: getting the right slice of responses out of MongoDB, cleaning them, and turning them into embeddings without paying to re-embed anything twice. Part 8 will pick up from there and cover clustering, LLM summarization, and the mapping stage along with the resumability logic that lets a killed job pick up mid-clustering instead of starting over.
Every stage below reports a milestone through the same progress_cb we introduced in Part 5 "data fetched", "data prune completed", "embeddings loaded", and so on. That's not just a progress bar for the frontend; each milestone also resets the watchdog's staleness clock from Part 4. A stage that takes forty minutes to embed ten thousand responses is fine. A stage that goes forty minutes without reporting anything looks exactly like a hung process, and the watchdog will kill it. Keep that in mind as we go it's why almost every stage below is chunked and batched rather than done in one giant pass.
Step 1: Fetching the Right Slice of Data
The pipeline doesn't operate on an entire dataset at once. It operates on one field within one dataset, seen through one caller's access scope. fetch_data delegates to a helper that builds a MongoDB query against the response records and pulls out just the answers for the target field:
query = {"datasetId": dataset_id}
if filters:
and_conditions = []
for field_id, values in filters.items():
and_conditions.append({
"answers": {
"$elemMatch": {
"fieldId": field_id,
"values": {"$in": values[0].split("_")}
}
}
})
query["$and"] = and_conditions
records = await db_client.fetch_many(
"responses", query, {"answers": 1, "_id": 1}
)
Two details here are easy to miss but matter a lot in practice:
Multi-value answers get split into pseudo-records. An open-ended field like "what could we improve?" often comes back as a single comma-separated blob "better pay, more flexible hours, less micromanagement." Rather than treating that as one response, the pipeline splits it and gives each fragment its own synthetic ID ({response_id}_{index}), so three distinct themes don't get flattened into one incoherent response during clustering later. The split isn't naive string-splitting on commas, either a small guard list of conjunctions (NO_SPLIT_WORDS: "and", "because", "however", "for example"...) prevents cutting a sentence like "pay is fine, but the hours are brutal" into two meaningless halves.
Hierarchy-scoped access is a BFS, not a flat lookup. When a dataset has organizational-hierarchy scoping enabled, a small resolver walks down from the caller's scope through a parent-scope field on each record, breadth-first, collecting every descendant scope before the main query runs. Someone scoped to one business unit should see that unit and everything beneath it but the query can't express "everything beneath it" directly, so the pipeline has to compute the full set of scopes first and filter against it.
The output isn't a list of strings , it's a pandas.DataFrame with response_id, response, and redacted_response columns, plus the raw set of response_ids. That set matters more than it looks: it's the key the next two stages use to avoid redoing work.
Step 2: Pruning Garbage Before It Costs You Anything
Free-text fields attract noise: "n/a", "....", "xxxxx", single characters, and the occasional string of keyboard mashing. None of that is worth an embedding call or a clustering pass, so prune_data filters it out early before any paid API call touches it.
DataFramePruner._is_meaningful runs each response through a small stack of cheap filters, ordered from cheapest to most expensive:
def _is_meaningful(self, value: str) -> bool:
v = value.strip()
if self.pattern.match(v): # "na", "n/a", "tbd", "--", single letters...
return False
if self.gibberish_pattern.match(v): # "aaaa", "!!!!", repeated chars
return False
if self.ai_detector:
result = self.ai_detector(v[:512], top_k=None)
gibberish_score = next((r['score'] for r in result if r['label'].lower() == 'noise'), 0.0)
if gibberish_score > 0.8:
return False
if self.use_nlp_filter:
meaningful_words = [t for t in tokens if t not in self.stopwords]
if len(meaningful_words) == 0: # response is *only* stopwords
return False
return True
The two regexes catch the obvious junk for free. The optional AI detector a locally-loaded gibberish-classification model, only wired up if AI_DETECTOR_PATH is set , it catches subtler noise the regexes miss.
The stopword check catches the residual case where a response is grammatically valid but semantically empty, like "it is what it is."
Pruning isn't just cosmetic it's a gate. If fewer than 50 responses survive, the pipeline stops outright and marks the run INSUFFICIENT_DATA instead of running an entire clustering pipeline on a handful of data points:
if len(df_pruned) < 50:
await self.update_state(self.dataset_id, self.field_id, "INSUFFICIENT_DATA")
return None, response_info
Key insight: every expensive stage downstream: embedding, clustering, LLM refinement costs real money and real time per response. Pruning aggressively at the top of the funnel isn't an optimization, it's what makes the economics of the rest of the pipeline work.
Step 3: Don't Pay to Embed the Same Response Twice
This is the stage I'd call the most architecturally interesting of the six, and it's easy to walk right past it. Before generating a single new embedding, the pipeline checks what it's already embedded for this exact dataset/field pair, using the response_ids collected back in Step 1:
async def fetch_with_retry(chunk):
async with semaphore:
return await self.client.fetch_many(
"embeddings_new",
{"datasetId": self.dataset_id, "fieldId": self.field_id,
"responseId": {"$in": chunk}},
{"_id": 0, "responseId": 1, "processedResponse": 1, "embedding": 1}
)
tasks = [fetch_with_retry(value_list[start:start + batch_size])
for start in range(0, len(value_list), batch_size)]
results = await asyncio.gather(*tasks)
Response IDs are chunked into batches of 1,000, fetched concurrently under a semaphore (so a dataset with fifty thousand responses doesn't fire fifty thousand-ID queries or fifty unbounded ones), with exponential backoff plus jitter on failure. The result is two dictionaries existing_responses and existing_embedding_map keyed by response ID.
split_new_existing then does exactly what it sounds like: any response already found gets routed into df_existing (its previously-computed clean text gets reattached), and everything else goes into df_new; the only rows that will actually touch the embedding API in Step 5.
This is a cache lookup, functionally, but it's worth naming as a distinct architectural decision rather than an implementation detail. Datasets get re-analyzed as new responses trickle in over time. Without this step, every re-run would re-embed every response ever collected, and embedding cost would grow without bound instead of scaling with new responses only.
Step 4: Normalizing Text for the Model
Only df_new the responses that survived pruning and aren't already embedded goes through TextPreprocessor. The interesting design choice here is that the preprocessing steps are configuration, not code:
self.step_funcs = {
"remove_placeholder": self._remove_placeholders,
"lowercase": self._lowercase,
"remove_punctuation": self._remove_punctuation,
"remove_whitespace": self._remove_whitespace,
"remove_stopwords": self._remove_stopwords,
"lemmatize": self._lemmatize,
}
PREPROCESS_CONFIG["steps"] is just a list of strings naming which functions to run and in what order:
"steps": ["remove_placeholder", "lowercase", "remove_punctuation",
"remove_whitespace", "lemmatize", "remove_stopwords"]
transform() walks that list and calls self.step_funcs[step] for each name, batch by batch. Want to drop lemmatization for a faster run, or reorder stopword removal to happen before lowercasing? That's a config change, not a code change ; the same "swap behavior without touching the class" instinct we saw with PIPELINE_CLASS_PATH dynamic loading back in Part 5, just applied one level down.
Lemmatization is the expensive step it runs every response through spaCy. The preprocessor checks for a GPU up front and falls back to spreading the work across CPU cores if theisn't one:
if self.use_gpu_if_available and torch.cuda.is_available():
spacy.require_gpu()
n_process = 1
else:
spacy.require_cpu()
n_process = multiprocessing.cpu_count() - 1
self.nlp = spacy.load(self.spacy_model, disable=["ner", "parser"])
Note the disable=["ner", "parser"] the pipeline only needs lemmas and part-of-speech tags, so named-entity recognition and dependency parsing (two of spaCy's moexpensive components) aswitched off entirely. It's a small line, but on a dataset with tens of thousands of responses it's the difference between a preprocessing pass that takes seconds and one that takes minutes.
Step 5: Generating Embeddings Without Melting the API
Only truly new responses reach this stage, and it reuses a pattern this codebase clearly likes: a thin interface with swappable providers.
class BaseEmbeddingProvider(ABC):
@abstractmethod
def embed_data(self, texts: List[str]) -> List[List[float]]:
pass
def get_embedding_provider(provider: str, model_type: str = None, **kwargs):
provider = provider.lower()
if provider == "openai":
return OpenAIEmbedding(model=model_type, **kwargs)
elif provider == "vertex":
return VertexEmbedding(model=model_type, **kwargs)
elif provider == "huggingface":
return HuggingFaceEmbedding(model_type=model_type)
elif provider == "bedrock":
return BedrockEmbedding(model_id=model_type, **kwargs)
This is the same shape as the PipeLineInterface / dynamic-import pattern from Part 5, just scoped to one concern: which embedding backend to call. Config picks openai / text-embedding-3-large today; swapping to a self-hosted HuggingFace model for cost reasons is a config change, not a rewrite of the pipeline.
The orchestration around the actual API calls is where the concurrency discipline from earlier posts shows up again this time applied to an external, rate-limited API instead of internal processes:
async def process_chunk(start_idx, end_idx):
chunk_raw = raw_responses[start_idx:end_idx]
async with self.api_semaphore:
embeddings = await asyncio.to_thread(
self.embedding_service.generate_embeddings, chunk_raw
)
async with self.db_semaphore:
await self.db_service.save_embeddings(embeddings, ..., collection_name)
return embeddings
tasks = [process_chunk(i, min(i + self.chunk_size, len(raw_responses)))
for i in range(0, len(raw_responses), self.chunk_size)]
chunk_results = await asyncio.gather(*tasks)
Responses are split into chunks of 200, and two independent semaphores bound the work: one caps how many embedding API calls are in flight at once, the other caps concurrent MongoDB writes. They're deliberately separate the API and the database have completely different capacity limits, and coupling them would mean tuning for whichever one is more fragile.
EmbeddingService.generate_embeddings adds its own retry loop with backoff on top of this so a transient rate-limit error on one chunk doesn't take down the whole batch, it just delays that one chunk's asyncio.gather slot.
Step 6: Recombining Old and New
The last stage in this post is the shortest, but it's where the two paths from Step 3 rejoin:
def combine_data_for_clustering(self, df_new_processed, df_existing, new_embedding_map, existing_embedding_map):
df_all = pd.concat([df_new_processed, df_existing], ignore_index=True)
embedding_vectors = [
new_embedding_map[rid] if rid in new_embedding_map else existing_embedding_map.get(rid)
for rid in df_all["response_id"]
]
df_all['redacted_response'] = df_all['redacted_response'].combine_first(df_all["response"])
df_all.rename(columns={"redacted_response": "response"}, inplace=True)
return df_all, embedding_vectors
Two things worth calling out. First, the embedding vector list is rebuilt by looking up each response ID in either map this is the point where "responses we just embedded" and "responses we already had embeddings for" become a single aligned dataset, ready to hand to clustering as one responses/embeddings pair. Second, and easy to miss: combine_first means the redacted version of a response wins whenever one exists, falling back to the raw response only when no redaction was recorded. Every downstream stage clustering, LLM refinement, summarization sees the PII-safe version of the text, not the original.
What to Notice Across All Six Steps
Look back across fetch → prune → dedupe → preprocess → embed → combine and a few patterns repeat on purpose:
Provider abstraction shows up twice (embeddings here, LLMs in Part 8) using the same "interface + factory function" shape as the dynamic pipeline loading from Part 5. It's the same idea applied at three different layers of the system.
Every I/O-bound stage is chunked and semaphore-bounded response ID lookups, embedding API calls, and DB writes all cap their own concurrency independently, because each has a different real-world capacity limit.
Config, not code, controls behavior preprocessing steps, batch sizes, chunk sizes, and provider choice all live in plain dictionaries in pipelines/config/, not scattered through the pipeline logic.
Cheap filters run before expensive ones regex before AI-model inference in pruning; a cache lookup before any paid API call in the embedding stage.
None of this is exotic engineering. It's the boring, disciplined kind bound your concurrency, cache what you can, gate expensive work behind cheap checks applied consistently at every stage. That discipline is exactly what makes it possible for a job to survive a SIGTERM mid-embedding and pick back up later without corrupting state or double-billing an API.
That survivability becomes the whole story in Part 8, where we cover the three stages that actually produce a "theme": clustering, LLM-based topic refinement, map-reduce summarization across an arbitrary number of themes, and the resume logic that lets the pipeline recognize "clustering already finished, just re-enter at the mapping stage" instead of starting from fetch_data again.
