OpenAI LLM.py

llm.py

from openai import AsyncOpenAI
from typing import Type, TypeVar, Optional
import logging
import os
from pydantic import BaseModel

logger = logging.... for logger

T = TypeVar('T', bound=BaseModel)

class ResponseGenerator:
    """Class for generating responses using OpenAI's API"""

    def __init__(self):
        logger.info("🔧 Initializing ResponseGenerator")
        self.client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])
        if not self.client.api_key:
            raise ValueError("OPENAI_API_KEY must be set")
        logger.info("✅ ResponseGenerator initialized successfully")

    async def generate_response(
        self,
        prompt: str,
        user_input: Optional[str] = None,
        response_type: Type[T] = None,
        image_data: Optional[str] = None,
        temperature: float = 0.7
    ) -> Optional[T]:
        model = "gpt-4.1-mini"
        logger.info("🤖 Generating response using model: %s", model)
        logger.debug("📝 Prompt: %s", prompt)

        messages = [{"role": "system", "content": prompt}]
        # Build user message content

        if user_input:
            logger.debug("🗣️ Adding user input to message")
            messages.append({"role": "user", "content": user_input})

        if image_data:
            logger.debug("🖼️ Adding base64 image to message")
            messages.append({
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": "data:image/jpeg;base64,%s" % image_data,
                            "detail": "high"
                        }
                    }
                ]
            })
        try:
            logger.info("🔄 Making API call to OpenAI")
            completion = await self.client.beta.chat.completions.parse(
                model=model,
                messages=messages,
                temperature=temperature,
                response_format=response_type
            )
            message = completion.choices[0].message
            if message.parsed:
                logger.info("✅ Successfully generated response")
                return message.parsed
            else:
                logger.error(
                    "❌ Failed to generate response: %s",
                    message.refusal)
                logger.error("⚠️ Error details: %s", message)
                return None
        except Exception as e:
            logger.error("❌ Error generating response: %s", str(e))
            raise e