Smart Streaming: Cut Gen AI Costs with Dataflow Pre-Filtering

Real-time streaming pipelines are the unsung heroes of modern enterprises, quietly processing everything from support tickets to transaction logs. But here’s the rub: traditional streaming DAGs are static—once deployed, their logic is set in stone. That’s where generative AI agents come in, turning rigid pipelines into adaptive systems that can plan, query, and act on the fly.

Imagine a customer fires off an angry message about a damaged order. Instead of just logging it, your pipeline could look up the order, decide on a fix (replacement or refund), email the customer, and log the resolution—all automatically. That’s the promise of agentic streaming, but it comes with a catch: scale, latency, and cost. Sending every raw event to a heavyweight model is like using a caiman to catch a minnow—overkill and expensive.

Enter Google Dataflow and the Agent Development Kit (ADK). This pattern combines a lightweight CPU model to filter events, sending only the tricky cases to a gen AI agent. It’s a universal blueprint for any high-volume stream where most events are routine and only a few need deep reasoning.

Why Pre-Filter Streaming Events?

In a high-throughput stream, the vast majority of messages are routine—positive feedback, neutral queries, simple requests. Routing every single one to a heavyweight LLM creates three bottlenecks:

• API cost: Frontier models charge per token, so costs scale linearly with volume.
• Latency: Multi-step workflows with database lookups take seconds, clogging the stream.
• Quotas: External APIs have rate limits that streaming workers can easily exhaust.

To avoid these, you build a pre-filtered pipeline in Apache Beam/Dataflow. A lightweight sentiment classifier runs on CPU, filtering out the noise. Only negative messages trigger the agent, which dynamically decides what actions to take—introducing adaptive branching without hardcoding thousands of conditionals.

Pipeline Flow: From Pub/Sub to Agentic Action

1. Ingestion: Read raw messages from Google Pub/Sub.
2. Lightweight sentiment classifier (CPU): Run all messages through a Hugging Face model (distilbert-base-uncased-finetuned-sst-2-english) using Beam’s RunInference transform. This executes locally on Dataflow workers, avoiding external API costs.
3. Pre-qualification Gate: A simple DoFn filters the stream. Positive or neutral messages are acknowledged and dropped.
4. Automated Remediation (ADK): Only negative messages 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 an email via the Gmail API.

Adaptive Execution: Making the Beam DAG Dynamic

In traditional streaming, the DAG is rigid—change means redeploying the whole pipeline. By placing a gen AI agent downstream of the sentiment filter, you introduce a dynamic node. For the 95% of records that are positive or neutral, the pipeline runs along a fast, static path. But when a negative record comes through, the agent evaluates the payload and dynamically selects the correct sequence of API tools at runtime. This eliminates the need to build and maintain thousands of hardcoded conditional branches.

Implementing the Pipeline

Here’s a taste of the implementation. First, define the lightweight sentiment model:

model_handler = HuggingFacePipelineModelHandler(
    task="sentiment-analysis",
    model="distilbert-base-uncased-finetuned-sst-2-english"
)

Then build the ADK agent with tools like lookup_user, lookup_orders, and send_email. Configure the LlmAgent and package it in the ADKAgentModelHandler:

adk_agent = LlmAgent(
    name="remediation_agent",
    model="gemini-3.5-flash",
    instruction=(
        "You are a customer service remediation assistant with access to "
        "BigQuery lookup tools and an email sending tool. "
        "When given a prompt describing a customer situation, follow the "
        "numbered steps exactly and use your tools to complete the task."
    ),
    tools=adk_tools,
)

adk_handler = ADKAgentModelHandler(agent=adk_agent)

Finally, assemble the Dataflow DAG:

with beam.Pipeline(options=pipeline_options) as p:
    sentiment_results = (
        p
        | "ReadFromPubSub" >> beam.io.ReadFromPubSub(topic=known_args.input_topic)
        | "DecodeMessages" >> beam.Map(lambda x: x.decode('utf-8'))
        | "SentimentInference" >> RunInference(model_handler)
    )

    _ = (
        sentiment_results
        | "FilterNegativeADK" >> beam.ParDo(FilterNegativeAndPromptADK())
        | "ADKInference" >> RunInference(adk_handler)
        | "LogADKResults" >> beam.ParDo(LogADKResponse())
    )

Cost and Performance Advantages

By introducing this filtering step, you gain major advantages:

1. Significant Cost Reductions

Instead of paying for Gemini tokens on 100% of events, you pay only for the fraction that are negative (typically < 5%). The other 95% are classified locally on CPU at zero incremental API cost.

2. High Streaming Throughput

Dataflow distributes the CPU classification across many instances. Since CPU inference takes milliseconds, the pipeline scales horizontally. The heavyweight LLM agent, which can take seconds per request, is called sparingly, preventing backlog.

3. Native Apache Beam Integration

Adding the agent into the DAG requires no complex orchestration logic. Using ADKAgentModelHandler with Beam’s native RunInference transform handles parallel worker threads, batching, and integration automatically, keeping the codebase clean.

Key Takeaways

Streaming data is fast and high-volume, while heavyweight gen 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 models, and the deep, automated capabilities of Gemini-backed agents. It’s like having a capybara’s efficiency with a caiman’s bite—only when needed.

For the complete codebase, check out the next-2026-demo GitHub repository.

If you’re exploring similar backend platforms, you might also want to check our Google Cloud Review and Supabase Review for more insights.

Original announcement published on Google Cloud.