Agentic AI on Google Cloud, Part 3: Making Answers Verifiable with Claim Checking and Citations
The assistant can now look up a claim and search the policy. But a fluent answer is not a correct answer. This post adds the layer that checks what the assistant says against the evidence it actually retrieved, and attaches citations a customer can inspect.
The series. Part 1: the GCP map and the first agent. Part 2: knowledge preparation, vectorization and automated ingestion. Part 3 (this post): verification and citations. Part 4: testing, CI/CD and monitoring.
1. Why a correct tool is not enough
Suppose the policy passage the agent retrieved says: "Outpatient consultations are reimbursed at 80%, subject to the applicable annual limit." The agent writes this to the customer:
Every tool worked. The claim lookup was right and the retrieval found the right clause. Yet the answer contains two errors: it invents "no annual limit", and it promises a payment nobody can promise. Language models do this because they write the most plausible text, not the most faithful text. Better prompts reduce it but never remove it, so we add a control the model cannot talk its way around.
There are several distinct ways an answer can be wrong, and each needs its own check:
| Failure | Example | Caught by |
|---|---|---|
| Unsupported claim | "No annual limit" | Claim-to-evidence check |
| Wrong source cited | Cites an old policy version as current | Citation and version check |
| Misread clause | 80% rate read as 80% chance of approval | Model-based check |
| Fabricated tool result | "I checked the claim" when the tool failed | Evidence ledger |
| Wrong arithmetic | Ignores the remaining annual limit | Deterministic calculator |
| Overpromise | "You will be paid" | Response policy |
2. The idea, and where it lives on GCP
The principle is simple: nothing the customer reads should be a factual statement we cannot trace to evidence the tools returned. We build this as a pipeline between the agent's draft and the user.
On GCP there is no separate "verification service" to deploy. The verifier is a Python module inside the same Cloud Run API service from Part 1. It uses two things you already have: Gemini on Vertex AI (called with the same api-sa identity, so no new permissions) for extracting claims and for judging nuanced ones, and Cloud Logging to record every verification result. A log sink to BigQuery will turn those records into the reliability dashboard in Step 8.
3. Hands-on: build the verification layer
Seven steps, in dependency order. We first create the evidence record, then the model-facing rules for citing, then the checks, then the decision logic, and finally wire everything into the API and measure it.
Step 1. Add a settings value and the evidence schema
The next job: define what "evidence" is. A single typed object makes every later step simple, because each check receives the same shape no matter which tool produced it. We also add a model name for the verifier; using a smaller, cheaper Gemini model here is usual, since the work is short and structured.
# src/common/config.py (add one field)
verifier_model: str # a smaller Gemini model offered in your region
# src/verify/evidence.py
from pydantic import BaseModel
class Evidence(BaseModel):
source_id: str # stable id, e.g. "claims:CLM-1042" or a chunk id
kind: str # "api" or "policy"
title: str # shown to the customer
version: int | None = None
section: str | None = None
page: int | None = None
retrieved_at: str
data: dict # structured record, or {"text": passage}Step 2. Fill the evidence ledger automatically
The next job: record every tool result as it happens. ADK lets you attach a callback that runs after each tool call. The callback copies the result into session state under a stable id. The model cannot edit it, which is exactly the point: this is the ground truth we will check the answer against.
# src/agent/callbacks.py
from datetime import datetime, timezone
def record_evidence(tool, args, tool_context, tool_response):
now = datetime.now(timezone.utc).isoformat()
ledger = list(tool_context.state.get("evidence", []))
if tool.name == "get_claim_status" and tool_response.get("success"):
ledger.append({"source_id": f'claims:{tool_response["claim_id"]}', "kind": "api",
"title": "Claims system", "retrieved_at": now, "data": tool_response})
if tool.name == "search_policy_documents":
for p in tool_response.get("passages", []):
ledger.append({"source_id": p["source_id"], "kind": "policy", "title": "Health policy",
"version": p["version"], "section": p["section"], "page": p["page"],
"retrieved_at": now, "data": {"text": p["text"]}})
tool_context.state["evidence"] = ledger
return None # leave the tool response unchangedAttach it to the agent from Part 1 with after_tool_callback=record_evidence, and extend the instruction so the model cites ids in square brackets: "After every factual statement, cite its evidence as [source_id], for example [claims:CLM-1042]. Only cite ids returned by tools. Never promise a reimbursement." ADK's callback signature and state handling can differ between versions, so check the documentation for yours.
Check: run a question in adk web and open the session state. You should see an evidence list with one entry for the claim and several for policy passages.
Step 3. Split the draft into single claims
The next job: break a paragraph into pieces we can check one by one. A sentence like "rejected because the invoice is missing and you are covered at 80%" contains several independent facts, and one of them may be wrong while the rest are fine. We ask Gemini to do the splitting and to return strict JSON so the code can rely on the shape.
# src/verify/claims.py
import os
from google import genai
from google.genai import types
from pydantic import BaseModel
from src.common.config import settings
client = genai.Client(vertexai=True, project=os.environ["GOOGLE_CLOUD_PROJECT"],
location=os.environ["GOOGLE_CLOUD_LOCATION"])
class AtomicClaim(BaseModel):
text: str
cited: list[str] # source ids attached in the draft
kind: str # "status", "policy_fact", "amount", "advice"
value: str | None = None # for status claims: the stated status, e.g. "rejected"
class ClaimList(BaseModel):
claims: list[AtomicClaim]
def extract_claims(draft: str) -> list[AtomicClaim]:
r = client.models.generate_content(
model=settings.verifier_model,
contents=("Split the answer into single, self-contained factual claims. Keep the "
"[source ids] cited for each claim. Mark promises and recommendations as "
"kind=advice.\n\nANSWER:\n" + draft),
config=types.GenerateContentConfig(response_mime_type="application/json",
response_schema=ClaimList, temperature=0))
return r.parsed.claimsStep 4. Check claims: deterministic first, model second
The next job: label each claim supported, unsupported or contradicted. We follow one rule: wherever a claim maps to a structured field, compare it with code, because code does not get tired or persuaded. Only claims that need reading comprehension go to a model. Unsupported does not mean false; it means the evidence does not establish it.
# src/verify/check.py
from pydantic import BaseModel
from src.verify.claims import client, AtomicClaim
from src.verify.evidence import Evidence
from google.genai import types
from src.common.config import settings
class Verdict(BaseModel):
label: str # supported | unsupported | contradicted
reason: str
class Checked(BaseModel):
claim: AtomicClaim
label: str
method: str # deterministic | model
reason: str
def check_claim(c: AtomicClaim, ledger: dict[str, Evidence]) -> Checked:
if c.kind == "advice":
return Checked(claim=c, label="unverifiable", method="deterministic", reason="advice, not a fact")
missing = [s for s in c.cited if s not in ledger]
if not c.cited or missing: # no citation, or an id no tool returned
return Checked(claim=c, label="unsupported", method="deterministic",
reason=f"no valid source ({missing or 'none cited'})")
api = [ledger[s] for s in c.cited if ledger[s].kind == "api"]
if c.kind == "status" and api and c.value: # compare to the field, no model needed
ok = c.value.strip().casefold() == api[0].data["status"].strip().casefold()
return Checked(claim=c, label="supported" if ok else "contradicted",
method="deterministic", reason=f'record says {api[0].data["status"]}')
evidence = "\n".join(f'[{s}] {ledger[s].data.get("text", ledger[s].data)}' for s in c.cited)
r = client.models.generate_content(
model=settings.verifier_model,
contents=(f"EVIDENCE:\n{evidence}\n\nCLAIM: {c.text}\n\nDoes the evidence support the claim, "
"contradict it, or fail to establish it? Be strict: a claim that adds a detail "
"not in the evidence is unsupported."),
config=types.GenerateContentConfig(response_mime_type="application/json",
response_schema=Verdict, temperature=0))
return Checked(claim=c, label=r.parsed.label, method="model", reason=r.parsed.reason)Step 5. Decide what the customer sees
The next job: turn labels into action. A detector that finds problems but still shows the answer protects nobody. The decision policy is a small, testable function; the response is fixed, corrected once, or replaced by an honest "I can't confirm that".
# src/verify/policy.py
from src.verify.claims import extract_claims, client
from src.verify.check import check_claim
from src.common.config import settings
from google.genai import types
ABSTAIN = ("I can't confirm that from the claim record and policy documents I can access. "
"Please contact an advisor, who can check it with you.")
def decide(checked) -> str:
labels = {c.label for c in checked}
if "contradicted" in labels: return "fix"
if "unsupported" in labels: return "fix"
return "allow"
def verify_answer(draft: str, ledger: dict, rewrite) -> tuple[str, str, list]:
checked = [check_claim(c, ledger) for c in extract_claims(draft)]
if decide(checked) == "allow":
return draft, "answered", checked
fixed = rewrite(draft, [c for c in checked if c.label in {"unsupported", "contradicted"}])
checked2 = [check_claim(c, ledger) for c in extract_claims(fixed)]
if decide(checked2) == "allow":
return fixed, "answered_after_fix", checked2
return ABSTAIN, "abstained", checked2 # still failing: do not guessdef rewrite_with_gemini(draft: str, flagged) -> str:
problems = "\n".join(f"- {c.claim.text} ({c.label}: {c.reason})" for c in flagged)
r = client.models.generate_content(model=settings.verifier_model, config=types.GenerateContentConfig(temperature=0),
contents=("Rewrite the answer so that it no longer contains the flagged claims, or states them "
"correctly. Keep every other sentence and its [source ids]. Add nothing new.\n\n"
f"FLAGGED:\n{problems}\n\nANSWER:\n{draft}"))
return r.textThe rewrite function (above) asks Gemini to produce a new answer that removes or corrects only the flagged claims, using the evidence. We allow one repair attempt and re-verify it, then stop. An unbounded loop would add latency and cost for no guarantee.
Step 6. Build the citations from evidence
The next job: show the customer where each fact came from. Citations are never typed by the model. We take the ids the verified answer uses and render them from the ledger's metadata, so a page number or version can only appear if a tool returned it.
# src/verify/cite.py
import re
def render(answer: str, ledger: dict):
order, sources = {}, []
def sub(m):
sid = m.group(1)
if sid not in ledger: return "" # drop ids no tool returned
if sid not in order:
order[sid] = len(order) + 1
e = ledger[sid]
sources.append({"citation_id": str(order[sid]), "type": e.kind, "title": e.title,
"version": e.version, "section": e.section, "page": e.page,
"record": sid if e.kind == "api" else None, "retrieved_at": e.retrieved_at})
return f"[{order[sid]}]"
return re.sub(r"\[([^\]]+)\]", sub, answer), sourcesThe final payload the frontend receives looks like this:
{ "answer": "Your claim CLM-1042 is rejected because an itemized invoice is missing [1]. Outpatient costs are reimbursed at 80%, subject to the annual limit [2]. I can't promise the outcome of a reassessment.",
"sources": [
{"citation_id":"1","type":"api","title":"Claims system","record":"claims:CLM-1042","retrieved_at":"2026-10-24T10:02:11Z"},
{"citation_id":"2","type":"policy","title":"Health policy","version":4,"section":"6.2 Outpatient","page":12} ],
"status": "answered" }Step 7. Wire it into the API
The next job: put the verifier between the agent and the response in the endpoint from Part 1. The endpoint now reads the ledger from the session after the run, verifies, renders citations and logs a structured record Cloud Logging will pick up.
# src/api/main.py (the chat handler, updated)
import json, sys
from src.verify.evidence import Evidence
from src.verify.policy import verify_answer
from src.verify.cite import render
# ... after the runner loop. Rename the variable `answer` from Part 1 to `draft` ...
s = await sessions.get_session(app_name="insurance", user_id=cust, session_id=sid)
ledger = {e["source_id"]: Evidence(**e) for e in s.state.get("evidence", [])}
text, status, checked = verify_answer(draft, ledger, rewrite=rewrite_with_gemini)
answer, sources = render(text, ledger)
print(json.dumps({"severity": "INFO", "event": "verification", "status": status,
"total": len(checked), "supported": sum(c.label == "supported" for c in checked),
"unsupported": sum(c.label == "unsupported" for c in checked),
"contradicted": sum(c.label == "contradicted" for c in checked)}), file=sys.stdout)
return {"answer": answer, "sources": sources, "status": status}Cloud Run sends JSON written to stdout to Cloud Logging as structured entries, so each field becomes queryable. Note what we log: counts and a status, not the customer's text.
Redeploy with the same build and gcloud run deploy commands from Part 1, adding AGENT_VERIFIER_MODEL to the environment variables. No new IAM roles are needed.
4. Try it, including the failures
Verification is only trustworthy if you have watched it catch real errors. Try these four on the deployed service or in adk web:
| Scenario | What the verifier should do |
|---|---|
| Ordinary question about CLM-1042 | All claims supported; status answered; two citations |
| Force an overstated draft ("no annual limit") | Claim labelled unsupported or contradicted; one repair; status answered_after_fix |
| Draft cites an id no tool returned | Claim unsupported; the fake id never appears in sources |
| Stop the claims API and ask again | No claims evidence, so status claims are unsupported; the answer says it cannot confirm, or abstains |
To force the second case without waiting for a model slip, add a temporary line in a test that replaces the draft with the overstated text and calls verify_answer directly. This is the first piece of the evaluation suite we build in Part 4.
5. Measure reliability, do not just hope
A verifier gives you a number you did not have before. Define a few metrics exactly and track them separately, because they measure different things.
| Metric | Definition | What it tells you |
|---|---|---|
| Claim support rate | supported / (supported + unsupported + contradicted) | How much of what the agent says is backed by evidence |
| Hallucination rate | (unsupported + contradicted) / all claims | The operational complement; report both parts separately |
| Citation correctness | cited claims whose source truly supports them / cited claims | Separate from whether the answer is right |
| Abstention rate | answers replaced by the "cannot confirm" message | Too high means retrieval is weak; too low may mean the verifier is lenient |
These are operational measures under your chosen verification procedure, not universal truths, and a high claim support rate can hide one critical error. Also, a model's self-reported confidence is not a probability until calibrated on labelled examples.
To turn the structured logs into a table you can chart, create a Log Router sink to BigQuery:
bq --location=$REGION mk --dataset $PROJECT:ops
gcloud logging sinks create verification-sink \
bigquery.googleapis.com/projects/$PROJECT/datasets/ops \
--log-filter='resource.type="cloud_run_revision" AND jsonPayload.event="verification"'
# grant the sink's writer identity (printed by the command) BigQuery Data Editor on dataset ops-- daily reliability (table name depends on your sink; adjust field types if needed)
SELECT DATE(timestamp) AS day,
SUM(CAST(jsonPayload.supported AS INT64)) / SUM(CAST(jsonPayload.total AS INT64)) AS claim_support_rate,
COUNTIF(jsonPayload.status = "abstained") / COUNT(*) AS abstention_rate
FROM `ops.run_googleapis_com_stdout`
GROUP BY day ORDER BY day6. Costs, limits and gotchas
| Topic | What to know |
|---|---|
| Latency and cost | The verifier adds at least two model calls per answer (extract, then judge nuanced claims). Use a smaller model, run deterministic checks first, and skip the model check when every claim is an API field |
| The verifier is a model | It can approve a subtle error or reject a good claim. Keep deterministic checks for anything structured, and review a sample of results by hand |
| Evidence size | Passing every retrieved passage to the judge can be large. Pass only the passages the claim cites |
| Streaming | Do not stream draft text to the customer as if it were final; show progress, then the verified answer |
| Structured output | Schemas limit what the model can return; invalid or empty output must be handled and treated as "unverified", never as "allowed" |
| Privacy | Log counts and statuses, not customer text or whole documents |
7. What you have and what is next
Every answer now passes through a gate. Tools fill an evidence ledger, the draft is split into claims, each claim is labelled by code or by a model, a small policy decides whether to allow, repair once or abstain, and citations are built from the ledger instead of from the model's imagination. The reliability numbers flow into BigQuery through a log sink.
Part 4 makes all of this testable and operable: a golden evaluation set (including the failure cases above) run in CI, security tests, Terraform for the whole environment, a canary release process, JWT validation and a shared session store, and monitoring and alerts built on the verification logs. Not covered here: calibration of confidence scores, a deterministic reimbursement calculator, and graph-based consistency checks. I have not run this code, so check the ADK callback signature and state API, the Gen AI SDK structured-output call, the model names, and the Log Router table names against current documentation.
References: Agent Development Kit, Vertex AI generative AI, Log Router sinks, Cloud Run logging.
Comments
Post a Comment