F5 AI Gateway Python SDK Quickstart¶

Note

Contact your F5 representative for access to evaluate the SDK.

About processors¶

There are three types of alterations which a processor may make on a request.

Annotation
Rejection
Modification

A processor can only make these types of alterations when a specific parameter is provided.

If there is an attempt to make a modification without the required parameter, the SDK will drop the modification, and it will log a warning. For example:

WARNING:root:CatClassifierProcessor tried to annotate request with tags when parameters.annotate was set to false, tags will be dropped

Annotation¶

A processor can add tags to a request which can be used by AI Gateway to route the request. This is controlled by the parameter annotate, which defaults to true.

Rejection¶

A processor can reject a request which causes a rejection message to be returned to the client. This is controlled by the parameter reject, which defaults to false.

Modification¶

A processor can modify the prompt that is sent by a client or modify the response from an upstream model. This is controlled by the parameter modify, which defaults to false.

Before you begin¶

To follow the steps in this guide you will need:

A Python 3.11+ environment with pip or an alternative installed
An ASGI server

In this example we will use uvicorn as our ASGI server. Install it first if you don’t have it already or adjust the commands to use the server of your choice.

Note

You may want to use a Python virtual environment to manage your dependencies. Below is an example of setting up and activating a virtual environment.

python3 -m venv .venv
source .venv/bin/activate

pip install uvicorn

Install the SDK¶

The F5 AI Gateway Python SDK is designed to handle requests and responses between the F5 AI Gateway and your application, allowing you to focus on developing your processor logic.

pip install f5-ai-gateway-sdk

Create a basic processor which annotates a request¶

In this tutorial we create a simple processor that inspects the prompt and response for any mentions of a ‘cat’ and tags it accordingly. Tags can be used in the AI Gateway configuration to route to a specific upstream LLM or run a particular set of processors.

Create a file called cat_classifier.py with the following contents.

from f5_ai_gateway_sdk.parameters import Parameters
from f5_ai_gateway_sdk.processor import Processor
from f5_ai_gateway_sdk.processor_routes import ProcessorRoutes
from f5_ai_gateway_sdk.request_input import Message, MessageRole
from f5_ai_gateway_sdk.result import Result
from f5_ai_gateway_sdk.signature import BOTH_SIGNATURE
from f5_ai_gateway_sdk.tags import Tags
from f5_ai_gateway_sdk.type_hints import Metadata
from starlette.applications import Starlette


class CatClassifierProcessor(Processor):
    """
    A simple processor which adds a tag when a cat is found in the prompt or response
    """

    def __init__(self):
        super().__init__(
            name="cat-classifier",
            version="v1",
            namespace="tutorial",
            signature = BOTH_SIGNATURE
        )

    def process(self, prompt, response, metadata, parameters, request):
        my_tags = Tags()
        cat_found = False

        if response:
            cat_found = any(
                "cat" in choice.message.content for choice in response.choices
            )
        elif prompt:
            cat_found = any("cat" in message.content for message in prompt.messages)

        result = Metadata({"cat_found": cat_found})
        if cat_found:
            my_tags.add_tag("animals-found", "cat")

        return Result(
            processor_result=result, tags=my_tags if parameters.annotate else None
        )


app = Starlette(
    routes=ProcessorRoutes([CatClassifierProcessor()]),
)

Note

This example contains all of the imports that will be needed for this tutorial so your editor may warn about unused imports.

Run the processor locally¶

Run the processor locally using uvicorn; this will start a server on port 9999. Ensure that you run this command from the directory where your cat_classifier.py file is located.

python -m uvicorn cat_classifier:app --host 127.0.0.1 --port 9999 --reload

The --reload flag will automatically restart the server when you make changes to your code.

We can check if the server is running by sending a curl request to the signature endpoint of the processor.

Request

curl -i http://localhost:9999/api/v1/signature/tutorial/cat-classifier

Response

{
    "fields": [
        {
            "type": "response.choices",
            "required": false
        },
        {
            "type": "input.messages",
            "required": false
        }
    ],
    "parameters": {
        "description": "Default empty parameters class for processors that do not require parameters.",
        "properties": {
            "annotate": {
                "default": true,
                "description": "Whether the processor can annotate the input with tags.",
                "title": "Annotate",
                "type": "boolean"
            },
            "modify": {
                "default": false,
                "description": "Whether the processor can modify the input.",
                "title": "Modify",
                "type": "boolean"
            },
            "reject": {
                "default": false,
                "description": "Whether the processor can reject requests.",
                "title": "Reject",
                "type": "boolean"
            }
        },
        "title": "Default Parameters",
        "type": "object"
    }
}

In the above endpoint, tutorial is the namespace of your processor, and cat-classifier is the name of your processor.

The response should be similar to the following example and contains information that is used by AI Gateway to send correctly formatted requests to the processor:

Request

