Handling Structured Output with OpenAI Function Calling

· 9 min read ·
Handling Structured Output with OpenAI Function Calling

What You’ll Need

Before diving into implementation, make sure you have the following prerequisites ready:

  • A cloud server hosted on Hetzner VPS or Contabo VPS running Ubuntu 22.04 LTS or 24.04 LTS. You can also host on DigitalOcean as an alternative.
  • A domain name managed via Namecheap if you plan to expose webhooks or backend endpoints.
  • An active OpenAI API key with access to the gpt-4o or gpt-4o-mini models.
  • Python 3.10 or higher installed on your environment.
  • PostgreSQL database server running locally or on a remote host.
  • Optional automation tools such as n8n Cloud or self-hosted n8n for workflow orchestration, or Make.com for standard cloud automation comparisons.

Table of Contents

Understanding Structured Outputs and Function Calling

When building backend services powered by Large Language Models, working with unstructured text output introduces reliability issues. Models can output malformed JSON, omit required keys, or introduce unexpected keys that break downstream code. Early solutions relied on prompt engineering or basic JSON mode, but these approaches provided no absolute runtime guarantee that output matched your database schema.

OpenAI solved this problem by introducing native Structured Outputs alongside Function Calling. By leveraging strict JSON Schema enforcement at the constrained decoding sampler level, the model is physically constrained from generating tokens that violate your specified JSON schema. This guarantees that when the model generates a response, it strictly matches your predefined Pydantic models or JSON schemas.

This reliability transforms LLMs from unpredictable text generators into deterministic data transformation steps inside your backend infrastructure. Whether you are extracting invoice fields to save into application databases or populating decoupled data backends like when deploying self-hosted PocketBase on cloud servers, guaranteed structured payloads eliminate parsing failures across production software stack.

Environment Setup and Pydantic Schema Definition

To implement strict structured extraction, we use Python along with the openai and pydantic SDKs. First, set up a virtual environment and install the required dependencies:

python3 -m venv venv
source venv/bin/activate
pip install openai pydantic psycopg2-binary

We define our schemas using Pydantic. Pydantic classes compile directly down to JSON Schema definitions, which the OpenAI API uses to control model token generation.

Create a file named schema.py and populate it with explicit type declarations:

from enum import Enum
from typing import List, Optional
from pydantic import BaseModel, Field

class Currency(str, Enum):
    USD = "USD"
    EUR = "EUR"
    GBP = "GBP"
    CAD = "CAD"

class InvoiceLineItem(BaseModel):
    description: str = Field(description="Description of the item or service provided")
    quantity: int = Field(description="Number of units billed")
    unit_price: float = Field(description="Price per individual unit")
    line_total: float = Field(description="Calculated total price for this line item")

class InvoiceData(BaseModel):
    vendor_name: str = Field(description="The formal business name of the vendor or supplier")
    invoice_number: str = Field(description="Unique alphanumeric identifier of the invoice")
    invoice_date: str = Field(description="Date the invoice was issued in YYYY-MM-DD format")
    due_date: Optional[str] = Field(default=None, description="Due date for payment in YYYY-MM-DD format")
    currency: Currency = Field(description="Standard 3-letter currency code")
    subtotal: float = Field(description="Total invoice value before tax")
    tax_amount: float = Field(description="Total tax applied to the invoice")
    total_amount: float = Field(description="Final total balance due including taxes")
    line_items: List[InvoiceLineItem] = Field(description="Array of individual line items included in the invoice")

Tutorial 1: Extracting Structured Data with Strict Schema Enforcement

Now we implement our primary extraction script. We will use the OpenAI Beta SDK features, specifically client.beta.chat.completions.parse, which automatically passes the Pydantic class to the API and parses the returned JSON string directly into a validated Python object.

Create a file named extractor.py:

import os
import json
from openai import OpenAI
from schema import InvoiceData

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

unstructured_invoice_text = """
INVOICE #INV-90214
Issued By: Acme Cloud Services LLC
Date: 2026-03-15
Payment Due: 2026-04-15

Billed Items:
1. Virtual Private Server Hosting (2 months) - Qty: 2 - Price per unit: $45.00 - Total: $90.00
2. Managed Domain Name Registration - Qty: 1 - Price per unit: $15.00 - Total: $15.00
3. SSL Certificate Setup - Qty: 1 - Price per unit: $25.00 - Total: $25.00

Subtotal: $130.00
Sales Tax (8%): $10.40
Total Amount Due: $140.40
Currency: USD
"""

def extract_invoice_payload(text_content: str) -> InvoiceData:
    response = client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "You are an expert document extraction engine. Extract all detailed billing fields from the raw unstructured text provided."
            },
            {
                "role": "user",
                "content": text_content
            }
        ],
        response_format=InvoiceData,
    )

    parsed_object = response.choices[0].message.parsed
    if parsed_object is None:
        refusal = response.choices[0].message.refusal
        raise ValueError(f"Model refused to process request: {refusal}")

    return parsed_object

