convert_to_openai_object in new API

[mjedmonds]

This function was quite useful for unit testing - was this moved to another location / under a different name? I cannot find similar functionality in the new API.

Thanks in advance.

[RobertCraigie]

How were you using this function? In v1 you can simply construct the object yourself without the need for a helper function, e.g.

from openai.types.chat import ChatCompletionMessage
from openai.types.chat.chat_completion import ChatCompletion, Choice

completion = ChatCompletion(
    id="foo",
    model="gpt-4",
    object="chat.completion",
    choices=[
        Choice(
            finish_reason="stop",
            index=0,
            message=ChatCompletionMessage(
                content="Hello world!",
                role="assistant",
            ),
        )
    ],
    created=int(datetime.now().timestamp()),
)

More details: #398 (comment)

[mjedmonds]

Thanks! That's a much cleaner interface...I started using convert_to_openai_object because creating mocked objects was difficult with the previous API. This is a great improvement.

However I'm still having issues mocking the request; I still get an auth error when mocking the v1/chat/completions endpoint (note the switch to model_dump_json:

openai.AuthenticationError: Error code: 401 - {'error': {'message': 'Incorrect API key provided: OPENAI_A**_KEY. You can find your API key at https://platform.openai.com/account/api-keys.', 'type': 'invalid_request_error', 'param': None, 'code': 'invalid_api_key'}}

Here's the unit test:

import datetime
import unittest
from unittest.mock import patch, MagicMock
import os

import httpx
from respx import MockRouter
from openai import OpenAI
from openai.types.chat import ChatCompletionMessage
from openai.types.chat.chat_completion import ChatCompletion, Choice

from x_lib import gpt_lib
from x_lib.model_config import ModelConfig

os.environ["OPENAI_API_KEY"] = "OPENAI_API_KEY"
client = OpenAI()

class TestRunGptPrompt(unittest.IsolatedAsyncioTestCase):

    @patch('x_lib.gpt_lib.retry')
    async def test_run_gpt_prompt(self, _):
        mock_model_config = MagicMock(spec=ModelConfig)
        mock_model_config.api_env_var = "OPENAI_API_KEY"
        mock_model_config.model_version = "gpt-version"
        mock_model_config.model_temperature = 0.5
        mock_model_config.output_token_limit = None
        mock_model_config.seed = None
        mock_model_config.supports_json_mode = False

        mocked_content = "mocked content"
        completion = ChatCompletion(
            id="foo",
            model=mock_model_config.model_version,
            object="chat.completion",
            choices=[
                Choice(
                    finish_reason="stop",
                    index=0,
                    message=ChatCompletionMessage(
                        content=mocked_content,
                        role="assistant",
                    ),
                )
            ],
            created=int(datetime.datetime.now().timestamp()),
        )

        mock_router = MockRouter()
        mock_router.post("/v1/chat/completions").mock(
            return_value=httpx.Response(200, json=completion.model_dump_json())
        )

        output = await gpt_lib.run_gpt_prompt(
            mock_model_config, messages=[{"role": "system", "content": "test"}])

        self.assertEqual(output, mocked_content)


if __name__ == "__main__":
    unittest.main()

For reference run_gpt_prompt is just a light wrapper around create:

import os
from typing import Dict, List

from tenacity import (
    retry,
    stop_after_attempt,
    wait_random_exponential,
)  # for exponential backoff
import openai

from x_lib.model_config import ModelConfig
from x_lib.logging import LogSeverity, log_exception, log_status

DEFAULT_MAX_TOKENS = 3000


