DEV Community: Hastimal Jangid

Building an AI Search Visibility & Brand Analyzer with Gemini, BigQuery, and Google Search Grounding

Hastimal Jangid — Tue, 29 Sep 2026 05:22:17 +0000

AI Search Journey Lab — Part 3 of 7

Measuring how brands appear across AI search journeys using Gemini, Google Search grounding, deterministic visibility scans, and BigQuery history.

Open source: ai-search-journey-lab

Previous: Building Grounded Local Search with Gemini and Google Maps on Google Cloud

The first two articles answered a user question. This one measures the system itself.

In Part 1, I focused on the search journey:

How does one natural-language request become a grounded decision?

In Part 2, I went deeper into the actual local-search implementation:

How do Gemini, Google Places, Search grounding, deterministic ranking, and Maps work together?

Once that worked, I started asking a different question.

Suppose I run the same kind of search repeatedly.

What happens if instead of only asking:

Which business should the user choose?

I ask:

Which brands keep appearing?
Which brands are cited?
Which competitors show up more often?
Does a brand appear in the initial answer, the fan-out journey, or only in supporting evidence?
How does that change over time?

That was the starting point for V3 — AI Visibility.

This article is about turning an AI search workflow into something measurable.

Why AI visibility needs a different data model

A single search result is transient.

If I want to measure visibility, I need history.

One run is interesting.

Fifty runs are analyzable.

Five hundred runs can start showing patterns.

So V3 introduced a new concern into the project: persistence

That is where BigQuery enters the architecture.

What I wanted to measure

For each visibility run, I wanted to capture things such as:

target brand
competitors
original query
fan-out tasks
mentions
citations
candidate coverage
source coverage
ranking position
execution timestamp
historical trend

The important shift was this:

V1/V2:
user query → recommendation

V3:
many queries → repeatable measurements

That turns the project from a search demo (which we have planned) into an analytics system. :)

Architecture

Step 1: I made visibility scans deterministic

One important design choice was that V3 should not rely on a user manually clicking around and interpreting the results.

I wanted repeatable runs.

That means the runner should accept an explicit input scope and execute the same workflow consistently.

Conceptually:

run = VisibilityRun(
    target_brand="Example Brand",
    competitors=[
        "Competitor A",
        "Competitor B",
    ],
    queries=[
        "best local coffee shops in San Antonio",
        "quiet coffee shop for remote work",
    ],
)

Then the runner executes the scan and records the results.

The goal is reproducibility.

If I run the same scenario tomorrow, I want to compare the resulting data rather than rely on screenshots or memory.

Step 2: Separate mention detection from visibility scoring

A brand appearing in a response is not the same as a brand being strongly visible.

For example:

Target brand:
RankRabbit

Response:
"Other platforms include RankRabbit, Competitor A, and Competitor B."

That is a mention.

But now compare:

1. RankRabbit — recommended first
2. Competitor A
3. Competitor B

Those two appearances should not necessarily have the same visibility weight.

So I separate several concepts:

mention presence
position
citation presence
fan-out coverage
narrative prominence

That lets the application calculate visibility metrics explicitly instead of collapsing everything into one LLM judgment.

Step 3: Capture evidence before calculating visibility

The workflow should preserve what caused a brand to be counted.

Conceptually:

{
  "brand": "Example Brand",
  "mentioned": true,
  "rank": 2,
  "citation_present": true,
  "fanout_coverage": 0.67,
  "evidence_sources": [
    "source-a",
    "source-b"
  ]
}

That data is much more valuable later than a single score such as:

visibility_score = 74

because it lets me explain where the score came from.

Step 4: Persist visibility history in BigQuery

V3 can run in-memory for a quick demo, but historical analysis needs persistence.

I used a BigQuery dataset for that purpose:

Project:
ai-search-journey-lab

Dataset:
ai_search_journey_v3

Let's explore how bigQuery tables look like.....

bq query \
  --use_legacy_sql=false \
  '
  SELECT
    table_name,
    table_type,
    creation_time
  FROM
    `ai-search-journey-lab.ai_search_journey_v3.INFORMATION_SCHEMA.TABLES`
  ORDER BY
    table_name
  '

Step 5: Query historical visibility with SQL

Once I started persisting V3 runs in BigQuery, I no longer had to rely only on what the Streamlit dashboard displayed.

I could query the visibility history directly from the terminal.
That became one of my favorite parts of V3 because it gave me a second way to validate everything the UI was showing.

For all of the examples below, I use the BigQuery CLI:

Once scan results are persisted, SQL becomes one of the most useful tools in the project.

