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.
A universal blueprint for high-volume streams
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.
The architecture: Why pre-filter streaming events?
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:
Pipeline flow
-
Ingestion: Read raw customer messages from Google Pub/Sub.
-
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’sRunInferencetransform. This executes locally on the Dataflow worker CPUs, avoiding external API costs. -
Pre-qualification Gate: A simple
DoFnfilters the stream. Messages withPOSITIVEorNEUTRALsentiment are acknowledged and dropped. -
Automated Remediation (ADK): If and only if a message is classified as
NEGATIVE, we trigger the gen AI agent backed bygemini-3.5-flashusing theADKAgentModelHandler. 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.
Adaptive execution: Making the Beam DAG dynamic
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.
Implementing the pipeline
Here is an example implementation in Apache Beam using the Google Agent Development Kit (ADK) and the RunInference framework.
1. Defining the lightweight sentiment model
We define the upstream CPU model using HuggingFacePipelineModelHandler. This model classifies sentiment into POSITIVE, NEUTRAL, or NEGATIVE on the worker instance.
- code_block
- <ListValue: [StructValue([('code', 'model_handler = HuggingFacePipelineModelHandler(rn task="sentiment-analysis",rn model="distilbert-base-uncased-finetuned-sst-2-english"rn)'), ('language', ''), ('caption', )])]>
2. Building the heavyweight ADK agent
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.
- code_block
- dict:rn “””Look up user information (email address) from BigQuery by user ID.”””rn from google.cloud import bigqueryrnrn client = bigquery.Client(project=project)rn query = (rn f”SELECT user_id, user_email “rn f”FROM `{project}.{dataset}.users` “rn f”WHERE user_id = @user_id”rn )rn job_config = bigquery.QueryJobConfig(rn query_parameters=[bigquery.ScalarQueryParameter(“user_id”, “INT64”, user_id)]rn )rn try:rn results = list(client.query(query, job_config=job_config).result())rn if results:rn row = results[0]rn return {“user_id”: row.user_id, “user_email”: row.user_email}rn return {“error”: f”No user found with user_id={user_id}”}rn except Exception as exc:rn return {“error”: str(exc)}rnrn def lookup_orders(user_id: int) -> dict:rn “””Look up a user’s orders and current product inventory from BigQuery.”””rn from google.cloud import bigqueryrnrn client = bigquery.Client(project=project)rn query = (rn f”SELECT p.order_id, p.product_id, pr.remaining_inventory, pr.price “rn f”FROM `{project}.{dataset}.purchases` p “rn f”JOIN `{project}.{dataset}.products` pr ON p.product_id = pr.product_id “rn f”WHERE p.user_id = @user_id”rn )rn job_config = bigquery.QueryJobConfig(rn query_parameters=[bigquery.ScalarQueryParameter(“user_id”, “INT64”, user_id)]rn )rn try:rn results = list(client.query(query, job_config=job_config).result())rn orders = [rn {rn “order_id”: row.order_id,rn “product_id”: row.product_id,rn “remaining_inventory”: row.remaining_inventory,rn “price”: float(row.price),rn }rn for row in resultsrn ]rn return {“orders”: orders}rn except Exception as exc:rn return {“error”: str(exc)}rnrn def send_email(to_address: str, subject: str, body: str) -> str:rn “””Send a plain-text email to the customer via the Gmail API.”””rn import google.authrn import googleapiclient.discoveryrn import email.mime.textrn import base64rnrn try:rn creds, _ = google.auth.default(rn scopes=[“https://www.googleapis.com/auth/gmail.send”]rn )rn service = googleapiclient.discovery.build(“gmail”, “v1”, credentials=creds)rnrn mime_msg = email.mime.text.MIMEText(body)rn mime_msg[“to”] = to_addressrn mime_msg[“subject”] = subjectrn raw = base64.urlsafe_b64encode(mime_msg.as_bytes()).decode(“utf-8″)rn service.users().messages().send(userId=”me”, body={“raw”: raw}).execute()rn return “Email sent successfully”rn except Exception as exc:rn return f”Failed to send email: {exc}”rnrn return [lookup_user, lookup_orders, send_email]’), (‘language’, ”), (‘caption’, )])]>
We configure the LlmAgent and package it in the ADKAgentModelHandler:
- code_block
- <ListValue: [StructValue([('code', 'adk_agent = LlmAgent(rn name="remediation_agent",rn model="gemini-3.5-flash",rn instruction=(rn "You are a customer service remediation assistant with access to "rn "BigQuery lookup tools and an email sending tool. "rn "When given a prompt describing a customer situation, follow the "rn "numbered steps exactly and use your tools to complete the task."rn ),rn tools=adk_tools,rn)rnrn# RunInference handler for the ADK agentrnadk_handler = ADKAgentModelHandler(agent=adk_agent)'), ('language', ''), ('caption', )])]>
3. Assembling the Dataflow DAG
The entire pipeline is declared cleanly. The upstream sentiment inference feeds directly into the filtering step (FilterNegativeADK), which then conditionally executes the downstream ADKInference:
- code_block
- > beam.io.ReadFromPubSub(topic=known_args.input_topic)rn | “DecodeMessages” >> beam.Map(lambda x: x.decode(‘utf-8′))rn | “SentimentInference” >> RunInference(model_handler)rn )rnrn # 2. Filter out non-negative sentiment and invoke the ADK Agentrn _ = (rn sentiment_resultsrn | “FilterNegativeADK” >> beam.ParDo(FilterNegativeAndPromptADK())rn | “ADKInference” >> RunInference(adk_handler)rn | “LogADKResults” >> beam.ParDo(LogADKResponse())rn )’), (‘language’, ”), (‘caption’, )])]>
Cost and performance advantages
By introducing this filtering step, we gain major engineering and operational advantages:
1. Significant cost reductions
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.
2. High streaming throughput
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.
3. Native Apache Beam integration
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.
Key takeaways
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