curl -i -X POST http://localhost:9999/api/v1/execute/tutorial/cat-classifier \
    -H "Content-Type: multipart/form-data" \
    --form 'input.parameters={};type=application/json' \
    --form 'metadata={};type=application/json' \
    --form 'input.messages={
    "messages": [
        {
        "content": "Curious cats gracefully roam through cozy homes, chasing shadows and basking in warm sunlit windows."
        }
    ]
    };type=application/json'

Response

HTTP/1.1 200 OK
date: Fri, 22 Nov 2024 16:33:10 GMT
server: uvicorn
content-type: multipart/form-data;charset=utf-8;boundary="3zWBuRUjyupW5lUbCEITz3riZBm2dPztxregsT4fMZJBHjmoEDYUq5fqAXGCKHr6"
Transfer-Encoding: chunked

--3zWBuRUjyupW5lUbCEITz3riZBm2dPztxregsT4fMZJBHjmoEDYUq5fqAXGCKHr6
Content-Disposition: form-data; name="metadata"
Content-Type: application/json

{"processor_id": "example:cat-classifier", "processor_version": "v1", "processor_result": {"cat_found": true}, "tags": {"animals-found": ["cat"]}}
--3zWBuRUjyupW5lUbCEITz3riZBm2dPztxregsT4fMZJBHjmoEDYUq5fqAXGCKHr6--

Use the processor with AI Gateway¶

To configure this processor in AI Gateway add the following to your aigw.yml.

processors:
  - name: cat-classifier
    type: external
    config:
      endpoint: "localhost:9999"
      namespace: tutorial
      version: 1

See the configure processors page for more details.

Configure the processor to reject a request¶

Processors can be configured on a per-request basis by setting the params object in either the processors or steps sections of the AI Gateway configuration file.

Let’s use our default parameters to reject any requests which don’t contain cats. Update the end of the process function in cat_classifier.py to add the reject variable and reference it in the return statement:

        reject = parameters.reject and not cat_found

        return Result(
            processor_result=result,
            tags=my_tags if parameters.annotate else None,
            rejected=reject,
        )

We can now make a request to the processor with the reject parameter set to true:

Request

curl -i -X POST http://localhost:9999/api/v1/execute/tutorial/cat-classifier \
    -H "Content-Type: multipart/form-data" \
    --form 'input.parameters={"reject":true};type=application/json' \
    --form 'metadata={};type=application/json' \
    --form 'input.messages={
    "messages": [
        {
        "content": "Curious dogs gracefully roam through cozy homes, chasing shadows and basking in warm sunlit windows."
        }
    ]
    };type=application/json'

Response

HTTP/1.1 422 Unprocessable Content
date: Fri, 22 Nov 2024 16:32:45 GMT
server: uvicorn
content-type: multipart/form-data;charset=utf-8;boundary="dkQBsMcvc3JXNWKnlTdRw8dOfsD3KWjrZnJvgZ6z0nFS8hNBCnSmiJ2eRldNXTnR"
Transfer-Encoding: chunked

--dkQBsMcvc3JXNWKnlTdRw8dOfsD3KWjrZnJvgZ6z0nFS8hNBCnSmiJ2eRldNXTnR
Content-Disposition: form-data; name="metadata"
Content-Type: application/json

{"processor_id": "example:cat-classifier", "processor_version": "v1", "processor_result": {"cat_found": false}}
--dkQBsMcvc3JXNWKnlTdRw8dOfsD3KWjrZnJvgZ6z0nFS8hNBCnSmiJ2eRldNXTnR--

The 422 status code is used to indicate that the request was rejected by the processor.

Configure the processor to modify a request¶

Processors also have the ability to make modifications to the prompt or response before it is routed to the next stage. Make the following changes to add a system message to the request if the processor is used in an input stage.

Additional parameters can be added to the processor by defining a class that extends f5_ai_gateway_sdk.parameters.Parameters. Add the following after the imports section at the top of the cat_classifier.py file:

class CatClassifierParameters(Parameters):
    enforce_cat_message: str = "Remember to talk about cats in your response"

We then need to update the __init__ function for the processor to specify this class.

    def __init__(self):
        super().__init__(
            name="cat-classifier",
            version="v1",
            namespace="tutorial",
            signature=BOTH_SIGNATURE,
            parameters_class=CatClassifierParameters
        )

If we make a request to the signature endpoint, we can see the newly added parameter.

Request

curl -i http://localhost:9999/api/v1/signature/tutorial/cat-classifier

Response

{
    "fields": [
        {
            "type": "input.messages",
            "required": false
        },
        {
            "type": "response.choices",
            "required": false
        }
    ],
    "parameters": {
        "properties": {
            "annotate": {
                "default": true,
                "description": "Whether the processor can annotate the input with tags.",
                "title": "Annotate",
                "type": "boolean"
            },
            "modify": {
                "default": false,
                "description": "Whether the processor can modify the input.",
                "title": "Modify",
                "type": "boolean"
            },
            "reject": {
                "default": false,
                "description": "Whether the processor can reject requests.",
                "title": "Reject",
                "type": "boolean"
            },
            "enforce_cat_message": {
                "default": "Remember to talk about cats in your response",
                "title": "Enforce Cat Message",
                "type": "string"
            }
        },
        "title": "CatClassifierParameters",
        "type": "object"
    }
}