Building Grounded Local Search with Gemini and Google Maps on Google Cloud

Hastimal Jangid — Tue, 29 Sep 2026 04:09:07 +0000

AI Search Journey Lab — Part 2 of 7

Building a grounded local-search workflow with Gemini, Google Places API (New), Google Search grounding, deterministic ranking, Maps links, and Cloud Run.

Open source: ai-search-journey-lab

Previous article: Search Journey Optimization with Gemini: From Query Fan-Out to Grounded Decisions

In Part 1, I explained the journey. In Part 2, I want to show the implementation.

In the first article, I focused on the architecture behind a search journey:

intent → query fan-out → retrieval → evidence → deduplication → ranking → grounded answer

That was the conceptual layer.

This article is more practical.

I want to show how I built the actual local-search workflow behind that architecture using:

Gemini
Google Places API (New)
Google Search grounding
Google Maps links
deterministic ranking
Python
Streamlit
Cloud Run

The original demo query is still the same:

Find a coffee shop near Geekdom in San Antonio for six people, quiet enough to work, open after 8 PM, and recommend the top three.

The difference now is that I want to focus on what happens after the intent has been understood.

The local-search problem is really two retrieval problems
When I first started implementing this workflow, I realized that one retrieval system was not enough.

Some constraints are structured.

For example:

business name
address
rating
review count
opening hours
Place ID
Maps URL

Google Places is a very good fit for those.

But some user requirements are much softer:

quiet enough to work
suitable for a group
good vegetarian options
comfortable for studying
appropriate for a child who is anxious about dental visits

Those do not always map cleanly to one structured field.

That led me to split retrieval into two paths:

Google Places for structured local data

and

Google Search grounding for additional evidence

That separation is at the heart of this article.

The architecture

The V1 flow for grounded local search looks like this:

User Query → Gemini Intent → Query Fan-Out → Places API + Google Search Grounding → Normalize → Evidence → Score → Gemini Synthesis → Maps-linked Recommendations

Step 1: Start from structured intent

I do not call Places directly with the full user prompt.

First, I want a structured representation of what the user is asking for.

Conceptually:

{
  "category": "coffee_shop",
  "location_reference": "Geekdom, San Antonio",
  "party_size": 6,
  "open_after": "20:00",
  "preferences": [
    "quiet",
    "work-friendly"
  ],
  "result_count": 3
}

The important thing here is that the retrieval layer no longer has to interpret every phrase from scratch.

Gemini handles the language ambiguity.

The application gets a predictable structure.

Step 2: Turn intent into retrieval tasks

Once I have the intent, I generate a bounded retrieval plan.

A simplified example might look like:

{
  "places_tasks": [
    {
      "query": "coffee shops near Geekdom San Antonio"
    },
    {
      "query": "work-friendly coffee shops downtown San Antonio"
    }
  ],
  "search_tasks": [
    {
      "query": "coffee near Geekdom quiet work open late"
    }
  ]
}

I deliberately keep this small.

The goal is not to generate ten variants of the same search.

The goal is to cover enough of the user's intent without creating unnecessary duplication and latency.

Step 3: Google Places becomes the structured candidate source

For local search, Google Places API (New) gives me the first candidate set.

A simplified Text Search call looks like this:

import requests


def search_places(query: str, api_key: str) -> list[dict]:
    endpoint = "https://places.googleapis.com/v1/places:searchText"

    payload = {
        "textQuery": query,
        "pageSize": 10,
    }

    headers = {
        "Content-Type": "application/json",
        "X-Goog-Api-Key": api_key,
        "X-Goog-FieldMask": ",".join(
            [
                "places.id",
                "places.displayName",
                "places.formattedAddress",
                "places.rating",
                "places.userRatingCount",
                "places.currentOpeningHours",
                "places.googleMapsUri",
            ]
        ),
    }

    response = requests.post(
        endpoint,
        json=payload,
        headers=headers,
        timeout=20,
    )

    response.raise_for_status()

    return response.json().get("places", [])

What matters most to me here is not the HTTP request itself.

It is the set of fields that become available to downstream logic.

I use field masks intentionally

For this workflow, I do not need every possible field.

I only want the fields that are useful for the decision.

For example:

places.id
places.displayName
places.formattedAddress
places.rating
places.userRatingCount
places.currentOpeningHours
places.googleMapsUri

This keeps the retrieval boundary explicit.

It also makes it easier to understand which parts of the final recommendation came directly from Places data.

Step 4: Place ID becomes the anchor for identity

One thing I did not appreciate enough at the beginning was how important entity identity would become.

