OpenRouter Pydantic Code sample

This method lets you call the OpenRouter API, pass in a specific Pydantic schema, and get a response which should conform to the schema that you sent.

def extract_autism_info(custom_input_text=None, custom_input_name=None, custom_model_name=None,
                        custom_system_prompt=None, custom_reasoning_effort=None, custom_force_override=None,
                        vaers_id=None, custom_experiment=None):
    """
    Extract autism information from VAERS report text.
    
    Args:
        custom_input_text (str, optional): Custom input text to use instead of predefined inputs
        custom_input_name (str, optional): Custom name for the input (used in filename)
        custom_model_name (str, optional): Custom model name to use
        custom_system_prompt (str, optional): Custom system prompt to use
        custom_reasoning_effort (str, optional): Custom reasoning effort level
        custom_force_override (bool, optional): Custom force override setting
        vaers_id (str, optional): VAERS ID for the report (used in filename, defaults to 0)
    
    Returns:
        dict: The processed response data
    """
    # Use custom values if provided, otherwise use defaults
    model_name = custom_model_name or MODEL_NAME
    system_prompt = custom_system_prompt or SystemPrompt.spas2
    reasoning_effort = custom_reasoning_effort or REASONING_EFFORT
    force_override = custom_force_override if custom_force_override is not None else True  # Default to True

    # Handle VAERS ID
    vaers_id_str = str(vaers_id) if vaers_id is not None else "0"

    experiment = custom_experiment or EXPERIMENT_NAME
    model_name_str = model_name.replace('/', '-').replace(':', '-')

    # Handle input text
    if custom_input_text:
        input_info = custom_input_text
        input_name = custom_input_name or "custom_input"
    else:
        # Use default input
        selected_input = Inputs.ias190064
        input_info = selected_input.value
        input_name = selected_input.name

    formatted_input = format_input(input_info)

    # Construct filename and directory structure
    model_name_slug = slugify.slugify(model_name_str)
    filename = f'autism_info_{model_name_slug}_{vaers_id_str}.json'
    # Create subfolder based on model name to prevent too many files in one folder
    model_folder = f'json/{experiment}/{model_name_slug}'
    filepath = f'{model_folder}/{filename}'

    # Always try to load existing file first
    existing_data = None
    if os.path.exists(filepath):
        print(f"Found existing file: {filepath}")
        try:
            with open(filepath, 'r') as f:
                existing_data = json.load(f)
            
            # Check if the file already contains valid JSON
            if existing_data.get('contains_valid_json', False):
                print("File contains valid JSON. Skipping API call.")
                return existing_data
            else:
                print("File exists but does not contain valid JSON. Making API call.")
        except Exception as e:
            print(f"Error loading existing file: {e}")
            print("Making API call.")
            existing_data = None
    else:
        print(f"No existing file found. Making API call.")

    if not force_override and existing_data and existing_data.get('contains_valid_json', False):
        print("Using existing data with valid JSON without API call.")
        return existing_data

    # If we reach here, we need to make an API call
    json_schema_text = json.dumps(VAERSReport.model_json_schema(), indent=2)
    prompt = f"""
    {system_prompt} 
    
    Report: {formatted_input}
    
    JSON Schema:
    {json_schema_text}
    """

    before = time.time()

    # Initialize response variables
    api_error = None

    try:
        url = "https://openrouter.ai/api/v1/chat/completions"

        headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
            "HTTP-Referer": "https://botflo.com/", 
            "X-Title": "LLM Accuracy Comparisons for Structured Outputs",
        }

        payload = {
            "model": model_name,
            "messages": [
                {"role": "user", "content": f'{prompt}'}
            ],
            "reasoning": {
                "effort": "high"
            }

        }

        response_full = requests.post(url, headers=headers, data=json.dumps(payload))
        response = response_full.json()
        after = time.time()
        elapsed = after - before
        inner_response_text = response['choices'][0]['message']['content']
        full_response_json = response_full.json()
        now = datetime.datetime.now()
        timestamp = now.strftime('%Y_%m_%d_%H_%M_%S')
        log_file = f'{model_name_str}_{vaers_id_str}_{timestamp}.json'
        with open(f'json/logs/{log_file}', 'w+') as f:
            json.dump(response, f, indent=2)
    except Exception as e:
        after = time.time()
        elapsed = after - before
        api_error = str(e)
        inner_response_text = f"API Error: {api_error}"
        full_response_json = json.dumps({"error": api_error, "timestamp": time.time()})

    # No JSON processing here - just save the raw response

    # Extract token usage from full_response_json
    prompt_tokens = None
    completion_tokens = None
    if not api_error and full_response_json:
        try:
            # full_response_json is already a dict, not a string
            if isinstance(full_response_json, dict) and 'usage' in full_response_json:
                usage = full_response_json['usage']
                prompt_tokens = usage.get('prompt_tokens')
                completion_tokens = usage.get('completion_tokens')
            elif isinstance(full_response_json, str):
                # Handle case where it might be a JSON string
                full_response_data = json.loads(full_response_json)
                if 'usage' in full_response_data:
                    usage = full_response_data['usage']
                    prompt_tokens = usage.get('prompt_tokens')
                    completion_tokens = usage.get('completion_tokens')
        except Exception as e:
            print(f"Error extracting token usage: {e}")

    print('Before new item assignment')
    try:
        new_item = {
            "model_full_name": get_model_full_name(model_name),
            "input_text": formatted_input,
            "system_prompt": system_prompt,
            'json_schema_text': json_schema_text,
            "inner_response_text": inner_response_text,
            "full_response_json": full_response_json,
            "time_elapsed": elapsed,
            "api_error": api_error,
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens
        }
    except Exception as e:
        print(f"Error creating new_item: {e}")
        # In case of exception, create a minimal item with basic fields
        new_item = {
            "model_full_name": "unknown",
            "input_text": formatted_input,
            "system_prompt": system_prompt,
            'json_schema_text': json_schema_text,
            "inner_response_text": inner_response_text,
            "full_response_json": full_response_json,
            "time_elapsed": elapsed,
            "api_error": str(e),
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens
        }

    print('After new item assignment')

    # Create experiment directory and model subfolder if they don't exist
    os.makedirs(f'json/{experiment}', exist_ok=True)
    os.makedirs(model_folder, exist_ok=True)
    with open(filepath, 'w+') as f:
        json.dump(new_item, f, indent=2)

    return new_item

