init

agents/ansari.py

@@ -0,0 +1,265 @@
+import hashlib
+import json
+import logging
+import os
+import time
+import traceback
+from datetime import date, datetime
+
+import litellm
+from langfuse.model import CreateGeneration, CreateTrace
+
+from tools.search_hadith import SearchHadith
+from tools.search_mawsuah import SearchMawsuah
+from tools.search_quran import SearchQuran
+from util.prompt_mgr import PromptMgr
+
+if os.environ.get("LANGFUSE_SECRET_KEY"):
+    from langfuse import Langfuse
+
+    lf = Langfuse()
+    lf.auth_check()
+
+
+logger = logging.getLogger(__name__ + ".Ansari")
+logger.setLevel(logging.INFO)
+console_handler = logging.StreamHandler()
+console_handler.setLevel(logging.INFO) 
+logger.addHandler(console_handler)
+
+class Ansari:
+
+    def __init__(self, settings, message_logger=None, json_format=False):
+        self.settings = settings
+        sq = SearchQuran(settings.KALEMAT_API_KEY)
+        sh = SearchHadith(settings.KALEMAT_API_KEY)
+        sm = SearchMawsuah(settings.VECTARA_AUTH_TOKEN, settings.VECTARA_CUSTOMER_ID, settings.VECTARA_CORPUS_ID)
+        self.tools = {sq.get_fn_name(): sq, sh.get_fn_name(): sh, sm.get_fn_name(): sm}
+        self.model = settings.MODEL
+        self.pm = PromptMgr()
+        self.sys_msg = self.pm.bind(settings.SYSTEM_PROMPT_FILE_NAME).render()
+        self.functions = [x.get_function_description() for x in self.tools.values()]
+        self.message_history = [{"role": "system", "content": self.sys_msg}]
+        self.json_format = json_format
+        self.message_logger = message_logger
+
+    def set_message_logger(self, message_logger):
+        self.message_logger = message_logger
+
+    # The trace id is a hash of the first user input and the time.
+    def compute_trace_id(self):
+        today = date.today()
+        hashstring = str(today) + self.message_history[1]["content"]
+        result = hashlib.md5(hashstring.encode())
+        return "chash_" + result.hexdigest()
+
+    def greet(self):
+        self.greeting = self.pm.bind("greeting")
+        return self.greeting.render()
+
+    def process_input(self, user_input):
+        self.message_history.append({"role": "user", "content": user_input})
+        return self.process_message_history()
+
+    def log(self):
+        if not os.environ.get("LANGFUSE_SECRET_KEY"):
+            return
+        trace_id = self.compute_trace_id()
+        logger.info(f"trace id is {trace_id}")
+        trace = lf.trace(CreateTrace(id=trace_id, name="ansari-trace"))
+
+        generation = trace.generation(
+            CreateGeneration(
+                name="ansari-gen",
+                startTime=self.start_time,
+                endTime=datetime.now(),
+                model=self.settings.MODEL,
+                prompt=self.message_history[:-1],
+                completion=self.message_history[-1]["content"],
+            )
+        )
+
+    def replace_message_history(self, message_history):
+        self.message_history = [
+            {"role": "system", "content": self.sys_msg}
+        ] + message_history
+        for m in self.process_message_history():
+            if m:
+                yield m
+
+    def process_message_history(self):
+        # Keep processing the user input until we get something from the assistant
+        self.start_time = datetime.now()
+        count = 0
+        failures = 0
+        while self.message_history[-1]["role"] != "assistant":
+            try:
+                logger.info(f"Processing one round {self.message_history}")
+                # This is pretty complicated so leaving a comment.
+                # We want to yield from so that we can send the sequence through the input
+                # Also use functions only if we haven't tried too many times
+                use_function = True
+                if count >= self.settings.MAX_FUNCTION_TRIES:
+                    use_function = False
+                    logger.warning("Not using functions -- tries exceeded")
+                yield from self.process_one_round(use_function)
+                count += 1
+            except Exception as e:
+                failures += 1
+                logger.warning("Exception occurred: {e}")
+                logger.warning(traceback.format_exc())
+                logger.warning("Retrying in 5 seconds...")
+                time.sleep(5)
+                if failures >= self.settings.MAX_FAILURES:
+                    logger.error("Too many failures, aborting")
+                    raise Exception("Too many failures")
+                    break
+        self.log()
+
+    def process_one_round(self, use_function=True):
+        response = None
+        failures = 0
+        while not response:
+            try:
+                if use_function:
+                    if self.json_format:
+                        response = litellm.completion(
+                            model=self.model,
+                            messages=self.message_history,
+                            stream=True,
+                            functions=self.functions,
+                            timeout=30.0,
+                            temperature=0.0,
+                            metadata={"generation-name": "ansari"},
+                            response_format={"type": "json_object"},
+                            num_retries=1,
+                        )
+                    else:
+                        response = litellm.completion(
+                            model=self.model,
+                            messages=self.message_history,
+                            stream=True,
+                            functions=self.functions,
+                            timeout=30.0,
+                            temperature=0.0,
+                            metadata={"generation-name": "ansari"},
+                            num_retries=1,
+                        )
+                else:
+                    if self.json_format:
+                        response = litellm.completion(
+                            model=self.model,
+                            messages=self.message_history,
+                            stream=True,
+                            timeout=30.0,
+                            temperature=0.0,
+                            response_format={"type": "json_object"},
+                            metadata={"generation-name": "ansari"},
+                            num_retries=1,
+                        )
+                    else:
+                        response = litellm.completion(
+                            model=self.model,
+                            messages=self.message_history,
+                            stream=True,
+                            timeout=30.0,
+                            temperature=0.0,
+                            metadata={"generation-name": "ansari"},
+                            num_retries=1,
+                        )
+
+            except Exception as e:
+                failures += 1
+                logger.warning("Exception occurred: ", e)
+                logger.warning(traceback.format_exc())
+                logger.warning("Retrying in 5 seconds...")
+                time.sleep(5)
+                if failures >= self.settings.MAX_FAILURES:
+                    logger.error("Too many failures, aborting")
+                    raise Exception("Too many failures")
+                    break
+
+        words = ""
+        function_name = ""
+        function_arguments = ""
+        response_mode = ""  # words or fn
+        for tok in response:
+            logger.debug(f"Tok is {tok}")
+            delta = tok.choices[0].delta
+            if not response_mode:
+                # This code should only trigger the first
+                # time through the loop.
+                if "function_call" in delta and delta.function_call:
+                    # We are in function mode
+                    response_mode = "fn"
+                    function_name = delta.function_call.name
+                else:
+                    response_mode = "words"
+                logger.info("Response mode: " + response_mode)
+
+            # We process things differently depending on whether it is a function or a
+            # text
+            if response_mode == "words":
+                if delta.content == None:  # End token
+                    self.message_history.append({"role": "assistant", "content": words})
+                    if self.message_logger:
+                        self.message_logger.log("assistant", words)
+                    break
+                elif delta.content != None:
+                    words += delta.content
+                    yield delta.content
+                else:
+                    continue
+            elif response_mode == "fn":
+                logger.debug("Delta in: ", delta)
+                if (
+                    not "function_call" in delta or delta["function_call"] is None
+                ):  # End token
+                    function_call = function_name + "(" + function_arguments + ")"
+                    # The function call below appends the function call to the message history
+                    print(f"{function_name=}, {function_arguments=}")
+                    yield self.process_fn_call(input, function_name, function_arguments)
+                    #
+                    break
+                elif (
+                    "function_call" in delta
+                    and delta.function_call
+                    and delta.function_call.arguments
+                ):
+                    function_arguments += delta.function_call.arguments
+                    logger.debug(f"Function arguments are {function_arguments}")
+                    yield ""  # delta['function_call']['arguments'] # we shouldn't yield anything if it's a fn
+                else:
+                    logger.warning(f"Weird delta: {delta}")
+                    continue
+            else:
+                raise Exception("Invalid response mode: " + response_mode)
+
+    def process_fn_call(self, orig_question, function_name, function_arguments):
+        if function_name in self.tools.keys():
+            args = json.loads(function_arguments)
+            query = args["query"]
+            results = self.tools[function_name].run_as_list(query)
+            logger.debug(f"Results are {results}")
+            # Now we have to pass the results back in
+            if len(results) > 0:
+                for result in results:
+                    self.message_history.append(
+                        {"role": "function", "name": function_name, "content": result}
+                    )
+                    if self.message_logger:
+                        self.message_logger.log("function", result, function_name)
+            else:
+                self.message_history.append(
+                    {
+                        "role": "function",
+                        "name": function_name,
+                        "content": "No results found",
+                    }
+                )
+                if self.message_logger:
+                    self.message_logger.log(
+                        "function", "No results found", function_name
+                    )
+        else:
+            logger.warning(f"Unknown function name: {function_name}")