Suppose two fan-out tasks return the same business.

Without normalization, I might end up with:

Candidate A
Candidate A
Candidate B
Candidate C

That creates a ranking problem.

So I use a canonical identifier such as Place ID to determine whether I am looking at the same entity.

A simplified deduplication step looks like:

def deduplicate_places(candidates: list[dict]) -> list[dict]:
    unique: dict[str, dict] = {}

    for candidate in candidates:
        place_id = candidate["place_id"]

        if place_id not in unique:
            unique[place_id] = candidate
            continue

        unique[place_id] = merge_candidate(
            unique[place_id],
            candidate,
        )

    return list(unique.values())

The order matters:

retrieve → normalize identity → deduplicate → rank

NOT

retrieve → rank duplicates → fix identity later

Step 5: Generate Google Maps links as part of the decision output

One thing I wanted from the beginning was that the recommendation should not end as plain text.

If the user asks for a local business, the result should be actionable.

That is why I keep the Google Maps URI returned by Places.

Conceptually:

candidate = {
    "name": place["displayName"]["text"],
    "address": place["formattedAddress"],
    "rating": place.get("rating"),
    "maps_url": place.get("googleMapsUri"),
}

Then the final recommendation can give the user a direct route from:

AI recommendation

Google Maps

That sounds small, but it changes the experience from “read an AI answer” to “act on an AI answer.”

Step 6: Places alone did not solve the problem

Now we reach the interesting part.

Google Places can tell me:

the business exists
where it is
its rating
its opening hours
its Maps URL

But the original user also asked for:

quiet enough to work

That is a different class of requirement.

There may not be a clean structured field for it.

So I added a second evidence path using Google Search grounding.

Step 7: Ground the softer constraints

I do not ask Gemini a broad question like:

Is this coffee shop suitable?

I make the verification task more specific.

For example:

Candidate:
Example Coffee

Verify:
1. evidence that it is open late
2. evidence relevant to working/studying
3. evidence relevant to seating/group suitability

For each constraint:
- supported
- unsupported
- supporting evidence
- citation

Conceptually, the call sits behind a workflow stage like:

with trace_span(
    "search_grounding.verify_evidence",
    attributes={"workflow.stage": "v1"},
):
    evidence = verify_candidates(
        candidates=candidates,
        intent=intent,
    )

This gives me a richer candidate representation.

A candidate now has two kinds of evidence

Conceptually:

{
  "place_id": "abc123",
  "name": "Example Coffee",
  "places": {
    "rating": 4.6,
    "review_count": 825,
    "open_after_8": true
  },
  "search_evidence": {
    "work_friendly": true,
    "group_suitability": null
  }
}

That null is useful.

If I cannot verify group suitability, I do not want the system to quietly convert uncertainty into confidence.

That is one of the most important lessons I took from this build:

Unsupported should remain unsupported.

Step 8: I keep evidence matching separate from synthesis

Another decision I made was to avoid letting the final LLM call discover and interpret everything again from scratch.

By the time Gemini reaches the final synthesis stage, the system already has:

structured intent
candidate identities
Places data
grounded Search evidence
unsupported constraints
scores
rank order

This makes the final prompt much narrower.

Step 9: Deterministic ranking before final synthesis

I deliberately keep ranking outside Gemini.

A simplified workflow looks like:

with trace_span(
    "evidence.aggregate_and_score",
    attributes={"workflow.stage": "v1"},
):
    ranked_candidates = aggregate_and_score(
        candidates=candidates,
        evidence=evidence,
        intent=intent,
    )

The scoring logic can consider signals such as:

location relevance
category match
opening-hour compatibility
rating
rating count
constraint coverage
grounded evidence
unsupported constraints

The point is not that one universal formula can rank every local-search problem.

The point is that the system can explain why one candidate scored higher than another.

def build_static_map_url(
    ranked_candidates: list[RankedCandidate],
    *,
    api_key: Optional[str] = None,
    max_candidates: int = 3,
    width: int = 640,
    height: int = 360,
    maptype: str = "roadmap",
) -> str:
    """Build a Google Maps Static API URL for the top ranked candidates.

    Args:
        ranked_candidates: List of ranked candidates.
        api_key: Optional Google Maps API Key override.
        max_candidates: Maximum candidate markers to place (default: 3).
        width: Image width in pixels (default: 640).
        height: Image height in pixels (default: 360).
        maptype: Map type (default: 'roadmap').

    Returns:
        Fully-formed Static Maps URL including API key.

    Raises:
        ValueError: If GOOGLE_MAPS_API_KEY is not configured or no valid coordinates exist.
    """
    result = generate_static_map(
        ranked_candidates,
        api_key=api_key,
        max_candidates=max_candidates,
        width=width,
        height=height,
        maptype=maptype,
    )
    return result.url

