Agentic AI on Google Cloud, Part 2: Building RAG with Cloud Storage, Document AI, Vertex AI and BigQuery
The agent can look up a claim. It cannot yet explain why a rule applies. This post gives it knowledge: policy documents that are stored, extracted, chunked, embedded, indexed and searched, and then kept up to date automatically, with each GCP service doing one clear job.
Where we are. Part 1 gave us the GCP map and a deployed agent with a claims tool. Steps 1–8 here build and test the knowledge path by hand so you can see every stage. Steps 9–13 automate it with Pub/Sub, a Cloud Run Service, a Cloud Run Job and Cloud Scheduler.
1. What RAG is, in plain terms
A language model knows general language, not your contract. Retrieval-augmented generation fixes that in two moves. Retrieve: find the few passages of your documents most relevant to the question. Generate: hand them to the model as evidence and ask it to answer from them.
The quality of the answer is capped by the quality of retrieval. If the right clause is not among the passages, the model cannot cite it, and may improvise. So most of the engineering is not about the model. It is about how documents are prepared, stored and searched.
2. The pipeline and which GCP service does each stage
| Stage | Service | Why |
|---|---|---|
| Store originals | Cloud Storage | Source of truth for documents; the file path becomes the citation link |
| Extract text | Document AI (or plain parsing for clean digital PDFs) | Handles scans, tables, layout |
| Chunk | Your Python code, optionally with Hugging Face models | Only you know your document structure; small open models can find topic boundaries and count tokens locally |
| Embed | Vertex AI embeddings | Turns each chunk into a vector capturing meaning |
| Store and search | BigQuery | Vector search plus ordinary SQL filters on version, date and status |
Where does the chunking code actually run?
No GCP product "does chunking". The parser, the chunker and the indexing logic are ordinary Python inside a Cloud Run container. When a chunking strategy needs a model, your code calls Vertex AI. Cloud Run executes; Vertex AI supplies embeddings or generation. You choose how the container runs: as a Service for documents arriving one by one (events or an upload API), or as a Job for bulk and scheduled rebuilds (details in Part 1 and Steps 9–13). Both use the same image and the same code, so a nightly rebuild produces the same chunks as a live upload.
Why BigQuery, and when to choose something else
For this insurer, the hard requirement is version-aware retrieval: a customer insured in March 2026 must be answered from the policy version effective then. BigQuery lets us express that as a plain SQL filter next to the vector search, which is easy to audit and test. The managed options are real alternatives:
| Option | Pick it when | Trade-off |
|---|---|---|
| Vertex AI RAG Engine | You want a managed corpus, import and retrieval API quickly | Less control over filtering and ranking; check current feature support |
| Vertex AI Search | You want managed enterprise search with ranking built in | Opinionated; verify filter and metadata support for your needs |
| Vertex AI Vector Search | Very large corpora, low-latency serving | More infrastructure to run |
| BigQuery (this post) | Moderate corpora, strict SQL filters, analytics next to the data | Not built for millisecond serving at huge scale |
A sensible workflow is to prototype on the managed option, measure with a small evaluation set, and move to custom only if the evals show you need control.
3. Prepare the data: what a good chunk looks like
Two decisions matter more than the embedding model.
Chunk by meaning. If "80% reimbursement" lands in one chunk and "subject to a €500 annual limit" in the next, a search may return only the first and the answer will omit the cap. Split at clause or section boundaries and keep the heading with its text.
Attach metadata. Each chunk carries what you need to filter and cite:
| Field | Used for |
|---|---|
chunk_id, document_id, version | Stable citation ids |
effective_from, effective_to | Pick the version in force on the expense date |
status (staged / active / retired) | Never retrieve drafts; keep old versions for history |
section, page, source_uri | Clickable citations |
embedding | Vector search |
4. Hands-on: build the knowledge path
Before the steps, here is the whole knowledge path as one list. Every stage you expect from a RAG system has a step, and the headings below carry the stage name so you can always tell where you are.
| Stage | What happens | Steps |
|---|---|---|
| Infrastructure | Bucket, dataset, table, permissions | 1–2 |
| Data preparation | Upload, extract text, clean it, chunk it, attach metadata | 3–4 |
| Vectorization | Turn each chunk into an embedding vector and load it with its metadata | 5 |
| Retrieval | Embed the question, search by meaning, filter by version and date | 6 |
| Agent integration | Expose retrieval as a tool the agent can call | 7 |
| Evaluation | Measure whether retrieval finds the right passages | 8 |
| Automation | Run all of the above on events, in bulk and on a schedule | 9–13 |
The files you create in this part live in two folders of the project from Part 1 (see Figure 6 there for the complete tree):
src/
├── ingestion/
│ ├── extract.py Step 3 Document AI text extraction
│ ├── chunk.py Step 4 clean_text, chunk_by_section, semantic_chunks
│ ├── embeddings.py Step 5 Vertex AI vectors (3 task types)
│ ├── pipeline.py Step 5 parse → clean → chunk → embed → load (shared)
│ ├── main.py Step 9 Pub/Sub endpoint (Cloud Run Service)
│ └── batch.py Step 13 Cloud Run Job entrypoint
├── rag/
│ └── retrieval.py Step 6 vector search with version and date filters
└── agent/tools/
└── policies.py Step 7 the tool the agent callsStep 1. Infrastructure: create the bucket, the dataset and the table
gcloud storage buckets create gs://$PROJECT-policies --location=$REGION
bq --location=$REGION mk --dataset $PROJECT:kb
bq query --use_legacy_sql=false "
CREATE TABLE kb.chunks (
chunk_id STRING, document_id STRING, version INT64,
effective_from DATE, effective_to DATE, status STRING,
section STRING, page INT64, source_uri STRING,
text STRING, embedding ARRAY<FLOAT64>,
chunker_version STRING, embedding_model STRING)"Check: bq show kb.chunks lists the columns. Keep bucket and dataset in the same region as the rest of the system.
Step 2. Infrastructure: grant the ingestion identity its roles
ING=ingest-sa@$PROJECT.iam.gserviceaccount.com
for ROLE in roles/aiplatform.user roles/bigquery.dataEditor roles/bigquery.jobUser \
roles/storage.objectViewer roles/documentai.apiUser; do
gcloud projects add-iam-policy-binding $PROJECT --member=serviceAccount:$ING --role=$ROLE
done
gcloud services enable documentai.googleapis.comCompare with Part 1: api-sa only read BigQuery; ingest-sa can write it. In Steps 1–8 you run the ingestion code under your own credentials; from Step 9 it runs as ingest-sa in Cloud Run.
Step 3. Data preparation: upload the policy and extract its text
gcloud storage cp health-policy-v4.pdf gs://$PROJECT-policies/policies/HEALTH-2026-001/v4.pdf \
--custom-metadata=effective_from=2026-01-01The path convention policies/<policy id>/v<N>.pdf plus an effective_from metadata value lets the automated pipeline later derive the document id, the version and the effective date without any manual input. In the console, open Document AI and create an OCR processor in the eu location. Note its ID. For clean digital PDFs a plain parser may be enough, but OCR also handles scans.
# src/ingestion/extract.py
from google.cloud import documentai, storage
def extract_pages(gcs_uri: str, processor_name: str) -> list[str]:
bucket, blob = gcs_uri.removeprefix("gs://").split("/", 1)
data = storage.Client().bucket(bucket).blob(blob).download_as_bytes()
client = documentai.DocumentProcessorServiceClient(
client_options={"api_endpoint": "eu-documentai.googleapis.com"}) # match the processor location
result = client.process_document(request=documentai.ProcessRequest(
name=processor_name,
raw_document=documentai.RawDocument(content=data, mime_type="application/pdf")))
doc = result.document
return [doc.text[p.layout.text_anchor.text_segments[0].start_index:
p.layout.text_anchor.text_segments[0].end_index] for p in doc.pages]us or eu multi-region locations, and the processor name has the form projects/PROJECT/locations/eu/processors/ID. The client must use the matching endpoint.Step 4. Data preparation: clean the text, chunk it, attach metadata
Extracted text is rarely clean. PDFs bring running page footers, words split across lines by hyphens, and strange spacing, and every one of these quirks ends up inside a chunk and then inside a vector, where it adds noise. So data preparation in this step has three parts, in order: clean the text, chunk it along its natural structure, and attach metadata to each chunk. Cleaning first:
# src/ingestion/chunk.py (top of file)
import re
def clean_text(t: str) -> str:
t = t.replace("\u00ad", "") # soft hyphens
t = re.sub(r"-\n(?=[a-z])", "", t) # re-join words split across lines
t = re.sub(r"^Page \d+ of \d+\s*$", "", t, flags=re.M) # running footers
t = re.sub(r"[ \t]+", " ", t) # collapse spaces
t = re.sub(r"\n{3,}", "\n\n", t) # collapse blank lines
return t.strip()Adapt the patterns to your own documents: open a few extracted pages, find the recurring noise, and write one rule per kind of noise. Then chunk the cleaned pages. The chunker below follows the document's numbered headings; the metadata (document id, version, effective dates, section, page, source file) is attached when the pipeline builds each row in Step 5, so every chunk carries what it needs for filtering and citation.
# src/ingestion/chunk.py
import re
HEADING = re.compile(r"^(\d+(\.\d+)*)\s+[A-Z].+$", re.M) # e.g. "6.2 Outpatient care"
def chunk_by_section(pages: list[str], max_chars: int = 1800):
chunks = []
for page_no, text in enumerate(pages, start=1):
marks = [m.start() for m in HEADING.finditer(text)] + [len(text)]
for a, b in zip(marks, marks[1:]):
body = text[a:b].strip()
if not body: continue
title = body.splitlines()[0]
for i in range(0, len(body), max_chars): # split only oversized sections
chunks.append({"section": title, "page": page_no, "text": f"{title}\n{body[i:i+max_chars]}"})
return chunksThis is a starting point. Real policies have tables and cross-references, so inspect your chunks by eye: print ten of them and ask whether each stands alone as an answer. Re-chunking is cheap; discovering bad chunks after launch is not.
Alternative chunking: semantic boundaries
Section chunking suits policies with numbered headings. For text without reliable structure (emails, free-form manuals) you can detect topic changes instead: split into sentences, embed each, compare consecutive sentences with cosine similarity, and start a new chunk where similarity drops below a threshold.
# src/ingestion/chunk.py (addition)
import numpy as np
def cosine_similarity(a, b) -> float:
a, b = np.asarray(a), np.asarray(b)
d = np.linalg.norm(a) * np.linalg.norm(b)
return 0.0 if d == 0 else float(np.dot(a, b) / d)
def semantic_chunks(sentences, embeddings, threshold=0.65):
if len(sentences) != len(embeddings): raise ValueError("counts must match")
if not sentences: return []
chunks, current = [], [sentences[0]]
for i in range(1, len(sentences)):
if cosine_similarity(embeddings[i-1], embeddings[i]) < threshold:
chunks.append(" ".join(current)); current = []
current.append(sentences[i])
if current: chunks.append(" ".join(current))
return chunks0.65 is illustrative. Tune it on your corpus using the recall@k harness in Step 8; comparing sliding windows of sentences usually finds boundaries better than adjacent pairs. Use structural chunking first where structure exists, semantic chunking where it does not.Using Hugging Face models for chunking
Semantic chunking needs a model that turns sentences into vectors, but those boundary vectors are used once and thrown away. Paying for an API call per sentence is wasteful when a small open model from the Hugging Face hub can do the same job inside your own container, for free, with no network round trip. Hugging Face models help in two places in the chunker:
- Boundary detection: a sentence-transformers model embeds each sentence so we can spot topic changes.
- Token counting: a model's tokenizer tells us how many tokens a chunk really contains, so chunk size is controlled in tokens (what models care about) instead of characters.
Install the libraries (the CPU build of PyTorch is enough for small models and keeps the image smaller):
uv add sentence-transformers transformers# src/ingestion/chunk.py (addition: Hugging Face models)
import os
from functools import lru_cache
from sentence_transformers import SentenceTransformer
from transformers import AutoTokenizer
# A multilingual model suits French and English policies. Check the model card
# for its language coverage, maximum input length and licence before using it.
HF_MODEL = os.environ.get("HF_BOUNDARY_MODEL",
"sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2")
@lru_cache(maxsize=1)
def _encoder() -> SentenceTransformer:
return SentenceTransformer(HF_MODEL) # loaded once per container, not per document
@lru_cache(maxsize=1)
def _tokenizer():
return AutoTokenizer.from_pretrained(HF_MODEL)
def embed_for_boundaries(sentences: list[str]) -> list[list[float]]:
return _encoder().encode(sentences, batch_size=64, normalize_embeddings=True).tolist()
def n_tokens(text: str) -> int:
return len(_tokenizer().encode(text, add_special_tokens=False))
def split_oversized(chunks: list[str], max_tokens: int = 300) -> list[str]:
out = []
for c in chunks:
if n_tokens(c) <= max_tokens:
out.append(c); continue
cur = []
for sent in re.split(r"(?<=[.!?])\s+", c): # break an oversized chunk at sentence ends
if cur and n_tokens(" ".join(cur + [sent])) > max_tokens:
out.append(" ".join(cur)); cur = []
cur.append(sent)
if cur: out.append(" ".join(cur))
return outNow point the fallback chunker from Step 5's pipeline.py at these functions, replacing the call to the Vertex AI boundary embeddings:
# src/ingestion/pipeline.py (semantic_fallback, updated)
from src.ingestion.chunk import embed_for_boundaries, split_oversized
def semantic_fallback(pages):
out = []
for n, text in enumerate(pages, 1):
sents = [s for s in re.split(r"(?<=[.!?])\s+", text) if s.strip()]
if not sents: continue
vecs = embed_for_boundaries(sents) # Hugging Face, local
for t in split_oversized(semantic_chunks(sents, vecs)): # then cap chunk size in tokens
out.append({"section": "(unstructured)", "page": n, "text": t})
return outThree things to understand about this design.
The boundary model and the stored-vector model are different, and that is fine. The Hugging Face model only decides where to cut. The vectors you store and search are still produced by the Vertex AI model in Step 5, computed on the final chunks. Boundary vectors are never compared with stored vectors, so the two models do not need to match. (You could also use a Hugging Face model for the stored vectors, but then queries must use the same model and the vector length in BigQuery changes, so treat that as a separate, deliberate migration.)
Token counts are approximate for the stored model. The tokenizer belongs to the Hugging Face model, not to Gemini, so its counts will not match exactly. Use it to keep chunks comfortably below the embedding model's input limit, with some margin, rather than to hit a precise number.
Do not download models at request time. Cloud Run has no persistent disk, and a model fetched from the hub on first use makes every cold start slow and dependent on an external site. Bake the model into the container image at build time:
# Dockerfile.ingest (additions)
ENV HF_HOME=/app/.hf
RUN python -c "from sentence_transformers import SentenceTransformer as S; \
S('sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2')"This makes the image larger (the model plus PyTorch), so give the Service enough memory (the 2 Gi in Step 10 is a starting point; measure it) and expect slower builds. In exchange, boundary detection costs nothing per call and works the same on your laptop, in the Service and in the Job. Step 4's structural chunker needs none of this, so keep these libraries out of the API image from Part 1; only the ingestion image needs them.
Step 5. Vectorization: embed the chunks and load BigQuery
What vectorization is. A search engine that matches words fails when the customer says "specialist appointment" and the contract says "outpatient consultation". Vectorization solves this by turning each chunk of text into a vector: a long list of numbers (hundreds or thousands of them) produced by an embedding model. The model is trained so that texts with similar meaning receive vectors that point in similar directions, and unrelated texts end up far apart. You can picture each chunk as a point in a very high-dimensional space, with related ideas forming neighbourhoods.
"Closeness" is measured with a simple formula, cosine similarity, which compares the angle between two vectors. We let BigQuery compute it in Step 6. Our job in this step is only to produce one vector per chunk and store it next to the chunk's text and metadata. One row therefore holds three things: what the chunk says (text), what it means (the vector) and what it is (version, dates, page, source).
Embeddings come from Vertex AI through the Google Gen AI SDK. Model names and input limits change, so check the current documentation before deploying (the models documented at the time of writing include gemini-embedding-001 and gemini-embedding-2). Note that there are three different jobs, each with its own task type:
# src/ingestion/embeddings.py
import os
from google import genai
from google.genai.types import EmbedContentConfig
client = genai.Client(vertexai=True, project=os.environ["GOOGLE_CLOUD_PROJECT"],
location=os.environ["GOOGLE_CLOUD_LOCATION"])
MODEL = os.environ.get("EMBEDDING_MODEL", "gemini-embedding-001")
def embed(texts: list[str], task_type: str) -> list[list[float]]:
out = []
for i in range(0, len(texts), 50): # batch; respect quotas, add retries
r = client.models.embed_content(model=MODEL, contents=texts[i:i+50],
config=EmbedContentConfig(task_type=task_type))
out += [e.values for e in r.embeddings]
return out
def embed_for_boundaries(sentences): return embed(sentences, "SEMANTIC_SIMILARITY") # thrown away
def embed_for_index(chunks): return embed(chunks, "RETRIEVAL_DOCUMENT") # stored
def embed_query(question): return embed([question], "RETRIEVAL_QUERY")[0] # at search timeSentence embeddings are used only to find boundaries and are discarded. The vectors you store are computed on the final chunks. Two rules follow. The stored model must match the query model, because vectors from different models are not comparable; changing the model means re-embedding everything (the Job in Step 13 does this). And the vector length must be identical in every row.
Now the pipeline that ties parsing, chunking, embedding and loading together. It is the one piece of code both the Service and the Job will call, so write it as a plain function that can be tested without HTTP:
# src/ingestion/pipeline.py
import re
from google.cloud import bigquery
from src.ingestion import embeddings
from src.ingestion.extract import extract_pages
from src.ingestion.chunk import clean_text, chunk_by_section, semantic_chunks
CHUNKER_VERSION = "2026-10-a"
bq = bigquery.Client()
def semantic_fallback(pages):
out = []
for n, text in enumerate(pages, 1):
sents = [s for s in re.split(r"(?<=[.!?])\s+", text) if s.strip()]
if not sents: continue
vecs = embeddings.embed_for_boundaries(sents)
out += [{"section": "(unstructured)", "page": n, "text": t} for t in semantic_chunks(sents, vecs)]
return out
def process(document_id, version, source_uri, meta, processor_name) -> int:
pages = [clean_text(p) for p in extract_pages(source_uri, processor_name)]
chunks = chunk_by_section(pages) or semantic_fallback(pages)
vecs = embeddings.embed_for_index([c["text"] for c in chunks])
rows = [{**meta, "chunk_id": f"{document_id}-{i}", "document_id": document_id,
"version": version, "source_uri": source_uri, "status": "staged",
"chunker_version": CHUNKER_VERSION, "embedding_model": embeddings.MODEL,
"section": c["section"], "page": c["page"], "text": c["text"], "embedding": v}
for i, (c, v) in enumerate(zip(chunks, vecs))]
bq.query("DELETE FROM kb.chunks WHERE document_id=@d AND status='staged'",
job_config=bigquery.QueryJobConfig(query_parameters=[
bigquery.ScalarQueryParameter("d", "STRING", document_id)])).result() # safe re-run
bq.load_table_from_json(rows, "kb.chunks").result() # load job, not streaming
return len(rows)Three design choices matter. chunk_id is deterministic, so re-processing the same document produces the same ids. Staged rows are deleted before loading, so a retry never duplicates. And the chunker_version and embedding_model columns record how every row was made, which is what lets you find and rebuild stale rows later. Run it once by hand:
python -c "from src.ingestion.pipeline import process; print(process('HEALTH-2026-001-v4', 4, \
'gs://$PROJECT-policies/policies/HEALTH-2026-001/v4.pdf', \
{'effective_from': '2026-01-01', 'effective_to': None}, \
'projects/$PROJECT/locations/eu/processors/PROCESSOR_ID'))"Notice the rows are loaded as staged. Retrieval only reads active chunks, so a half-processed document can never reach a customer. After a quick validation, promote it:
UPDATE kb.chunks SET status='active' WHERE document_id='HEALTH-2026-001-v4';
UPDATE kb.chunks SET status='retired', effective_to=DATE '2025-12-31'
WHERE document_id='HEALTH-2026-001-v3'; -- kept for historical claimsCheck: SELECT section, page, SUBSTR(text,1,80) FROM kb.chunks LIMIT 10 shows readable chunks, and SELECT COUNT(*) FROM kb.chunks WHERE ARRAY_LENGTH(embedding)=0 returns 0.
Step 6. Retrieval: search with version and date filters
Retrieval embeds the question with the query task type, then runs VECTOR_SEARCH on a pre-filtered set of rows. The filter lives in SQL, not in the prompt, because a prompt is a request and a filter is a guarantee.
# src/rag/retrieval.py
from google.cloud import bigquery
from src.ingestion.embeddings import embed_query
SQL = """
SELECT base.chunk_id, base.document_id, base.version, base.section, base.page,
base.source_uri, base.text, distance
FROM VECTOR_SEARCH(
(SELECT * FROM kb.chunks
WHERE status IN UNNEST(@statuses)
AND effective_from <= @as_of AND (effective_to IS NULL OR effective_to >= @as_of)),
'embedding', (SELECT @q AS embedding), top_k => @k, distance_type => 'COSINE')
ORDER BY distance"""
def search(question: str, as_of: str, k: int = 5, statuses=("active",)):
q = embed_query(question)
job = bigquery.Client().query(SQL, job_config=bigquery.QueryJobConfig(query_parameters=[
bigquery.ArrayQueryParameter("q", "FLOAT64", q),
bigquery.ArrayQueryParameter("statuses", "STRING", list(statuses)),
bigquery.ScalarQueryParameter("as_of", "DATE", as_of),
bigquery.ScalarQueryParameter("k", "INT64", k)]))
return [dict(r) for r in job.result()]Try it. The question never says "outpatient", yet the right section should rank first:
python -c "from src.rag.retrieval import search; \
[print(r['section'], r['version'], round(r['distance'],3)) for r in search('How much do I get back for a specialist appointment?', '2026-03-10')]"6.2 Outpatient consultations 4 0.211
6.4 Specialist referrals 4 0.298
2.1 Eligible expenses 4 0.352Now run the same query with as_of='2025-06-01'. You should get version 3 passages, because that was in force then. That one behavior is why we chose a table with dates over a generic index.
HEALTH-2026-001, semantic search can miss. BigQuery also supports text search indexes and the SEARCH function, and a hybrid approach merges both result lists. We add that when the evaluation shows identifier queries failing.Step 7. Agent integration: give the agent a policy tool
# src/agent/tools/policies.py
from src.rag.retrieval import search
def search_policy_documents(question: str, expense_date: str) -> dict:
"""Search approved insurance policy text. expense_date is the date the
expense occurred, as YYYY-MM-DD, so the correct policy version is used.
Use for any question about coverage, limits, waiting periods or rules."""
hits = search(question, as_of=expense_date)
return {"success": True, "passages": [
{"source_id": h["chunk_id"], "version": h["version"], "section": h["section"],
"page": h["page"], "text": h["text"]} for h in hits]}Add it to the agent from Part 1: tools=[get_claim_status, search_policy_documents], and extend the instruction: "For coverage questions use search_policy_documents and answer only from the returned passages. Cite source_id. If no passage supports the answer, say you cannot confirm it." Redeploy with the image build from Part 1. The expense date is something the agent can get from the claim record; if unknown it should ask the customer.
Check: ask "Why was CLM-1042 rejected and what is the reimbursement rate for outpatient care?" In the trace you should see both tools called, and an answer that quotes the section and version.
Step 8. Evaluation: measure retrieval before you trust it
Write 20 questions with the section that should answer each, and compute recall@k, the share of questions where the right chunk appears in the top k results.
cases = [("How much for a specialist appointment?", "2026-03-10", "6.2"), ...]
hits = sum(any(r["section"].startswith(sec) for r in search(q, d, k=5)) for q, d, sec in cases)
print("recall@5 =", hits / len(cases))Change one thing at a time (chunk size, headings kept, embedding model) and compare scores. This tiny harness will grow into the evaluation suite in Part 4.
5. Automate: the event-driven Service and the bulk Job
Steps 1–8 ran ingestion by hand. In production it must start by itself when a file arrives, and never block the person who uploaded it. Google Cloud's tool for "something happened, someone must react" is Pub/Sub.
First the big picture, which the rest of this section unpacks. Read the flow below, with the identity that does each hop:
file lands in Cloud Storage
→ notification published to a Pub/Sub topic (Cloud Storage may publish)
→ Pub/Sub pushes it to the Cloud Run Service (as pubsub-invoker-sa, role run.invoker)
→ Service runs pipeline.process() as ingest-sa (BigQuery, Vertex AI, Document AI, Firestore)
→ staged rows in kb.chunks; promotion stays a separate validated stepBefore touching a command, let us understand Pub/Sub with a story.
Pub/Sub in plain words: a mailroom
Imagine a company mailroom. The warehouse (Cloud Storage) drops a note into a mailbox saying "a new policy PDF arrived". The mailbox does not read the note. A courier picks it up and delivers it to the ingestion team (our Cloud Run Service). The team signs for the delivery. If nobody signs, the courier comes back later. After five failed attempts, the note goes to a "problem" shelf where a human can look at it.
| Pub/Sub word | Mailroom word | In our system |
|---|---|---|
| Message | The note | A small JSON: "object policies/HEALTH-2026-001/v5.pdf was created in bucket X". It does not contain the PDF itself |
| Publisher | Whoever writes the note | Cloud Storage (the bucket notification) |
| Topic | The mailbox | policy-files |
| Subscription | A delivery contract: "bring every note from this mailbox to this address" | policy-ingest, type push: Pub/Sub calls our URL |
| Subscriber | The team that reads the note | Cloud Run Service rag-ingestion, route /pubsub |
| Ack (acknowledge) | The signature | Our service answers HTTP 2xx: "got it, stop sending" |
| Retry | Courier comes back | Any non-2xx answer or a timeout: Pub/Sub sends the same note again, later |
| Dead-letter topic | The problem shelf | policy-files-dlq, used after 5 failed attempts |
Follow one message from start to finish
A colleague uploads v5.pdf. Here is everything that happens, in order.
What does the message look like when it reaches our code? Pub/Sub wraps it in a JSON envelope and POSTs it to /pubsub. The useful information is in attributes:
{
"message": {
"attributes": {
"eventType": "OBJECT_FINALIZE",
"bucketId": "my-project-policies",
"objectId": "policies/HEALTH-2026-001/v5.pdf",
"objectGeneration": "1761390001234567"
},
"data": "eyJuYW1lIjoi...", <- base64 copy of the object's metadata; we do not need it
"messageId": "1234567890",
"publishTime": "2026-10-25T09:00:02Z"
},
"subscription": "projects/my-project/subscriptions/policy-ingest"
}Our code needs only three attributes, and each one has a job:
| Attribute | Value in the example | What our code does with it |
|---|---|---|
bucketId | my-project-policies | Where to download the PDF from |
objectId | policies/HEALTH-2026-001/v5.pdf | The regular expression extracts policy HEALTH-2026-001 and version 5, so we build the document id HEALTH-2026-001-v5 with no manual input |
objectGeneration | 1761390001234567 | Changes every time a file is overwritten. It makes the Firestore key unique per upload, so "same message twice" is detected but "new file with the same name" is processed |
The Firestore key for this example is therefore policies_HEALTH-2026-001_v5.pdf:1761390001234567:2026-10-a: object, generation and chunker version.
Step 9. The Pub/Sub endpoint, explained line by line
The code is unchanged from before; read it with this map next to it.
# src/ingestion/main.py
import os, re
from fastapi import FastAPI, Request, HTTPException
from google.cloud import firestore, storage
from src.ingestion import pipeline
app, db, gcs = FastAPI(), firestore.Client(), storage.Client()
PROCESSOR = os.environ["DOCAI_PROCESSOR"]
NAME = re.compile(r"policies/(?P<policy>[^/]+)/v(?P<ver>\d+)\.pdf$")
@app.post("/pubsub") # (A)
async def handle(req: Request):
a = (await req.json())["message"].get("attributes", {}) # (B)
obj, bucket, gen = a["objectId"], a["bucketId"], a["objectGeneration"]
m = NAME.match(obj)
if not m: return {"status": "ignored"} # (C)
ref = db.collection("ingestion").document(
f'{obj.replace("/", "_")}:{gen}:{pipeline.CHUNKER_VERSION}') # (D)
snap = ref.get()
if snap.exists and snap.to_dict()["status"] == "done":
return {"status": "duplicate"} # (E)
ref.set({"status": "running"}) # (F)
try:
meta = {"effective_from": gcs.bucket(bucket).get_blob(obj).metadata["effective_from"],
"effective_to": None} # (G)
n = pipeline.process(f'{m["policy"]}-v{m["ver"]}', int(m["ver"]),
f"gs://{bucket}/{obj}", meta, PROCESSOR) # (H)
ref.set({"status": "done", "chunks": n}) # (I)
except Exception:
ref.set({"status": "failed"})
raise HTTPException(500) # (J)
return {"status": "ok"} # (K)| Mark | What it does | Why |
|---|---|---|
| A | Exposes the route Pub/Sub will call | The push subscription URL ends with /pubsub |
| B | Reads the attributes from the envelope | That is where Cloud Storage puts the file name |
| C | Files that do not match policies/<id>/v<N>.pdf are ignored but acknowledged (HTTP 200) | Returning an error here would make Pub/Sub retry a file we will never want |
| D | Builds the idempotency key | Same object + same generation + same chunker = same work |
| E | Already done: answer 200 and stop | Pub/Sub delivers at least once: duplicates are normal |
| F | Marks the job as running | Lets you see in Firestore what is in progress |
| G | Reads effective_from from the file's custom metadata (set in Step 3) | The version date comes from the upload, not from a person typing it |
| H | Runs the shared pipeline: extract, clean, chunk, embed, load as staged | Same function the Job uses |
| I | Marks done | Next duplicate stops at E |
| J | On any error: mark failed and answer HTTP 500 | The 500 is the signal "please retry" |
| K | Normal exit: HTTP 200 | The 200 is the signature: Pub/Sub stops |
The one rule to remember: your HTTP status code is the conversation with Pub/Sub.
| Your answer | Pub/Sub understands | Use it when |
|---|---|---|
| 200 / 204 | "Delivered, forget it" | Success, duplicate, or a file we deliberately ignore |
| 500 or any error, or no answer before the ack deadline | "Failed, send again later" | A temporary problem (quota, network, Document AI hiccup) |
| 401 / 403 | Same as an error: retried, then dead-lettered | You never choose this; it means the invoker permission is wrong (see Step 11) |
Step 10. Deploy the Service, command by command
gcloud services enable pubsub.googleapis.com firestore.googleapis.com cloudscheduler.googleapis.com
gcloud firestore databases create --location=$REGION
gcloud projects add-iam-policy-binding $PROJECT --member=serviceAccount:$ING --role=roles/datastore.user
gcloud artifacts repositories create rag --repository-format=docker --location=$REGION
IMAGE=$REGION-docker.pkg.dev/$PROJECT/rag/ingestion:v1
gcloud builds submit --tag $IMAGE -f Dockerfile.ingest .
gcloud run deploy rag-ingestion --image $IMAGE --region $REGION --service-account $ING \
--memory 2Gi --cpu 2 --concurrency 4 --timeout 600 --no-allow-unauthenticated \
--set-env-vars GOOGLE_CLOUD_PROJECT=$PROJECT,GOOGLE_CLOUD_LOCATION=$REGION,DOCAI_PROCESSOR=projects/$PROJECT/locations/eu/processors/PROCESSOR_ID| Command or flag | Meaning |
|---|---|
services enable ... | Switches on the three APIs we use. A GCP API is off until you enable it in the project |
firestore databases create | Creates the small database that remembers "this file was processed" |
add-iam-policy-binding ... roles/datastore.user | Lets ingest-sa read and write Firestore (nothing more) |
artifacts repositories create | Creates a private shelf in Artifact Registry to store container images |
builds submit --tag $IMAGE | Builds the image from your Dockerfile in Google's cloud and stores it under that name |
run deploy rag-ingestion | Creates the Service from the image |
--service-account $ING | The identity the code runs as: this is what allows BigQuery, Vertex AI and Document AI calls |
--no-allow-unauthenticated | The URL is private: only a caller holding run.invoker gets in. Without this, anyone on the internet could trigger ingestion |
--timeout 600 | Maximum seconds for one request. Cloud Run's default is 300; a long PDF may need more. Keep it consistent with the subscription's ack deadline (Step 11) |
--concurrency 4, --memory, --cpu | Starting points: four documents at once per instance, 2 GiB, 2 vCPU. Lower concurrency if memory runs out |
Step 11. Connect Cloud Storage to the Service, command by command
Think of this step as laying three pipes and giving three keys. Pipes first:
# Pipe 1 and the problem shelf: two topics
gcloud pubsub topics create policy-files
gcloud pubsub topics create policy-files-dlq
# Pipe 2: the bucket publishes to the topic whenever a file is created under policies/
gcloud storage buckets notifications create gs://$PROJECT-policies \
--topic=policy-files --event-types=OBJECT_FINALIZE --object-prefix=policies/
# The identity Pub/Sub will use to knock on the Service's door
gcloud iam service-accounts create pubsub-invoker-sa
# Pipe 3: the subscription that pushes each message to the Service URL
gcloud pubsub subscriptions create policy-ingest --topic=policy-files \
--push-endpoint=$(gcloud run services describe rag-ingestion --region $REGION --format='value(status.url)')/pubsub \
--push-auth-service-account=pubsub-invoker-sa@$PROJECT.iam.gserviceaccount.com \
--dead-letter-topic=policy-files-dlq --max-delivery-attempts=5 --ack-deadline=600| Piece | Meaning |
|---|---|
topics create policy-files | The mailbox. Created empty; it only holds messages in transit |
--event-types=OBJECT_FINALIZE | Only "a new object was fully written" events. Deletions and metadata changes are ignored |
--object-prefix=policies/ | Only files whose name starts with policies/; other files in the bucket never create messages |
--push-endpoint=.../pubsub | The address Pub/Sub calls. $(gcloud run services describe ...) asks Cloud Run for the Service URL and appends our route |
--push-auth-service-account | Pub/Sub attaches a signed token proving it acts as pubsub-invoker-sa; Cloud Run checks it. This is how a private URL accepts Pub/Sub |
--max-delivery-attempts=5 | Give up after 5 deliveries and move the message to the dead-letter topic (allowed range is 5 to 100) |
--ack-deadline=600 | Seconds Pub/Sub waits for our answer before assuming failure. Must cover your slowest document; 600 is the maximum for a subscription |
Now the three keys. A message can only travel if each hop has permission. Figure 6 shows who needs what.
# Key 3: Pub/Sub may call the private Service as pubsub-invoker-sa
gcloud run services add-iam-policy-binding rag-ingestion --region $REGION \
--member=serviceAccount:pubsub-invoker-sa@$PROJECT.iam.gserviceaccount.com --role=roles/run.invoker
# Key 1: Cloud Storage may publish to the topic
GCS_SA=$(gcloud storage service-agent --project=$PROJECT)
gcloud pubsub topics add-iam-policy-binding policy-files \
--member=serviceAccount:$GCS_SA --role=roles/pubsub.publisher
# Key 2: Pub/Sub's own service agent may dead-letter and acknowledge
PROJECT_NUMBER=$(gcloud projects describe $PROJECT --format='value(projectNumber)')
PS_SA=service-$PROJECT_NUMBER@gcp-sa-pubsub.iam.gserviceaccount.com
gcloud pubsub topics add-iam-policy-binding policy-files-dlq \
--member=serviceAccount:$PS_SA --role=roles/pubsub.publisher
gcloud pubsub subscriptions add-iam-policy-binding policy-ingest \
--member=serviceAccount:$PS_SA --role=roles/pubsub.subscriber
# A shelf is useless if nobody can read it: give the dead-letter topic its own subscription
gcloud pubsub subscriptions create policy-files-dlq-pull --topic=policy-files-dlqpolicy-files-dlq-pull, dead-lettered messages are lost. A "service agent" is a Google-managed identity that represents the product itself (Cloud Storage, Pub/Sub). You do not create it; you only grant it roles. If a command complains that an agent does not exist yet, run the notification or subscription creation once, then retry.Try it with a fake message (no upload needed)
You can play the publisher yourself. This is the fastest way to understand and debug the chain. Use a file that already exists in the bucket:
gcloud pubsub topics publish policy-files --message='{}' \
--attribute=eventType=OBJECT_FINALIZE,bucketId=$PROJECT-policies,objectId=policies/HEALTH-2026-001/v5.pdf,objectGeneration=111
gcloud run services logs read rag-ingestion --region $REGION --limit 20Run the first command twice with the same objectGeneration=111. What you should see:
| Attempt | What happens | Where you see it |
|---|---|---|
| 1st publish | Firestore has no record: pipeline runs, staged rows appear, status done | Logs, kb.chunks, Firestore |
| 2nd publish, same generation | Record is done: answer duplicate, no new rows | Logs only; row count unchanged |
Publish with objectGeneration=222 | New key: the file is processed again, as if re-uploaded | A second Firestore record |
Publish with a bad objectId (notes.txt) | Regex does not match: ignored, HTTP 200 | Logs; nothing retried |
What happens when it fails: a timeline
Suppose Document AI is down. Pub/Sub does not hammer the Service; it waits longer between attempts (exponential backoff, by default between 10 and 600 seconds, unless you configure a retry policy).
| Attempt | Our answer | Firestore | Then |
|---|---|---|---|
| 1 | 500 | failed | Pub/Sub waits, retries |
| 2 to 4 | 500 | failed → running → failed | Waits grow |
| 5 | 500 | failed | Message moves to policy-files-dlq |
| (if the service recovers at attempt 3) | 200 | done | Message finished, nothing in the DLQ |
To look at what landed on the problem shelf, and put it back after you fix the cause:
gcloud pubsub subscriptions pull policy-files-dlq-pull --limit=5 # read, do not delete
# fix the bug, redeploy, then re-publish the same attributes to policy-filesWhen nothing arrives: a checklist
| Symptom | Likely cause | Check with |
|---|---|---|
| Upload done, no log line at all | The bucket notification is missing or the prefix is wrong | gcloud storage buckets notifications list gs://$PROJECT-policies |
| Notification exists, still nothing | Key 1 missing: Cloud Storage cannot publish | Topic IAM: gcloud pubsub topics get-iam-policy policy-files |
| Messages pile up, Service shows 401/403 | Key 3 missing or wrong service account on the subscription | gcloud pubsub subscriptions describe policy-ingest and the Service's IAM |
| Service runs, answers 500 | Code or ingest-sa roles (Key 4) | gcloud run services logs read rag-ingestion --region $REGION |
| Same message processed repeatedly | Service answers too slowly (longer than the ack deadline) so Pub/Sub resends | Raise --ack-deadline and --timeout together, or speed up the pipeline |
| Dead-letter topic empty although errors occurred | No subscription on the DLQ topic, or Key 2 missing | gcloud pubsub subscriptions list |
Check: after Step 11, publishing the fake message once should leave a done record in Firestore and staged rows in BigQuery. If it does, all four keys work, and the real upload in Step 12 will work too. I have not run these commands; if one fails, read its error message and compare the flag with the current gcloud pubsub documentation.
Step 12. Test it, including the failures
gcloud storage cp health-policy-v5.pdf gs://$PROJECT-policies/policies/HEALTH-2026-001/v5.pdf \
--custom-metadata=effective_from=2026-09-01
gcloud run services logs read rag-ingestion --region $REGION --limit 20Check: logs show the request, SELECT COUNT(*) FROM kb.chunks WHERE status='staged' AND document_id='HEALTH-2026-001-v5' is above zero, and Firestore has a done record. Then try the failure paths: re-send the same notification (expect duplicate and no new rows), and make the parser fail on purpose (expect five retries, then a message in policy-files-dlq).
After checking the chunks, promote by SQL as in Step 5. Making promotion a deliberate step, never automatic, is what keeps half-processed or wrong documents away from customers.
Step 13. The Cloud Run Job for bulk and nightly rebuilds
Suppose 100,000 documents already exist. Sending them to an HTTP endpoint and waiting would be fragile. A Job runs the same image as a batch; only the command changes.
gcloud run jobs create rag-batch-ingestion --image $IMAGE --region $REGION --service-account $ING \
--command python --args=-m,src.ingestion.batch \
--set-env-vars GOOGLE_CLOUD_PROJECT=$PROJECT,GOOGLE_CLOUD_LOCATION=$REGION,SOURCE_BUCKET=$PROJECT-policies,DOCAI_PROCESSOR=projects/$PROJECT/locations/eu/processors/PROCESSOR_ID \
--memory 4Gi --cpu 2 --tasks 10 --max-retries 2 --task-timeout 3600
gcloud run jobs execute rag-batch-ingestion --region $REGIONTen tasks start in parallel, but nothing divides the work for you. If each task lists the whole bucket, all ten process the same files. Cloud Run gives every task its index and the task count as environment variables; use them to take a distinct slice:
# src/ingestion/batch.py
import os, re
from google.cloud import storage
from src.ingestion import pipeline
idx, n = int(os.environ.get("CLOUD_RUN_TASK_INDEX", 0)), int(os.environ.get("CLOUD_RUN_TASK_COUNT", 1))
bucket, processor = os.environ["SOURCE_BUCKET"], os.environ["DOCAI_PROCESSOR"]
NAME = re.compile(r"policies/(?P<policy>[^/]+)/v(?P<ver>\d+)\.pdf$")
blobs = sorted((b for b in storage.Client().list_blobs(bucket, prefix="policies/") if NAME.match(b.name)),
key=lambda b: b.name)
for b in blobs[idx::n]: # this task's share only
m = NAME.match(b.name)
meta = {"effective_from": (b.metadata or {})["effective_from"], "effective_to": None}
pipeline.process(f'{m["policy"]}-v{m["ver"]}', int(m["ver"]), f"gs://{bucket}/{b.name}", meta, processor)
print({"task": idx, "document": b.name}) # structured log lineBecause process() deletes staged rows first, a retried task safely redoes its documents. When a rebuild replaces an existing document (new chunker or model), promote it with one transaction so queries never see a gap: BEGIN TRANSACTION; DELETE FROM kb.chunks WHERE document_id=@d AND status='active'; UPDATE kb.chunks SET status='active' WHERE document_id=@d AND status='staged'; COMMIT TRANSACTION;. Promoting a new version is different: retire the previous version instead of deleting it. For a nightly rebuild, create a Cloud Scheduler job that executes the Cloud Run Job through the Cloud Run Admin API using a service account allowed to run it; check the current Cloud Scheduler documentation for the exact target syntax.
6. Cost, limits and gotchas
| Topic | What to know |
|---|---|
| Embedding cost | You pay per character or token embedded. Ingest once, not per query; query embedding is tiny |
| BigQuery cost | Charged by bytes scanned. Vector search scans embeddings, so filter first and consider a vector index as the table grows |
| Region | Bucket, dataset and Vertex AI calls in the same region avoid latency and residency surprises |
| Embedding model change | Vectors from different models are not comparable. Changing the model means re-embedding everything |
| Quotas | Embedding and Document AI have per-minute limits. Batch and back off |
| Streaming inserts | Rows added by streaming insert cannot be deleted for a while, which breaks delete-then-reload. The pipeline uses load jobs for that reason |
Production checklist for the ingestion pipeline
| Concern | What this pipeline does |
|---|---|
| Idempotency | Firestore key per object generation and chunker version; deterministic chunk ids; staged rows deleted before load |
| Versioning | chunker_version and embedding_model on every row, so stale rows can be found and rebuilt |
| Retries | Pub/Sub retries and a dead-letter topic; add backoff on embedding calls; Job --max-retries |
| Batching | 50 texts per embedding request; respect model quotas |
| Security | Dedicated ingest-sa with least privilege; no key files in the image; services deployed with --no-allow-unauthenticated |
| Observability | Log document id, chunk count, duration and errors; alert on dead-letter depth and on failed Firestore records |
| Data quality | Keep document id, headings, page numbers and source offsets so citations can point to exact passages |
| Reindexing | Replace chunks per document and version; retire old versions rather than deleting history |
Where the vectors live
This series keeps chunks and vectors in BigQuery because version and date filters are plain SQL. Another common design pairs Vertex AI Vector Search for similarity with a separate metadata store (Firestore or Cloud SQL). That suits very large corpora or low-latency serving, but you then keep two stores consistent. The pipeline changes only in its last step, the write.
7. What you have and what is next
The agent now has two kinds of knowledge, each from the right place: live facts from the claims API, and explanations from versioned policy text in BigQuery. New documents flow from Cloud Storage through Pub/Sub into a Cloud Run Service that parses, chunks, embeds and stages them, while a Job handles bulk and nightly rebuilds with the same code. Each hop has its own identity and only the permissions it needs.
Part 3 adds the verification and citation layer, so answers are checked against this evidence. Not covered here: hybrid lexical search, reranking, a vector index, an automated validated promotion step, and Terraform, which come later. I have not run this code, so check the embedding model name and output size, the Gen AI SDK call shape, the Document AI client usage and location, the notification attribute names, and the Cloud Scheduler syntax against current documentation.
References: BigQuery vector search, Document AI, Vertex AI embeddings, RAG Engine.
Comments
Post a Comment