if __name__ == "__main__":
    extracted_data = extract_invoice_payload(unstructured_invoice_text)
    print("Successfully Extracted and Validated Schema:")
    print(json.dumps(extracted_data.model_dump(), indent=2))

Execute the script to observe guaranteed structural output:

export OPENAI_API_KEY="your-actual-api-key"
python3 extractor.py

💡 Fast-Track Your Project: Don’t want to configure this yourself? I build custom n8n pipelines and bots. Message me with code SYS3-HUGO.

Tutorial 2: Inserting Structured Payload into PostgreSQL

Once data is parsed into strongly-typed objects, saving it to relational databases becomes straightforward and safe from SQL injection or schema drift. If you want high availability or data redundancy across multiple nodes, check out our operational guide on how to set up PostgreSQL logical replication.

Create a file named db_ingest.py that takes our validated Pydantic model and performs an atomic transaction inserting both the main invoice record and its associated child line items into a PostgreSQL database.

import os
import psycopg2
from psycopg2.extras import execute_batch
from schema import InvoiceData
from extractor import extract_invoice_payload, unstructured_invoice_text

DB_HOST = os.getenv("DB_HOST", "localhost")
DB_NAME = os.getenv("DB_NAME", "billing_db")
DB_USER = os.getenv("DB_USER", "postgres")
DB_PASS = os.getenv("DB_PASS", "postgres_password")
DB_PORT = os.getenv("DB_PORT", "5432")

def initialize_database_schema(conn):
    with conn.cursor() as cur:
        cur.execute("""
            CREATE TABLE IF NOT EXISTS invoices (
                id SERIAL PRIMARY KEY,
                vendor_name VARCHAR(255) NOT NULL,
                invoice_number VARCHAR(100) UNIQUE NOT NULL,
                invoice_date DATE NOT NULL,
                due_date DATE,
                currency VARCHAR(10) NOT NULL,
                subtotal NUMERIC(12, 2) NOT NULL,
                tax_amount NUMERIC(12, 2) NOT NULL,
                total_amount NUMERIC(12, 2) NOT NULL,
                created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
            );
        """)
        
        cur.execute("""
            CREATE TABLE IF NOT EXISTS invoice_line_items (
                id SERIAL PRIMARY KEY,
                invoice_number VARCHAR(100) REFERENCES invoices(invoice_number) ON DELETE CASCADE,
                description TEXT NOT NULL,
                quantity INT NOT NULL,
                unit_price NUMERIC(12, 2) NOT NULL,
                line_total NUMERIC(12, 2) NOT NULL
            );
        """)
    conn.commit()

def save_invoice_to_db(invoice: InvoiceData):
    conn = psycopg2.connect(
        host=DB_HOST,
        dbname=DB_NAME,
        user=DB_USER,
        password=DB_PASS,
        port=DB_PORT
    )
    
    try:
        initialize_database_schema(conn)
        
        with conn.cursor() as cur:
            insert_invoice_sql = """
                INSERT INTO invoices (
                    vendor_name, invoice_number, invoice_date, due_date, currency, subtotal, tax_amount, total_amount
                ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
                ON CONFLICT (invoice_number) DO UPDATE SET
                    subtotal = EXCLUDED.subtotal,
                    tax_amount = EXCLUDED.tax_amount,
                    total_amount = EXCLUDED.total_amount;
            """
            
            cur.execute(insert_invoice_sql, (
                invoice.vendor_name,
                invoice.invoice_number,
                invoice.invoice_date,
                invoice.due_date,
                invoice.currency.value,
                invoice.subtotal,
                invoice.tax_amount,
                invoice.total_amount
            ))

            cur.execute("DELETE FROM invoice_line_items WHERE invoice_number = %s;", (invoice.invoice_number,))

            insert_items_sql = """
                INSERT INTO invoice_line_items (
                    invoice_number, description, quantity, unit_price, line_total
                ) VALUES (%s, %s, %s, %s, %s);
            """
            
            items_tuples = [
                (
                    invoice.invoice_number,
                    item.description,
                    item.quantity,
                    item.unit_price,
                    item.line_total
                )
                for item in invoice.line_items
            ]
            
            execute_batch(cur, insert_items_sql, items_tuples)

        conn.commit()
        print(f"Successfully saved invoice {invoice.invoice_number} with {len(invoice.line_items)} line items to database.")

    except Exception as e:
        conn.rollback()
        print(f"Database insertion error: {str(e)}")
        raise e
    finally:
        conn.close()

if __name__ == "__main__":
    invoice_obj = extract_invoice_payload(unstructured_invoice_text)
    save_invoice_to_db(invoice_obj)

Tutorial 3: Scheduling Production Pipelines with Systemd

To run automated document processing on set schedules without relying on external webhooks, you can host your execution script directly on your server and schedule it using systemd. Learn step by step how to schedule Python scripts with systemd to handle long-running backend processes reliably.

Let us build a production wrapper named run_pipeline.py:

import sys
import logging
from extractor import extract_invoice_payload
from db_ingest import save_invoice_to_db

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)]
)

