DEV Community

Cover image for Event-Driven Multi-Agent Pipelines on Google Cloud with ADK, Gemma 3 and Eventarc
Google Developer Experts profile image Robina Mirbahar
Robina Mirbahar for Google Developer Experts

Posted on Edited on

Event-Driven Multi-Agent Pipelines on Google Cloud with ADK, Gemma 3 and Eventarc

#googleai #gemma #googlecloud #ai

This article provides a step by step architectural and deployment guide for StreamForge, a real-time event-driven multi-agent pipeline built with the Google Agent Development Kit (ADK), open-weights Gemma 3 (gemma3:12b), Google Cloud Eventarc, Cloud Pub/Sub, and Google Cloud Run.

Repository: github.com/RobinaMirbahar/streamforge-event-adk

Enterprise data platforms operate on real-time state changes: a product engineering team uploads a release specification (vertex-vector-sync-v4.2.json) to a Google Cloud Storage bucket, a streaming job commits a partition to BigQuery, or a partner service publishes a schema event to Cloud Pub/Sub. Connecting Google Cloud Eventarc directly to a multi-agent Google ADK workflow runtime on Cloud Run triggers autonomous workflows within 200 ms of an object landing in storage while scaling to 0 instances when the bucket is idle.

Why Combine Eventarc with ADK ParallelAgent and LoopAgent?

Processing a multi-regional product release requires three distinct tasks: extracting technical benchmarks from the raw JSON payload, translating launch messaging across target locales (en_US, de_DE, ja_JP), and simulating 12-month Annual Recurring Revenue (ARR) expansion under regional B2B currency rules. Running those three steps sequentially increases end-to-end latency, while running them in parallel without a verification step risks publishing unrounded foreign-exchange prices.

Architecture Pattern Trigger Mechanism Execution Topology Cross-Agent Consistency Check Duplicate Event Protection
🥇 Eventarc + ADK ParallelAgent & LoopAgent Real-time CloudEvents v1.0 (< 200 ms lag) 3 concurrent branches (3x fan-out) + LoopAgent critic Iterative audit repairs EUR/JPY pricing until quality_score == 1.0 Firestore ce-id ledger short-circuits replays in < 5 ms
🥈 Scheduled ADK SequentialAgent Fixed cron schedule Linear 3-stage chain Single-pass sequential handoff Date-keyed Firestore lock
🥉 Single Monolithic Prompt Manual human chat prompt Single LLM call Unchecked single-pass output Manual retry handling

End-to-End Event-Driven Architecture on Google Cloud

Figure 1: Real-Time Event-Driven Agent Architecture on Google Cloud

  1. Tier 1: Event Ingestion and Routing: Uploading releases/2026-10/vertex-vector-sync-v4.2.json to gs://streamforge-launch-inbox emits a native google.cloud.storage.object.v1.finalized event. Google Cloud Eventarc wraps the notification in a CNCF CloudEvents v1.0 envelope (ce-id, ce-type, ce-source, ce-time) and delivers it over an OIDC-authenticated Cloud Pub/Sub push subscription.
  2. Tier 2: Serverless Multi-Agent Runtime (Cloud Run): A private FastAPI receiver checks the incoming ce-id header (ce-98104271-gcs-finalized) against Cloud Firestore. New events execute the Google ADK workflow (ParallelAgent → LoopAgent → LlmAgent) powered by Gemma 3 12B via LiteLlm.
  3. Tier 3: State and Downstream Sink: The final publisher agent validates the output against the Pydantic EventDrivenLaunchManifest schema and commits both the manifest (MANIFEST-SF-VECTOR-SYNC-4.2) and the ce-id deduplication record to Cloud Firestore.

Orchestrating ParallelAgent Fan-Out and LoopAgent Self-Correction

Figure 2: Orchestrating ADK ParallelAgent Fan-Out and LoopAgent Self-Correction

In Google ADK, all branches inside a ParallelAgent share the same InvocationContext.session.state. Assigning a distinct output_key to each branch isolates their writes cleanly:

  • TechnicalSpecExtractorAgent writes engineering benchmarks to output_key="extracted_specs".
  • GlobalLocalizationAgent writes localized headlines and draft currency tiers to output_key="localized_copy".
  • RevenuePricingSimulatorAgent writes the 12-month ARR forecast and regional FX compliance rules to output_key="pricing_simulation".

