
# Edit Your Images Easily with Inpainting and Diffusion

Wondering how new technology supports us in image modification and editing? It turns out **diffusion** can help us a lot. Among these techniques, **image-to-image** methods like **impainting** are especially useful. Instead of generating a whole new image, we focus only on modifying specific parts of an existing image.

![example_mug.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/f16d2fe39613591d49925714f12473289fa989753b0677faab39f767a88e36f5.jpg)

Whether using **Stable Diffusion** or other variants, each can demonstrate benefits in specific fields.

In this tutorial, we will guide you through:
- Preparing the environment  
- Installing the library  
- Loading the model and performing inference in Python  
- Creating an interactive UI with **Gradio** for a better experience  

![example_necklake.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/46a2de1e80e17f8643ca6fb2ae7c63d9b048fbe53ff40c64a8c15401027d68a3.jpg)

---

**Requirements**: To run this lab, we suggest having a **GPU** with 12GB of VRAM, or using **Google Colab** is preferred.

---

## Prepare the Environment

In this tutorial, we run the **diffusion model** in a **Python 3.9** environment using the **Anaconda** manager. If you are using Google Colab, you can skip this step.

Below is how you can set up the environment:

```bash
# Create a new conda environment
conda create -n diffusion python=3.9
conda activate diffusion
```

Then, we need to install **Torch** and **Transformers**. These libraries are essential for running the diffusion model.

```bash
# Install the transformers library
pip install transformers
```

Pay attention to the version of **CUDA** on your system. If you are using Google Colab, you don’t need to install **Torch**. For example, in our system, we are using **CUDA 12.4** and above, so the installation will be:

```bash
# Install torch based on your CUDA version
pip install torch torchvision torchaudio --extra-index-url https://download.pytorch.org/whl/cu124
```

Next, we will install **Diffusers**, the module that enables inpainting.

```bash
# Install the diffusers library
pip install diffusers
```

---

## Load the Model

If you check out **Hugging Face**, several models are available. Remember to select the **impainting model**.

In this tutorial, let's use **diffusers/stable-diffusion-xl-1.0-inpainting-0.1** as our starting point.

The code to load a pipeline that applies this model is as follows:

```python
import torch
from diffusers import AutoPipelineForInpainting

# Load model and move it to GPU
pipe = AutoPipelineForInpainting.from_pretrained(
    "diffusers/stable-diffusion-xl-1.0-inpainting-0.1", 
    torch_dtype=torch.float16
).to("cuda")
```

**Explanation of the code above**:  
- We import **torch** and the **AutoPipelineForInpainting** from the **Diffusers** library.  
- The `from_pretrained` method loads the specified model, and we set it to run on the **GPU** using `torch.float16` for better performance.

---

## Load Image and Mask

The idea of **impainting** is to modify parts of an image.  

**Images** are the visuals we want to edit, while **masks** specify the areas we want to change.

We will load an available image and mask for testing as shown below:

```python
# Load an image to inpaint
url = "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/diffusers/inpaint.png"
response = requests.get(url)
image = Image.open(BytesIO(response.content)).convert("RGB")

# Load the mask image
mask_url = "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/diffusers/inpaint_mask.png"
mask_response = requests.get(mask_url)
mask_image = Image.open(BytesIO(mask_response.content)).convert("RGB")

# Resize images if necessary
image = image.resize((1024, 1024))
mask_image = mask_image.resize((1024, 1024))
```

**Explanation of the code**:  
- We load the image and the mask using URLs.  
- The images are converted to **RGB** format and resized to **1024x1024 pixels**.

Let’s visualize the images we loaded from the above URLs:

```python
# Visualize both images in the same chart using matplotlib 
import matplotlib.pyplot as plt

fig, ax = plt.subplots(1, 2, figsize=(10, 5))
ax[0].imshow(image)
ax[0].set_title("Image")
ax[0].axis("off")
ax[1].imshow(mask_image)
ax[1].set_title("Mask")
ax[1].axis("off")
plt.show()
```

![1.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/687275847dfbdadd4b50c7282b696d78907422d337a1affdfdd4119b778d4572.jpg)

As you can see, the original image shows a picture of a road and a mountain in the center. The mask covers the part of the mountain. Our aim is to use **impainting** to change this mountain using a prompt.

---

## Set Up the Prompt

A **prompt** is a textual description of what we want to generate in the masked area.

```python
generator = torch.Generator("cuda").manual_seed(92)
prompt = 'an elven castle, in the mountain mist'
```

We use a **generator** to ensure the output is consistent whenever we use the same prompt.

Next, we input the prompt, image, and mask into the pipeline:

```python
# Generate the inpainted image
impainted_image = pipe(prompt=prompt, image=image, mask_image=mask_image, generator=generator).images[0]
```

It may take some time to process, depending on the speed of your GPU.

Let’s visualize the results of inpainting:

```python
# Use matplotlib to visualize the image, mask, and inpainted image
fig, ax = plt.subplots(1, 3, figsize=(15, 5))
ax[0].imshow(image)
ax[0].set_title("Image")
ax[0].axis("off")
ax[1].imshow(mask_image)
ax[1].set_title("Mask")
ax[1].axis("off")
ax[2].imshow(impainted_image)
ax[2].set_title("Inpainted Image")
ax[2].axis("off")
plt.show()
```

![2.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/68a7eb4fa32fb99c3510a5a1227fcbc90bf4eeb1929bcf5a7213a0963f4d972c.jpg)

You can see that the area we masked before has been changed, while the rest of the image remains intact.

---

## Change the Parameters

We can adjust the parameters in the pipeline, such as **guidance scale**, **number of inference steps**, and **strength**.

```python
impainted_image = pipe(
    prompt=prompt,
    image=image,
    mask_image=mask_image,
    guidance_scale=8.0,
    num_inference_steps=20,  # Steps between 15 and 30 work well for us
    strength=0.99,  # Ensure `strength` is below 1.0
).images[0]
```

**Explanation of each parameter**:  
- **guidance_scale**: This controls how closely the generated image should match the prompt. Higher values make the image adhere more closely to the prompt.  
- **num_inference_steps**: This determines how many steps the model takes to generate the image. More steps can lead to better results.  
- **strength**: This adjusts how much of the original image is preserved. A value below 1.0 means some original features are retained.

---

## Interactive UI with Gradio

We can use **Gradio** to create an AI tool that allows users to draw a mask and apply inpainting.

```bash
# Install Gradio
pip install gradio
```

First, we create a function to receive the image, prompt, mask, and parameters:

```python
# Dummy inpainting function with hyperparameters
def impaint(img, prompt, num_steps, guidance_scale, strength):
    # Extract alpha channel to create a mask for inpainting
    background = img["background"][:, :, :3]  # Discard the alpha channel, keep RGB
    
    # Extract alpha channel to create a mask for inpainting
    alpha_channel = img["layers"][0][:, :, 3]
    mask = np.where(alpha_channel == 0, 0, 255).astype(np.uint8)
    
    background = Image.fromarray(background)    
    mask = Image.fromarray(mask)
    
    # Apply inpainting with the given prompt and hyperparameters
    impainted_image = pipe(prompt=prompt, image=background, 
                            mask_image=mask,
                            num_steps=num_steps, guidance_scale=guidance_scale, strength=strength
                            ).images[0]
    
    return impainted_image
```

- The **impaint** function takes in an image, prompt, and parameters.  
- It extracts the background and the alpha channel to create a mask.  
- The inpainting is performed with the given parameters, and the inpainted image is returned.

Next, we set up the Gradio interface:

```python
with gr.Blocks() as demo:
    
    with gr.Row():
        # Textbox to enter the inpainting prompt
        prompt_input = gr.Textbox(label="Inpainting Prompt", placeholder="Enter text prompt for inpainting")
        
    with gr.Row():
        # Numeric inputs for editing hyperparameters
        num_steps = gr.Slider(minimum=1, maximum=100, value=50, step=1, label="Number of Inference Steps")
        guidance_scale = gr.Slider(minimum=1.0, maximum=20.0, value=7.5, step=0.1, label="Guidance Scale")
        strength = gr.Slider(minimum=0.0, maximum=1.0, value=0.5, step=0.1, label="Strength")
        
    with gr.Row():
        img = gr.ImageEditor(
            crop_size="1:1",
            height="30vw"
        )
        im_preview = gr.Image(height="30vw")
        
    # Automatically process the uploaded image and show it immediately    
    btn = gr.Button("Process Image")

    btn.click(impaint, [img, prompt_input, num_steps, guidance_scale, strength], [im_preview])
```

- We set up the **Gradio** interface with input fields for the prompt and hyperparameters.  
- The **ImageEditor** allows users to draw the mask directly on the image.  
- When the **Process Image** button is clicked, the **impaint** function processes the inputs and displays the inpainted image.

Finally, we run the code:

```python
demo.launch()
```

To access the web app, go to: http://127.0.0.1:7860 in the browser. You can see the app interface as shown below:

![web.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/945119d1940ec1207c5c25fa26d27961718735a6d7a40a6653fcfc49a1c8a79c.jpg)

In this interface, you can change a variety of input options and upload your own image.

After uploading, you can use the drawing tool to draw a mask.  
![draw.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/11bd8fc78cb90d273af9a74813b4c5a00fdff7477efc3b27b5f12ba55e97b466.jpg)

After that, type in the form, adjust the values of parameters, and click the **Process Image** button.

![running.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/44bf66d7f841206a1cb692b5490f7b16135a6b15ae1664a03f1c6a5ae1bf7962.jpg)

After some time (depending on your GPU), we can view the result in the right panel.

![result.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/e2c0d0bb730b8c746e6b7e135c30e63f83bfcc455f6419d61d50f01439fd8d1c.jpg)