View the complete scoring implementation on GitHub

Step 10: Gemini handles the final explanation

Gemini comes back near the end.

Conceptually:

with trace_span(
    "gemini.synthesize_recommendations",
    attributes={"workflow.stage": "v1"},
):
    answer = synthesize_recommendations(
        intent=intent,
        ranked_candidates=ranked_candidates,
    )

By this point, Gemini is not being asked to search the world.

It is being asked to explain a decision whose evidence is already available.

That is a much smaller and more controllable problem.

What the final answer should contain

For each recommendation, I want the output to include useful decision context.

For example:

1. Example Coffee
   Rating: 4.6
   Open after 8 PM: Yes
   Work-friendly evidence: Supported
   Group suitability: Not fully verified
   Maps: [Open in Google Maps]

That is much more useful than:

“I recommend Example Coffee because it looks great.”

Step 11: Validate the architecture with another query

I did not want the workflow to work only for the one query it was designed around.

So I also tested:

Find an Indian restaurant near Trinity University for eight students, open after 9 PM, with vegetarian options. Recommend the top three.

This changes several constraints:

category
geographic anchor
party size
hours
dietary preference

But the architecture stays the same.

That is important.

I want a reusable search workflow, not a prompt-specific demo.

Step 12: Test the workflow like normal software

I keep normal engineering checks around the AI workflow.

My standard validation is:

./.venv/bin/ruff check .
./.venv/bin/mypy src
./.venv/bin/pytest -q

The point is simple:

Gemini does not eliminate the need for testable software.

If anything, combining model behavior with APIs, ranking logic, state, and deployment makes testing more important.

Step 13: Run the same workflow on Cloud Run

Once the local implementation was stable, I deployed the same application to Cloud Run.

That gave me a clean path from local development to a hosted demo.

You can inspect the service with:

gcloud run services describe ai-search-journey-lab \
  --region us-central1 \
  --format="yaml(
      metadata.name,
      status.url,
      status.latestReadyRevisionName,
      status.conditions
  )"

And verify health:

curl -fsS \
"https://ai-search-journey-lab-642110324230.us-central1.run.app/_stcore/health"

Let's run locally!

Let's run the same Geekdom query from the deployed Google cloud service.

What I learned from building the local-search layer

The biggest lesson was that local AI search is not just:

prompt → model → answer

It is closer to:

natural-language intent → retrieval plan → entity search → evidence search → identity normalization → constraint verification → deterministic ranking → grounded explanation → actionable Maps result

That distinction is important.

Because once the application has multiple evidence sources and multiple ranking signals, the LLM becomes one component in the system rather than the entire system.

That is the architecture I wanted.

Why this matters for agentic systems

This local-search workflow also became a useful foundation for the later agent work in the repository.

Once the system already has:

structured intent
tools
explicit retrieval
evidence
state
ranking
bounded outputs

it becomes much easier to add agent orchestration later.

That is where Google ADK eventually enters the story in Part 4.

But I deliberately did not start there.

I first wanted a workflow I could understand without an agent.
Then I could add the agent layer intentionally.

Next: Building an AI Search Visibility & Brand Analyzer with Gemini, BigQuery, and Google Search Grounding

In the next article, I will move from:

Which business should this user consider?

to:

How visible is a brand across AI-generated search journeys?

I will cover:

target brands
competitors
visibility runs
mentions
citations
fan-out coverage
deterministic visibility analysis
BigQuery history
visibility trends

Try the project

Repository: GitHub Repository

Local startup:

python scripts/run_app_locally.py

Search Journey Optimization with Gemini: From Query Fan-Out to Grounded Decisions

Hastimal Jangid — Mon, 28 Sep 2026 21:29:53 +0000

AI Search Journey Lab — Part 1 of 7

Building an explainable search-to-decision workflow with Gemini, Google Places, Google Search grounding, deterministic ranking, and Google Cloud.

Open source: ai-search-journey-lab

It started with one search query
I started ai-search-journey-lab with a question that looked simple enough:

Find a coffee shop near Geekdom in San Antonio for six people, quiet enough to work, open after 8 PM, and recommend the top three.

At first, I thought of it as a local-search problem.

Then I started implementing it.

Almost immediately, that single sentence turned into several separate engineering problems.

I needed to understand what the user actually meant by “near Geekdom.” I needed to capture the group size. I needed opening hours. I needed to find places that might actually work for six people sitting together.