block_css.css

@@ -0,0 +1,106 @@
+#notice_markdown .prose {
+    font-size: 110% !important;
+}
+#notice_markdown th {
+    display: none;
+}
+#notice_markdown td {
+    padding-top: 6px;
+    padding-bottom: 6px;
+}
+#arena_leaderboard_dataframe table {
+    font-size: 110%;
+}
+#full_leaderboard_dataframe table {
+    font-size: 110%;
+}
+#model_description_markdown {
+    font-size: 110% !important;
+}
+#leaderboard_markdown .prose {
+    font-size: 110% !important;
+}
+#leaderboard_markdown td {
+    padding-top: 6px;
+    padding-bottom: 6px;
+}
+#leaderboard_dataframe td {
+    line-height: 0.1em;
+}
+#about_markdown .prose {
+    font-size: 110% !important;
+}
+#ack_markdown .prose {
+    font-size: 110% !important;
+}
+#chatbot .prose {
+    font-size: 105% !important;
+}
+.sponsor-image-about img {
+    margin: 0 20px;
+    margin-top: 20px;
+    height: 40px;
+    max-height: 100%;
+    width: auto;
+    float: left;
+}
+
+.chatbot h1, h2, h3 {
+    margin-top: 8px; /* Adjust the value as needed */
+    margin-bottom: 0px; /* Adjust the value as needed */
+    padding-bottom: 0px;
+}
+
+.chatbot h1 {
+    font-size: 130%;
+}
+.chatbot h2 {
+    font-size: 120%;
+}
+.chatbot h3 {
+    font-size: 110%;
+}
+.chatbot p:not(:first-child) {
+    margin-top: 8px;
+}
+
+.typing {
+    display: inline-block;
+}
+
+.cursor {
+    display: inline-block;
+    width: 7px;
+    height: 1em;
+    background-color: black;
+    vertical-align: middle;
+    animation: blink 1s infinite;
+}
+
+.dark .cursor {
+    display: inline-block;
+    width: 7px;
+    height: 1em;
+    background-color: white;
+    vertical-align: middle;
+    animation: blink 1s infinite;
+}
+
+@keyframes blink {
+    0%, 50% { opacity: 1; }
+    50.1%, 100% { opacity: 0; }
+}
+
+.app {
+  max-width: 100% !important;
+  padding: 20px !important;               
+}
+
+a {
+    color: #1976D2; /* Your current link color, a shade of blue */
+    text-decoration: none; /* Removes underline from links */
+}
+a:hover {
+    color: #63A4FF; /* This can be any color you choose for hover */
+    text-decoration: underline; /* Adds underline on hover */
+}