You can also change the prompt and other parameters to generate your desired output:

![result 2.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/d9a1ac55c0e8ca1650095fd525262bafad89126044d80d7714beecd625e79daa.jpg)

Finally, you can save the output image by clicking the download button.

---

## Conclusion

In this tutorial, we explored how to use **impainting** with **diffusion models** to modify specific parts of an image effectively. With libraries like **Diffusers** and **Gradio**, we can create interactive applications that make image editing easy and accessible.

Feel free to experiment with different prompts and parameters to see how they affect the inpainting results!

---

## References

Hugging Face. (n.d.). Inpainting. In *diffusers documentation*. Retrieved December 16, 2024, from [https://huggingface.co/docs/diffusers/using-diffusers/inpaint](https://huggingface.co/docs/diffusers/using-diffusers/inpaint)

Wondering how new technology supports us in image modification and editing? It turns out **diffusion** can help us a lot. Among these techniques, **image-to-image** methods like **impainting** are especially useful. Instead of generating a whole new image, we focus only on modifying specific parts of an existing image.

Edit Your Images Easily with Inpainting and Diffusion

# Airflow Macros and Jinja: Enhancing Machine Learning Workflows

## 1. Introduction:

Apache Airflow is a powerful tool for creating and managing data pipelines. In the world of Machine Learning (ML), where data processing and model training often involve complex, multi-step workflows, **Airflow shines as a way to organize and automate these processes**. For example:

![1.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/77cd873b1b2506602c861df5b09a7456ba03d5f1ab829aabf86ca423de07bbaa.jpg)

One of the biggest challenges in ML workflows is ***the need for flexibility***:
- Data can change
- Models need updating
- parameters require tuning
- etc.

**This is where Airflow's dynamic pipelines come in handy.** 

👉 They allow us to create workflows that can adapt to changing conditions without needing to rewrite our code constantly.

## 2. Airflow Macros: The Powerhouse of Dynamic Workflows

Airflow macros are pre-defined variables or functions that you can use in your Airflow tasks. They act like **placeholders** that get filled with real values when your task runs. 

This means you can write code once that works for different dates, times, or other changing values.

![2.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/1d4deef3578bc7e247b7991be859b0985fc677d992ae2a293c05becaab9c83ae.jpg)


**Note:**
- Imagine you have a pipeline that needs to process yesterday's data every day. Instead of changing the date manually each time, you can use a macro that automatically gives you yesterday's date. This makes your pipeline dynamic and saves you from potential errors.

### Jinja templating: The foundation for macros
Airflow uses a system called Jinja templating to make macros work. Jinja is like a smart find-and-replace tool. It looks for special markers in your code (usually wrapped in double curly braces like this: `{{ }}`) and replaces them with actual values when the code runs.

![3.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/1d742d50cb87e1df9579885c33c86d7b87c08c273930da55c2eb7025dfa0154f.jpg)

## 3. Essential Airflow Macros for ML Pipelines

Let's look at some of the most useful macros for ML workflows:

### Datetime macros for time-based operations

- `{{ ds }}`: This gives you the date of the current run in the format 'YYYY-MM-DD'. It's super useful for processing data for a specific day.

- `{{ yesterday_ds }}`: As the name suggests, this gives you yesterday's date. Perfect for when you need to look at recent data.

### ds, ds_nodash, ts, and their applications

- `{{ ds }}`: We mentioned this earlier. It's great for naming files or folders by date.
  Example: `output_{{ ds }}.csv` might create a file named `output_2023-09-14.csv`.

- `{{ ds_nodash }}`: This is like `ds`, but without the dashes. Useful when you can't have dashes in names.
  Example: `data_{{ ds_nodash }}.parquet` could create `data_20230914.parquet`.

- `{{ ts }}`: This gives you the current timestamp. It's handy for very precise timing or creating unique identifiers.
  Example: `model_{{ ts }}.pkl` might save a model as `model_2023-09-14T10:30:00+00:00.pkl`.

### params and their use in parametrizing DAGs

The `params` macro lets you pass custom parameters to your tasks. This is incredibly useful for making your DAGs flexible.

Example:
```python
default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 1, 1),
    'params': {
        'model_type': 'random_forest',
        'n_estimators': 100
    }
}

with DAG('ml_training_dag', default_args=default_args) as dag:
    train_model = PythonOperator(
        task_id='train_model',
        python_callable=train_model_func,
        op_kwargs={
            'model_type': '{{ params.model_type }}',
            'n_estimators': '{{ params.n_estimators }}'
        }
    )
```

In this example, we can easily change the model type or number of estimators without changing the DAG code itself.

### Other useful macros

- `{{ macros.ds_add(ds, days) }}`: This lets you add or subtract days from a given date. Useful for creating date ranges or looking at specific time periods.

- `{{ task_instance }}`: This gives you access to the current task instance, which can be helpful for getting runtime information or previous task results.

- `{{ dag }}`: This provides access to the DAG object, allowing you to get information about the DAG itself.

### Example ML Workflow with Macros:
![4.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/5cfa8dfce6403ae4a2fbf0c722b5d18491a53446330d6a9f9f0fb27e5304bda9.jpg)


**In this image:**
- The **Fetch Data** task dynamically pulls the data for the day using the `{{ ds }}` macro.
- The **Process Data** task uses `{{ yesterday_ds }}` to process data from the previous day.
- The **Train Model** task uses the `{{ params.model_type }}` and `{{ params.task_instance }}` macros to train the specified model type with instance-specific parameters.
- The **Deploy Model** task ensures that the model is deployed based on the model's instance data.

👉 This shows how macros and Jinja templating help in creating flexible and dynamic pipelines that adapt to different conditions without needing constant manual updates.

## 4. Creating Custom Macros

### How to create Custom Macros

You can create your own macros to encapsulate complex logic or frequently used operations. Here's how:

1. Define your macro function in a Python file (e.g., `custom_macros.py`):

```python
def days_ago(n):
    from datetime import datetime, timedelta
    return (datetime.now() - timedelta(days=n)).strftime('%Y-%m-%d')
```

2. Add the path to this file in your Airflow configuration (`airflow.cfg`):

```
user_defined_macros = /path/to/custom_macros.py
```

3. Use your custom macro in your DAG:

```python
task = BashOperator(
    task_id='print_date',
    bash_command='echo "Date from 7 days ago: {{ days_ago(7) }}"',
    dag=dag
)
```

### Custom Macros for ML-specific Operations

Let's create a macro for data normalization:

```python
def normalize_column(column_name, min_val, max_val):
    return f"""
    ({column_name} - {min_val}) / ({max_val} - {min_val}) AS normalized_{column_name}
    """

# In your DAG:
normalize_task = BigQueryOperator(
    task_id='normalize_data',
    sql=f"""
    SELECT 
        {{ normalize_column('feature1', 0, 100) }},
        {{ normalize_column('feature2', -50, 50) }}
    FROM `my_dataset.my_table`
    """,
    use_legacy_sql=False,
    dag=dag
)
```

This macro helps standardize the process of normalizing columns in your dataset.

## 5. Airflow Macros in Machine Learning Workflows

### Use Case: Automated Hyperparameter Tuning

We can use macros to dynamically set hyperparameters:

```python
def set_learning_rate(base_rate, decay_factor):
    return f"""
    {base_rate} * pow({decay_factor}, {{{{ task_instance.try_number - 1 }}}})
    """

train_model_task = PythonOperator(
    task_id='train_model',
    python_callable=train_model_func,
    op_kwargs={
        'learning_rate': "{{ set_learning_rate(0.1, 0.9) }}"
    },
    dag=dag
)
```

This macro adjusts the learning rate based on the number of task attempts, implementing a simple learning rate decay.

## 6. Advanced Techniques: Combining Macros with Airflow Operators

### PythonOperator with macros

```python
def process_data(date, feature_count):
    # Your data processing logic here
    pass

process_task = PythonOperator(
    task_id='process_data',
    python_callable=process_data,
    op_kwargs={
        'date': '{{ ds }}',
        'feature_count': '{{ task_instance.xcom_pull(task_ids="count_features") }}'
    },
    dag=dag
)
```

This example shows how to use macros to pass dynamic values to a Python function.

### BranchPythonOperator with macros

```python
def choose_model(ti):
    data_size = ti.xcom_pull(task_ids='check_data_size')
    if data_size > 1000000:
        return 'train_large_model'
    else:
        return 'train_small_model'

branch_task = BranchPythonOperator(
    task_id='choose_model',
    python_callable=choose_model,
    dag=dag
)

train_large_model = PythonOperator(
    task_id='train_large_model',
    python_callable=train_large_model_func,
    op_kwargs={'date': '{{ ds }}'},
    dag=dag
)

train_small_model = PythonOperator(
    task_id='train_small_model',
    python_callable=train_small_model_func,
    op_kwargs={'date': '{{ ds }}'},
    dag=dag
)

branch_task >> [train_large_model, train_small_model]
```

This example uses a BranchPythonOperator to dynamically choose which model to train based on the data size.

## 7. Best Practices and Conclusion

### Tips for effective use of macros in ML workflows

1. Keep macros simple and focused on a single task.
2. Use descriptive names for custom macros to enhance readability.
3. Document your macros, especially if they contain complex logic.
4. Test your macros thoroughly to ensure they behave as expected in different scenarios.

### Potential pitfalls and how to avoid them

1. Overuse of macros can make DAGs hard to read. Use them judiciously.
2. Be cautious with macros that access external resources, as they may slow down DAG parsing.
3. Remember that macros are evaluated at runtime, not during DAG parsing. This can lead to unexpected behavior if not considered.

## References:
[1] https://airflow.apache.org/docs/apache-airflow/1.10.12/macros-ref.html
[2] https://medium.com/@laxmirathaur11/airflow-beyond-basics-part-4-6215e8e5d6be

Apache Airflow is a powerful tool for creating and managing data pipelines. In the world of Machine Learning (ML), where data processing and model training often involve complex, multi-step workflows, **Airflow shines as a way to organize and automate these processes**. For example:

Airflow Macros and Jinja: Enhancing Machine Learning Workflows

In this article, we'll explore how companies are transitioning from ETL to ELT and why **"shift-left"** is becoming a key trend in modern data architecture. From optimizing data processing workflows to reducing costs and latency, shift-left ensures high-quality data right from the source, ready for real-time analysis and decision-making.

## 1. From ETL to ELT

**ETL (Extract, Transform, Load)** is a widely used data processing workflow for moving data from various sources into a target data warehouse or database, where it can be analyzed and effectively utilized. This process includes three main steps:

* **Extract:** Collect data from multiple sources, regardless of format, to prepare for further processing.  
* **Transform:** Clean and standardize the data, applying business rules to ensure consistency and relevance for analysis.  
* **Load:** Store the transformed data in the target warehouse, ready for analysis and use.  

![Screenshot 2024-11-11 at 10.30.13.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/90b10c2042e08984c2ff51266fd476631be28057a20afe16fdbc871a78245a2d.jpg)

### **Problems with ETL:**

* **Time and Resource Intensive:** The transformation phase occurs before loading, requiring heavy compute resources and extending processing times—especially with large data volumes.  
* **Scalability and Flexibility Issues:** Traditional ETL doesn’t scale well as data grows, and pre-load transformations limit flexibility when analytics needs change frequently.  
* **High Complexity:** ETL pipelines are complex to build and maintain, prone to errors, and rigid in execution order (extract → transform → load).  
* **Doesn’t Leverage Cloud Power Effectively:** Modern cloud data warehouses are highly scalable, but traditional ETL fails to fully utilize this due to pre-load transformations.  

![Screenshot 2024-11-11 at 10.30.16.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/e8b70f159d7ff85e478b2140bda6a7f67a366426be09e143e25a11003c4fad28.jpg)

### **The Emergence of ELT:**

* To address ETL's limitations, the **ELT (Extract, Load, Transform)** approach was introduced. ELT reverses the order: data is extracted and loaded directly into the data warehouse first, and transformations are performed afterward within the destination. This leverages the powerful processing capabilities of cloud platforms (e.g., BigQuery, Snowflake, or Redshift) for faster and more efficient data transformations.
* With ELT, raw data can be ingested immediately, allowing flexible, on-demand transformation based on analytical needs. ELT not only reduces cost and processing time but also scales better with growing data volumes—ideal for modern applications that demand high-throughput, complex data handling.

**However…**

## 2. When ELT Is Too Easy, It Comes With Trade-Offs

As data grows in **volume**, **velocity**, and **variety**, companies are increasingly adopting ELT—moving data to the cloud and processing it conveniently. But when this process becomes too fast and uncontrolled, it can lead to **"data swamp"** — massive, unstructured, and low-value datasets. This drives up costs without delivering proportional value.

A more critical issue arises when data loaded into storage systems is accessed by multiple teams, often resulting in repeated cleaning or transformation efforts. Additionally, the **"reverse ETL"** process (pushing data back into operational systems) can complicate data flows, increase latency, and make maintenance harder.

![Screenshot 2024-11-10 at 21.44.59.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/4f21ff5e5f16fb9f9f6fea86a80a82c2b0244e810f664ea81a54eee83d7cf440.jpg)

## 3. Time to "Shift-Left"

The **shift-left** architecture addresses ELT’s challenges by moving data processing and governance closer to the source systems. This ensures that data is processed accurately at the origin—once—and reused consistently across downstream systems.

So how do we process data right as it's created? The answer is **stream processing**, with a workflow like this:

1. Connect directly to the data stream.  
2. Use a **data contract** to clean and validate data quality from the source.  
3. Data engineers write transformation logic on the stream, preparing it for downstream systems.

> **Note:** A *data contract* is an agreement between data producers and data consumers, defining structure and quality expectations. It includes schema, metadata, encoding rules, and versioning policies to minimize impact on downstream systems.

![Screenshot 2024-11-11 at 10.56.19.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/ac2578516f73f106f6fb881b7a1c29ca3b1363e44d2abe7f8939c33b4bbc54bd.jpg)

Besides reducing latency and duplicated data processing, **shift-left** ensures fresher, more reliable data and helps control storage costs in cloud warehouses or other storage systems.

## 4. Netflix Uses "Shift-Left" Too

The **shift-left** approach is increasingly adopted by major tech companies like **Netflix**. With over 450 million events daily from 100+ million users across 190 countries, Netflix uses shift-left to optimize data processing. Instead of running batch jobs that take up to 8 hours, they now deliver **real-time recommendations** based on immediate user activity.

## 5. Conclusion

**Shift-left** is becoming a strategic solution in modern data management, enabling fast, accurate, and efficient processing at the earliest stages. By pushing data quality and transformation upstream, it reduces costs and latency while ensuring data readiness for real-time analytics and decision-making.

As data becomes more dynamic and diverse, **shift-left** is a critical step to maintain competitive advantage, optimize infrastructure, and support real-time needs.

> If you'd like to learn more about modern data architectures, practical implementation at scale, and how to optimize your data systems, don’t forget to sign up for FSDS's MLOps and Data Engineer courses.

## 6. References

* [Shifting Left: Discover What's Possible When You Process Data Closer to the Source](https://www.quix.io/blog/shifting-left-discover-whats-possible-when-you-process-data-closer-to-the-source)  
* [The Shift Left Architecture – From Batch and Lakehouse to Real-Time Data Products with Data Streaming](https://www.kai-waehner.de/blog/2024/06/15/the-shift-left-architecture-from-batch-and-lakehouse-to-real-time-data-products-with-data-streaming/)  
* [What is Shift Left?](https://www.confluent.io/learn/what-is-shift-left/)  
* [Apache Kafka + Flink + Snowflake: Cost Efficient Analytics and Data Governance](https://www.kai-waehner.de/blog/2024/04/26/apache-kafka-flink-snowflake-cost-efficient-analytics-and-data-governance/)  
* [Data Contracts for Schema Registry on Confluent Platform](https://docs.confluent.io/platform/current/schema-registry/fundamentals/data-contracts.html)

Discover Shift-Left Data Architecture – a solution that optimizes cost and improves data quality by moving data governance and validation processes closer to the source systems.

Shift-Left Data Architecture: Reducing Costs and Improving Data Quality at the Source

# Functools in Machine Learning

## Introduction:

You're working on a machine learning project:
Your model prediction function is slow because it performs the same calculations repeatedly for the same inputs.
- You've written a complex data preprocessing function, and you notice it's running the same operations on the same data multiple times, wasting time and resources.
- You've created a custom class to represent your data, but now you need to compare and sort instances of this class, which requires writing multiple comparison methods.

👉 This is where `functools` comes in.


## What is `functools`?

`functools` is a **module in Python's standard library** that provides **higher-order functions** and operations on **callable objects** (Functions, Classes, Built-in Functions, etc.) 
### Higher-Order Functions:

Higher-order functions are:
- functions that can either take other functions as arguments
- return functions as their results. 

![1.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/c360ce742dee6cf4d28249bfffad2881bad8e4d5c824a27a87466dd3ed636165.jpg)


Let's look at a simple example to understand this better:

```python
def apply_operation(func, x, y):
    return func(x, y)

def add(a, b):
    return a + b

def multiply(a, b):
    return a * b

# Using the higher-order function
result1 = apply_operation(add, 5, 3)
print(result1)  # Output: 8

result2 = apply_operation(multiply, 5, 3)
print(result2)  # Output: 15
```

In this example:

- `apply_operation` is a higher-order function because it takes another function (`func`) as an argument.
- We can pass different functions (`add` or `multiply`) to `apply_operation`, and it will use whichever function we provide.

#### Higher-Order Functions in Machine Learning example:

```python
import numpy as np

def apply_to_dataset(func, dataset):
	return func(dataset)

def normalize(dataset):
	return (dataset - np.mean(dataset)) / np.std(dataset)

def standardize(dataset):
	return (dataset - np.min(dataset)) / (np.max(dataset) - np.min(dataset))

# Sample dataset
data = np.array([1, 2, 3, 4, 5])
# Apply different preprocessing functions using apply_to_dataset
normalized_data = apply_to_dataset(normalize, data)
standardized_data = apply_to_dataset(standardize, data)

print("Normalized:", normalized_data)
print("Standardized:", standardized_data)
```

In this ML-related example, `apply_to_dataset` is a higher-order function that can apply any preprocessing function to our dataset.

We can easily switch between different preprocessing methods (`normalize` or `standardize`) without changing the main function.

### Key Functions in `functools`

#### 1. `functools.partial`

The `partial` function from the `functools` module is a powerful tool that allows you to "freeze" some portion of the function's arguments, creating a new function that you can call with fewer arguments. This is particularly useful for reusing functions with different fixed arguments.

Example:

```python
from functools import partial

def greet(greeting, name):
    return f"{greeting}, {name}!"

# Creating a new function 'hello' where the 'greeting' argument is pre-set to "Hello"
hello = partial(greet, "Hello")

# Now, you only need to pass the 'name' argument when calling 'hello'
print(hello("Alice"))  # Output: Hello, Alice!
print(hello("Bob"))    # Output: Hello, Bob!

```


- The `partial` function is used to create a new function, `hello`, which is a version of the `greet` function where the `greeting` argument is pre-set to "Hello".
- When you call `hello("Alice")`, it effectively calls `greet("Hello", "Alice")`.
- This is very useful when you need to reuse the same function but with some arguments already fixed.
##### Real-world use case:
For instance, if you are working with a mathematical function that frequently requires the same parameters, you can use `partial` to simplify your code and reduce redundancy.

#### 2. `functools.lru_cache`

`lru_cache` is a decorator that implements memoization, caching a function's results based on its arguments.

Example:

```python
from functools import lru_cache

# Counter to keep track of function calls
call_counter = 0

@lru_cache(maxsize=None)
def fibonacci(n):
    global call_counter
    call_counter += 1  # Increment the counter each time the function is called
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

# Compute the Fibonacci number for 100
print(f"Fibonacci(100): {fibonacci(100)}")

# Verify how many times the function was called
print(f"Function was called {call_counter} times.")
```

Output:

```shell
Fibonacci(100): 354224848179261915075
Function was called 101 times.
```

In this case, `fibonacci(100)` is computed efficiently with only 101 calls to the function (instead of an exponentially large number without caching).

#### 3. `functools.cache`

`cache` is similar to `lru_cache` but provides simpler syntax for unlimited caching.

Example:

```python
from functools import cache

# Counter to track the number of function calls
call_counter = 0

@cache
def expensive_computation(x):
    global call_counter
    call_counter += 1  # Increment the counter each time the function is called
    # Simulate an expensive operation
    return x ** 2

# First call - function is computed
print(expensive_computation(10))  # Output: 100 (computed)

# Second call - result is retrieved from the cache
print(expensive_computation(10))  # Output: 100 (cached)

# Verify how many times the function was called
print(f"Function was called {call_counter} times.")

```

```shell
100  # Computed
100  # Retrieved from cache
Function was called 1 times.
```

**This output verifies that the second call retrieves the result from the cache rather than recomputing it.**

#### 4. `functools.total_ordering`

`total_ordering` is a class decorator that fills in missing comparison methods in a class that defines at least one comparison method.

Example:

```python
from functools import total_ordering

@total_ordering
class Person:
    def __init__(self, name, age):
        self.name = name
        self.age = age

    def __eq__(self, other):
        return self.age == other.age

    def __lt__(self, other):
        return self.age < other.age

p1 = Person("Alice", 30)
p2 = Person("Bob", 25)
print(p1 > p2)  # True
```

## Why `functools` is Valuable in Machine Learning

In machine learning projects, we often deal with complex, computationally expensive operations, large datasets, and the need for flexible, reusable code. `functools` provides tools that can help address these challenges. 

Let's explore the main functions in `functools` and see how they can be applied in ML contexts.

### 1. `functools.partial`

**What**: `partial` creates a new function with some of the arguments of the original function pre-set.

**Why it's useful in ML**: 
- Simplifies the creation of specialized functions from more general ones
- Helps in creating consistent preprocessing or model configuration functions

**Example in ML**:

```python
from functools import partial
from sklearn.ensemble import RandomForestClassifier

# Create a partially configured RandomForest
rf_classifier = partial(RandomForestClassifier, 
                        n_estimators=100, 
                        random_state=42)

# Now you can create classifiers with different max_depths easily
clf_shallow = rf_classifier(max_depth=5)
clf_deep = rf_classifier(max_depth=15)

# This is especially useful in grid search or pipelines
from sklearn.model_selection import GridSearchCV

param_grid = {'max_depth': [5, 10, 15, 20]}
grid_search = GridSearchCV(rf_classifier(), param_grid, cv=5)

# You can fit this grid_search object to your data
# grid_search.fit(X, y)
```

In this example, `partial` allows us to create a base RandomForest configuration, making it easier to experiment with different `max_depth` values while keeping other parameters constant.

### 2. `functools.lru_cache`

**What**: `lru_cache` is a decorator that caches the results of a function, avoiding repeated computations for the same inputs.

**Why it's useful in ML**: 
- Speeds up expensive computations, especially in feature engineering or model inference
- Reduces redundant calculations in iterative processes like cross-validation

**Example in ML**:

```python
from functools import lru_cache
import numpy as np

@lru_cache(maxsize=100)
def expensive_feature_computation(data):
    # Simulate an expensive computation
    return np.mean(np.array(data) ** 2)

# In your feature engineering pipeline
def engineer_features(dataset):
    return [expensive_feature_computation(tuple(row)) for row in dataset]

# The function will only compute for unique rows, reusing cached results
dataset = [(1, 2, 3), (4, 5, 6), (1, 2, 3)]
features = engineer_features(dataset)
print(features)
```

Here, `lru_cache` ensures that the expensive feature computation is only done once for each unique input, significantly speeding up the feature engineering process for datasets with repeated or similar samples.

### 3. `functools.cache`

**What**: Similar to `lru_cache`, but with no size limit on the cache.

**Why it's useful in ML**: 
- Ideal for pure functions with a limited input domain
- Useful in recursive algorithms or dynamic programming approaches in ML

**Example in ML**:

```python
from functools import cache

@cache
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

# This can be used in feature engineering or certain ML algorithms
def fib_feature(x):
    return fibonacci(int(x))

# Apply to a dataset
dataset = [5, 8, 13, 5, 8]
fib_features = [fib_feature(x) for x in dataset]
print(fib_features)
```

While this is a simple example, it demonstrates how `cache` can optimize recursive computations, which could be part of more complex feature engineering or custom ML algorithms.

### 4. `functools.total_ordering`

**What**: A class decorator that automatically generates ordering methods (`__lt__`, `__gt__`, etc.) based on `__eq__` and one other comparison method.

**Why it's useful in ML**: 
- Simplifies the creation of custom, comparable objects (e.g., for custom metrics or model results)
- Useful in sorting or ranking scenarios in ML pipelines

**Example in ML**:

```python
from functools import total_ordering

@total_ordering
class ModelPerformance:
    def __init__(self, accuracy, f1_score):
        self.accuracy = accuracy
        self.f1_score = f1_score
    
    def __eq__(self, other):
        return (self.accuracy, self.f1_score) == (other.accuracy, other.f1_score)
    
    def __lt__(self, other):
        return (self.accuracy, self.f1_score) < (other.accuracy, other.f1_score)

# Now you can easily compare model performances
model1 = ModelPerformance(0.85, 0.82)
model2 = ModelPerformance(0.82, 0.85)

print(model1 > model2)  # True
print(max(model1, model2))  # ModelPerformance(0.85, 0.82)

# This is useful for sorting model results
models = [ModelPerformance(0.8, 0.75), ModelPerformance(0.85, 0.8), ModelPerformance(0.82, 0.78)]
best_model = max(models)
print(f"Best model: Accuracy {best_model.accuracy}, F1 {best_model.f1_score}")
```

This example shows how `total_ordering` can be used to create a custom class for model performance that can be easily compared and sorted, which is often necessary in model selection or ensemble methods.


### 5. Putting It All Together: `functools` in a ML System:

Here's a simple example showing how you might use `functools` in different parts of a machine learning system:

```python
from functools import lru_cache, partial, cache
import numpy as np
from sklearn.ensemble import RandomForestClassifier

# Data handling
@cache
def load_data(file_name):
    # Pretend we're loading data from a file
    return np.random.rand(1000, 10)

# Model setup
base_model = partial(RandomForestClassifier, n_estimators=100, random_state=42)

# Model evaluation
@lru_cache(maxsize=10)
def evaluate_model(model, X, y):
    # Pretend we're doing cross-validation here
    return np.mean(model.predict(X) == y)

# Prediction function
@lru_cache(maxsize=1000)
def predict(model, features):
    return model.predict([features])[0]

# Using these functions
X = load_data("fake_data.csv")
y = np.random.randint(0, 2, 1000)  # Fake labels

model = base_model(max_depth=5)
model.fit(X, y)

print(f"Model score: {evaluate_model(model, tuple(map(tuple, X)), tuple(y))}")

# Pretend API call
print(f"Prediction: {predict(model, tuple(X[0]))}")
```



## Conclusion

Throughout this tutorial, we've explored how the `functools` module can be a powerful ally in your machine learning and MLOps workflows. Let's recap the key takeaways:

1. **Efficiency Boost**: Functions like `lru_cache` and `cache` can significantly speed up your ML pipelines by avoiding redundant computations. This is particularly valuable in scenarios involving expensive feature engineering or repeated model inferences.

2. **Flexibility in Code**: `partial` allows you to create specialized versions of functions, making it easier to experiment with different model configurations or preprocessing steps. This flexibility is crucial in the iterative nature of ML development.

3. **Simplified Comparisons**: `total_ordering` simplifies the creation of custom, comparable objects, which can be invaluable when dealing with complex model metrics or custom performance measures.

4. **Cleaner, More Maintainable Code**: By leveraging these tools, you can write more modular and reusable code, which is essential in managing the complexity of ML projects.

5. **Optimized Resources**: Proper use of `functools` can lead to more efficient use of computational resources, a critical factor when working with large datasets or complex models.

## References:
[1] https://docs.python.org/3/library/functools.html
[2] https://www.geeksforgeeks.org/functools-module-in-python/
[3] https://towardsdatascience.com/introducing-pythons-functools-module-2c4cba4774e


`functools` is a **module in Python's standard library** that provides **higher-order functions** and operations on **callable objects** (Functions, Classes, Built-in Functions, etc.) 

 Functools in Machine Learning

Large Language Models (LLMs) are revolutionizing technology, driving applications like chatbots, content generators, and more. But despite their power, LLMs can be unpredictable—hallucinating facts, giving inconsistent answers, or degrading over time. To build trust and ensure reliability, evaluating and monitoring LLMs is essential.

LLM Evaluation and Observability Using Evidently

# Rate Limiter in Triton: Managing Resource Allocation for MLOps

## 1. Introduction

Deploying machine learning models can be tricky. One big challenge is managing computer resources like memory and processing power (GPU utilization). 

When you're running multiple models at once::

- Your GPU VRam might run out of memory
- Some models might use up all the resources, leaving others waiting
- The whole system could slow down or crash

This is where Triton Inference Server comes in handy. Triton is a tool that helps deploy and run machine learning models. It has a feature called the Rate Limiter, which helps manage resources better.

In this tutorial, we'll learn about Triton and its Rate Limiter. We'll start with the basics and then see how the Rate Limiter can solve resource problems.

## 2. Triton Inference Server Fundamentals

### What is Triton Inference Server?

Triton Inference Server is like a helpful manager for your machine learning models. It's **a software** that:

- Runs your models
- Handles requests to use these models
- Manages resources for smooth operation

### Key Features of Triton

- Supports many types of models (TensorFlow, PyTorch, ONNX, etc.)
- Can run multiple models at the same time
- Allows models to run on CPUs or GPUs
- Has tools to manage resources (like the Rate Limiter we'll learn about)

### Basic Architecture

Let's imagine Triton as a building with many rooms:

- Each room can hold a different model
- There's a front desk (API) where requests come in
- The building manager (Triton) decides **which** room (model) to use and **when**

## 3. Getting Started with Triton

Let's set up a simple Triton server and run a basic model.

### Installation

1. The easiest way to get started is using Docker. Install Docker if you haven't already.
2. Pull the Triton Docker image:
```shell
docker pull nvcr.io/nvidia/tritonserver:23.09-py3-sdk
```

### Running a Simple Triton Server

1. Create a folder for your models:
   ```shell
   mkdir -p models/simple_model/1
   ```

2. For this example, let's use a pre-trained ONNX model. Download a simple model (like a small image classifier) and place it in `models/simple_model/1/model.onnx`.

3. Create a configuration file `models/simple_model/config.pbtxt`:
   ```txt
   name: "simple_model"
   platform: "onnxruntime_onnx"
   max_batch_size: 8
   input [
     {
       name: "input"
       data_type: TYPE_FP32
       dims: [ 3, 224, 224 ]
     }
   ]
   output [
     {
       name: "output"
       data_type: TYPE_FP32
       dims: [ 1000 ]
     }
   ]
   ```

4. Start the Triton server:
```shell
docker run --gpus all --rm -p 8000:8000 -p 8001:8001 -p 8002:8002 -v $(pwd)/models:/models nvcr.io/nvidia/tritonserver:21.09-py3 tritonserver --model-repository=/models
```
You now have a Triton server running with a simple model!


**Note:**
- The `--gpus all` flag ensures Triton can access all available GPUs.
- Use `nvidia-smi` to verify GPU availability before starting Triton.


## 4. Understanding the Rate Limiter

### What is a Rate Limiter?

A Rate Limiter is like a traffic controller for your models. It decides:
- How many requests each model can handle at once
- Which models get to run when resources are limited

### How Triton's Rate Limiter Works

Triton's Rate Limiter uses two main concepts:

1. Resources: These are things your models need, like GPU memory or CPU power.
2. Priorities: This decides which models are more important when resources are limited.

The Rate Limiter keeps track of available resources and assigns them to models based on their needs and priorities.


## 5. Configuring the Rate Limiter in Triton

Let's set up the Rate Limiter for our Triton server.

### Defining Resources

Resources are identified by a unique name and a count indicating the number of copies of the resource. Here's how to define resources in your model's config file (`config.pbtxt`):

```txt
instance_group [
  {
    count: 1
    kind: KIND_GPU
    gpus: [ 0, 1, 2 ]
    rate_limiter {
      resources [
        {
          name: "R1"
          count: 4
        },
        {
          name: "R2"
          global: True
          count: 2
        }
      ]
      priority: 2
    }
  }
]
```

In this example:

- The model instance requires 4 units of resource "R1" and 2 units of resource "R2".
- "R2" is specified as a global resource, meaning it's shared across all devices.
- This configuration creates 3 model instances, one on each GPU (0, 1, and 2).

The three instances won't contend for "R1" among themselves as it's local to each device. However, they will contend for "R2" because it's a global resource shared across the system.

**Note:** When configuring the Rate Limiter, it's important to understand what the resource numbers means that from the pool of all resources (global or per-device), this model requires 4 R1 resources to execute.

For example, if you have 10 R1 resources for a device in total, and you have 3 instances of model A, each requiring 4 R1 resources, only two instances of model A can be executed simultaneously.

### Setting Total Resources

By default, the available number of resource copies is the maximum across all model instances that list that resource. For example, if you have three model instances with the following resource requirements:

```
A: [R1: 4, R2: 4]
B: [R2: 5, R3: 10, R4: 5]
C: [R1: 1, R3: 7, R4: 2]
```

Triton will create the following resources:

```
R1: 4
R2: 5
R3: 10
R4: 5
```

You can override these defaults using the `--rate-limit-resource` option when starting Triton:
```shell
docker run --gpus all --rm -p 8000:8000 -p 8001:8001 -p 8002:8002 -v $(pwd)/models:/models nvcr.io/nvidia/tritonserver:21.09-py3 tritonserver --model-repository=/models --rate-limit-resource=R1:10 --rate-limit-resource=R2:5:0 --rate-limit-resource=R2:8:1 --rate-limit-resource=R3:2:global
```

This command sets up the following resources:
```
GLOBAL   => [R3: 2]
DEVICE 0 => [R1: 10, R2: 5]
DEVICE 1 => [R1: 10, R2: 8]
```

 #### `--rate-limit-resource` Flags:
- **`--rate-limit-resource=R1:10`**:
    - **R1**: The name of the resource.
    - **10**: The limit for this specific resource.
- **`--rate-limit-resource=R2:5:0`**:
    - **R2**: The resource name.
    - **5**: The limit for this specific resource.
    - **0**: The resource's instance index (e.g., the 0th instance of **R2**).
- **`--rate-limit-resource=R2:8:1`**:
    - Similar to the above, this specifies that the 1st instance of **R2** has a limit of 8.
- **`--rate-limit-resource=R3:2:global`**:
    - **R3**: Another resource.
    - **2**: Limit for this resource.
    - **global**: Instead of being instance-specific, this limit applies globally across all instances of this resource.

### Configuring Priorities:

Priority serves as a weighting value for scheduling across all model instances. 

An instance with priority 2 will be given half the number of scheduling chances as an instance with priority 1. You can set the priority in the rate limiter configuration:

```
rate_limiter {
  priority: 2
}
```

#### Example:

Imagine you are running two model instances on the same Triton server:

1. **Instance A** (latency-sensitive): Should handle requests faster, so you set `priority: 1`.
2. **Instance B** (less critical): Doesn't need as much scheduling attention, so you set `priority: 2`.

Here’s how you can configure this in Triton’s `config.pbtxt` file:

```protobuf
instance_group [
  {
    kind: KIND_GPU
    count: 1
    rate_limiter {
      priority: 1
    }
  },
  {
    kind: KIND_CPU
    count: 1
    rate_limiter {
      priority: 2
    }
  }
]

```

**Instance A (GPU)** with `priority: 1` will be scheduled twice as often as **Instance B (CPU)** with `priority: 2`, giving it more chances to process requests faster.

## 7. Practical Example: Configuring Rate Limiter for Multiple Models

Here is a practical example of configuring the Rate Limiter for multiple models.
### Step 1: Calculate Total GPU Memory

First, determine the total memory of your GPU. Let's say we have a GPU with 16GB (16,384 MB) of memory.

### Step 2: Calculate Base Memory Usage

Calculate the memory usage of each model when loaded into Triton (not serving any inference requests). For this example, let's assume:

- Model A: 500 MB
- Model B: 700 MB
- Model C: 600 MB

Total base memory usage: 500 + 700 + 600 = 1,800 MB

### Step 3: Calculate Available Memory

Subtract the base memory usage from the total GPU memory:
16,384 MB - 1,800 MB = 14,584 MB

This 14,584 MB is our available memory for inference (let's call this value m).

### Step 4: Determine Maximum Memory Usage

Find out the maximum memory usage of each model instance during inference. For this example, let's assume:

- Model A: 1,000 MB
- Model B: 1,500 MB
- Model C: 1,200 MB

### Step 5: Set Resource Values

Now, we'll set the R1 values for each model based on their maximum memory usage. We'll use 1 R1 unit to represent 100 MB of GPU memory:

```
instance_group [
  {
    count: 1
    kind: KIND_GPU
    rate_limiter {
      resources [
        {
          name: "R1"
          count: 10  # Model A: 1,000 MB / 100 MB per unit = 10 units
        }
      ]
    }
  }
]
```

Similarly, for Model B, count would be 15, and for Model C, it would be 12.

### Step 6: Set Total Per-Device Resource Value

Set the total per-device R1 value to m (our available memory for inference):

```shell
tritonserver --rate-limit-resource=R1:145 --model-repository=/path/to/model_repository
```

Here, 145 represents 14,584 MB / 100 MB per unit, rounded down.

### Complete Configuration Example

Here's how the complete configuration might look:

```
# Model A config.pbtxt
name: "model_a"
platform: "onnxruntime_onnx"
max_batch_size: 8
instance_group [
  {
    count: 1
    kind: KIND_GPU
    rate_limiter {
      resources [
        {
          name: "R1"
          count: 10
        }
      ]
    }
  }
]

# Model B config.pbtxt
name: "model_b"
platform: "onnxruntime_onnx"
max_batch_size: 8
instance_group [
  {
    count: 1
    kind: KIND_GPU
    rate_limiter {
      resources [
        {
          name: "R1"
          count: 15
        }
      ]
    }
  }
]

# Model C config.pbtxt
name: "model_c"
platform: "onnxruntime_onnx"
max_batch_size: 8
instance_group [
  {
    count: 1
    kind: KIND_GPU
    rate_limiter {
      resources [
        {
          name: "R1"
          count: 12
        }
      ]
    }
  }
]
```

With this configuration:
- Model A can use up to 1,000 MB of GPU memory during inference
- Model B can use up to 1,500 MB
- Model C can use up to 1,200 MB

 **Note:** The total usage across all models won't exceed 14,500 MB, leaving some buffer and accounting for the base memory usage


## 8. Best Practices and Tips

- Use the Model Analyzer (https://github.com/triton-inference-server/model_analyzer) to help with memory profiling steps.
- Monitor your system's performance and adjust resource allocations as needed.
- Remember to account for base memory usage of loaded models.
- Leave some buffer in your calculations to account for system overhead and variations in memory usage.
- For models with highly variable memory usage, consider setting the resource count to the maximum expected usage to ensure stability.

## 9. Advanced Rate Limiter Configurations

### Global vs. Per-Device Resources

- Per-device resources: By default, resources are per-device. Resource requirements for a model instance are enforced against the resources associated with the device where the model instance runs.
- Global resources: Instead of creating resource copies per-device, a global resource has a single copy shared across the entire system. You can specify a global resource in the model configuration or when starting Triton.

### Fine-tuning Priorities

In a resource-constrained system, there will be contention for resources among model instances. The priority setting helps determine which model instance to select for the next execution. Lower priority values indicate higher priority (more scheduling chances).


## 10. Conclusion

The Rate Limiter is a powerful tool in Triton that helps you:
- Manage resources efficiently
- Run multiple models safely
- Improve system stability

Triton is a tool that helps deploy and run machine learning models. It has a feature called the Rate Limiter, which helps manage resources better.

Rate Limiter in Triton: Managing Resource Allocation for MLOps

# Versioning Data in MLOps with DVC (Data Version Control)

## 1. Introduction

Have you ever worked on a machine learning project and found yourself drowning in data files? Maybe you've accidentally overwritten important data, or **you can't remember which dataset version gave you the best results**. 

![1.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/fcfa0847e776c5919fd4e9b517c6fdbac2be480c4b750e6a5c9abf2350e169fe.jpg)

Enter DVC, or Data Version Control. It's a tool that helps **manage and version your data**, much like how Git manages your code. 


## 2. Understanding Data Versioning in MLOps

Data versioning is about tracking changes in datasets over time. It's essential in machine learning operations (MLOps) for:

- Reproducing results
- Facilitating collaboration
- Tracking data lineage

![2.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/0926624668c6b98817b8d26dd22cc1303a54842b5128632b6814c3f9d125f699.jpg)

## 3. Introduction to DVC

DVC is an open-source tool that works alongside Git to handle large files, datasets, and machine learning models. Key features:

- Tracks changes in data without storing the data itself in Git
- Works with existing Git workflows
- Connects with cloud storage for large files

![3.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/2165695c1812fa81775292080190935a9a556a689bcc41eb7ad7710c27136d6b.jpg)

## 4. Setting up DVC

Let's set up a project with DVC:

1. Create a new directory and initialize Git:
  ```bash
mkdir my_ml_project
cd my_ml_project
git init
   ```

2. Install DVC:
  ```bash
pip install dvc==3.55.2
   ```

3. Initialize DVC:
  ```bash
dvc init
git commit -m "Initialize DVC"
   ```

Your project structure now looks like this:
```
📦my_ml_project  
 ┣ 📂.dvc  
 ┃ ┣ 📂tmp  
 ┃ ┃ ┗ 📜btime  
 ┃ ┣ 📜.gitignore  
 ┃ ┗ 📜config  
 ┣ 📂.git  
 ┗ 📜.dvcignore
```

![4.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/c067b95b990083839c495592b5cacbd00fcc6a2486b5b8a2c6adf09408c62078.jpg)

## 5. Basic DVC Workflow

Let's walk through a basic DVC workflow using an image classification project as an example.

1. Set up your project structure:
   ```
   my_ml_project/
   ├── data/
   │   └── images/
   ├── model.py
   ├── train.py
   └── .gitignore
   ```
   
   This structure keeps your project organized. The `data` directory will store your datasets, while separate Python files will handle model definition and training.

2. Add your dataset to DVC:
   ```bash
   dvc add data/images
   ```

   This command tells DVC to start tracking the `images` directory. DVC creates a small metadata file (`data/images.dvc`) that Git will version, while the actual image data is stored separately. This approach allows you to version your data without bloating your Git repository.

   The `data/images.dvc` file will look something like this:

   ```yaml
   outs:
   - md5: d41d8cd98f00b204e9800998ecf8427e  # Unique identifier for the current state of the data
     size: 0  # Total size of the tracked data
     path: images  # Location of the data relative to this .dvc file
   ```

3. Commit changes to Git:
   ```bash
   git add data/images.dvc .gitignore
   git commit -m "Add image dataset"
   ```

   By committing the DVC metadata file, you're essentially versioning a pointer to your data. This allows you to track which version of the data was used at this point in your project's history.

4. Create a simple training script (`train.py`):
   ```python
   import numpy as np
   from sklearn.model_selection import train_test_split
   from sklearn.ensemble import RandomForestClassifier
   from sklearn.metrics import accuracy_score
   import joblib
   import json

   def load_image_data(path):
       # Simulating image data loading
       # In a real scenario, you'd load and preprocess images here
       X = np.random.rand(1000, 224, 224, 3)  # 1000 images of 224x224 with 3 color channels
       y = np.random.randint(0, 10, 1000)  # 10 classes
       return X, y

   # Load and preprocess data
   X, y = load_image_data("data/images/")
   X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

   # Train model
   model = RandomForestClassifier(n_estimators=100)
   model.fit(X_train.reshape(X_train.shape[0], -1), y_train)

   # Evaluate model
   y_pred = model.predict(X_test.reshape(X_test.shape[0], -1))
   accuracy = accuracy_score(y_test, y_pred)
   print(f"Model accuracy: {accuracy}")

   # Save model
   joblib.dump(model, "model.joblib")

   # Save accuracy for DVC to track
   with open("accuracy.json", "w") as f:
       json.dump({"accuracy": accuracy}, f)
   ```

   This script defines our machine learning pipeline. It loads data, trains a model, evaluates it, and saves both the model and its accuracy. In a real scenario, you'd replace the data loading function with actual image processing code.

5. Create a DVC pipeline (`dvc.yaml`):
   ```yaml
   stages:
     train:
       cmd: python train.py  # Command to run for this stage
       deps:
         - data/images  # Input data dependency
         - train.py  # Code file dependency
       outs:
         - model.joblib  # Output model file
       metrics:
         - accuracy.json:  # Output metric file
             cache: false
   ```

   This YAML file defines a DVC pipeline. It specifies how to run your training script, what it depends on, and what it produces. DVC uses this information to determine when the pipeline needs to be re-run based on changes to the inputs, ensuring your outputs are always up-to-date.

6. Run the DVC pipeline:
   ```bash
   dvc repro
   ```

   This command runs the pipeline defined in `dvc.yaml`. DVC checks if any dependencies have changed and, if so, re-runs the affected stages. This ensures your outputs are always consistent with your inputs and code.

7. Commit changes:
   ```bash
   git add .
   git commit -m "Add training pipeline"
   ```

   This step versions your pipeline definition, training script, and the current state of your model (via the DVC metadata files). It captures the entire state of your ML project at this point, making it easy to reproduce or revert to this exact state later.


By following these steps, you've created a versioned, reproducible machine learning pipeline. You can now track changes to your data, code, and model performance over time, and easily revert to previous states if needed. 

## 6. Advanced DVC Features

DVC offers advanced features for more complex workflows:

### Data Pipelines

DVC pipelines allow you to define and run multi-stage data processing and model training workflows. This is particularly useful for more complex projects where you have multiple interdependent steps.

Here's an example `dvc.yaml` for a multi-stage pipeline:

```yaml
stages:
  preprocess:
    cmd: python preprocess.py  # Command to run the preprocessing script
    deps:
      - data/raw  # Input raw data
    outs:
      - data/processed  # Output processed data

  train:
    cmd: python train.py  # Command to run the training script
    deps:
      - data/processed  # Input processed data
    outs:
      - model.joblib  # Output trained model
    metrics:
      - accuracy.json:  # Output accuracy metric
          cache: false

  evaluate:
    cmd: python evaluate.py  # Command to run the evaluation script
    deps:
      - model.joblib  # Input trained model
      - data/test  # Input test data
    metrics:
      - evaluation.json:  # Output evaluation metrics
          cache: false
```

This pipeline defines three stages: preprocessing, training, and evaluation. Each stage has its own dependencies and outputs, allowing DVC to determine which stages need to be re-run when something changes.

### Metrics Tracking

DVC can track and compare metrics across experiments, which is crucial for monitoring your model's performance over time.

To view metrics:

```bash
dvc metrics show  # Displays the current metrics
```

To compare metrics with a previous version:

```bash
dvc metrics diff  # Shows how metrics have changed since the last commit
```

These commands help you quickly assess how changes in your data or code have affected your model's performance.

### Remote Storage

DVC can store your data in remote locations like S3, Google Cloud Storage, or SSH servers. This is essential for backing up your data and collaborating with team members.

To add a remote storage:

```bash
dvc remote add -d myremote s3://mybucket/dvcstore
# -d sets this remote as the default
# myremote is the name we're giving to this remote
# s3://mybucket/dvcstore is the S3 bucket path
```

To push data to the remote:

```bash
dvc push  # Uploads tracked data to the remote storage
```

To pull data from the remote:

```bash
dvc pull  # Downloads tracked data from the remote storage
```

Using remote storage allows you to keep your Git repository light while still having full version control over your large datasets and models.

### Experiment Management

DVC offers features to help manage and compare different experiments:

```bash
dvc exp run --set-param train.epochs=10  # Run an experiment with a specific parameter
dvc exp show  # Display a table of experiments and their metrics
dvc exp apply <exp-id>  # Apply the changes from a specific experiment
```

These commands allow you to easily try different parameters, compare results, and apply the most promising changes to your project.


## 7. Best Practices

1. Use meaningful stage names in your DVC pipeline.
2. Keep your raw data separate from processed data.
3. Use `.gitignore` to prevent large files from being added to Git.
4. Regularly push your data to remote storage for backup and collaboration.
5. Use metrics to track and compare experiment results.

## 8. Conclusion

DVC brings the power of version control to your data and ML pipelines. By using DVC, you can:

- Track changes in your datasets
- Create reproducible workflows
- Collaborate more effectively with your team
- Manage experiments and compare results easily

Enter DVC, or Data Version Control. It's a tool that helps **manage and version your data**, much like how Git manages your code. 

Versioning Data in MLOps with DVC (Data Version Control)

The big question is: What can we do with this dataset? How can we use it to evaluate company performance and deliver insights for improvement?

Identify and Measure Key Metrics to Analyze the Company’s Performance

# Rate Limiting in FastAPI: Essential Protection for ML API Endpoints

## What is Rate Limiting?

Rate limiting is a mechanism to control how many requests a client can make to your API within a specific timeframe. It acts as a gatekeeper, preventing abuse, ensuring fair resource distribution, and protecting your infrastructure from overload. For ML APIs, where inference tasks can be computationally expensive, rate limiting becomes critical to maintain stability and cost efficiency.

![intro.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/de44f91deb81887f9c2cf9c50bfd32a20295d549f6d89214cf144c5054dd1367.jpg)



In this guide, we'll explore rate limiting in FastAPI, a modern Python web framework ideal for building high-performance ML-powered APIs. We'll focus on practical implementations that are particularly relevant for machine learning engineers and data scientists.

## Installation Requirements

Before starting, ensure you have Python installed (I tested with **Python 3.9.12**), then install the required dependencies for each approach:

- **Base requirements (all examples):**
    
```bash
pip install fastapi==0.115.10 uvicorn==0.34.0
```
    
- **For Redis-based rate limiting (Approach 2):**
    
```bash
pip install fastapi-limiter==0.1.6 aioredis==2.0.1
```
    
- **For `slowapi` rate limiting (Approach 3):**
    
```bash
pip install slowapi==0.1.9
```
## Why Rate Limiting Matters for ML Engineers

Machine learning endpoints often require significant computational resources. Consider these scenarios where rate limiting becomes critical:

1. **Preventing resource exhaustion**: ML inference, especially for large models, can consume substantial CPU/GPU resources
2. **Protecting against abuse**: Without limits, a single client could monopolize your service
3. **Managing costs**: For cloud-deployed models, each prediction may have associated costs
4. **Ensuring fair service**: Distributing resources fairly across all clients
5. **Stabilizing performance**: Preventing traffic spikes that could impact model serving latency

Unlike traditional web applications, ML endpoints might perform complex operations involving large models loaded in memory. A sudden influx of requests could lead to out-of-memory errors, degraded performance, or even service outages.

## Implementation Options for Rate Limiting in FastAPI

FastAPI doesn't include built-in rate limiting, but several approaches are available. Let's explore them in order of complexity and features.

### Approach 1: In-Memory Rate Limiting (No External Dependencies)

This approach is ideal for development environments, single-server deployments, or when you're just starting to implement rate limiting. It's perfect when you need something quick to implement without adding infrastructure complexity.

![Approach1.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/d70377927409c8d4ca9d628d6c4eacbb99d66dd70f57ac6c38c99c637d73af1d.jpg)

For simpler applications or testing environments, a lightweight in-memory solution works well. This approach doesn't require external databases, making it perfect for development or smaller-scale deployments.


```python
from fastapi import FastAPI, Request, HTTPException, Depends
import time

app = FastAPI()

# In-memory storage for request counters
request_counters = {}

class RateLimiter:
    def __init__(self, requests_limit: int, time_window: int):
        self.requests_limit = requests_limit
        self.time_window = time_window

    def __call__(self, request: Request):
        client_ip = request.client.host
        route_path = request.url.path
        current_time = int(time.time())
        key = f"{client_ip}:{route_path}"

        if key not in request_counters:
            request_counters[key] = {"timestamp": current_time, "count": 1}
        else:
            if current_time - request_counters[key]["timestamp"] > self.time_window:
                request_counters[key] = {"timestamp": current_time, "count": 1}
            elif request_counters[key]["count"] >= self.requests_limit:
                raise HTTPException(status_code=429, detail="Too Many Requests")
            else:
                request_counters[key]["count"] += 1

        # Clean up expired entries
        for k in list(request_counters.keys()):
            if current_time - request_counters[k]["timestamp"] > self.time_window:
                request_counters.pop(k)

        return True

@app.post("/predict", dependencies=[Depends(RateLimiter(requests_limit=10, time_window=60))])
def predict_endpoint(data: dict):
    return {"prediction": 0.95, "confidence": 0.87}

@app.post("/batch-predict", dependencies=[Depends(RateLimiter(requests_limit=2, time_window=60))])
def batch_predict_endpoint(data: dict):
    return {"results": [{"prediction": 0.95}, {"prediction": 0.85}]}

@app.get("/model-info")
def model_info():
    return {"model_version": "v1.2.3", "framework": "PyTorch"}

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


This implementation tracks requests by client IP and endpoint path. For each request, it checks if the client has exceeded their limit within the defined time window.

**How to Run:**
1. Save as `app.py`.
2. Run:
```bash
uvicorn app:app --host 0.0.0.0 --port 8000
```

**How to Test:**
- With `curl`:
```bash
for i in {1..11}; do curl -X POST http://localhost:8000/predict -H "Content-Type: application/json" -d '{"data": [1,2,3]}'; echo ""; done
```
  - After 10 requests, you’ll get a "429 Too Many Requests" response.

**Pros:**
- No external dependencies.
- Easy to implement.
- Endpoint-specific limits.

**Cons:**
- Not suitable for distributed systems.
- Memory usage scales with clients.

### Approach 2: Redis-Based Rate Limiting with fastapi-limiter (With Docker)

For production-level applications, using a caching layer like Redis provides a scalable and persistent solution:

![Approach2.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/4d164981440796ae9edc5560fda2affb68ee5fecb4bc86b75dcc28222667ab1e.jpg)

Redis (an in-memory data store) offers a scalable way to enforce rate limits across distributed systems. Redis centralizes request tracking, ensuring consistency even if your FastAPI app runs on multiple servers.

**Pros**
- Supports distributed rate limiting across multiple app instances.
- Persists rate limit counters across restarts (with Redis persistence enabled).
- Scalable with minimal performance overhead.
- Well-maintained library with good documentation.

**Cons**
- Requires Redis infrastructure, adding setup complexity.
- Introduces an additional dependency to manage.

#### **Step-by-Step Setup with Docker:**
 
**Install Docker:**
Ensure Docker is installed on your machine. Download it from [Docker’s official site](https://www.docker.com/get-started) and follow the installation instructions for your operating system (Windows, macOS, or Linux).

**Pull the Redis Docker Image**
Use the official Redis image from Docker Hub:
```bash
docker pull redis
```
This ensures you’re using a stable, well-maintained version of Redis.

**Run Redis in a Docker Container**
Start a Redis container with the following command:
```bash
docker run -d --name redis-server -p 6379:6379 redis
```
- `-d`: Runs the container in detached mode (background).
- `--name redis-server`: Names the container for easy reference.
- `-p 6379:6379`: Maps port 6379 on your host to port 6379 in the container (Redis’s default port).

Redis will now be accessible at `localhost:6379` on your host machine.

**Verify Redis Is Running**
Confirm the container is active:
```bash
docker ps
```
You should see `redis-server` in the output.

Optionally, test the Redis instance by connecting to it:
```bash
docker exec -it redis-server redis-cli
```
In the Redis CLI, type `PING`. If Redis responds with `PONG`, it’s running correctly.

**Set Up and test the FastAPI Application**
The `fastapi-limiter` package integrates with Redis to enforce rate limits. Below is the complete FastAPI code:

```python
import aioredis
import uvicorn
from fastapi import Depends, FastAPI
from pydantic import BaseModel
from fastapi_limiter import FastAPILimiter
from fastapi_limiter.depends import RateLimiter

app = FastAPI()

class PredictionRequest(BaseModel):
    features: list
    model_version: str = "default"

@app.on_event("startup")
async def startup():
    # Connect to Redis - typically you'd get this from environment variables
    redis = await aioredis.from_url("redis://localhost:6379")
    await FastAPILimiter.init(redis)

# Standard prediction endpoint with rate limiting
@app.post("/predict", dependencies=[Depends(RateLimiter(times=20, seconds=60))])
def predict(request: PredictionRequest):
    # Your ML model inference code goes here
    result = 0.95  # Placeholder for actual prediction
    return {"prediction": result, "model_version": request.model_version}

# More restrictive rate limiting for resource-intensive operations
@app.post("/predict/batch", dependencies=[Depends(RateLimiter(times=5, seconds=300))])
def batch_predict(requests: list[PredictionRequest]):
    # Resource-intensive batch prediction
    return {"batch_predictions": [0.95, 0.85, 0.75]}

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

**Note:** 
- The FastAPI code for **Approach 2** uses `aioredis` to connect to Redis. By default, it connects to `redis://localhost:6379`, which matches the Docker setup.
- Here’s the complete FastAPI code with rate limiting:

**Run the FastAPI Application**
- Start the FastAPI server:
```bash
uvicorn app:app --host 0.0.0.0 --port 8000
```
- The app will connect to the Redis container during startup and use it for rate limiting.

**How to Test:**
   - Test `/predict`:
```bash
for i in {1..21}; do curl -X POST http://localhost:8000/predict -H "Content-Type: application/json" -d '{"features": [1,2,3]}'; echo ""; done
```
     - 429 error after 20 requests.
   - Test `/predict/batch`:
```bash
for i in {1..6}; do curl -X POST http://localhost:8000/predict/batch -H "Content-Type: application/json" -d '[{"features": [1,2,3]}]'; echo ""; done
```
     - 429 error after 5 requests.

### Approach 3: Rate Limiting with slowapi
When you have complex rate limiting requirements such as different limits for different user tiers, varying limits based on request complexity, or when you need sophisticated rate limiting algorithms beyond simple counters.

![Approach3.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/f2ada36a8b4acc85acc3f88b27eba08b44177de536a34bc8e9622a8aa42173f3.jpg)

For more complex rate limiting needs, `slowapi` provides additional features like different rate limit algorithms and customizable responses:

```python
from fastapi import FastAPI, Request, HTTPException
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from pydantic import BaseModel

# Initialize Limiter with a default key function
limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

class ModelInput(BaseModel):
    data: list
    parameters: dict = {}

# Default endpoint with IP-based rate limiting
@app.post("/predict")
@limiter.limit("10/minute")
async def predict(request: Request, input_data: ModelInput):
    return {"result": "prediction", "status": "success"}

# Custom key function for premium endpoint
def get_user_tier(request: Request):
    return request.headers.get("X-User-Tier", "free")

# Premium endpoint with tier-based rate limiting
@app.post("/premium/predict")
@limiter.limit("30/minute", key_func=get_user_tier)
async def premium_predict(request: Request, input_data: ModelInput):
    # Validate that the user is premium
    tier = get_user_tier(request)
    if tier != "premium":
        raise HTTPException(status_code=403, detail="Premium access required")
    return {"result": "premium prediction", "status": "success"}

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

**Notes:**
- `/premium/predict` requires `X-User-Tier: premium` to access; otherwise, it returns a 403 error.

**How to Run:**
1. Save as `app.py`.
2. Run:
```bash
uvicorn app:app --host 0.0.0.0 --port 8000
```

**How to Test:**
- Free tier on `/predict`:
```bash
  for i in {1..11}; do curl -X POST http://localhost:8000/predict -H "Content-Type: application/json" -d '{"data": [1,2,3]}'; echo ""; done
```
  - 429 error after 10 requests.
- Premium tier on `/premium/predict`:
```bash
  for i in {1..31}; do curl -X POST http://localhost:8000/premium/predict -H "Content-Type: application/json" -H "X-User-Tier: premium" -d '{"data": [1,2,3]}'; echo ""; done
```
  - 429 error after 30 requests.
- Non-premium user on `/premium/predict`:
```bash
  curl -X POST http://localhost:8000/premium/predict -H "Content-Type: application/json" -H "X-User-Tier: free" -d '{"data": [1,2,3]}'
```
  - Expected: `{"detail": "Premium access required"}` with a 403 status code.


**Pros:**

- Feature-rich with multiple rate limiting strategies
- Can define limits based on custom keys 
- Supports Redis, Memcached, or in-memory storage
- Good integration with FastAPI

**Cons:**

- Slightly more complex setup
- May have more overhead than simpler solutions
## ML-Specific Considerations for Rate Limiting

When implementing rate limiting for ML APIs, consider these specialized factors:

![Monitoring.jpg](https://fsds-content.s3.ap-southeast-1.amazonaws.com/uploads/8ce7519071bd3aeec49c7c140a65f5cd05f2b7dfaf5e4263a9c69c7366c0a9a7.jpg)

### 1. Resource-Based Differentiation

Not all ML endpoints consume equal resources. Consider implementing different rate limits based on:

- **Computational complexity**: More complex models might need stricter limits
- **Batch sizes**: Large batch requests consume more resources
- **Model size**: Larger models with more parameters will require more memory

Here's how you might implement this:

```python
# Light model with generous limits
@app.post("/models/lightweight/predict", 
          dependencies=[Depends(RateLimiter(requests_limit=100, time_window=60))])
async def lightweight_predict(data: dict):
    # Inference with small, fast model
    return {"prediction": 0.95}

# Heavy model with stricter limits
@app.post("/models/heavy/predict", 
          dependencies=[Depends(RateLimiter(requests_limit=10, time_window=60))])
async def heavy_predict(data: dict):
    # Inference with large, resource-intensive model
    return {"prediction": 0.92}

# Very restrictive limits for batch processing
@app.post("/models/batch-process", 
          dependencies=[Depends(RateLimiter(requests_limit=2, time_window=300))])
async def batch_process(data: list):
    # Batch processing logic
    return {"results": [...]}
```

### 2. User Tiers for ML Services

For commercial ML APIs, consider implementing tiered access with different rate limits:

```python
async def get_user_tier(request: Request):
    # This could retrieve tier info from a database, token, or header
    api_key = request.headers.get("X-API-Key")
    # Lookup tier based on API key (simplified example)
    if api_key == "premium-user-key":
        return "premium"
    return "free"

class TieredRateLimiter(RateLimiter):
    async def __call__(self, request: Request):
        # Override the base rate limits based on user tier
        tier = await get_user_tier(request)
        if tier == "premium":
            self.requests_limit = self.requests_limit * 5  # 5x higher limit for premium
        return await super().__call__(request)

# Use the tiered rate limiter
@app.post("/predict", dependencies=[Depends(TieredRateLimiter(requests_limit=20, time_window=60))])
async def predict(data: dict):
    # Standard limits for free tier (20/minute)
    # Premium users get 100/minute through the TieredRateLimiter
    return {"prediction": 0.95}
```

### 3. Managing Model Warm-up Periods

ML models often need a "warm-up" period when first loaded. You might implement a dynamic rate limiter that adjusts limits during warm-up:

```python
class ModelStatus:
    def __init__(self):
        self.is_warming_up = True
        # When the model starts, it's in warm-up mode

model_status = ModelStatus()

# After model initialization completes
@app.on_event("startup")
async def startup_event():
    # Model loading and initialization
    # ...
    # After warm-up completes
    model_status.is_warming_up = False

class AdaptiveRateLimiter(RateLimiter):
    async def __call__(self, request: Request):
        # Reduce limits during warm-up
        if model_status.is_warming_up:
            self.requests_limit = max(1, self.requests_limit // 5)  # Much lower during warm-up
        return await super().__call__(request)

@app.post("/predict", dependencies=[Depends(AdaptiveRateLimiter(requests_limit=50, time_window=60))])
async def predict(data: dict):
    return {"prediction": 0.95}
```

## Monitoring and Debugging Rate Limits

You also need visibility into how your limits are affecting users and system performance. This monitoring approach provides the feedback loop necessary to tune your rate limits over time.

Here's a simple approach:

```python
from fastapi import FastAPI, Request, HTTPException, Depends
import time
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("rate_limiter")

# Initialize counters for metrics
rate_limit_metrics = {
    "total_requests": 0,
    "limited_requests": 0,
    "endpoints": {}
}

class MonitoredRateLimiter(RateLimiter):
    async def __call__(self, request: Request):
        endpoint = request.url.path
        client_ip = request.client.host
        
        # Update metrics
        rate_limit_metrics["total_requests"] += 1
        if endpoint not in rate_limit_metrics["endpoints"]:
            rate_limit_metrics["endpoints"][endpoint] = {
                "total": 0, "limited": 0
            }
        rate_limit_metrics["endpoints"][endpoint]["total"] += 1
        
        try:
            result = await super().__call__(request)
            return result
        except HTTPException as e:
            if e.status_code == 429:  # Rate limit exceeded
                rate_limit_metrics["limited_requests"] += 1
                rate_limit_metrics["endpoints"][endpoint]["limited"] += 1
                logger.warning(f"Rate limit exceeded for {client_ip} on {endpoint}")
            raise

@app.get("/metrics")
async def get_metrics():
    return rate_limit_metrics
```

This abstract approach provides basic metrics and logging for rate limiting events, which can be valuable when tuning your limits or diagnosing issues.

## Conclusion

Rate limiting is an essential component of production ML APIs, protecting your resources from overuse and ensuring fair service distribution. FastAPI offers several implementation options, from simple in-memory solutions to more robust distributed approaches.

For ML engineers, the right approach depends on your deployment scale, infrastructure, and specific requirements. The in-memory approach works well for smaller deployments, while Redis-based solutions offer more robustness for production-grade systems.

Remember that effective rate limiting is about finding the right balance—too restrictive, and you limit legitimate use; too permissive, and you risk resource exhaustion or service degradation.

## References

1. [FastAPI Official Documentation](https://fastapi.tiangolo.com/)
2. [Redis Rate Limiting Pattern](https://redis.io/commands/incr#pattern-rate-limiter)
3. [Best Practices for ML API Design](https://cloud.google.com/architecture/ml-on-gcp-best-practices)
4. [Rate Limiting Algorithms Explained](https://konghq.com/blog/how-to-design-a-scalable-rate-limiting-algorithm/)

Rate limiting is a mechanism to control how many requests a client can make to your API within a specific timeframe. It acts as a gatekeeper, preventing abuse, ensuring fair resource distribution, and protecting your infrastructure from overload. For ML APIs, where inference tasks can be computationally expensive, rate limiting becomes critical to maintain stability and cost efficiency.

FSDS Blog