But once the business model includes a deposit now, a balance later, and referral or coupon attribution in between, the problem changes completely.

At that point, you're not just collecting money. You're managing a payment workflow.

In this tutorial, I’ll show you how to build a referral-aware split payment flow in Django that:

• tracks Step 2 deposit and balance separately

• supports coupon and partner linkage

• prevents duplicate payment processing

• uses database transactions safely

• keeps referral payouts consistent

• unlocks deliverables only when the workflow is complete

The main idea is simple: treat payment as a state transition, not just a webhook event.

Table of Contents

• Prerequisites

• Project Structure

• Designing the Data Model

• How Split Payments Work

• Finalizing Payments Safely

• Handling Webhooks Idempotently

• Applying Coupons and Referral Attribution

• Why the Referral Payout Should Be Explicit

• Unlocking Deliverables at the Right Time

• Common Mistakes

• Conclusion

Prerequisites

Before following along, you should already be comfortable with:

• Django models, views, and querysets

• database transactions in Django

• basic webhook concepts

• Python class-based or function-based view patterns

• how payment providers like Stripe or Paystack send event callbacks

You don't need to be an expert in payments, but you should understand how Django talks to the database and how to store state safely.

Project Structure

Here's a simple structure for the parts we need:

payments/
├── models.py
├── services.py
├── views.py
├── urls.py
└── webhooks.py

This separation matters.

• models.py stores the business state

• services.py contains the finalization logic

• views.py handles user-facing payment actions

• webhooks.py receives gateway callbacks

• urls.py connects endpoints

Keeping payment logic out of views makes the system easier to test and much harder to break.

Designing the Data Model

The most important decision is to model the payment stages clearly.

Instead of storing one vague “paid” flag, define the stages your business actually uses. For example:

from django.db import models
from django.conf import settings

class Journey(models.Model):
    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
    deposit_paid = models.BooleanField(default=False)
    balance_paid = models.BooleanField(default=False)
    deliverables_released = models.BooleanField(default=False)
    referral_code = models.CharField(max_length=50, blank=True, default="")
    partner_name = models.CharField(max_length=120, blank=True, default="")
    created_at = models.DateTimeField(auto_now_add=True)

class Payment(models.Model):
    STAGE_DEPOSIT = "deposit"
    STAGE_BALANCE = "balance"

    STAGE_CHOICES = [
        (STAGE_DEPOSIT, "Deposit"),
        (STAGE_BALANCE, "Balance"),
    ]

    STATUS_PENDING = "pending"
    STATUS_SUCCEEDED = "succeeded"
    STATUS_FAILED = "failed"

    STATUS_CHOICES = [
        (STATUS_PENDING, "Pending"),
        (STATUS_SUCCEEDED, "Succeeded"),
        (STATUS_FAILED, "Failed"),
    ]

    journey = models.ForeignKey(Journey, on_delete=models.CASCADE, related_name="payments")
    stage = models.CharField(max_length=20, choices=STAGE_CHOICES)
    gateway_reference = models.CharField(max_length=120, unique=True)
    amount = models.DecimalField(max_digits=10, decimal_places=2)
    discount_amount = models.DecimalField(max_digits=10, decimal_places=2, default=0)
    net_amount = models.DecimalField(max_digits=10, decimal_places=2)
    status = models.CharField(max_length=20, choices=STATUS_CHOICES, default=STATUS_PENDING)
    raw_payload = models.JSONField(null=True, blank=True)
    finalized_at = models.DateTimeField(null=True, blank=True)

class ReferralPayout(models.Model):
    payment = models.OneToOneField(Payment, on_delete=models.CASCADE, related_name="referral_payout")
    partner_name = models.CharField(max_length=120)
    amount = models.DecimalField(max_digits=10, decimal_places=2)
    is_paid = models.BooleanField(default=False)
    created_at = models.DateTimeField(auto_now_add=True)

This model design gives you a clean separation:

• Journey represents the customer’s overall progress

• Payment represents each financial event

• ReferralPayout represents what the partner earns from that payment

That separation is what keeps the logic manageable.

How Split Payments Work

Split payments usually follow a simple pattern:

1. the customer pays a deposit

2. the system records that deposit

3. a later payment clears the balance

4. the full workflow becomes complete

5. deliverables unlock only after the right stage

The important part is that each payment stage should be explicit.

If you treat the deposit and balance as two different milestones, then:

• discounts can apply to one stage and not the other

• referral attribution can be recorded per stage

• payouts can happen only when the stage is truly completed

• admin users can see the exact status of the workflow

That's much safer than trying to infer meaning from the amount alone.

Finalizing Payments Safely

The finalization logic should live in a service function, not directly inside the webhook view.

Here's a simple example:

from django.db import transaction
from django.utils import timezone

def finalize_payment(*, payment):
    with transaction.atomic():
        locked_payment = Payment.objects.select_for_update().select_related("journey").get(pk=payment.pk)

        if locked_payment.status == Payment.STATUS_SUCCEEDED:
            return locked_payment

        locked_payment.status = Payment.STATUS_SUCCEEDED
        locked_payment.finalized_at = timezone.now()
        locked_payment.save(update_fields=["status", "finalized_at"])

        journey = locked_payment.journey

        if locked_payment.stage == Payment.STAGE_DEPOSIT:
            journey.deposit_paid = True
        elif locked_payment.stage == Payment.STAGE_BALANCE:
            journey.balance_paid = True

        if journey.deposit_paid and journey.balance_paid:
            journey.deliverables_released = True

        journey.save(update_fields=["deposit_paid", "balance_paid", "deliverables_released"])

        if journey.referral_code and not hasattr(locked_payment, "referral_payout"):
            ReferralPayout.objects.create(
                payment=locked_payment,
                partner_name=journey.partner_name,
                amount=locked_payment.net_amount * 0.10,
            )

        return locked_payment

There are three important ideas here.

First, transaction.atomic() makes sure the update happens as one unit.

Second, select_for_update() locks the row so two processes don't finalize the same payment at the same time.

Third, the function checks whether the payment was already processed before doing any work.

That gives you a safe and repeatable finalization path.

Handling Webhooks Idempotently

Payment gateways can send the same webhook more than once.

That means your webhook handler must be idempotent, which simply means it can safely run multiple times without creating duplicate records or breaking state.

Here's a clean pattern:

import json
from django.http import HttpResponse, JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST

@csrf_exempt
@require_POST
def payment_webhook(request):
    payload = json.loads(request.body.decode("utf-8"))

    event_type = payload.get("event")
    data = payload.get("data", {})
    reference = data.get("reference")

    if not reference:
        return JsonResponse({"error": "Missing reference"}, status=400)

    if event_type != "charge.success":
        return HttpResponse(status=200)

    payment = Payment.objects.filter(gateway_reference=reference).first()
    if not payment:
        return JsonResponse({"error": "Payment not found"}, status=404)

    finalize_payment(payment=payment)
    return HttpResponse(status=200)

This view stays intentionally small.

It doesn't try to decide business rules. It only reads the webhook, finds the payment, and passes it to the service layer.

That makes it much easier to test and debug.

Applying Coupons and Referral Attribution

Coupons and partner codes become tricky when the payment is split across stages.

For example, a coupon might apply only to the deposit. Or it might apply to the balance only. Or it might affect both.

The best solution is to store that rule explicitly.

Here's a simple model for stage-aware coupon logic:

class DiscountCode(models.Model):
    APPLIES_DEPOSIT = "deposit"
    APPLIES_BALANCE = "balance"
    APPLIES_BOTH = "both"

    APPLIES_CHOICES = [
        (APPLIES_DEPOSIT, "Deposit only"),
        (APPLIES_BALANCE, "Balance only"),
        (APPLIES_BOTH, "Both stages"),
    ]

    code = models.CharField(max_length=50, unique=True)
    partner_name = models.CharField(max_length=120, blank=True, default="")
    applies_to = models.CharField(max_length=20, choices=APPLIES_CHOICES, default=APPLIES_BOTH)
    percent_off = models.PositiveSmallIntegerField(default=0)
    is_active = models.BooleanField(default=True)