@retry(
    wait=wait_random_exponential(min=1, max=60),
    stop=stop_after_attempt(1),
)
async def run_gpt_prompt(
        model_config: ModelConfig,
        messages: List[Dict[str, str]],
        max_tokens: int = DEFAULT_MAX_TOKENS,
        **kwargs) -> str:
    try:
        client = openai.OpenAI()
        openai.api_key = os.getenv(model_config.api_env_var)
        if not openai.api_key:
            raise ValueError(
                f"OpenAI API key not set. Please set the {model_config.api_env_var} environment variable.")
        if model_config.output_token_limit:
            max_tokens = min(max_tokens, model_config.output_token_limit)
        response = client.chat.completions.create(
            model=model_config.model_version,
            messages=messages,
            temperature=model_config.model_temperature,
            seed=model_config.seed,
            max_tokens=max_tokens,
            response_format={"type": "json_object"} if model_config.supports_json_mode else None,
        )

        choice = response.choices[0]
        if choice.finish_reason == 'length':
            log_status(LogSeverity.WARNING, "OpenAI GPT response exceeded token limits", **kwargs)
        return choice.message.content
    except openai.APITimeoutError as exc:
        log_exception(LogSeverity.WARNING, "OpenAI Timeout error running GPT", exception=exc, **kwargs)
        raise exc
    except openai.RateLimitError as exc:
        log_exception(LogSeverity.WARNING, "OpenAI RateLimitError exceeded running GPT", exception=exc, **kwargs)
        raise exc
    except openai.APIStatusError as exc:
        log_exception(LogSeverity.WARNING, "OpenAI APIStatusError running GPT", exception=exc, **kwargs)
        raise exc
    except openai.APIError as exc:
        log_exception(LogSeverity.ERROR, "OpenAI APIError running GPT", exception=exc, **kwargs)
        raise exc
    except openai.OpenAIError as exc:
        log_exception(LogSeverity.ERROR, "OpenAI OpenAIError running GPT", exception=exc, **kwargs)
        raise exc
    except Exception as exc:  # pylint: disable=broad-except
        log_exception(LogSeverity.ERROR, "Error running GPT", exception=exc, **kwargs)
        raise exc

[RobertCraigie]

It looks like respx isn't overring httpx correctly and you're hitting the live API servers, I think that "override" step is never ran because you're using unittest.

I'm not sure if respx supports unittest properly, you'll likely have to come up with your own wrapper to use it properly if you want to continue using unittest otherwise I'd highly recommend pytest, where it should just work™️

Please ask for any further help using respx in their repo! https://github.com/lundberg/respx

[mjedmonds]

For those reading, this is the working version of the test

import datetime
import unittest
from unittest.mock import patch, MagicMock
import os

import httpx
import respx
from openai import OpenAI
from openai.types.chat import ChatCompletionMessage
from openai.types.chat.chat_completion import ChatCompletion, Choice

from x_lib import gpt_lib
from x_lib.model_config import ModelConfig

os.environ["OPENAI_API_KEY"] = "OPENAI_API_KEY"
client = OpenAI()

class TestRunGptPrompt(unittest.IsolatedAsyncioTestCase):

    @respx.mock
    @patch('owler_lib.gpt_lib.retry')
    async def test_run_gpt_prompt(self, _):
        mock_model_config = MagicMock(spec=ModelConfig)
        mock_model_config.api_env_var = "OPENAI_API_KEY"
        mock_model_config.model_version = "gpt-version"
        mock_model_config.model_temperature = 0.5
        mock_model_config.output_token_limit = None
        mock_model_config.seed = None
        mock_model_config.supports_json_mode = False

        mocked_content = "mocked content"
        completion = ChatCompletion(
            id="foo",
            model=mock_model_config.model_version,
            object="chat.completion",
            choices=[
                Choice(
                    finish_reason="stop",
                    index=0,
                    message=ChatCompletionMessage(
                        content=mocked_content,
                        role="assistant",
                    ),
                )
            ],
            created=int(datetime.datetime.now().timestamp()),
        )


        respx.post("https://api.openai.com/v1/chat/completions").mock(
            return_value=httpx.Response(200, json=completion.dict())
        )

        output = await gpt_lib.run_gpt_prompt(
            mock_model_config, messages=[{"role": "system", "content": "test"}])

        self.assertEqual(output, mocked_content)


if __name__ == "__main__":
    unittest.main()

[elvinagam]

a little more intuitive version of unit test with v2 for OpenAIAssistant class

from config import OPENAI_API_KEY