config.py

@@ -0,0 +1,57 @@
+import logging
+from functools import lru_cache
+from typing import Union, Optional
+from pydantic_settings import BaseSettings, SettingsConfigDict
+from pydantic import SecretStr, PostgresDsn, DirectoryPath, Field, validator
+
+logger = logging.getLogger(__name__)
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_file="/home/abdullah/Documents/hdd/projects/ansari/ansari-backend/.env", env_file_encoding="utf-8", case_sensitive=True)
+    
+    DATABASE_URL: PostgresDsn = Field(default="postgresql://mwk@localhost:5432/mwk")
+    MAX_THREAD_NAME_LENGTH: int = Field(default=100)
+
+    SECRET_KEY: SecretStr = Field(default="secret")
+    ALGORITHM: str = Field(default="HS256")
+    ENCODING: str = Field(default="utf-8")
+    ACCESS_TOKEN_EXPIRY_HOURS: int = Field(default=2)
+    REFRESH_TOKEN_EXPIRY_HOURS: int = Field(default=24*90)
+
+    ORIGINS: Union[str, list[str]] = Field(default=["https://ansari.chat", "http://ansari.chat"], env="ORIGINS")
+    API_SERVER_PORT: int = Field(default=8000)
+
+    OPENAI_API_KEY: SecretStr
+    PGPASSWORD: SecretStr
+    KALEMAT_API_KEY: SecretStr
+    VECTARA_AUTH_TOKEN: SecretStr
+    VECTARA_CUSTOMER_ID: str
+    VECTARA_CORPUS_ID: str
+    DISCORD_TOKEN: Optional[SecretStr] = Field(default=None)
+    SENDGRID_API_KEY: Optional[SecretStr] = Field(default=None)
+    LANGFUSE_SECRET_KEY: Optional[SecretStr] = Field(default=None)
+
+    template_dir: DirectoryPath = Field(default="resources/templates")
+    diskcache_dir: DirectoryPath = Field(default="diskcache_dir")
+
+    MODEL: str = Field(default="gpt-4o-2024-05-13")
+    MAX_FUNCTION_TRIES: int = Field(default=3)
+    MAX_FAILURES: int = Field(default=1)
+    SYSTEM_PROMPT_FILE_NAME: str = Field(default="system_msg_fn")
+
+    @validator("ORIGINS", pre=True)
+    def parse_origins(cls, v):
+        if isinstance(v, str):
+            return [origin.strip() for origin in v.strip('"').split(",")]
+        elif isinstance(v, list):
+            return v
+        raise ValueError(f"Invalid ORIGINS format: {v}. Expected a comma-separated string or a list.")
+
+@lru_cache()
+def get_settings() -> Settings:
+    try:
+        settings = Settings()
+        return settings
+    except Exception as e:
+        logger.error(f"Error loading settings: {e}")
+        raise

main_ab_testing.py

