def _transform()

in pipelines/ml_ai_python/ml_ai_pipeline/pipeline.py [0:0]

• 5 lines of code
• 1 McCabe index (conditional complexity)

def _transform(msgs: PCollection[str], model_path: str) -> PCollection[str]:
  preds: PCollection[
      PredictionResult] = msgs | "RunInference-Gemma" >> RunInference(
          GemmaModelHandler(model_path))
  return preds | "Format Output" >> beam.Map(_format_output)