There is a lot more going on in this code than what is the absolute minimum necessary, so let me focus on the main part

This is the prompt which is sent to the API

json_schema_text = json.dumps(VAERSReport.model_json_schema(), indent=2)
prompt = f"""
{system_prompt} 

Report: {formatted_input}

JSON Schema:
{json_schema_text}
"""

There are three parts to this prompt:

The system instruction

The formatted input

The JSON schema in text format

You construct the prompt by concatenating all these three elements and send it to the OpenRouter API

url = "https://openrouter.ai/api/v1/chat/completions"

headers = {
    "Authorization": f"Bearer {api_key}",
    "Content-Type": "application/json",
    "HTTP-Referer": "https://botflo.com/",  # Optional. Site URL for rankings on openrouter.ai.
    "X-Title": "LLM Accuracy Comparisons for Structured Outputs",
}

payload = {
    "model": model_name,
    "messages": [
        {"role": "user", "content": f'{prompt}'}
    ],
    "reasoning": {
        "effort": "high"
    }

}

response_full = requests.post(url, headers=headers, data=json.dumps(payload))
response = response_full.json()

As I have written before I prefer to directly call the OpenRouter API rather than using the Python SDK

One of the most important things you need to know about LLMs when you are trying to extract structured output:

There are many ways for this LLM response to fail

A simple API error (usually LLM provider is down, or some system issue on OpenRouter’s side). This is very rare
The response text does not contain pure JSON (even though that is what we request in the prompt)
The response text does not contain valid JSON even as a substring
There is valid JSON, but it does not conform to the schema we went to the LLM

Handing these errors elegantly (for example, in some cases a simple retry is sufficient to fix the problem) so that you can still get the results that you are expecting is worthy of its own course!