Now your payment flow can check whether the coupon is valid for the current stage before applying it.

A helper function might look like this:

def calculate_discount(amount, coupon, stage):
    if not coupon or not coupon.is_active:
        return 0

    if coupon.applies_to == DiscountCode.APPLIES_DEPOSIT and stage != Payment.STAGE_DEPOSIT:
        return 0

    if coupon.applies_to == DiscountCode.APPLIES_BALANCE and stage != Payment.STAGE_BALANCE:
        return 0

    return amount * (coupon.percent_off / 100)

This keeps referral and coupon logic predictable.

Why the Referral Payout Should Be Explicit

A lot of systems accidentally mix these ideas:

• payment received

• coupon applied

• referral credited

• referral paid out

Those aren't the same thing.

A referral code can be attached at checkout, but the actual payout should be created only when the business rules say it's safe.

For example, you might decide:

• the partner gets credited when the deposit is paid

• the payout is created only after the balance clears

• the payout amount is based on the final net payment

That way, you don't pay out early if the customer never completes the full flow.

Unlocking Deliverables at the Right Time

One of the biggest mistakes in split payment systems is unlocking everything after the first payment.

That creates operational problems and trust issues.

A better rule is:

• deposit confirms intent

• balance confirms completion

• deliverables unlock only after the balance is received

You can keep that logic very simple in the Journey model:

def update_delivery_state(journey):
    journey.deliverables_released = journey.deposit_paid and journey.balance_paid
    journey.save(update_fields=["deliverables_released"])

The logic is readable, testable, and easy for an admin to understand.

Common Mistakes

Here are the mistakes that usually cause trouble in split payment systems:

1. Using one payment flag for everything

A single paid=True field isn't enough when the business has multiple payment stages.

2. Letting the webhook write directly to many tables

That makes the flow hard to test and easy to break. Use a service layer instead.

3. Forgetting idempotency

If the gateway retries a webhook, you shouldn't create duplicate payouts or double-update the journey.

4. Applying coupons without checking the stage

A code that's valid for the deposit may not be valid for the balance.

5. Releasing deliverables too early

Payment received doesn't always mean the workflow is complete.

Conclusion

Referral-aware split payment systems aren't hard because of the payment gateway. They're hard because the business rules are multi-step.

If you want the system to stay reliable, you should:

• model each payment stage explicitly

• store coupon and referral logic separately

• finalize payments inside transaction.atomic()

• lock rows with select_for_update()

• make webhook handling idempotent

• unlock deliverables only when the full workflow is complete

That approach keeps your Django app honest, traceable, and much easier to maintain as the product grows.

The goal isn't to build a flashy demo. The goal is to show how an AI agent can work on a stack that is realistic for many developers: PHP for request handling, MySQL for persistence, Gemini Flash for reasoning and function calling, and cPanel for deployment on standard shared hosting.

By the end of this article, you’ll understand:

• How to structure an agent loop

• How tool calling works in practice

• How to store and reload conversation memory

• How to expose the system through a public API endpoint

• How to deploy the project on standard shared hosting

Table of Contents

• Prerequisites

• Architecture Overview

• Project Structure

• Set Up the MySQL Database

• Connect PHP to MySQL

• Call Gemini Flash from PHP

• Define the Tool Registry

• Build the Tools

• Add MySQL Conversation Memory

• Create the Agent Loop

• Expose the Public API Endpoint

• Deploy on cPanel

• Test the Agent

• Production Hardening Ideas

• Conclusion

Prerequisites

Before you start, you should have:

• PHP 8.1 or newer

• MySQL access

• cURL enabled in PHP

• A Gemini API key from Google AI Studio

• Basic familiarity with PHP arrays, JSON, and SQL

• Access to cPanel and phpMyAdmin

You don't need a separate application server. The PHP files can run on ordinary shared hosting, provided your account supports PHP, cURL, MySQL, and outbound HTTPS requests.

In a typical cPanel setup, the project will be stored inside a folder under public_html.

Architecture Overview

The system has five main parts:

1. A public endpoint receives the user request.

2. An agent loop sends the conversation to Gemini Flash.

3. A tool registry tells Gemini which functions are available.

4. PHP tools perform actions such as saving notes, searching the web, or sending email.

5. MySQL stores the conversation so the agent can continue across requests.

The flow is straightforward. A user sends a message to the PHP endpoint. The endpoint passes the message and session ID to the agent. The agent loads previous messages and sends the conversation to Gemini along with the available tools. Gemini then decides what to do.

If the request can be answered directly, Gemini returns text. If an action is required, Gemini returns a function call containing the tool name and its arguments.

PHP receives the function call, runs the matching tool, and adds the result to the conversation. That result is sent back to Gemini, which can then call another tool or return a final answer.

This loop is what makes the application an agent rather than a basic chatbot.

Project Structure

Create a folder named agent inside public_html:

/public_html/agent/
├── index.php
├── agent.php
├── gemini.php
├── db.php
├── memory.php
├── tool_registry.php
├── tools/
│   ├── save_note.php
│   ├── search_web.php
│   ├── send_email.php
│   └── .htaccess
└── .htaccess

Each file has one main responsibility:

• index.php receives the HTTP request and returns JSON.

• agent.php contains the reasoning loop.

• gemini.php communicates with Gemini.

• db.php creates the MySQL connection.

• memory.php loads and saves conversation history.

• tool_registry.php describes the available tools.

• The tools folder contains the functions that perform actions.

Keeping the files separate makes the project easier to maintain. You can add a new tool without changing the rest of the application.

Add this to tools/.htaccess:

Deny from all

The tools folder shouldn't be accessible through a public URL. These files can write to the database, make external requests, or send email.

Set Up the MySQL Database

The application needs one table for conversation history and another for saved notes.

Run this SQL in phpMyAdmin:

CREATE DATABASE IF NOT EXISTS ai_agent;

USE ai_agent;

CREATE TABLE agent_memory (
    id INT AUTO_INCREMENT PRIMARY KEY,
    session_id VARCHAR(64) NOT NULL,
    role VARCHAR(20) NOT NULL,
    content TEXT NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_session_id (session_id)
);

