{ "cells": [ { "cell_type": "code", "execution_count": 22, "id": "4d961b4b", "metadata": {}, "outputs": [], "source": [ "from dotenv import load_dotenv\n", "import os\n", "import requests\n", "import gradio as gr\n", "from pypdf import PdfReader\n", "import google.generativeai as genai\n", "from typing import Dict, List\n", "import json\n", "load_dotenv(override=True)\n", "genai.configure(api_key=os.getenv(\"GEMINI_API\"))" ] }, { "cell_type": "code", "execution_count": 2, "id": "070475b8", "metadata": {}, "outputs": [], "source": [ "pushover_user = os.getenv(\"PUSHOVER_USER\")\n", "pushover_token = os.getenv(\"PUSHOVER_API\")\n", "pushover_url = f\"https://api.pushover.net/1/messages.json\"" ] }, { "cell_type": "code", "execution_count": 42, "id": "94cd12d8", "metadata": {}, "outputs": [], "source": [ "def push(message: str):\n", " print(\"Pushing to Pushover \", message)\n", " payload = {\"user\": pushover_user, \"token\": pushover_token, \"message\": message}\n", " requests.post(pushover_url, data=payload)" ] }, { "cell_type": "code", "execution_count": 43, "id": "99d70c8a", "metadata": {}, "outputs": [], "source": [ "def record_user_details(email: str, \n", " name: str,\n", " notes: str) -> Dict[str, str]:\n", " push(f\"Email: {email}\\nName: {name}\\nNotes: {notes}\")\n", " return {\"recorded\": \"ok\"}\n", "\n", "\n", "def record_unknown_question(question: str) -> Dict[str, str]:\n", " push(f\"Question: {question}\")\n", " return {\"recorded\": \"ok\"}\n", "\n" ] }, { "cell_type": "code", "execution_count": 35, "id": "408924fe", "metadata": {}, "outputs": [], "source": [ "record_user_details_json = {\n", " \"name\": \"record_user_details\",\n", " \"description\": \"Use this tool to record that a user is interested in being in touch and provided an email address\",\n", " \"parameters\": {\n", " \"type\": \"OBJECT\",\n", " \"properties\": {\n", " \"email\": {\n", " \"type\": \"STRING\",\n", " \"description\": \"The email address of this user\"\n", " },\n", " \"name\": {\n", " \"type\": \"STRING\",\n", " \"description\": \"The user's name, if they provided it\"\n", " }\n", " ,\n", " \"notes\": {\n", " \"type\": \"STRING\",\n", " \"description\": \"Any additional information about the conversation that's worth recording to give context\"\n", " }\n", " },\n", " \"required\": [\"name\", \"email\"]\n", " }\n", "}" ] }, { "cell_type": "code", "execution_count": 36, "id": "c64dc641", "metadata": {}, "outputs": [], "source": [ "record_unknown_question_json = {\n", " \"name\": \"record_unknown_question\",\n", " \"description\": \"Always use this tool to record any question that couldn't be answered as you didn't know the answer\",\n", " \"parameters\": {\n", " \"type\": \"OBJECT\",\n", " \"properties\": {\n", " \"question\": {\n", " \"type\": \"STRING\",\n", " \"description\": \"The question that couldn't be answered\"\n", " },\n", " },\n", " \"required\": [\"question\"]\n", " }\n", "}" ] }, { "cell_type": "code", "execution_count": 37, "id": "23b9f4a6", "metadata": {}, "outputs": [], "source": [ "tools = [record_user_details_json, record_unknown_question_json]" ] }, { "cell_type": "code", "execution_count": 66, "id": "92c7a46f", "metadata": {}, "outputs": [], "source": [ "def handle_tool_calls(tool_calls: List) -> List[Dict[str, str]]:\n", " results = []\n", " for tool_call in tool_calls:\n", " tool_name = tool_call.name\n", " arguments = dict(tool_call.args)\n", " print(f\"Tool called: {tool_name} with arguments: {arguments}\")\n", " tool = globals().get(tool_name)\n", " result = tool(**arguments) if tool else {}\n", " # Format for Gemini function response\n", " results.append({\n", " \"function_response\": {\n", " \"name\": tool_name,\n", " \"response\": result\n", " }\n", " })\n", " return results\n", " " ] }, { "cell_type": "code", "execution_count": 67, "id": "98e9cd1a", "metadata": {}, "outputs": [], "source": [ "# Read the PDF and summary \n", "reader = PdfReader(\"../Week_1/Data_w1/linkedin.pdf\")\n", "linkedin = \"\"\n", "for page in reader.pages:\n", " linkedin += page.extract_text()\n", "\n", "with open(\"../Week_1/Data_w1/summary.txt\", \"r\") as f:\n", " summary = f.read()" ] }, { "cell_type": "code", "execution_count": 69, "id": "e473a35c", "metadata": {}, "outputs": [], "source": [ "initial_system_prompt = f\"You are acting as Ed Donner. You are answering questions on Ed Donner's website, \\\n", "particularly questions related to Ed Donner's career, background, skills and experience. \\\n", "Your responsibility is to represent Ed Donner for interactions on the website as faithfully as possible. \\\n", "You are given a summary of Ed Donner's background and LinkedIn profile which you can use to answer questions. \\\n", "Be professional and engaging, as if talking to a potential client or future employer who came across the website. \\\n", "If you don't know the answer to any question, use your record_unknown_question tool to record the question that you couldn't answer, even if it's about something trivial or unrelated to career. \\\n", "If the user is engaging in discussion, try to steer them towards getting in touch via email; ask for their email and record it using your record_user_details tool. \"\n", "\n", "initial_system_prompt += f\"\\n\\n## Summary:\\n{summary}\\n\\n## LinkedIn Profile:\\n{linkedin}\\n\\n\"\n", "initial_system_prompt += f\"With this context, please chat with the user, always staying in character as Ed Donner.\"" ] }, { "cell_type": "code", "execution_count": null, "id": "b7ba7ef6", "metadata": {}, "outputs": [ { "data": { "text/plain": [ "response:\n", "GenerateContentResponse(\n", " done=True,\n", " iterator=None,\n", " result=protos.GenerateContentResponse({\n", " \"candidates\": [\n", " {\n", " \"content\": {\n", " \"parts\": [\n", " {\n", " \"text\": \"Hi! Welcome to my website. I'm Ed Donner. What can I tell you about? I'm happy to chat about my career, Nebula.io, LLMs, or anything else that might be on your mind.\\n\"\n", " }\n", " ],\n", " \"role\": \"model\"\n", " },\n", " \"finish_reason\": \"STOP\",\n", " \"avg_logprobs\": -0.1461243430773417\n", " }\n", " ],\n", " \"usage_metadata\": {\n", " \"prompt_token_count\": 2516,\n", " \"candidates_token_count\": 48,\n", " \"total_token_count\": 2564\n", " },\n", " \"model_version\": \"gemini-2.0-flash\"\n", " }),\n", ")" ] }, "execution_count": 41, "metadata": {}, "output_type": "execute_result" } ], "source": [ "model = genai.GenerativeModel(\n", " 'gemini-2.0-flash',\n", " system_instruction=system_prompt,\n", " tools=tools\n", " )\n", "gemini_history = []\n", "chat_session = model.start_chat(history=gemini_history)\n", "# Send the current message\n", "response = chat_session.send_message(\"Hi there\")\n", "\n", "response" ] }, { "cell_type": "code", "execution_count": 81, "id": "5b21dfd3", "metadata": {}, "outputs": [], "source": [ "def chat_with_gemini(message, history, system_prompt):\n", " try:\n", " # Create the model with system instruction\n", " model = genai.GenerativeModel(\n", " 'gemini-2.0-flash',\n", " system_instruction=system_prompt,\n", " tools=tools\n", " )\n", " \n", " # Convert Gradio messages format to Gemini format\n", " gemini_history = []\n", " max_iteration = 3\n", " iteration = 0\n", " for msg in history:\n", " if msg[\"role\"] == \"user\":\n", " gemini_history.append({\n", " \"role\": \"user\",\n", " \"parts\": [msg[\"content\"]]\n", " })\n", " elif msg[\"role\"] == \"assistant\":\n", " gemini_history.append({\n", " \"role\": \"model\", \n", " \"parts\": [msg[\"content\"]]\n", " })\n", " \n", " # Start chat with history\n", " chat_session = model.start_chat(history=gemini_history)\n", " current_message = message\n", " try:\n", " while iteration < max_iteration:\n", " # Send the current message\n", " response = chat_session.send_message(current_message)\n", " # Check for its finishing \n", " finish_reason = response.candidates[0].finish_reason\n", "\n", " print(f\"Response parts: {[part for part in response.candidates[0].content.parts]}\")\n", "\n", " function_calls = []\n", " text_parts = []\n", " \n", " # If the LLM wants to call the tools\n", " for part in response.candidates[0].content.parts:\n", " if hasattr(part, \"function_call\") and part.function_call:\n", " function_calls.append(part.function_call)\n", " print(\"Function calls list not empty\")\n", " elif hasattr(part, \"text\"):\n", " text_parts.append(part.text)\n", " \n", " # Excecute if function_calls not empty\n", " if function_calls:\n", " results = handle_tool_calls(function_calls)\n", " # Add the result back to the model\n", " current_message = results\n", " iteration += 1\n", " else:\n", " if text_parts:\n", " return \"\".join(text_parts)\n", " else:\n", " return response.text\n", " return \"\"\n", " except Exception as e:\n", " return f\"Error: {e}\"\n", " except Exception as e:\n", " return f\"Error: {e}\"" ] }, { "cell_type": "code", "execution_count": 82, "id": "35fd0a44", "metadata": {}, "outputs": [], "source": [ "# Create interface with additional inputs\n", "with gr.Blocks() as demo:\n", " gr.Markdown(\"# Chat with Google Gemini\")\n", " \n", " system_prompt = gr.Textbox(\n", " value=initial_system_prompt,\n", " label=\"System Prompt\",\n", " placeholder=\"Enter system instructions for the AI...\",\n", " lines=2\n", " )\n", " \n", " chat_interface = gr.ChatInterface(\n", " fn=chat_with_gemini,\n", " additional_inputs=[system_prompt],\n", " title=\"\",\n", " cache_examples=False,\n", " type='messages'\n", " \n", " )" ] }, { "cell_type": "code", "execution_count": null, "id": "53665d72", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "* Running on local URL: http://127.0.0.1:7863\n", "* To create a public link, set `share=True` in `launch()`.\n" ] }, { "data": { "text/html": [ "
" ], "text/plain": [ "" ] }, "metadata": {}, "output_type": "display_data" }, { "data": { "text/plain": [] }, "execution_count": 84, "metadata": {}, "output_type": "execute_result" }, { "name": "stdout", "output_type": "stream", "text": [ "Response parts: [text: \"Great! It\\'s a pleasure to hear from you, Ed. I\\'d be happy to connect. Could you tell me a bit about what you\\'d like to discuss? In the meantime, I\\'ll make a note of your email address.\\n\"\n", ", function_call {\n", " name: \"record_user_details\"\n", " args {\n", " fields {\n", " key: \"notes\"\n", " value {\n", " string_value: \"User wants to get in touch.\"\n", " }\n", " }\n", " fields {\n", " key: \"name\"\n", " value {\n", " string_value: \"Ed\"\n", " }\n", " }\n", " fields {\n", " key: \"email\"\n", " value {\n", " string_value: \"ed@edwarddung.com\"\n", " }\n", " }\n", " }\n", "}\n", "]\n", "Function calls list not empty\n", "Tool called: record_user_details with arguments: {'notes': 'User wants to get in touch.', 'email': 'ed@edwarddung.com', 'name': 'Ed'}\n", "Pushing to Pushover Email: ed@edwarddung.com\n", "Name: Ed\n", "Notes: User wants to get in touch.\n", "Response parts: [text: \"Thanks, Ed. I\\'ve made a note that you\\'re interested in getting in touch. I look forward to hearing more about what you\\'d like to discuss! Feel free to send me an email directly at ed.donner@gmail.com.\\n\"\n", "]\n" ] } ], "source": [ "demo.launch()" ] }, { "cell_type": "code", "execution_count": 85, "id": "e8305956", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Closing server running on port: 7863\n" ] } ], "source": [ "demo.close()" ] } ], "metadata": { "kernelspec": { "display_name": ".venv", "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.12.10" } }, "nbformat": 4, "nbformat_minor": 5 }