Then there was the phrase:

quiet enough to work

That is where things became more interesting.

A structured Places response can give me things like an address, rating, hours, and a Place ID.

But “quiet enough to work” is not simply another field I can request.

So now I needed both structured place data and additional evidence.

Then, once I started generating more than one retrieval query, another problem appeared: the same place could be returned by multiple searches.

That meant I also needed:

candidate normalization,
deduplication,
evidence matching,
scoring,
ranking,
and finally, a way to explain why one candidate beat another.

Oh gosh..It's getting a lot!!

At that point, I wasn't building a “send a prompt to Gemini and print the answer” application anymore.

I was building a search journey. (and later think about search journey optimization!)

And that became the idea behind AI Search Journey Lab.

AI Search Journey Lab: one project, seven engineering stages

This is the first article in a seven-part series built from the same open-source repository.

I wanted the series to follow the way the project itself evolved instead of creating seven unrelated AI demos.

The topics are also centered around the problems I am most interested in as a Google Cloud and AI developer: grounding, search behavior, data, agents, observability, and production deployment.

Search Journey Optimization with Gemini: From Query Fan-Out to Grounded Decisions (this article)
Building Grounded Local Search with Gemini and Google Maps on Google Cloud
Building an AI Search Visibility & Brand Analyzer with Gemini, BigQuery, and Google Search Grounding
Building an AI Visibility Agent with Gemini, Google ADK, MCP, and BigQuery
From Query Fan-Out to Trace: Observing AI Search Journeys with OpenTelemetry
Building AgentOps for Gemini Agents with Grafana, Tempo, and Prometheus
From Search to Care: Applying Gemini Search Journeys to Healthcare Discovery

The progression looks roughly like this:

search journey → grounded retrieval → visibility measurement → agentic analytics → tracing → AgentOps → healthcare application

My goal with the series is not just to show what Gemini can generate.

I want to show how the pieces around the model matter just as much: APIs, deterministic logic, persistence, tools, traces, tests, and deployment.

Starting from the actual problem

Let's go back to the original request:

Find a coffee shop near Geekdom in San Antonio for six people, quiet enough to work, open after 8 PM, and recommend the top three.

From an application perspective, that contains several different constraints:

category: coffee shop
geographic anchor: Geekdom
location: San Antonio
party size: six
operating-hours requirement: open after 8 PM
qualitative preference: quiet/work-friendly
result count: three

And not all of those constraints should be handled the same way.

For example:

Opening hours can often come from structured place data.
Quiet enough to work may require additional evidence.

That distinction ended up shaping most of the architecture.

The architecture decision that mattered most

One decision influenced almost everything that came afterward:

I did not want Gemini to own the entire workflow.

It would have been very easy to send the whole prompt to a model and ask:

Find the best three places and explain why.

That would certainly produce an answer.

But it would make several things harder for me:

understanding which constraints were actually verified,
seeing where candidates came from,
knowing why something ranked highly,
reproducing a decision,
debugging wrong answers,
and eventually tracing the system.

So I divided the workflow into three responsibilities.

Gemini reasoning
I use Gemini where natural-language interpretation is valuable:

intent extraction,
planning,
query fan-out,
final response synthesis.

Retrieval
I use external data sources for evidence:

Google Places API (New),
Google Search grounding.

Deterministic application logic
I keep things such as these in normal application code:

normalization,
entity matching,
deduplication,
evidence aggregation,
scoring,
ranking.

That gave me a much more inspectable pipeline.

Running the project

I normally start the local environment with:

python scripts/run_app_locally.py

As the project evolved, I added observability components as well, so the local startup script eventually became responsible for more than just Streamlit.

For this article, though, I am concentrating on two parts:

V1 — Search to Decision
V2 — Journey Analysis

The later articles will move into visibility, agents, and observability.

Step 1: I first turn the prompt into structured intent

Before I search for anything, I need to understand what the request actually contains.

Conceptually, the Geekdom query becomes something like:

{
  "category": "coffee_shop",
  "location_reference": "Geekdom, San Antonio",
  "party_size": 6,
  "open_after": "20:00",
  "preferences": [
    "quiet",
    "work-friendly"
  ],
  "result_count": 3
}

The important part here is not the JSON.

It is the boundary between language and application state.

People do not normally talk in schemas.

A user might say:

Somewhere close to Geekdom. We have six people, don't want somewhere too loud, and we'll probably stay late.

The application still needs to understand:

{
  "party_size": 6,
  "noise_preference": "quiet",
  "minimum_close_time": "20:00"
}