CREATE TABLE agent_notes (
    id INT AUTO_INCREMENT PRIMARY KEY,
    session_id VARCHAR(64) NOT NULL,
    note TEXT NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

The agent_memory table stores the messages that make up each conversation. The session_id column separates one conversation from another, while role identifies whether the message came from the user, model, or a function.

The agent_notes table stores information that the user deliberately asks the agent to remember. Keeping notes separate from conversation history makes them easier to retrieve and use as application data.

If cPanel adds an account prefix to your database name, use the complete name in db.php. For example, ai_agent may become account_ai_agent.

Connect PHP to MySQL

Create db.php:

<?php

function db(): PDO
{
    static $pdo = null;

    if ($pdo === null) {
        $pdo = new PDO(
            "mysql:host=localhost;dbname=ai_agent;charset=utf8mb4",
            "db_user",
            "db_password",
            [
                PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
                PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC,
            ]
        );
    }

    return $pdo;
}

The function uses PDO to connect to MySQL. The static variable ensures that the same connection is reused for subsequent requests, rather than opening a new one each time the agent accesses the database.

PDO::ATTR_ERRMODE makes database failures throw exceptions. This makes errors easier to detect and handle.

Replace ai_agent, db_user, and db_password with the actual database details from cPanel. The utf8mb4 character set allows the database to store a wide range of characters, including emoji and non-English text.

Call Gemini Flash from PHP

Gemini supports function calling. It can decide that a tool is needed and return the tool name and arguments. It doesn't execute the PHP function itself. Your application is responsible for validating the request and running the tool.

For example, Gemini might return:

{
  "note": "Our launch is on 1 September 2026"
}

Create gemini.php:

<?php

function gemini_request(array $contents, array $tools = []): array
{
    $apiKey = "YOUR_GEMINI_API_KEY";

    $url =
        "https://generativelanguage.googleapis.com/v1beta/models/" .
        "gemini-1.5-flash:generateContent?key=" . $apiKey;

    $payload = [
        "contents" => $contents
    ];

    if (!empty($tools)) {
        $payload["tools"] = [
            [
                "functionDeclarations" => $tools
            ]
        ];
    }

    $ch = curl_init($url);

    curl_setopt_array($ch, [
        CURLOPT_RETURNTRANSFER => true,
        CURLOPT_POST => true,
        CURLOPT_HTTPHEADER => [
            "Content-Type: application/json"
        ],
        CURLOPT_POSTFIELDS => json_encode($payload),
        CURLOPT_TIMEOUT => 30
    ]);

    $response = curl_exec($ch);
    $httpCode = curl_getinfo($ch, CURLINFO_HTTP_CODE);

    if ($response === false) {
        $error = curl_error($ch);
        curl_close($ch);
        throw new RuntimeException("Gemini request failed: " . $error);
    }

    curl_close($ch);

    if ($httpCode !== 200) {
        throw new RuntimeException("Gemini API error: " . $response);
    }

    $decoded = json_decode($response, true);

    if (!is_array($decoded)) {
        throw new RuntimeException("Gemini returned invalid JSON.");
    }

    return $decoded;
}

function parse_gemini_response(array $response): array
{
    $part = $response["candidates"][0]["content"]["parts"][0] ?? [];

    if (isset($part["functionCall"])) {
        return [
            "type" => "function_call",
            "name" => $part["functionCall"]["name"],
            "args" => $part["functionCall"]["args"] ?? []
        ];
    }

    return [
        "type" => "text",
        "text" => $part["text"] ?? ""
    ];
}

The gemini_request function sends the conversation and tool definitions to Gemini. The parser converts Gemini's response into either a text response or a function call.

The agent loop can then make a simple decision:

• If the type is function_call, execute the requested tool.

• If the type is text, return the answer to the user.

For a real deployment, store the API key outside the public web directory whenever possible.

Define the Tool Registry

The tool registry tells Gemini which tools exist and what arguments they require.

Create tool_registry.php:

<?php

function tool_definitions(): array
{
    return [
        [
            "name" => "save_note",
            "description" => "Save an important note to the database.",
            "parameters" => [
                "type" => "object",
                "properties" => [
                    "note" => [
                        "type" => "string"
                    ]
                ],
                "required" => ["note"]
            ]
        ],
        [
            "name" => "search_web",
            "description" => "Search the web for current information.",
            "parameters" => [
                "type" => "object",
                "properties" => [
                    "query" => [
                        "type" => "string"
                    ]
                ],
                "required" => ["query"]
            ]
        ],
        [
            "name" => "send_email",
            "description" => "Send an email when the user explicitly asks.",
            "parameters" => [
                "type" => "object",
                "properties" => [
                    "to" => [
                        "type" => "string"
                    ],
                    "subject" => [
                        "type" => "string"
                    ],
                    "body" => [
                        "type" => "string"
                    ]
                ],
                "required" => ["to", "subject", "body"]
            ]
        ]
    ];
}

The description helps Gemini decide when to use a tool. The parameters describe the values that Gemini should provide.

The registry doesn't replace server-side validation. Every PHP tool must still check its own arguments before performing an action.

Build the Tools

Each tool should:

1. Read the arguments from Gemini.

2. Validate the input.

3. Perform the action.

4. Return a JSON result.

Save Note Tool

Create tools/save_note.php:

<?php

require_once __DIR__ . "/../db.php";

function save_note_tool(array $args, string $sessionId): string
{
    $note = trim($args["note"] ?? "");

    if ($note === "") {
        return json_encode([
            "success" => false,
            "message" => "Empty note"
        ]);
    }

    $stmt = db()->prepare(
        "INSERT INTO agent_notes (session_id, note)
         VALUES (:session_id, :note)"
    );

    $stmt->execute([
        ":session_id" => $sessionId,
        ":note" => $note
    ]);

    return json_encode([
        "success" => true,
        "message" => "Note saved"
    ]);
}

The note is trimmed and checked before it is saved. The prepared statement prevents SQL injection and safely handles the input.

Search Web Tool

Create tools/search_web.php:

<?php

function search_web_tool(array $args): string
{
    $query = trim($args["query"] ?? "");

    if ($query === "") {
        return json_encode([
            "success" => false,
            "message" => "Search query is empty"
        ]);
    }

    $url = "https://api.example.com/search?q=" . urlencode($query);

    $ch = curl_init($url);

    curl_setopt_array($ch, [
        CURLOPT_RETURNTRANSFER => true,
        CURLOPT_TIMEOUT => 15
    ]);

    $response = curl_exec($ch);

    if ($response === false) {
        $error = curl_error($ch);
        curl_close($ch);

        return json_encode([
            "success" => false,
            "message" => "Search failed",
            "error" => $error
        ]);
    }

    curl_close($ch);

    return $response;
}

The URL is a placeholder. Replace it with your chosen search provider and add any required API key or authentication header.

The timeout prevents a slow external service from keeping the PHP request open indefinitely.

Send Email Tool

Create tools/send_email.php:

<?php

function send_email_tool(array $args): string
{
    $to = filter_var(
        $args["to"] ?? "",
        FILTER_VALIDATE_EMAIL
    );

    $subject = trim($args["subject"] ?? "");
    $body = trim($args["body"] ?? "");

    if (!$to) {
        return json_encode([
            "success" => false,
            "message" => "Invalid email address"
        ]);
    }

    if ($subject === "" || $body === "") {
        return json_encode([
            "success" => false,
            "message" => "Email subject and body are required"
        ]);
    }

    $headers = "From: agent@yourdomain.com\r\n";
    $headers .= "Content-Type: text/plain; charset=UTF-8\r\n";

    $sent = mail($to, $subject, $body, $headers);

    return json_encode([
        "success" => $sent,
        "message" => $sent ? "Email sent" : "Email failed"
    ]);
}

The recipient address, subject, and body are checked before sending. Replace the From address with one belonging to your domain.

The mail() function may be available on shared hosting, but an authenticated email service or SMTP provider is usually more reliable for production applications.

Add MySQL Conversation Memory

Gemini doesn't automatically remember previous API requests. The application must load the conversation from MySQL before each request and save the updated history afterwards.

Create memory.php:

<?php

require_once __DIR__ . "/db.php";

function load_memory(string $sessionId): array
{
    $stmt = db()->prepare(
        "SELECT role, content
         FROM agent_memory
         WHERE session_id = :session_id
         ORDER BY id ASC"
    );

    $stmt->execute([
        ":session_id" => $sessionId
    ]);

    $history = [];

    foreach ($stmt->fetchAll() as $row) {
        $history[] = [
            "role" => $row["role"],
            "parts" => [
                [
                    "text" => $row["content"]
                ]
            ]
        ];
    }

    return $history;
}

function save_memory(string $sessionId, array $history): void
{
    $pdo = db();

    $delete = $pdo->prepare(
        "DELETE FROM agent_memory
         WHERE session_id = :session_id"
    );

    $delete->execute([
        ":session_id" => $sessionId
    ]);

    $insert = $pdo->prepare(
        "INSERT INTO agent_memory
         (session_id, role, content)
         VALUES (:session_id, :role, :content)"
    );

    foreach ($history as $turn) {
        $part = $turn["parts"][0] ?? [];

        $text = isset($part["text"])
            ? $part["text"]
            : json_encode($part);

        $insert->execute([
            ":session_id" => $sessionId,
            ":role" => $turn["role"],
            ":content" => $text
        ]);
    }
}

load_memory Retrieves messages for the current session and rebuilds them in the format expected by Gemini.

save_memory replaces the stored history with the current history. This is simple and suitable for a small tutorial application. A larger system could append only new messages, use a JSON column, or summarise older conversations to reduce database and API usage.

Create the Agent Loop

The agent loop connects the database, Gemini, memory, and tools.

Create agent.php:

<?php

require_once __DIR__ . "/gemini.php";
require_once __DIR__ . "/memory.php";
require_once __DIR__ . "/tool_registry.php";
require_once __DIR__ . "/tools/save_note.php";
require_once __DIR__ . "/tools/search_web.php";
require_once __DIR__ . "/tools/send_email.php";

function run_tool(
    string $name,
    array $args,
    string $sessionId
): string {
    return match ($name) {
        "save_note" => save_note_tool($args, $sessionId),
        "search_web" => search_web_tool($args),
        "send_email" => send_email_tool($args),
        default => json_encode([
            "success" => false,
            "message" => "Unknown tool"
        ])
    };
}

function run_agent(
    string $message,
    string $sessionId
): string {
    $history = load_memory($sessionId);

    $history[] = [
        "role" => "user",
        "parts" => [
            [
                "text" => $message
            ]
        ]
    ];

    $tools = tool_definitions();
    $limit = 5;
    $step = 0;

    while ($step < $limit) {
        $step++;

        $response = gemini_request($history, $tools);
        $parsed = parse_gemini_response($response);

        if ($parsed["type"] === "text") {
            $history[] = [
                "role" => "model",
                "parts" => [
                    [
                        "text" => $parsed["text"]
                    ]
                ]
            ];

            save_memory($sessionId, $history);

            return $parsed["text"];
        }

        if ($parsed["type"] === "function_call") {
            $toolName = $parsed["name"];
            $toolArgs = $parsed["args"];
            $result = run_tool($toolName, $toolArgs, $sessionId);

            $history[] = [
                "role" => "model",
                "parts" => [
                    [
                        "functionCall" => [
                            "name" => $toolName,
                            "args" => $toolArgs
                        ]
                    ]
                ]
            ];

            $history[] = [
                "role" => "function",
                "parts" => [
                    [
                        "functionResponse" => [
                            "name" => $toolName,
                            "response" => [
                                "content" => $result
                            ]
                        ]
                    ]
                ]
            ];
        }
    }

    save_memory($sessionId, $history);

    return "I could not complete the task within the allowed number of steps.";
}

The run_tool function routes the requested tool to the correct PHP function. The default case handles unexpected tool names safely.

The run_agent function first loads the existing history and adds the new user message. It then sends the conversation to Gemini.

If Gemini returns text, the response is saved and returned to the user.

If Gemini returns a function call, PHP executes the tool. The function call and its result are both added to the history before the next loop iteration.

The five-step limit prevents the model from repeatedly calling tools without finishing. You can adjust the limit according to the needs of your application.

Expose the Public API Endpoint

Create index.php:

<?php

require_once __DIR__ . "/agent.php";

header("Content-Type: application/json");

$input = json_decode(
    file_get_contents("php://input"),
    true
);

if (!is_array($input)) {
    http_response_code(400);

    echo json_encode([
        "error" => "Invalid JSON body"
    ]);

    exit;
}

$message = trim($input["message"] ?? "");
$sessionId = trim($input["session_id"] ?? "");

if ($message === "" || $sessionId === "") {
    http_response_code(400);

    echo json_encode([
        "error" => "message and session_id are required"
    ]);

    exit;
}

try {
    $reply = run_agent($message, $sessionId);

    echo json_encode([
        "reply" => $reply,
        "session_id" => $sessionId
    ]);
} catch (Throwable $e) {
    http_response_code(500);

    echo json_encode([
        "error" => $e->getMessage()
    ]);
}

The endpoint expects a JSON request containing a message and session ID:

{
  "message": "Save a note that our launch is on 1 September 2026",
  "session_id": "demo123"
}

The frontend should reuse the same session ID for messages in the same conversation. A new session ID creates a separate conversation.

During development, returning the exception message can help with debugging. In production, log detailed errors privately and return a general error message to users.

Deploy on cPanel

Follow these steps:

1. Upload the project to /public_html/agent/.

2. Create a database and user in cPanel.

3. Grant the user access to the database.

4. Run the SQL in phpMyAdmin.

5. Update the credentials in db.php.

6. Add the Gemini API key in gemini.php.

7. Select PHP 8.1 or newer.

8. Enable the cURL extension.

9. Add the .htaccess file to the tools folder.

The endpoint should then be available at:

https://yourdomain.com/agent/index.php

Test the Agent

Save a note:

curl -X POST https://yourdomain.com/agent/index.php \
  -H "Content-Type: application/json" \
  -d '{"message":"Save a note that our launch is on 1 September 2026","session_id":"demo123"}'

The agent should call save_note, store the note in MySQL, and return a confirmation.

You can test memory by sending another request with the same session ID:

curl -X POST https://yourdomain.com/agent/index.php \
  -H "Content-Type: application/json" \
  -d '{"message":"What note did I save earlier?","session_id":"demo123"}'

The agent should load the previous conversation from agent_memory and use it to answer.

If something fails, check the database credentials, database tables, Gemini API key, PHP version, cURL extension, and server error logs. Also make sure the search tool doesn't still point to the placeholder API URL.

Production Hardening Ideas

Before allowing real users to access the application, consider adding:

Authentication

Require an API token or another authentication method. Otherwise, anyone who discovers the endpoint may be able to use your tools and Gemini account.

Rate limiting

Limit requests by session ID, IP address, or authenticated user to prevent abuse and unexpected usage.

Tool logging

Store the session ID, tool name, arguments, result, and execution time. This helps you investigate unexpected behaviour.

Stronger validation

Validate every tool argument. Check email addresses, reject empty values, restrict string lengths, and validate database identifiers.

Better memory management

Long conversations can make API requests larger and less efficient. Consider keeping recent messages, summarising older messages, or storing structured tool calls separately.

Confirmation for sensitive actions

For actions such as sending email, ask the user for confirmation before executing the tool. Prompt instructions are helpful, but the application should also handle confirmation.

Credential protection

Don't store API keys and database passwords in a public repository. Keep sensitive configuration outside the public web directory when possible.

Conclusion

You don't need a complex cloud stack to build a useful AI agent.

With PHP, MySQL, Gemini Flash, and cPanel, you can create a system that reasons, calls tools, stores memory, and runs on infrastructure that many developers already understand.

The architecture is built around a simple process:

1. The endpoint receives the user's request.

2. MySQL provides the previous conversation.

3. Gemini decides whether a tool is needed.

4. PHP executes the tool.

5. The result is sent back to Gemini.

6. Gemini produces the final answer.

7. The updated conversation is saved.

This gives you a practical foundation for building agent-based applications on standard shared hosting.

For my work on circular economy and battery recycling, I needed a way to query multiple databases at once without the manual fatigue.

In this tutorial, you'll build an automated research pipeline using n8n that reduces roughly six hours of manual literature review into a five-minute automated process.

This isn’t a “cool demo workflow.” It’s a production-minded pipeline with parallel collection, normalisation, deduplication, structured AI extraction, scoring, and practical error handling.

Table of Contents

1. Prerequisites

2. The Problem: Research Takes Too Long

3. The Tech Stack

4. The Project Structure: How to Think About an n8n Workflow Like Software

5. Stage 1: Centralised Configuration

6. Stage 2: Parallel API Collection (With Failure Isolation)

7. Stage 3: Normalisation and Deduplication (DOI-first, Title fallback)

8. Stage 4: AI-Powered Content Extraction (Strict JSON)

9. Stage 5: Scoring and Synthesis

10. [Beginner-Friendly Evals (Retrieval and Extraction QA)(#heading-beginnerfriendly-evals-retrieval-and-extraction-qa)

11. Key Learnings and Error Handling

12. Conclusion

Prerequisites

You don’t need to be a DevOps engineer to follow this, but you should have:

• Basic comfort with APIs and JSON (request/response payloads)

• Familiarity with spreadsheets (Google Sheets basics)

• Willingness to use a small amount of JavaScript inside n8n Function/Code nodes

Access to:

• An n8n instance (self-hosted or cloud)

• A Groq API key (or a compatible LLM provider)

• Optional API keys, depending on the databases you use

What you’ll build assumes:

• You’re extracting from metadata + abstracts (not downloading full PDFs).

• You can accept that some sources will occasionally rate-limit or return partial results (and your workflow will be designed to survive this).

The Problem: Research Takes Too Long

Manual research is often a bottleneck for innovation. Before building this automation, my workflow involved searching multiple academic databases, scanning abstracts, and manually extracting key findings. This process was not only slow but also prone to human error and inconsistent note-taking.

The goal of this automation is to provide a “full-stack research assistant” that handles the heavy lifting of collecting candidate papers, removing duplicates, extracting consistent fields, scoring relevance and quality, and delivering a curated daily or weekly report, so you can spend your time on high-level synthesis rather than repetitive collection.

The Tech Stack

This workflow leverages a combination of automation tooling, high-speed LLM inference, and academic metadata providers.

Notes: coverage varies by provider. Some APIs return abstracts reliably, while others may omit them. Your pipeline should treat missing abstracts as a normal case, not a failure.

The Project Structure: How to Think About an n8n Workflow Like Software

While n8n is a visual tool, it helps to design your workflow as modular stages to avoid the “spaghetti workflow” problem.

.
├── configuration/         # Keywords, thresholds, limits, date filters
├── collectors/            # Parallel HTTP request nodes (multiple sources)
├── processing/            # Normalization + deduplication code nodes
├── extraction/            # LLM extraction nodes (strict JSON)
├── scoring/               # Relevance + quality scoring + filtering
└── delivery/              # Google Sheets + email/HTML report

Design principle: each stage should produce a clean, predictable output shape that the next stage can rely on.

Stage 1: Centralised Configuration

Instead of hardcoding search parameters (keywords, min year, citation thresholds) across multiple nodes, use one configuration node to define workflow variables.

This matters for maintainability (change a value once, not in ten nodes), reusability (repurpose the entire pipeline by swapping one config object), and debuggability (log the config at the start of each run so you can reproduce results).

Use a Set node, or a Code node returning JSON like this:

{
  "keywords": "circular economy battery recycling remanufacturing",
  "min_year": 2020,
  "max_results_per_source": 10,
  "min_citations": 2,
  "relevance_threshold": 15,
  "batch_size": 10
}

Tip: keep numeric fields as numbers (not strings) to avoid scoring bugs later.

Stage 2: Parallel API Collection (With Failure Isolation)

Your workflow should query multiple sources simultaneously. In n8n, you can branch from your configuration node into multiple HTTP Request nodes, and then merge results later.

The production mindset here is simple: APIs fail. Rate limits happen. Providers return partial data. The key is to prevent one failing collector from crashing the whole run.

To implement this, on each HTTP Request node, enable Continue On Fail (or the equivalent “don’t stop workflow” behaviour). Then, in the normalisation stage, treat missing or failed outputs as empty arrays so downstream stages still run.

In practice, it also helps to set explicit timeouts and add a small retry policy (one to two retries) for transient failures. “Good” looks like this: if two out of five sources fail, you still produce a useful report from the remaining three, and you log which sources failed so you can investigate later.

Stage 3: Normalisation and Deduplication (DOI-first, Title fallback)

Each academic API returns different field names and shapes. One might use title, another display_name, another paper_title. Your next stage should normalise all inputs into one schema.

Target normalised schema

Here’s a simple baseline schema (expand later as needed):

{
  "title": "string",
  "abstract": "string|null",
  "doi": "string|null",
  "year": 2024,
  "citations": 12,
  "url": "string|null",
  "source": "Semantic Scholar|OpenAlex|arXiv|PubMed"
}

What deduping by DOI means (and what a DOI is)

A DOI (Digital Object Identifier) is a unique, persistent identifier assigned to many scholarly publications. If a paper has a DOI, that DOI functions like a stable ID: the same paper may appear in multiple databases with slightly different metadata, but the DOI should remain consistent.

So, deduping by DOI means: if two records share the same DOI, treat them as the same paper and keep only one.

When a DOI is missing (which is common for some preprints and some API responses), the fallback is to dedupe using a normalised title key, lowercased, trimmed, punctuation stripped, and whitespace collapsed. It’s not as perfect as DOI-based matching, but it’s a strong pragmatic backup.

What “normalise into a unified object” means (what’s happening in the code)

“Normalise into a unified object” simply means converting every provider’s raw response into the same predictable shape (the schema above). Once everything looks the same, downstream steps, such as deduplication, scoring, AI extraction, and storage, become straightforward because they don’t need provider-specific logic.

In the code below, that’s what the normalized object is: it maps Semantic Scholar’s fields (paper.title, paper.externalIds.DOI, paper.citationCount) into your standard fields (title, doi, citations, etc.). After that, the workflow generates a dedupe key (doi:... if DOI exists, otherwise title:...) and uses a Set to keep only the first occurrence.

Example n8n Code Node (Normalisation + Dedupe Pattern)

const itemsIn = $input.all();

const seen = new Set();
const results = [];

function titleKey(t) {
  return (t || "")
    .toLowerCase()
    .replace(/[\W_]+/g, " ")
    .replace(/\s+/g, " ")
    .trim();
}

for (const item of itemsIn) {
  // Example: Semantic Scholar response shape
  const papers = item.json?.data || [];

  for (const paper of papers) {
    // "Normalize into a unified object":
    // take the provider-specific fields and map them into our standard schema.
    const normalized = {
      title: paper.title || null,
      abstract: paper.abstract || null,
      doi: paper.externalIds?.DOI || null,
      year: paper.year || null,
      citations: paper.citationCount || 0,
      url: paper.url || null,
      source: "Semantic Scholar",
    };

    if (!normalized.title) continue;

    // Dedupe key: DOI is strongest; title is fallback
    const key = normalized.doi
      ? `doi:${normalized.doi.toLowerCase()}`
      : `title:${titleKey(normalized.title)}`;

    if (seen.has(key)) continue;
    seen.add(key);

    results.push(normalized);
  }
}

return results.map(r => ({ json: r }));

Production-minded note: keep a field like source so you can debug where bad metadata is coming from later.

Stage 4: AI-Powered Content Extraction (Strict JSON)

Once you have a deduplicated list of papers, you can send each paper (or a small batch) to Groq for structured extraction.

Why structured output matters

If your LLM returns narrative text instead of JSON, misses fields, or emits malformed JSON, your workflow breaks downstream. In a production workflow, that’s not a rare edge case; it’s something you should expect and design around.

That’s why you’ll use strict schema prompting and validate responses downstream.

System prompt vs user prompt (and how to compose them)

A helpful way to think about prompts in production is:

• The system prompt defines the non-negotiable contract: output format, allowed keys, no commentary, and what to do in uncertain cases. This is where you say “return ONLY valid JSON” and “no extra keys.”

• The user prompt provides the variable data for this specific request: title, year, citations, abstract, and the exact schema you want filled.

Composing them this way keeps your workflow stable. The system prompt stays mostly constant (your formatting contract), while the user prompt changes per paper (your payload). It also makes debugging easier: if outputs start failing, you can adjust the system constraints without rewriting every payload template.

Suggested extraction schema

Extract only what you can support from abstract-level data:

• research_question

• methodology

• key_findings

• limitations

• notes (for missing abstract / ambiguity)

Example prompt (system + user)

System:

You are a research extraction engine. You must return ONLY valid JSON.
No markdown. No extra keys. No commentary.
If the abstract is missing or too vague, set fields to null and include a reason in "notes".

User:

Extract structured fields from this paper.

TITLE: {{title}}
YEAR: {{year}}
CITATIONS: {{citations}}
ABSTRACT: {{abstract}}

Return JSON with keys:
research_question (string|null)
methodology (string|null)
key_findings (array of strings)
limitations (array of strings)
notes (string)

Model settings: keep temperature low (around 0.2–0.3) and keep responses short and structured.

Batch processing to avoid timeouts

Instead of sending 50 papers at once, process them in batches (for example, 10). This reduces latency spikes, failure blast radius, and cost surprises. Smaller batches also make it easier to retry only the failing chunk rather than re-running everything.

Stage 5: Scoring and Synthesis

Not every retrieved paper is worth your time. Without scoring, your pipeline becomes a firehose: you’ve automated collection, but you still have to manually decide what to read. Scoring is what turns “a big list of results” into a shortlist you can trust.

I recommend computing two signals:

• Relevance: Is this actually about your research question?

• Quality/priority: If it’s relevant, is it worth reading first?

For relevance, keep it simple and explainable. Count keyword hits in the title and abstract (and optionally in extracted key_findings). Title matches should be weighted higher because titles are deliberately compact summaries. Abstract hits are useful too, but cap them so long abstracts don’t dominate the score.

For quality/priority, use lightweight metadata you already have. Recency is a strong signal in fast-moving areas, and citations can help, but they should be treated as a weak signal (and capped) so newer high-value papers aren’t unfairly penalised.

A solid first scoring model is: add a title bonus, add a capped abstract bonus, add a capped citations bonus, and add a small recency bonus for papers from the last two years. Then filter using the relevance_threshold results from Stage 1. The advantage of this approach is that it’s easy to debug and tune: you can always explain why a paper passed or failed.

Once you’ve filtered down to your “gold” set, synthesis becomes safer and more useful. Write one row per accepted paper to Google Sheets, then generate a daily/weekly HTML summary (for example, top 5 papers with 1–2 key findings each) and include links so you can verify quickly.

Beginner-Friendly Evals: Retrieval and Extraction QA

AI workflows regress silently. A prompt tweak, a model update, or an API schema change can break extraction without throwing an obvious error. Adding lightweight evals is the difference between “it worked last week” and “it’s reliable.”

The goal here isn’t to build a full evaluation framework. It’s to add small, cheap checks that catch the most common failure modes:

• Are collectors still returning results?

• Are we actually removing duplicates?

• Is the LLM returning valid JSON with the keys we require?

What it looks like in n8n (a concrete example)

A simple implementation is to add an “Assertions” Code node immediately after your extraction step, plus (optionally) another one after normalisation/deduplication.

At a high level, the workflow section looks like:

1. Collectors (parallel HTTP Request nodes)

2. Merge results

3. Normalise + dedupe (Code node)

4. Split in Batches (optional)

5. LLM extraction (Groq/OpenAI-compatible node)

6. Assertions (Code node)

7. If node (pass/fail)

8. Delivery (Sheets + email)

Example: Assertions code node after extraction

This code node assumes each item is a paper with:

• title, abstract in the normalised fields, and

• an extraction field (or whatever you name it) containing the LLM response as an object or JSON string.

Adapt the field name to match your actual node output, but the pattern is the same: parse, validate required keys, compute percentages, then decide whether to fail or warn.

const items = $input.all();

let total = items.length;
let withTitle = 0;
let withAbstract = 0;

let parseOk = 0;
let schemaOk = 0;

const requiredKeys = [
  "research_question",
  "methodology",
  "key_findings",
  "limitations",
  "notes",
];

const failures = [];

for (let i = 0; i < items.length; i++) {
  const p = items[i].json;

  if (p.title && String(p.title).trim().length > 0) withTitle++;
  if (p.abstract && String(p.abstract).trim().length > 0) withAbstract++;

  // Adjust this depending on where you store the model output:
  const raw = p.extraction ?? p.llm ?? p.model_output;

  let obj = null;
  try {
    obj = typeof raw === "string" ? JSON.parse(raw) : raw;
    parseOk++;
  } catch (e) {
    failures.push({ index: i, title: p.title || null, reason: "JSON parse failed" });
    continue;
  }

  const hasAllKeys = requiredKeys.every(k => Object.prototype.hasOwnProperty.call(obj, k));
  if (!hasAllKeys) {
    failures.push({ index: i, title: p.title || null, reason: "Missing required keys" });
    continue;
  }

  // Optional: ensure arrays are arrays
  const arraysOk = Array.isArray(obj.key_findings) && Array.isArray(obj.limitations);
  if (!arraysOk) {
    failures.push({ index: i, title: p.title || null, reason: "key_findings/limitations not arrays" });
    continue;
  }

  schemaOk++;
}

const pct = (n) => (total === 0 ? 0 : Math.round((n / total) * 100));

const report = {
  total_papers: total,
  pct_with_title: pct(withTitle),
  pct_with_abstract: pct(withAbstract),
  pct_extraction_json_parse_ok: pct(parseOk),
  pct_extraction_schema_ok: pct(schemaOk),
  failures_sample: failures.slice(0, 5),
};

// Decide pass/fail thresholds
const HARD_FAIL_PARSE_BELOW = 90;
const HARD_FAIL_SCHEMA_BELOW = 85;

const shouldFail =
  report.pct_extraction_json_parse_ok < HARD_FAIL_PARSE_BELOW ||
  report.pct_extraction_schema_ok < HARD_FAIL_SCHEMA_BELOW;

return [
  {
    json: {
      eval_report: report,
      shouldFail,
    },
  },
];

Then add an If node:

• If shouldFail is true, then route to an “Alert/Stop” branch (Slack/email/log) and optionally stop the workflow.

• If false, then continue to the delivery stage.

This is the automation equivalent of unit tests: small, cheap, and extremely effective. It also gives you a concrete paper trail when something changes upstream.

Key Learnings and Error Handling

Building this automation taught me that the best workflows are designed for failure.

First, error resilience is not optional. Never let one failing API crash the workflow. Use “Continue On Fail” on your HTTP nodes, merge partial results, and log which sources failed in your final report so you can debug without losing an entire run.

Second, batching is your friend. Process papers in batches (often 5–15) to reduce timeouts and cost spikes. Keep LLM payloads small and focused on what you actually need (metadata + abstract), and retry transient failures once rather than repeatedly hammering the model or API.

Third, structured prompting is what makes AI reliable in automation. A strict JSON schema is the difference between a workflow that runs unattended and one that breaks randomly. Keep temperature low, enforce the schema in the system prompt, and validate everything downstream with simple parse-and-assert checks.

Conclusion

A good research pipeline doesn’t just retrieve papers – it turns scattered results into a consistent, deduplicated, scored, and review-ready shortlist you can trust.

By treating your n8n workflow like software modular stages, strict contracts between steps, and lightweight eval checks, you can reduce hours of manual literature review into a fast, repeatable process that survives real-world API failures and model quirks.

If you build this with good defaults (failure isolation, batching, normalisation, strict JSON extraction, and simple scoring), you end up with something you can run daily or weekly and actually rely on without the manual fatigue.

About Me

I am Chidozie Managwu, an award-winning AI Product Architect and founder focused on helping global tech talent build real, production-ready skills. I contribute to global AI initiatives as a GAFAI Delegate and lead the AI Titans Network, a community for developers learning how to ship AI products.

My work has been recognised with the Global Tech Hero award and featured on platforms like HackerNoon.

They answer questions they should not, they break when document retrieval is weak, they time out due to network latency, and nobody can tell exactly what happened because there are no logs and no tests.

In this tutorial, you’ll build a beginner-friendly Retrieval Augmented Generation (RAG) application designed to survive production realities. This isn’t just a script that calls an API. It’s a system featuring a FastAPI backend, a persisted FAISS vector store, and essential safety guardrails (including a retrieval gate and fallbacks).

Table of Contents

1. Why RAG Alone Does Not Equal Production-Ready

2. The Architecture You Are Building

3. Project Setup and Structure

4. How to Build the RAG Layer with FAISS

5. How to Add the LLM Call with Structured Output

6. How to Add Guardrails: Retrieval Gate and Fallbacks

7. FastAPI App: Creating the /answer Endpoint

8. How to Add Beginner-Friendly Evals

9. What to Improve Next: Realistic Upgrades

Why RAG Alone Does Not Equal Production-Ready

Retrieval Augmented Generation (RAG) is often hailed as the hallucination killer. By grounding the model in retrieved text, we provide it with the facts it needs to be accurate. But simply connecting a vector database to an LLM isn’t enough for a production environment.

Production issues usually arise from the silent failures in the system surrounding the model:

• Weak retrieval: If the app retrieves irrelevant chunks of text, the model tries to bridge the gap by inventing an answer anyway. Without a designated “I do not know” path, the model is essentially forced to hallucinate.

• Lack of visibility: Without structured outputs and basic logging, you can’t tell if bad retrieval, a confusing prompt, or a model update caused a wrong answer.

• Fragility: A simple API timeout or malformed provider response becomes a user-facing outage if you don’t implement fallbacks.

• No regression testing: In traditional software, we have unit tests. In AI, we need evals. Without them, a small tweak to your prompt might fix one issue but break ten others without you realising it.

We’ll solve each of these issues systematically in this guide.

Prerequisites

This tutorial is beginner-friendly, but it assumes you have a few basics in place so you can focus on building a robust RAG system instead of getting stuck on setup issues.

Knowledge

You should be comfortable with:

• Python fundamentals (functions, modules, virtual environments)

• Basic HTTP + JSON (requests, response payloads)

• APIs with FastAPI (what an endpoint is and how to run a server)

• High-level LLM concepts (prompting, temperature, structured outputs)

Tools + Accounts

You’ll need:

• Python 3.10+

• A working OpenAI-compatible API key (OpenAI or any provider that supports the same request/response shape)

• A local environment where you can run a FastAPI app (Mac/Linux/Windows)

What This Tutorial Covers (and What It Doesn’t)

We’ll build a production-minded baseline:

• A FAISS-backed retriever with a persisted index + metadata

• A retrieval gate to prevent “forced hallucination”

• Structured JSON outputs so your backend is stable

• Fallback behavior for timeouts and provider errors

• A small eval harness to prevent regressions

We won’t implement advanced upgrades such as rerankers, semantic chunking, auth, background jobs beyond a roadmap at the end.

The Architecture You Are Building

The flow of our application follows a disciplined path so every answer is grounded in evidence:

1. User query: The user submits a question via a FastAPI endpoint.

2. Retrieval: The system embeds the question and retrieves the top-k most similar document chunks.

3. The retrieval gate: We evaluate the similarity score. If the context is not relevant enough, we stop immediately and refuse the query.

4. Augmentation and generation: If the gate passes, we send a context-augmented prompt to the LLM.

5. Structured response: The model returns a JSON object containing the answer, sources used, and a confidence level.

Project Setup and Structure

To keep things organized and maintainable, we’ll use a modular structure. This allows you to swap out your LLM provider or your vector database without rewriting your entire core application.

Project Structure

.
├── app.py              # FastAPI entry point and API logic
├── rag.py              # FAISS index, persistence, and document retrieval
├── llm.py              # LLM API interface and JSON parsing
├── prompts.py          # Centralized prompt templates
├── data/               # Source .txt documents
├── index/              # Persisted FAISS index and metadata
└── evals/              # Evaluation dataset and runner script
    ├── eval_set.json
    └── run_evals.py

Install Dependencies

First, create a virtual environment to isolate your project:

python -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate
pip install fastapi uvicorn faiss-cpu numpy pydantic requests python-dotenv

Configure the Environment

Create a .env file in the root directory. We are targeting OpenAI-compatible providers:

OPENAI_API_KEY=your_actual_api_key_here
OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_MODEL=gpt-4o-mini

Important note on compatibility: The code below assumes an OpenAI-style API. If you use a provider that is not compatible, you must change the URL, headers (for example X-API-Key), and the way you extract embeddings and final message content in embed_texts() and call_llm().

How to Build the RAG Layer with FAISS

In rag.py, we handle the “Retriever” part of RAG. This involves turning raw text into mathematical vectors that the computer can compare.

What is FAISS (and What Does It Do)?

FAISS (Facebook AI Similarity Search) is a fast library for vector similarity search. In a RAG system, each chunk of text becomes an embedding vector (a list of floats). FAISS stores those vectors in an index so you can quickly ask:

“Given this question embedding, which document chunks are closest to it?”

In this tutorial, we use IndexFlatIP inner product and normalise vectors with faiss.normalize_L2(...). With normalised vectors, the inner product behaves like cosine similarity, giving us a stable score we can use for a retrieval gate.

Chunking Strategy With Overlap

We’ll use chunking with overlap. If we split a document at exactly 1,000 characters, we might cut a sentence in half, losing its meaning. By using an overlap, for example, 200 characters, we ensure that the end of one chunk and the beginning of the next share context.

Implementation of rag.py

import os
import faiss
import numpy as np
import requests
import json
from typing import List, Dict
from dotenv import load_dotenv

load_dotenv()

INDEX_PATH = "index/faiss.index"
META_PATH = "index/meta.json"

def chunk_text(text: str, size: int = 1000, overlap: int = 200) -> List[str]:
    chunks = []
    step = max(1, size - overlap)
    for i in range(0, len(text), step):
        chunk = text[i : i + size].strip()
        if chunk:
            chunks.append(chunk)
    return chunks

def embed_texts(texts: List[str]) -> np.ndarray:
    # Note: If your provider is not OpenAI-compatible, change this URL and headers
    url = f"{os.getenv('OPENAI_BASE_URL')}/embeddings"
    headers = {"Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}"}
    payload = {"input": texts, "model": "text-embedding-3-small"}

    resp = requests.post(url, headers=headers, json=payload, timeout=30)
    resp.raise_for_status()
    # If your provider uses a different response format, change the line below
    vectors = np.array([item["embedding"] for item in resp.json()["data"]], dtype="float32")
    return vectors

def build_index() -> None:
    all_chunks: List[str] = []
    metadata: List[Dict] = []

    if not os.path.exists("data"):
        os.makedirs("data")
        return

    for file in os.listdir("data"):
        if not file.endswith(".txt"):
            continue

        with open(f"data/{file}", "r", encoding="utf-8") as f:
            text = f.read()

        chunks = chunk_text(text)
        all_chunks.extend(chunks)
        for c in chunks:
            metadata.append({"source": file, "text": c})

    if not all_chunks:
        return

    embeddings = embed_texts(all_chunks)
    faiss.normalize_L2(embeddings)

    dim = embeddings.shape[1]
    index = faiss.IndexFlatIP(dim)
    index.add(embeddings)

    os.makedirs("index", exist_ok=True)
    faiss.write_index(index, INDEX_PATH)

    with open(META_PATH, "w", encoding="utf-8") as f:
        json.dump(metadata, f, ensure_ascii=False)

def load_index():
    if not (os.path.exists(INDEX_PATH) and os.path.exists(META_PATH)):
        raise FileNotFoundError(
            "FAISS index not found. Add .txt files to data/ and run build_index()."
        )

    index = faiss.read_index(INDEX_PATH)
    with open(META_PATH, "r", encoding="utf-8") as f:
        metadata = json.load(f)
    return index, metadata

def retrieve(query: str, k: int = 5) -> List[Dict]:
    index, metadata = load_index()

    q_emb = embed_texts([query])
    faiss.normalize_L2(q_emb)

    scores, ids = index.search(q_emb, k)
    results = []
    for score, idx in zip(scores[0], ids[0]):
        if idx == -1:
            continue
        m = metadata[idx]
        results.append(
            {"score": float(score), "source": m["source"], "text": m["text"], "id": int(idx)}
        )
    return results

How to Add the LLM Call with Structured Output

A major failure point in AI apps is the “chatty” nature of LLMs. If your backend expects a list of sources but the LLM returns conversational filler, your code will crash.

We solve this with structured output: instruct the model to return a strict JSON object, then parse it safely.

Implementation of llm.py

import json
import requests
import os
from typing import Dict, Any

def call_llm(system_prompt: str, user_prompt: str) -> Dict[str, Any]:
    # Note: Change URL/Headers if using a non-OpenAI compatible provider
    url = f"{os.getenv('OPENAI_BASE_URL')}/chat/completions"
    headers = {
        "Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}",
        "Content-Type": "application/json",
    }

    payload = {
        "model": os.getenv("OPENAI_MODEL"),
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
        "response_format": {"type": "json_object"},
        "temperature": 0,
    }

    try:
        resp = requests.post(url, headers=headers, json=payload, timeout=30)
        resp.raise_for_status()
        content = resp.json()["choices"][0]["message"]["content"]

        parsed = json.loads(content)
        parsed.setdefault("answer", "")
        parsed.setdefault("refusal", False)
        parsed.setdefault("confidence", "medium")
        parsed.setdefault("sources", [])
        return parsed

    except (requests.Timeout, requests.ConnectionError):
        return {
            "answer": "The system is temporarily unavailable (network issue). Please try again.",
            "refusal": True,
            "confidence": "low",
            "sources": [],
            "error_type": "network_error",
        }
    except Exception:
        return {
            "answer": "A system error occurred while generating the answer.",
            "refusal": True,
            "confidence": "low",
            "sources": [],
            "error_type": "unknown_error",
        }

How to Add Guardrails: Retrieval Gate and Fallbacks

Guardrails are interceptors. They sit between the user and the model to prevent predictable failures.

The Retrieval Gate: How It Works and How to Add It

In a standard RAG pipeline, the system always calls the LLM. If the user asks an irrelevant question, the retriever will still return the “closest” (but wrong) chunks.

The solution is the retrieval gate:

1. Retrieve top-k chunks and get the top similarity score

2. If the score is below a threshold (for example 0.30), refuse immediately

3. Only call the LLM when retrieval is strong enough to ground the answer

A threshold of 0.30 is a reasonable starting point when using normalised cosine similarity, but you should tune it using evals (next section).

Fallbacks and Why They Matter

Fallbacks ensure that if an API fails or times out, the user gets a helpful message instead of a crash. They also keep your API response shape consistent, which prevents frontend errors and makes logging meaningful.

In this tutorial, fallbacks are implemented inside call_llm() so your FastAPI layer stays simple.

FastAPI App: Creating the /answer Endpoint

The app.py file is the conductor. It ties retrieval, guardrails, prompting, and generation together.

Implementation of app.py

from fastapi import FastAPI
from pydantic import BaseModel
from rag import retrieve
from llm import call_llm
import prompts
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("rag_app")

app = FastAPI(title="Production-Ready RAG")

class QueryRequest(BaseModel):
    question: str

@app.post("/answer")
async def get_answer(req: QueryRequest):
    start_time = time.time()
    question = (req.question or "").strip()

    if not question:
        return {
            "answer": "Please provide a non-empty question.",
            "refusal": True,
            "confidence": "low",
            "sources": [],
            "latency_sec": round(time.time() - start_time, 2),
        }

    # 1) Retrieval
    results = retrieve(question, k=5)
    top_score = results[0]["score"] if results else 0.0

    logger.info("query=%r top_score=%.3f num_results=%d", question, top_score, len(results))

    # 2) Retrieval Gate (Guardrail)
    if top_score < 0.30:
        return {
            "answer": "I do not have documents to answer that question.",
            "refusal": True,
            "confidence": "low",
            "sources": [],
            "latency_sec": round(time.time() - start_time, 2),
            "retrieval": {"top_score": top_score, "k": 5},
        }

    # 3) Augment
    context_text = "\n\n".join([f"Source {r['source']}: {r['text']}" for r in results])
    user_prompt = f"Context:\n{context_text}\n\nQuestion: {question}"

    # 4) Generation with Fallback
    response = call_llm(prompts.SYSTEM_PROMPT, user_prompt)

    # 5) Attach debug metadata
    response["latency_sec"] = round(time.time() - start_time, 2)
    response["retrieval"] = {"top_score": top_score, "k": 5}
    return response

Centralized Prompt – Template: prompts.py

A small but important habit: keep prompts centralised so they’re versionable and easy to evaluate.

Example prompts.py

SYSTEM_PROMPT = """You are a RAG assistant. Use ONLY the provided Context to answer.
If the context does not contain the answer, respond with refusal=true.

Return a valid JSON object with exactly these keys:
- answer: string
- refusal: boolean
- confidence: "low" | "medium" | "high"
- sources: array of strings (source filenames you used)

Do not include any extra keys. Do not include markdown. Do not include commentary."""

How to Add Beginner-Friendly Evals

In AI systems, outputs are probabilistic. This makes testing harder than traditional software. Evals (evaluations) are a set of “golden questions” and “expected behaviours” you run repeatedly to detect regressions.

Instead of “does it output exactly this string,” you test:

• Should the app refuse when the retrieval is weak?

• When it answers, does it include sources?

• Is the behaviour stable across prompt tweaks and model changes?

Step 1: Create evals/eval_set.json

This should contain both positive and negative cases.

[
  {
    "id": "in_scope_01",
    "question": "What is a retrieval gate and why is it important?",
    "expect_refusal": false,
    "notes": "Should explain gating and relate it to hallucination prevention."
  },
  {
    "id": "out_of_scope_01",
    "question": "What is the capital of France?",
    "expect_refusal": true,
    "notes": "If the knowledge base only includes our docs, the app should refuse."
  },
  {
    "id": "edge_01",
    "question": "",
    "expect_refusal": true,
    "notes": "Empty input should not call the LLM."
  }
]

Step 2: Create evals/run_evals.py

This runner calls your API endpoint (end-to-end) and checks expected behaviours.

import json
import requests

API_URL = "http://127.0.0.1:8000/answer"

def run():
    with open("evals/eval_set.json", "r", encoding="utf-8") as f:
        cases = json.load(f)

    passed = 0
    failed = 0

    for case in cases:
        resp = requests.post(API_URL, json={"question": case["question"]}, timeout=60)
        resp.raise_for_status()
        out = resp.json()

        got_refusal = bool(out.get("refusal", False))
        expect_refusal = bool(case["expect_refusal"])

        ok = (got_refusal == expect_refusal)

        # Beginner-friendly: if it answers, sources should exist and be a list
        if not got_refusal:
            ok = ok and isinstance(out.get("sources"), list)

        if ok:
            passed += 1
            print(f"PASS {case['id']}")
        else:
            failed += 1
            print(f"FAIL {case['id']} expected_refusal={expect_refusal} got_refusal={got_refusal}")
            print("Output:", json.dumps(out, indent=2))

    print(f"\nDone. Passed={passed} Failed={failed}")
    if failed:
        raise SystemExit(1)

if __name__ == "__main__":
    run()

How to Use Evals in Practice

Run your server:

uvicorn app:app --reload

In another terminal, run evals:

python evals/run_evals.py

If an eval fails, you have a concrete signal that something changed in retrieval, gating, prompting, or provider behaviour.

What to Improve Next: Realistic Upgrades

Building a reliable RAG app is iterative. Here are realistic next steps:

• Semantic chunking: Break text based on meaning instead of character count.

• Reranking: Use a cross-encoder reranker to reorder the top-k chunks for higher precision.

• Metadata filtering: Filter results by category, date, or department to reduce false positives.

• Better citations: Store chunk IDs and show exactly which chunk(s) the answer came from.

• Observability: Add request IDs, structured logs, and traces so “what happened?” is answerable.

• Async + background indexing: Move index building to a background job and keep the API responsive.

Final Thoughts: Production-Ready Is a Set of Habits

Building an AI application that survives in the real world is about building a system that is predictable, measurable, and safe.

• Retrieval quality is measurable: Use similarity scores to gate your LLM.

• Refusal is a feature: It is better to say “I do not know” than to lie.

• Fallbacks are mandatory: Design for the moment the API goes down.

• Evals prevent regressions: Never deploy a change without running your tests.

About Me

I am Chidozie Managwu, an award-winning AI Product Architect and founder focused on helping global tech talent build real, production-ready skills. I contribute to global AI initiatives as a GAFAI Delegate and lead AI Titans Network, a community for developers learning how to ship AI products.

My work has been recognized with the Global Tech Hero award and featured on platforms like HackerNoon.