# Building a Document-RAG Agent on GCP's Agent Development Kit (ADK) > Source: > Published: 2026-07-14 14:06:30+00:00 [LogiChat](https://logichat.io) is a chatbot platform. Customers upload their docs, get a chat widget, never touch a model. For two years, "training the bot" meant hand-curating a Q&A list in a dashboard form — `question` and `answer` pairs, one row at a time, dumped into the prompt as few-shot pairs. I just deleted that and rebuilt the whole pipeline on ** Google's Agent Development Kit (ADK)**, with It's not a "hello world" agent demo — it's the architecture I landed on after rebuilding the whole pipeline over the last few months, three Cloud Run services touched (the fourth, the Stripe top-up token subscriber, was untouched), and the one design choice that turned out to matter more than anything else: **retrieval as a pre-model step, not an ADK tool.** The pipeline below touches three of those services — `apps/api` , `apps/subscribers/doc-processor` , and `apps/agent` — plus the Firestore vector index. `apps/api` appears twice in the diagram because it acts as both the upload ingress AND the gateway that mints ID tokens for the agent. Click through the five stages to see which service is active at each hop and what payload crosses each boundary: Interactive diagram:step through the five-stage pipeline —Upload → Process → Retrieve → Ground → Answer— on the[original post on dalenguyen.me]. Each hop shows which service is active (dashboard →`apps/api` → Cloud Storage →`doc-processor` → Firestore →`apps/agent` → Vertex AI) and what payload crosses the boundary. Two pieces are different from a textbook RAG stack: `apps/api` , Express) is a separate service that mints Cloud Run ID tokens to call the agent. The agent is not publicly reachable.Both are deliberate. Both are what this post is really about. I picked [Google's Agent Development Kit](https://github.com/google/adk-python) over LangGraph, CrewAI, and a hand-rolled loop for one reason: **the agent service runs entirely on GCP, and ADK is the only one of those that was built for Vertex AI from the ground up.** The Python SDK ships a `google.adk.agents.Agent` and a `google.adk.runners.InMemoryRunner` that route through `google-genai` , which already speaks Vertex AI when `GOOGLE_GENAI_USE_VERTEXAI=TRUE` . No LangChain adapter, no provider shim, no surprise model name drift between staging and prod. Concretely, my `apps/agent` is a 53-line `pyproject.toml` Python 3.12 service: ``` [project] name = "logichat-agent" version = "0.1.0" description = "LogiChat ADK agent service (Python) — Gemini 2.5 Flash via Vertex AI." requires-python = ">=3.12" dependencies = [ "fastapi>=0.115", "pydantic>=2.7", "uvicorn>=0.30", # google-adk is the Agent Development Kit Python SDK. Pinned loosely # because ADK's API is still moving; follow-up issues will exercise # the real Runner (sessions / tools / streaming). "google-adk>=0.1.0", # google-genai comes with ADK, but I import google.genai.types # directly in the runner adapter, so declare it explicitly. "google-genai>=1.0", # google-cloud-firestore is the client I use for the per-app chunk # vector search. Required by the retrieval layer. "google-cloud-firestore>=2.20", ] ``` Five runtime deps. FastAPI for the HTTP surface, Pydantic for the wire contract, uvicorn for the Cloud Run entrypoint, `google-adk` for the agent, `google-genai` for the typed message parts, `google-cloud-firestore` for the KNN search. ADC works for all of it — no API keys anywhere. `apps/agent` is split into four files, each with one job: | File | Job | |---|---| `src/agent/main.py` | uvicorn entrypoint; binds `$PORT` (Cloud Run injects) | `src/agent/agent_definition.py` | Builds the ADK `Agent` + `InMemoryRunner` , wraps in `RunnerProtocol` | `src/agent/retrieval.py` | Firestore `find_nearest` + `text-embedding-005` via Vertex AI | `src/agent/api/app.py` | FastAPI factory: `POST /run` , `GET /healthz` | The `main.py` is six lines of orchestration: ``` php # apps/agent/src/agent/main.py def main() -> None: runner = build_agent() # builds the ADK-backed adapter app = create_app(runner=runner) # wires the runner into FastAPI port = int(os.environ.get("PORT", "8080")) uvicorn.run(app, host="0.0.0.0", port=port, log_level="info") ``` `build_agent()` is where ADK gets wired up. Here's the whole thing: ``` # apps/agent/src/agent/agent_definition.py MODEL_ID = "gemini-2.5-flash" SYSTEM_INSTRUCTION = ( "You are LogiChat's support assistant. Answer the user's question using " "ONLY the context retrieved from the company's knowledge base. If the " "retrieved context does not contain the answer, say you don't know and " "suggest the user rephrase or contact a human. Never invent facts, " "product names, or policies that are not present in the retrieved " "context. Each retrieved block is prefixed with an internal " "[doc:] marker so you can tell sources apart while reasoning — " "these markers are for your own use only and must never appear " "anywhere in the text you show the user." ) def build_agent() -> RunnerProtocol: _require_env("GOOGLE_GENAI_USE_VERTEXAI") # TRUE → route via Vertex AI _require_env("GOOGLE_CLOUD_PROJECT") # logichat-dev / logichat-prod _require_env("GOOGLE_CLOUD_LOCATION") # us-central1 # Inline import: lets the unit tests run without google-adk installed. from google.adk.agents import Agent from google.adk.runners import InMemoryRunner root_agent = Agent( name="logichat_faq_agent", model=MODEL_ID, instruction=SYSTEM_INSTRUCTION, # tools=[firestore_retrieval_tool] ← intentionally NOT an ADK tool ) runner = InMemoryRunner(agent=root_agent) return _AdkRunnerAdapter(runner) ``` The env vars are validated at startup so a misconfigured deploy fails the first request — not the first deploy. The model id is **pinned in code** so a stray env var override can't silently swap in a different family. The region and project still come from env so staging and prod can diverge without a rebuild. `RunnerProtocol` seam The `google.adk.runners.InMemoryRunner` exposes more surface than my HTTP layer needs — sessions, event streams, function calls. I wrap it in a protocol so the route handler is stable while the upstream SDK moves: ``` # apps/agent/src/agent/api/dependencies.py class RunnerProtocol(Protocol): async def run( self, *, app_id: str, session_id: str, question: str, config: dict[str, Any], ) -> RunnerResult: ... ``` `_AdkRunnerAdapter` implements that protocol and forwards to the real ADK runner. The HTTP layer only knows about `RunnerProtocol` — it never imports `google.adk` . The reason this matters: **the ADK Python SDK's API is still in motion.** Wrapping it lets me swap implementations later (sessions, eval harness, streaming) without touching `app.py` . The adapter runs retrieval *before* the ADK Runner call. This is the design choice worth understanding. This is the single most important architectural decision in the whole system, and it's the one most ADK tutorials get wrong. The "obvious" pattern in ADK is to expose retrieval as a tool: ``` # What I did NOT do. firestore_retrieval_tool = FunctionTool(firestore_retrieval) root_agent = Agent( name="logichat_faq_agent", model="gemini-2.5-flash", instruction=..., tools=[firestore_retrieval_tool], # model invokes at its discretion ) ``` I evaluated this and rejected it. Two reasons: **Threshold short-circuit.** When no chunk is close enough to the question, I want to skip the model call entirely and return `defaultAnswer` . ADK tools fire *inside* the model loop — by the time the tool runs, you've already paid the input-token cost. A tool-based design cannot enforce a "no useful context" short-circuit because the model has to decide to invoke the tool first. **Cross-tenant safety.** `app_id` must never become a model-controlled argument. Exposing it as a tool input would let a crafted prompt pivot retrieval to another tenant's collection. I scope retrieval server-side from the trusted caller (`config` / `RunnerProtocol.run` 's `app_id` ), not from the model. So retrieval lives in the adapter, *before* the ADK runner is invoked: ``` python # apps/agent/src/agent/agent_definition.py async def run(self, *, app_id, session_id, question, config): # 1. Retrieval — pre-model, NOT an ADK tool chunks = await self._retrieve(app_id=app_id, question=question) if not chunks: # No useful context → short-circuit to defaultAnswer. # Also covers the case where retrieval raised: _retrieve returns # an empty list and a flag on failure. return RunnerResult(answer=_default_answer(config), grounded=False, retrieval=[]) # 2. Compose the grounded user message chunk_block = "\n\n".join(_format_chunk(c) for c in chunks) user_text = f"{chunk_block}\n\nUser question: {question.strip()}" # 3. Run the ADK flow against the grounded message runner = self._adk_runner user_id = app_id # one ADK user per LogiChat app session = await runner.session_service.get_session( app_name=runner.app_name, user_id=user_id, session_id=session_id ) if session is None: await runner.session_service.create_session( app_name=runner.app_name, user_id=user_id, session_id=session_id ) message = types.Content(role="user", parts=[types.Part(text=user_text)]) final_text = "" async for event in runner.run_async( user_id=user_id, session_id=session_id, new_message=message ): if event.is_final_response() and event.content and event.content.parts: final_text = "".join(part.text or "" for part in event.content.parts) return RunnerResult( answer=sanitize_answer(final_text), grounded=True, retrieval=[ RetrievalHit( chunk_id=c.chunk_id, document_id=c.document_id, distance=c.distance ) for c in chunks ], ) ``` Three things worth pointing out: `asyncio.to_thread` `google-cloud-firestore` and `google-genai` calls so a slow Firestore round-trip doesn't block the uvicorn event loop. ADK itself is async-native; the retrieval layer is the only sync part of the request path.`_retrieve` swallows exceptions, logs a warning, returns `[]` . The request still succeeds via the ungrounded ADK flow. A transient Firestore outage degrades answer quality; it doesn't take the service down.`RunnerResult` carries `grounded` + `retrieval` `answer` . The gateway logs retrieval metadata on the conversation doc so I can evaluate retrieval quality offline. The original ADK Runner contract only returns text; I extended it through my adapter.`apps/agent/src/agent/retrieval.py` is small on purpose — it owns the two operations the adapter needs: embed the question, pull the most relevant chunks. Nothing else in the agent service touches Firestore or the embedding API directly. ``` EMBEDDING_MODEL = "text-embedding-005" EMBEDDING_DIMENSION = 768 DEFAULT_DISTANCE_THRESHOLD = 0.5 # COSINE: 0 = identical, 2 = opposite DEFAULT_TOP_K = 5 ``` The KNN search uses COSINE distance: ``` # NB: neither symbol is re-exported from the package roots. Both # raise ImportError against the real google-cloud-firestore 2.x # install. Don't "fix" this to the package root — it'll silently # fall through to the ungrounded fallback on every request. from google.cloud.firestore_v1.base_vector_query import DistanceMeasure from google.cloud.firestore_v1.vector import Vector query = db.collection(f"logichat-apps/{app_id}/chunks").find_nearest( vector_field="embedding", query_vector=Vector(query_vector), distance_measure=DistanceMeasure.COSINE, limit=limit, distance_result_field="vector_distance", ) ``` The threshold is applied in Python, after KNN returns. The reason: `find_nearest` returns the nearest `limit` chunks by distance; the threshold is a separate concern (recall vs precision). Conflating them in the query means every app owner has to know the right top-k to balance both. Better to keep the parameters independent. **Both write path and read path must agree on dimension.** The KNN index in `firestore.indexes.json` is built for 768 dimensions, flat index: ``` { "collectionGroup": "chunks", "queryScope": "COLLECTION", "fields": [ { "fieldPath": "embedding", "vectorConfig": { "dimension": 768, "flat": {} } } ] } ``` Drift here is silent and catastrophic — a chunk embedded with a 1536-dim model stored against a 768-dim index either fails the write or, worse, produces results that look reasonable but are nonsense. I pin `EMBEDDING_MODEL` and `EMBEDDING_DIMENSION` as module constants and fail loud if the API returns the wrong shape: ``` if len(values) != EMBEDDING_DIMENSION: raise RuntimeError( f"embedding model {EMBEDDING_MODEL} returned {len(values)} dims, " f"expected {EMBEDDING_DIMENSION}" ) ``` The agent's job is to answer questions. The doc-processor's job is to make sure there *are* chunks to answer from. `apps/subscribers/doc-processor` is a Cloud Run service subscribed to `google.cloud.storage.object.v1.finalized` events on the project bucket. Trigger filter is the path prefix `apps/{appId}/docs/{documentId}/{filename}` — anything outside that prefix is ignored silently. Why drive off the Storage event instead of a Firestore write? Because the API writes the `pending` doc *before* it returns the signed URL, so the metadata can lag the actual upload by seconds. Driving off Storage means I wait for the real bytes. When the event fires: ``` // apps/subscribers/doc-processor/src/app/doc-processor.handler.ts const [buffer] = await storage.bucket(event.bucket).file(event.name).download() const parsedDoc = await parseDocument(buffer, mimeType) const chunks = chunkText(parsedDoc.text) const vectors = await embedBatch(chunks) await persistChunks(appId, documentId, uid, chunks, vectors, embeddingModel) await markReady(appId, documentId, chunks.length, embeddingModel) ``` Three of those lines hide real choices. **Chunker ( libs/server/ai/src/lib/chunker.utils.ts).** 800-token chunks with 100-token overlap. Splits on paragraph boundaries first, falls back to sentence boundaries when a paragraph overflows, and applies overlap by prepending the tail of the previous chunk to the next. Pure function in `*.utils.ts` — no injected dependencies, no lifecycle, that's the repo convention.**Parser.** Plain text, markdown, HTML, and PDF. The HTML parser uses cheerio and **strips