Add a /query command. (#28)

.github/workflows/prodution-deploy.yml

@@ -4,6 +4,7 @@ on:
   push:
     branches:
       - master
+      - realistik-changes
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.ref }}

app/discord_message_utils.py

@@ -36,3 +36,28 @@ def split_message_into_chunks(message: str, *, max_length: int) -> list[str]:
         return [message[:split_index]] + split_message_into_chunks(
             message[split_index:], max_length=max_length
         )
+
+
+def smart_split_message_into_chunks(message: str, *, max_length: int) -> list[str]:
+    """Like `split_message_into_chunks`, but also considers code blocks."""
+
+    split_messages = split_message_into_chunks(
+        message,
+        # Magic number to account for MD code embed headers.
+        max_length=max_length - 15,
+    )
+    output_messages = []
+    code_block_language = None
+
+    for message_chunk in split_messages:
+        if code_block_language is not None:
+            message_chunk = f"```{code_block_language}\n" + message_chunk
+            code_block_language = None
+
+        code_block_language = get_unclosed_code_block_language(message_chunk)
+        if code_block_language is not None:
+            message_chunk += "\n```"
+
+        output_messages.append(message_chunk)
+
+    return output_messages

app/main.py

@@ -467,5 +467,34 @@ async def transcript(
         )
 
 
+@command_tree.command(name=command_name("query"))
+async def query(
+    interaction: discord.Interaction,
+    query: str,
+    model: gpt.OpenAIModel = gpt.OpenAIModel.GPT_4_OMNI,
+):
+    """Query a model without any context."""
+
+    await interaction.response.defer()
+
+    result = await ai_conversations.send_message_without_context(
+        bot,
+        interaction,
+        query,
+        model,
+    )
+
+    # I do not think interactions allow multiple messages.
+    messages_to_send: list[str] = []
+    if isinstance(result, Error):
+        messages_to_send = result.messages
+    else:
+        messages_to_send = result.response_messages
+
+    # I have no idea whether they actually allow you to send multiple follow-ups.
+    for message_text in messages_to_send:
+        await interaction.followup.send(message_text)
+
+
 if __name__ == "__main__":
     bot.run(settings.DISCORD_TOKEN)

app/usecases/ai_conversations.py

@@ -1,5 +1,6 @@
 import json
 import traceback
+from typing import NamedTuple
 
 import discord
 from pydantic import BaseModel
@@ -44,6 +45,107 @@ def get_author_name(discord_author_name: str) -> str:
     return f"User #{_get_author_id(discord_author_name)}"
 
 