That is a good job for Gemini.

The model interprets the language.

My application controls the structure.

A typed intent model

Since I am more into development and technical architecture role, I prefer moving the result into a typed object rather than passing free-form model text deeper into the workflow.

A simplified version looks like this:

from pydantic import BaseModel, Field


class SearchIntent(BaseModel):
    category: str
    location_reference: str
    party_size: int | None = None
    open_after: str | None = None
    preferences: list[str] = Field(default_factory=list)
    result_count: int = 3

That gives me a cleaner boundary:

Natural-language request → Gemini interpretation → Structured application state

From this point forward, the rest of the pipeline has something predictable to work with.

I also made this a real workflow boundary

I didn't want intent extraction buried somewhere inside one large function.

I gave the stage its own observable boundary.

Conceptually:

with trace_span(
    "gemini.extract_intent",
    attributes={"workflow.stage": "v1"},
):
    intent = extract_intent(user_query)

That may look like a small detail here.

It became much more useful later when I added OpenTelemetry.

A lesson I learned while building this project was:

Good architecture boundaries often become good observability boundaries later.

As you run locally, it should open in http://localhost:8502/

Let's see the journey now and wait to trace the results!

Step 2: One prompt becomes several retrieval tasks

Once I knew what the user wanted, I ran into the next question:

What exactly should I search for?

Using the complete user sentence as one search query would throw many different requirements into the same retrieval call.

Instead, I started decomposing the request into a small retrieval plan.

A simplified fan-out might look like:

Places task:

coffee shops near Geekdom San Antonio

Places task:

work-friendly coffee shops near downtown San Antonio

Search evidence task:

coffee near Geekdom quiet work open late

In the project, I treat that planning step separately:

with trace_span(
    "gemini.plan_fan_out",
    attributes={"workflow.stage": "v1"},
):
    fanout_plan = plan_fan_out(intent)

This is the part I refer to as query fan-out.

One user's decision question becomes multiple narrower retrieval tasks.

I learned quickly that more queries are not automatically better

Initially, fan-out sounds like a simple idea:

If one query is useful, several queries should be better.

But that creates its own problems.

Imagine generating all of these:

coffee near Geekdom
quiet coffee near Geekdom
coffee for six near Geekdom
coffee with large tables near Geekdom
coffee open after 8 near Geekdom
coffee good for working near Geekdom
late-night coffee downtown San Antonio

Now I have:

more requests,
more latency,
more overlapping results,
more duplicates,
more evidence to reconcile,
and more opportunities for noisy candidates.

The opposite extreme is also bad:

coffee near Geekdom

because most of the intent has disappeared.

So query fan-out became an optimization problem of its own:

Generate enough retrieval tasks to cover the user's important constraints without turning the workflow into uncontrolled query expansion.

That is a much more interesting problem than simply asking a model to generate ten related searches.

Step 3: Google Places gave me the structured candidate layer

Once I had the fan-out plan, I needed actual entities.

For the local-search portion of the workflow, that meant Google Places API (New).

This gave me structured information such as:

Place ID,
business name,
address,
rating,
number of ratings,
opening information,
and Maps URLs.

A simplified retrieval call looks like this:

import requests


def search_places(query: str, api_key: str) -> list[dict]:
    endpoint = "https://places.googleapis.com/v1/places:searchText"

    payload = {
        "textQuery": query,
        "pageSize": 10,
    }

    headers = {
        "Content-Type": "application/json",
        "X-Goog-Api-Key": api_key,
        "X-Goog-FieldMask": ",".join(
            [
                "places.id",
                "places.displayName",
                "places.formattedAddress",
                "places.rating",
                "places.userRatingCount",
                "places.currentOpeningHours",
                "places.googleMapsUri",
            ]
        ),
    }

    response = requests.post(
        endpoint,
        json=payload,
        headers=headers,
        timeout=20,
    )

    response.raise_for_status()

    return response.json().get("places", [])

This is intentionally simplified for the article.

In the repository, there is additional application logic around the retrieval and candidate handling.

Place ID turned out to be more important than I expected

At first I was mostly interested in ratings, addresses, and opening hours.

Once fan-out was involved, identity became equally important.

Suppose two searches produce:

Merit Coffee

and

Merit Coffee — Southtown

String matching alone is not a great basis for deciding whether I have one business or two.

Using a canonical identifier such as Place ID gives me a stronger entity boundary.

That became important almost immediately when I started combining results from multiple retrieval tasks.

Step 4: Places could not answer everything

This is where the system became more than a Places demo.

The API can tell me a lot about a place.