Note

Due to the CatClassifierParameters used in this example being a subclass of Parameters, it automatically has access to the common parameters of annotate, reject, and modify.

Add the following to the process function before the return statement

        if (
            not reject  # don't attempt to add a message if we are already rejecting
            and parameters.modify
            and prompt
        ):
            prompt.messages.append(
                Message(
                    content=parameters.enforce_cat_message,
                    role=MessageRole.SYSTEM,
                )
            )
            return Result(
                processor_result=result,
                tags=my_tags if parameters.annotate else None,
                rejected=reject,
                modified_prompt=prompt,
            )

Verify that the system message has been added to the request.

Request

curl -i -X POST http://localhost:9999/api/v1/execute/tutorial/cat-classifier \
    -H "Content-Type: multipart/form-data" \
    --form 'input.parameters={"modify":true,"enforce_cat_message":"Remember to talk about how cute cats are in your response"};type=application/json' \
    --form 'metadata={};type=application/json' \
    --form 'input.messages={
    "messages": [
        {
        "content": "Curious dogs gracefully roam through cozy homes, chasing shadows and basking in warm sunlit windows."
        }
    ]
    };type=application/json'

Response

HTTP/1.1 200 OK
date: Thu, 12 Dec 2024 16:03:26 GMT
server: uvicorn
content-type: multipart/form-data;charset=utf-8;boundary="iBDsmhfPKiowYNQTb9bbfPT76c26q3y8Zj9o7qFHrgDIsIlLqDQi5E5bQaqjIkLy"
Transfer-Encoding: chunked

--iBDsmhfPKiowYNQTb9bbfPT76c26q3y8Zj9o7qFHrgDIsIlLqDQi5E5bQaqjIkLy
Content-Disposition: form-data; name="input.messages"
Content-Type: text/plain;charset=utf-8

{"messages":[{"content":"Curious dogs gracefully roam through cozy homes, chasing shadows and basking in warm sunlit windows.","role":"user"},{"content":"Remember to talk about how cute cats are in your response","role":"system"}]}
--iBDsmhfPKiowYNQTb9bbfPT76c26q3y8Zj9o7qFHrgDIsIlLqDQi5E5bQaqjIkLy
Content-Disposition: form-data; name="metadata"
Content-Type: application/json

{"processor_id": "tutorial:cat-classifier", "processor_version": "v1", "processor_result": {"cat_found": false}}
--iBDsmhfPKiowYNQTb9bbfPT76c26q3y8Zj9o7qFHrgDIsIlLqDQi5E5bQaqjIkLy--

Completed tutorial processor¶

from f5_ai_gateway_sdk.parameters import Parameters
from f5_ai_gateway_sdk.processor import Processor
from f5_ai_gateway_sdk.processor_routes import ProcessorRoutes
from f5_ai_gateway_sdk.request_input import Message, MessageRole
from f5_ai_gateway_sdk.result import Result
from f5_ai_gateway_sdk.signature import BOTH_SIGNATURE
from f5_ai_gateway_sdk.tags import Tags
from f5_ai_gateway_sdk.type_hints import Metadata
from starlette.applications import Starlette


class CatClassifierParameters(Parameters):
    enforce_cat_message: str = "Remember to talk about cats in your response"


class CatClassifierProcessor(Processor):
    """
    A simple processor which adds a tag when a cat is found in the prompt or response
    """

    def __init__(self):
        super().__init__(
            name="cat-classifier",
            version="v1",
            namespace="tutorial",
            signature=BOTH_SIGNATURE,
            parameters_class=CatClassifierParameters,
        )

    def process(self, prompt, response, metadata, parameters, request):
        my_tags = Tags()
        cat_found = False

        if response:
            cat_found = any(
                "cat" in choice.message.content for choice in response.choices
            )
        elif prompt:
            cat_found = any("cat" in message.content for message in prompt.messages)

        result = Metadata({"cat_found": cat_found})
        if cat_found:
            my_tags.add_tag("animals-found", "cat")

        reject = parameters.reject and not cat_found

        if (
            not reject  # don't attempt to add a message if we are already rejecting
            and parameters.modify
            and prompt
        ):
            prompt.messages.append(
                Message(
                    content=parameters.enforce_cat_message,
                    role=MessageRole.SYSTEM,
                )
            )
            return Result(
                processor_result=result,
                tags=my_tags if parameters.annotate else None,
                rejected=reject,
                modified_prompt=prompt,
            )

        return Result(
            processor_result=result,
            tags=my_tags if parameters.annotate else None,
            rejected=reject,
        )


app = Starlette(
    routes=ProcessorRoutes([CatClassifierProcessor()]),
)

Previous Next