function_call_to_db.py

from openai import OpenAI
import csv
import os
import time
import json
from typing import Dict, Any

# docker run --pull=always -p 8080:8080 ghcr.io/edgelesssys/privatemode/privatemode-proxy:latest
# PRIVATEMODE_API_KEY=<> uv run --with openai function_call.py

# Sample patient data for demonstration
sample_patient_data = """patient_id,name,age,last_visit_date,diagnosis,doctor
001,John Smith,45,2024-01-15,Hypertension,Dr. Johnson
002,Mary Johnson,32,2024-02-03,Diabetes,Dr. Williams  
003,Robert Brown,67,2024-01-28,Heart Disease,Dr. Davis
004,Lisa Wilson,29,2024-02-10,Asthma,Dr. Miller
005,David Jones,54,2024-01-22,High Cholesterol,Dr. Johnson"""

query="What was John Smith's last visit date and what was his diagnosis?"

# Example output:
# ============================================================
# EXAMPLE 1: CSV Data in System Prompt
# ============================================================
# Response: According to the hospital records, John Smith's last visit date was 2024-01-15, and his diagnosis was Hypertension.
# Request-to-response time: 1.28 seconds

# ============================================================
# EXAMPLE 2: Function Calling with Tools
# ============================================================
# Function called: getLastHospitalVisitByName
# Arguments: {'name': 'John Smith'}
# Function result: 2024-01-15
# Response: John Smith's last hospital visit was on January 15, 2024.
# Request-to-response time: 1.65 seconds

api_key = os.environ.get("PRIVATEMODE_API_KEY")  # insert
api_base = "http://localhost:8080/v1"
csv_file_path = "patient_data.csv"  # insert path to your CSV file
model="latest"

client = OpenAI(
    api_key=api_key,
    base_url=api_base,
)

def read_csv_content(file_path):
    """Read CSV file and return content as string"""
    try:
        with open(file_path, 'r', encoding='utf-8') as file:
            return file.read()
    except FileNotFoundError:
        print(f"CSV file not found at {file_path}. Using sample data.")
        return sample_patient_data

# EXAMPLE 1: CSV data embedded in system prompt
def example_1_system_prompt():
    print("=" * 60)
    print("EXAMPLE 1: CSV Data in System Prompt")
    print("=" * 60)
    
    csv_content = read_csv_content(csv_file_path)
    
    system_prompt = f"""You are a hospital information assistant. You have access to patient data from our hospital records. The data is provided in CSV format below:

{csv_content}

Please help answer questions about patients based on this data. Be precise and only provide information that exists in the records."""

    start_time = time.time()
    
    chat_response = client.chat.completions.create(
        model=model,
        messages=[
            {
                "role": "system",
                "content": system_prompt
            },
            {
                "role": "user",
                "content": "What was John Smith's last visit date and what was his diagnosis?"
            }
        ],
    )
    
    end_time = time.time()
    response_time = end_time - start_time
    
    print("Response:", chat_response.choices[0].message.content)
    print(f"Request-to-response time: {response_time:.2f} seconds")
    print()

def getLastHospitalVisitByName(name: str) -> str:
    """Get the last hospital visit date for a patient by their name
    
    Args:
        name: The full name of the patient
        
    Returns:
        The last visit date as a string, or error message if patient not found
    """
    csv_content = read_csv_content(csv_file_path)
    
    # Parse CSV content
    csv_reader = csv.DictReader(csv_content.splitlines())
    
    for row in csv_reader:
        if row['name'].lower() == name.lower():
            return row['last_visit_date']
    
    return f"Patient '{name}' not found in records"
def example_2_function_calling():
    print("=" * 60)
    print("EXAMPLE 2: Function Calling with Tools")
    print("=" * 60)

    available_functions = {
        "getLastHospitalVisitByName": getLastHospitalVisitByName,
    }
    
    tools = [
        {
            "type": "function",
            "function": {
                "name": "getLastHospitalVisitByName",
                "description": "Get the last hospital visit date for a patient by their name",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "name": {
                            "type": "string",
                            "description": "The full name of the patient"
                        }
                    },
                    "required": ["name"]
                }
            }
        }
    ]
    
    start_time = time.time()
    
    # First API call with function available
    chat_response = client.chat.completions.create(
        model=model,
        messages=[
            {
                "role": "system",
                "content": "You are a hospital information assistant. Use the available functions to help answer questions about patient records."
            },
            {
                "role": "user",
                "content": "When was John Smith's last hospital visit?"
            }
        ],
        tools=tools,
        tool_choice="auto"
    )
    
    response_message = chat_response.choices[0].message
    
    # Check if the model wants to call a function
    if response_message.tool_calls:
        # Execute the function call
        tool_call = response_message.tool_calls[0]
        function_name = tool_call.function.name
        function_args = json.loads(tool_call.function.arguments)
        
        print(f"Function called: {function_name}")
        print(f"Arguments: {function_args}")
        
        if function_name in available_functions:
            function_response = available_functions[function_name](**function_args)
            print(f"Function result: {function_response}")
            
            # Second API call with function result
            chat_response = client.chat.completions.create(
                model=model,
                messages=[
                    {
                        "role": "system",
                        "content": "You are a hospital information assistant. Use the available functions to help answer questions about patient records."
                    },
                    {
                        "role": "user",
                        "content": "When was John Smith's last hospital visit?"
                    },
                    {
                        "role": "assistant",
                        "content": None,
                        "tool_calls": [tool_call]
                    },
                    {
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": function_response
                    }
                ],
                tools=None,
            )
    
    end_time = time.time()
    response_time = end_time - start_time
    
    print("Response:", chat_response.choices[0].message.content)
    print(f"Request-to-response time: {response_time:.2f} seconds")
    print()

if __name__ == "__main__":
    # Create sample CSV file if it doesn't exist
    if not os.path.exists(csv_file_path):
        with open(csv_file_path, 'w', encoding='utf-8') as f:
            f.write(sample_patient_data)
        print(f"Created sample CSV file: {csv_file_path}")
    
    # Run both examples
    example_1_system_prompt()
    example_2_function_calling()