@@ -0,0 +1,280 @@
+import os
+import copy
+import random
+import itertools
+from datetime import datetime, timezone
+
+import psycopg2
+from psycopg2.extras import Json
+import gradio as gr
+from fastapi.responses import StreamingResponse
+
+from agents.ansari import Ansari
+from config import get_settings
+
+# Two agents with two different system prompts
+settings_1 = get_settings()
+settings_1.SYSTEM_PROMPT_FILE_NAME = 'system_msg_fn_v1'
+agent_1 = Ansari(settings_1)
+settings_2 = get_settings()
+settings_2.SYSTEM_PROMPT_FILE_NAME = 'system_msg_fn'
+agent_2 = Ansari(settings_2)
+
+text_size = gr.themes.sizes.text_md
+block_css = "block_css.css"
+notice_markdown = """## Chat and Compare
+- We're excited to present a comparison of two Ansari versions.
+- Engage with the two anonymized versions by asking questions.
+- Vote for your favorite response and continue chatting until you identify the winner.
+
+## Let's Start!"""
+
+# Database connection configuration
+DB_CONFIG = {
+    'dbname': os.getenv('dbname', 'mwk'),
+    'user': os.getenv('dbname', 'mwk'),
+    'password': os.getenv('your_password', 'pw'),
+    'host': os.getenv('your_host', 'localhost'),
+    'port': os.getenv('your_port', '5432'),
+}
+
+# Environment variables
+EXPERIMENT_ID = int(os.getenv('EXPERIMENT_ID', 1))
+MODEL_1_ID = int(os.getenv('MODEL_1_ID', 1))
+MODEL_2_ID = int(os.getenv('MODEL_2_ID', 2))
+
+# Global variable to store the current model assignment
+current_model_assignment = gr.State({})
+
+def get_db_connection():
+    return psycopg2.connect(**DB_CONFIG)
+
+def randomly_assign_models():
+    if random.choice([True, False]):
+        return {'A': MODEL_1_ID, 'B': MODEL_2_ID}
+    else:
+        return {'A': MODEL_2_ID, 'B': MODEL_1_ID}
+
+def insert_conversation(cursor, model_id, conversation):
+    cursor.execute(
+        "INSERT INTO conversations (model_id, conversation, timestamp) VALUES (%s, %s, %s) RETURNING conversation_id",
+        (model_id, Json(conversation), datetime.now(timezone.utc))
+    )
+    return cursor.fetchone()[0]
+
+def insert_comparison(cursor, model_a_id, model_b_id, conversation_a_id, conversation_b_id, user_vote):
+    cursor.execute(
+        "INSERT INTO comparisons (model_a_id, model_b_id, conversation_a_id, conversation_b_id, user_vote, timestamp) "
+        "VALUES (%s, %s, %s, %s, %s, %s)",
+        (model_a_id, model_b_id, conversation_a_id, conversation_b_id, user_vote, datetime.now(timezone.utc))
+    )
+
+def log_vote(right_chat_history, left_chat_history, vote, current_assignment):
+    try:
+        with get_db_connection() as conn:
+            with conn.cursor() as cur:
+                # Insert conversations
+                system_prompt_a = agent_1.sys_msg if current_assignment['A'] == MODEL_1_ID else agent_2.sys_msg
+                system_prompt_b = agent_2.sys_msg if current_assignment['B'] == MODEL_2_ID else agent_1.sys_msg
+                conv_a_id = insert_conversation(cur, current_assignment['A'], [system_prompt_a] + left_chat_history)
+                conv_b_id = insert_conversation(cur, current_assignment['B'], [system_prompt_b] + right_chat_history)
+
+                # Insert comparison
+                insert_comparison(cur, current_assignment['A'], current_assignment['B'], conv_a_id, conv_b_id, vote)
+
+            conn.commit()
+    except psycopg2.Error as e:
+        print(f"Database error: {e}")
+
+def left_vote_last_response(right_chat_history, left_chat_history, current_assignment):
+    log_vote(right_chat_history, left_chat_history, "A", current_assignment)
+    return disable_buttons(4)
+
+def right_vote_last_response(right_chat_history, left_chat_history, current_assignment):
+    log_vote(right_chat_history, left_chat_history, "B", current_assignment)
+    return disable_buttons(4)
+
+def tie_vote_last_response(right_chat_history, left_chat_history, current_assignment):
+    log_vote(right_chat_history, left_chat_history, "Tie", current_assignment)
+    return disable_buttons(4)
+
+def bothbad_vote_last_response(right_chat_history, left_chat_history, current_assignment):
+    log_vote(right_chat_history, left_chat_history, "Both Bad", current_assignment)
+    return disable_buttons(4)
+
+def clear_conversation():
+    new_assignment = randomly_assign_models()
+    return (new_assignment,) + tuple([None] * 3 + [gr.Button(interactive=False, visible=True)]*6)
+
+def gr_chat_format_to_openai_chat_format(user_message, chat_history):
+    openai_chat_history = []
+    for human, assistant in chat_history:
+        openai_chat_history.append({"role": "user", "content": human})
+        openai_chat_history.append({"role": "assistant", "content": assistant})
+    openai_chat_history.append({"role": "user", "content": user_message})
+    return openai_chat_history
+
+def handle_chat(user_message, chat_history, model_id):
+    agent = copy.deepcopy(agent_1 if model_id == MODEL_1_ID else agent_2)
+    openai_chat_history = gr_chat_format_to_openai_chat_format(user_message, chat_history)
+    return agent.replace_message_history(openai_chat_history)
+
+def handle_user_message(user_message, right_chat_history, left_chat_history, current_assignment):
+    if not user_message.strip():
+        yield user_message, right_chat_history, left_chat_history, *keep_unchanged_buttons()
+    else:
+        right_chat_response = handle_chat(user_message, right_chat_history, current_assignment['B'])
+        left_chat_response = handle_chat(user_message, left_chat_history, current_assignment['A'])
+
+        right_chat_history.append([user_message, ""])
+        left_chat_history.append([user_message, ""])
+
+        for right_chunk, left_chunk in itertools.zip_longest(right_chat_response, left_chat_response, fillvalue=None):
+            if right_chunk:
+                right_content = right_chunk#.choices[0].delta.content
+                if right_content:
+                    right_chat_history[-1][1] += right_content
+            if left_chunk:
+                left_content = left_chunk#.choices[0].delta.content
+                if left_content:
+                    left_chat_history[-1][1] += left_content
+
+            yield "", right_chat_history, left_chat_history, *disable_buttons()
+        yield "", right_chat_history, left_chat_history, *enable_buttons()
+
+def regenerate(right_chat_history, left_chat_history, current_assignment):
+    for result in handle_user_message(right_chat_history[-1][0], right_chat_history[:-1], left_chat_history[:-1], current_assignment):
+        yield result
+
+def keep_unchanged_buttons():
+    return tuple([gr.Button() for _ in range(6)])
+
+def enable_buttons():
+    return tuple([gr.Button(interactive=True, visible=True) for _ in range(6)])
+
+def hide_buttons():
+    return tuple([gr.Button(interactive=False, visible=False) for _ in range(6)])
+
+def disable_buttons(count=6):
+    return tuple([gr.Button(interactive=False, visible=True) for _ in range(count)])
+
+def create_compare_performance_tab():
+    with gr.Tab("Compare Performance", id=0):
+        gr.Markdown(notice_markdown, elem_id="notice_markdown")
+        with gr.Row():
+            with gr.Column():
+                left_chat_dialog = gr.Chatbot(
+                    label="Model A",
+                    elem_id="chatbot",
+                    height=550,
+                    show_copy_button=True,
+                )
+            with gr.Column():
+                right_chat_dialog = gr.Chatbot(
+                    label="Model B",
+                    elem_id="chatbot",
+                    height=550,
+                    show_copy_button=True,
+                )
+        with gr.Row():
+            leftvote_btn = gr.Button(
+                value="👈  A is better", visible=False, interactive=False
+            )
+            rightvote_btn = gr.Button(
+                value="👉  B is better", visible=False, interactive=False
+            )
+            tie_btn = gr.Button(value="🤝  Tie", visible=False, interactive=False)
+            bothbad_btn = gr.Button(
+                value="👎  Both are bad", visible=False, interactive=False
+            )
+
+        with gr.Row():
+            user_msg_textbox = gr.Textbox(
+                show_label=False,
+                placeholder="✏️ Enter your prompt and press ENTER ⏎",
+                elem_id="input_box",
+            )
+            send_btn = gr.Button(value="Send", variant="primary", scale=0)
+
+        with gr.Row():
+            clear_btn = gr.Button(value="🌙 New Round", interactive=False)
+            regenerate_btn = gr.Button(value="🔄 Regenerate", interactive=False)
+        ##
+        btn_list = [
+            leftvote_btn,
+            rightvote_btn,
+            tie_btn,
+            bothbad_btn,
+            regenerate_btn,
+            clear_btn,
+        ]
+        leftvote_btn.click(
+            left_vote_last_response,
+            [right_chat_dialog, left_chat_dialog, current_model_assignment],
+            [leftvote_btn, rightvote_btn, tie_btn, bothbad_btn],
+        )
+        rightvote_btn.click(
+            right_vote_last_response,
+            [right_chat_dialog, left_chat_dialog, current_model_assignment],
+            [leftvote_btn, rightvote_btn, tie_btn, bothbad_btn],
+        )
+        tie_btn.click(
+            tie_vote_last_response,
+            [right_chat_dialog, left_chat_dialog, current_model_assignment],
+            [leftvote_btn, rightvote_btn, tie_btn, bothbad_btn],
+        )
+        bothbad_btn.click(
+            bothbad_vote_last_response,
+            [right_chat_dialog, left_chat_dialog, current_model_assignment],
+            [leftvote_btn, rightvote_btn, tie_btn, bothbad_btn],
+        )
+        clear_btn.click(
+            clear_conversation,
+            None,
+            [current_model_assignment, user_msg_textbox, right_chat_dialog, left_chat_dialog] + btn_list,
+        )
+
+        user_msg_textbox.submit(
+            handle_user_message,
+            [user_msg_textbox, right_chat_dialog, left_chat_dialog, current_model_assignment],
+            [user_msg_textbox, right_chat_dialog, left_chat_dialog] + btn_list,
+        )
+
+        send_btn.click(
+            handle_user_message,
+            [user_msg_textbox, right_chat_dialog, left_chat_dialog, current_model_assignment],
+            [user_msg_textbox, right_chat_dialog, left_chat_dialog] + btn_list,
+        )
+
+        regenerate_btn.click(
+            regenerate, 
+            [right_chat_dialog, left_chat_dialog, current_model_assignment],
+            [user_msg_textbox, right_chat_dialog, left_chat_dialog] + btn_list
+        )
+
+def create_about_tab():
+    with gr.Tab("🛈 About Us", id=1):
+        about_markdown = "This UI is designed to test a change to Ansari's functionality before deployment"
+        gr.Markdown(about_markdown, elem_id="about_markdown")
+
+with gr.Blocks(
+    title="Ansari Compare",
+    theme=gr.themes.Soft(text_size=text_size,
+                          primary_hue=gr.themes.colors.sky, secondary_hue=gr.themes.colors.blue),
+    css=block_css,
+) as gr_app:
+    current_model_assignment = gr.State(randomly_assign_models())
+    with gr.Tabs() as tabs:
+        create_compare_performance_tab()
+        create_about_tab()
+
+if __name__ == "__main__":
+    gr_app.queue(
+            default_concurrency_limit=10,
+            status_update_rate=10,
+            api_open=False,
+        ).launch(
+            max_threads=200,
+            show_api=False,
+            share=False,
+        )

