{ "cells": [ { "cell_type": "code", "execution_count": null, "metadata": { "id": "u2_t_yaIyHSc" }, "outputs": [], "source": [ "import os\n", "import openai\n", "import sys\n", "sys.path.append('../..')\n", "import utils\n", "\n", "import panel as pn # GUI\n", "pn.extension()\n", "\n", "from dotenv import load_dotenv, find_dotenv\n", "_ = load_dotenv(find_dotenv()) # read local .env file\n", "\n", "openai.api_key = os.environ['OPENAI_API_KEY']" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "1YOdJ1dhyKH_" }, "outputs": [], "source": [ "def get_completion_from_messages(messages, model=\"gpt-3.5-turbo\", temperature=0, max_tokens=500):\n", " response = openai.ChatCompletion.create(\n", " model=model,\n", " messages=messages,\n", " temperature=temperature,\n", " max_tokens=max_tokens,\n", " )\n", " return response.choices[0].message[\"content\"]" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "Z25P1M2jyKKj" }, "outputs": [], "source": [ "def process_user_message(user_input, all_messages, debug=True):\n", " delimiter = \"```\"\n", "\n", " # Step 1: Check input to see if it flags the Moderation API or is a prompt injection\n", " response = openai.Moderation.create(input=user_input)\n", " moderation_output = response[\"results\"][0]\n", "\n", " if moderation_output[\"flagged\"]:\n", " print(\"Step 1: Input flagged by Moderation API.\")\n", " return \"Sorry, we cannot process this request.\"\n", "\n", " if debug: print(\"Step 1: Input passed moderation check.\")\n", "\n", " category_and_product_response = utils.find_category_and_product_only(user_input, utils.get_products_and_category())\n", " #print(print(category_and_product_response)\n", " # Step 2: Extract the list of products\n", " category_and_product_list = utils.read_string_to_list(category_and_product_response)\n", " #print(category_and_product_list)\n", "\n", " if debug: print(\"Step 2: Extracted list of products.\")\n", "\n", " # Step 3: If products are found, look them up\n", " product_information = utils.generate_output_string(category_and_product_list)\n", " if debug: print(\"Step 3: Looked up product information.\")\n", "\n", " # Step 4: Answer the user question\n", " system_message = f\"\"\"\n", " You are a customer service assistant for a large electronic store. \\\n", " Respond in a friendly and helpful tone, with concise answers. \\\n", " Make sure to ask the user relevant follow-up questions.\n", " \"\"\"\n", " messages = [\n", " {'role': 'system', 'content': system_message},\n", " {'role': 'user', 'content': f\"{delimiter}{user_input}{delimiter}\"},\n", " {'role': 'assistant', 'content': f\"Relevant product information:\\n{product_information}\"}\n", " ]\n", "\n", " final_response = get_completion_from_messages(all_messages + messages)\n", " if debug:print(\"Step 4: Generated response to user question.\")\n", " all_messages = all_messages + messages[1:]\n", "\n", " # Step 5: Put the answer through the Moderation API\n", " response = openai.Moderation.create(input=final_response)\n", " moderation_output = response[\"results\"][0]\n", "\n", " if moderation_output[\"flagged\"]:\n", " if debug: print(\"Step 5: Response flagged by Moderation API.\")\n", " return \"Sorry, we cannot provide this information.\"\n", "\n", " if debug: print(\"Step 5: Response passed moderation check.\")\n", "\n", " # Step 6: Ask the model if the response answers the initial user query well\n", " user_message = f\"\"\"\n", " Customer message: {delimiter}{user_input}{delimiter}\n", " Agent response: {delimiter}{final_response}{delimiter}\n", "\n", " Does the response sufficiently answer the question?\n", " \"\"\"\n", " messages = [\n", " {'role': 'system', 'content': system_message},\n", " {'role': 'user', 'content': user_message}\n", " ]\n", " evaluation_response = get_completion_from_messages(messages)\n", " if debug: print(\"Step 6: Model evaluated the response.\")\n", "\n", " # Step 7: If yes, use this answer; if not, say that you will connect the user to a human\n", " if \"Y\" in evaluation_response: # Using \"in\" instead of \"==\" to be safer for model output variation (e.g., \"Y.\" or \"Yes\")\n", " if debug: print(\"Step 7: Model approved the response.\")\n", " return final_response, all_messages\n", " else:\n", " if debug: print(\"Step 7: Model disapproved the response.\")\n", " neg_str = \"I'm unable to provide the information you're looking for. I'll connect you with a human representative for further assistance.\"\n", " return neg_str, all_messages\n", "\n", "user_input = \"tell me about the smartx pro phone and the fotosnap camera, the dslr one. Also what tell me about your tvs\"\n", "response,_ = process_user_message(user_input,[])\n", "print(response)" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "mtAZM_EJyKNL" }, "outputs": [], "source": [ "def collect_messages(debug=False):\n", " user_input = inp.value_input\n", " if debug: print(f\"User Input = {user_input}\")\n", " if user_input == \"\":\n", " return\n", " inp.value = ''\n", " global context\n", " #response, context = process_user_message(user_input, context, utils.get_products_and_category(),debug=True)\n", " response, context = process_user_message(user_input, context, debug=False)\n", " context.append({'role':'assistant', 'content':f\"{response}\"})\n", " panels.append(\n", " pn.Row('User:', pn.pane.Markdown(user_input, width=600)))\n", " panels.append(\n", " pn.Row('Assistant:', pn.pane.Markdown(response, width=600, style={'background-color': '#F6F6F6'})))\n", "\n", " return pn.Column(*panels)" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "BDCSKqdmyKPr" }, "outputs": [], "source": [ "panels = [] # collect display\n", "\n", "context = [ {'role':'system', 'content':\"You are Service Assistant\"} ]\n", "\n", "inp = pn.widgets.TextInput( placeholder='Enter text here…')\n", "button_conversation = pn.widgets.Button(name=\"Service Assistant\")\n", "\n", "interactive_conversation = pn.bind(collect_messages, button_conversation)\n", "\n", "dashboard = pn.Column(\n", " inp,\n", " pn.Row(button_conversation),\n", " pn.panel(interactive_conversation, loading_indicator=True, height=300),\n", ")\n", "\n", "dashboard" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "h4x_VfZNyKSf" }, "outputs": [], "source": [] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "Zchl49g2yKVg" }, "outputs": [], "source": [] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "OmGP6B6lyKXk" }, "outputs": [], "source": [] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "oHbmL7_SyKaS" }, "outputs": [], "source": [] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "hjvXWF-6yKdB" }, "outputs": [], "source": [] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "Bc1YQguFyKfb" }, "outputs": [], "source": [] } ], "metadata": { "colab": { "provenance": [] }, "kernelspec": { "display_name": "Python 3 (ipykernel)", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.10.6" } }, "nbformat": 4, "nbformat_minor": 1 }