Spaces:
Running
Running
File size: 5,805 Bytes
72eef4f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 |
# app.py
import os
import requests # To download the audio file from Twilio's URL
from flask import Flask, request
from dotenv import load_dotenv
# Use the new top-level import and types
from google import genai
from google.genai import types
from twilio.twiml.messaging_response import MessagingResponse
# --- Define Calculator Functions for the Model to Use ---
# The SDK uses the function signature (name, parameters) and docstring
# to tell the model how and when to use these tools.
def add(a: float, b: float):
"""
Adds two numbers together.
Args:
a: The first number.
b: The second number.
"""
print(f"Tool Call: add(a={a}, b={b})")
return a + b
def subtract(a: float, b: float):
"""
Subtracts the second number from the first.
Args:
a: The number to subtract from.
b: The number to subtract.
"""
print(f"Tool Call: subtract(a={a}, b={b})")
return a - b
def multiply(a: float, b: float):
"""
Multiplies two numbers.
Args:
a: The first number.
b: The second number.
"""
print(f"Tool Call: multiply(a={a}, b={b})")
return a * b
def divide(a: float, b: float):
"""
Divides the first number by the second.
Args:
a: The numerator.
b: The denominator.
"""
print(f"Tool Call: divide(a={a}, b={b})")
if b == 0:
return "Error: Cannot divide by zero."
return a / b
# A list of all the functions the model can call
calculator_tools = [add, subtract, multiply, divide]
# --- Initialize Flask App and APIs ---
# Load environment variables from .env file
load_dotenv(r"C:\Users\Vaibhav Arora\Documents\MyExperimentsandCodes\APPS_WEBSITES\CANADA_WHOLESALE_PROJECT\GITHUB_REPOS\mvp-vue\wholesale-grocery-app\AIAPPS\.env")
app = Flask(__name__)
# Configure the Gemini API and initialize the client
try:
client = genai.Client(api_key=os.environ.get("GEMINI_API_KEY"))
print("Gemini client and calculator tools initialized successfully.")
except Exception as e:
print(f"Error initializing Gemini client: {e}")
client = None
# --- Define the Webhook Endpoint ---
@app.route("/sms", methods=['POST'])
def sms_reply():
"""Respond to incoming text or audio messages, using calculator functions if needed."""
twilio_resp = MessagingResponse()
if not client:
twilio_resp.message("The AI client is not configured correctly. Please check the server logs.")
return str(twilio_resp)
# Prepare the contents list for the Gemini API call
contents = []
# Check if the incoming message contains media
num_media = int(request.values.get('NumMedia', 0))
if num_media > 0:
media_url = request.values.get('MediaUrl0')
mime_type = request.values.get('MediaContentType0')
# Process only if the media is audio
if 'audio' in mime_type:
print(f"Received audio message. URL: {media_url}, MIME Type: {mime_type}")
# Download the audio file from the Twilio URL
audio_response = requests.get(media_url)
if audio_response.status_code == 200:
audio_bytes = audio_response.content
audio_part = types.Part.from_bytes(data=audio_bytes, mime_type=mime_type)
prompt = "Please transcribe this audio. If it contains a calculation or a question, please answer it."
contents = [prompt, audio_part]
else:
error_message = "Sorry, I couldn't download the audio file to process it. Please try again."
twilio_resp.message(error_message)
return str(twilio_resp)
else:
# Handle non-audio media like images or videos
twilio_resp.message("Sorry, I can only process text and audio messages.")
return str(twilio_resp)
else:
# Fallback to text message processing
incoming_msg = request.values.get('Body', '').strip()
print(f"Received text message: '{incoming_msg}'")
if not incoming_msg:
twilio_resp.message("Please send a text or audio message to get a response.")
return str(twilio_resp)
contents = [incoming_msg]
if not contents:
twilio_resp.message("Could not determine content to process. Please send a message.")
return str(twilio_resp)
try:
print("Sending content to Gemini with calculator tools...")
# Configure the request to use our calculator functions
config = types.GenerateContentConfig(tools=calculator_tools)
# The SDK handles the multi-turn process for function calling automatically
gemini_response = client.models.generate_content(
model="gemini-2.5-flash",
contents=contents, # This will contain either text or [prompt, audio_part]
config=config,
)
# This is the final text answer after any function calls have been resolved.
ai_answer = gemini_response.text
print(f"Gemini final response: '{ai_answer}'")
# Add the Gemini response to the TwiML message
twilio_resp.message(ai_answer)
except Exception as e:
print(f"An error occurred with the Gemini API: {e}")
# Send a user-friendly error message
error_message = "Sorry, I'm having trouble connecting to my brain right now. Please try again later."
twilio_resp.message(error_message)
return str(twilio_resp)
# --- Run the Flask App ---
if __name__ == "__main__":
app.run(debug=True, port=5000) |