+class _GptRequestResponse(NamedTuple):
+    response_content: str
+    input_tokens: int
+    output_tokens: int
+
+
+async def _make_gpt_request(
+    message_history: list[gpt.Message], model: gpt.OpenAIModel
+) -> _GptRequestResponse | Error:
+    functions = openai_functions.get_full_openai_functions_schema()
+    try:
+        gpt_response = await gpt.send(
+            model=model,
+            messages=message_history,
+            functions=functions,
+        )
+    except Exception as exc:
+        traceback.print_exc()
+        # NOTE: this is *generally* bad practice to expose this information
+        # to end users, and should be removed if we are to deploy this app
+        # more widely. Right now it's okay because it's a private bot.
+        return Error(
+            code=ErrorCode.UNEXPECTED_ERROR,
+            messages=[
+                f"Request to OpenAI failed with the following error:\n```\n{exc}```"
+            ],
+        )
+
+    gpt_choice = gpt_response.choices[0]
+    gpt_message = gpt_choice.message
+
+    if gpt_choice.finish_reason == "stop":
+        assert gpt_message.content is not None
+        gpt_response_content: str = gpt_message.content
+        message_history.append(
+            {
+                "role": gpt_message.role,
+                "content": [
+                    {
+                        "type": "text",
+                        "text": gpt_message.content,
+                    }
+                ],
+            }
+        )
+    elif (
+        gpt_choice.finish_reason == "function_call"
+        and gpt_message.function_call is not None
+    ):
+        function_name = gpt_message.function_call.name
+        function_kwargs = json.loads(gpt_message.function_call.arguments)
+
+        ai_function = openai_functions.ai_functions[function_name]
+        function_response = await ai_function["callback"](**function_kwargs)
+
+        # send function response back to gpt for the final response
+        # TODO: could it call another function?
+        #       i think they may expect/support recursive calls
+        message_history.append(
+            {
+                "role": "function",
+                "name": function_name,
+                "content": [function_response],
+            }
+        )
+        try:
+            gpt_response = await gpt.send(
+                model=model,
+                messages=message_history,
+            )
+        except Exception as exc:
+            traceback.print_exc()
+            # NOTE: this is *generally* bad practice to expose this information
+            # to end users, and should be removed if we are to deploy this app
+            # more widely. Right now it's okay because it's a private bot.
+            return Error(
+                code=ErrorCode.UNEXPECTED_ERROR,
+                messages=[
+                    f"Request to OpenAI failed with the following error:\n```\n{exc}```"
+                ],
+            )
+
+        assert gpt_response.choices[0].message.content is not None
+        gpt_response_content = gpt_response.choices[0].message.content
+
+    else:
+        raise NotImplementedError(
+            f"Unknown chatgpt finish reason: {gpt_choice.finish_reason}"
+        )
+
+    assert gpt_response.usage is not None
+    input_tokens = gpt_response.usage.prompt_tokens
+    output_tokens = gpt_response.usage.completion_tokens
+
+    return _GptRequestResponse(
+        gpt_response_content,
+        input_tokens,
+        output_tokens,
+    )
+
+
 class SendAndReceiveResponse(BaseModel):
     response_messages: list[str]
 
@@ -136,127 +238,92 @@ async def send_message_to_thread(
             }
         )
 
-        functions = openai_functions.get_full_openai_functions_schema()
-        try:
-            gpt_response = await gpt.send(
-                model=tracked_thread.model,
-                messages=message_history,
-                functions=functions,
-            )
-        except Exception as exc:
-            traceback.print_exc()
-            # NOTE: this is *generally* bad practice to expose this information
-            # to end users, and should be removed if we are to deploy this app
-            # more widely. Right now it's okay because it's a private bot.
-            return Error(
-                code=ErrorCode.UNEXPECTED_ERROR,
-                messages=[
-                    f"Request to OpenAI failed with the following error:\n```\n{exc}```"
-                ],
-            )
-
-        gpt_choice = gpt_response.choices[0]
-        gpt_message = gpt_choice.message
-
-        if gpt_choice.finish_reason == "stop":
-            assert gpt_message.content is not None
-            gpt_response_content: str = gpt_message.content
-
-            message_history.append(
-                {
-                    "role": gpt_message.role,
-                    "content": [
-                        {
-                            "type": "text",
-                            "text": gpt_message.content,
-                        }
-                    ],
-                }
-            )
-        elif (
-            gpt_choice.finish_reason == "function_call"
-            and gpt_message.function_call is not None
-        ):
-            function_name = gpt_message.function_call.name
-            function_kwargs = json.loads(gpt_message.function_call.arguments)
-
-            ai_function = openai_functions.ai_functions[function_name]
-            function_response = await ai_function["callback"](**function_kwargs)
-
-            # send function response back to gpt for the final response
-            # TODO: could it call another function?
-            #       i think they may expect/support recursive calls
-            message_history.append(
-                {
-                    "role": "function",
-                    "name": function_name,
-                    "content": [function_response],
-                }
-            )
-            try:
-                gpt_response = await gpt.send(
-                    model=tracked_thread.model,
-                    messages=message_history,
-                )
-            except Exception as exc:
-                traceback.print_exc()
-                # NOTE: this is *generally* bad practice to expose this information
-                # to end users, and should be removed if we are to deploy this app
-                # more widely. Right now it's okay because it's a private bot.
-                return Error(
-                    code=ErrorCode.UNEXPECTED_ERROR,
-                    messages=[
-                        f"Request to OpenAI failed with the following error:\n```\n{exc}```"
-                    ],
-                )
-
-            assert gpt_response.choices[0].message.content is not None
-            gpt_response_content = gpt_response.choices[0].message.content
-
-        else:
-            raise NotImplementedError(
-                f"Unknown chatgpt finish reason: {gpt_choice.finish_reason}"
-            )
-
-        assert gpt_response.usage is not None
-        input_tokens = gpt_response.usage.prompt_tokens
-        output_tokens = gpt_response.usage.completion_tokens
+        gpt_response = await _make_gpt_request(
+            message_history,
+            tracked_thread.model,
+        )
+        if isinstance(gpt_response, Error):
+            return gpt_response
 
         # Handle code blocks which may exceed the previous message.
