Repository: deepset-ai/haystack
Pull request: feat: Add ChatGPT PromptNode layer
### Related Issues
- no relevant issue

### Proposed Changes:
Adds ChatGPTInvocationLayer as PromptModelInvocationLayer implementation for the support of [ChatGPT](https://platform.openai.com/docs/guides/chat) model

Note that `gpt-3.5-turbo` model requires special payload messages to be specified. Having said that PromptNode should not store any state (list of user messages) but this should be handled by the higher abstraction layer using PromptNode. The task of this layer is to integrate chatgpt API support and enable PromptNode to send/receive requests from the chat completions `gpt-3.5-turbo` model. 

### How did you test it?
No tests added yet; unit tests and documentation are coming soon


### Notes for the reviewer
This is a draft proposal, do not yet integrate.


### Checklist
- [ ] I have read the [contributors guidelines](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md) and the [code of conduct](https://github.com/deepset-ai/haystack/blob/main/code_of_conduct.txt)
- [ ] I have updated the related issue with new insights and changes
- [ ] I added tests that demonstrate the correct behavior of the change
- [ ] I've used one of the [conventional commit types](https://www.conventionalcommits.org/en/v1.0.0/) for my PR title: `fix:`, `feat:`, `build:`, `chore:`, `ci:`, `docs:`, `style:`, `refactor:`, `perf:`, `test:`.
- [ ] I documented my code
- [ ] I ran [pre-commit hooks](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md#installation) and fixed any issue


Problem statement:
add support for gpt-3.5
**Is your feature request related to a problem? Please describe.**

No.

**Describe the solution you'd like**

OpenAI recently launched the API for GPT-3.5, but currently the `OpenAIInvocationLayer` only supports the GPT-3 models (i.e. "ada", "babbage", "davinci", "curie"), not GPT-3.5 (gpt-3.5-turbo). GPT-3.5 is relatively cheaper and more powerful, so it would be very useful to add support for it in Haystack. Therefore, I hope to add support for GPT-3.5 in Haystack.

**Describe alternatives you've considered**

**Additional context**

Hints from issue discussion:
Thanks for raising this point @observerw ; already in the works. Should be available next week.

Commit message: align method name

Patch to review:
diff --git a/haystack/nodes/answer_generator/openai.py b/haystack/nodes/answer_generator/openai.py
index ad44c03ffe..8ae2fd9325 100644
--- a/haystack/nodes/answer_generator/openai.py
+++ b/haystack/nodes/answer_generator/openai.py
@@ -11,7 +11,7 @@
     openai_request,
     count_openai_tokens,
     _openai_text_completion_tokenization_details,
-    _check_openai_text_completion_answers,
+    _check_openai_finish_reason,
 )
 
 logger = logging.getLogger(__name__)
@@ -229,7 +229,7 @@ def predict(
             headers["Authorization"] = f"Bearer {self.api_key}"
 
         res = openai_request(url=url, headers=headers, payload=payload, timeout=timeout)
-        _check_openai_text_completion_answers(result=res, payload=payload)
+        _check_openai_finish_reason(result=res, payload=payload)
         generated_answers = [ans["text"] for ans in res["choices"]]
         answers = self._create_answers(generated_answers, input_docs)
         result = {"query": query, "answers": answers}
diff --git a/haystack/nodes/prompt/prompt_node.py b/haystack/nodes/prompt/prompt_node.py
index bec4552cb2..64255c3898 100644
--- a/haystack/nodes/prompt/prompt_node.py
+++ b/haystack/nodes/prompt/prompt_node.py
@@ -3,7 +3,7 @@
 import re
 from abc import ABC
 from string import Template
-from typing import Dict, List, Optional, Tuple, Union, Any, Iterator, Type
+from typing import Dict, List, Optional, Tuple, Union, Any, Iterator, Type, overload
 
 import torch
 
@@ -245,7 +245,7 @@ def create_invocation_layer(
             "PromptModelInvocationLayer."
         )
 
-    def invoke(self, prompt: Union[str, List[str]], **kwargs) -> List[str]:
+    def invoke(self, prompt: Union[str, List[str], List[Dict[str, str]]], **kwargs) -> List[str]:
         """
         It takes in a prompt, and returns a list of responses using the underlying invocation layer.
 
@@ -256,7 +256,15 @@ def invoke(self, prompt: Union[str, List[str]], **kwargs) -> List[str]:
         output = self.model_invocation_layer.invoke(prompt=prompt, **kwargs)
         return output
 
+    @overload
     def _ensure_token_limit(self, prompt: str) -> str:
+        ...
+
+    @overload
+    def _ensure_token_limit(self, prompt: List[Dict[str, str]]) -> List[Dict[str, str]]:
+        ...
+
+    def _ensure_token_limit(self, prompt: Union[str, List[Dict[str, str]]]) -> Union[str, List[Dict[str, str]]]:
         """Ensure that length of the prompt and answer is within the maximum token length of the PromptModel.
 
         :param prompt: Prompt text to be sent to the generative model.
@@ -487,7 +495,7 @@ def prompt(self, prompt_template: Optional[Union[str, PromptTemplate]], *args, *
         send_event("PromptNode.prompt()", event_properties={"template": str(prompt_template)})
         results = []
         # we pop the prompt_collector kwarg to avoid passing it to the model
-        prompt_collector: List[str] = kwargs.pop("prompt_collector", [])
+        prompt_collector: List[Union[str, List[Dict[str, str]]]] = kwargs.pop("prompt_collector", [])
         if isinstance(prompt_template, str) and not self.is_supported_template(prompt_template):
             raise ValueError(
                 f"{prompt_template} not supported, please select one of: {self.get_prompt_template_names()} "
diff --git a/haystack/nodes/prompt/providers.py b/haystack/nodes/prompt/providers.py
index d945da0c1a..8dfcf0f721 100644
--- a/haystack/nodes/prompt/providers.py
+++ b/haystack/nodes/prompt/providers.py
@@ -1,7 +1,8 @@
 import json
 import logging
 from abc import abstractmethod, ABC
-from typing import Dict, List, Optional, Union, Type
+from typing import Dict, List, Optional, Union, Type, cast
+
 
 import sseclient
 import torch
@@ -22,8 +23,9 @@
     openai_request,
     _openai_text_completion_tokenization_details,
     load_openai_tokenizer,
-    _check_openai_text_completion_answers,
+    _check_openai_finish_reason,
     count_openai_tokens,
+    count_openai_tokens_messages,
 )
 
 logger = logging.getLogger(__name__)
@@ -113,7 +115,7 @@ def supports(cls, model_name_or_path: str, **kwargs) -> bool:
         return False
 
     @abstractmethod
-    def _ensure_token_limit(self, prompt: str) -> str:
+    def _ensure_token_limit(self, prompt: Union[str, List[Dict[str, str]]]) -> Union[str, List[Dict[str, str]]]:
         """Ensure that length of the prompt and answer is within the maximum token length of the PromptModel.
 
         :param prompt: Prompt text to be sent to the generative model.
@@ -273,7 +275,7 @@ def invoke(self, *args, **kwargs):
                     generated_texts[idx] = generated_texts[idx].replace(stop_word, "").strip()
         return generated_texts
 
-    def _ensure_token_limit(self, prompt: str) -> str:
+    def _ensure_token_limit(self, prompt: Union[str, List[Dict[str, str]]]) -> Union[str, List[Dict[str, str]]]:
         """Ensure that the length of the prompt and answer is within the max tokens limit of the model.
         If needed, truncate the prompt text so that it fits within the limit.
 
@@ -445,7 +447,7 @@ def invoke(self, *args, **kwargs):
         }
         if not stream:
             res = openai_request(url=self.url, headers=self.headers, payload=payload)
-            _check_openai_text_completion_answers(result=res, payload=payload)
+            _check_openai_finish_reason(result=res, payload=payload)
             responses = [ans["text"].strip() for ans in res["choices"]]
             return responses
         else:
@@ -466,13 +468,13 @@ def invoke(self, *args, **kwargs):
                 client.close()
             return ["".join(tokens)]  # return a list of strings just like non-streaming
 
-    def _ensure_token_limit(self, prompt: str) -> str:
+    def _ensure_token_limit(self, prompt: Union[str, List[Dict[str, str]]]) -> Union[str, List[Dict[str, str]]]:
         """Ensure that the length of the prompt and answer is within the max tokens limit of the model.
         If needed, truncate the prompt text so that it fits within the limit.
 
         :param prompt: Prompt text to be sent to the generative model.
         """
-        n_prompt_tokens = count_openai_tokens(prompt, self._tokenizer)
+        n_prompt_tokens = count_openai_tokens(cast(str, prompt), self._tokenizer)
         n_answer_tokens = self.max_length
         if (n_prompt_tokens + n_answer_tokens) <= self.max_tokens_limit:
             return prompt
@@ -545,3 +547,111 @@ def supports(cls, model_name_or_path: str, **kwargs) -> bool:
         return (
             valid_model and kwargs.get("azure_base_url") is not None and kwargs.get("azure_deployment_name") is not None
         )
+
+
+class ChatGPTInvocationLayer(OpenAIInvocationLayer):
+    """
+    ChatGPT Invocation Layer
+
+    PromptModelInvocationLayer implementation for OpenAI's GPT-3 ChatGPT API. Invocations are made using REST API.
+    See [OpenAI ChatGPT API](https://platform.openai.com/docs/guides/chat) for more details.
+
+    Note: kwargs other than init parameter names are ignored to enable reflective construction of the class
+    as many variants of PromptModelInvocationLayer are possible and they may have different parameters.
+    """
+
+    def __init__(
+        self, api_key: str, model_name_or_path: str = "gpt-3.5-turbo", max_length: Optional[int] = 500, **kwargs
+    ):
+        super().__init__(api_key, model_name_or_path, max_length, **kwargs)
+
+    def invoke(self, *args, **kwargs):
+        """
+        It takes in either a prompt or a list of messages and returns a list of responses, using a REST invocation.
+
+        :return: A list of generated responses.
+
+        Note: Only kwargs relevant to OpenAI are passed to OpenAI rest API. Others kwargs are ignored.
+        For more details, see [OpenAI ChatGPT API reference](https://platform.openai.com/docs/api-reference/chat).
+        """
+        prompt = kwargs.get("prompt", None)
+
+        if isinstance(prompt, str):
+            messages = [{"role": "user", "content": prompt}]
+        elif isinstance(prompt, list) and len(prompt) > 0 and isinstance(prompt[0], dict):
+            messages = prompt
+        else:
+            raise ValueError(
+                f"The prompt format is different than what the model expects."
+                f"The model {self.model_name_or_path} requires either a string or messages in the ChatML format."
+                f"For more details, see this [GitHub discussion](https://github.com/openai/openai-python/blob/main/chatml.md)."
+            )
+
+        kwargs_with_defaults = self.model_input_kwargs
+        if kwargs:
+            # we use keyword stop_words but OpenAI uses stop
+            if "stop_words" in kwargs:
+                kwargs["stop"] = kwargs.pop("stop_words")
+            if "top_k" in kwargs:
+                top_k = kwargs.pop("top_k")
+                kwargs["n"] = top_k
+                kwargs["best_of"] = top_k
+            kwargs_with_defaults.update(kwargs)
+        payload = {
+            "model": self.model_name_or_path,
+            "messages": messages,
+            "max_tokens": kwargs_with_defaults.get("max_tokens", self.max_length),
+            "temperature": kwargs_with_defaults.get("temperature", 0.7),
+            "top_p": kwargs_with_defaults.get("top_p", 1),
+            "n": kwargs_with_defaults.get("n", 1),
+            "stream": False,  # no support for streaming
+            "stop": kwargs_with_defaults.get("stop", None),
+            "presence_penalty": kwargs_with_defaults.get("presence_penalty", 0),
+            "frequency_penalty": kwargs_with_defaults.get("frequency_penalty", 0),
+            "logit_bias": kwargs_with_defaults.get("logit_bias", {}),
+        }
+        response = openai_request(url=self.url, headers=self.headers, payload=payload)
+        _check_openai_finish_reason(result=response, payload=payload)
+        assistant_response = [choice["message"]["content"].strip() for choice in response["choices"]]
+
+        # Although ChatGPT generates text until stop words are encountered, unfortunately it includes the stop word
+        # We want to exclude it to be consistent with other invocation layers
+        if "stop" in kwargs_with_defaults and kwargs_with_defaults["stop"] is not None:
+            stop_words = kwargs_with_defaults["stop"]
+            for idx, _ in enumerate(assistant_response):
+                for stop_word in stop_words:
+                    assistant_response[idx] = assistant_response[idx].replace(stop_word, "").strip()
+
+        return assistant_response
+
+    def _ensure_token_limit(self, prompt: Union[str, List[Dict[str, str]]]) -> Union[str, List[Dict[str, str]]]:
+        """Make sure the length of the prompt and answer is within the max tokens limit of the model.
+        If needed, truncate the prompt text so that it fits within the limit.
+
+        :param prompt: Prompt text to be sent to the generative model.
+        """
+        if isinstance(prompt, str):
+            messages = [{"role": "user", "content": prompt}]
+        elif isinstance(prompt, list) and len(prompt) > 0 and isinstance(prompt[0], dict):
+            messages = prompt
+
+        n_prompt_tokens = count_openai_tokens_messages(messages, self._tokenizer)
+        n_answer_tokens = self.max_length
+        if (n_prompt_tokens + n_answer_tokens) <= self.max_tokens_limit:
+            return prompt
+
+        # TODO: support truncation as in _ensure_token_limit methods for other invocation layers
+        raise ValueError(
+            f"The prompt or the messages are too long ({n_prompt_tokens} tokens). "
+            f"The length of the prompt or messages and the answer ({n_answer_tokens} tokens) should be within the max token limit ({self.max_tokens_limit} tokens). "
+            f"Reduce the length of the prompt or messages."
+        )
+
+    @property
+    def url(self) -> str:
+        return "https://api.openai.com/v1/chat/completions"
+
+    @classmethod
+    def supports(cls, model_name_or_path: str, **kwargs) -> bool:
+        valid_model = any(m for m in ["gpt-3.5-turbo"] if m in model_name_or_path)
+        return valid_model
diff --git a/haystack/utils/openai_utils.py b/haystack/utils/openai_utils.py
index 939f58e082..6bbd479fd2 100644
--- a/haystack/utils/openai_utils.py
+++ b/haystack/utils/openai_utils.py
@@ -4,7 +4,7 @@
 import platform
 import sys
 import json
-from typing import Dict, Union, Tuple, Optional
+from typing import Dict, Union, Tuple, Optional, List
 import requests
 
 from transformers import GPT2TokenizerFast
@@ -69,6 +69,28 @@ def count_openai_tokens(text: str, tokenizer) -> int:
         return len(tokenizer.tokenize(text))
 
 
+def count_openai_tokens_messages(messages: List[Dict[str, str]], tokenizer) -> int:
+    """Count the number of tokens in `messages` based on the OpenAI `tokenizer` provided.
+
+    :param messages: The messages to be tokenized.
+    :param tokenizer: An OpenAI tokenizer.
+    """
+    # adapted from https://platform.openai.com/docs/guides/chat/introduction
+    # should be kept up to date
+    num_tokens = 0
+    for message in messages:
+        num_tokens += 4  # every message follows <im_start>{role/name}\n{content}<im_end>\n
+        for key, value in message.items():
+            if USE_TIKTOKEN:
+                num_tokens += len(tokenizer.encode(value))
+            else:
+                num_tokens += len(tokenizer.tokenize(value))
+            if key == "name":  # if there's a name, the role is omitted
+                num_tokens += -1  # role is always required and always 1 token
+    num_tokens += 2  # every reply is primed with <im_start>assistant
+    return num_tokens
+
+
 def _openai_text_completion_tokenization_details(model_name: str):
     """Return the tokenizer name and max tokens limit for a given OpenAI `model_name`.
 
@@ -79,6 +101,10 @@ def _openai_text_completion_tokenization_details(model_name: str):
         max_tokens_limit = 4000
         if USE_TIKTOKEN:
             tokenizer_name = MODEL_TO_ENCODING.get(model_name, "p50k_base")
+    elif "gpt-3.5-turbo" in model_name:
+        max_tokens_limit = 4096
+        if USE_TIKTOKEN:
+            tokenizer_name = MODEL_TO_ENCODING.get(model_name, "cl100k_base")
     else:
         max_tokens_limit = 2048
     return tokenizer_name, max_tokens_limit
@@ -125,9 +151,9 @@ def openai_request(
         return response
 
 
-def _check_openai_text_completion_answers(result: Dict, payload: Dict) -> None:
-    """Check the `finish_reason` the answers returned by OpenAI completions endpoint. If the `finish_reason` is `length`,
-    log a warning to the user.
+def _check_openai_finish_reason(result: Dict, payload: Dict) -> None:
+    """Check the `finish_reason` the answers returned by OpenAI completions endpoint.
+    If the `finish_reason` is `length` or `content_filter`, log a warning to the user.
 
     :param result: The result returned from the OpenAI API.
     :param payload: The payload sent to the OpenAI API.
@@ -140,3 +166,13 @@ def _check_openai_text_completion_answers(result: Dict, payload: Dict) -> None:
             number_of_truncated_completions,
             payload["n"],
         )
+
+    number_of_content_filtered_completions = sum(
+        1 for ans in result["choices"] if ans["finish_reason"] == "content_filter"
+    )
+    if number_of_content_filtered_completions > 0:
+        logger.warning(
+            "%s out of the %s completions have omitted content due to a flag from OpenAI content filters.",
+            number_of_truncated_completions,
+            payload["n"],
+        )
diff --git a/test/conftest.py b/test/conftest.py
index 2808d75271..b0c4cf7a77 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -1000,6 +1000,14 @@ def prompt_model(request):
         return PromptModel("google/flan-t5-base", devices=["cpu"])
 
 
+@pytest.fixture
+def chatgpt_prompt_model():
+    api_key = os.environ.get("OPENAI_API_KEY", "KEY_NOT_FOUND")
+    if api_key is None or api_key == "":
+        api_key = "KEY_NOT_FOUND"
+    return PromptModel("gpt-3.5-turbo", api_key=api_key)
+
+
 @pytest.fixture
 def azure_conf():
     return haystack_azure_conf()
diff --git a/test/prompt/test_prompt_node.py b/test/prompt/test_prompt_node.py
index b23ec9cd77..50cf234637 100644
--- a/test/prompt/test_prompt_node.py
+++ b/test/prompt/test_prompt_node.py
@@ -973,3 +973,48 @@ def test_simple_pipeline_batch_query_multiple_doc_list(self, prompt_model):
 def test_HFLocalInvocationLayer_supports():
     assert HFLocalInvocationLayer.supports("philschmid/flan-t5-base-samsum")
     assert HFLocalInvocationLayer.supports("bigscience/T0_3B")
+
+
+@pytest.mark.integration
+def test_chatgpt_direct_prompting(chatgpt_prompt_model):
+    skip_test_for_invalid_key(chatgpt_prompt_model)
+    pn = PromptNode(chatgpt_prompt_model)
+    result = pn("Hey, I need some Python help. When should I use list comprehension?")
+    assert len(result) == 1 and all(w in result[0] for w in ["comprehension", "list"])
+
+
+@pytest.mark.integration
+def test_chatgpt_direct_prompting_w_messages(chatgpt_prompt_model):
+    skip_test_for_invalid_key(chatgpt_prompt_model)
+    pn = PromptNode(chatgpt_prompt_model)
+
+    messages = [
+        {"role": "system", "content": "You are a helpful assistant."},
+        {"role": "user", "content": "Who won the world series in 2020?"},
+        {"role": "assistant", "content": "The Los Angeles Dodgers won the World Series in 2020."},
+        {"role": "user", "content": "Where was it played?"},
+    ]
+
+    result = pn(messages)
+    assert len(result) == 1 and all(w in result[0].casefold() for w in ["arlington", "texas"])
+
+
+@pytest.mark.integration
+@pytest.mark.skipif(
+    not os.environ.get("OPENAI_API_KEY", None),
+    reason="No OpenAI API key provided. Please export an env var called OPENAI_API_KEY containing the OpenAI API key to run this test.",
+)
+def test_chatgpt_promptnode():
+    pn = PromptNode(model_name_or_path="gpt-3.5-turbo", api_key=os.environ.get("OPENAI_API_KEY", None))
+
+    result = pn("Hey, I need some Python help. When should I use list comprehension?")
+    assert len(result) == 1 and all(w in result[0] for w in ["comprehension", "list"])
+
+    messages = [
+        {"role": "system", "content": "You are a helpful assistant."},
+        {"role": "user", "content": "Who won the world series in 2020?"},
+        {"role": "assistant", "content": "The Los Angeles Dodgers won the World Series in 2020."},
+        {"role": "user", "content": "Where was it played?"},
+    ]
+    result = pn(messages)
+    assert len(result) == 1 and all(w in result[0].casefold() for w in ["arlington", "texas"])


Code review comments:
diff_hunk: @@ -545,3 +547,111 @@ def supports(cls, model_name_or_path: str, **kwargs) -> bool:
         return (
             valid_model and kwargs.get("azure_base_url") is not None and kwargs.get("azure_deployment_name") is not None
         )
+
+
+class ChatGPTInvocationLayer(OpenAIInvocationLayer):
+    """
+    ChatGPT Invocation Layer
+
+    PromptModelInvocationLayer implementation for OpenAI's GPT-3 ChatGPT API. Invocations are made using REST API.
+    See [OpenAI ChatGPT API](https://platform.openai.com/docs/guides/chat) for more details.
+
+    Note: kwargs other than init parameter names are ignored to enable reflective construction of the class
+    as many variants of PromptModelInvocationLayer are possible and they may have different parameters.
+    """
+
+    def __init__(
+        self, api_key: str, model_name_or_path: str = "gpt-3.5-turbo", max_length: Optional[int] = 500, **kwargs
+    ):
+        super().__init__(api_key, model_name_or_path, max_length, **kwargs)
+
+    def invoke(self, *args, **kwargs):
+        """
+        It takes in either a prompt or a list of messages and returns a list of responses, using a REST invocation.
+
+        :return: A list of generated responses.
+
+        Note: Only kwargs relevant to OpenAI are passed to OpenAI rest API. Others kwargs are ignored.
+        For more details, see [OpenAI ChatGPT API reference](https://platform.openai.com/docs/api-reference/chat).
+        """
+        prompt = kwargs.get("prompt", None)
+
+        if isinstance(prompt, str):
+            messages = [{"role": "user", "content": prompt}]
+        elif isinstance(prompt, list) and len(prompt) > 0 and isinstance(prompt[0], dict):
+            messages = prompt
+        else:
+            raise ValueError(
+                f"The prompt format is different than what the model expects."
path: haystack/nodes/prompt/providers.py
text: @user1:
Add a space at the end of the string, otherwise the error message will be missing a space after the period.
```suggestion
                f"The prompt format is different than what the model expects. "
```