mirror of
https://github.com/vee1e/InstaFuel_Chatbot_public.git
synced 2026-09-01 10:49:13 +00:00
342 lines
12 KiB
Python
342 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Professional Chat Interface for Fitness Supplement Chatbot
|
|
Uses Google Gemini API for natural conversation with clean output
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import os
|
|
import sys
|
|
|
|
# Configure logging to suppress Google library warnings and log to file
|
|
logging.basicConfig(
|
|
level=logging.WARNING,
|
|
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
|
handlers=[logging.FileHandler("chatbot.log"), logging.StreamHandler(sys.stderr)],
|
|
)
|
|
|
|
# Suppress specific Google library logs
|
|
logging.getLogger("absl").setLevel(logging.ERROR)
|
|
logging.getLogger("grpc").setLevel(logging.ERROR)
|
|
logging.getLogger("google").setLevel(logging.ERROR)
|
|
logging.getLogger("google.auth").setLevel(logging.ERROR)
|
|
logging.getLogger("google.cloud").setLevel(logging.ERROR)
|
|
logging.getLogger("google.generativeai").setLevel(logging.ERROR)
|
|
|
|
# Also suppress via environment variables
|
|
os.environ["GRPC_VERBOSITY"] = "ERROR"
|
|
os.environ["GRPC_TRACE"] = ""
|
|
os.environ["TF_CPP_MIN_LOG_LEVEL"] = "3"
|
|
|
|
# Add src to path to allow imports
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "src")))
|
|
|
|
# Load environment variables
|
|
try:
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
except ImportError:
|
|
logging.warning("python-dotenv not found. Using environment variables directly.")
|
|
|
|
try:
|
|
from llm.models import Model, load_model_config
|
|
except ImportError:
|
|
print("ERROR: Model class not found. Please ensure src/llm/models.py exists.")
|
|
sys.exit(1)
|
|
|
|
|
|
class SimpleChatbot:
|
|
"""Simple chatbot using LiteLLM for natural conversation."""
|
|
|
|
# Define gratitude detection sets as class variables for efficiency
|
|
_GRATITUDE_KEYWORDS = frozenset([
|
|
"thank you",
|
|
"thanks",
|
|
"appreciate",
|
|
"much obliged",
|
|
"grateful",
|
|
])
|
|
_FOLLOW_UP_MARKERS = frozenset([
|
|
"?",
|
|
"but",
|
|
"also",
|
|
"another",
|
|
"else",
|
|
"recommend",
|
|
"suggest",
|
|
])
|
|
# Pre-computed gratitude response as a class variable
|
|
_GRATITUDE_RESPONSE = (
|
|
"You're very welcome! If you want to fine-tune your stack or have new goals, just let me know. "
|
|
"Keep up the great work with your training!"
|
|
)
|
|
|
|
def __init__(self):
|
|
"""Initialize the chatbot with the configured LiteLLM model."""
|
|
self.model = Model()
|
|
self.config = self.model.config
|
|
self.product_catalog = self._load_product_catalog()
|
|
self.catalog_context = self._build_catalog_context(self.product_catalog)
|
|
self.history = []
|
|
self._display_welcome()
|
|
|
|
def _get_system_prompt(self):
|
|
"""Get the professional system prompt."""
|
|
return f"""
|
|
You are a professional fitness supplement consultant for InstaFuel.
|
|
|
|
Communication style:
|
|
- Professional, encouraging, and easy to follow
|
|
- Cite concrete product details from the catalog
|
|
- Explain why each suggestion fits the user's stated goals or constraints
|
|
|
|
Grounding requirements:
|
|
- Base every recommendation ONLY on the product catalog provided below. Do not invent items.
|
|
- If nothing in the catalog fits, clearly state that no available product matches and invite the user to share more details.
|
|
- When offering a product, include its name, price, key benefits, and shopping link in natural language. Bullet points or short paragraphs are both acceptable.
|
|
- When the user asks for a complementary item, use the catalog's stack_partners to explain why the pairing works.
|
|
- If the user expresses gratitude without a new question, respond warmly and avoid introducing new products unless they ask for more.
|
|
|
|
Product catalog:
|
|
{self.catalog_context}
|
|
"""
|
|
|
|
@staticmethod
|
|
def _load_product_catalog():
|
|
"""Return an in-memory catalog to ground the assistant's suggestions."""
|
|
# Prefer a path relative to this file so the script works when invoked
|
|
# from other working directories. Give clear errors on failure.
|
|
catalog_path = os.path.abspath(
|
|
os.path.join(os.path.dirname(__file__), "data", "all_products.json")
|
|
)
|
|
try:
|
|
with open(catalog_path, "r", encoding="utf-8") as fh:
|
|
data = json.load(fh)
|
|
|
|
if not isinstance(data, list):
|
|
raise RuntimeError(
|
|
f"Product catalog must be a JSON list, got {type(data)}"
|
|
)
|
|
|
|
return data
|
|
except FileNotFoundError:
|
|
raise RuntimeError(f"Product catalog not found at {catalog_path}")
|
|
except json.JSONDecodeError as exc:
|
|
raise RuntimeError(f"Failed to parse product catalog JSON: {exc}")
|
|
except Exception as exc:
|
|
raise RuntimeError(f"Unable to load product catalog: {exc}")
|
|
|
|
@staticmethod
|
|
def _build_catalog_context(products):
|
|
"""Render the catalog into a concise text block for the system prompt."""
|
|
lines = []
|
|
for product in products:
|
|
# Basic identity
|
|
pid = product.get("id", "unknown-id")
|
|
name = product.get("name", "Unnamed product")
|
|
lines.append(f"- {name} (id: {pid})")
|
|
|
|
# Meta: category / price / link
|
|
meta = []
|
|
if product.get("category"):
|
|
meta.append(f"Category: {product['category']}")
|
|
price = product.get("price")
|
|
if isinstance(price, (int, float)):
|
|
meta.append(f"Price: ${price:.2f}")
|
|
if product.get("link"):
|
|
meta.append(f"Link: {product['link']}")
|
|
if meta:
|
|
lines.append(f" {' | '.join(meta)}")
|
|
|
|
# Benefits
|
|
benefits = product.get("benefits")
|
|
if benefits:
|
|
if isinstance(benefits, list):
|
|
lines.append(f" Benefits: {', '.join(benefits)}")
|
|
else:
|
|
lines.append(f" Benefits: {benefits}")
|
|
|
|
# Nutrition (dict -> key: value)
|
|
nutrition = product.get("nutrition")
|
|
if nutrition and isinstance(nutrition, dict):
|
|
nut_parts = []
|
|
for k, v in nutrition.items():
|
|
label = str(k).replace("_", " ")
|
|
nut_parts.append(f"{label}: {v}")
|
|
lines.append(f" Nutrition: {', '.join(nut_parts)}")
|
|
|
|
# Directions of use
|
|
directions = product.get("directions_of_use") or product.get("directions")
|
|
if directions:
|
|
if isinstance(directions, list):
|
|
lines.append(f" Directions: {'; '.join(directions)}")
|
|
else:
|
|
lines.append(f" Directions: {directions}")
|
|
|
|
# Image (optional)
|
|
if product.get("image_url"):
|
|
lines.append(f" Image: {product['image_url']}")
|
|
lines.append("")
|
|
return "\n".join(lines).strip()
|
|
|
|
@classmethod
|
|
def _is_gratitude(cls, message: str) -> bool:
|
|
"""Return True when the message is primarily gratitude without a new request."""
|
|
lowered = message.lower()
|
|
# Use class-level frozensets for O(1) lookups instead of recreating lists
|
|
has_gratitude = any(keyword in lowered for keyword in cls._GRATITUDE_KEYWORDS)
|
|
has_followup = any(marker in lowered for marker in cls._FOLLOW_UP_MARKERS)
|
|
return has_gratitude and not has_followup
|
|
|
|
@classmethod
|
|
def _gratitude_response(cls) -> str:
|
|
"""Friendly closing response when user expresses gratitude."""
|
|
return cls._GRATITUDE_RESPONSE
|
|
|
|
def _display_welcome(self):
|
|
"""Display professional welcome message."""
|
|
print("\n" + "=" * 60)
|
|
print(" INSTAFUEL FITNESS SUPPLEMENT ASSISTANT")
|
|
print("=" * 60)
|
|
print("Welcome! I'm your professional fitness supplement consultant.")
|
|
print("I can help you with:")
|
|
print("• Personalized supplement recommendations")
|
|
print("• Product information and usage guidance")
|
|
print("• Fitness and nutrition advice")
|
|
print("• Order support and tracking")
|
|
print("\nCurrent model configuration:")
|
|
print(f"• Provider: {self.config.provider}")
|
|
print(f"• Model: {self.config.name}")
|
|
print("\nType your question below, or 'quit' to exit.")
|
|
print("-" * 60 + "\n")
|
|
|
|
def chat_loop(self):
|
|
"""Main chat loop for user interaction."""
|
|
while True:
|
|
try:
|
|
# Get user input with professional formatting
|
|
user_input = input("You: ").strip()
|
|
|
|
# Check for exit commands
|
|
if user_input.lower() in ["quit", "exit", "bye", "goodbye"]:
|
|
self._display_goodbye()
|
|
break
|
|
|
|
if not user_input:
|
|
continue
|
|
|
|
# Show processing indicator
|
|
print("Assistant: ", end="", flush=True)
|
|
|
|
# Add user message to history
|
|
self.history.append({"role": "user", "content": user_input})
|
|
|
|
if self._is_gratitude(user_input):
|
|
response_text = self._gratitude_response()
|
|
else:
|
|
response_text = self.model.respond(
|
|
conv_history=self.history,
|
|
system_prompt=self._get_system_prompt(),
|
|
)
|
|
|
|
# Add assistant response to history
|
|
self.history.append({"role": "assistant", "content": response_text})
|
|
|
|
# Format and display response
|
|
self._display_response(response_text)
|
|
|
|
except KeyboardInterrupt:
|
|
self._display_goodbye()
|
|
break
|
|
except Exception as e:
|
|
logging.error(f"Chat error: {e}")
|
|
print(f"\nERROR: {str(e)}")
|
|
print(
|
|
"Assistant: I apologize, but I'm experiencing a technical issue. Please try again.\n"
|
|
)
|
|
|
|
def _display_response(self, response_text):
|
|
"""Display response with professional formatting."""
|
|
# Clean up the response text
|
|
response_text = response_text.strip()
|
|
|
|
# Handle multi-line responses with proper formatting
|
|
lines = response_text.split("\n")
|
|
first_line = True
|
|
|
|
for line in lines:
|
|
if first_line:
|
|
print(line)
|
|
first_line = False
|
|
else:
|
|
# Indent continuation lines
|
|
if line.strip():
|
|
print(f" {line}")
|
|
else:
|
|
print()
|
|
|
|
print() # Add spacing after response
|
|
|
|
def _display_goodbye(self):
|
|
"""Display professional goodbye message."""
|
|
print("\n" + "-" * 60)
|
|
print("Thank you for consulting with InstaFuel!")
|
|
print("Stay committed to your fitness journey.")
|
|
print("We're here to support your goals every step of the way.")
|
|
print("-" * 60)
|
|
|
|
def get_chat_history(self):
|
|
"""Get the conversation history."""
|
|
return self.history
|
|
|
|
|
|
async def main():
|
|
"""Main function to run the chatbot."""
|
|
# Clear screen for clean start
|
|
os.system("cls" if os.name == "nt" else "clear")
|
|
|
|
print("Initializing InstaFuel Fitness Assistant...")
|
|
print("Loading AI models and configurations...")
|
|
|
|
try:
|
|
config = load_model_config()
|
|
print(f"Configured provider: {config.provider}")
|
|
print(f"Configured model: {config.name}")
|
|
except RuntimeError as exc:
|
|
logging.error("Model configuration error: %s", exc)
|
|
print("\n" + "!" * 60)
|
|
print("CONFIGURATION ERROR")
|
|
print("!" * 60)
|
|
print(str(exc))
|
|
print("!" * 60)
|
|
return
|
|
|
|
try:
|
|
chatbot = SimpleChatbot()
|
|
chatbot.chat_loop()
|
|
|
|
except RuntimeError as exc:
|
|
logging.error("Chatbot initialization error: %s", exc)
|
|
print("\n" + "!" * 60)
|
|
print("INITIALIZATION ERROR")
|
|
print("!" * 60)
|
|
print(str(exc))
|
|
print("!" * 60)
|
|
except Exception as e:
|
|
logging.error(f"Critical error in main: {e}")
|
|
print("\n" + "!" * 60)
|
|
print("SYSTEM ERROR")
|
|
print("!" * 60)
|
|
print(f"Failed to start chatbot: {str(e)}")
|
|
print("Please check:")
|
|
print("1. Your internet connection")
|
|
print("2. The configured API key is valid")
|
|
print("3. All required packages are installed")
|
|
print("!" * 60)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|