Chaining Activities — from text to vectors

This is a guest post written by Houman Kargaran, Engineering Lead at ANZ Bank

The full working code is available at https://github.com/houmanka/PII-Compliant-RAG.

In the previous blog, we covered how IngestFileActivity enforces the PII boundary by keeping the redaction, classification, and storage inside a single Activity. The only thing that leaves that Activity is a file_id — an integer pointing to a batch of redacted complaints in Postgres.

From that point, the Workflow has everything it needs to continue. This article covers what happens next: embedding those complaints using an on-premises model, caching the vectors, pushing them to a vector store, and cleaning up — all as a chain of discrete Temporal Activities.

It might sound a bit complicated, but I will try to explain it in as much detail as possible.

The Workflow orchestration#

The full chain is defined in ComplaintWorkflow. Each step awaits an Activity call. Temporal records the sequence and can retry failed Activities according to their retry policies:

@workflow.defn(name="ComplaintWorkflow")
class ComplaintWorkflow:
   @workflow.run
   async def run(self, file_input: FileInput) -> bool:
       file_id = await workflow.execute_activity(
           IngestFileActivity.ingest_file_activity,
           FileDetails(path=file_input.path, provider=file_input.provider),
           start_to_close_timeout=timedelta(seconds=120),
       )

       embedding_result: EmbeddingActivityResult = await workflow.execute_activity(
           EmbeddingActivity.embedding_activity,
           file_id,
           start_to_close_timeout=timedelta(seconds=120),
       )

       vector_storage: VectorStorageActivityResult = await workflow.execute_activity(
           VectorStorageActivity.store_vector,
           embedding_result.cache_id,
           start_to_close_timeout=timedelta(seconds=120),
           retry_policy=retry()
       )

       await workflow.execute_activity(
           IngestFileActivity.update_embedded_records,
           file_id,
           start_to_close_timeout=timedelta(seconds=120),
       )

       await workflow.execute_activity(
           VectorStorageActivity.query_vector,
           embedding_result.cache_id,
           start_to_close_timeout=timedelta(seconds=120),
       )

       await workflow.execute_activity(
           CacheActivity.delete_cache,
           embedding_result.cache_id,
           start_to_close_timeout=timedelta(seconds=120),
       )

       return True

I keep vector_storage here so it is easy to inspect or log while debugging.

The Workflow keeps the control flow deterministic, while the external work happens inside Activities. Here, each Activity handles a discrete operation, and the Workflow sequences those operations.

EmbeddingActivity — on-premise embeddings#

If you recall the functional requirement, embedding generation stays within the infrastructure. That rules out external embedding APIs. Instead, for this blog, we use all-MiniLM-L6-v2, a sentence-transformer model that runs entirely on-premises.

The EmbeddingActivity fetches the unembedded complaints from Postgres using the file_id, embeds them, and caches the result in Redis.

Why Redis? This client already had an established caching layer, so we reused it here. Another option is the Claim Check pattern, which keeps larger payloads outside Temporal’s Event History and passes a lightweight reference through the Workflow instead. In this implementation, Redis serves that external storage role for the generated vectors.

Reference: https://docs.temporal.io/ai-cookbook/claim-check-pattern-python

@activity.defn
async def embedding_activity(self, file_id: int) -> EmbeddingActivityResult:
   complaint_list = self.data_store.fetch_unembedded(file_id)
   unique_cache_key = await self.get_a_unique_cache_key(file_id)
   dragonfly_key = f"vectors:{unique_cache_key}"

   vectored_ready = [complaint.text_redacted for complaint in complaint_list]
   embedded_text = self.embedding_provider.embed_texts(vectored_ready)

   cache_with_case = [
       (complaint.case_id, vector, complaint.classification.name)
       for complaint, vector in zip(complaint_list, embedded_text)
   ]

   self.cache_provider.create(dragonfly_key, cache_with_case)
   return EmbeddingActivityResult(file_id=file_id, cache_id=dragonfly_key)

