This happens a lot for the local Prism runner. Have you tried to run this with Dataflow?
On Sat, Oct 18, 2025 at 7:01 AM Marc _ <[email protected]> wrote: > HI Danny > thank a lot!.. worked as a charm for some simple instructions and 'low > tokens request' > But , when using a complex system instruction which requires some work on > the Model side i am getting back this > > Traceback (most recent call last): > File "C:\Users\Marco\envs\obb-dataflow\Lib\threading.py", line 1045, in > _bootstrap_inner > self.run() > File "C:\Users\Marco\envs\obb-dataflow\Lib\threading.py", line 982, in > run > self._target(*self._args, **self._kwargs) > File > "C:\Users\Marco\envs\obb-dataflow\Lib\site-packages\apache_beam\runners\portability\portable_runner.py", > line 533, in read_messages > for message in self._message_stream: > File > "C:\Users\Marco\envs\obb-dataflow\Lib\site-packages\grpc\_channel.py", line > 543, in __next__ > return self._next() > ^^^^^^^^^^^^ > File > "C:\Users\Marco\envs\obb-dataflow\Lib\site-packages\grpc\_channel.py", line > 969, in _next > raise self > grpc._channel._MultiThreadedRendezvous: <_MultiThreadedRendezvous of RPC > that terminated with: > status = StatusCode.DEADLINE_EXCEEDED > details = "Deadline Exceeded" > debug_error_string = "UNKNOWN:Error received from peer > {grpc_message:"Deadline Exceeded", grpc_status:4, > created_time:"2025-10-18T10:58:33.9731013+00:00"}" > > > > Any idea on how i can get around it> can i somehow control how much to > wait for a Model response? > > Kind regards > Marco > > > > > On Mon, Oct 13, 2025 at 4:43 PM Danny McCormick <[email protected]> > wrote: > >> The model handler lets you pass in any inference function you want. For >> example, the notebook uses the default one defined here - >> https://github.com/apache/beam/blob/08b480000ec859292d0f7bbadafb72328d3e9e16/sdks/python/apache_beam/ml/inference/gemini_inference.py#L54 >> >> So you could define an inference function which passes in additional >> config, e.g. from https://pypi.org/project/google-genai/ >> >> ``` >> config=types.GenerateContentConfig( >> system_instruction='I say high, you say low', >> max_output_tokens=3, >> temperature=0.3, >> ), >> ``` >> >> On Sat, Oct 11, 2025 at 3:46 PM Marc _ <[email protected]> wrote: >> >>> Hello >>> i m a muppet. Did not notice this in the colab sample >>> https://github.com/blueviggen/beam-remote-llm-examples/blob/main/gemini_runinference_example.ipynb >>> >>> # Only supported for genai package 1.21.1 or earlier >>> output_text = gemini_response.content.parts[0].text >>> >>> Using that package i can run sample pipeline on my local machine >>> I have further questions on the GeminiModelHandler as i could not find >>> anything via google.. >>> >>> How can i specify system instructions ? I was able to do so with the >>> OpenAIHandler - below a snippet copied from a prev mail Danny sent me few >>> months ago >>> >>> class SampleOpenAIHandler(ModelHandler): >>> """DoFn that accepts a batch of images as bytearray >>> and sends that batch to the Cloud Vision API for remote inference""" >>> def __init__(self, oai_key, llm_instructions): >>> self.oai_key = oai_key >>> self.llm_instructions = llm_instructions >>> >>> def load_model(self): >>> """Initiate the Google Vision API client.""" >>> """Initiate the OAI API client.""" >>> client = openai.OpenAI( >>> # This is the default and can be omitted >>> api_key=self.oai_key, >>> ) >>> return client >>> >>> >>> def run_inference(self, batch, model, inference): >>> >>> >>> response = model.responses.create( >>> model="gpt-4o", >>> instructions=self.llm_instructions, >>> input=batch[0], >>> ) >>> return [response.output_text] >>> >>> >>> >>> Kind regards >>> Marco >>> >>> >>> On Fri, Oct 10, 2025 at 10:51 PM Marc _ <[email protected]> wrote: >>> >>>> Danny / XQ >>>> got some setback. >>>> I copied the colab gemini sample,. >>>> https://github.com/blueviggen/beam-remote-llm-examples/blob/main/gemini_runinference_example.ipynb >>>> >>>> I Have added the following lines to post processor to figure out what >>>> is going on >>>> >>>> input_prompt = element.example >>>> >>>> # The API response is in `element.inference` >>>> # Path to text: response -> candidates -> content -> parts -> text >>>> gemini_inference = element.inference >>>> print(f'element.inference is {gemini_inference}') >>>> print(gemini_inference[1]) >>>> >>>> The code works fine in colab and i can see that the response matches >>>> the docs >>>> element.inference is ('candidates', >>>> [Candidate(content=Content(parts=[Part(video_metadata=None, thought=None, >>>> inline_data=None, file_data=None, thought_signature=None, >>>> code_execution_result=None, executable_code=None, function_call=None, >>>> function_response=None, text='```json\n{\n "question": "What is 1+2?",\n >>>> "answer": 3\n}\n```')], role='model'), citation_metadata=None, >>>> finish_message=None, token_count=None, finish_reason=<FinishReason.STOP: >>>> 'STOP'>, url_context_metadata=None, avg_logprobs=None, >>>> grounding_metadata=None, index=0, logprobs_result=None, >>>> safety_ratings=None)]) >>>> >>>> But when i run the same pipeline on GCP DataFlow, it seems i can >>>> only capture the first HttpResponse with all headers. the rest of the >>>> response is gone....... >>>> Plus, the response is not a Candidate but an HttpResponse and hence >>>> the following code fails miserably >>>> >>>> Is it because GCP is running multiple workers and somehow i am only >>>> capturing the first 'streamed' response from the model? >>>> >>>> Kind regards >>>> Marco >>>> >>>> >>>> >>>> >>>> element.inference is ('sdk_http_response', HttpResponse( headers=<dict >>>> len=10> )) >>>> >>>> <https://console.cloud.google.com/logs/query;query=resource.type%3D%22dataflow_step%22%20resource.labels.job_id%3D%222025-10-10_14_40_13-2699715226748068892%22%20logName%3D%22projects%2Fdatascience-projects%2Flogs%2Fdataflow.googleapis.com%252Fworker%22%20resource.labels.step_id%3D%2528%22PostProcess%22%2529%20timestamp%20%3E%3D%20%222025-10-10T21:40:13.801Z%22%20timestamp%20%3C%3D%20%222025-10-10T21:46:28.995Z%22%20severity%3E%3DDEFAULT;timeRange=2025-10-10T21:44:25.660899877Z%2F2025-10-10T21:44:25.660899877Z--PT1H;storageScope=project;pinnedLogId=2025-10-10T21:44:25.660899877Z%2F6182574922792223526:174769:0:14686?hl=en&project=datascience-projects> >>>> >>>> <https://console.cloud.google.com/logs/query;query=resource.type%3D%22dataflow_step%22%20resource.labels.job_id%3D%222025-10-10_14_40_13-2699715226748068892%22%20logName%3D%22projects%2Fdatascience-projects%2Flogs%2Fdataflow.googleapis.com%252Fworker%22%20resource.labels.step_id%3D%2528%22PostProcess%22%2529%20timestamp%20%3E%3D%20%222025-10-10T21:40:13.801Z%22%20timestamp%20%3C%3D%20%222025-10-10T21:46:28.995Z%22%20severity%3E%3DDEFAULT;timeRange=2025-10-10T21:44:25.660899877Z%2F2025-10-10T21:44:25.660899877Z--PT1H;storageScope=project;pinnedLogId=2025-10-10T21:44:25.660899877Z%2F6182574922792223526:174769:0:14686?hl=en&project=datascience-projects> >>>> >>>> <https://console.cloud.google.com/logs/query;query=resource.type%3D%22dataflow_step%22%20resource.labels.job_id%3D%222025-10-10_14_40_13-2699715226748068892%22%20logName%3D%22projects%2Fdatascience-projects%2Flogs%2Fdataflow.googleapis.com%252Fworker%22%20resource.labels.step_id%3D%2528%22PostProcess%22%2529%20timestamp%20%3E%3D%20%222025-10-10T21:40:13.801Z%22%20timestamp%20%3C%3D%20%222025-10-10T21:46:28.995Z%22%20severity%3E%3DDEFAULT;timeRange=2025-10-10T21:44:25.660899877Z%2F2025-10-10T21:44:25.660899877Z--PT1H;storageScope=project;pinnedLogId=2025-10-10T21:44:25.660899877Z%2F6182574922792223526:174769:0:14686?hl=en&project=datascience-projects> >>>> >>>> <https://console.cloud.google.com/logs/query;query=resource.type%3D%22dataflow_step%22%20resource.labels.job_id%3D%222025-10-10_14_40_13-2699715226748068892%22%20logName%3D%22projects%2Fdatascience-projects%2Flogs%2Fdataflow.googleapis.com%252Fworker%22%20resource.labels.step_id%3D%2528%22PostProcess%22%2529%20timestamp%20%3E%3D%20%222025-10-10T21:40:13.801Z%22%20timestamp%20%3C%3D%20%222025-10-10T21:46:28.995Z%22%20severity%3E%3DDEFAULT;timeRange=2025-10-10T21:44:25.660899877Z%2F2025-10-10T21:44:25.660899877Z--PT1H;storageScope=project;pinnedLogId=2025-10-10T21:44:25.660899877Z%2F6182574922792223526:174769:0:14686?hl=en&project=datascience-projects> >>>> >>>> <https://console.cloud.google.com/logs/query;query=resource.type%3D%22dataflow_step%22%20resource.labels.job_id%3D%222025-10-10_14_40_13-2699715226748068892%22%20logName%3D%22projects%2Fdatascience-projects%2Flogs%2Fdataflow.googleapis.com%252Fworker%22%20resource.labels.step_id%3D%2528%22PostProcess%22%2529%20timestamp%20%3E%3D%20%222025-10-10T21:40:13.801Z%22%20timestamp%20%3C%3D%20%222025-10-10T21:46:28.995Z%22%20severity%3E%3DDEFAULT;timeRange=2025-10-10T21:44:25.660899877Z%2F2025-10-10T21:44:25.660899877Z--PT1H;storageScope=project;pinnedLogId=2025-10-10T21:44:25.660899877Z%2F6182574922792223526:174769:0:14686?hl=en&project=datascience-projects> >>>> >>>> >>>> >>>> >>>> >>>> >>>> >>>> >>>> >>>> >>>> >>>> >>>> On Thu, Oct 9, 2025 at 2:51 PM Marc _ <[email protected]> wrote: >>>> >>>>> thanks Danny / XQ, will have a look and report back if i am stuck >>>>> kr >>>>> >>>>> On Thu, Oct 9, 2025 at 2:22 PM Danny McCormick via user < >>>>> [email protected]> wrote: >>>>> >>>>>> If you're using the Dataflow runner, >>>>>> https://cloud.google.com/dataflow/docs/concepts/security-and-permissions#permissions >>>>>> has >>>>>> some info as well - basically you'd want to make sure the worker service >>>>>> account has access to the Vertex endpoint you're using. >>>>>> >>>>>> Thanks, >>>>>> Danny >>>>>> >>>>>> On Thu, Oct 9, 2025 at 9:18 AM XQ Hu via user <[email protected]> >>>>>> wrote: >>>>>> >>>>>>> I think >>>>>>> https://cloud.google.com/dataflow/docs/notebooks/run_inference_vertex_ai >>>>>>> has more details for you to get started. >>>>>>> >>>>>>> On Thu, Oct 9, 2025 at 7:14 AM Marc _ <[email protected]> wrote: >>>>>>> >>>>>>>> Hello all >>>>>>>> i want to port this example to a real dataflow pipeline i am >>>>>>>> running, as i want to move from >>>>>>>> OpenAI to dataflow >>>>>>>> >>>>>>>> >>>>>>>> https://github.com/blueviggen/beam-remote-llm-examples/blob/main/gemma_runinference_example.ipynb >>>>>>>> >>>>>>>> Could anyone advise on the authentication side for accessing >>>>>>>> VertexAI? >>>>>>>> >>>>>>>> Kind regards >>>>>>>> Marco >>>>>>>> >>>>>>>
