The Pipeline Bridge listens for pipeline trigger requests on NATS and submits them to the appropriate workflow engine (Argo Workflows or Kubeflow Pipelines). It monitors execution and publishes status updates back to NATS.

NATS Subjects

Subject	Direction	Description
`ai.pipeline.trigger`	Subscribe	Pipeline trigger requests
`ai.pipeline.status.{request_id}`	Publish	Pipeline status updates

Supported Pipelines

Pipeline	Engine	Description
`document-ingestion`	Argo	Ingest documents into Milvus
`batch-inference`	Argo	Run batch LLM inference
`model-evaluation`	Argo	Evaluate model performance
`rag-query`	Kubeflow	Execute RAG query pipeline
`voice-pipeline`	Kubeflow	Full voice assistant pipeline

Request Format

{
  "request_id": "uuid",
  "pipeline": "document-ingestion",
  "parameters": {
    "source-url": "s3://bucket/docs/",
    "collection-name": "knowledge_base"
  }
}

Response Format

{
  "request_id": "uuid",
  "status": "submitted",
  "pipeline": "document-ingestion",
  "engine": "argo",
  "run_id": "document-ingestion-abc123",
  "message": "Pipeline submitted successfully",
  "timestamp": "2026-01-03T12:00:00Z"
}

Status Updates

The bridge publishes status updates as the workflow progresses:

submitted - Workflow created
pending - Waiting to start
running - In progress
succeeded - Completed successfully
failed - Failed
error - System error

Implementation

The pipeline bridge uses the handler-base library for standardized NATS handling, telemetry, and health checks.

Environment Variables

Variable	Default	Description
`NATS_URL`	`nats://nats.ai-ml.svc.cluster.local:4222`	NATS server URL
`KUBEFLOW_HOST`	`http://ml-pipeline.kubeflow.svc.cluster.local:8888`	Kubeflow Pipelines API
`ARGO_HOST`	`http://argo-server.argo.svc.cluster.local:2746`	Argo Workflows API
`ARGO_NAMESPACE`	`ai-ml`	Namespace for Argo Workflows

Building

docker build -t pipeline-bridge:latest .

# With specific handler-base tag
docker build --build-arg BASE_TAG=latest -t pipeline-bridge:latest .

Testing

# Port-forward NATS
kubectl port-forward -n ai-ml svc/nats 4222:4222

# Trigger document ingestion
nats pub ai.pipeline.trigger '{
  "request_id": "test-1",
  "pipeline": "document-ingestion",
  "parameters": {"source-url": "https://example.com/docs.txt"}
}'

# Monitor status
nats sub "ai.pipeline.status.>"

License

MIT