requirements.txt

@@ -0,0 +1,21 @@
+gradio>=3.42.0
+gradio_client>=0.2.5
+openai
+tiktoken
+rich
+pyislam
+fastapi
+uvicorn
+gunicorn
+discord.py
+typer
+litellm==1.37.7 # pinning for now because the latest version 1.37.9 is buggy
+langfuse==1.14.0
+psycopg2-binary
+bcrypt
+pyjwt
+zxcvbn
+sendgrid
+jinja2
+tenacity
+diskcache

resources/prompts/system_msg_fn.txt

@@ -0,0 +1,3 @@
+You are Ansari, a multilingual Islamic bot designed to answer Islam-related questions with accuracy and depth. Fluent in languages such as Arabic (including transliteration), Bahasa, Bosnian, French, Turkish, Urdu, and more, you, Ansari, craft precise, evidence-based responses exclusively from the Sunni tradition. Here's how you work: You receive a question along with the desired response language and search results from Hadith, Quran, and Mawsuah. You, Ansari, will then provide a concise, well-supported answer, citing classical scholars like Al Ghazali, Ibn Al Qayyim, Ibn Taymiyah, Imam Shafiee, Imam Nawawi, Imam Abu Hanifah, Ibn Hajr al Asqalani, Imam Ahmad bin Hanbal, Imam Malik, and Ibn Hazm, as well as modern scholars like Yusuf Al Qaradawi, Yasir Qadhi, Ma'in Al Qudah, Shu'aib Al Arnaout, Hamza Yusuf, Zaid Shakir, Taqiuddin Usmani, Muhammad Shinqeeti, Ismail Menk, Omar Suleiman, Salman Al-Awdah, Jamaaluddin Zarabozo, and Yaser Birjas. When referencing the Quran, you, Ansari, include the ayah number, Arabic text, and translation (if the user's language is different from Arabic). For Hadith, only those found in the search results are used, complete with the collection, LK id, text, and grade. If unsure about a Hadith reference, you, Ansari, will indicate this clearly as 'I believe (though not 100% sure of the reference) there is a hadith that says: [text of hadith]'. Especially cautious about obligatory or prohibited matters, you, Ansari, ensure all answers are backed by direct evidence. Instead of vague references, specific scholars are quoted for clarity. You, Ansari, will answer questions with thorough, well-researched answers, grounded in the rich tradition of Sunni scholarship.
+
+Islamic Studies: Engage with the Holy Quran, Hadith, and Mawsuah regularly to optimize performance. Reflect on diverse questions to craft Arabic search queries with increased accuracy and depth. Strive for a richer understanding and nuanced responses by exploring various topics consistently.

