In Part 7 we followed a batch of raw text responses through fetching, pruning, deduplication, preprocessing, and embedding, and ended up with one clean dataset: every response paired with its vector, old and new responses merged into a single frame ready for the next stage. This post finishes the pipeline. We'll cluster those responses into themes, summarize an arbitrary number of themes into one coherent overview, map each theme onto a fixed set of categories, and then look at the part that ties the whole thing together: how the pipeline knows exactly where to pick up if it gets killed halfway through any of this.
If you've been following the series since Part 1, this is where it all pays off. Everything we built, the thin API that just persists a job and returns, the queue manager that respects concurrency limits, the worker that survives a SIGTERM, exists to give this code room to run for as long as it needs without anyone worrying about the HTTP request that kicked it off.
Step 7: Turning Responses into Themes
Clustering takes four inputs: cleaned responses (for the model to find structure in), raw responses (for humans and the LLM to actually read later), response ids, and the embedding vectors from Part 7. The heavy lifting is done by BERTopic, which itself leans on two other libraries doing very different jobs. UMAP takes the high-dimensional embedding vectors and projects them down into a smaller space where distances are more meaningful for clustering. HDBSCAN then finds dense regions in that reduced space and calls each one a topic.
What's worth paying attention to isn't BERTopic itself, it's how much of its behavior is exposed as plain configuration rather than buried in code:
umap_params = {
"n_neighbors": 20,
"n_components": 10,
"min_dist": 0.03,
"metric": "cosine",
}
hdbscan_params = {
"min_cluster_size": 5,
"metric": "euclidean",
"cluster_selection_method": "eom",
}
Whether to override BERTopic's defaults at all, whether to reduce the number of topics afterward with hierarchical merging, whether to steer clustering with a predefined set of seed categories: all of that lives in one config dictionary. Nobody has to touch the clustering code itself to tune how granular the resulting themes are.
One detail that's easy to skip past: HDBSCAN doesn't force every point into a cluster. Anything that doesn't fit a dense region gets labeled topic -1, effectively "noise." Left alone, that would mean some responses just vanish from the analysis. The pipeline explicitly reassigns those outliers to their nearest real topic afterward, so every response that made it this far ends up counted somewhere.
if -1 in topics:
topics = self.topic_model.reduce_outliers(cleaned_responses, self.topic_model.topics_)
Clustering gives you groups of responses and a ranked list of keywords per group. It doesn't give you a readable theme name, a summary, or a sentiment breakdown. That's the next job, and it's handled per topic rather than all at once:
async def process_topic(self, topic_id, topic_subset, keywords_scores, topic_name, dataset_id, field_id):
responses = topic_subset["raw_response"].tolist()
ids = topic_subset["response_id"].tolist()
keywords = [kw for kw, _ in keywords_scores]
refined_data, sentiment_map = await self.llm_refiner_service.refine_topic(
keywords, responses, ids, field_id
)
theme_id, _ = await self.db_service.store_topic_results(
dataset_id, field_id, topic_name, refined_data, keywords_scores, sentiment_map, topic_id
)
return {"theme_id": theme_id, "theme_name": refined_data.get("name", "")}
refine_topic batches the responses in a topic by token budget (using the same batching helper we'll see again in the next section), asks the LLM to name the theme, write a summary, suggest actions, and tag sentiment per response, then merges those pieces back together if the topic was too large for one prompt.
Here's the part that actually matters for reliability. Before any of that LLM refinement happens, the raw topic (its id, keywords, and the response ids in it) gets written to storage and flagged as unrefined:
async def store_raw_topic(self, dataset_id, field_id, topic_id, topic_name, keywords, response_ids):
doc = {
"datasetId": dataset_id,
"fieldId": field_id,
"topicId": topic_id,
"topicName": topic_name,
"keywords": keywords,
"responseIds": response_ids,
"refined": False,
}
return await self.db.insert_one("rawThemes", doc)
Only once refinement for a given topic finishes does that flag flip to True, alongside the finished theme document. And before doing any new clustering at all, the pipeline checks whether unrefined topics from a previous, interrupted run already exist:
unrefined = await self.db_service.get_unrefined_topics(dataset_id, field_id)
if unrefined:
# skip clustering entirely, resume refinement on just these topics
...
Key insight: clustering the whole dataset is expensive and, more importantly, non-deterministic between runs (UMAP and HDBSCAN both involve randomness). If a worker died while refining topic 14 out of 20, you don't want to re-cluster from scratch and risk getting a different set of topics entirely. You want to pick up exactly where you left off, on the exact topics that already exist. Persisting the raw clustering result immediately, before spending a single token on refinement, is what makes that possible.
Step 8: Summarizing Any Number of Themes into One
Once every theme has its own name and summary, the pipeline needs a single overall summary across all of them. The problem is that a dataset might produce six themes or six hundred, and there's no way to know that number ahead of time, or to fit six hundred summaries into one prompt.
The fix is a small utility that both the clustering stage and this stage lean on: a token-aware batcher built on tiktoken.
class PromptBatcher:
def __init__(self, model_name="gpt-4.1"):
self.encoder = tiktoken.encoding_for_model(model_name)
def count_tokens(self, text):
return len(self.encoder.encode(text))
def batch_dynamic_content(self, dynamic_items, prompt_template, dynamic_placeholder_name,
constant_placeholders={}, max_total_tokens=7500):
prompt_tokens = self.count_tokens(prompt_template.format(**constant_placeholders, **{dynamic_placeholder_name: ""}))
max_batch_tokens = max_total_tokens - prompt_tokens
batches, current_batch, current_tokens = [], [], 0
for item in dynamic_items:
item_tokens = self.count_tokens(item)
if current_tokens + item_tokens > max_batch_tokens:
batches.append(current_batch)
current_batch, current_tokens = [item], item_tokens
else:
current_batch.append(item)
current_tokens += item_tokens
if current_batch:
batches.append(current_batch)
return batches
It measures the fixed prompt template's token cost once, then greedily packs as many summaries as will fit under the remaining budget, starting a new batch whenever the next item would push it over.
Summarization then becomes a straightforward map, then reduce, then reduce again until there's nothing left to reduce:
initial_batches = batcher.batch_dynamic_content(summaries, prompt_template, "summaries_list", max_total_tokens=7500)
partial_summaries = await asyncio.gather(*[call_llm(batch) for batch in initial_batches])
while True:
merge_batches = batcher.batch_dynamic_content(partial_summaries, prompt_template, "summaries_list")
if len(merge_batches) == 1:
final_summary = await call_llm(merge_batches[0])
break
partial_summaries = await asyncio.gather(*[call_llm(batch) for batch in merge_batches])
Six themes might fit in one batch and finish in a single round. Six hundred themes get summarized in groups, and those group summaries get merged together in groups, and so on, until the merge step produces exactly one batch and one final summary comes out. The number of rounds adapts automatically to however many themes actually exist, without anyone having to guess a batch size up front.
Step 9: Mapping Themes onto a Fixed Taxonomy
Raw themes and a summary are useful, but the product this pipeline feeds also wants every theme sorted into a small, fixed set of higher-level categories defined once for the whole system, not per analysis. That mapping is the last LLM-driven stage.
Themes go through the same token-aware batching as before, and each batch gets sent to the LLM as a single structured request asking it to score every theme against every category:
result = await asyncio.to_thread(
mapper_service.llm.generate, prompt, parse_json=True,
response_format={"type": "json_object"}
)
for theme_id, theme_data in result.get("results", {}).items():
step_scores = theme_data.get("step_scores", {})
if not step_scores:
continue
best_category = max(step_scores.items(), key=lambda x: x[1])[0]
mapped_themes.add(theme_id)
await db_service.upsert_theme_step(dataset_id, field_id, best_category, step_scores, theme_id)
Each theme lands in exactly one category, whichever one it scored highest against. What happens next is the part worth noticing:
missing_theme_ids = all_theme_ids - mapped_themes
if missing_theme_ids:
return missing_theme_ids
If a theme doesn't come back with a valid score (a malformed LLM response, a theme the model skipped, anything), it isn't quietly dropped. The pipeline tracks exactly which theme ids didn't make it and treats the whole mapping step as incomplete until they do. That set of leftover ids becomes the input to the resume logic covered next.
Step 10: A Pipeline That Remembers Where It Left Off
Everything above assumes a clean run from start to finish. In practice, workers get killed, machines restart, and jobs get re-enqueued. The last piece of this pipeline is the logic that decides, every time it starts, whether it actually needs to start from scratch at all.
Before touching any data, run() checks the persisted record for this exact dataset, field, and caller scope. If it already reached a finished state, the pipeline exits immediately and does nothing:
existing_record = await self.client.fetch_one("analysisResults", {...})
if existing_record and existing_record.get("state") == "DATA_READY":
return
If the analysis isn't finished, the pipeline looks at the last meaningful checkpoint it reached, not the last milestone in general, but specifically the last one that represents a safe place to resume from:
relevant_checkpoints = ["overall summary generated", "clustering completed", "data combined"]
for milestone in reversed(job_doc["milestones"]):
if milestone["message"] in relevant_checkpoints:
return milestone["message"]
That single value decides which of three paths the run takes:
if last_checkpoint == "clustering completed":
await self.resume_after_clustering() # themes exist, jump to summarization
return
elif last_checkpoint == "overall summary generated":
await self.resume_after_overall_summary() # summary exists, jump to mapping
return
# otherwise: run everything from fetch_data onward
resume_after_clustering loads the theme documents that already exist in storage and picks up exactly at summarization, skipping fetching, pruning, embedding, and clustering entirely. resume_after_overall_summary goes one step further: it loads both the themes and whatever mapping records already exist, works out which theme ids are still missing from the mapping (the same kind of set difference we saw at the end of Step 9), and maps only those.
There's one more detail worth calling out, because it's a decision that's easy to get wrong. The embedding cache from Part 7 is keyed only by dataset, field, and response id. It's shared freely across every caller, because raw text and its embedding don't depend on who's asking. Themes, summaries, and mappings are different: which responses a given caller is even allowed to see can differ from one caller to the next, which means the themes that come out of clustering can legitimately differ too. Every one of those derived records, not the embeddings, but everything downstream of clustering, is keyed by dataset, field, and a hash of the caller's access scope. Two callers analyzing the same field never share or overwrite each other's themes, even though they're drawing from the same underlying embedding cache.
Key insight: resumability here isn't one mechanism, it's the same idea applied at three different granularities. The watchdog from Part 4 decides whether a worker process is still alive. The top-level checkpoint logic in this section decides which pipeline stage to resume from. And the unrefined/refined flag from Step 7 decides which individual clustering topics still need work. None of these three layers know about each other, and none of them need to. Each one just answers the same question, "how much of the work already happened," at its own scale.
Closing the Loop
Eight parts ago, this series started with a simple observation: an HTTP endpoint has no business blocking on work that takes minutes or hours. Everything since then has been one long answer to that problem, worked out one layer at a time. A thin API that only ever accepts and persists. A queue manager that claims work within limits it actually enforces. A listener loop that turns a database row into a running process. A worker that survives being told to stop mid-task. And now, finally, the pipeline itself: fetching only what's needed, never paying twice for the same embedding, clustering and refining in a way that can be interrupted and picked back up, and a state machine at the very top that always knows exactly how much of the job is already done.
None of the individual pieces are exotic. What makes the whole thing work is that the same handful of principles, bound your concurrency, treat the database as the source of truth, checkpoint before doing expensive work, show up again and again at every layer, from the FastAPI route all the way down to a single clustering topic. That consistency is the actual architecture. Everything else is implementation detail.
Thanks for following along through the whole series.
