	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Generative AI > Google Gemini API > Vertex AI Pipelines with ADK

Vertex AI Pipelines with ADK

Author: Venkata Sudhakar

Vertex AI Pipelines is a managed workflow orchestration service that runs sequences of steps as a directed acyclic graph (DAG). Combining it with ADK lets you build automated pipelines where agent reasoning is one step in a larger data processing or ML workflow. A nightly pipeline might: extract new customer reviews from BigQuery, run them through a Gemini structured output agent to classify sentiment and extract issues, write the structured results back to BigQuery, and trigger an alert if negative sentiment exceeds a threshold. This entire workflow runs on a schedule with no human intervention and full observability in the GCP Console.

Vertex AI Pipelines uses the Kubeflow Pipelines SDK (KFP) where you define components as Python functions decorated with @dcc.component and wire them into a pipeline with @pipeline. Each component runs in its own isolated container. ADK agent calls fit naturally as pipeline components ï¿½ a component receives data, runs an ADK agent query, and outputs the result to the next step. The pipeline is compiled to a YAML spec and submitted to Vertex AI which handles scheduling, retry, and logging automatically.

The below example builds a nightly product review processing pipeline that reads new reviews, runs them through a Gemini classification agent, writes structured results to BigQuery, and sends a summary report ï¿½ all orchestrated as a Vertex AI Pipeline that runs on a schedule.

import json
from kfp import dsl, compiler
from kfp.dsl import component, Output, Input, Dataset, Artifact
from google.cloud import aiplatform

PROJECT = "your-gcp-project"
REGION  = "us-central1"
BUCKET  = "gs://your-pipeline-bucket"

# Component 1: Extract new reviews from BigQuery
@component(base_image="python:3.11", packages_to_install=["google-cloud-bigquery"])
def extract_reviews(project: str, days_back: int, reviews_output: Output[Dataset]):
    from google.cloud import bigquery
    import json
    bq = bigquery.Client(project=project)
    sql = (
        "SELECT review_id, product_name, review_text, review_date "
        "FROM " + project + ".shopmax_analytics.product_reviews "
        "WHERE review_date >= DATE_SUB(CURRENT_DATE(), INTERVAL "
        + str(days_back) + " DAY)"
    )
    rows = list(bq.query(sql).result())
    reviews = [{"id": r.review_id, "product": r.product_name,
                "text": r.review_text} for r in rows]
    with open(reviews_output.path, "w") as f:
        json.dump(reviews, f)
    print("Extracted", len(reviews), "reviews")

# Component 2: Classify reviews using Gemini structured output
@component(
    base_image="python:3.11",
    packages_to_install=["google-genai"]
)
def classify_reviews(
    api_key: str,
    reviews_input: Input[Dataset],
    classified_output: Output[Dataset]
):
    import json
    from google import genai
    from google.genai import types

client = genai.Client(api_key=api_key)
    schema = types.Schema(
        type=types.Type.OBJECT,
        required=["rating", "sentiment", "issues"],
        properties={
            "rating":    types.Schema(type=types.Type.INTEGER),
            "sentiment": types.Schema(type=types.Type.STRING,
                                      enum=["positive","neutral","negative","mixed"]),
            "issues":    types.Schema(type=types.Type.ARRAY,
                                      items=types.Schema(type=types.Type.STRING))
        }
    )
    with open(reviews_input.path) as f:
        reviews = json.load(f)

classified = []
    for review in reviews:
        resp = client.models.generate_content(
            model="gemini-2.0-flash",
            config=types.GenerateContentConfig(
                response_mime_type="application/json",
                response_schema=schema,
                max_output_tokens=200
            ),
            contents=[review["text"]]
        )
        result = json.loads(resp.text)
        classified.append({**review, **result})
    with open(classified_output.path, "w") as f:
        json.dump(classified, f)
    print("Classified", len(classified), "reviews")

Components 3 and 4 for writing results and sending the daily report,