Once all 3 parallel branches complete, QualityCritiqueRefinementLoop (LoopAgent with max_iterations=3) invokes LaunchReadinessCriticAgent to audit the combined state. When GlobalLocalizationAgent converts the $499/mo Growth Tier at the raw 0.92 EUR exchange rate (€459.08/mo), Iteration 1 of the LoopAgent flags the violation against the de_DE rule requiring VAT-exclusive B2B prices rounded to the nearest €5 (quality_score = 0.72) and updates the pricing tier to €460/mo. On Iteration 2, the audit reaches quality_score = 1.0 and invokes exit_quality_loop(), setting tool_context.actions.escalate = True to exit the loop immediately.

At This Point You Should Have

  • Python 3.10+ (3.11 recommended) with google-adk>=1.0.0, litellm, fastapi, and pydantic installed.
  • A Google Cloud project with Cloud Run, Eventarc, Cloud Pub/Sub, Cloud Storage, and Cloud Firestore APIs enabled.
  • Access to Gemma 3 12B (ollama_chat/gemma3:12b locally via Ollama or google/gemma-3-12b-it on a Cloud Run NVIDIA L4 24 GB GPU service).

Step 1: Define the CloudEvent and Launch Manifest Schemas

The pipeline enforces typed contracts at both boundaries: CloudEventEnvelope validates the incoming Eventarc HTTP headers and payload, while EventDrivenLaunchManifest constrains the final ReleasePublisherAgent output in streamforge_agent/schemas.py.

==> [1/6] Enabling Cloud Run, Eventarc, Pub/Sub, Storage, and Firestore APIs...
==> [2/6] Creating Cloud Storage Inbox Bucket & Dead-Letter Pub/Sub Topic...
==> [3/6] Creating Least-Privilege Service Accounts...
==> [4/6] Deploying StreamForge ADK Service to Cloud Run (private ingress)...
Service [streamforge-event-adk] revision [streamforge-event-adk-00001-x8k] has been deployed and is serving 100 percent of traffic.
==> [5/6] Granting Cloud Run Invoker + Eventarc EventReceiver to Eventarc SA...
==> [6/6] Creating Eventarc Trigger for google.cloud.storage.object.v1.finalized...
Created trigger [streamforge-gcs-release-trigger] in [us-central1].
Enter fullscreen mode Exit fullscreen mode

Step 2: Implement the Parallel Enrichment and Loop Escalation Tools

In streamforge_agent/tools.py, each specialist tool exposes typed parameters and docstrings for Gemma 3 function calling, while exit_quality_loop sets tool_context.actions.escalate = True on the injected ADK ToolContext.

==> [1/6] Enabling Cloud Run, Eventarc, Pub/Sub, Storage, and Firestore APIs...
==> [2/6] Creating Cloud Storage Inbox Bucket & Dead-Letter Pub/Sub Topic...
==> [3/6] Creating Least-Privilege Service Accounts...
==> [4/6] Deploying StreamForge ADK Service to Cloud Run (private ingress)...
Service [streamforge-event-adk] revision [streamforge-event-adk-00001-x8k] has been deployed and is serving 100 percent of traffic.
==> [5/6] Granting Cloud Run Invoker + Eventarc EventReceiver to Eventarc SA...
==> [6/6] Creating Eventarc Trigger for google.cloud.storage.object.v1.finalized...
Created trigger [streamforge-gcs-release-trigger] in [us-central1].
Enter fullscreen mode Exit fullscreen mode

Step 3: Wire ParallelAgent, LoopAgent, and SequentialAgent

In streamforge_agent/agent.py, build_streamforge_event_pipeline() composes the 3 parallel enrichment agents, the iterative quality critic loop, and the schema-constrained publisher agent into a single parent SequentialAgent.

==> [1/6] Enabling Cloud Run, Eventarc, Pub/Sub, Storage, and Firestore APIs...
==> [2/6] Creating Cloud Storage Inbox Bucket & Dead-Letter Pub/Sub Topic...
==> [3/6] Creating Least-Privilege Service Accounts...
==> [4/6] Deploying StreamForge ADK Service to Cloud Run (private ingress)...
Service [streamforge-event-adk] revision [streamforge-event-adk-00001-x8k] has been deployed and is serving 100 percent of traffic.
==> [5/6] Granting Cloud Run Invoker + Eventarc EventReceiver to Eventarc SA...
==> [6/6] Creating Eventarc Trigger for google.cloud.storage.object.v1.finalized...
Created trigger [streamforge-gcs-release-trigger] in [us-central1].
Enter fullscreen mode Exit fullscreen mode