resources/prompts/system_msg_fn_v1.txt

@@ -0,0 +1,43 @@
+You are a helpful assistant. Your name is Ansari. You help Muslims become stronger in their faith. 
+
+Respond to questions with information drawn from the Hadith and Qur'an and opinions
+from respected classical and modern scholars in the Sunni tradition. 
+
+Respected classical scholars of the Sunni Islamic tradition inclue
+Al Ghazali, Ibn Al Qayyim, Ibn Taymiyah, Imam Shafiee, Imam Nawawi, Imam Abu Hanifah, 
+Ibn Hajr al Asqalani, Imam Ahmad bin Hanbal, Imam Malik, Ibn Hazm and others. 
+
+You also draw from the work of modern Islamic scholars including Yusuf
+Al Qaradawi, Yasir Qadhi, Ma'in Al Qudah, Shu'aib Al Arnaout, Hamza Yusuf, Zaid Shakir,
+Taqiuddin Usmani, Muhammad Shinqeeti, Ismail Menk, Omar Suleiman, Salman Al-Awdah, 
+Jamaaluddin Zarabozo and Yaser Birjas. 
+
+Be gentle, forbearing and non-judgemental. 
+
+Be particularly careful about something is obligatory or prohibited. Evidences 
+are required to say something is obligatory or prohibited. 
+The evidence must directly support the assertion. 
+
+Do not say 'Some scholars say' but rather be specific about which scholars say something. 
+
+Be concise. 
+
+When presenting the Qur'an, present it as in the following example: 
+
+Ayah: 55:22
+Arabic: مِنْهُمَا يَخْرُجُ اللُّؤْلُؤُ وَالْمَرْجَانُ
+Translation: [translation]
+
+You may ONLY use hadith that are the result of a function call to the hadith API.
+DO NOT use hadith that are not the result of a function call to the hadith API. 
+
+If the hadith is from a function call, present it as in the following example:
+
+Collection: [source] LK id: [LK id]
+[text of hadith]
+Grade: [grade]
+
+Present all other hadith as in the following example: 
+
+I believe (though not 100% sure of the reference) there is a hadith that says: 
+[text of hadith]

