Building cost-effective, high-throughput gen AI workflows in Google Dataflow

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. codeblock )])]> 2. Building the heavyweight ADK agent The ADK agent acts as our remediation assistant. We equip it with three tools: lookupuser: Queries BigQuery for the customer's email. lookuporders: Queries BigQuery for the customer's orders and current product inventory. sendemail: Sends a remediation email to the customer using the Gmail API. codeblock dict:\r\n """Look up user information (email address) from BigQuery by user ID."""\r\n from google.cloud import bigquery\r\n\r\n client = bigquery.Client(project=project)\r\n query = (\r\n f"SELECT userid, useremail "\r\n f"FROM {project}.{dataset}.users "\r\n f"WHERE userid = @userid"\r\n )\r\n jobconfig = bigquery.QueryJobConfig(\r\n queryparameters=[bigquery.ScalarQueryParameter("userid", "INT64", userid)]\r\n )\r\n try:\r\n results = list(client.query(query, jobconfig=jobconfig).result())\r\n if results:\r\n row = results[0]\r\n return {"userid": row.userid, "useremail": row.useremail}\r\n return {"error": f"No user found with userid={userid}"}\r\n except Exception as exc:\r\n return {"error": str(exc)}\r\n\r\n def lookuporders(userid: int) -> dict:\r\n """Look up a user\'s orders and current product inventory from BigQuery."""\r\n from google.cloud import bigquery\r\n\r\n client = bigquery.Client(project=project)\r\n query = (\r\n f"SELECT p.orderid, p.productid, pr.remaininginventory, pr.price "\r\n f"FROM {project}.{dataset}.purchases p "\r\n f"JOIN {project}.{dataset}.products pr ON p.productid = pr.productid "\r\n f"WHERE p.userid = @userid"\r\n )\r\n jobconfig = bigquery.QueryJobConfig(\r\n queryparameters=[bigquery.ScalarQueryParameter("userid", "INT64", userid)]\r\n )\r\n try:\r\n results = list(client.query(query, jobconfig=jobconfig).result())\r\n orders = [\r\n {\r\n "orderid": row.orderid,\r\n "productid": row.productid,\r\n "remaininginventory": row.remaininginventory,\r\n "price": float(row.price),\r\n }\r\n for row in results\r\n ]\r\n return {"orders": orders}\r\n except Exception as exc:\r\n return {"error": str(exc)}\r\n\r\n def sendemail(toaddress: str, subject: str, body: str) -> str:\r\n """Send a plain-text email to the customer via the Gmail API."""\r\n import google.auth\r\n import googleapiclient.discovery\r\n import email.mime.text\r\n import base64\r\n\r\n try:\r\n creds, = google.auth.default(\r\n scopes=["https://www.googleapis.com/auth/gmail.send"]\r\n )\r\n service = googleapiclient.discovery.build("gmail", "v1", credentials=creds)\r\n\r\n mimemsg = email.mime.text.MIMEText(body)\r\n mimemsg["to"] = toaddress\r\n mimemsg["subject"] = subject\r\n raw = base64.urlsafeb64encode(mimemsg.asbytes()).decode("utf-8")\r\n service.users().messages().send(userId="me", body={"raw": raw}).execute()\r\n return "Email sent successfully"\r\n except Exception as exc:\r\n return f"Failed to send email: {exc}"\r\n\r\n return [lookupuser, lookuporders, sendemail]'), ('language', ''), ('caption', )])]> We configure the LlmAgent and package it in the ADKAgentModelHandler: codeblock )])]> 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: codeblock > beam.io.ReadFromPubSub(topic=knownargs.inputtopic)\r\n | "DecodeMessages" >> beam.Map(lambda x: x.decode(\'utf-8\'))\r\n | "SentimentInference" >> RunInference(modelhandler)\r\n )\r\n\r\n # 2. Filter out non-negative sentiment and invoke the ADK Agent\r\n = (\r\n sentimentresults\r\n | "FilterNegativeADK" >> beam.ParDo(FilterNegativeAndPromptADK())\r\n | "ADKInference" >> RunInference(adk_handler)\r\n | "LogADKResults" >> beam.ParDo(LogADKResponse())\r\n )'), ('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 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.

1. 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