mitzen

 What will happen when we set output wrongly?   pipeline_runner = LlmAgent (     name = "pipeline_runner" ,     model = MODEL ,     description = (         "Runs and coordinates the delivery pipeline from dev to test to "         "pre-prod to prod, enforcing promotion gates between stages."     ),     instruction = RUNNER_INSTRUCTION ,     tools = [ run_pipeline_stage , get_stage_status , get_stage_logs ],     output_key = "pipeline_run_report1111" , ) Generally this is what you will get:-  raise KeyError(f'Context variable not found: `{var_name}`.') KeyError: 'Context variable not found: `pipeline_run_report`.' And it looksl like this:

If we specify output_schema here for our agent and provide tools for it to call - you will notice that tool calling stops and other times it return default python model (using default values) :- pipeline_runner = LlmAgent (     name = "pipeline_runner" ,     model = MODEL ,     description = (         "Runs and coordinates the delivery pipeline from dev to test to "         "pre-prod to prod, enforcing promotion gates between stages."     ),     instruction = RUNNER_INSTRUCTION ,     output_schema = PipelineRunReport ,     tools = [ run_pipeline_stage , get_stage_status , get_stage_logs ],     output_key = "pipeline_run_report" , ) And this is our agentic setup:-  pipeline_coordinator = SequentialAgent (     name = "pipeline_coordinator" ,     description = (         "Coordinates a pipeline run: execute stages, log what ...

To fine tune gemini using kubeflow pipeline, we are using the following setup. There is no validation being configure here. You can easily added it tho. This will be a much more practical example of using kubeflow for :-  from kfp import compiler, dsl   # ---------------------------- edit these ---------------------------- PROJECT_ID = "project-id" LOCATION = "us-central1" BUCKET = "my-vertex-bucket-ai"   # the bucket where you uploaded train.jsonl   TRAIN_URI = f "gs:// {BUCKET} /text/train.jsonl" VALIDATION_URI = f "gs:// {BUCKET} /text/validation.jsonl"   # set to "" to skip validation   # Must be a Gemini model that supports supervised tuning. Check Google's # supported-models list, because it changes over time. SOURCE_MODEL = "gemini-2.5-flash" TUNED_MODEL_NAME = "ticket-classifier"   PIPELINE_ROOT = f "gs:// {BUCKET} /pipeline-root"   # where pipeline files are stored # ------------...

We are going to look at a simple kubeflow condition and parallel task when working with pipeline, This pipeline will greet pet owner and if there's pets we greet them too.  This is our example components:- @component def sayhi ( name : str ) -> int :   return 1 if name == "eric" else 2 @component def args_generator_op () -> list :     return [{ "cats" : "1" , "dogs" : "2" }, { "cats" : "10" , "dogs" : "20" }] And this is our pipeline code:- import json from kfp import dsl @dsl.pipeline (name= "control" , pipeline_root=PIPELINE_ROOT) def pipeline ():     name = sayhi(name= "greet-human" )       with dsl.If(name.output == 1 , "if has pets" ):           cat_and_dog = args_generator_op()           with dsl.Else( "if has pets" ):         cat_and_dog = args_generator_op()       with dsl.ParallelFor(cat_and_dog.output, name= "hi-to-pe...

We can easily run ML pipeline using aiplatfor sdk using the following setup - notebook  PROJECT_ID = "project-your-id"   # @param {type:"string"} LOCATION = "us-central1"   # @param {type:"string"} BUCKET_URI = f "gs://your-bucket-name- {PROJECT_ID} -unique"   # @param {type:"string"} ! gcloud storage buckets create --location={LOCATION} --project={PROJECT_ID} {BUCKET_URI} SERVICE_ACCOUNT = "ml-pipeline-dev-sa"   # @param {type:"string"} import sys IS_COLAB = "google.colab" in sys.modules if (     SERVICE_ACCOUNT == ""     or SERVICE_ACCOUNT is None     or SERVICE_ACCOUNT == "ml-pipeline-dev-sa" ):     # Get your service account from gcloud     if not IS_COLAB:         shell_output = ! gcloud auth list 2 >/dev/null         SERVICE_ACCOUNT = shell_output[ 2 ].replace( "*" , "" ).strip()     if IS_COLAB:         shell_output = !...

We can create gcp agent platform pipeline using kubeflow or tensorflow SDK (this is for large dataset which span terrabytes).  Kubeflow sdk are basically python script that you defines pipelines or components. @kfp.dsl.pipeline and @kfp.dsl.componet. Lets gets started by installing the required modules  uv install kfp.dsl  google-cloud-pipeline-components  docker In its simplistic form,  a pipeline can look like this (this will not work for Windows setup with docker) local_pipeline.py from kfp import dsl # Define a component to add two numbers. @ dsl . component def add ( a : int , b : int ) -> int :     return a + b # Define a simple pipeline using the component. @ dsl . pipeline def addition_pipeline ( x : int , y : int , z : int ) -> int :     task1 = add ( a = x , b = y )     task2 = add ( a = task1 . output , b = z )     return task2 . output And here is our runner  from kfp import local fro...

Let's say you would like to ensure that you pipeline operates in a sequential manner - lockBehaviour  option- sequential in a sense that if user A trigger pipeline at 12:00 and then another user B trigger pipeline at 12:01, user B will have to wait until user A completes.  If user a cancel it, User B will have to wait until the cancellation completes. This can take a while before next pipeline start executing.  There are 2 option available - sequential and runLatest. stages: - stage: A   lockBehavior: sequential   jobs:   - job: Job     steps:     - script: Hey! If we do not specify, lockBehaviour is runLatest. Potential issue :- Queued runs waiting on a check can eventually time out, so B could fail if it waits too long behind other runs. Only runs targeting the same protected resource (for example, the same environment, repository, agent pool, service connection) compete for the lock.

With Azure Devops future uncertain - find myself hardly read or keep up to date with it. Well just in case, I do, there is always this link here to goto  :) https://learn.microsoft.com/en-us/azure/devops/release-notes/2026/sprint-280-update

In gcp agentic platform mcp server is an abtraction that allows your agent to call a actual MCP hosted elsewhere.  It is not to create and deploy an MCP server. You can of course do that via Google Cloud run.  Typically you can use the following code in your agentic runtime to call this MCP server  import os from google.adk.integrations.agent_registry import AgentRegistry from google.auth import default from google.adk.agents import Agent from google.adk.models import Gemini from google.genai import types _ , project_id = default() LOCATION = os.environ.setdefault( "GOOGLE_CLOUD_LOCATION" , "us-central1" ) MCP_SERVER_NAME = os.environ.get( "MCP_SERVER_NAME" , "agentregistry-00000000-0000-0000-bd07-35253020786b" ) os.environ[ "GOOGLE_GENAI_USE_VERTEXAI" ] = "True" registry = AgentRegistry(project_id=project_id, location=LOCATION) mcp_toolset = registry.get_mcp_toolset(     f "projects/project-id/locations/us-cen...

 This is an example of how we can deployment agent runtime agent with environment variables. The key settings here is the "env_vars".  import asyncio from urllib import response from google . adk . agents import Agent from google . adk . runners import Runner from google . adk . sessions import InMemorySessionService from google . genai import types from google . adk . agents . callback_context import CallbackContext from google . adk . models . llm_request import LlmRequest from vertexai import types as vertex_types from google . genai import types import vertexai import agentplatform from agentplatform import types   # check below: this is my assumption from agentplatform . frameworks import AdkApp # 1. Define the callback function def inspect_llm_input ( callback_context : CallbackContext , llm_request : LlmRequest ):     """Callback triggered directly before ADK sends payload to the LLM."""     print ( ...