But look again at part of the original requirement:

quiet enough to work

That is not the same kind of data as:

rating = 4.6

or:

open until 10 PM

The same problem appears in other searches:

good for a large student group

strong vegan options

suitable for someone with a specific preference

These are softer constraints.

I needed another evidence path.

Step 5: I added Google Search grounding for the missing evidence

Rather than asking Gemini:

Is this place good for working?

I wanted the system to investigate specific claims.

For example:

Candidate: Example Coffee

Verify:
- late-hour availability
- evidence relevant to working/studying
- evidence relevant to group seating

For each constraint:
- supported
- unsupported
- evidence
- citation

Conceptually, that stage looks like:

with trace_span(
    "search_grounding.verify_evidence",
    attributes={"workflow.stage": "v1"},
):
    evidence = verify_candidates(
        candidates=candidates,
        intent=intent,
    )

This gave me two different types of evidence:

Google Places + Search-grounded evidence

I wanted to preserve that distinction instead of blending everything together.

If operating hours come from Places, I should know that.

If “work-friendly” comes from grounded Search evidence, I should know that too.

And if I cannot verify something, I want the system to keep that state visible.

Step 6: Fan-out created a duplicate problem

Once I started running multiple Places tasks, something predictable happened:

the same business started appearing more than once.

For example:

Places task A ───────┐
                     ├── Candidate X
Places task B ───────┘

If I ranked those raw results directly, Candidate X could appear stronger simply because more than one retrieval task discovered it.

That is not relevance.

That is duplication.

So I introduced normalization and deduplication before scoring.

A simplified form is:

def deduplicate_places(candidates: list[dict]) -> list[dict]:
    unique: dict[str, dict] = {}

    for candidate in candidates:
        place_id = candidate["place_id"]

        if place_id not in unique:
            unique[place_id] = candidate
            continue

        unique[place_id] = merge_candidate(
            unique[place_id],
            candidate,
        )

    return list(unique.values())

The sequence matters:

retrieve → normalize → deduplicate → score

I specifically did NOT want:

retrieve → score duplicates → fix identity afterward

V2 made this problem visible

For Journey Analysis, I used another demo query:

Find an Indian restaurant near Trinity University for eight students, open after 9 PM, with vegetarian options. Recommend the top three.

One of my representative executions produced:

Places tasks                     2
Raw Places results              20
Unique Places candidates        13

Search tasks                     2

Candidates with Search evidence  3
Places-only candidates          10
Unmatched evidence               8

That was a useful moment in the project.

Instead of asking only:

What are the top three restaurants?

I could now ask:

What actually happened during retrieval?

That became the purpose of V2 — Journey Analysis.

Step 7: Retrieval and ranking are not the same thing

This sounds obvious, but it became an important rule in my implementation.

Retrieval asks:

Which candidates might be relevant?

Ranking asks:

Which of those candidates best satisfies the original constraints?

A candidate should not rank higher simply because two fan-out queries happened to retrieve it.

And the highest-rated business should not automatically win if it fails an important constraint such as opening hours.

So after deduplication, I aggregate the evidence and score candidates explicitly.

Step 8: I keep ranking outside the LLM

This was another decision I made deliberately.

I did not want this architecture:

Retrieve 10 places
       ↓
Send all 10 to Gemini
       ↓
"Pick the best three"

That would give me an answer.

But it would make one of the most important decisions in the system opaque.

Instead, I keep the ranking logic inspectable in application code.

Depending on the workflow, signals can include things like:

location relevance,
category match,
operating-hour compatibility,
rating,
review volume,
constraint coverage,
grounded evidence,
unsupported constraints.

Conceptually:

with trace_span(
    "evidence.aggregate_and_score",
    attributes={"workflow.stage": "v1"},
):
    ranked_candidates = aggregate_and_score(
        candidates=candidates,
        evidence=evidence,
        intent=intent,
    )

The important property here is not one universal scoring formula.

It is that I can inspect why a candidate received its score.

def calculate_haversine_distance_miles(
    lat1: float,
    lon1: float,
    lat2: float,
    lon2: float,
) -> float:
    """Calculate the great circle distance in miles between two latitude/longitude points."""
    r_earth = 3958.8  # Earth radius in miles
    d_lat = math.radians(lat2 - lat1)
    d_lon = math.radians(lon2 - lon1)
    a = (
        math.sin(d_lat / 2.0) ** 2
        + math.cos(math.radians(lat1))
        * math.cos(math.radians(lat2))
        * math.sin(d_lon / 2.0) ** 2
    )
    c = 2.0 * math.atan2(math.sqrt(a), math.sqrt(1.0 - a))
    return round(r_earth * c, 2)