# Component 3: Write classified results to BigQuery
@component(
    base_image="python:3.11",
    packages_to_install=["google-cloud-bigquery"]
)
def write_to_bigquery(
    project: str,
    classified_input: Input[Dataset],
    negative_count_output: Output[Artifact]
):
    import json
    from google.cloud import bigquery
    bq   = bigquery.Client(project=project)
    with open(classified_input.path) as f:
        rows = json.load(f)
    table = project + ".shopmax_analytics.classified_reviews"
    errors = bq.insert_rows_json(table, rows)
    if errors:
        raise RuntimeError("BigQuery insert errors: " + str(errors))
    negative = sum(1 for r in rows if r["sentiment"] == "negative")
    with open(negative_count_output.path, "w") as f:
        f.write(str(negative))
    print("Inserted", len(rows), "rows |", negative, "negative reviews")

# Component 4: Send summary report if negative rate is high
@component(base_image="python:3.11")
def send_alert(negative_count_input: Input[Artifact], threshold: int):
    with open(negative_count_input.path) as f:
        negative = int(f.read().strip())
    if negative > threshold:
        print("ALERT: " + str(negative) + " negative reviews today - above threshold of "
              + str(threshold))
        # In production: send email via SendGrid or Pub/Sub notification
    else:
        print("All good: " + str(negative) + " negative reviews (below threshold " + str(threshold) + ")")

# Wire components into a pipeline
@dsl.pipeline(
    name="nightly-review-classification",
    description="Classify product reviews nightly using Gemini and load to BigQuery"
)
def review_pipeline(
    project: str = PROJECT,
    gemini_api_key: str = "your-key",
    days_back: int = 1,
    alert_threshold: int = 10
):
    extract_task = extract_reviews(project=project, days_back=days_back)
    classify_task = classify_reviews(
        api_key=gemini_api_key,
        reviews_input=extract_task.outputs["reviews_output"]
    )
    write_task = write_to_bigquery(
        project=project,
        classified_input=classify_task.outputs["classified_output"]
    )
    send_alert(
        negative_count_input=write_task.outputs["negative_count_output"],
        threshold=alert_threshold
    )

# Compile and submit to Vertex AI
compiler.Compiler().compile(review_pipeline, "review_pipeline.yaml")
aiplatform.init(project=PROJECT, location=REGION)
job = aiplatform.PipelineJob(
    display_name="nightly-review-classification",
    template_path="review_pipeline.yaml",
    pipeline_root=BUCKET,
    parameter_values={"days_back": 1, "alert_threshold": 10}
)
job.submit()
print("Pipeline submitted:", job.resource_name)

It gives the following output when the pipeline runs on Vertex AI,

Pipeline submitted: projects/my-project/locations/us-central1/pipelineJobs/nightly-review-123

[Vertex AI Pipelines executes each component in order]
Component 1 - extract_reviews:    Extracted 47 reviews
Component 2 - classify_reviews:   Classified 47 reviews
Component 3 - write_to_bigquery:  Inserted 47 rows | 8 negative reviews
Component 4 - send_alert:         All good: 8 negative reviews (below threshold 10)

Pipeline Status: SUCCEEDED
Duration: 4m 32s

# Schedule this pipeline to run nightly via Vertex AI Pipelines scheduler:
# GCP Console > Vertex AI > Pipelines > Schedules > Create Schedule
# Or programmatically using aiplatform.PipelineJobSchedule

The pipeline now runs automatically every night at 2am IST,

Schedule created: projects/my-project/locations/us-central1/schedules/abc123

# Pipeline runs every night: 2am IST = 8:30pm UTC
# Each run processes previous day reviews automatically
# All run history visible in GCP Console > Vertex AI > Pipelines
# Failed runs auto-retry and trigger alerts via Cloud Monitoring

Vertex AI Pipelines with ADK is the right architecture when you need: scheduled batch processing of large datasets, reproducible and auditable AI workflows, automatic retry and failure handling at the step level, and integration with the broader GCP data ecosystem (BigQuery, Cloud Storage, Pub/Sub). For real-time agent workflows use Agent Engine (Tutorial 320). For scheduled batch AI processing use Vertex AI Pipelines. The two complement each other: Agent Engine for interactive user-facing agents, Vertex AI Pipelines for back-office automation that runs while your team sleeps.

Send your comments, suggestions or queries regarding this site to [email protected].