class OpenAIAssistant:
    def __init__(self, model_name="gpt-4"):
        self.model_name = model_name

    def generate_response(self, conversation):
        client = OpenAI(api_key=OPENAI_API_KEY)

        response = client.chat.completions.create(
            model=self.model_name,
            messages=conversation,
            temperature=0,
            top_p=1,
            frequency_penalty=0,    
            presence_penalty=0
        )
        return response.choices[0].message.content```


**Unittest**


import datetime
import unittest
from unittest.mock import patch, MagicMock

from openai.types.chat import ChatCompletionMessage
from openai.types.chat.chat_completion import ChatCompletion, Choice

from openai_chat import OpenAIAssistant

class TestRunGptPrompt(unittest.TestCase):

    @patch('openai_chat.OpenAIAssistant', autospec=True)
    def test_run_gpt_prompt(self, MockOpenAIAssistant):
        mock_model_config = MagicMock()
        mock_model_config.api_env_var = "OPENAI_API_KEY"
        mock_model_config.model_version = "gpt-4"
        mock_model_config.model_temperature = 0.5
        mock_model_config.output_token_limit = None
        mock_model_config.seed = None
        mock_model_config.supports_json_mode = False

        mocked_content = "This is not a question or a prompt. Could you please provide more information or context?"
        ChatCompletion(
            id="foo",
            model=mock_model_config.model_version,
            object="chat.completion",
            choices=[
                Choice(
                    finish_reason="stop",
                    index=0,
                    message=ChatCompletionMessage(
                        content=mocked_content,
                        role="assistant",
                    ),
                )
            ],
            created=int(datetime.datetime.now().timestamp()),
        )

        # Mock the OpenAIAssistant class and its generate_response method
        mock_instance = MockOpenAIAssistant.return_value
        mock_instance.generate_response.return_value = mocked_content

        # Create an instance of the OpenAIAssistant and call the generate_response method
        assistant = OpenAIAssistant()
        output = assistant.generate_response([{"role": "system", "content": "test"}])

        self.assertEqual(output, mocked_content)


if __name__ == "__main__":
    unittest.main()

[sfc-gh-jcarroll]

Here's an example without any helper stuff, just using the raw API. Includes both basic response and streamed response mocks.

import datetime
from unittest.mock import patch

from openai.types.chat import ChatCompletionMessage
from openai.types.chat.chat_completion import ChatCompletion, Choice
from openai.types.chat.chat_completion_chunk import ChatCompletionChunk, ChoiceDelta
from openai.types.chat.chat_completion_chunk import Choice as StreamChoice


def create_chat_completion(response: str, role: str = "assistant") -> ChatCompletion:
    return ChatCompletion(
        id="foo",
        model="gpt-3.5-turbo",
        object="chat.completion",
        choices=[
            Choice(
                finish_reason="stop",
                index=0,
                message=ChatCompletionMessage(
                    content=response,
                    role=role,
                ),
            )
        ],
        created=int(datetime.datetime.now().timestamp()),
    )


@patch("openai.resources.chat.Completions.create")
def test_chat_completion(openai_create):
    from openai import OpenAI
    EXPECTED_RESPONSE = "The mock is working! ;)"
    openai_create.return_value = create_chat_completion(EXPECTED_RESPONSE)
    client = OpenAI(api_key="sk-...")
    r = client.chat.completions.create(
        messages=[{"role": "user", "content": "Do you know any jokes?"}],
        model="gpt-3.5-turbo",
    )
    response = r.choices[0].message.content
    assert response == EXPECTED_RESPONSE


def create_stream_chat_completion(response: str, role: str = "assistant"):
    for token in response:
        yield ChatCompletionChunk(
            id="foo",
            model="gpt-3.5-turbo",
            object="chat.completion.chunk",
            choices=[
                StreamChoice(
                    index=0,
                    finish_reason=None,
                    delta=ChoiceDelta(
                        content=token,
                        role=role,
                    )
                ),
            ],
            created=int(datetime.datetime.now().timestamp()),
        )


@patch("openai.resources.chat.Completions.create")
def test_stream_chat_completion(openai_create):
    from openai import OpenAI
    EXPECTED_RESPONSE = "The mock is STILL working! ;)"
    openai_create.return_value = create_stream_chat_completion(EXPECTED_RESPONSE)
    client = OpenAI(api_key="sk-...")
    stream = client.chat.completions.create(
        messages=[{"role": "user", "content": "Do you know any jokes?"}],
        model="gpt-3.5-turbo",
        stream=True,
    )

    response = ""
    chunk_count = 0
    for chunk in stream:
        response += (chunk.choices[0].delta.content or "")
        chunk_count += 1
    assert response == EXPECTED_RESPONSE
    assert chunk_count == len(EXPECTED_RESPONSE)

[dahifi]

Dude, thanks, I spent hours yesterday trying to mock one of the endpoints through client.chat and couldn't figure out why it was working until I saw you do this through openai.resources.chat. Thank you so much!

[pamelafox]

If it's helpful for anyone else, I just made mocks for the Async Azure versions, using pytest monkeypatch instead of mock.patch:

https://github.com/pamelafox/chatgpt-backend-fastapi/blob/main/tests/conftest.py