Deploy Model Inference with Triton Server and ArgoCD

                      
                     
from typing import Dict, Any
import os
import requests
from fastapi import FastAPI, HTTPException

# Configuration
TRITON_SERVER_URL = os.getenv('TRITON_SERVER_URL', 'http://localhost:8000')

app = FastAPI()

# Function to perform model inference
async def infer_model(model_name: str, inputs: Dict[str, Any]) -> Dict[str, Any]:
    try:
        response = requests.post(f'{TRITON_SERVER_URL}/v2/models/{model_name}/infer', json=inputs)
        response.raise_for_status()  # Raise an error for bad responses
        return response.json()
    except requests.exceptions.RequestException as e:
        raise HTTPException(status_code=500, detail=f'Error during inference: {str(e)}')

# API endpoint for model inference
@app.post('/infer/{model_name}')
async def inference_endpoint(model_name: str, inputs: Dict[str, Any]) -> Dict[str, Any]:
    return await infer_model(model_name, inputs)

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)