Simplifying AI Development with PydanticAI and Serverless Functions on Civo

This tutorial will show you how to build an AI-powered medical symptom checker using the Standard Treatment Guidelines – Common Diseases PDF, published by the Rajasthan Medical Services Corporation (RMSC) as the dataset. With PydanticAI ensuring that your AI model outputs are structured and reliable and with serverless functions on Civo (powered by OpenFaaS) taking care of the heavy lifting on the infrastructure side, you can focus on solving real-world problems without the usual operational headaches.

Introduction

What if you could build AI-driven applications without worrying about complex infrastructure or cumbersome model handling? Enter PydanticAI and serverless computing—a combination that simplifies developing, deploying, and scaling AI applications. PydanticAI is a powerful tool that provides type safety and structured outputs for AI models, ensuring that your application receives data in a predictable format. At the same time, serverless functions abstract away the intricacies of managing servers and infrastructure.

In this tutorial, we’ll deploy a serverless AI-powered medical symptom checker on Civo using OpenFaaS as our orchestration platform. We’ll utilize a dataset from the Common Diseases PDF—which includes common diseases, their symptoms, and recommended treatments—to demonstrate how you can process user input (medical symptoms) and generate structured, reliable responses. By the end, you’ll see how the combination of PydanticAI and serverless computing not only simplifies AI development but also ensures scalability and efficiency.

What is PydanticAI?

PydanticAI is an extension of the popular Pydantic library, designed specifically to integrate AI models into applications with a structured approach. It leverages Python’s type hints to enforce strict data validation and ensure that the outputs from your AI models are predictable and consistent. This structure is crucial when building AI applications where the integrity and reliability of data can impact user experience and overall system performance.

What is PydanticAI

Source

Key benefits of PydanticAI

Type Safety: By using Python’s type annotations, PydanticAI ensures that inputs and outputs adhere to specified formats. This reduces bugs and unexpected behavior.

Structured Outputs: This forces your AI models to return data in a defined schema. For instance, you can structure responses for a medical symptom checker as follows:

{
  "disease": "Flu",
  "recommended_action": "Stay hydrated and rest. If symptoms worsen, see a doctor."
}

Model-Agnostic Compatibility: Whether you’re using a deep learning model or a simple statistical method, PydanticAI can work with various types of AI models, making it highly versatile.

Easy Deployment: It simplifies the integration of AI models into your serverless functions, ensuring that your application remains robust and maintainable as it scales.

Why are we using PydanticAI in this tutorial?

PydanticAI will help us structure the responses from our AI model, making it easy to interpret and reliable for end users. This ensures that every response is consistent, making it simpler to debug, monitor, and scale your application.

PydanticAI stands out among other integration tools because it combines the flexibility of modern AI with the reliability of structured programming, streamlining the entire development process.

Setting up your environment on Civo

Before we can deploy our AI function, we need to set up the environment on Civo—a cloud platform that makes it easy to spin up Kubernetes clusters. Follow these steps to get started:

Step 1: Create a Civo account

Visit Civo’s website and sign up for an account. Once this is done, verify your email and log in to your new account.

Step 2: Install the Civo CLI

The Civo CLI allows you to manage cloud resources directly from your terminal. Follow the Civo CLI installation guide for your operating system.

Once installed, run the following command to log in:

civo apikey add myapikey <your-api-key>

Step 3: Create a Kubernetes cluster on Civo

Since OpenFaaS runs on Kubernetes, you need a cluster to deploy your serverless functions. Create a new cluster with the following command:

civo kubernetes create my-cluster --nodes 3 --size g4s.kube.medium

Wait for the cluster to be ready, and then configure your kubectl to point to your new cluster:

civo kubernetes config my-cluster > kubeconfig.yaml
export KUBECONFIG=kubeconfig.yaml

Now, your Civo environment is set up and ready to deploy serverless functions.

Deploying OpenFaaS for serverless AI

With your Kubernetes cluster running on Civo, the next step is to deploy OpenFaaS—a framework for building serverless functions that can scale with demand.

Step 1: Install required tools

kubectl: Ensure you have kubectl installed. Follow the official guide.

faas-cli: On Windows, you can install faas-cli via Chocolatey:

choco install faas-cli

Alternatively, download the Windows binary from the faas-cli GitHub releases page and add it to your system PATH.

Step 2: Deploy OpenFaaS using Helm

The previous method of applying static YAML files from the faas-netes repository has evolved. The current recommended approach is to use Helm charts:

Add the OpenFaaS Helm Repository:

helm repo add openfaas https://openfaas.github.io/faas-netes/
helm repo update

Create the openfaas-fn namespace: This is where OpenFaaS functions will be deployed:

kubectl create namespace openfaas-fn

Install OpenFaaS: Deploy the control plane into the openfaas namespace and specify that functions run in the openfaas-fn namespace. In PowerShell, you can use the backtick (`) for line continuations:

helm install openfaas openfaas/openfaas `
--namespace openfaas `
--create-namespace `
--set functionNamespace=openfaas-fn `
--set basic_auth=false

Note: If you require authentication, adjust the basic_auth setting accordingly.

Verify the Deployment: Check that the OpenFaaS control plane pods are up and running:

kubectl get pods -n openfaas

And later, when you deploy functions, they will appear in the openfaas-fn namespace:

kubectl get pods -n openfaas-fn

Step 3: Access the OpenFaaS Dashboard

Port-Forward the Gateway Service: Once the gateway pod is running and healthy, forward its port to your local machine:

kubectl port-forward -n openfaas svc/gateway 8080:8080

Note: On Windows, ensure you’re running this in PowerShell or Git Bash with the appropriate environment settings.

Open the Dashboard: In your browser, navigate to http://localhost:8080 to access the OpenFaaS dashboard where you can manage functions, view logs, and monitor scaling.

Building and deploying a PydanticAI function

Now that our environment is ready, it’s time to build our serverless AI function that uses the COMMON DISEASES dataset.

Our dataset contains information on around 20 common diseases, along with their symptoms and recommended treatments. This dataset will serve as the knowledge base for our medical symptom checker. When a user inputs their symptoms, the AI function will reference this dataset to suggest possible diseases and actions.

Step 1: Create the AI Function

Using OpenFaaS, we will build a function that:

Reads the dataset at runtime.
Processes the input symptoms.
Returns a structured response with a disease and recommended action.

Here’s a simplified version of what the function code might look like (in Python):

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import PyPDF2
import re
import os


app = FastAPI()


# Define response model
class DiagnosisResponse(BaseModel):
    disease: str
    recommended_action: str


class SymptomInput(BaseModel):
    symptoms: str


# Function to extract text from PDF
def extract_text_from_pdf(pdf_path):
    with open(pdf_path, "rb") as file:
        reader = PyPDF2.PdfReader(file)
        text = "\n".join([page.extract_text() for page in reader.pages if page.extract_text()])
    return text

The code starts by importing necessary libraries, setting up the FastAPI app, and defining data models with Pydantic. It also provides a utility function to extract text from a PDF file, laying the foundation for further data processing.

# Function to parse diseases and symptoms from PDF text
def parse_disease_data(text):
    diseases = {}
    current_disease = None


    for line in text.split("\n"):
        line = line.strip()
        # Identify disease names (assuming they are in uppercase)
        if re.match(r"^[A-Z ]+$", line) and len(line) > 5:
            current_disease = line
            diseases[current_disease] = {"symptoms": [], "treatment": ""}
        elif current_disease:
            if "Diagnosis" in line:
                continue
            if "Treatment" in line:
                diseases[current_disease]["treatment"] = line.split("Treatment")[-1].strip()
            else:
                diseases[current_disease]["symptoms"].append(line)


    return diseases


# Load and process PDF at startup
pdf_path = os.path.join(os.path.dirname(__file__), "Chapter-01.indd.pdf")
pdf_text = extract_text_from_pdf(pdf_path)
disease_data = parse_disease_data(pdf_text)

Next, it processes the extracted PDF text by parsing it to identify diseases along with their symptoms and treatment details. This part reads through the text line-by-line, detects disease names (assumed to be in uppercase), and organizes the corresponding details, loading the data when the application starts.

# Function to match symptoms
def check_symptoms(user_symptoms):
    matched_diseases = []


    for disease, data in disease_data.items():
        if any(symptom.lower() in " ".join(data["symptoms"]).lower() for symptom in user_symptoms.split(",")):
            matched_diseases.append({"disease": disease, "recommended_action": data["treatment"]})


    return matched_diseases[:3] if matched_diseases else [{"disease": "Unknown", "recommended_action": "Consult a healthcare professional."}]


@app.post("/", response_model=list[DiagnosisResponse])
def diagnose(input_data: SymptomInput):
    if not input_data.symptoms:
        raise HTTPException(status_code=400, detail="No symptoms provided")
   
    results = check_symptoms(input_data.symptoms)
    return results


@app.get("/_/health")
def health_check():
    return {"status": "ok"}

# Function to match symptoms
def check_symptoms(user_symptoms):
    matched_diseases = []


    for disease, data in disease_data.items():
        if any(symptom.lower() in " ".join(data["symptoms"]).lower() for symptom in user_symptoms.split(",")):
            matched_diseases.append({"disease": disease, "recommended_action": data["treatment"]})


    return matched_diseases[:3] if matched_diseases else [{"disease": "Unknown", "recommended_action": "Consult a healthcare professional."}]


@app.post("/", response_model=list[DiagnosisResponse])
def diagnose(input_data: SymptomInput):
    if not input_data.symptoms:
        raise HTTPException(status_code=400, detail="No symptoms provided")
   
    results = check_symptoms(input_data.symptoms)
    return results


@app.get("/_/health")
def health_check():
    return {"status": "ok"}

Finally, the code implements logic to match user-provided symptoms with the stored disease data and exposes API endpoints. One endpoint accepts symptoms and returns up to three matched diseases with recommended actions, while another serves as a simple health check.

Step 2: Package the Function in a Docker Container

Next, package the function inside a Docker container. Create a Dockerfile like the following:


FROM tiangolo/uvicorn-gunicorn-fastapi:python3.9


# Set working directory
WORKDIR /app


# Copy function code and dataset into the container
COPY . /app


# Install dependencies
RUN pip install --no-cache-dir fastapi uvicorn pydantic PyPDF2


# Expose port
EXPOSE 8080


# Start the FastAPI server
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8080"]

Build the Docker image with:

docker build -t my-diagnosis-function .

Tag the Image for Docker Hub. OpenFaaS Community Edition requires a public image on Docker Hub:

docker tag my-diagnosis-function YOUR_DOCKER_USERNAME/my-diagnosis-function:latest

Push the Image to Docker Hub: If you’re not already logged in, log in to Docker Hub:

docker login

Then push your tagged image:

docker push YOUR_DOCKER_USERNAME/my-diagnosis-function:latest

This step makes your image available publicly so that the OpenFaaS gateway can pull it during deployment.

Step 3: Deploy the Function on OpenFaaS

Use the faas-cli to deploy your function. Create a stack.yml file for your function:

provider:
  name: openfaas
  gateway: http://localhost:8080

functions:
  diagnosis:
    lang: python3-flask
    handler: ./diagnosis
    image: YOUR_DOCKER_USERNAME/my-diagnosis-function:latest

Ensure that you replace YOURDOCKERUSERNAME with your actual Docker Hub username.

Deploy with:

faas-cli deploy -f stack.yml

Deploy the Function on OpenFaaS

Step 4: Expose and Test the Function

After deployment, your function will be accessible via an API endpoint on the OpenFaaS gateway. Test the function by sending a POST request to the /diagnose endpoint. You can use tools like curl or Postman:

curl -X POST http://localhost:8080/function/diagnosis -H "Content-Type: application/json" -d '{"symptoms": "Fever, Abdominal pain, Headache"}'

You should receive a structured JSON response similar to:

[
    {
        "disease": "ACUTE FEVER",
        "recommended_action": ""
    },
    {
        "disease": "FEVER IN CHILDREN",
        "recommended_action": ""
    },
    {
        "disease": "SALIENT FEATURES",
        "recommended_action": "• Salbutamol inhalation 2.5 mg/dose (5 mg/ml"
    }
]

This test confirms that the function processes the input, accesses the dataset, and returns a properly structured response thanks to PydanticAI.

Observing Automatic Scaling

One of the great benefits of using OpenFaaS on Civo is its ability to automatically scale based on demand. As requests to your function increase, OpenFaaS will spawn more instances of your Docker container to handle the load, ensuring that your symptom checker remains responsive under heavy traffic.

Key takeaways

Throughout this tutorial, we have gone through the following:

PydanticAI Simplifies AI Development: PydanticAI helps you build robust AI applications by ensuring that model outputs are structured and type-safe, reducing the risk of errors in production.
Serverless Functions Remove Infrastructure Overhead: Using OpenFaaS on Civo, you can deploy and scale your functions without worrying about server management, letting you focus on code and functionality.
Seamless Integration: By combining PydanticAI with serverless computing, you can create scalable AI applications like a medical symptom checker with minimal effort.
Scalability in Action: Leveraging Civo’s Kubernetes clusters, OpenFaaS automatically handles scaling, ensuring your application performs well under increasing demand.
Practical and Extensible: The tutorial using the COMMON DISEASES dataset demonstrates a real-world scenario where structured AI responses improve reliability and user experience. The same approach can be extended to other domains and datasets.

Conclusion

This tutorial walked you through the process of setting up a Civo environment, deploying OpenFaaS for serverless computing, and building a PydanticAI-powered medical symptom checker using a COMMON DISEASES dataset. The result is a scalable, efficient, and responsive application that leverages modern cloud and AI technologies.

The combination of PydanticAI and serverless functions not only simplifies development but also provides the scalability needed to handle growing workloads. As you explore further, you’ll find that this approach can be adapted for a variety of applications—from personalized recommendation engines to automated monitoring systems.

Simplifying AI Development with PydanticAI and Serverless Functions on Civo

Introduction

What is PydanticAI?

Key benefits of PydanticAI

Why are we using PydanticAI in this tutorial?

Setting up your environment on Civo

Step 1: Create a Civo account

Step 2: Install the Civo CLI

Step 3: Create a Kubernetes cluster on Civo

Deploying OpenFaaS for serverless AI

Step 1: Install required tools

Step 2: Deploy OpenFaaS using Helm

Step 3: Access the OpenFaaS Dashboard

Building and deploying a PydanticAI function

Step 1: Create the AI Function

Step 2: Package the Function in a Docker Container

Step 3: Deploy the Function on OpenFaaS

Step 4: Expose and Test the Function

Observing Automatic Scaling

Key takeaways

Conclusion

Mostafa Ibrahim

These may also be of interest

Pushing the limits: Guide to stress testing ML Models on Civo GPUs

Guide to Helm 3 with an Express.js microservice

Deploying Llama 3.1 with Kubeflow on Kubernetes: A CPU-Focused tutorial

Kubernetes

Compute

Databases

CivoStack Enterprise

Civo FlexCore

CivoStack for Service Providers

Cloud GPU

Carbon neutral GPU

Kubeflow as a Service

Case studies & testimonials

Learn

Blog

White papers

Documentation

Civo news

Meetups

Marketplace

Use Civo for your demos

Simplifying AI Development with PydanticAI and Serverless Functions on Civo

Introduction

What is PydanticAI?

Key benefits of PydanticAI

Why are we using PydanticAI in this tutorial?

Setting up your environment on Civo

Step 1: Create a Civo account

Step 2: Install the Civo CLI

Step 3: Create a Kubernetes cluster on Civo

Deploying OpenFaaS for serverless AI

Step 1: Install required tools

Step 2: Deploy OpenFaaS using Helm

Step 3: Access the OpenFaaS Dashboard

Building and deploying a PydanticAI function

Step 1: Create the AI Function

Step 2: Package the Function in a Docker Container

Step 3: Deploy the Function on OpenFaaS

Step 4: Expose and Test the Function

Observing Automatic Scaling

Key takeaways

Conclusion

Mostafa Ibrahim

These may also be of interest

Pushing the limits: Guide to stress testing ML Models on Civo GPUs

Guide to Helm 3 with an Express.js microservice

Deploying Llama 3.1 with Kubeflow on Kubernetes: A CPU-Focused tutorial