Snippet for Structured Output Prompt OpenAI API, Loop over DataFrame

structured_output_prompt.py

import json
from pathlib import Path

import pandas as pd
from dotenv import load_dotenv
from openai import OpenAI
from tqdm import tqdm

# .env
# OPENAI_API_KEY=<API-Key>

load_dotenv()
client = OpenAI()


def run_prompt(input_text: str):
    system_prompt = """
    """

    user_prompt = f"""
    {input_text}
    """

    completion = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
        temperature=0,  # deterministic
        response_format={
            "type": "json_schema",
            "json_schema": {
                "name": "name_of_schema",  # TODO: rename
                "schema": {
                    "type": "object",
                    "properties": {
                        "var1": {"type": "string", "description": "<var1 prompt>"},
                        "var2": {"type": "boolean", "description": "<var2 prompt>"},
                        "reasoning": {
                            "type": "array",
                            "items": {"type": "string"},
                            "description": "Reasoning for answer.",
                        },
                    },
                    "required": [
                        "var1",
                        "var2",
                        "reasoning",
                    ],
                    "additionalProperties": False,
                },
                "strict": True,
            },
        },
    )
    return completion.choices[0].message.content


data_folder = Path("data")
df = pd.read_csv(data_folder / "input" / "input.csv")

results = []
for _, row in tqdm(df.iterrows(), desc="Processing data"):
    try:
        message_content = run_prompt(row["content"])
        results.append(json.loads(message_content))
    except Exception as e:
        print(f"Error processing {row['content']}: {e}")
        results.append({"error": str(e)})


results_df = pd.DataFrame(results)
results_df.to_csv(data_folder / "output" / "output.csv", index=False)