-        response_messages: list[str] = []
-        code_block_language: str | None = None
-        for chunk in discord_message_utils.split_message_into_chunks(
-            gpt_response_content,
-            max_length=1985,
-        ):
-            if code_block_language is not None:
-                chunk = f"```{code_block_language}\n" + chunk
-                code_block_language = None
-
-            code_block_language = (
-                discord_message_utils.get_unclosed_code_block_language(chunk)
+        response_messages: list[str] = (
+            discord_message_utils.smart_split_message_into_chunks(
+                gpt_response.response_content,
+                max_length=2000,
             )
-            if code_block_language is not None:
-                chunk += "\n```"
-
-            response_messages.append(chunk)
+        )
 
         await thread_messages.create(
             message.channel.id,
             prompt,
             discord_user_id=message.author.id,
             role="user",
-            tokens_used=input_tokens,
+            tokens_used=gpt_response.input_tokens,
         )
 
         await thread_messages.create(
             message.channel.id,
-            gpt_response_content,
+            gpt_response.response_content,
             discord_user_id=bot.user.id,
             role="assistant",
-            tokens_used=output_tokens,
+            tokens_used=gpt_response.output_tokens,
         )
 
     return SendAndReceiveResponse(
         response_messages=response_messages,
     )
+
+
+async def send_message_without_context(
+    bot: DiscordBot,
+    interaction: discord.Interaction,
+    message_content: str,
+    model: gpt.OpenAIModel,
+) -> SendAndReceiveResponse | Error:
+    if bot.user is None:
+        return Error(
+            code=ErrorCode.NOT_READY,
+            messages=["The server is not ready to handle requests"],
+        )
+
+    if interaction.user.id == bot.user.id:
+        return Error(code=ErrorCode.SKIP, messages=[])
+
+    if interaction.user.id not in DISCORD_USER_ID_WHITELIST:
+        return Error(
+            code=ErrorCode.UNAUTHORIZED,
+            messages=["User is not authorised to use this bot"],
+        )
+
+    # author_name = get_author_name(interaction.user.name)
+    # prompt = f"{author_name}: {message_content}"
+
+    # Since there is no context nor multi-user convos, we can just send the message as is
+    prompt = message_content
+
+    user_messages: list[MessageContent] = [
+        {
+            "type": "text",
+            "text": prompt,
+        }
+    ]
+    message_context: list[gpt.Message] = [
+        {
+            "role": "user",
+            "content": user_messages,
+        }
+    ]
+
+    gpt_response = await _make_gpt_request(message_context, model)
+    if isinstance(gpt_response, Error):
+        return gpt_response
+
+    # TODO: Track input and output tokens here.
+
+    response_messages: list[str] = (
+        discord_message_utils.smart_split_message_into_chunks(
+            gpt_response.response_content,
+            max_length=2000,
+        )
+    )
+    return SendAndReceiveResponse(response_messages=response_messages)