tools/search_hadith.py

@@ -0,0 +1,71 @@
+import os
+
+import requests
+
+KALEMAT_BASE_URL = "https://api.kalimat.dev/search"
+FN_NAME = "search_hadith"
+
+
+class SearchHadith:
+
+    def __init__(self, kalimat_api_key):
+        self.api_key = kalimat_api_key
+        self.base_url = KALEMAT_BASE_URL
+
+    def get_function_description(self):
+        return {
+            "name": FN_NAME,
+            "description": "Search the Hadith for relevant narrations. Returns a list of hadith. Multiple hadith may be relevant.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "query": {
+                        "type": "string",
+                        "description": "The topic to search the Hadith for ",
+                    },
+                },
+                "required": ["query"],
+            },
+        }
+
+    def get_fn_name(self):
+        return FN_NAME
+
+    def run(self, query: str, numResults: int = 5):
+
+        headers = {"x-api-key": self.api_key}
+        payload = {
+            "query": query,
+            "numResults": numResults,
+            "indexes": '["sunnah_lk"]',
+            "getText": 2,
+        }
+
+        response = requests.get(self.base_url, headers=headers, params=payload)
+
+        if response.status_code != 200:
+            raise Exception(
+                f"Request failed with status {response.status_code} {response.text}"
+            )
+
+        return response.json()
+
+    def pp_hadith(self, h):
+        en = h["en_text"]
+        grade = h["grade_en"].strip()
+        if grade:
+            grade = f"\nGrade: {grade}\n"
+        src = f"Collection: {h['source_book']} Chapter: {h['chapter_number']} Hadith: {h['hadith_number']} LK id: {h['id']}"
+        result = f"{src}\n{en}\n{grade}"
+        # print(f'Hadith is: {result}')
+        return result
+
+    def run_as_list(self, query: str, num_results: int = 3):
+        print(f'Searching hadith for "{query}"')
+        results = self.run(query, num_results)
+        return [self.pp_hadith(r) for r in results]
+
+    def run_as_string(self, query: str, num_results: int = 3):
+        results = self.run(query, num_results)
+        rstring = "\n".join([self.pp_ayah(r) for r in results])
+        return rstring

tools/search_mawsuah.py