Step 4: Verify the Pipeline and CloudEvent Deduplication Locally

Run scripts/simulate_cloudevent_pipeline.py to execute the 3-branch ParallelAgent fan-out, inspect the 2-iteration LoopAgent self-correction trace, and verify the ce-id idempotency guard against a duplicate Pub/Sub delivery.

==> [1/6] Enabling Cloud Run, Eventarc, Pub/Sub, Storage, and Firestore APIs...
==> [2/6] Creating Cloud Storage Inbox Bucket & Dead-Letter Pub/Sub Topic...
==> [3/6] Creating Least-Privilege Service Accounts...
==> [4/6] Deploying StreamForge ADK Service to Cloud Run (private ingress)...
Service [streamforge-event-adk] revision [streamforge-event-adk-00001-x8k] has been deployed and is serving 100 percent of traffic.
==> [5/6] Granting Cloud Run Invoker + Eventarc EventReceiver to Eventarc SA...
==> [6/6] Creating Eventarc Trigger for google.cloud.storage.object.v1.finalized...
Created trigger [streamforge-gcs-release-trigger] in [us-central1].
Enter fullscreen mode Exit fullscreen mode 
==> [1/6] Enabling Cloud Run, Eventarc, Pub/Sub, Storage, and Firestore APIs...
==> [2/6] Creating Cloud Storage Inbox Bucket & Dead-Letter Pub/Sub Topic...
==> [3/6] Creating Least-Privilege Service Accounts...
==> [4/6] Deploying StreamForge ADK Service to Cloud Run (private ingress)...
Service [streamforge-event-adk] revision [streamforge-event-adk-00001-x8k] has been deployed and is serving 100 percent of traffic.
==> [5/6] Granting Cloud Run Invoker + Eventarc EventReceiver to Eventarc SA...
==> [6/6] Creating Eventarc Trigger for google.cloud.storage.object.v1.finalized...
Created trigger [streamforge-gcs-release-trigger] in [us-central1].
Enter fullscreen mode Exit fullscreen mode

Step 5: Deploy the Private Cloud Run Receiver and Eventarc Trigger

Run deploy/deploy_eventarc_adk.sh to deploy the container to Google Cloud Run (2 vCPU, 2 GiB memory, --no-allow-unauthenticated, --min-instances=0, --max-instances=10) and bind the Eventarc trigger to google.cloud.storage.object.v1.finalized.

==> [1/6] Enabling Cloud Run, Eventarc, Pub/Sub, Storage, and Firestore APIs...
==> [2/6] Creating Cloud Storage Inbox Bucket & Dead-Letter Pub/Sub Topic...
==> [3/6] Creating Least-Privilege Service Accounts...
==> [4/6] Deploying StreamForge ADK Service to Cloud Run (private ingress)...
Service [streamforge-event-adk] revision [streamforge-event-adk-00001-x8k] has been deployed and is serving 100 percent of traffic.
==> [5/6] Granting Cloud Run Invoker + Eventarc EventReceiver to Eventarc SA...
==> [6/6] Creating Eventarc Trigger for google.cloud.storage.object.v1.finalized...
Created trigger [streamforge-gcs-release-trigger] in [us-central1].
Enter fullscreen mode Exit fullscreen mode 
==> [1/6] Enabling Cloud Run, Eventarc, Pub/Sub, Storage, and Firestore APIs...
==> [2/6] Creating Cloud Storage Inbox Bucket & Dead-Letter Pub/Sub Topic...
==> [3/6] Creating Least-Privilege Service Accounts...
==> [4/6] Deploying StreamForge ADK Service to Cloud Run (private ingress)...
Service [streamforge-event-adk] revision [streamforge-event-adk-00001-x8k] has been deployed and is serving 100 percent of traffic.
==> [5/6] Granting Cloud Run Invoker + Eventarc EventReceiver to Eventarc SA...
==> [6/6] Creating Eventarc Trigger for google.cloud.storage.object.v1.finalized...
Created trigger [streamforge-gcs-release-trigger] in [us-central1].
Enter fullscreen mode Exit fullscreen mode

Official Google Cloud and ADK Documentation

