import os import time import asyncio import logging import sqlite3 import tiktoken from uuid import uuid4 from functools import lru_cache from typing import Optional, List, Dict, Literal from fastapi import FastAPI, HTTPException, Depends, Security, BackgroundTasks from fastapi.security import APIKeyHeader from fastapi.responses import StreamingResponse from pydantic import BaseModel, Field from openai import OpenAI # ============================================================================ # Configuration and Setup # ============================================================================ # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler("app.log"), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) # FastAPI app setup app = FastAPI() # API key configuration API_KEY_NAME = "X-API-Key" API_KEY = os.environ.get("CHAT_AUTH_KEY", "default_secret_key") api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False) # Model definitions ModelID = Literal[ "openai/gpt-4o-mini", "meta-llama/llama-3-70b-instruct", "anthropic/claude-3.5-sonnet", "deepseek/deepseek-coder", "anthropic/claude-3-haiku", "openai/gpt-3.5-turbo-instruct", "qwen/qwen-72b-chat", "google/gemma-2-27b-it" ] # Pydantic models class LLMAgentQueryModel(BaseModel): prompt: str = Field(..., description="User's query or prompt") system_message: Optional[str] = Field(None, description="Custom system message for the conversation") model_id: ModelID = Field( default="openai/gpt-4o-mini", description="ID of the model to use for response generation" ) conversation_id: Optional[str] = Field(None, description="Unique identifier for the conversation") user_id: str = Field(..., description="Unique identifier for the user") class Config: schema_extra = { "example": { "prompt": "How do I implement a binary search in Python?", "system_message": "You are a helpful coding assistant.", "model_id": "meta-llama/llama-3-70b-instruct", "conversation_id": "123e4567-e89b-12d3-a456-426614174000", "user_id": "user123" } } # API key and client setup @lru_cache() def get_api_keys(): logger.info("Loading API keys") return { "OPENROUTER_API_KEY": f"sk-or-v1-{os.environ['OPENROUTER_API_KEY']}", } api_keys = get_api_keys() or_client = OpenAI(api_key=api_keys["OPENROUTER_API_KEY"], base_url="https://openrouter.ai/api/v1") # In-memory storage for conversations conversations: Dict[str, List[Dict[str, str]]] = {} last_activity: Dict[str, float] = {} # Token encoding encoding = tiktoken.encoding_for_model("gpt-3.5-turbo") # ============================================================================ # Database Functions # ============================================================================ DB_PATH = '/app/data/conversations.db' def init_db(): logger.info("Initializing database") os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) conn = sqlite3.connect(DB_PATH) c = conn.cursor() c.execute('''CREATE TABLE IF NOT EXISTS conversations (id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT, conversation_id TEXT, message TEXT, response TEXT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP)''') conn.commit() conn.close() logger.info("Database initialized successfully") def update_db(user_id, conversation_id, message, response): logger.info(f"Updating database for conversation: {conversation_id}") conn = sqlite3.connect(DB_PATH) c = conn.cursor() c.execute('''INSERT INTO conversations (user_id, conversation_id, message, response) VALUES (?, ?, ?, ?)''', (user_id, conversation_id, message, response)) conn.commit() conn.close() logger.info("Database updated successfully") # ============================================================================ # Utility Functions # ============================================================================ def calculate_tokens(msgs): return sum(len(encoding.encode(str(m))) for m in msgs) def limit_conversation_history(conversation: List[Dict[str, str]], max_tokens: int = 4000) -> List[Dict[str, str]]: """Limit the conversation history to a maximum number of tokens.""" limited_conversation = [] current_tokens = 0 for message in reversed(conversation): message_tokens = calculate_tokens([message]) if current_tokens + message_tokens > max_tokens: break limited_conversation.insert(0, message) current_tokens += message_tokens return limited_conversation async def verify_api_key(api_key: str = Security(api_key_header)): if api_key != API_KEY: logger.warning("Invalid API key used") raise HTTPException(status_code=403, detail="Could not validate credentials") return api_key # ============================================================================ # LLM Interaction Functions # ============================================================================ def chat_with_llama_stream(messages, model="meta-llama/llama-3-70b-instruct", max_output_tokens=2500): logger.info(f"Starting chat with model: {model}") try: response = or_client.chat.completions.create( model=model, messages=messages, max_tokens=max_output_tokens, stream=True ) full_response = "" for chunk in response: if chunk.choices[0].delta.content is not None: content = chunk.choices[0].delta.content full_response += content yield content # After streaming, add the full response to the conversation history messages.append({"role": "assistant", "content": full_response}) logger.info("Chat completed successfully") except Exception as e: logger.error(f"Error in model response: {str(e)}") raise HTTPException(status_code=500, detail=f"Error in model response: {str(e)}") # ============================================================================ # Background Tasks # ============================================================================ async def clear_inactive_conversations(): while True: logger.info("Clearing inactive conversations") current_time = time.time() inactive_convos = [conv_id for conv_id, last_time in last_activity.items() if current_time - last_time > 1800] # 30 minutes for conv_id in inactive_convos: if conv_id in conversations: del conversations[conv_id] if conv_id in last_activity: del last_activity[conv_id] logger.info(f"Cleared {len(inactive_convos)} inactive conversations") await asyncio.sleep(60) # Check every minute # ============================================================================ # FastAPI Events and Endpoints # ============================================================================ @app.on_event("startup") async def startup_event(): logger.info("Starting up the application") init_db() asyncio.create_task(clear_inactive_conversations()) @app.post("/llm-agent") async def llm_agent(query: LLMAgentQueryModel, background_tasks: BackgroundTasks, api_key: str = Depends(verify_api_key)): """ LLM agent endpoint that provides responses based on user queries, maintaining conversation history. Accepts custom system messages and allows selection of different models. Requires API Key authentication via X-API-Key header. """ logger.info(f"Received LLM agent query: {query.prompt}") # Generate a new conversation ID if not provided if not query.conversation_id: query.conversation_id = str(uuid4()) # Initialize or retrieve conversation history if query.conversation_id not in conversations: system_message = query.system_message or "You are a helpful assistant. Provide concise and accurate responses." conversations[query.conversation_id] = [ {"role": "system", "content": system_message} ] elif query.system_message: # Update system message if provided conversations[query.conversation_id][0] = {"role": "system", "content": query.system_message} # Add user's prompt to conversation history conversations[query.conversation_id].append({"role": "user", "content": query.prompt}) last_activity[query.conversation_id] = time.time() # Limit tokens in the conversation history limited_conversation = limit_conversation_history(conversations[query.conversation_id]) def process_response(): full_response = "" for content in chat_with_llama_stream(limited_conversation, model=query.model_id): full_response += content yield content # Add the assistant's response to the conversation history conversations[query.conversation_id].append({"role": "assistant", "content": full_response}) background_tasks.add_task(update_db, query.user_id, query.conversation_id, query.prompt, full_response) logger.info(f"Completed LLM agent response for query: {query.prompt}") return StreamingResponse(process_response(), media_type="text/event-stream") import edge_tts import io @app.get("/tts") async def text_to_speech( text: str = Query(..., description="Text to convert to speech"), voice: str = Query(default="en-GB-SoniaNeural", description="Voice to use for speech") ): communicate = edge_tts.Communicate(text, voice) async def generate(): async for chunk in communicate.stream(): if chunk["type"] == "audio": yield chunk["data"] return StreamingResponse(generate(), media_type="audio/mpeg") # ============================================================================ # Main Execution # ============================================================================ if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)