chat_handler.py

import os
import openai
import json
from dotenv import load_dotenv
from functions.tools import available_functions
from functions.tools import client_functions
import asyncio

load_dotenv()
openai.api_key = os.getenv("OPENAI_API_KEY")

# Model config
model = "gpt-3.5-turbo-0125"  
max_tokens = 1000
temperature = 0.9
function_call = "auto"

# get chat history from messages.json
chat_history = []
chat_history_json = json.load(open("messages.json", "r"))
chat_history = chat_history_json.get("chat_history", '')

functions = [
    {
        "name": "control_light",
        "description": "Control smart lights in a room",
        "parameters": {
            "type": "object",
            "properties": {
                "action": {
                    "type": "string",
                    "enum": ["on", "off", "color", "brightness"],
                    "description": "Action to perform on the lights"
                },
                "color": {
                    "type": "object",
                    "properties": {
                        "r": {
                            "type": "integer",
                            "minimum": 0,
                            "maximum": 255,
                            "description": "Red value (0-255)"
                        },
                        "g": {
                            "type": "integer",
                            "minimum": 0,
                            "maximum": 255,
                            "description": "Green value (0-255)"
                        },
                        "b": {
                            "type": "integer",
                            "minimum": 0,
                            "maximum": 255,
                            "description": "Blue value (0-255)"
                        }
                    },
                    "description": "RGB color to set the lights to (optional, required if action is 'color')"
                },
                "brightness": {
                    "type": "integer",
                    "minimum": 0,
                    "maximum": 1000,
                    "description": "Brightness level to set the lights to (optional, required if action is 'brightness')"
                }
            },
            "required": ["action"]
        },
    },
    {
        "name": "get_weather",
        "description": "Get current weather of a city",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "City name"}
            },
            "required": ["city"]
        },
    },
    {
        "name": "music_control",
        "description": "Control music playback",
        "parameters": {
            "type": "object",
            "properties": {
                "action": {
                    "type": "string",
                    "enum": ["play", "stop", "skip", "pause", "resume", "next", "previous", "status"],
                    "description": "Action to perform on music playback"
                },
                "track": {"type": "string", "description": "Track name (optional)"}
            },
            "required": ["action"]
        },
    }
]

def truncate_chat_history(messages, max_turns=10):
    if not messages or messages[0]['role'] != 'system':
        raise ValueError("First message must be a system message.")

    system_message = messages[0]
    body_messages = messages[1:]
    # Get the last 2 * max_turns messages
    truncated = body_messages[-max_turns:]

    return [system_message] + truncated

def add_chat_history(role, content):
    global chat_history
    chat_history.append({"role": role, "content": content})
    # Save updated chat history to messages.json
    with open("messages.json", "w") as f:
        json.dump({"chat_history": chat_history}, f, indent=4)

def get_model_response():
    trimmed = truncate_chat_history(chat_history, max_turns=10)
    response = openai.chat.completions.create(
        model=model,
        messages=trimmed,
        functions=functions,
        function_call=function_call,
        max_tokens=max_tokens,
        temperature=temperature
    )
    return response.choices[0].message

def get_chat_response(user_input):
    # Append the user input to chat history
    add_chat_history("user", user_input)
    response_message = get_model_response()
    print("response: ", response_message)

    # If tool function is required
    if response_message.function_call:
        func_name = response_message.function_call.name
        print(f"Function call detected: {func_name}")
        arguments = json.loads(response_message.function_call.arguments)
        if func_name in available_functions:
            print(f"Executing function: {func_name} with arguments: {arguments}")
            if "color" in arguments:
                if isinstance(arguments["color"], str):
                    try:
                        arguments["color"] = json.loads(arguments["color"])
                    except json.JSONDecodeError:
                        raise ValueError("'color' must be a dict or valid JSON string.")
                elif not isinstance(arguments["color"], dict):
                    raise ValueError("'color' must be a dictionary.")
            if func_name == "get_weather":
                result = asyncio.run(available_functions[func_name](**arguments))
            else:
                result = available_functions[func_name](**arguments)
            add_chat_history("developer", f"Function {func_name} - Result: {result}")
            response_message = get_model_response()
            add_chat_history("assistant", response_message.content)
            return {"tools": False, "response": response_message.content}
        elif func_name in client_functions:
            return {"tools": True, "function": func_name, "arguments": arguments}
        else:
            add_chat_history("developer", f"Unable to fetch or perform the task")
            response_message = get_model_response()
            return {"tools": False, "response": response_message.content}
    else:
        add_chat_history("assistant", response_message.content)
        return {"tools": False, "response": response_message.content}