Summary

The goal of this article was to build and verify a real-time, event-driven multi-agent pipeline on Google Cloud using Eventarc, Cloud Pub/Sub, Cloud Run, open-weights Gemma 3, and the Google Agent Development Kit. The key to the solution was pairing ParallelAgent for concurrent 3-branch fan-out with LoopAgent for iterative cross-branch self-correction and a Cloud Firestore ce-id deduplication ledger. The verification results were:

  • 🟢 Concurrent Fan-Out (ParallelAgent): Extracted engineering specs (73.8% latency drop from 145 ms to 38 ms at 250,000 rows/sec), localized copy across 3 locales (en_US, de_DE, ja_JP), and forecasted $2,856,000 in 12-month Net New ARR lift across 340 upgraded accounts (18.5% of 1,840 active Growth Tier accounts at $8,400 annual expansion).
  • 🟢 Self-Correcting Quality Loop (LoopAgent): Detected the €459.08/mo and ¥75,848/mo currency formatting violations on Iteration 1 (quality_score = 0.72), repaired them to €460/mo and ¥75,800/mo, and exited cleanly on Iteration 2 (quality_score = 1.0) via exit_quality_loop().
  • 🟢 At-Least-Once Idempotency: Short-circuited duplicate ce-98104271-gcs-finalized redelivery in < 5 ms with 0 redundant parallel branches or LLM calls.

The strategy for orchestrating event-driven multi-agent pipelines with Google ADK and Gemma 3 on Cloud Run was validated with an incremental step by step approach.

Top comments (2)

Collapse
 
stefan_07d23f966225156b50 profile image
Stefan •

For teaching, I'd want to review the four-panel script before generating the images. If the RAG explanation is wrong, I'd rather fix it while it's still text than after I've got four lovely panels explaining the wrong thing. Could that be an optional pause in the workflow?

Collapse
 
robinamirbahar profile image
Robina Mirbahar Google Developer Experts •

Thanks so much for reading closely, @stefan_07d23f966225156b50 ! Have you tried building something similar for your own teaching materials yet? Burning image generation credits on four panels only to realize Panel 2 explained vector retrieval backwards is painful, both for teaching accuracy and your cloud bill.

Yes, you can add an optional human review pause right between the script stage and the image rendering stage. Because Cloud Run bills by active request time, you don't want the container sitting idle waiting for a human click. The cleanest way to do this in Google ADK + Eventarc is a two-stage state handoff using Firestore (or ADK session state):

  1. Stage 1 (Script Draft Only): Run the storyboard agent to generate the 4-panel text and save it to Firestore (or session.state["panel_script"]) with status: "PENDING_REVIEW". If you pass a flag like auto_render: false, the pipeline stops right here and returns the editable JSON text for all 4 panels.

  2. Human Edit & Approval: You review the RAG explanation in text, tweak the dialogue or visual prompts on any panel you don't like, and flip status to "APPROVED".

  3. Stage 2 (Render on Approval): An Eventarc trigger listening for google.cloud.firestore.document.v1.updated (where status == "APPROVED", or a simple /render-approved endpoint call) wakes up the second ADK agent, pulls your edited panel_script, and fans out the 4 image generation calls.

If you are running it inside a single ADK runner with session_service, it looks like this:

# Step 1: Generate the 4-panel script and pause
await script_runner.run_async(user_id=user_id, session_id=session_id, new_message=topic_msg)
session = await session_service.get_session(app_name="comic_pipeline", user_id=user_id, session_id=session_id)

# Return session.state["panel_script"] to the teacher for review/edits
# ... teacher edits panel 2 text and clicks "Approve" ...

# Step 2: Write the teacher's edited script back into state and run the image agent
session.state["panel_script"] = edited_four_panel_script
await image_runner.run_async(user_id=user_id, session_id=session_id, new_message=approve_msg)
Enter fullscreen mode Exit fullscreen mode

Bonus tip if you build this for teaching:
When you pause at the text stage, have the script agent also output a shared character_bible field in the JSON (for example, "Robot tutor wearing round brass glasses and a navy hoodie, flat vector style") and prepend that exact string to all 4 panel prompts during Stage 2. Also, add a dirty_panels: [2] parameter to your render endpoint so if Panel 2's visual metaphor still looks slightly off after rendering, you can re-roll just Panel 2 without regenerating Panels 1, 3, and 4!