Real-time streaming pipelines are the operational backbone of modern enterprises, continuously processing everything from customer support interactions to transaction logs. Traditionally, streaming DAGs are static; once deployed, their processing logic and execution paths are fixed. However, by integrating generative AI agents, we can move beyond static logic to adaptive execution. This allows streaming workflows to dynamically construct plans, query databases, and trigger custom remediation paths at runtime depending on the content of the data.
For example, when a customer sends an angry message about a damaged order, a pipeline shouldn't just log the error or flag a dashboard. It should look up the order in the database that holds customer order and inventory records, decide on a remediation action (like shipping a replacement or issuing a refund), email the customer, and log the final resolution. However, streaming systems face a fundamental engineering hurdle when executing gen AI workflows: scale, latency, and cost. Sending every raw event directly to a heavyweight model or multi-step agent equipped with external database and email tools is prohibitively expensive, introduces high latency, and quickly exhausts API rate limits.
This pattern addresses the scale and complexity challenge by combining Google Dataflow, Google Cloud's fully managed, serverless execution service for Apache Beam, and the Agent Development Kit (ADK) to build a hybrid streaming pipeline. By using a lightweight, CPU-bound machine learning model upstream to filter and qualify events, we keep the pipeline highly cost-effective, routing only the complex cases to the downstream agent. There, the agent dynamically decides what actions to take, introducing dynamic branching to the stream without hardcoding thousands of conditional steps into the pipeline's static DAG.
While we use a customer support triage scenario below, this pre-filter + agentic action pattern is a universal paradigm. It applies to any stream where a high volume (>9X%) of events are routine, and only a small number require complex, contextual reasoning. IT Operations & DevOps: Filtering millions of routine system logs on CPU, and triggering an agent to run diagnostics and open bug tickets only when a critical anomaly is flagged.
Financial Fraud Triaging: Passing millions of transactions through lightweight, local rules, and calling an agent to execute multi-database lookup tools only for highly suspicious patterns.
Industrial IoT: Monitoring normal telemetry on the edge, and routing erratic spikes to an agent to coordinate equipment shutdowns and email field engineers.
In a high-throughput stream, the vast majority of messages do not require complex reasoning or remediation. They might be positive feedback, neutral inquiries, or simple queries.
Routing every single event to a heavyweight LLM workflow creates three primary bottlenecks:
API cost: Frontier models charge per token. Under high throughput, cost scales linearly with stream volume.
Latency: Multi-step workflows (which involve database lookups and external API calls) take seconds, creating a bottleneck in streaming DAGs.
Quotas: External APIs have strict rate limits that streaming workers can easily exhaust.
To prevent this, we build a pre-filtered pipeline in Apache Beam/Dataflow:
**Ingestion:** Read raw customer messages from [Google Pub/Sub](https://cloud.google.com/pubsub).
**Lightweight sentiment classifier (CPU):** Run all messages through a lightweight, CPU-based Hugging Face model (`distilbert-base-uncased-finetuned-sst-2-english`
) using Apache Beam’s RunInference
transform. This executes locally on the Dataflow worker CPUs, avoiding external API costs.
Pre-qualification Gate: A simple DoFn
filters the stream. Messages with POSITIVE
or NEUTRAL
sentiment are acknowledged and dropped.
Automated Remediation (ADK): If and only if a message is classified as NEGATIVE
, we trigger the gen AI agent backed by gemini-3.5-flash
using the ADKAgentModelHandler
. The agent uses tools to look up the user in BigQuery, fetch orders, choose a remediation plan, and send a notification email via the Gmail API.
In traditional streaming architectures, the pipeline's Directed Acyclic Graph (DAG) is rigid. Once deployed to Dataflow, the sequence of transforms is set. If you need to handle new types of alerts or change how specific events are routed, you have to modify, test, and redeploy the entire pipeline.
By placing a gen AI agent downstream of our sentiment pre-filter, we introduce a dynamic, adaptive node inside the static DAG.
For the 95% of records that are positive or neutral, the pipeline runs along a fast, static path. But when the filter gates a negative record, the agent evaluates the payload and dynamically selects the correct sequence of API tools (e.g., database query, inventory check, or email notification) at runtime. This allows the pipeline to execute complex decision trees dynamically, eliminating the need to build and maintain thousands of hardcoded conditional branches in the static Apache Beam code.
Here is an example implementation in Apache Beam using the Google Agent Development Kit (ADK) and the RunInference
framework.
We define the upstream CPU model using HuggingFacePipelineModelHandler
. This model classifies sentiment into POSITIVE
, NEUTRAL
, or NEGATIVE
on the worker instance.
The ADK agent acts as our remediation assistant. We equip it with three tools:
lookup_user
: Queries BigQuery for the customer's email.
lookup_orders
: Queries BigQuery for the customer's orders and current product inventory.
send_email
: Sends a remediation email to the customer using the Gmail API.
We configure the LlmAgent
and package it in the ADKAgentModelHandler
:
The entire pipeline is declared cleanly. The upstream sentiment inference feeds directly into the filtering step (FilterNegativeADK
), which then conditionally executes the downstream ADKInference
:
By introducing this filtering step, we gain major engineering and operational advantages:
Instead of paying for Gemini input/output tokens on 100% of incoming events, we pay only for the fraction that represent negative customer sentiment (typically < 5% of messages). The other 95% are classified locally on CPU instances at zero incremental API cost.
Dataflow distributes the CPU classification workload across many instances. Since CPU inference takes milliseconds, the pipeline scales horizontally to handle high-throughput event streams. The heavyweight LLM agent, which can take seconds per request due to tool execution, is called sparingly, preventing backlog.
Adding the agent into the DAG requires no complex orchestration logic or manual thread pools. Using ADKAgentModelHandler
with Beam's native RunInference
transform handles parallel worker threads, batching, and integration automatically, keeping the codebase maintainable and clean.
Streaming data is fast and high-volume, while heavyweight generative AI reasoning is slow and costly.
By building a pre-filtered pipeline with Google Dataflow and the ADK, you get the best of both worlds: the cost and speed of local CPU-based models, and the deep, automated capabilities of Gemini-backed agents.
To see the complete codebase and deploy this yourself, check out the next-2026-demo GitHub repository. Apache Beam is a trademark of the Apache Software Foundation