The embedding model follows the same provider pattern we use for storage, with an EmbeddingProvider contract and all-MiniLM-L6-v2 wired up in the Worker.

In an enterprise environment, keeping the embedding model behind a provider interface makes it easier to swap implementations and test them independently. In this implementation, the application-level configuration change happens in one place:

# worker.py — the only place the concrete model is named
embedding_provider = build_embedding_provider(kind=EmbeddingProviderKind.ALL_MINILM, config=conf)

A model change may still require corresponding index configuration or re-embedding existing data if the new model produces a different vector space.

The embedded flag pattern#

After the vectors are stored in Pinecone, we call update_embedded_records to mark the complaints as embedded=True in Postgres:

@activity.defn
async def update_embedded_records(self, file_id: int) -> None:
   self.data_store.mark_embedded(file_id)

This flag decouples Postgres from Pinecone. After store_vector completes successfully, update_embedded_records marks the complaints as embedded=True in Postgres. The flag becomes useful during reconciliation or a later rerun because the embedding step can select only records that have not yet been successfully written to the vector store.

case_id provides a stable identifier across the pipeline. Postgres ignores duplicate inserts for an existing case_id, while a Pinecone upsert using an existing record ID overwrites that record rather than creating another one. This makes retries safe without requiring previously completed Activities to run again.

VectorStorageActivity — upsert and query check#

The VectorStorageActivity reads the vectors from Redis and upserts them into Pinecone:

@Activity.defn
async def store_vector(self, unique_cache_id: str) -> VectorStorageActivityResult:
   vector_payloads: list[VectorRecord] = self.get_vectors_from_cache(unique_cache_id)
   count = self.vector_db_provider.upsert_vectors(vector_payloads)
   if count != len(vector_payloads):
       raise RuntimeError(
           f"Pinecone upsert mismatch: expected {len(vector_payloads)}, got {count}"
       )
   return VectorStorageActivityResult(number_of_vectors_stored=count)

After the upsert, we run a sanity query using the same cache key — take the first vector and query Pinecone for its top three nearest neighbors. After the upsert, we exercise the query path using the first cached vector and request its top three nearest neighbors. This gives us a lightweight smoke test of the retrieval path. Because Pinecone is eventually consistent, an immediate query should not be treated as definitive proof that a recent write is already visible. You can do something clever here and write a small Workflow which you might want to run in your CI to check it. Or you can write an integration test which runs this in CI to make sure everything is working fine.

@activity.defn
async def query_vector(self, unique_cache_id: str) -> list[SimilarityResponse]:
   vector_payloads: list[VectorRecord] = self.get_vectors_from_cache(unique_cache_id)
   response = self.vector_db_provider.query(
       vector=vector_payloads[0].vector,
       top_k=3,
       namespace="default",
       filters=None,
   )
   return response

The SimilarityResponse results are visible in the Temporal UI — this is intentional. The embeddings are generated only from text that has already passed through the redaction boundary. Even so, we treat embeddings as sensitive application data rather than assuming that a numeric representation is inherently private.

Cache cleanup#

The final step deletes the Redis key:

@activity.defn
async def delete_cache(self, key: str) -> None:
   self.cache_provider.delete(key)

Simple, but important. The cache key is scoped to a single Workflow execution — once the vectors are in Pinecone and the records are marked embedded, the cache has no further use. Cleaning up the key prevents stale vectors from accumulating after the Workflow no longer needs them.

Conclusion#

To quickly summarize this: Each Activity in this chain has a single responsibility and a clear contract with the next one. The only data crossing Activity boundaries is a file_id and a Redis cache key — both small, neither containing PII. Idempotency is baked in at every layer: case_id in Postgres, upsert semantics in Pinecone, and the embedded flag as the reconciliation mechanism between the two.

Temporal makes the control flow durable, while the application’s idempotency strategy keeps the Postgres and Pinecone side effects safe to retry. The pipeline can resume after failures without rebuilding work that has already completed.