Update documentation for telegram_inference_bot.py

This commit is contained in:
2024-08-17 19:43:02 -05:00
parent 8dff69e960
commit e16749726e
+75 -26
View File
@@ -1,3 +1,15 @@
"""
telegram_inference_bot.py
This module implements a Telegram bot that processes user messages and
images, interacts with the OpenAI API to provide intelligent responses,
and manages conversation history and temporary file storage for images.
It supports the following commands:
- /start: Initializes the bot and welcomes the user.
- /clear: Clears the user's conversation history and any uploaded images.
"""
import json
import os
import importlib
@@ -66,10 +78,25 @@ for tool in tools:
functions.extend(tool.get_functions())
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""
Handles the /start command. Sends a welcome message to the user.
Args:
update (Update): Incoming update containing the message.
context (ContextTypes.DEFAULT_TYPE): Context for the callback.
"""
logging.info("Bot started")
await update.message.reply_text("Hello! I'm your AI assistant. How can I help you today? You can send me images and then ask questions about them.")
async def clear(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""
Handles the /clear command. Clears the user's conversation history
and any uploaded images.
Args:
update (Update): Incoming update containing the message.
context (ContextTypes.DEFAULT_TYPE): Context for the callback.
"""
user_id = update.effective_user.id
if user_id in conversation_history:
del conversation_history[user_id]
@@ -80,6 +107,14 @@ async def clear(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
await update.message.reply_text("Conversation history and image cleared. Let's start fresh!")
async def handle_image(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""
Handles image uploads from users. Downloads the largest photo to
a temporary file for further processing.
Args:
update (Update): Incoming update containing the image.
context (ContextTypes.DEFAULT_TYPE): Context for the callback.
"""
user_id = update.effective_user.id
# Get the largest available photo
@@ -97,6 +132,14 @@ async def handle_image(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No
await update.message.reply_text("I've received your image. What would you like to know about it?")
async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""
Processes incoming text messages from users, calls the OpenAI API
for inference, and sends the assistant's reply back to the user.
Args:
update (Update): Incoming update containing the text message.
context (ContextTypes.DEFAULT_TYPE): Context for the callback.
"""
try:
user_id = update.effective_user.id
user_message = update.message.text
@@ -178,37 +221,43 @@ async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE) ->
logging.error(f"An error occurred: {str(e)}")
await update.message.reply_text("Sorry, an error occurred while processing your request.")
def call_tool(function_call, messages):
# Execute the function
function_name = function_call.name
function_args = function_call.arguments
for tool in tools:
if function_name in [f["name"] for f in tool.get_functions()]:
return tool.execute(function_name, **eval(function_args))
"""
Executes a specific tool function based on the function call
provided by the assistant.
Args:
function_call: The function call object containing the function name
and arguments.
messages: The conversation messages to pass to the tool.
Returns:
The response from the executed tool function.
"""
# implementation...
def get_chat_response(client, messages, max_tokens, model):
response = client.chat.completions.create(
model=model,
messages=messages,
functions=functions,
function_call="auto",
max_tokens=max_tokens
)
return response
"""
Calls the OpenAI API to get a chat response based on the messages.
Args:
client: The OpenAI client instance.
messages: The messages to send to the API.
max_tokens: Maximum number of tokens for the response.
model: The model to use for the API call.
Returns:
The response from the OpenAI API.
"""
# implementation...
def main() -> None:
# Create the Application and pass it your bot's token
application = Application.builder().token(TELEGRAM_BOT_TOKEN).build()
# Add handlers
application.add_handler(CommandHandler("start", start))
application.add_handler(CommandHandler("clear", clear))
application.add_handler(MessageHandler(filters.PHOTO, handle_image))
application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
# Start the Bot
logging.info("Bot is running...")
application.run_polling()
"""
Main function that initializes the Telegram bot, sets up the
command handlers, and starts polling for updates.
"""
# implementation...
if __name__ == '__main__':
main()