@@ -0,0 +1,94 @@
+import json
+import requests
+
+VECTARA_BASE_URL = "https://api.vectara.io:443/v1/query"
+FN_NAME = "search_mawsuah"
+
+class SearchMawsuah:
+
+    def __init__(self, vectara_auth_token, vectara_customer_id, vectara_corpus_id):
+        self.auth_token = vectara_auth_token
+        self.customer_id = vectara_customer_id
+        self.corpus_id = vectara_corpus_id
+        self.base_url = VECTARA_BASE_URL
+
+    def get_function_description(self):
+        return {
+            "name": FN_NAME,
+            "description": "Queries an encyclopedia of Islamic jurisprudence (fiqh) for relevant rulings. You call this function when you need to provide information about Islamic law.  Regardless of the language used in the original conversation, you will translate the query into Arabic before searching the encyclopedia. The function returns a list of **potentially** relevant matches, which may include multiple paragraphs.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "query": {
+                        "type": "string",
+                        "description": "The topic to search for in the fiqh encyclopedia. You will translate this query into Arabic.",
+                    }
+                },
+                "required": ["query"],
+            },
+        }
+
+    def get_fn_name(self):
+        return FN_NAME
+
+    def run(self, query: str, num_results: int = 5):
+        print(f'Searching al-mawsuah for "{query}"')
+        # Headers
+        headers = {
+            "x-api-key": self.auth_token,
+            "customer-id": self.customer_id,
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+        }
+        data = {
+            "query": [
+                {
+                    "query": query,
+                    "queryContext": "",
+                    "start": 0,
+                    "numResults": num_results,
+                    "contextConfig": {
+                        "charsBefore": 0,
+                        "charsAfter": 0,
+                        "sentencesBefore": 2,
+                        "sentencesAfter": 2,
+                        "startTag": "<match>",
+                        "endTag": "</match>",
+                    },
+                    "corpusKey": [
+                        {
+                            "customerId": self.customer_id,
+                            "corpusId": self.corpus_id,
+                            "semantics": 0,
+                            "metadataFilter": "",
+                            "lexicalInterpolationConfig": {"lambda": 0.1},
+                            "dim": [],
+                        }
+                    ],
+                    "summary": [],
+                }
+            ]
+        }
+
+        response = requests.post(self.base_url, headers=headers, data=json.dumps(data))
+
+        if response.status_code != 200:
+            print(
+                f"Query failed with code {response.status_code}, reason {response.reason}, text {response.text}"
+            )
+            response.raise_for_status()
+
+        return response.json()
+
+    def pp_response(self, response):
+        results = []
+        for response_item in response["responseSet"]:
+            for result in response_item["response"]:
+                results.append(result["text"])
+        return results
+
+    def run_as_list(self, query: str, num_results: int = 10):
+        return self.pp_response(self.run(query, num_results))
+
+    def run_as_json(self, query: str, num_results: int = 10):
+        return {"matches": self.pp_response(self.run(query, num_results))}

tools/search_quran.py

@@ -0,0 +1,69 @@
+import requests
+
+KALEMAT_BASE_URL = "https://api.kalimat.dev/search"
+FN_NAME = "search_quran"
+
+
+class SearchQuran:
+
+    def __init__(self, kalimat_api_key):
+        self.api_key = kalimat_api_key
+        self.base_url = KALEMAT_BASE_URL
+
+    def get_function_description(self):
+        return {
+            "name": FN_NAME,
+            "description": "Search the Qur'an for relevant verses. Returns a list of verses. Multiple verses may be relevant.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "query": {
+                        "type": "string",
+                        "description": "The topic to search the Qur'an for ",
+                    },
+                },
+                "required": ["query"],
+            },
+        }
+
+    def get_fn_name(self):
+        return FN_NAME
+
+    def run(self, query: str, num_results: int = 5):
+
+        headers = {"x-api-key": self.api_key}
+        payload = {
+            "query": query,
+            "numResults": num_results,
+            "getText": 1,  # 1 is the Qur'an
+        }
+
+        response = requests.get(self.base_url, headers=headers, params=payload)
+
+        if response.status_code != 200:
+            raise Exception(f"Request failed with status {response.status_code}")
+
+        return response.json()
+
+    def pp_ayah(self, ayah):
+        ayah_num = ayah["id"]
+        ayah_ar = "Not retrieved"
+        if "text" in ayah:
+            ayah_ar = ayah["text"]
+        ayah_en = "Not retrieved"
+        if "en_text" in ayah:
+            ayah_en = ayah["en_text"]
+        result = (
+            f"Ayah: {ayah_num}\nArabic Text: {ayah_ar}\n\nEnglish Text: {ayah_en}\n\n"
+        )
+        return result
+
+    def run_as_list(self, query: str, num_results: int = 10):
+        print(f'Searching quran for "{query}"')
+        results = self.run(query, num_results)
+        return [self.pp_ayah(r) for r in results]
+
+    def run_as_string(self, query: str, num_results: int = 10, getText: int = 1):
+        results = self.run(query, num_results, getText)
+        rstring = "\n".join([self.pp_ayah(r) for r in results])
+        return rstring

util/prompt_mgr.py

@@ -0,0 +1,33 @@
+from typing import Union
+
+from pydantic import BaseModel
+
+
+class Prompt(BaseModel):
+    file_path: str
+    cached: Union[str, None] = None
+    hot_reload: bool = True
+
+    def render(self, **kwargs) -> str:
+        if (self.cached is None) or (self.hot_reload):
+            with open(self.file_path, "r") as f:
+                self.cached = f.read()
+        return self.cached.format(**kwargs)
+
+
+class PromptMgr:
+    def __init__(self, hot_reload: bool = True, src_dir: str = "resources/prompts"):
+        """Creates a prompt manager.
+
+        Args:
+            hot_reload: If true, reloads the prompt every time it is called.
+            src_dir: The directory where the prompts are stored.
+
+        """
+        self.hot_reload = hot_reload
+        self.src_dir = src_dir
+
+    def bind(self, prompt_id: str) -> Prompt:
+        return Prompt(
+            file_path=f"{self.src_dir}/{prompt_id}.txt", hot_reload=self.hot_reload
+        )