def run_batch_job():
    logging.info("Starting invoice extraction pipeline job...")
    
    sample_text = """
    INVOICE #INV-88301
    Issued By: Cloud Hosting Provider Inc.
    Date: 2026-03-20
    Payment Due: 2026-04-20

    Items:
    1. Dedicated Bare Metal Server - Qty: 1 - Unit Price: $180.00 - Total: $180.00
    2. Block Storage 1TB - Qty: 2 - Unit Price: $20.00 - Total: $40.00

    Subtotal: $220.00
    Tax: $0.00
    Total: $220.00
    Currency: USD
    """
    
    try:
        invoice = extract_invoice_payload(sample_text)
        logging.info(f"Successfully processed invoice {invoice.invoice_number}")
        save_invoice_to_db(invoice)
        logging.info("Pipeline job executed successfully.")
    except Exception as e:
        logging.error(f"Pipeline job failed: {str(e)}")
        sys.exit(1)

if __name__ == "__main__":
    run_batch_job()

Next, create the systemd service file at /etc/systemd/system/invoice_processor.service:

[Unit]
Description=OpenAI Invoice Processing Pipeline Worker
After=network.target postgresql.service

[Service]
Type=oneshot
User=ubuntu
WorkingDirectory=/home/ubuntu/invoice_pipeline
Environment="PATH=/home/ubuntu/invoice_pipeline/venv/bin"
Environment="OPENAI_API_KEY=your-actual-api-key"
Environment="DB_HOST=localhost"
Environment="DB_NAME=billing_db"
Environment="DB_USER=postgres"
Environment="DB_PASS=postgres_password"
ExecStart=/home/ubuntu/invoice_pipeline/venv/bin/python3 /home/ubuntu/invoice_pipeline/run_pipeline.py

[Install]
WantedBy=multi-user.target

Create the companion timer file at /etc/systemd/system/invoice_processor.timer:

[Unit]
Description=Run Invoice Processing Pipeline Every 15 Minutes

[Timer]
OnCalendar=*:0/15
Persistent=true

[Install]
WantedBy=timers.target

Enable and activate the timer:

sudo systemctl daemon-reload
sudo systemctl enable --now invoice_processor.timer
sudo systemctl status invoice_processor.timer

Handling Schema Refusals and Runtime Validation Exceptions

Although OpenAI Structured Outputs guarantee schema compliance when the model responds, there are two exception types you must handle in production backend applications:

  1. Model Refusals: The model may refuse to answer if the input text contains safety violations or instructions that contradict safety parameters.
  2. Schema Definition Constraints: Not all JSON schema features are supported in strict mode. For example, all fields in object definitions must be marked as required, and optional fields must be configured using explicit union types with null or Pydantic’s Optional[].

Here is a robust module demonstrating proper error management and refusal checking:

import os
from typing import Optional
from pydantic import BaseModel, Field
from openai import OpenAI, APIError

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

class SimpleOutputSchema(BaseModel):
    summary: str = Field(description="A concise summary of the provided text")
    action_required: bool = Field(description="Indicates whether immediate operator attention is needed")

def safe_parse_text(text: str) -> Optional[SimpleOutputSchema]:
    try:
        response = client.beta.chat.completions.parse(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "Extract and summarize incoming operational logs."},
                {"role": "user", "content": text}
            ],
            response_format=SimpleOutputSchema,
        )
        
        message = response.choices[0].message
        
        if message.refusal:
            print(f"Safety Refusal triggered by model: {message.refusal}")
            return None
            
        return message.parsed

    except APIError as api_err:
        print(f"OpenAI API transport level error encountered: {api_err}")
        return None
    except Exception as general_err:
        print(f"Unexpected operational failure: {general_err}")
        return None

if __name__ == "__main__":
    valid_log = "CRITICAL: Database server disk usage exceeded 95 percent on cluster node db-01."
    result = safe_parse_text(valid_log)
    if result:
        print(f"Summary: {result.summary}")
        print(f"Action Required: {result.action_required}")

Getting Started

To implement production grade structured processing pipelines on your own infrastructure:

  1. Spin up a cloud server on Hetzner VPS or Contabo VPS running Ubuntu 22.04 LTS. You can also use DigitalOcean if preferred.
  2. Provision domain infrastructure via Namecheap if building exposed API endpoints.
  3. Configure your automated workflow triggers using n8n Cloud or write background Python execution scripts managed directly by systemd.

Outsource Your Automation

Don’t have time? I build production n8n workflows, WhatsApp bots, and fully automated YouTube Shorts pipelines. Hire me on Fiverr, mention SYS3-HUGO for priority. Or DM at chasebot.online.

Want to automate this yourself?

Start with n8n Cloud (free tier available) or self-host on a Hetzner VPS for full control.

Want this engine running on your own VPS?

This blog publishes itself — daily, unattended, on free API tiers. The full engine, Hugo theme, and setup guide are available as System 3.

Get System 3

Related Guides

system online