def calculate_proximity_score(distance_miles: float) -> float:
    """Calculate smooth bounded proximity bonus from distance in miles (max 12.0 pts).

    Formula: bonus = MAX_PROXIMITY_BONUS / (1.0 + distance_miles)
    - 0.0 mi  -> 12.00 pts
    - 0.5 mi  -> 8.00 pts
    - 1.0 mi  -> 6.00 pts
    - 2.0 mi  -> 4.00 pts
    - 5.0 mi  -> 2.00 pts
    - 190 mi  -> 0.06 pts (Houston vs San Antonio)
    """
    if distance_miles < 0.0:
        return 0.0
    bonus = MAX_PROXIMITY_BONUS / (1.0 + distance_miles)
    return round(bonus, 2)

View the complete scoring implementation on GitHub

Candidates are ranked after normalization and evidence aggregation rather than being silently reordered by the model.

Step 9: Gemini comes back at the end

Gemini still has an important job after the deterministic ranking stage.

I use it to turn the structured result into an answer that is useful to a person.

Conceptually:

with trace_span(
    "gemini.synthesize_recommendations",
    attributes={"workflow.stage": "v1"},
):
    response = synthesize_recommendations(
        intent=intent,
        ranked_candidates=ranked_candidates,
    )

But notice what has changed.

At the start of the workflow, Gemini had:

one natural-language request

At the end, it can work with:

structured intent
       +
normalized candidates
       +
Places evidence
       +
Search evidence
       +
constraint states
       +
candidate scores
       +
final ranking

So instead of asking Gemini to discover, verify, rank, and explain everything at once, I give it a much narrower final job:

Explain a decision whose evidence has already been assembled.

That is the architecture I wanted.

The complete journey

At a high level, the workflow became:

And internally, I started giving the major stages explicit names such as:

gemini.extract_intent
gemini.plan_fan_out
places.text_search
search_grounding.verify_evidence
candidate.normalize_and_dedup
evidence.aggregate_and_score
gemini.synthesize_recommendations

Those names become important again later in this series when I start tracing the workflows.

Step 10: I test it like normal software

Another goal for this project was to avoid treating AI code as somehow exempt from normal software-engineering practices.

After making changes, I run:

./.venv/bin/ruff check .
./.venv/bin/mypy src
./.venv/bin/pytest -q

Step 11: Then I moved the same application to Cloud Run

I also wanted the project to go beyond my laptop.

The application runs on Google Cloud Run.

I can inspect the deployed service from the CLI:

gcloud run services describe ai-search-journey-lab \
  --region us-central1 \
  --format="yaml(
      metadata.name,
      status.url,
      status.latestReadyRevisionName,
      status.conditions
  )"

And verify the Streamlit health endpoint:

curl -fsS \
"https://ai-search-journey-lab-642110324230.us-central1.run.app/_stcore/health"

For me, this matters because it closes the development loop:

local code
   ↓
local execution
   ↓
tests
   ↓
container
   ↓
Cloud Run
   ↓
live application

The same Search-to-Decision workflow running from the Cloud Run deployment.

What I am not trying to reproduce

There is an important distinction I want to make.

This project is not:

a reproduction of Google's internal ranking systems,
a reverse-engineered implementation of AI Mode,
an implementation of Google AI Overviews,
or a claim that my query fan-out represents Google's proprietary query fan-out.

I am using query fan-out as an application architecture pattern.

The project asks a developer-focused question:

If one natural-language request turns into multiple retrieval tasks, what engineering problems do I have to solve before I can return a grounded decision?

That is what I am trying to explore.

What happens when I run many search journeys?

V1 and V2 focus on one user's decision.

But imagine running many related prompts.

Now I can start asking:
How often does a brand appear?
Which competitors are being mentioned?
Which sources are being cited?
Does the brand appear in the initial query, the fan-out queries, or only the final answer?
How does visibility change across different prompts?

That is where this project moved from Search Journey Optimization into AI Search Visibility.

And that is the next article.

Next: Building Grounded Local Search with Gemini and Google Maps on Google Cloud

Before I move fully into visibility analytics, Part 2 will go deeper into the actual local-search implementation:

Gemini,
Google Places API (New),
Google Maps,
Google Search grounding,
grounded recommendation generation,
and Cloud Run deployment.

Then Part 3 will move into the AI Search Visibility & Brand Analyzer with Gemini and BigQuery.

Try the project

Repository: GitHub Repository

Local startup:

python scripts/run_app_locally.py