Building AlexAI: A Complete Local AI Assistant
Building AlexAI, a complete local AI assistant with Ollama, MCP, RAG, and pydantic-ai.
Part 1: Setting Up the Foundation
Project Structure and Dependencies
Let’s start by creating our project structure and understanding why each component is essential.
# Create project directory
mkdir alexai-assistant
cd alexai-assistant
# Create virtual environment
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
# Install core dependencies
pip install pydantic-ai[openai] ollama-python chromadb whisper-cpp-python pyttsx3 pytesseract pillow fastapi uvicorn asyncio-mqtt sqlalchemy alembicWhy Each Dependency?
- pydantic-ai: Type-safe AI framework that validates inputs/outputs
- ollama-python: Interface with local Ollama models
- chromadb: Vector database for RAG and memory
- whisper-cpp-python: Fast, local speech-to-text
- pyttsx3: Text-to-speech synthesis
- pytesseract: OCR for image text extraction
- fastapi: Web framework for API and MCP server
- asyncio-mqtt: A2A communication protocol
Core Configuration
# config.py
from pydantic import BaseSettings
from typing import Optional, List
import os
class Config(BaseSettings):
# Ollama Configuration
OLLAMA_BASE_URL: str = "http://localhost:11434"
PRIMARY_MODEL: str = "llama3.1:8b"
EMBEDDING_MODEL: str = "nomic-embed-text"
VISION_MODEL: str = "llava:7b"
# Vector Database
CHROMA_PERSIST_DIR: str = "./data/chroma_db"
# Memory Configuration
MEMORY_DB_URL: str = "sqlite:///./data/memory.db"
MAX_MEMORY_TOKENS: int = 4000
# Voice Configuration
STT_MODEL: str = "base" # Whisper model size
TTS_VOICE_RATE: int = 200
# MCP Configuration
MCP_SERVER_PORT: int = 8000
# A2A Protocol
MQTT_BROKER: str = "localhost"
MQTT_PORT: int = 1883
AGENT_ID: str = "alexai-001"
class Config:
env_file = ".env"
config = Config()Why this configuration structure?
- Centralized settings: Easy to modify behavior without code changes
- Environment variables: Secure credential management
- Type hints: Catch configuration errors early
- Defaults: Works out of the box for development
Part 2: Local LLM with Ollama Integration
Why Ollama?
Ollama provides several advantages:
- Privacy: Your data never leaves your machine
- Speed: No network latency for inference
- Cost: No API fees for unlimited usage
- Customization: Fine-tune models for your specific needs
Setting Up Ollama Models
# models/ollama_manager.py
import ollama
from typing import AsyncGenerator, Dict, Any, Optional
import asyncio
import logging
from config import config
class OllamaManager:
def __init__(self):
self.client = ollama.AsyncClient(host=config.OLLAMA_BASE_URL)
self.models_cache = {}
async def ensure_model_available(self, model_name: str) -> bool:
"""Ensure a model is pulled and available"""
try:
# Check if model exists locally
models = await self.client.list()
model_names = [m['name'] for m in models['models']]
if model_name not in model_names:
logging.info(f"Pulling model {model_name}...")
await self.client.pull(model_name)
logging.info(f"Model {model_name} pulled successfully")
return True
except Exception as e:
logging.error(f"Failed to ensure model {model_name}: {e}")
return False
async def generate_response(
self,
prompt: str,
model: Optional[str] = None,
system_prompt: Optional[str] = None,
stream: bool = False
) -> AsyncGenerator[str, None] | str:
"""Generate response with specified model"""
model = model or config.PRIMARY_MODEL
await self.ensure_model_available(model)
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": prompt})
try:
if stream:
async for chunk in await self.client.chat(
model=model,
messages=messages,
stream=True
):
if chunk['message']['content']:
yield chunk['message']['content']
else:
response = await self.client.chat(
model=model,
messages=messages
)
return response['message']['content']
except Exception as e:
logging.error(f"Generation failed: {e}")
return "I apologize, but I encountered an error generating a response."
async def generate_embeddings(self, texts: list[str]) -> list[list[float]]:
"""Generate embeddings for texts"""
await self.ensure_model_available(config.EMBEDDING_MODEL)
embeddings = []
for text in texts:
try:
response = await self.client.embeddings(
model=config.EMBEDDING_MODEL,
prompt=text
)
embeddings.append(response['embedding'])
except Exception as e:
logging.error(f"Embedding generation failed for text: {e}")
embeddings.append([0.0] * 768) # Fallback zero embedding
return embeddings
# Usage example
async def main():
manager = OllamaManager()
# Ensure our models are available
await manager.ensure_model_available(config.PRIMARY_MODEL)
await manager.ensure_model_available(config.EMBEDDING_MODEL)
# Generate a response
response = await manager.generate_response(
"Explain why local AI models are important for privacy",
system_prompt="You are a helpful AI assistant focused on privacy and security."
)
print(response)
if __name__ == "__main__":
asyncio.run(main())Why this architecture?
- Async operations: Prevents blocking during model loading/inference
- Model caching: Avoids repeated model pulls
- Error handling: Graceful degradation when models fail
- Flexible interface: Supports both streaming and non-streaming responses
Part 3: RAG Implementation with ChromaDB
Why RAG + Local Embeddings?
RAG (Retrieval Augmented Generation) solves the knowledge limitation problem of LLMs by:
- Current information: Access to up-to-date data beyond training cutoff
- Personal knowledge: Integration with your documents and notes
- Factual accuracy: Grounding responses in retrieved information
- Source attribution: Transparent information sourcing
Vector Database Setup
# rag/vector_store.py
import chromadb
from chromadb.config import Settings
import uuid
from typing import List, Dict, Any, Optional
from datetime import datetime
import logging
from models.ollama_manager import OllamaManager
from config import config
class VectorStore:
def __init__(self):
# Initialize ChromaDB with persistence
self.client = chromadb.PersistentClient(
path=config.CHROMA_PERSIST_DIR,
settings=Settings(
anonymized_telemetry=False,
allow_reset=True
)
)
self.ollama_manager = OllamaManager()
# Create collections for different data types
self.collections = {
'documents': self._get_or_create_collection('documents'),
'conversations': self._get_or_create_collection('conversations'),
'web_searches': self._get_or_create_collection('web_searches'),
'personal_notes': self._get_or_create_collection('personal_notes')
}
def _get_or_create_collection(self, name: str):
"""Get or create a collection with custom embedding function"""
try:
return self.client.get_collection(name)
except ValueError:
# Collection doesn't exist, create it
return self.client.create_collection(
name=name,
metadata={"hnsw:space": "cosine"}
)
async def add_document(
self,
content: str,
metadata: Dict[str, Any],
collection_name: str = 'documents'
) -> str:
"""Add document to vector store"""
# Generate embedding using local model
embeddings = await self.ollama_manager.generate_embeddings([content])
# Create unique ID
doc_id = str(uuid.uuid4())
# Add timestamp
metadata['timestamp'] = datetime.now().isoformat()
metadata['content_length'] = len(content)
# Store in ChromaDB
self.collections[collection_name].add(
documents=[content],
embeddings=embeddings,
metadatas=[metadata],
ids=[doc_id]
)
logging.info(f"Added document {doc_id} to {collection_name}")
return doc_id
async def similarity_search(
self,
query: str,
collection_name: str = 'documents',
n_results: int = 5,
filter_metadata: Optional[Dict] = None
) -> List[Dict[str, Any]]:
"""Search for similar documents"""
# Generate query embedding
query_embeddings = await self.ollama_manager.generate_embeddings([query])
# Search in ChromaDB
results = self.collections[collection_name].query(
query_embeddings=query_embeddings,
n_results=n_results,
where=filter_metadata
)
# Format results
formatted_results = []
for i in range(len(results['ids'][0])):
formatted_results.append({
'id': results['ids'][0][i],
'content': results['documents'][0][i],
'metadata': results['metadatas'][0][i],
'similarity_score': 1 - results['distances'][0][i] # Convert distance to similarity
})
return formatted_results
async def get_relevant_context(
self,
query: str,
max_tokens: int = 2000
) -> str:
"""Get relevant context for RAG"""
# Search across all collections
all_results = []
for collection_name in self.collections.keys():
try:
results = await self.similarity_search(
query=query,
collection_name=collection_name,
n_results=3
)
for result in results:
result['collection'] = collection_name
all_results.append(result)
except Exception as e:
logging.warning(f"Search failed in {collection_name}: {e}")
# Sort by similarity score
all_results.sort(key=lambda x: x['similarity_score'], reverse=True)
# Build context within token limit
context_parts = []
current_tokens = 0
for result in all_results:
content = result['content']
# Rough token estimation (1 token ≈ 4 characters)
content_tokens = len(content) // 4
if current_tokens + content_tokens > max_tokens:
break
source_info = f"[Source: {result['collection']}]"
context_parts.append(f"{source_info}\n{content}")
current_tokens += content_tokens
return "\n\n".join(context_parts)
# RAG Integration
class RAGGenerator:
def __init__(self):
self.vector_store = VectorStore()
self.ollama_manager = OllamaManager()
async def generate_with_context(
self,
query: str,
system_prompt: Optional[str] = None
) -> str:
"""Generate response using RAG"""
# Get relevant context
context = await self.vector_store.get_relevant_context(query)
# Build prompt with context
if context:
enhanced_prompt = f"""Context Information:
{context}
User Query: {query}
Please provide a comprehensive answer based on the context information provided. If the context doesn't contain relevant information, clearly state that and provide a general response."""
else:
enhanced_prompt = f"""User Query: {query}
No specific context information was found. Please provide a helpful general response."""
# Generate response
response = await self.ollama_manager.generate_response(
prompt=enhanced_prompt,
system_prompt=system_prompt
)
return response
# Usage example
async def demo_rag():
rag_gen = RAGGenerator()
# Add some sample documents
await rag_gen.vector_store.add_document(
content="AlexAI is a personal assistant that prioritizes user privacy by running all AI models locally. It uses Ollama for LLM hosting and ChromaDB for vector storage.",
metadata={"type": "system_info", "category": "documentation"}
)
# Query with RAG
response = await rag_gen.generate_with_context(
"What is AlexAI and how does it protect privacy?"
)
print("RAG Response:", response)
if __name__ == "__main__":
asyncio.run(demo_rag())Why this RAG implementation?
- Local embeddings: No data sent to external services
- Multiple collections: Organized knowledge by type
- Similarity scoring: Ranked retrieval results
- Token management: Prevents context overflow
- Source attribution: Transparent information sourcing
Part 4: Memory System Implementation
Why Persistent Memory?
Memory systems enable AI assistants to:
- Learn preferences: Remember user likes and dislikes
- Maintain context: Continue conversations across sessions
- Build relationships: Develop personalized interactions
- Improve over time: Accumulate knowledge about the user
Database Schema for Memory
# memory/models.py
from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, Float, JSON
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime
from config import config
Base = declarative_base()
class ConversationMemory(Base):
__tablename__ = "conversation_memory"
id = Column(Integer, primary_key=True)
session_id = Column(String(255), index=True)
user_message = Column(Text)
assistant_response = Column(Text)
timestamp = Column(DateTime, default=datetime.utcnow)
importance_score = Column(Float, default=0.5) # 0-1 scale
tags = Column(JSON) # For categorizing memories
embedding_id = Column(String(255)) # Reference to vector store
class UserPreferences(Base):
__tablename__ = "user_preferences"
id = Column(Integer, primary_key=True)
category = Column(String(100), index=True) # e.g., "communication_style", "interests"
key = Column(String(100))
value = Column(Text)
confidence = Column(Float, default=0.5) # How confident we are about this preference
last_updated = Column(DateTime, default=datetime.utcnow)
source = Column(String(100)) # How we learned this preference
class LongTermMemory(Base):
__tablename__ = "long_term_memory"
id = Column(Integer, primary_key=True)
memory_type = Column(String(50)) # "fact", "preference", "relationship", etc.
content = Column(Text)
importance = Column(Float, default=0.5)
last_accessed = Column(DateTime, default=datetime.utcnow)
access_count = Column(Integer, default=1)
metadata = Column(JSON)
# Database setup
engine = create_engine(config.MEMORY_DB_URL, echo=False)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def create_tables():
Base.metadata.create_all(bind=engine)Memory Manager Implementation
# memory/memory_manager.py
from typing import List, Dict, Any, Optional, Tuple
from sqlalchemy.orm import Session
from datetime import datetime, timedelta
import json
import logging
from .models import SessionLocal, ConversationMemory, UserPreferences, LongTermMemory
from rag.vector_store import VectorStore
from models.ollama_manager import OllamaManager
class MemoryManager:
def __init__(self):
self.vector_store = VectorStore()
self.ollama_manager = OllamaManager()
def get_db(self) -> Session:
return SessionLocal()
async def store_conversation(
self,
session_id: str,
user_message: str,
assistant_response: str,
tags: Optional[List[str]] = None
) -> str:
"""Store conversation in both SQL and vector databases"""
db = self.get_db()
try:
# Calculate importance score
importance = await self._calculate_importance(user_message, assistant_response)
# Store in SQL database
memory = ConversationMemory(
session_id=session_id,
user_message=user_message,
assistant_response=assistant_response,
importance_score=importance,
tags=tags or []
)
db.add(memory)
db.commit()
# Store in vector database for semantic search
conversation_text = f"User: {user_message}\nAssistant: {assistant_response}"
embedding_id = await self.vector_store.add_document(
content=conversation_text,
metadata={
"type": "conversation",
"session_id": session_id,
"importance": importance,
"tags": tags or []
},
collection_name="conversations"
)
# Update SQL record with embedding ID
memory.embedding_id = embedding_id
db.commit()
logging.info(f"Stored conversation with importance {importance}")
return embedding_id
except Exception as e:
db.rollback()
logging.error(f"Failed to store conversation: {e}")
raise
finally:
db.close()
async def _calculate_importance(self, user_message: str, assistant_response: str) -> float:
"""Calculate importance score for a conversation"""
# Use LLM to assess importance
assessment_prompt = f"""Assess the importance of this conversation on a scale of 0.0 to 1.0:
User: {user_message}
Assistant: {assistant_response}
Consider:
- Is this about personal preferences or information?
- Does it contain important facts or decisions?
- Is it likely to be referenced in future conversations?
- Does it reveal emotional or relationship context?
Respond with only a number between 0.0 and 1.0:"""
try:
response = await self.ollama_manager.generate_response(
assessment_prompt,
system_prompt="You are an expert at assessing conversation importance. Respond only with a decimal number."
)
# Extract number from response
import re
numbers = re.findall(r'0\.\d+|1\.0|0\.0', response)
if numbers:
return float(numbers[0])
else:
return 0.5 # Default importance
except Exception as e:
logging.warning(f"Importance calculation failed: {e}")
return 0.5
async def get_conversation_context(
self,
session_id: str,
query: Optional[str] = None,
max_messages: int = 10
) -> List[Dict[str, Any]]:
"""Get relevant conversation context"""
db = self.get_db()
try:
if query:
# Semantic search for relevant conversations
search_results = await self.vector_store.similarity_search(
query=query,
collection_name="conversations",
n_results=max_messages
)
# Get detailed conversation records
context = []
for result in search_results:
# Find matching conversation in SQL
conversation = db.query(ConversationMemory).filter(
ConversationMemory.embedding_id == result['id']
).first()
if conversation:
context.append({
'user_message': conversation.user_message,
'assistant_response': conversation.assistant_response,
'timestamp': conversation.timestamp,
'importance': conversation.importance_score,
'similarity': result['similarity_score']
})
return context
else:
# Get recent conversations from current session
conversations = db.query(ConversationMemory).filter(
ConversationMemory.session_id == session_id
).order_by(ConversationMemory.timestamp.desc()).limit(max_messages).all()
return [{
'user_message': conv.user_message,
'assistant_response': conv.assistant_response,
'timestamp': conv.timestamp,
'importance': conv.importance_score
} for conv in reversed(conversations)]
except Exception as e:
logging.error(f"Failed to get conversation context: {e}")
return []
finally:
db.close()
async def learn_preference(
self,
category: str,
key: str,
value: str,
confidence: float = 0.5,
source: str = "conversation"
):
"""Learn and store user preference"""
db = self.get_db()
try:
# Check if preference already exists
existing = db.query(UserPreferences).filter(
UserPreferences.category == category,
UserPreferences.key == key
).first()
if existing:
# Update existing preference
existing.value = value
existing.confidence = min(1.0, existing.confidence + confidence * 0.1)
existing.last_updated = datetime.utcnow()
existing.source = source
else:
# Create new preference
preference = UserPreferences(
category=category,
key=key,
value=value,
confidence=confidence,
source=source
)
db.add(preference)
db.commit()
logging.info(f"Learned preference: {category}.{key} = {value}")
except Exception as e:
db.rollback()
logging.error(f"Failed to learn preference: {e}")
finally:
db.close()
def get_user_preferences(self, category: Optional[str] = None) -> Dict[str, Any]:
"""Get user preferences"""
db = self.get_db()
try:
query = db.query(UserPreferences)
if category:
query = query.filter(UserPreferences.category == category)
preferences = query.all()
result = {}
for pref in preferences:
if pref.category not in result:
result[pref.category] = {}
result[pref.category][pref.key] = {
'value': pref.value,
'confidence': pref.confidence,
'last_updated': pref.last_updated
}
return result
except Exception as e:
logging.error(f"Failed to get preferences: {e}")
return {}
finally:
db.close()
# Usage example
async def demo_memory():
memory = MemoryManager()
# Store a conversation
await memory.store_conversation(
session_id="demo_session",
user_message="I prefer coffee over tea in the morning",
assistant_response="I'll remember that you prefer coffee in the morning. Would you like me to remind you about coffee-related deals or recipes?",
tags=["preference", "beverages"]
)
# Learn a preference
await memory.learn_preference(
category="beverages",
key="morning_preference",
value="coffee",
confidence=0.8,
source="direct_statement"
)
# Get conversation context
context = await memory.get_conversation_context("demo_session")
print("Context:", context)
# Get preferences
preferences = memory.get_user_preferences()
print("Preferences:", preferences)
if __name__ == "__main__":
import asyncio
asyncio.run(demo_memory())Why this memory architecture?
- Dual storage: SQL for structured data, vectors for semantic search
- Importance scoring: Prioritizes meaningful conversations
- Preference learning: Automatic extraction of user preferences
- Context retrieval: Relevant memory based on current conversation
- Confidence tracking: Measures certainty of learned information
Part 5: Voice Processing - Natural Interaction
Why Voice Processing?
Voice interaction makes AI assistants more natural and accessible:
- Hands-free operation: Use while driving, cooking, or working
- Natural communication: More intuitive than typing
- Accessibility: Helps users with visual or mobility impairments
- Multitasking: Continue other activities while interacting
Speech-to-Text Implementation
# voice/speech_processor.py
import whisper
import pyaudio
import wave
import tempfile
import asyncio
import logging
from typing import Optional, Callable
import threading
import queue
import numpy as np
from config import config
class SpeechToTextProcessor:
def __init__(self):
# Load Whisper model
self.model = whisper.load_model(config.STT_MODEL)
# Audio configuration
self.chunk_size = 1024
self.format = pyaudio.paInt16
self.channels = 1
self.rate = 16000 # Whisper works best with 16kHz
# Initialize PyAudio
self.audio = pyaudio.PyAudio()
# Recording state
self.is_recording = False
self.audio_queue = queue.Queue()
def _record_audio(self, duration: Optional[float] = None) -> bytes:
"""Record audio from microphone"""
stream = self.audio.open(
format=self.format,
channels=self.channels,
rate=self.rate,
input=True,
frames_per_buffer=self.chunk_size
)
frames = []
if duration:
# Record for specified duration
for _ in range(0, int(self.rate / self.chunk_size * duration)):
data = stream.read(self.chunk_size)
frames.append(data)
else:
# Record until stopped
self.is_recording = True
while self.is_recording:
try:
data = stream.read(self.chunk_size, exception_on_overflow=False)
frames.append(data)
except Exception as e:
logging.warning(f"Audio recording error: {e}")
break
stream.stop_stream()
stream.close()
return b''.join(frames)
def start_recording(self):
"""Start continuous recording in background thread"""
if self.is_recording:
return
def record_thread():
audio_data = self._record_audio()
self.audio_queue.put(audio_data)
thread = threading.Thread(target=record_thread)
thread.daemon = True
thread.start()
def stop_recording(self) -> Optional[bytes]:
"""Stop recording and return audio data"""
if not self.is_recording:
return None
self.is_recording = False
try:
# Get recorded audio data
return self.audio_queue.get(timeout=1.0)
except queue.Empty:
return None
async def transcribe_audio(self, audio_data: bytes) -> str:
"""Transcribe audio data to text"""
# Save audio data to temporary file
with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as temp_file:
# Convert raw audio to WAV format
with wave.open(temp_file.name, 'wb') as wav_file:
wav_file.setnchannels(self.channels)
wav_file.setsampwidth(self.audio.get_sample_size(self.format))
wav_file.setframerate(self.rate)
wav_file.writeframes(audio_data)
try:
# Transcribe using Whisper
result = self.model.transcribe(temp_file.name)
return result['text'].strip()
except Exception as e:
logging.error(f"Transcription failed: {e}")
return ""
finally:
# Clean up temporary file
import os
try:
os.unlink(temp_file.name)
except:
pass
async def transcribe_file(self, file_path: str) -> str:
"""Transcribe audio file to text"""
try:
result = self.model.transcribe(file_path)
return result['text'].strip()
except Exception as e:
logging.error(f"File transcription failed: {e}")
return ""
def cleanup(self):
"""Clean up audio resources"""
if hasattr(self, 'audio'):
self.audio.terminate()
class VoiceActivityDetector:
"""Simple voice activity detection"""
def __init__(self, threshold: float = 0.01, min_duration: float = 0.5):
self.threshold = threshold
self.min_duration = min_duration
def detect_speech(self, audio_data: bytes, sample_rate: int = 16000) -> bool:
"""Detect if audio contains speech"""
# Convert bytes to numpy array
audio_array = np.frombuffer(audio_data, dtype=np.int16)
# Normalize audio
audio_normalized = audio_array.astype(np.float32) / 32768.0
# Calculate RMS energy
rms_energy = np.sqrt(np.mean(audio_normalized ** 2))
# Check if energy exceeds threshold
return rms_energy > self.threshold
# Usage example
async def demo_speech_to_text():
stt = SpeechToTextProcessor()
vad = VoiceActivityDetector()
print("Starting voice recording... Press Enter to stop")
# Start recording
stt.start_recording()
# Wait for user input to stop
input()
# Stop recording and get audio
audio_data = stt.stop_recording()
if audio_data and vad.detect_speech(audio_data):
print("Transcribing...")
text = await stt.transcribe_audio(audio_data)
print(f"Transcription: {text}")
else:
print("No speech detected")
stt.cleanup()Text-to-Speech Implementation
# voice/text_to_speech.py
import pyttsx3
import asyncio
import tempfile
import threading
from typing import Optional, Dict, Any
import logging
from config import config
class TextToSpeechProcessor:
def __init__(self):
# Initialize pyttsx3 engine
self.engine = pyttsx3.init()
# Configure voice settings
self._configure_voice()
# Speech queue for async processing
self.speech_queue = asyncio.Queue()
self.is_speaking = False
def _configure_voice(self):
"""Configure TTS engine settings"""
# Set speech rate
self.engine.setProperty('rate', config.TTS_VOICE_RATE)
# Set volume (0.0 to 1.0)
self.engine.setProperty('volume', 0.9)
# Get available voices and set preferred voice
voices = self.engine.getProperty('voices')
if voices:
# Prefer female voice if available
for voice in voices:
if 'female' in voice.name.lower() or 'zira' in voice.name.lower():
self.engine.setProperty('voice', voice.id)
break
else:
# Use first available voice
self.engine.setProperty('voice', voices[0].id)
def get_available_voices(self) -> list[Dict[str, Any]]:
"""Get list of available voices"""
voices = self.engine.getProperty('voices')
return [{
'id': voice.id,
'name': voice.name,
'languages': getattr(voice, 'languages', []),
'gender': getattr(voice, 'gender', 'unknown')
} for voice in voices] if voices else []
def set_voice(self, voice_id: str) -> bool:
"""Set specific voice by ID"""
try:
self.engine.setProperty('voice', voice_id)
return True
except Exception as e:
logging.error(f"Failed to set voice {voice_id}: {e}")
return False
def speak_sync(self, text: str):
"""Speak text synchronously"""
try:
self.engine.say(text)
self.engine.runAndWait()
except Exception as e:
logging.error(f"TTS sync error: {e}")
async def speak_async(self, text: str):
"""Speak text asynchronously"""
def speak_thread():
try:
self.engine.say(text)
self.engine.runAndWait()
except Exception as e:
logging.error(f"TTS async error: {e}")
finally:
self.is_speaking = False
if self.is_speaking:
# Queue the speech request
await self.speech_queue.put(text)
else:
self.is_speaking = True
# Start speaking in background thread
thread = threading.Thread(target=speak_thread)
thread.daemon = True
thread.start()
# Process queued speech requests
asyncio.create_task(self._process_speech_queue())
async def _process_speech_queue(self):
"""Process queued speech requests"""
while not self.speech_queue.empty():
try:
text = await asyncio.wait_for(self.speech_queue.get(), timeout=0.1)
await asyncio.sleep(0.1) # Small delay between speeches
await self.speak_async(text)
except asyncio.TimeoutError:
break
def save_to_file(self, text: str, filename: str) -> bool:
"""Save speech to audio file"""
try:
self.engine.save_to_file(text, filename)
self.engine.runAndWait()
return True
except Exception as e:
logging.error(f"Failed to save TTS to file: {e}")
return False
def stop_speaking(self):
"""Stop current speech"""
try:
self.engine.stop()
self.is_speaking = False
except Exception as e:
logging.error(f"Failed to stop TTS: {e}")
# Combined Voice Interface
class VoiceInterface:
"""Combined speech-to-text and text-to-speech interface"""
def __init__(self):
self.stt = SpeechToTextProcessor()
self.tts = TextToSpeechProcessor()
self.vad = VoiceActivityDetector()
# Conversation state
self.listening = False
self.wake_words = ['alex', 'alexa', 'assistant']
async def listen_for_wake_word(self, timeout: float = 30.0) -> bool:
"""Listen for wake word activation"""
print("Listening for wake word...")
start_time = asyncio.get_event_loop().time()
while (asyncio.get_event_loop().time() - start_time) < timeout:
# Record short audio snippet
audio_data = self.stt._record_audio(duration=2.0)
if self.vad.detect_speech(audio_data):
# Transcribe and check for wake word
text = await self.stt.transcribe_audio(audio_data)
text_lower = text.lower()
for wake_word in self.wake_words:
if wake_word in text_lower:
await self.tts.speak_async("Yes, I'm listening.")
return True
await asyncio.sleep(0.1)
return False
async def voice_conversation(self, callback: Callable[[str], str]) -> str:
"""Conduct a voice conversation"""
await self.tts.speak_async("I'm ready. Please speak your message.")
# Start recording
self.stt.start_recording()
# Wait for silence or timeout
await asyncio.sleep(5.0) # Max 5 seconds of recording
# Stop recording
audio_data = self.stt.stop_recording()
if audio_data and self.vad.detect_speech(audio_data):
# Transcribe user input
user_text = await self.stt.transcribe_audio(audio_data)
if user_text:
print(f"User said: {user_text}")
# Get response from callback
response = await callback(user_text)
# Speak response
await self.tts.speak_async(response)
return response
else:
await self.tts.speak_async("I didn't catch that. Could you repeat?")
return ""
else:
await self.tts.speak_async("I didn't hear anything. Please try again.")
return ""
def cleanup(self):
"""Clean up voice interface resources"""
self.stt.cleanup()
# Usage example
async def demo_voice_interface():
voice = VoiceInterface()
# Define response callback
async def respond_to_user(text: str) -> str:
# Simple echo response
return f"You said: {text}. How can I help you with that?"
# Wait for wake word
if await voice.listen_for_wake_word():
# Conduct conversation
response = await voice.voice_conversation(respond_to_user)
print(f"Assistant responded: {response}")
voice.cleanup()
if __name__ == "__main__":
asyncio.run(demo_voice_interface())Why this voice processing architecture?
- Local processing: Whisper runs locally for privacy
- Async operations: Non-blocking audio processing
- Voice activity detection: Reduces unnecessary processing
- Wake word activation: Natural activation method
- Queue management: Handles multiple speech requests
- Error handling: Graceful degradation on audio issues
Part 6: OCR and Image Processing
Why OCR and Image Processing?
Visual document understanding enables:
- Document digitization: Convert paper documents to searchable text
- Screenshot analysis: Extract text from screen captures
- Handwriting recognition: Process handwritten notes
- Multi-modal interaction: Understand visual context in conversations
OCR Implementation with Tesseract
# vision/ocr_processor.py
import pytesseract
from PIL import Image, ImageEnhance, ImageFilter
import cv2
import numpy as np
import logging
from typing import List, Dict, Any, Optional, Tuple
import tempfile
import os
from pathlib import Path
class OCRProcessor:
def __init__(self):
# Configure Tesseract (adjust path as needed)
# pytesseract.pytesseract.tesseract_cmd = r'/usr/bin/tesseract' # Linux
# pytesseract.pytesseract.tesseract_cmd = r'C:\Program Files\Tesseract-OCR\tesseract.exe' # Windows
# Supported image formats
self.supported_formats = {'.png', '.jpg', '.jpeg', '.tiff', '.bmp', '.gif'}
# OCR configuration for different document types
self.ocr_configs = {
'default': '--oem 3 --psm 6',
'single_block': '--oem 3 --psm 6',
'single_line': '--oem 3 --psm 7',
'single_word': '--oem 3 --psm 8',
'digits_only': '--oem 3 --psm 6 -c tessedit_char_whitelist=0123456789',
'handwritten': '--oem 3 --psm 6',
}
def preprocess_image(self, image: Image.Image, enhancement_type: str = 'default') -> Image.Image:
"""Preprocess image for better OCR accuracy"""
# Convert to RGB if necessary
if image.mode != 'RGB':
image = image.convert('RGB')
# Apply different preprocessing based on type
if enhancement_type == 'scan':
# For scanned documents
image = self._enhance_scanned_document(image)
elif enhancement_type == 'photo':
# For photos of documents
image = self._enhance_photo_document(image)
elif enhancement_type == 'screenshot':
# For screenshots
image = self._enhance_screenshot(image)
else:
# Default enhancement
image = self._default_enhancement(image)
return image
def _enhance_scanned_document(self, image: Image.Image) -> Image.Image:
"""Enhance scanned documents"""
# Convert to OpenCV format
cv_image = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2BGR)
# Convert to grayscale
gray = cv2.cvtColor(cv_image, cv2.COLOR_BGR2GRAY)
# Apply Gaussian blur to reduce noise
blurred = cv2.GaussianBlur(gray, (5, 5), 0)
# Apply threshold to get binary image
_, thresh = cv2.threshold(blurred, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)
# Remove noise with morphological operations
kernel = np.ones((1, 1), np.uint8)
cleaned = cv2.morphologyEx(thresh, cv2.MORPH_CLOSE, kernel)
# Convert back to PIL
return Image.fromarray(cleaned)
def _enhance_photo_document(self, image: Image.Image) -> Image.Image:
"""Enhance photos of documents"""
# Convert to OpenCV format
cv_image = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2BGR)
# Convert to grayscale
gray = cv2.cvtColor(cv_image, cv2.COLOR_BGR2GRAY)
# Apply adaptive threshold
adaptive_thresh = cv2.adaptiveThreshold(
gray, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY, 11, 2
)
# Apply dilation and erosion to connect text components
kernel = np.ones((2, 2), np.uint8)
processed = cv2.morphologyEx(adaptive_thresh, cv2.MORPH_CLOSE, kernel)
return Image.fromarray(processed)
def _enhance_screenshot(self, image: Image.Image) -> Image.Image:
"""Enhance screenshots"""
# Screenshots are usually already high quality, minimal processing needed
enhancer = ImageEnhance.Contrast(image)
enhanced = enhancer.enhance(1.2)
enhancer = ImageEnhance.Sharpness(enhanced)
enhanced = enhancer.enhance(1.1)
return enhanced
def _default_enhancement(self, image: Image.Image) -> Image.Image:
"""Default image enhancement"""
# Enhance contrast
enhancer = ImageEnhance.Contrast(image)
enhanced = enhancer.enhance(1.2)
# Enhance sharpness
enhancer = ImageEnhance.Sharpness(enhanced)
enhanced = enhancer.enhance(1.1)
# Apply slight blur to reduce noise
enhanced = enhanced.filter(ImageFilter.MedianFilter(size=3))
return enhanced
def extract_text(
self,
image_path: str,
config_type: str = 'default',
language: str = 'eng',
enhancement_type: str = 'default'
) -> Dict[str, Any]:
"""Extract text from image"""
try:
# Load image
image = Image.open(image_path)
# Preprocess image
processed_image = self.preprocess_image(image, enhancement_type)
# Get OCR configuration
config = self.ocr_configs.get(config_type, self.ocr_configs['default'])
config += f' -l {language}'
# Extract text
text = pytesseract.image_to_string(processed_image, config=config)
# Get detailed information
data = pytesseract.image_to_data(processed_image, config=config, output_type=pytesseract.Output.DICT)
# Calculate confidence scores
confidences = [int(conf) for conf in data['conf'] if int(conf) > 0]
avg_confidence = sum(confidences) / len(confidences) if confidences else 0
# Extract word-level information
words = []
for i in range(len(data['text'])):
if int(data['conf'][i]) > 30: # Only include words with decent confidence
words.append({
'text': data['text'][i],
'confidence': int(data['conf'][i]),
'bbox': {
'x': data['left'][i],
'y': data['top'][i],
'width': data['width'][i],
'height': data['height'][i]
}
})
return {
'text': text.strip(),
'confidence': avg_confidence,
'word_count': len([w for w in words if w['text'].strip()]),
'words': words,
'image_size': image.size,
'processing_config': config_type
}
except Exception as e:
logging.error(f"OCR extraction failed: {e}")
return {
'text': '',
'confidence': 0,
'error': str(e)
}
def extract_text_from_regions(
self,
image_path: str,
regions: List[Tuple[int, int, int, int]],
config_type: str = 'default'
) -> List[Dict[str, Any]]:
"""Extract text from specific regions of an image"""
results = []
try:
image = Image.open(image_path)
for i, (x, y, width, height) in enumerate(regions):
# Crop region
region = image.crop((x, y, x + width, y + height))
# Save to temporary file
with tempfile.NamedTemporaryFile(suffix='.png', delete=False) as temp_file:
region.save(temp_file.name)
# Extract text from region
result = self.extract_text(temp_file.name, config_type)
result['region_id'] = i
result['bbox'] = {'x': x, 'y': y, 'width': width, 'height': height}
results.append(result)
# Clean up
os.unlink(temp_file.name)
except Exception as e:
logging.error(f"Region OCR failed: {e}")
results.append({'error': str(e), 'region_id': -1})
return results
def detect_document_structure(self, image_path: str) -> Dict[str, Any]:
"""Detect document structure (headers, paragraphs, etc.)"""
try:
# Get detailed OCR data
image = Image.open(image_path)
processed_image = self.preprocess_image(image)
data = pytesseract.image_to_data(
processed_image,
config='--oem 3 --psm 6',
output_type=pytesseract.Output.DICT
)
# Group text by blocks and paragraphs
blocks = {}
paragraphs = {}
for i in range(len(data['text'])):
if int(data['conf'][i]) > 30:
block_num = data['block_num'][i]
par_num = data['par_num'][i]
# Group by blocks
if block_num not in blocks:
blocks[block_num] = []
blocks[block_num].append({
'text': data['text'][i],
'bbox': {
'x': data['left'][i],
'y': data['top'][i],
'width': data['width'][i],
'height': data['height'][i]
}
})
# Group by paragraphs
par_key = f"{block_num}_{par_num}"
if par_key not in paragraphs:
paragraphs[par_key] = []
paragraphs[par_key].append({
'text': data['text'][i],
'bbox': {
'x': data['left'][i],
'y': data['top'][i],
'width': data['width'][i],
'height': data['height'][i]
}
})
# Reconstruct structured text
structured_text = []
for block_id in sorted(blocks.keys()):
block_text = ' '.join([item['text'] for item in blocks[block_id] if item['text'].strip()])
if block_text.strip():
structured_text.append({
'type': 'block',
'id': block_id,
'text': block_text.strip()
})
return {
'blocks': blocks,
'paragraphs': paragraphs,
'structured_text': structured_text,
'total_blocks': len(blocks),
'total_paragraphs': len(paragraphs)
}
except Exception as e:
logging.error(f"Document structure detection failed: {e}")
return {'error': str(e)}
# Vision-Language Integration
class VisionLanguageProcessor:
"""Integrate OCR with LLM for intelligent document understanding"""
def __init__(self, ollama_manager):
self.ocr = OCRProcessor()
self.ollama_manager = ollama_manager
async def analyze_document(self, image_path: str) -> Dict[str, Any]:
"""Analyze document with OCR + LLM"""
# Extract text with OCR
ocr_result = self.ocr.extract_text(image_path, enhancement_type='scan')
if not ocr_result['text']:
return {'error': 'No text found in image'}
# Analyze with LLM
analysis_prompt = f"""Analyze this document text and provide a structured summary:
Text:
{ocr_result['text']}
Please provide:
1. Document type (letter, invoice, report, etc.)
2. Key information extracted
3. Important dates, numbers, or entities
4. Summary of main content
5. Any action items or requirements
Format your response as structured information."""
analysis = await self.ollama_manager.generate_response(
analysis_prompt,
system_prompt="You are an expert document analyst. Provide clear, structured analysis of documents."
)
return {
'ocr_result': ocr_result,
'analysis': analysis,
'confidence': ocr_result.get('confidence', 0),
'word_count': ocr_result.get('word_count', 0)
}
async def answer_document_questions(self, image_path: str, question: str) -> str:
"""Answer questions about document content"""
# Extract text
ocr_result = self.ocr.extract_text(image_path)
if not ocr_result['text']:
return "I couldn't extract any text from this image."
# Answer question based on document content
qa_prompt = f"""Based on the following document text, answer the user's question.
Document text:
{ocr_result['text']}
User question: {question}
Provide a clear, accurate answer based only on the information in the document. If the information is not available, say so clearly."""
answer = await self.ollama_manager.generate_response(
qa_prompt,
system_prompt="You are a helpful assistant that answers questions based on document content."
)
return answer
# Usage example
async def demo_ocr_processing():
from models.ollama_manager import OllamaManager
# Initialize processors
ocr = OCRProcessor()
ollama = OllamaManager()
vision_processor = VisionLanguageProcessor(ollama)
# Example: Process a document image
image_path = "sample_document.jpg" # Replace with actual image path
if os.path.exists(image_path):
# Basic OCR
ocr_result = ocr.extract_text(image_path, enhancement_type='scan')
print(f"Extracted text: {ocr_result['text'][:200]}...")
print(f"Confidence: {ocr_result['confidence']:.2f}%")
# Document analysis
analysis = await vision_processor.analyze_document(image_path)
print(f"Document analysis: {analysis['analysis']}")
# Question answering
question = "What is the main topic of this document?"
answer = await vision_processor.answer_document_questions(image_path, question)
print(f"Q: {question}")
print(f"A: {answer}")
else:
print("Sample image not found. Please provide a document image to test.")
if __name__ == "__main__":
asyncio.run(demo_ocr_processing())Why this OCR architecture?
- Multiple enhancement strategies: Different preprocessing for different document types
- Confidence scoring: Quality assessment of OCR results
- Structured extraction: Beyond plain text to document structure
- LLM integration: Intelligent document analysis and Q&A
- Regional processing: Extract text from specific areas
- Error handling: Graceful degradation on processing failures
Part 7: MCP (Model Context Protocol) Integration
Why MCP?
The Model Context Protocol standardizes how AI models interact with external tools and data sources:
- Standardized interface: Consistent way to connect tools across different AI systems
- Security: Controlled access to external resources
- Extensibility: Easy addition of new capabilities
- Interoperability: Tools work across different AI platforms
MCP Server Implementation
# mcp/mcp_server.py
from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, Field
from typing import List, Dict, Any, Optional, Union
import asyncio
import logging
from datetime import datetime
import json
# MCP Protocol Models
class MCPResource(BaseModel):
uri: str
name: str
description: Optional[str] = None
mime_type: Optional[str] = None
class MCPTool(BaseModel):
name: str
description: str
input_schema: Dict[str, Any]
class MCPPrompt(BaseModel):
name: str
description: str
arguments: Optional[List[Dict[str, Any]]] = None
class MCPRequest(BaseModel):
method: str
params: Optional[Dict[str, Any]] = None
class MCPResponse(BaseModel):
result: Optional[Any] = None
error: Optional[Dict[str, Any]] = None
class MCPServer:
def __init__(self):
self.app = FastAPI(title="AlexAI MCP Server", version="1.0.0")
self.security = HTTPBearer()
# Register available tools and resources
self.tools = {}
self.resources = {}
self.prompts = {}
# Setup routes
self._setup_routes()
# Initialize integrations
self._setup_integrations()
def _setup_integrations(self):
"""Initialize all system integrations"""
from models.ollama_manager import OllamaManager
from rag.vector_store import VectorStore, RAGGenerator
from memory.memory_manager import MemoryManager
from voice.speech_processor import VoiceInterface
from vision.ocr_processor import VisionLanguageProcessor
self.ollama_manager = OllamaManager()
self.vector_store = VectorStore()
self.rag_generator = RAGGenerator()
self.memory_manager = MemoryManager()
self.voice_interface = VoiceInterface()
self.vision_processor = VisionLanguageProcessor(self.ollama_manager)
# Register tools
self._register_tools()
def _register_tools(self):
"""Register all available tools"""
# Chat tool
self.tools['chat'] = MCPTool(
name="chat",
description="Have a conversation with the AI assistant",
input_schema={
"type": "object",
"properties": {
"message": {"type": "string", "description": "User message"},
"session_id": {"type": "string", "description": "Session identifier"},
"use_memory": {"type": "boolean", "default": True},
"use_rag": {"type": "boolean", "default": True}
},
"required": ["message"]
}
)
# Document processing tool
self.tools['process_document'] = MCPTool(
name="process_document",
description="Process and analyze documents using OCR and LLM",
input_schema={
"type": "object",
"properties": {
"file_path": {"type": "string", "description": "Path to document image"},
"analysis_type": {"type": "string", "enum": ["extract", "analyze", "qa"], "default": "analyze"},
"question": {"type": "string", "description": "Question for QA mode"}
},
"required": ["file_path"]
}
)
# Voice interaction tool
self.tools['voice_chat'] = MCPTool(
name="voice_chat",
description="Voice-based conversation with the assistant",
input_schema={
"type": "object",
"properties": {
"mode": {"type": "string", "enum": ["listen", "speak", "conversation"], "default": "conversation"},
"text": {"type": "string", "description": "Text to speak (for speak mode)"},
"timeout": {"type": "number", "default": 30}
}
}
)
# Memory management tool
self.tools['manage_memory'] = MCPTool(
name="manage_memory",
description="Manage conversation memory and user preferences",
input_schema={
"type": "object",
"properties": {
"action": {"type": "string", "enum": ["get_context", "get_preferences", "learn_preference"]},
"session_id": {"type": "string"},
"category": {"type": "string"},
"key": {"type": "string"},
"value": {"type": "string"},
"confidence": {"type": "number", "default": 0.5}
},
"required": ["action"]
}
)
# Knowledge base tool
self.tools['knowledge_base'] = MCPTool(
name="knowledge_base",
description="Search and manage knowledge base",
input_schema={
"type": "object",
"properties": {
"action": {"type": "string", "enum": ["search", "add", "update", "delete"]},
"query": {"type": "string", "description": "Search query or content"},
"collection": {"type": "string", "default": "documents"},
"metadata": {"type": "object", "description": "Document metadata"}
},
"required": ["action"]
}
)
def _setup_routes(self):
"""Setup FastAPI routes for MCP protocol"""
@self.app.get("/")
async def root():
return {"service": "AlexAI MCP Server", "version": "1.0.0", "status": "running"}
@self.app.post("/mcp/initialize")
async def initialize(credentials: HTTPAuthorizationCredentials = Depends(self.security)):
"""Initialize MCP session"""
return MCPResponse(result={
"protocol_version": "1.0",
"server_info": {
"name": "AlexAI",
"version": "1.0.0"
},
"capabilities": {
"tools": True,
"resources": True,
"prompts": True
}
})
@self.app.get("/mcp/tools")
async def list_tools():
"""List available tools"""
return MCPResponse(result={"tools": list(self.tools.values())})
@self.app.post("/mcp/tools/call")
async def call_tool(request: MCPRequest):
"""Call a specific tool"""
tool_name = request.params.get("name")
arguments = request.params.get("arguments", {})
if tool_name not in self.tools:
return MCPResponse(error={"code": -32601, "message": f"Tool {tool_name} not found"})
try:
result = await self._execute_tool(tool_name, arguments)
return MCPResponse(result=result)
except Exception as e:
logging.error(f"Tool execution failed: {e}")
return MCPResponse(error={"code": -32603, "message": str(e)})
@self.app.get("/mcp/resources")
async def list_resources():
"""List available resources"""
return MCPResponse(result={"resources": list(self.resources.values())})
@self.app.post("/mcp/resources/read")
async def read_resource(request: MCPRequest):
"""Read a specific resource"""
uri = request.params.get("uri")
try:
content = await self._read_resource(uri)
return MCPResponse(result={"contents": [{"uri": uri, "text": content}]})
except Exception as e:
return MCPResponse(error={"code": -32603, "message": str(e)})
async def _execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
"""Execute a tool with given arguments"""
if tool_name == "chat":
return await self._handle_chat(arguments)
elif tool_name == "process_document":
return await self._handle_document_processing(arguments)
elif tool_name == "voice_chat":
return await self._handle_voice_chat(arguments)
elif tool_name == "manage_memory":
return await self._handle_memory_management(arguments)
elif tool_name == "knowledge_base":
return await self._handle_knowledge_base(arguments)
else:
raise ValueError(f"Unknown tool: {tool_name}")
async def _handle_chat(self, args: Dict[str, Any]) -> Dict[str, Any]:
"""Handle chat tool execution"""
message = args["message"]
session_id = args.get("session_id", "default")
use_memory = args.get("use_memory", True)
use_rag = args.get("use_rag", True)
# Get conversation context if memory is enabled
context = ""
if use_memory:
conversations = await self.memory_manager.get_conversation_context(
session_id=session_id,
query=message
)
if conversations:
context_parts = []
for conv in conversations[-3:]: # Last 3 relevant conversations
context_parts.append(f"User: {conv['user_message']}")
context_parts.append(f"Assistant: {conv['assistant_response']}")
context = "\n".join(context_parts)
# Generate response
if use_rag:
# Use RAG for enhanced responses
response = await self.rag_generator.generate_with_context(
query=message,
system_prompt=f"Previous conversation context:\n{context}" if context else None
)
else:
# Direct LLM response
prompt = f"{context}\n\nUser: {message}" if context else message
response = await self.ollama_manager.generate_response(
prompt=prompt,
system_prompt="You are AlexAI, a helpful personal assistant."
)
# Store conversation in memory
if use_memory:
await self.memory_manager.store_conversation(
session_id=session_id,
user_message=message,
assistant_response=response
)
return {
"response": response,
"session_id": session_id,
"used_memory": use_memory,
"used_rag": use_rag
}
async def _handle_document_processing(self, args: Dict[str, Any]) -> Dict[str, Any]:
"""Handle document processing tool"""
file_path = args["file_path"]
analysis_type = args.get("analysis_type", "analyze")
question = args.get("question")
if analysis_type == "extract":
# Simple OCR extraction
result = self.vision_processor.ocr.extract_text(file_path)
return {
"text": result["text"],
"confidence": result.get("confidence", 0),
"word_count": result.get("word_count", 0)
}
elif analysis_type == "analyze":
# Full document analysis
analysis = await self.vision_processor.analyze_document(file_path)
return analysis
elif analysis_type == "qa":
if not question:
raise ValueError("Question required for QA mode")
answer = await self.vision_processor.answer_document_questions(file_path, question)
return {
"question": question,
"answer": answer
}
else:
raise ValueError(f"Unknown analysis type: {analysis_type}")
async def _handle_voice_chat(self, args: Dict[str, Any]) -> Dict[str, Any]:
"""Handle voice chat tool"""
mode = args.get("mode", "conversation")
text = args.get("text")
timeout = args.get("timeout", 30)
if mode == "speak":
if not text:
raise ValueError("Text required for speak mode")
await self.voice_interface.tts.speak_async(text)
return {"status": "spoken", "text": text}
elif mode == "listen":
# Listen for wake word
detected = await self.voice_interface.listen_for_wake_word(timeout)
return {"wake_word_detected": detected}
elif mode == "conversation":
# Full voice conversation
async def chat_callback(user_text: str) -> str:
# Use chat tool to generate response
chat_result = await self._handle_chat({
"message": user_text,
"session_id": "voice_session",
"use_memory": True,
"use_rag": True
})
return chat_result["response"]
response = await self.voice_interface.voice_conversation(chat_callback)
return {"response": response}
else:
raise ValueError(f"Unknown voice mode: {mode}")
async def _handle_memory_management(self, args: Dict[str, Any]) -> Dict[str, Any]:
"""Handle memory management tool"""
action = args["action"]
if action == "get_context":
session_id = args.get("session_id", "default")
context = await self.memory_manager.get_conversation_context(session_id)
return {"context": context}
elif action == "get_preferences":
category = args.get("category")
preferences = self.memory_manager.get_user_preferences(category)
return {"preferences": preferences}
elif action == "learn_preference":
category = args["category"]
key = args["key"]
value = args["value"]
confidence = args.get("confidence", 0.5)
await self.memory_manager.learn_preference(category, key, value, confidence)
return {"status": "preference_learned"}
else:
raise ValueError(f"Unknown memory action: {action}")
async def _handle_knowledge_base(self, args: Dict[str, Any]) -> Dict[str, Any]:
"""Handle knowledge base tool"""
action = args["action"]
if action == "search":
query = args["query"]
collection = args.get("collection", "documents")
results = await self.vector_store.similarity_search(
query=query,
collection_name=collection
)
return {"results": results}
elif action == "add":
content = args["query"] # Using query field for content
collection = args.get("collection", "documents")
metadata = args.get("metadata", {})
doc_id = await self.vector_store.add_document(
content=content,
metadata=metadata,
collection_name=collection
)
return {"document_id": doc_id}
else:
raise ValueError(f"Unknown knowledge base action: {action}")
async def _read_resource(self, uri: str) -> str:
"""Read content from a resource URI"""
# Implement resource reading logic based on URI scheme
if uri.startswith("file://"):
file_path = uri[7:] # Remove file:// prefix
with open(file_path, 'r', encoding='utf-8') as f:
return f.read()
elif uri.startswith("memory://"):
# Read from memory system
# Implementation depends on specific memory resource format
return "Memory resource content"
else:
raise ValueError(f"Unsupported resource URI: {uri}")
# MCP Client for connecting to other services
class MCPClient:
def __init__(self, server_url: str, auth_token: Optional[str] = None):
self.server_url = server_url
self.auth_token = auth_token
self.session = None
async def initialize(self) -> Dict[str, Any]:
"""Initialize connection with MCP server"""
import aiohttp
headers = {}
if self.auth_token:
headers["Authorization"] = f"Bearer {self.auth_token}"
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.server_url}/mcp/initialize",
headers=headers
) as response:
result = await response.json()
return result
async def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Call a tool on the MCP server"""
import aiohttp
request_data = {
"method": "tools/call",
"params": {
"name": tool_name,
"arguments": arguments
}
}
headers = {}
if self.auth_token:
headers["Authorization"] = f"Bearer {self.auth_token}"
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.server_url}/mcp/tools/call",
json=request_data,
headers=headers
) as response:
result = await response.json()
return result
async def list_tools(self) -> List[Dict[str, Any]]:
"""List available tools"""
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.get(f"{self.server_url}/mcp/tools") as response:
result = await response.json()
return result.get("result", {}).get("tools", [])
# Usage example
async def demo_mcp_server():
"""Demonstrate MCP server functionality"""
# Start MCP server
server = MCPServer()
# Test chat tool
chat_result = await server._execute_tool("chat", {
"message": "Hello, how are you?",
"session_id": "demo_session"
})
print("Chat result:", chat_result)
# Test knowledge base
kb_result = await server._execute_tool("knowledge_base", {
"action": "add",
"query": "AlexAI is a privacy-focused personal assistant that runs locally.",
"metadata": {"category": "system_info"}
})
print("Knowledge base result:", kb_result)
# Search knowledge base
search_result = await server._execute_tool("knowledge_base", {
"action": "search",
"query": "privacy-focused assistant"
})
print("Search result:", search_result)
if __name__ == "__main__":
import uvicorn
# Create and run MCP server
server = MCPServer()
# Run with uvicorn
uvicorn.run(
server.app,
host="0.0.0.0",
port=8000,
log_level="info"
)Why this MCP implementation?
- Standard compliance: Follows MCP protocol specifications
- Tool integration: Unified interface for all AI capabilities
- Security: Authentication and authorization support
- Async operations: Non-blocking tool execution
- Error handling: Proper error responses and logging
- Extensibility: Easy addition of new tools and resources
Part 8: A2A (Agent-to-Agent) Protocol Implementation
Why A2A Communication?
Agent-to-agent communication enables:
- Distributed intelligence: Multiple AI agents working together
- Specialization: Each agent focuses on specific tasks
- Scalability: Horizontal scaling of AI capabilities
- Collaboration: Agents share information and coordinate actions
MQTT-Based A2A Protocol
# a2a/protocol.py
import asyncio
import json
import logging
from typing import Dict, Any, Optional, Callable, List
from datetime import datetime, timezone
import uuid
from dataclasses import dataclass, asdict
from enum import Enum
import paho.mqtt.client as mqtt
from config import config
class MessageType(Enum):
REQUEST = "request"
RESPONSE = "response"
BROADCAST = "broadcast"
HEARTBEAT = "heartbeat"
ERROR = "error"
class MessagePriority(Enum):
LOW = 1
NORMAL = 2
HIGH = 3
URGENT = 4
@dataclass
class A2AMessage:
id: str
sender_id: str
recipient_id: Optional[str] # None for broadcast
message_type: MessageType
priority: MessagePriority
timestamp: datetime
payload: Dict[str, Any]
correlation_id: Optional[str] = None # For request-response pairs
expires_at: Optional[datetime] = None
def to_json(self) -> str:
"""Convert message to JSON string"""
data = asdict(self)
data['timestamp'] = self.timestamp.isoformat()
data['message_type'] = self.message_type.value
data['priority'] = self.priority.value
if self.expires_at:
data['expires_at'] = self.expires_at.isoformat()
return json.dumps(data)
@classmethod
def from_json(cls, json_str: str) -> 'A2AMessage':
"""Create message from JSON string"""
data = json.loads(json_str)
data['timestamp'] = datetime.fromisoformat(data['timestamp'])
data['message_type'] = MessageType(data['message_type'])
data['priority'] = MessagePriority(data['priority'])
if data.get('expires_at'):
data['expires_at'] = datetime.fromisoformat(data['expires_at'])
return cls(**data)
class A2AProtocol:
def __init__(self, agent_id: str):
self.agent_id = agent_id
self.client = mqtt.Client()
self.is_connected = False
# Message handlers
self.message_handlers: Dict[MessageType, Callable] = {}
self.request_handlers: Dict[str, Callable] = {}
# Pending requests (for tracking responses)
self.pending_requests: Dict[str, asyncio.Future] = {}
# Known agents
self.known_agents: Dict[str, Dict[str, Any]] = {}
# Setup MQTT client
self._setup_mqtt_client()
# Topic structure: a2a/{recipient_id}/{message_type}
# Broadcast topic: a2a/broadcast/{message_type}
self.base_topic = "a2a"
def _setup_mqtt_client(self):
"""Setup MQTT client callbacks"""
def on_connect(client, userdata, flags, rc):
if rc == 0:
self.is_connected = True
logging.info(f"Agent {self.agent_id} connected to MQTT broker")
# Subscribe to our topics
topics = [
f"{self.base_topic}/{self.agent_id}/+", # Direct messages
f"{self.base_topic}/broadcast/+", # Broadcast messages
]
for topic in topics:
client.subscribe(topic)
logging.info(f"Subscribed to {topic}")
# Send heartbeat
asyncio.create_task(self._send_heartbeat())
else:
logging.error(f"Failed to connect to MQTT broker: {rc}")
def on_message(client, userdata, msg):
try:
message = A2AMessage.from_json(msg.payload.decode())
asyncio.create_task(self._handle_message(message))
except Exception as e:
logging.error(f"Failed to process A2A message: {e}")
def on_disconnect(client, userdata, rc):
self.is_connected = False
logging.info(f"Agent {self.agent_id} disconnected from MQTT broker")
self.client.on_connect = on_connect
self.client.on_message = on_message
self.client.on_disconnect = on_disconnect
async def connect(self):
"""Connect to MQTT broker"""
try:
self.client.connect(config.MQTT_BROKER, config.MQTT_PORT, 60)
self.client.loop_start()
# Wait for connection
while not self.is_connected:
await asyncio.sleep(0.1)
except Exception as e:
logging.error(f"Failed to connect to MQTT broker: {e}")
raise
async def disconnect(self):
"""Disconnect from MQTT broker"""
self.client.loop_stop()
self.client.disconnect()
def register_message_handler(self, message_type: MessageType, handler: Callable):
"""Register handler for specific message type"""
self.message_handlers[message_type] = handler
def register_request_handler(self, request_type: str, handler: Callable):
"""Register handler for specific request type"""
self.request_handlers[request_type] = handler
async def send_message(self, message: A2AMessage):
"""Send A2A message"""
if not self.is_connected:
raise RuntimeError("Not connected to MQTT broker")
# Determine topic
if message.recipient_id:
topic = f"{self.base_topic}/{message.recipient_id}/{message.message_type.value}"
else:
topic = f"{self.base_topic}/broadcast/{message.message_type.value}"
# Publish message
result = self.client.publish(topic, message.to_json())
if result.rc != mqtt.MQTT_ERR_SUCCESS:
raise RuntimeError(f"Failed to send message: {result.rc}")
logging.debug(f"Sent message {message.id} to {topic}")
async def send_request(
self,
recipient_id: str,
request_type: str,
payload: Dict[str, Any],
timeout: float = 30.0
) -> Dict[str, Any]:
"""Send request and wait for response"""
# Create request message
message = A2AMessage(
id=str(uuid.uuid4()),
sender_id=self.agent_id,
recipient_id=recipient_id,
message_type=MessageType.REQUEST,
priority=MessagePriority.NORMAL,
timestamp=datetime.now(timezone.utc),
payload={
"request_type": request_type,
**payload
},
correlation_id=str(uuid.uuid4())
)
# Create future for response
response_future = asyncio.Future()
self.pending_requests[message.correlation_id] = response_future
try:
# Send request
await self.send_message(message)
# Wait for response
response = await asyncio.wait_for(response_future, timeout=timeout)
return response
except asyncio.TimeoutError:
# Remove pending request
self.pending_requests.pop(message.correlation_id, None)
raise RuntimeError(f"Request timeout: {request_type}")
except Exception as e:
# Remove pending request
self.pending_requests.pop(message.correlation_id, None)
raise
async def send_response(
self,
request_message: A2AMessage,
response_payload: Dict[str, Any]
):
"""Send response to a request"""
response = A2AMessage(
id=str(uuid.uuid4()),
sender_id=self.agent_id,
recipient_id=request_message.sender_id,
message_type=MessageType.RESPONSE,
priority=request_message.priority,
timestamp=datetime.now(timezone.utc),
payload=response_payload,
correlation_id=request_message.correlation_id
)
await self.send_message(response)
async def broadcast_message(
self,
message_type: str,
payload: Dict[str, Any],
priority: MessagePriority = MessagePriority.NORMAL
):
"""Broadcast message to all agents"""
message = A2AMessage(
id=str(uuid.uuid4()),
sender_id=self.agent_id,
recipient_id=None,
message_type=MessageType.BROADCAST,
priority=priority,
timestamp=datetime.now(timezone.utc),
payload={
"broadcast_type": message_type,
**payload
}
)
await self.send_message(message)
async def _handle_message(self, message: A2AMessage):
"""Handle incoming A2A message"""
# Check if message is expired
if message.expires_at and datetime.now(timezone.utc) > message.expires_at:
logging.warning(f"Received expired message {message.id}")
return
# Handle response messages
if message.message_type == MessageType.RESPONSE:
await self._handle_response(message)
return
# Handle heartbeat messages
if message.message_type == MessageType.HEARTBEAT:
await self._handle_heartbeat(message)
return
# Handle request messages
if message.message_type == MessageType.REQUEST:
await self._handle_request(message)
return
# Handle broadcast messages
if message.message_type == MessageType.BROADCAST:
await self._handle_broadcast(message)
return
# Handle other message types
if message.message_type in self.message_handlers:
try:
await self.message_handlers[message.message_type](message)
except Exception as e:
logging.error(f"Message handler failed: {e}")
async def _handle_response(self, message: A2AMessage):
"""Handle response message"""
correlation_id = message.correlation_id
if correlation_id in self.pending_requests:
future = self.pending_requests.pop(correlation_id)
if not future.done():
future.set_result(message.payload)
async def _handle_request(self, message: A2AMessage):
"""Handle request message"""
request_type = message.payload.get("request_type")
if request_type in self.request_handlers:
try:
# Call request handler
response_payload = await self.request_handlers[request_type](message.payload)
# Send response
await self.send_response(message, response_payload)
except Exception as e:
logging.error(f"Request handler failed: {e}")
# Send error response
error_payload = {
"error": str(e),
"request_type": request_type
}
await self.send_response(message, error_payload)
else:
logging.warning(f"No handler for request type: {request_type}")
async def _handle_broadcast(self, message: A2AMessage):
"""Handle broadcast message"""
broadcast_type = message.payload.get("broadcast_type")
# Update known agents list
if broadcast_type == "agent_announcement":
agent_info = message.payload.get("agent_info", {})
self.known_agents[message.sender_id] = {
**agent_info,
"last_seen": datetime.now(timezone.utc)
}
logging.info(f"Discovered agent: {message.sender_id}")
# Handle other broadcast types with registered handlers
handler_key = f"broadcast_{broadcast_type}"
if handler_key in self.request_handlers:
try:
await self.request_handlers[handler_key](message.payload)
except Exception as e:
logging.error(f"Broadcast handler failed: {e}")
async def _handle_heartbeat(self, message: A2AMessage):
"""Handle heartbeat message"""
# Update agent's last seen time
if message.sender_id in self.known_agents:
self.known_agents[message.sender_id]["last_seen"] = datetime.now(timezone.utc)
async def _send_heartbeat(self):
"""Send periodic heartbeat"""
while self.is_connected:
try:
heartbeat = A2AMessage(
id=str(uuid.uuid4()),
sender_id=self.agent_id,
recipient_id=None,
message_type=MessageType.HEARTBEAT,
priority=MessagePriority.LOW,
timestamp=datetime.now(timezone.utc),
payload={
"status": "active",
"capabilities": await self._get_capabilities()
}
)
await self.send_message(heartbeat)
await asyncio.sleep(30) # Send heartbeat every 30 seconds
except Exception as e:
logging.error(f"Heartbeat failed: {e}")
break
async def _get_capabilities(self) -> List[str]:
"""Get agent capabilities"""
return [
"chat",
"document_processing",
"voice_interaction",
"knowledge_base_search",
"memory_management"
]
async def announce_agent(self):
"""Announce agent presence to network"""
await self.broadcast_message(
"agent_announcement",
{
"agent_info": {
"name": "AlexAI",
"version": "1.0.0",
"capabilities": await self._get_capabilities(),
"description": "AI Personal Assistant"
}
},
priority=MessagePriority.HIGH
)
# Agent coordination system
class AgentCoordinator:
"""Coordinates multiple AI agents"""
def __init__(self, agent_id: str):
self.agent_id = agent_id
self.protocol = A2AProtocol(agent_id)
self.specialized_agents = {}
# Register coordination handlers
self._register_handlers()
def _register_handlers(self):
"""Register coordination message handlers"""
# Task delegation
self.protocol.register_request_handler(
"delegate_task",
self._handle_task_delegation
)
# Capability query
self.protocol.register_request_handler(
"query_capabilities",
self._handle_capability_query
)
# Knowledge sharing
self.protocol.register_request_handler(
"share_knowledge",
self._handle_knowledge_sharing
)
# Collaborative processing
self.protocol.register_request_handler(
"collaborative_processing",
self._handle_collaborative_processing
)
async def start(self):
"""Start agent coordination"""
await self.protocol.connect()
await self.protocol.announce_agent()
# Start periodic agent discovery
asyncio.create_task(self._discover_agents())
async def stop(self):
"""Stop agent coordination"""
await self.protocol.disconnect()
async def _discover_agents(self):
"""Discover other agents in the network"""
while self.protocol.is_connected:
try:
# Query for active agents
await self.protocol.broadcast_message(
"agent_discovery",
{"requesting_capabilities": True}
)
await asyncio.sleep(60) # Discover every minute
except Exception as e:
logging.error(f"Agent discovery failed: {e}")
async def delegate_task(
self,
task_type: str,
task_data: Dict[str, Any],
preferred_agent: Optional[str] = None
) -> Dict[str, Any]:
"""Delegate task to appropriate agent"""
# Find suitable agent
target_agent = preferred_agent or await self._find_best_agent(task_type)
if not target_agent:
raise RuntimeError(f"No agent available for task type: {task_type}")
# Send delegation request
response = await self.protocol.send_request(
recipient_id=target_agent,
request_type="delegate_task",
payload={
"task_type": task_type,
"task_data": task_data,
"delegating_agent": self.agent_id
}
)
return response
async def _find_best_agent(self, task_type: str) -> Optional[str]:
"""Find the best agent for a specific task type"""
# Simple capability matching
for agent_id, agent_info in self.protocol.known_agents.items():
capabilities = agent_info.get("capabilities", [])
if task_type in capabilities:
return agent_id
return None
async def _handle_task_delegation(self, payload: Dict[str, Any]) -> Dict[str, Any]:
"""Handle incoming task delegation"""
task_type = payload["task_type"]
task_data = payload["task_data"]
# Execute task based on type
if task_type == "document_analysis":
return await self._execute_document_analysis(task_data)
elif task_type == "knowledge_search":
return await self._execute_knowledge_search(task_data)
elif task_type == "voice_processing":
return await self._execute_voice_processing(task_data)
else:
raise ValueError(f"Unsupported task type: {task_type}")
async def _execute_document_analysis(self, task_data: Dict[str, Any]) -> Dict[str, Any]:
"""Execute document analysis task"""
# This would integrate with the MCP server
from mcp.mcp_server import MCPServer
mcp_server = MCPServer()
result = await mcp_server._execute_tool("process_document", task_data)
return {
"status": "completed",
"result": result,
"agent_id": self.agent_id
}
async def _execute_knowledge_search(self, task_data: Dict[str, Any]) -> Dict[str, Any]:
"""Execute knowledge search task"""
from mcp.mcp_server import MCPServer
mcp_server = MCPServer()
result = await mcp_server._execute_tool("knowledge_base", {
"action": "search",
"query": task_data["query"]
})
return {
"status": "completed",
"result": result,
"agent_id": self.agent_id
}
async def _execute_voice_processing(self, task_data: Dict[str, Any]) -> Dict[str, Any]:
"""Execute voice processing task"""
from mcp.mcp_server import MCPServer
mcp_server = MCPServer()
result = await mcp_server._execute_tool("voice_chat", task_data)
return {
"status": "completed",
"result": result,
"agent_id": self.agent_id
}
async def _handle_capability_query(self, payload: Dict[str, Any]) -> Dict[str, Any]:
"""Handle capability query"""
return {
"agent_id": self.agent_id,
"capabilities": await self.protocol._get_capabilities(),
"status": "active",
"load": "normal" # Could be dynamically calculated
}
async def _handle_knowledge_sharing(self, payload: Dict[str, Any]) -> Dict[str, Any]:
"""Handle knowledge sharing request"""
knowledge_type = payload.get("knowledge_type")
query = payload.get("query")
# Share relevant knowledge from our knowledge base
from mcp.mcp_server import MCPServer
mcp_server = MCPServer()
search_result = await mcp_server._execute_tool("knowledge_base", {
"action": "search",
"query": query
})
return {
"shared_knowledge": search_result,
"source_agent": self.agent_id,
"knowledge_type": knowledge_type
}
async def _handle_collaborative_processing(self, payload: Dict[str, Any]) -> Dict[str, Any]:
"""Handle collaborative processing request"""
task_id = payload.get("task_id")
subtask = payload.get("subtask")
# Process our part of the collaborative task
# This is a simplified example
return {
"task_id": task_id,
"subtask_result": f"Processed subtask: {subtask}",
"processing_agent": self.agent_id,
"status": "completed"
}
# Usage example and integration
async def demo_a2a_protocol():
"""Demonstrate A2A protocol functionality"""
# Create two agents for demonstration
agent1 = AgentCoordinator("alexai-main")
agent2 = AgentCoordinator("alexai-specialist")
try:
# Start both agents
await agent1.start()
await agent2.start()
# Wait for discovery
await asyncio.sleep(2)
# Agent1 delegates a task to Agent2
result = await agent1.delegate_task(
task_type="knowledge_search",
task_data={
"query": "privacy-focused AI assistants"
}
)
print("Task delegation result:", result)
# Demonstrate broadcast communication
await agent1.protocol.broadcast_message(
"system_update",
{
"update_type": "capability_enhancement",
"details": "Added new document processing capabilities"
}
)
await asyncio.sleep(1)
finally:
# Clean up
await agent1.stop()
await agent2.stop()
if __name__ == "__main__":
asyncio.run(demo_a2a_protocol())Why this A2A architecture?
- MQTT backbone: Reliable, scalable messaging infrastructure
- Message typing: Structured communication patterns
- Request-response: Synchronous interaction when needed
- Broadcasting: Efficient one-to-many communication
- Agent discovery: Dynamic network topology
- Task delegation: Intelligent workload distribution
- Error handling: Robust failure recovery
Part 9: Complete Pydantic AI Integration
Why Pydantic AI?
Pydantic AI provides type-safe AI application development:
- Type validation: Automatic input/output validation
- Schema generation: Self-documenting APIs
- Error handling: Clear validation error messages
- IDE support: Better code completion and error detection
Main Application with Pydantic AI
# main.py - Complete AlexAI Assistant
from pydantic_ai import Agent, RunContext
from pydantic import BaseModel, Field, ValidationError
from typing import List, Dict, Any, Optional, Union, Literal
import asyncio
import logging
from datetime import datetime
import sys
from pathlib import Path
# Import all our components
from config import config
from models.ollama_manager import OllamaManager
from rag.vector_store import VectorStore, RAGGenerator
from memory.memory_manager import MemoryManager, create_tables
from voice.speech_processor import VoiceInterface
from vision.ocr_processor import VisionLanguageProcessor
from mcp.mcp_server import MCPServer
from a2a.protocol import AgentCoordinator
# Pydantic Models for type safety
class ChatRequest(BaseModel):
message: str = Field(..., description="User message", min_length=1, max_length=10000)
session_id: str = Field(default="default", description="Session identifier")
use_memory: bool = Field(default=True, description="Whether to use conversation memory")
use_rag: bool = Field(default=True, description="Whether to use RAG for enhanced responses")
voice_response: bool = Field(default=False, description="Whether to respond with voice")
class ChatResponse(BaseModel):
response: str = Field(..., description="Assistant response")
session_id: str = Field(..., description="Session identifier")
used_memory: bool = Field(..., description="Whether memory was used")
used_rag: bool = Field(..., description="Whether RAG was used")
response_time: float = Field(..., description="Response time in seconds")
confidence_score: Optional[float] = Field(None, description="Response confidence (0-1)")
class DocumentProcessRequest(BaseModel):
file_path: str = Field(..., description="Path to document file")
analysis_type: Literal["extract", "analyze", "qa"] = Field(default="analyze")
question: Optional[str] = Field(None, description="Question for QA analysis")
enhancement_type: Literal["default", "scan", "photo", "screenshot"] = Field(default="default")
class DocumentProcessResponse(BaseModel):
text: Optional[str] = Field(None, description="Extracted text")
analysis: Optional[str] = Field(None, description="Document analysis")
confidence: float = Field(..., description="OCR confidence score")
word_count: int = Field(..., description="Number of words extracted")
processing_time: float = Field(..., description="Processing time in seconds")
class VoiceRequest(BaseModel):
mode: Literal["listen", "speak", "conversation"] = Field(default="conversation")
text: Optional[str] = Field(None, description="Text to speak (for speak mode)")
timeout: float = Field(default=30.0, description="Timeout in seconds")
wake_word_detection: bool = Field(default=True, description="Enable wake word detection")
class VoiceResponse(BaseModel):
status: str = Field(..., description="Operation status")
transcribed_text: Optional[str] = Field(None, description="Transcribed speech")
response_text: Optional[str] = Field(None, description="Assistant response")
wake_word_detected: Optional[bool] = Field(None, description="Wake word detection result")
class KnowledgeRequest(BaseModel):
action: Literal["search", "add", "update", "delete"] = Field(..., description="Knowledge base action")
query: str = Field(..., description="Search query or content")
collection: str = Field(default="documents", description="Collection name")
metadata: Optional[Dict[str, Any]] = Field(default_factory=dict, description="Document metadata")
class KnowledgeResponse(BaseModel):
results: Optional[List[Dict[str, Any]]] = Field(None, description="Search results")
document_id: Optional[str] = Field(None, description="Added document ID")
status: str = Field(..., description="Operation status")
class SystemStatus(BaseModel):
service: str = Field(..., description="Service name")
version: str = Field(..., description="Service version")
status: str = Field(..., description="Service status")
uptime: float = Field(..., description="Uptime in seconds")
memory_usage: Dict[str, Any] = Field(..., description="Memory usage statistics")
active_sessions: int = Field(..., description="Number of active sessions")
total_requests: int = Field(..., description="Total requests processed")
# Context for passing data between agent functions
class AlexAIContext(RunContext):
def __init__(self):
super().__init__()
self.start_time = datetime.now()
self.total_requests = 0
self.active_sessions = set()
# Initialize all components
self.ollama_manager = OllamaManager()
self.vector_store = VectorStore()
self.rag_generator = RAGGenerator()
self.memory_manager = MemoryManager()
self.voice_interface = VoiceInterface()
self.vision_processor = VisionLanguageProcessor(self.ollama_manager)
self.mcp_server = MCPServer()
self.agent_coordinator = AgentCoordinator(config.AGENT_ID)
# Create Pydantic AI Agent
alexai_agent = Agent(
'ollama:llama3.1:8b', # Primary model
result_type=str, # Default result type
system_prompt="""You are AlexAI, an advanced personal assistant that prioritizes user privacy and runs entirely locally.
You have access to:
- Local language models (Ollama)
- Vector search and RAG capabilities
- Persistent memory and learning
- Voice interaction (speech-to-text and text-to-speech)
- Document processing with OCR
- Agent-to-agent communication
- Comprehensive tool integration via MCP
Your core principles:
1. Privacy first - all processing happens locally
2. Helpful and accurate responses
3. Learn from interactions to improve over time
4. Be transparent about your capabilities and limitations
5. Provide actionable and contextual assistance
Always be friendly, professional, and focused on helping the user achieve their goals."""
)
# Agent tool functions
@alexai_agent.tool
async def chat_with_memory_and_rag(
ctx: AlexAIContext,
request: ChatRequest
) -> ChatResponse:
"""Enhanced chat with memory and RAG capabilities"""
start_time = datetime.now()
try:
# Update session tracking
ctx.active_sessions.add(request.session_id)
ctx.total_requests += 1
# Get conversation context if memory enabled
context = ""
if request.use_memory:
conversations = await ctx.memory_manager.get_conversation_context(
session_id=request.session_id,
query=request.message
)
if conversations:
context_parts = []
for conv in conversations[-3:]:
context_parts.append(f"User: {conv['user_message']}")
context_parts.append(f"Assistant: {conv['assistant_response']}")
context = "\n".join(context_parts)
# Generate response
if request.use_rag:
response = await ctx.rag_generator.generate_with_context(
query=request.message,
system_prompt=f"Previous conversation context:\n{context}" if context else None
)
else:
prompt = f"{context}\n\nUser: {request.message}" if context else request.message
response = await ctx.ollama_manager.generate_response(
prompt=prompt,
system_prompt="You are AlexAI, a helpful personal assistant."
)
# Store conversation in memory
if request.use_memory:
await ctx.memory_manager.store_conversation(
session_id=request.session_id,
user_message=request.message,
assistant_response=response
)
# Voice response if requested
if request.voice_response:
await ctx.voice_interface.tts.speak_async(response)
response_time = (datetime.now() - start_time).total_seconds()
return ChatResponse(
response=response,
session_id=request.session_id,
used_memory=request.use_memory,
used_rag=request.use_rag,
response_time=response_time,
confidence_score=0.9 # Could be calculated based on various factors
)
except Exception as e:
logging.error(f"Chat error: {e}")
raise
@alexai_agent.tool
async def process_document(
ctx: AlexAIContext,
request: DocumentProcessRequest
) -> DocumentProcessResponse:
"""Process documents with OCR and analysis"""
start_time = datetime.now()
try:
if request.analysis_type == "extract":
result = ctx.vision_processor.ocr.extract_text(
request.file_path,
enhancement_type=request.enhancement_type
)
return DocumentProcessResponse(
text=result["text"],
confidence=result.get("confidence", 0),
word_count=result.get("word_count", 0),
processing_time=(datetime.now() - start_time).total_seconds()
)
elif request.analysis_type == "analyze":
analysis = await ctx.vision_processor.analyze_document(request.file_path)
return DocumentProcessResponse(
text=analysis["ocr_result"]["text"],
analysis=analysis["analysis"],
confidence=analysis["confidence"],
word_count=analysis["word_count"],
processing_time=(datetime.now() - start_time).total_seconds()
)
elif request.analysis_type == "qa":
if not request.question:
raise ValueError("Question required for QA analysis")
answer = await ctx.vision_processor.answer_document_questions(
request.file_path,
request.question
)
# Also extract text for metadata
ocr_result = ctx.vision_processor.ocr.extract_text(request.file_path)
return DocumentProcessResponse(
text=answer,
analysis=f"Q: {request.question}\nA: {answer}",
confidence=ocr_result.get("confidence", 0),
word_count=len(answer.split()),
processing_time=(datetime.now() - start_time).total_seconds()
)
except Exception as e:
logging.error(f"Document processing error: {e}")
raise
@alexai_agent.tool
async def voice_interaction(
ctx: AlexAIContext,
request: VoiceRequest
) -> VoiceResponse:
"""Handle voice interactions"""
try:
if request.mode == "speak":
if not request.text:
raise ValueError("Text required for speak mode")
await ctx.voice_interface.tts.speak_async(request.text)
return VoiceResponse(
status="spoken",
response_text=request.text
)
elif request.mode == "listen":
if request.wake_word_detection:
detected = await ctx.voice_interface.listen_for_wake_word(request.timeout)
return VoiceResponse(
status="listening_completed",
wake_word_detected=detected
)
else:
# Direct listening without wake word
return VoiceResponse(
status="listening_ready"
)
elif request.mode == "conversation":
async def chat_callback(user_text: str) -> str:
chat_request = ChatRequest(
message=user_text,
session_id="voice_session",
use_memory=True,
use_rag=True
)
chat_response = await chat_with_memory_and_rag(ctx, chat_request)
return chat_response.response
if request.wake_word_detection:
wake_detected = await ctx.voice_interface.listen_for_wake_word(request.timeout)
if not wake_detected:
return VoiceResponse(
status="no_wake_word",
wake_word_detected=False
)
response_text = await ctx.voice_interface.voice_conversation(chat_callback)
return VoiceResponse(
status="conversation_completed",
response_text=response_text,
wake_word_detected=request.wake_word_detection
)
except Exception as e:
logging.error(f"Voice interaction error: {e}")
raise
@alexai_agent.tool
async def manage_knowledge_base(
ctx: AlexAIContext,
request: KnowledgeRequest
) -> KnowledgeResponse:
"""Manage knowledge base operations"""
try:
if request.action == "search":
results = await ctx.vector_store.similarity_search(
query=request.query,
collection_name=request.collection
)
return KnowledgeResponse(
results=results,
status="search_completed"
)
elif request.action == "add":
doc_id = await ctx.vector_store.add_document(
content=request.query,
metadata=request.metadata,
collection_name=request.collection
)
return KnowledgeResponse(
document_id=doc_id,
status="document_added"
)
else:
raise ValueError(f"Unsupported action: {request.action}")
except Exception as e:
logging.error(f"Knowledge base error: {e}")
raise
@alexai_agent.tool
async def get_system_status(ctx: AlexAIContext) -> SystemStatus:
"""Get system status and statistics"""
uptime = (datetime.now() - ctx.start_time).total_seconds()
# Simple memory usage (could be enhanced with psutil)
import sys
memory_usage = {
"python_objects": sys.getsizeof(ctx),
"active_sessions": len(ctx.active_sessions)
}
return SystemStatus(
service="AlexAI",
version="1.0.0",
status="running",
uptime=uptime,
memory_usage=memory_usage,
active_sessions=len(ctx.active_sessions),
total_requests=ctx.total_requests
)
# Main application class
class AlexAIAssistant:
def __init__(self):
self.context = AlexAIContext()
self.is_running = False
async def initialize(self):
"""Initialize all components"""
logging.info("Initializing AlexAI Assistant...")
# Create database tables
create_tables()
# Initialize Ollama models
await self.context.ollama_manager.ensure_model_available(config.PRIMARY_MODEL)
await self.context.ollama_manager.ensure_model_available(config.EMBEDDING_MODEL)
# Start agent coordination
await self.context.agent_coordinator.start()
logging.info("AlexAI Assistant initialized successfully")
async def shutdown(self):
"""Shutdown assistant gracefully"""
logging.info("Shutting down AlexAI Assistant...")
# Stop agent coordination
await self.context.agent_coordinator.stop()
# Clean up voice interface
self.context.voice_interface.cleanup()
logging.info("AlexAI Assistant shutdown complete")
async def run_interactive_mode(self):
"""Run in interactive command-line mode"""
print("AlexAI Assistant - Interactive Mode")
print("Commands: chat, voice, document, knowledge, status, quit")
print("=" * 50)
self.is_running = True
while self.is_running:
try:
command = input("\nAlexAI> ").strip().lower()
if command == "quit":
self.is_running = False
break
elif command == "chat":
await self._interactive_chat()
elif command == "voice":
await self._interactive_voice()
elif command == "document":
await self._interactive_document()
elif command == "knowledge":
await self._interactive_knowledge()
elif command == "status":
await self._interactive_status()
else:
print("Unknown command. Available: chat, voice, document, knowledge, status, quit")
except KeyboardInterrupt:
print("\nUse 'quit' to exit gracefully")
except Exception as e:
print(f"Error: {e}")
async def _interactive_chat(self):
"""Interactive chat mode"""
print("\nChat Mode (type 'back' to return)")
while True:
try:
message = input("You: ").strip()
if message.lower() == "back":
break
if not message:
continue
request = ChatRequest(message=message)
response = await chat_with_memory_and_rag(self.context, request)
print(f"AlexAI: {response.response}")
print(f"(Response time: {response.response_time:.2f}s)")
except Exception as e:
print(f"Chat error: {e}")
async def _interactive_voice(self):
"""Interactive voice mode"""
print("\nVoice Mode")
print("1. Speak text")
print("2. Listen for wake word")
print("3. Voice conversation")
choice = input("Choose option (1-3): ").strip()
try:
if choice == "1":
text = input("Enter text to speak: ")
request = VoiceRequest(mode="speak", text=text)
response = await voice_interaction(self.context, request)
print(f"Status: {response.status}")
elif choice == "2":
print("Listening for wake word...")
request = VoiceRequest(mode="listen", timeout=10)
response = await voice_interaction(self.context, request)
print(f"Wake word detected: {response.wake_word_detected}")
elif choice == "3":
print("Starting voice conversation...")
request = VoiceRequest(mode="conversation")
response = await voice_interaction(self.context, request)
print(f"Conversation completed: {response.response_text}")
except Exception as e:
print(f"Voice error: {e}")
async def _interactive_document(self):
"""Interactive document processing"""
file_path = input("Enter document file path: ").strip()
if not Path(file_path).exists():
print("File not found")
return
print("Analysis types: extract, analyze, qa")
analysis_type = input("Choose analysis type: ").strip()
question = None
if analysis_type == "qa":
# Building an AI-Powered Personal Assistant: A Complete Guide
## Series Overview: The Ultimate AI Assistant Stack
In this comprehensive tutorial series, we'll build **"AlexAI"** - a sophisticated personal assistant that combines the power of local AI models, vector databases, memory systems, and multimodal capabilities. Our assistant will be able to:
- Process voice commands and respond with natural speech
- Remember conversations and learn from interactions
- Extract text from images and documents (OCR)
- Manage your calendar, emails, and tasks
- Search through your personal knowledge base
- Communicate with other AI agents using A2A protocols
### Why This Technology Stack?
**Scenario**: Imagine you're a busy professional who needs an AI assistant that:
- Works offline for privacy and speed
- Remembers your preferences and past conversations
- Can read documents, screenshots, and handwritten notes
- Integrates with your existing tools and workflows
- Shares information with other AI systems securely
### Technologies We'll Use and Why
1. **Ollama**: Local LLM hosting for privacy and speed
2. **MCP (Model Context Protocol)**: Standardized tool integration
3. **RAG (Retrieval Augmented Generation)**: Smart knowledge retrieval
4. **Memory Systems**: Persistent conversation history and learning
5. **Local Embeddings**: Fast, private vector search
6. **Multi-Model Support**: Different models for different tasks
7. **Voice-to-Text/Text-to-Voice**: Natural interaction
8. **OCR & Image Processing**: Visual document understanding
9. **A2A Protocols**: Agent-to-agent communication
10. **Pydantic AI**: Type-safe AI application development
## Part 10: Vue.js Frontend Interface
### Why Vue.js for AI Assistant Interface?
Vue.js provides an excellent foundation for AI assistant interfaces:
- **Reactive data**: Real-time updates for chat conversations
- **Component-based**: Modular UI components for different features
- **Easy integration**: Simple API communication with our FastAPI backend
- **Progressive enhancement**: Can be added incrementally to existing apps
- **Developer experience**: Excellent tooling and debugging support
### Project Structure and Setup
First, let's understand our frontend architecture:
frontend/ ├── public/ │ ├── index.html │ └── favicon.ico ├── src/ │ ├── components/ │ │ ├── Chat/ │ │ ├── Document/ │ │ ├── Voice/ │ │ ├── Knowledge/ │ │ └── System/ │ ├── services/ │ ├── stores/ │ ├── utils/ │ ├── App.vue │ └── main.js ├── package.json └── vite.config.js
### Complete Vue.js Frontend Implementation
```vue
<!-- App.vue - Main Application Component -->
<template>
<div id="app" class="min-h-screen bg-gray-50">
<!-- Navigation Header -->
<nav class="bg-blue-600 text-white shadow-lg">
<div class="max-w-7xl mx-auto px-4">
<div class="flex justify-between items-center h-16">
<div class="flex items-center space-x-4">
<div class="flex items-center">
<svg class="w-8 h-8 mr-2" fill="currentColor" viewBox="0 0 20 20">
<path d="M9 12l2 2 4-4m6 2a9 9 0 11-18 0 9 9 0 0118 0z"/>
</svg>
<h1 class="text-xl font-bold">AlexAI Assistant</h1>
</div>
</div>
<div class="flex items-center space-x-4">
<!-- Status Indicator -->
<div class="flex items-center space-x-2">
<div :class="[
'w-3 h-3 rounded-full',
systemStatus.status === 'running' ? 'bg-green-400' : 'bg-red-400'
]"></div>
<span class="text-sm">{{ systemStatus.status }}</span>
</div>
<!-- Settings Button -->
<button
@click="showSettings = !showSettings"
class="p-2 rounded-lg hover:bg-blue-700 transition-colors"
>
<svg class="w-5 h-5" fill="currentColor" viewBox="0 0 20 20">
<path fill-rule="evenodd" d="M11.49 3.17c-.38-1.56-2.6-1.56-2.98 0a1.532 1.532 0 01-2.286.948c-1.372-.836-2.942.734-2.106 2.106.54.886.061 2.042-.947 2.287-1.561.379-1.561 2.6 0 2.978a1.532 1.532 0 01.947 2.287c-.836 1.372.734 2.942 2.106 2.106a1.532 1.532 0 012.287.947c.379 1.561 2.6 1.561 2.978 0a1.533 1.533 0 012.287-.947c1.372.836 2.942-.734 2.106-2.106a1.533 1.533 0 01.947-2.287c1.561-.379 1.561-2.6 0-2.978a1.532 1.532 0 01-.947-2.287c.836-1.372-.734-2.942-2.106-2.106a1.532 1.532 0 01-2.287-.947zM10 13a3 3 0 100-6 3 3 0 000 6z" clip-rule="evenodd"/>
</svg>
</button>
</div>
</div>
</div>
</nav>
<div class="flex h-[calc(100vh-4rem)]">
<!-- Sidebar -->
<aside class="w-64 bg-white shadow-lg border-r">
<div class="p-4">
<nav class="space-y-2">
<button
v-for="tab in tabs"
:key="tab.id"
@click="activeTab = tab.id"
:class="[
'w-full flex items-center px-4 py-3 text-left rounded-lg transition-colors',
activeTab === tab.id
? 'bg-blue-100 text-blue-700 border-l-4 border-blue-700'
: 'text-gray-600 hover:bg-gray-100'
]"
>
<component :is="tab.icon" class="w-5 h-5 mr-3" />
{{ tab.name }}
</button>
</nav>
</div>
<!-- System Status Panel -->
<div class="p-4 border-t">
<h3 class="text-sm font-semibold text-gray-700 mb-2">System Status</h3>
<div class="space-y-2 text-xs text-gray-600">
<div class="flex justify-between">
<span>Uptime:</span>
<span>{{ formatUptime(systemStatus.uptime) }}</span>
</div>
<div class="flex justify-between">
<span>Requests:</span>
<span>{{ systemStatus.total_requests }}</span>
</div>
<div class="flex justify-between">
<span>Sessions:</span>
<span>{{ systemStatus.active_sessions }}</span>
</div>
</div>
</div>
</aside>
<!-- Main Content -->
<main class="flex-1 flex flex-col">
<!-- Chat Interface -->
<ChatInterface
v-if="activeTab === 'chat'"
:messages="chatMessages"
:loading="chatLoading"
@send-message="sendChatMessage"
@clear-chat="clearChat"
/>
<!-- Document Processing -->
<DocumentProcessor
v-if="activeTab === 'documents'"
@process-document="processDocument"
/>
<!-- Voice Interaction -->
<VoiceInterface
v-if="activeTab === 'voice'"
:is-listening="isListening"
:is-speaking="isSpeaking"
@start-listening="startVoiceInteraction"
@stop-listening="stopVoiceInteraction"
@text-to-speech="textToSpeech"
/>
<!-- Knowledge Base -->
<KnowledgeBase
v-if="activeTab === 'knowledge'"
:search-results="knowledgeResults"
@search-knowledge="searchKnowledge"
@add-knowledge="addKnowledge"
/>
<!-- System Monitor -->
<SystemMonitor
v-if="activeTab === 'system'"
:status="systemStatus"
:logs="systemLogs"
@refresh-status="refreshSystemStatus"
/>
</main>
</div>
<!-- Settings Modal -->
<SettingsModal
v-if="showSettings"
:settings="appSettings"
@close="showSettings = false"
@update-settings="updateSettings"
/>
<!-- Loading Overlay -->
<div
v-if="globalLoading"
class="fixed inset-0 bg-black bg-opacity-50 flex items-center justify-center z-50"
>
<div class="bg-white rounded-lg p-6 flex items-center space-x-4">
<div class="animate-spin rounded-full h-8 w-8 border-b-2 border-blue-600"></div>
<span class="text-gray-700">{{ loadingMessage }}</span>
</div>
</div>
<!-- Toast Notifications -->
<div class="fixed top-4 right-4 space-y-2 z-40">
<div
v-for="notification in notifications"
:key="notification.id"
:class="[
'p-4 rounded-lg shadow-lg max-w-sm transition-all duration-300',
notification.type === 'success' ? 'bg-green-500 text-white' :
notification.type === 'error' ? 'bg-red-500 text-white' :
notification.type === 'warning' ? 'bg-yellow-500 text-white' :
'bg-blue-500 text-white'
]"
>
<div class="flex items-center justify-between">
<span>{{ notification.message }}</span>
<button @click="removeNotification(notification.id)" class="ml-2 text-white hover:text-gray-200">
<svg class="w-4 h-4" fill="currentColor" viewBox="0 0 20 20">
<path fill-rule="evenodd" d="M4.293 4.293a1 1 0 011.414 0L10 8.586l4.293-4.293a1 1 0 111.414 1.414L11.414 10l4.293 4.293a1 1 0 01-1.414 1.414L10 11.414l-4.293 4.293a1 1 0 01-1.414-1.414L8.586 10 4.293 5.707a1 1 0 010-1.414z" clip-rule="evenodd"/>
</svg>
</button>
</div>
</div>
</div>
</div>
</template>
<script>
import { ref, onMounted, computed } from 'vue'
import { apiService } from './services/api'
import { useNotifications } from './composables/useNotifications'
// Import components
import ChatInterface from './components/Chat/ChatInterface.vue'
import DocumentProcessor from './components/Document/DocumentProcessor.vue'
import VoiceInterface from './components/Voice/VoiceInterface.vue'
import KnowledgeBase from './components/Knowledge/KnowledgeBase.vue'
import SystemMonitor from './components/System/SystemMonitor.vue'
import SettingsModal from './components/Settings/SettingsModal.vue'
// Icons
import ChatIcon from './components/Icons/ChatIcon.vue'
import DocumentIcon from './components/Icons/DocumentIcon.vue'
import VoiceIcon from './components/Icons/VoiceIcon.vue'
import KnowledgeIcon from './components/Icons/KnowledgeIcon.vue'
import SystemIcon from './components/Icons/SystemIcon.vue'
export default {
name: 'App',
components: {
ChatInterface,
DocumentProcessor,
VoiceInterface,
KnowledgeBase,
SystemMonitor,
SettingsModal
},
setup() {
// Reactive state
const activeTab = ref('chat')
const showSettings = ref(false)
const globalLoading = ref(false)
const loadingMessage = ref('')
// System status
const systemStatus = ref({
service: 'AlexAI',
version: '1.0.0',
status: 'connecting',
uptime: 0,
active_sessions: 0,
total_requests: 0,
memory_usage: {}
})
// Chat state
const chatMessages = ref([])
const chatLoading = ref(false)
// Voice state
const isListening = ref(false)
const isSpeaking = ref(false)
// Knowledge state
const knowledgeResults = ref([])
// System logs
const systemLogs = ref([])
// App settings
const appSettings = ref({
theme: 'light',
voiceEnabled: true,
autoSave: true,
apiTimeout: 30000,
sessionId: 'web_session_' + Date.now()
})
// Notifications composable
const { notifications, addNotification, removeNotification } = useNotifications()
// Navigation tabs
const tabs = [
{ id: 'chat', name: 'Chat', icon: ChatIcon },
{ id: 'documents', name: 'Documents', icon: DocumentIcon },
{ id: 'voice', name: 'Voice', icon: VoiceIcon },
{ id: 'knowledge', name: 'Knowledge', icon: KnowledgeIcon },
{ id: 'system', name: 'System', icon: SystemIcon }
]
// Methods
const refreshSystemStatus = async () => {
try {
const status = await apiService.getSystemStatus()
systemStatus.value = status
if (status.status === 'running') {
addNotification('Connected to AlexAI', 'success')
}
} catch (error) {
console.error('Failed to get system status:', error)
systemStatus.value.status = 'error'
addNotification('Failed to connect to AlexAI', 'error')
}
}
const sendChatMessage = async (message) => {
if (!message.trim()) return
// Add user message to chat
chatMessages.value.push({
id: Date.now(),
type: 'user',
content: message,
timestamp: new Date()
})
chatLoading.value = true
try {
const response = await apiService.sendChatMessage({
message: message,
session_id: appSettings.value.sessionId,
use_memory: true,
use_rag: true,
voice_response: false
})
// Add assistant response to chat
chatMessages.value.push({
id: Date.now() + 1,
type: 'assistant',
content: response.response,
timestamp: new Date(),
metadata: {
response_time: response.response_time,
used_memory: response.used_memory,
used_rag: response.used_rag,
confidence_score: response.confidence_score
}
})
// Update system status
systemStatus.value.total_requests++
} catch (error) {
console.error('Chat error:', error)
chatMessages.value.push({
id: Date.now() + 1,
type: 'error',
content: 'Sorry, I encountered an error processing your message.',
timestamp: new Date()
})
addNotification('Chat error: ' + error.message, 'error')
} finally {
chatLoading.value = false
}
}
const clearChat = () => {
chatMessages.value = []
addNotification('Chat cleared', 'info')
}
const processDocument = async (file, analysisType, question) => {
globalLoading.value = true
loadingMessage.value = 'Processing document...'
try {
// Create FormData for file upload
const formData = new FormData()
formData.append('file', file)
formData.append('analysis_type', analysisType)
if (question) {
formData.append('question', question)
}
const response = await apiService.processDocument(formData)
// Show results in notification or modal
addNotification(`Document processed successfully. Confidence: ${response.confidence.toFixed(1)}%`, 'success')
// Could also switch to results tab or show in modal
return response
} catch (error) {
console.error('Document processing error:', error)
addNotification('Document processing failed: ' + error.message, 'error')
} finally {
globalLoading.value = false
loadingMessage.value = ''
}
}
const startVoiceInteraction = async () => {
isListening.value = true
try {
const response = await apiService.voiceInteraction({
mode: 'conversation',
timeout: 30,
wake_word_detection: true
})
if (response.response_text) {
// Add voice response to chat
chatMessages.value.push({
id: Date.now(),
type: 'assistant',
content: response.response_text,
timestamp: new Date(),
metadata: { source: 'voice' }
})
}
addNotification('Voice interaction completed', 'success')
} catch (error) {
console.error('Voice interaction error:', error)
addNotification('Voice interaction failed: ' + error.message, 'error')
} finally {
isListening.value = false
}
}
const stopVoiceInteraction = () => {
isListening.value = false
addNotification('Voice interaction stopped', 'info')
}
const textToSpeech = async (text) => {
if (!text.trim()) return
isSpeaking.value = true
try {
await apiService.voiceInteraction({
mode: 'speak',
text: text
})
addNotification('Text spoken successfully', 'success')
} catch (error) {
console.error('Text-to-speech error:', error)
addNotification('Text-to-speech failed: ' + error.message, 'error')
} finally {
isSpeaking.value = false
}
}
const searchKnowledge = async (query, collection = 'documents') => {
try {
const response = await apiService.searchKnowledge({
action: 'search',
query: query,
collection: collection
})
knowledgeResults.value = response.results || []
addNotification(`Found ${knowledgeResults.value.length} results`, 'success')
} catch (error) {
console.error('Knowledge search error:', error)
addNotification('Knowledge search failed: ' + error.message, 'error')
}
}
const addKnowledge = async (content, metadata = {}) => {
try {
const response = await apiService.addKnowledge({
action: 'add',
query: content,
metadata: metadata
})
addNotification(`Knowledge added with ID: ${response.document_id}`, 'success')
} catch (error) {
console.error('Add knowledge error:', error)
addNotification('Failed to add knowledge: ' + error.message, 'error')
}
}
const updateSettings = (newSettings) => {
appSettings.value = { ...appSettings.value, ...newSettings }
localStorage.setItem('alexai_settings', JSON.stringify(appSettings.value))
addNotification('Settings updated', 'success')
}
const formatUptime = (seconds) => {
if (seconds < 60) return `${seconds.toFixed(0)}s`
if (seconds < 3600) return `${(seconds / 60).toFixed(0)}m`
return `${(seconds / 3600).toFixed(1)}h`
}
// Initialize app
onMounted(async () => {
// Load settings from localStorage
const savedSettings = localStorage.getItem('alexai_settings')
if (savedSettings) {
appSettings.value = { ...appSettings.value, ...JSON.parse(savedSettings) }
}
// Initial system status check
await refreshSystemStatus()
// Set up periodic status updates
setInterval(refreshSystemStatus, 30000) // Every 30 seconds
// Welcome message
chatMessages.value.push({
id: Date.now(),
type: 'assistant',
content: 'Hello! I\'m AlexAI, your personal assistant. How can I help you today?',
timestamp: new Date()
})
})
return {
// State
activeTab,
showSettings,
globalLoading,
loadingMessage,
systemStatus,
chatMessages,
chatLoading,
isListening,
isSpeaking,
knowledgeResults,
systemLogs,
appSettings,
notifications,
tabs,
// Methods
refreshSystemStatus,
sendChatMessage,
clearChat,
processDocument,
startVoiceInteraction,
stopVoiceInteraction,
textToSpeech,
searchKnowledge,
addKnowledge,
updateSettings,
removeNotification,
formatUptime
}
}
}
</script>
<style>
/* Global styles */
#app {
font-family: 'Inter', -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
-webkit-font-smoothing: antialiased;
-moz-osx-font-smoothing: grayscale;
}
/* Custom scrollbar */
::-webkit-scrollbar {
width: 6px;
}
::-webkit-scrollbar-track {
background: #f1f1f1;
}
::-webkit-scrollbar-thumb {
background: #c1c1c1;
border-radius: 3px;
}
::-webkit-scrollbar-thumb:hover {
background: #a8a8a8;
}
/* Animation classes */
.fade-enter-active, .fade-leave-active {
transition: opacity 0.3s ease;
}
.fade-enter-from, .fade-leave-to {
opacity: 0;
}
.slide-up-enter-active {
transition: all 0.3s ease-out;
}
.slide-up-leave-active {
transition: all 0.2s ease-in;
}
.slide-up-enter-from {
transform: translateY(20px);
opacity: 0;
}
.slide-up-leave-to {
transform: translateY(-20px);
opacity: 0;
}
</style>API Service Layer
// services/api.js
class ApiService {
constructor() {
this.baseURL = process.env.VUE_APP_API_URL || 'http://localhost:8000'
this.timeout = 30000
}
async request(endpoint, options = {}) {
const url = `${this.baseURL}${endpoint}`
const config = {
timeout: this.timeout,
headers: {
'Content-Type': 'application/json',
...options.headers
},
...options
}
try {
const response = await fetch(url, config)
if (!response.ok) {
throw new Error(`HTTP ${response.status}: ${response.statusText}`)
}
const contentType = response.headers.get('content-type')
if (contentType && contentType.includes('application/json')) {
return await response.json()
}
return await response.text()
} catch (error) {
console.error(`API request failed: ${endpoint}`, error)
throw error
}
}
// System endpoints
async getSystemStatus() {
return this.request('/status')
}
// Chat endpoints
async sendChatMessage(data) {
return this.request('/chat', {
method: 'POST',
body: JSON.stringify(data)
})
}
// Document processing endpoints
async processDocument(formData) {
return this.request('/document', {
method: 'POST',
headers: {}, // Remove Content-Type to let browser set multipart boundary
body: formData
})
}
// Voice interaction endpoints
async voiceInteraction(data) {
return this.request('/voice', {
method: 'POST',
body: JSON.stringify(data)
})
}
// Knowledge base endpoints
async searchKnowledge(data) {
return this.request('/knowledge', {
method: 'POST',
body: JSON.stringify(data)
})
}
async addKnowledge(data) {
return this.request('/knowledge', {
method: 'POST',
body: JSON.stringify(data)
})
}
}
export const apiService = new ApiService()Key Components
Chat Interface Component
<!-- components/Chat/ChatInterface.vue -->
<template>
<div class="flex flex-col h-full">
<!-- Chat Header -->
<div class="bg-white border-b px-6 py-4 flex justify-between items-center">
<h2 class="text-lg font-semibold text-gray-800">Chat with AlexAI</h2>
<div class="flex space-x-2">
<button
@click="toggleVoiceResponse"
:class="[
'px-3 py-1 rounded text-sm transition-colors',
voiceEnabled ? 'bg-green-100 text-green-700' : 'bg-gray-100 text-gray-700'
]"
>
🔊 Voice {{ voiceEnabled ? 'On' : 'Off' }}
</button>
<button
@click="$emit('clear-chat')"
class="px-3 py-1 bg-red-100 text-red-700 rounded text-sm hover:bg-red-200 transition-colors"
>
Clear Chat
</button>
</div>
</div>
<!-- Messages Container -->
<div class="flex-1 overflow-y-auto p-6 space-y-4" ref="messagesContainer">
<div
v-for="message in messages"
:key="message.id"
:class="[
'flex',
message.type === 'user' ? 'justify-end' : 'justify-start'
]"
>
<div
:class="[
'max-w-xs lg:max-w-md px-4 py-2 rounded-lg',
message.type === 'user'
? 'bg-blue-600 text-white'
: message.type === 'error'
? 'bg-red-100 text-red-800 border border-red-200'
: 'bg-gray-100 text-gray-800'
]"
>
<div class="text-sm">{{ message.content }}</div>
<!-- Message metadata -->
<div
v-if="message.metadata"
class="text-xs mt-2 opacity-75"
>
<div v-if="message.metadata.response_time">
Response: {{ message.metadata.response_time.toFixed(2) }}s
</div>
<div v-if="message.metadata.confidence_score">
Confidence: {{ (message.metadata.confidence_score * 100).toFixed(1) }}%
</div>
<div class="flex space-x-2 mt-1">
<span v-if="message.metadata.used_memory" class="bg-blue-500 px-1 rounded">Memory</span>
<span v-if="message.metadata.used_rag" class="bg-green-500 px-1 rounded">RAG</span>
<span v-if="message.metadata.source" class="bg-purple-500 px-1 rounded">{{ message.metadata.source }}</span>
</div>
</div>
<div class="text-xs mt-1 opacity-75">
{{ formatTime(message.timestamp) }}
</div>
</div>
</div>
<!-- Loading indicator -->
<div v-if="loading" class="flex justify-start">
<div class="bg-gray-100 text-gray-800 px-4 py-2 rounded-lg flex items-center space-x-2">
<div class="animate-spin w-4 h-4 border-2 border-gray-400 border-t-transparent rounded-full"></div>
<span>AlexAI is thinking...</span>
</div>
</div>
</div>
<!-- Message Input -->
<div class="bg-white border-t p-4">
<div class="flex space-x-2">
<input
v-model="currentMessage"
@keyup.enter="sendMessage"
@keydown.ctrl.enter="sendMessage"
type="text"
placeholder="Type your message... (Enter to send, Ctrl+Enter for new line)"
class="flex-1 border border-gray-300 rounded-lg px-4 py-2 focus:ring-2 focus:ring-blue-500 focus:border-transparent"
:disabled="loading"
/>
<button
@click="sendMessage"
:disabled="loading || !currentMessage.trim()"
class="px-6 py-2 bg-blue-600 text-white rounded-lg hover:bg-blue-700 disabled:opacity-50 disabled:cursor-not-allowed transition-colors"
>
Send
</button>
</div>
<!-- Quick actions -->
<div class="flex space-x-2 mt-2">
<button
v-for="quick in quickActions"
:key="quick.text"
@click="sendQuickMessage(quick.text)"
class="px-3 py-1 bg-gray-100 text-gray-700 rounded text-sm hover:bg-gray-200 transition-colors"
>
{{ quick.label }}
</button>
</div>
</div>
</div>
</template>
<script>
import { ref, nextTick, watch } from 'vue'
export default {
name: 'ChatInterface',
props: {
messages: {
type: Array,
default: () => []
},
loading: {
type: Boolean,
default: false
}
},
emits: ['send-message', 'clear-chat'],
setup(props, { emit }) {
const currentMessage = ref('')
const voiceEnabled = ref(false)
const messagesContainer = ref(null)
const quickActions = [
{ label: 'Help', text: 'What can you help me with?' },
{ label: 'Status', text: 'What is your current status?' },
{ label: 'Capabilities', text: 'What are your capabilities?' },
{ label: 'Privacy', text: 'How do you protect my privacy?' }
]
const sendMessage = () => {
if (currentMessage.value.trim() && !props.loading) {
emit('send-message', currentMessage.value.trim())
currentMessage.value = ''
}
}
const sendQuickMessage = (message) => {
if (!props.loading) {
emit('send-message', message)
}
}
const toggleVoiceResponse = () => {
voiceEnabled.value = !voiceEnabled.value
}
const formatTime = (date) => {
return new Intl.DateTimeFormat('en-US', {
hour: '2-digit',
minute: '2-digit',
second: '2-digit'
}).format(date)
}
const scrollToBottom = () => {
nextTick(() => {
if (messagesContainer.value) {
messagesContainer.value.scrollTop = messagesContainer.value.scrollHeight
}
})
}
// Auto-scroll when new messages arrive
watch(() => props.messages.length, scrollToBottom)
return {
currentMessage,
voiceEnabled,
messagesContainer,
quickActions,
sendMessage,
sendQuickMessage,
toggleVoiceResponse,
formatTime
}
}
}
</script>Document Processor Component
<!-- components/Document/DocumentProcessor.vue -->
<template>
<div class="flex flex-col h-full">
<!-- Header -->
<div class="bg-white border-b px-6 py-4">
<h2 class="text-lg font-semibold text-gray-800">Document Processing</h2>
<p class="text-sm text-gray-600 mt-1">Upload images and documents for OCR and AI analysis</p>
</div>
<div class="flex-1 p-6 overflow-y-auto">
<!-- Upload Area -->
<div
@drop="handleDrop"
@dragover.prevent
@dragenter.prevent
:class="[
'border-2 border-dashed rounded-lg p-8 text-center transition-colors',
isDragging ? 'border-blue-500 bg-blue-50' : 'border-gray-300'
]"
>
<svg class="mx-auto h-12 w-12 text-gray-400" stroke="currentColor" fill="none" viewBox="0 0 48 48">
<path d="M28 8H12a4 4 0 00-4 4v20m32-12v8m0 0v8a4 4 0 01-4 4H12a4 4 0 01-4-4v-4m32-4l-3.172-3.172a4 4 0 00-5.656 0L28 28M8 32l9.172-9.172a4 4 0 015.656 0L28 28m0 0l4 4m4-24h8m-4-4v8m-12 4h.02" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"/>
</svg>
<p class="mt-2 text-sm text-gray-600">
<span class="font-medium">Drop files here</span> or
<label class="text-blue-600 hover:text-blue-500 cursor-pointer">
<span>browse</span>
<input
ref="fileInput"
type="file"
class="sr-only"
accept="image/*,.pdf"
multiple
@change="handleFileSelect"
/>
</label>
</p>
<p class="text-xs text-gray-500 mt-1">Supports: JPG, PNG, PDF (up to 10MB)</p>
</div>
<!-- Analysis Options -->
<div class="mt-6 bg-white rounded-lg border p-4">
<h3 class="text-sm font-medium text-gray-800 mb-3">Analysis Options</h3>
<div class="space-y-3">
<div>
<label class="block text-sm font-medium text-gray-700 mb-1">Analysis Type</label>
<select
v-model="analysisType"
class="w-full border border-gray-300 rounded-md px-3 py-2 focus:ring-2 focus:ring-blue-500 focus:border-transparent"
>
<option value="extract">Extract Text Only</option>
<option value="analyze">Full Analysis</option>
<option value="qa">Question & Answer</option>
</select>
</div>
<div>
<label class="block text-sm font-medium text-gray-700 mb-1">Enhancement Type</label>
<select
v-model="enhancementType"
class="w-full border border-gray-300 rounded-md px-3 py-2 focus:ring-2 focus:ring-blue-500 focus:border-transparent"
>
<option value="default">Default</option>
<option value="scan">Scanned Document</option>
<option value="photo">Photo of Document</option>
<option value="screenshot">Screenshot</option>
</select>
</div>
<div v-if="analysisType === 'qa'">
<label class="block text-sm font-medium text-gray-700 mb-1">Question</label>
<input
v-model="question"
type="text"
placeholder="What would you like to know about this document?"
class="w-full border border-gray-300 rounded-md px-3 py-2 focus:ring-2 focus:ring-blue-500 focus:border-transparent"
/>
</div>
</div>
</div>
<!-- File Queue -->
<div v-if="fileQueue.length > 0" class="mt-6">
<h3 class="text-sm font-medium text-gray-800 mb-3">Files to Process</h3>
<div class="space-y-2">
<div
v-for="(file, index) in fileQueue"
:key="index"
class="flex items-center justify-between bg-gray-50 rounded-lg p-3"
>
<div class="flex items-center space-x-3">
<div class="flex-shrink-0">
<svg class="h-8 w-8 text-gray-400" fill="currentColor" viewBox="0 0 20 20">
<path fill-rule="evenodd" d="M4 4a2 2 0 012-2h4.586A2 2 0 0112 2.586L15.414 6A2 2 0 0116 7.414V16a2 2 0 01-2 2H6a2 2 0 01-2-2V4zm2 6a1 1 0 011-1h6a1 1 0 110 2H7a1 1 0 01-1-1zm1 3a1 1 0 100 2h6a1 1 0 100-2H7z" clip-rule="evenodd"/>
</svg>
</div>
<div>
<p class="text-sm font-medium text-gray-900">{{ file.name }}</p>
<p class="text-xs text-gray-500">{{ formatFileSize(file.size) }}</p>
</div>
</div>
<div class="flex items-center space-x-2">
<span
v-if="file.status === 'processing'"
class="text-xs text-blue-600 bg-blue-100 px-2 py-1 rounded"
>
Processing...
</span>
<span
v-else-if="file.status === 'completed'"
class="text-xs text-green-600 bg-green-100 px-2 py-1 rounded"
>
Completed
</span>
<span
v-else-if="file.status === 'error'"
class="text-xs text-red-600 bg-red-100 px-2 py-1 rounded"
>
Error
</span>
<button
@click="removeFile(index)"
class="text-gray-400 hover:text-red-500 transition-colors"
>
<svg class="h-4 w-4" fill="currentColor" viewBox="0 0 20 20">
<path fill-rule="evenodd" d="M4.293 4.293a1 1 0 011.414 0L10 8.586l4.293-4.293a1 1 0 111.414 1.414L11.414 10l4.293 4.293a1 1 0 01-1.414 1.414L10 11.414l-4.293 4.293a1 1 0 01-1.414-1.414L8.586 10 4.293 5.707a1 1 0 010-1.414z" clip-rule="evenodd"/>
</svg>
</button>
</div>
</div>
</div>
<!-- Process Button -->
<div class="mt-4">
<button
@click="processFiles"
:disabled="processing || fileQueue.every(f => f.status === 'completed')"
class="w-full bg-blue-600 text-white py-2 px-4 rounded-lg hover:bg-blue-700 disabled:opacity-50 disabled:cursor-not-allowed transition-colors"
>
{{ processing ? 'Processing...' : 'Process Documents' }}
</button>
</div>
</div>
<!-- Results -->
<div v-if="results.length > 0" class="mt-6">
<h3 class="text-sm font-medium text-gray-800 mb-3">Processing Results</h3>
<div class="space-y-4">
<div
v-for="(result, index) in results"
:key="index"
class="bg-white border rounded-lg p-4"
>
<div class="flex justify-between items-start mb-3">
<h4 class="font-medium text-gray-900">{{ result.filename }}</h4>
<div class="flex space-x-2 text-xs">
<span class="bg-blue-100 text-blue-800 px-2 py-1 rounded">
{{ result.confidence.toFixed(1) }}% confidence
</span>
<span class="bg-gray-100 text-gray-800 px-2 py-1 rounded">
{{ result.word_count }} words
</span>
</div>
</div>
<!-- Extracted Text -->
<div v-if="result.text" class="mb-3">
<h5 class="text-sm font-medium text-gray-700 mb-1">Extracted Text:</h5>
<div class="bg-gray-50 rounded p-3 text-sm max-h-32 overflow-y-auto">
{{ result.text }}
</div>
</div>
<!-- Analysis -->
<div v-if="result.analysis" class="mb-3">
<h5 class="text-sm font-medium text-gray-700 mb-1">Analysis:</h5>
<div class="bg-blue-50 rounded p-3 text-sm max-h-32 overflow-y-auto">
{{ result.analysis }}
</div>
</div>
<!-- Actions -->
<div class="flex space-x-2">
<button
@click="copyToClipboard(result.text || result.analysis)"
class="text-xs bg-gray-100 text-gray-700 px-3 py-1 rounded hover:bg-gray-200 transition-colors"
>
Copy Text
</button>
<button
@click="addToKnowledge(result)"
class="text-xs bg-green-100 text-green-700 px-3 py-1 rounded hover:bg-green-200 transition-colors"
>
Add to Knowledge Base
</button>
</div>
</div>
</div>
</div>
</div>
</div>
</template>
<script>
import { ref } from 'vue'
export default {
name: 'DocumentProcessor',
emits: ['process-document'],
setup(props, { emit }) {
const fileInput = ref(null)
const isDragging = ref(false)
const fileQueue = ref([])
const processing = ref(false)
const results = ref([])
// Options
const analysisType = ref('analyze')
const enhancementType = ref('default')
const question = ref('')
const handleDrop = (e) => {
e.preventDefault()
isDragging.value = false
const files = Array.from(e.dataTransfer.files)
addFiles(files)
}
const handleFileSelect = (e) => {
const files = Array.from(e.target.files)
addFiles(files)
e.target.value = '' // Reset input
}
const addFiles = (files) => {
const validFiles = files.filter(file => {
const isValidType = file.type.startsWith('image/') || file.type === 'application/pdf'
const isValidSize = file.size <= 10 * 1024 * 1024 // 10MB
return isValidType && isValidSize
})
validFiles.forEach(file => {
fileQueue.value.push({
file,
name: file.name,
size: file.size,
status: 'pending'
})
})
}
const removeFile = (index) => {
fileQueue.value.splice(index, 1)
}
const processFiles = async () => {
if (processing.value) return
processing.value = true
try {
for (let i = 0; i < fileQueue.value.length; i++) {
const fileItem = fileQueue.value[i]
if (fileItem.status === 'completed') continue
fileItem.status = 'processing'
try {
const result = await emit('process-document',
fileItem.file,
analysisType.value,
question.value
)
results.value.push({
filename: fileItem.name,
...result
})
fileItem.status = 'completed'
} catch (error) {
console.error('Processing failed:', error)
fileItem.status = 'error'
}
}
} finally {
processing.value = false
}
}
const formatFileSize = (bytes) => {
if (bytes === 0) return '0 Bytes'
const k = 1024
const sizes = ['Bytes', 'KB', 'MB', 'GB']
const i = Math.floor(Math.log(bytes) / Math.log(k))
return parseFloat((bytes / Math.pow(k, i)).toFixed(2)) + ' ' + sizes[i]
}
const copyToClipboard = async (text) => {
try {
await navigator.clipboard.writeText(text)
// Could add notification here
} catch (error) {
console.error('Failed to copy:', error)
}
}
const addToKnowledge = (result) => {
// Could emit event to parent to add to knowledge base
emit('add-to-knowledge', {
content: result.text || result.analysis,
metadata: {
source: 'document_processing',
filename: result.filename,
confidence: result.confidence
}
})
}
return {
fileInput,
isDragging,
fileQueue,
processing,
results,
analysisType,
enhancementType,
question,
handleDrop,
handleFileSelect,
removeFile,
processFiles,
formatFileSize,
copyToClipboard,
addToKnowledge
}
}
}
</script>Voice Interface Component
<!-- components/Voice/VoiceInterface.vue -->
<template>
<div class="flex flex-col h-full">
<!-- Header -->
<div class="bg-white border-b px-6 py-4">
<h2 class="text-lg font-semibold text-gray-800">Voice Interaction</h2>
<p class="text-sm text-gray-600 mt-1">Talk to AlexAI using voice commands</p>
</div>
<div class="flex-1 p-6">
<!-- Voice Status -->
<div class="text-center mb-8">
<div
:class="[
'w-32 h-32 rounded-full mx-auto mb-4 flex items-center justify-center transition-all duration-300',
isListening ? 'bg-red-100 border-4 border-red-300 animate-pulse' :
isSpeaking ? 'bg-blue-100 border-4 border-blue-300 animate-pulse' :
'bg-gray-100 border-4 border-gray-300'
]"
>
<svg
:class="[
'w-16 h-16 transition-colors',
isListening ? 'text-red-600' :
isSpeaking ? 'text-blue-600' :
'text-gray-600'
]"
fill="currentColor"
viewBox="0 0 20 20"
>
<path fill-rule="evenodd" d="M7 4a3 3 0 016 0v4a3 3 0 11-6 0V4zm4 10.93A7.001 7.001 0 0017 8a1 1 0 10-2 0A5 5 0 015 8a1 1 0 00-2 0 7.001 7.001 0 006 6.93V17H6a1 1 0 100 2h8a1 1 0 100-2h-3v-2.07z" clip-rule="evenodd"/>
</svg>
</div>
<p class="text-lg font-medium text-gray-900 mb-2">
{{ statusText }}
</p>
<p class="text-sm text-gray-600">
{{ statusDescription }}
</p>
</div>
<!-- Voice Controls -->
<div class="max-w-md mx-auto space-y-4">
<!-- Quick Voice Actions -->
<div class="grid grid-cols-2 gap-4">
<button
@click="startListening"
:disabled="isListening || isSpeaking"
class="flex flex-col items-center p-4 border-2 border-dashed border-gray-300 rounded-lg hover:border-blue-500 hover:bg-blue-50 disabled:opacity-50 disabled:cursor-not-allowed transition-colors"
>
<svg class="w-8 h-8 text-gray-600 mb-2" fill="currentColor" viewBox="0 0 20 20">
<path fill-rule="evenodd" d="M7 4a3 3 0 016 0v4a3 3 0 11-6 0V4zm4 10.93A7.001 7.001 0 0017 8a1 1 0 10-2 0A5 5 0 015 8a1 1 0 00-2 0 7.001 7.001 0 006 6.93V17H6a1 1 0 100 2h8a1 1 0 100-2h-3v-2.07z" clip-rule="evenodd"/>
</svg>
<span class="text-sm font-medium">Start Conversation</span>
</button>
<button
@click="showTextToSpeech = true"
:disabled="isListening || isSpeaking"
class="flex flex-col items-center p-4 border-2 border-dashed border-gray-300 rounded-lg hover:border-green-500 hover:bg-green-50 disabled:opacity-50 disabled:cursor-not-allowed transition-colors"
>
<svg class="w-8 h-8 text-gray-600 mb-2" fill="currentColor" viewBox="0 0 20 20">
<path fill-rule="evenodd" d="M9.383 3.076A1 1 0 0110 4v12a1 1 0 01-1.617.768L4.383 13H2a1 1 0 01-1-1V8a1 1 0 011-1h2.383l4-3.232zm7.617 2.924a1 1 0 01.293.707v6.586a1 1 0 01-1.707.707L14 12.414V7.586l1.586-1.586a1 1 0 011.414 0z" clip-rule="evenodd"/>
</svg>
<span class="text-sm font-medium">Text to Speech</span>
</button>
</div>
<!-- Stop Button (when active) -->
<button
v-if="isListening"
@click="stopListening"
class="w-full bg-red-600 text-white py-3 px-6 rounded-lg hover:bg-red-700 transition-colors flex items-center justify-center"
>
<svg class="w-5 h-5 mr-2" fill="currentColor" viewBox="0 0 20 20">
<path fill-rule="evenodd" d="M10 18a8 8 0 100-16 8 8 0 000 16zM8 7a1 1 0 00-1 1v4a1 1 0 001 1h4a1 1 0 001-1V8a1 1 0 00-1-1H8z" clip-rule="evenodd"/>
</svg>
Stop Listening
</button>
<!-- Settings -->
<div class="bg-gray-50 rounded-lg p-4">
<h3 class="text-sm font-medium text-gray-800 mb-3">Voice Settings</h3>
<div class="space-y-3">
<div class="flex items-center justify-between">
<label class="text-sm text-gray-700">Wake Word Detection</label>
<input
v-model="wakeWordEnabled"
type="checkbox"
class="rounded border-gray-300 text-blue-600 focus:ring-blue-500"
/>
</div>
<div class="flex items-center justify-between">
<label class="text-sm text-gray-700">Auto Response</label>
<input
v-model="autoResponse"
type="checkbox"
class="rounded border-gray-300 text-blue-600 focus:ring-blue-500"
/>
</div>
<div>
<label class="block text-sm text-gray-700 mb-1">Timeout (seconds)</label>
<input
v-model.number="timeout"
type="range"
min="10"
max="60"
class="w-full"
/>
<div class="text-xs text-gray-500 text-center">{{ timeout }}s</div>
</div>
</div>
</div>
<!-- Voice History -->
<div v-if="voiceHistory.length > 0" class="bg-white border rounded-lg p-4">
<h3 class="text-sm font-medium text-gray-800 mb-3">Recent Voice Interactions</h3>
<div class="space-y-2 max-h-32 overflow-y-auto">
<div
v-for="(interaction, index) in voiceHistory.slice(-5)"
:key="index"
class="text-xs p-2 bg-gray-50 rounded"
>
<div class="font-medium text-gray-900">{{ interaction.timestamp }}</div>
<div class="text-gray-600">{{ interaction.text }}</div>
</div>
</div>
</div>
</div>
<!-- Text to Speech Modal -->
<div
v-if="showTextToSpeech"
class="fixed inset-0 bg-black bg-opacity-50 flex items-center justify-center z-50"
@click="showTextToSpeech = false"
>
<div
class="bg-white rounded-lg p-6 max-w-md w-full mx-4"
@click.stop
>
<h3 class="text-lg font-medium text-gray-900 mb-4">Text to Speech</h3>
<textarea
v-model="textToSpeak"
placeholder="Enter text to speak..."
class="w-full border border-gray-300 rounded-md px-3 py-2 h-24 resize-none focus:ring-2 focus:ring-blue-500 focus:border-transparent"
></textarea>
<div class="flex justify-end space-x-3 mt-4">
<button
@click="showTextToSpeech = false"
class="px-4 py-2 text-gray-700 border border-gray-300 rounded-md hover:bg-gray-50 transition-colors"
>
Cancel
</button>
<button
@click="speakText"
:disabled="!textToSpeak.trim() || isSpeaking"
class="px-4 py-2 bg-blue-600 text-white rounded-md hover:bg-blue-700 disabled:opacity-50 disabled:cursor-not-allowed transition-colors"
>
{{ isSpeaking ? 'Speaking...' : 'Speak' }}
</button>
</div>
</div>
</div>
</div>
</div>
</template>
<script>
import { ref, computed } from 'vue'
export default {
name: 'VoiceInterface',
props: {
isListening: {
type: Boolean,
default: false
},
isSpeaking: {
type: Boolean,
default: false
}
},
emits: ['start-listening', 'stop-listening', 'text-to-speech'],
setup(props, { emit }) {
const showTextToSpeech = ref(false)
const textToSpeak = ref('')
const wakeWordEnabled = ref(true)
const autoResponse = ref(true)
const timeout = ref(30)
const voiceHistory = ref([])
const statusText = computed(() => {
if (props.isListening) return 'Listening...'
if (props.isSpeaking) return 'Speaking...'
return 'Ready'
})
const statusDescription = computed(() => {
if (props.isListening) return 'Say something to AlexAI'
if (props.isSpeaking) return 'AlexAI is responding'
return 'Click to start voice interaction'
})
const startListening = () => {
emit('start-listening')
addToHistory('Voice interaction started')
}
const stopListening = () => {
emit('stop-listening')
addToHistory('Voice interaction stopped')
}
const speakText = () => {
if (textToSpeak.value.trim()) {
emit('text-to-speech', textToSpeak.value.trim())
addToHistory(`TTS: ${textToSpeak.value.substring(0, 50)}...`)
textToSpeak.value = ''
showTextToSpeech.value = false
}
}
const addToHistory = (text) => {
voiceHistory.value.push({
timestamp: new Date().toLocaleTimeString(),
text: text
})
// Keep only last 10 entries
if (voiceHistory.value.length > 10) {
voiceHistory.value = voiceHistory.value.slice(-10)
}
}
return {
showTextToSpeech,
textToSpeak,
wakeWordEnabled,
autoResponse,
timeout,
voiceHistory,
statusText,
statusDescription,
startListening,
stopListening,
speakText
}
}
}
</script>Notifications Composable
// composables/useNotifications.js
import { ref } from 'vue'
export function useNotifications() {
const notifications = ref([])
let notificationId = 0
const addNotification = (message, type = 'info', duration = 5000) => {
const id = ++notificationId
notifications.value.push({
id,
message,
type,
timestamp: new Date()
})
// Auto remove after duration
if (duration > 0) {
setTimeout(() => {
removeNotification(id)
}, duration)
}
return id
}
const removeNotification = (id) => {
const index = notifications.value.findIndex(n => n.id === id)
if (index > -1) {
notifications.value.splice(index, 1)
}
}
const clearNotifications = () => {
notifications.value = []
}
return {
notifications,
addNotification,
removeNotification,
clearNotifications
}
}Project Setup Files
// package.json
{
"name": "alexai-frontend",
"version": "1.0.0",
"scripts": {
"dev": "vite",
"build": "vite build",
"preview": "vite preview",
"lint": "eslint src --ext .vue,.js,.ts",
"format": "prettier --write src/"
},
"dependencies": {
"vue": "^3.4.0",
"@vue/composition-api": "^1.7.2",
"vue-router": "^4.2.0",
"pinia": "^2.1.0"
},
"devDependencies": {
"@vitejs/plugin-vue": "^4.5.0",
"vite": "^5.0.0",
"tailwindcss": "^3.3.0",
"autoprefixer": "^10.4.0",
"postcss": "^8.4.0",
"eslint": "^8.0.0",
"prettier": "^3.0.0"
}
}// vite.config.js
import { defineConfig } from 'vite'
import vue from '@vitejs/plugin-vue'
import path from 'path'
export default defineConfig({
plugins: [vue()],
resolve: {
alias: {
'@': path.resolve(__dirname, 'src')
}
},
server: {
port: 3000,
proxy: {
'/api': {
target: 'http://localhost:8000',
changeOrigin: true,
rewrite: (path) => path.replace(/^\/api/, '')
}
}
},
build: {
outDir: 'dist',
sourcemap: true
}
})// tailwind.config.js
module.exports = {
content: [
"./index.html",
"./src/**/*.{vue,js,ts,jsx,tsx}",
],
theme: {
extend: {
fontFamily: {
sans: ['Inter', 'system-ui', 'sans-serif'],
},
colors: {
primary: {
50: '#eff6ff',
500: '#3b82f6',
600: '#2563eb',
700: '#1d4ed8',
}
}
},
},
plugins: [],
}## Part 7: MCP (Model Context Protocol) Integration
### Why MCP?
The Model Context Protocol standardizes how AI models interact with external tools and data sources:
- **Standardized interface**: Consistent way to connect tools across different AI systems
- **Security**: Controlled access to external resources
- **Extensibility**: Easy addition of new capabilities
- **Interoperability**: Tools work across different AI platforms
### MCP Server Implementation
```python
# mcp/mcp_server.py
from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, Field
from typing import List, Dict, Any, Optional, Union
import asyncio
import logging
from datetime import datetime
import json
# MCP Protocol Models
class MCPResource(BaseModel):
uri: str
name: str
description: Optional[str] = None
mime_type: Optional[str] = None
class MCPTool(BaseModel):
name: str
description: str
input_schema: Dict[str, Any]
class MCPPrompt(BaseModel):
name: str
description: str
arguments: Optional[List[Dict[str, Any]]] = None
class MCPRequest(BaseModel):
method: str
params: Optional[Dict[str, Any]] = None
class MCPResponse(BaseModel):
result: Optional[Any] = None
error: Optional[Dict[str, Any]] = None
class MCPServer:
def __init__(self):
self.app = FastAPI(title="AlexAI MCP Server", version="1.0.0")
self.security = HTTPBearer()
# Register available tools and resources
self.tools = {}
self.resources = {}
self.prompts = {}
# Setup routes
self._setup_routes()
# Initialize integrations
self._setup_integrations()
def _setup_integrations(self):
"""Initialize all system integrations"""
from models.ollama_manager import OllamaManager
from rag.vector_store import VectorStore, RAGGenerator
from memory.memory_manager import MemoryManager
from voice.speech_processor import VoiceInterface
from vision.ocr_processor import VisionLanguageProcessor
self.ollama_manager = OllamaManager()
self.vector_store = VectorStore()
self.rag_generator = RAGGenerator()
self.memory_manager = MemoryManager()
self.voice_interface = VoiceInterface()
self.vision_processor = VisionLanguageProcessor(self.ollama_manager)
# Register tools
self._register_tools()
def _register_tools(self):
"""Register all available tools"""
# Chat tool
self.tools['chat'] = MCPTool(
name="chat",
description="Have a conversation with the AI assistant",
input_schema={
"type": "object",
"properties": {
"message": {"type": "string", "description": "User message"},
"session_id": {"type": "string", "description": "Session identifier"},
"use_memory": {"type": "boolean", "default": True},
"use_rag": {"type": "boolean", "default": True}
},
"required": ["message"]
}
)
# Document processing tool
self.tools['process_document'] = MCPTool(
name="process_document",
description="Process and analyze documents using OCR and LLM",
input_schema={
"type": "object",
"properties": {
"file_path": {"type": "string", "description": "Path to document image"},
"analysis_type": {"type": "string", "enum": ["extract", "analyze", "qa"], "default": "analyze"},
"question": {"type": "string", "description": "Question for QA mode"}
},
"required": ["file_path"]
}
)
# Voice interaction tool
self.tools['voice_chat'] = MCPTool(
name="voice_chat",
description="Voice-based conversation with the assistant",
input_schema={
"type": "object",
"properties": {
"mode": {"type": "string", "enum": ["listen", "speak", "conversation"], "default": "conversation"},
"text": {"type": "string", "description": "Text to speak (for speak mode)"},
"timeout": {"type": "number", "default": 30}
}
}
)
# Memory management tool
self.tools['manage_memory'] = MCPTool(
name="manage_memory",
description="Manage conversation memory and user preferences",
input_schema={
"type": "object",
"properties": {
"action": {"type": "string", "enum": ["get_context", "get_preferences", "learn_preference"]},
"session_id": {"type": "string"},
"category": {"type": "string"},
"key": {"type": "string"},
"value": {"type": "string"},
"confidence": {"type": "number", "default": 0.5}
},
"required": ["action"]
}
)
# Knowledge base tool
self.tools['knowledge_base'] = MCPTool(
name="knowledge_base",
description="Search and manage knowledge base",
input_schema={
"type": "object",
"properties": {
"action": {"type": "string", "enum": ["search", "add", "update", "delete"]},
"query": {"type": "string", "description": "Search query or content"},
"collection": {"type": "string", "default": "documents"},
"metadata": {"type": "object", "description": "Document metadata"}
},
"required": ["action"]
}
)
def _setup_routes(self):
"""Setup FastAPI routes for MCP protocol"""
@self.app.get("/")
async def root():
return {"service": "AlexAI MCP Server", "version": "1.0.0", "status": "running"}
@self.app.post("/mcp/initialize")
async def initialize(credentials: HTTPAuthorizationCredentials = Depends(self.security)):
"""Initialize MCP session"""
return MCPResponse(result={
"protocol_version": "1.0",
"server_info": {
"name": "AlexAI",
"version": "1.0.0"
},
"capabilities": {
"tools": True,
"resources": True,
"prompts": True
}
})
@self.app.get("/mcp/tools")
async def list_tools():
"""List available tools"""
return MCPResponse(result={"tools": list(self.tools.values())})
@self.app.post("/mcp/tools/call")
async def call_tool(request: MCPRequest):
"""Call a specific tool"""
tool_name = request.params.get("name")
arguments = request.params.get("arguments", {})
if tool_name not in self.tools:
return MCPResponse(error={"code": -32601, "message": f"Tool {tool_name} not found"})
try:
result = await self._execute_tool(tool_name, arguments)
return MCPResponse(result=result)
except Exception as e:
logging.error(f"Tool execution failed: {e}")
return MCPResponse(error={"code": -32603, "message": str(e)})
@self.app.get("/mcp/resources")
async def list_resources():
"""List available resources"""
return MCPResponse(result={"resources": list(self.resources.values())})
@self.app.post("/mcp/resources/read")
async def read_resource(request: MCPRequest):
"""Read a specific resource"""
uri = request.params.get("uri")
try:
content = await self._read_resource(uri)
return MCPResponse(result={"contents": [{"uri": uri, "text": content}]})
except Exception as e:
return MCPResponse(error={"code": -32603, "message": str(e)})
async def _execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
"""Execute a tool with given arguments"""
if tool_name == "chat":
return await self._handle_chat(arguments)
elif tool_name == "process_document":
return await self._handle_document_processing(arguments)
elif tool_name == "voice_chat":
return await self._handle_voice_chat(arguments)
elif tool_name == "manage_memory":
return await self._handle_memory_management(arguments)
elif tool_name == "knowledge_base":
return await self._handle_knowledge_base(arguments)
else:
raise ValueError(f"Unknown tool: {tool_name}")
async def _handle_chat(self, args: Dict[str, Any]) -> Dict[str, Any]:
"""Handle chat tool execution"""
message = args["message"]
session_id = args.get("session_id", "default")
use_memory = args.get("use_memory", True)
use_rag = args.get("use_rag", True)
# Get conversation context if memory is enabled
context = ""
if use_memory:
conversations = await self.memory_manager.get_conversation_context(
session_id=session_id,
query=message
)
if conversations:
context_parts = []
for conv in conversations[-3:]: # Last 3 relevant conversations
context_parts.append(f"User: {conv['user_message']}")
context_parts.append(f"Assistant: {conv['assistant_response']}")
context = "\n".join(context_parts)
# Generate response
if use_rag:
# Use RAG for enhanced responses
response = await self.rag_generator.generate_with_context(
query=message,
system_prompt=f"Previous conversation context:\n{context}" if context else None
)
else:
# Direct LLM response
prompt = f"{context}\n\nUser: {message}" if context else message
response = await self.ollama_manager.generate_response(
prompt=prompt,
system_prompt="You are AlexAI, a helpful personal assistant."
)
# Store conversation in memory
if use_memory:
await self.memory_manager.store_conversation(
session_id=session_id,
user_message=message,
assistant_response=response
)
return {
"response": response,
"session_id": session_id,
"used_memory": use_memory,
"used_rag": use_rag
}
async def _handle_document_processing(self, args: Dict[str, Any]) -> Dict[str, Any]:
"""Handle document processing tool"""
file_path = args["file_path"]
analysis_type = args.get("analysis_type", "analyze")
question = args.get("question")
if analysis_type == "extract":
# Simple OCR extraction
result = self.vision_processor.ocr.extract_text(file_path)
return {
"text": result["text"],
"confidence": result.get("confidence", 0),
"word_count": result.get("word_count", 0)
}
elif analysis_type == "analyze":
# Full document analysis
analysis = await self.vision_processor.analyze_document(file_path)
return analysis
elif analysis_type == "qa":
if not question:
raise ValueError("Question required for QA mode")
answer = await self.vision_processor.answer_document_questions(file_path, question)
return {
"question": question,
"answer": answer
}
else:
raise ValueError(f"Unknown analysis type: {analysis_type}")
async def _handle_voice_chat(self, args: Dict[str, Any]) -> Dict[str, Any]:
"""Handle voice chat tool"""
mode = args.get("mode", "conversation")
text = args.get("text")
timeout = args.get("timeout", 30)
if mode == "speak":
if not text:
raise ValueError("Text required for speak mode")
await self.voice_interface.tts.speak_async(text)
return {"status": "spoken", "text": text}
elif mode == "listen":
# Listen for wake word
detected = await self.voice_interface.listen_for_wake_word(timeout)
return {"wake_word_detected": detected}
elif mode == "conversation":
# Full voice conversation
async def chat_callback(user_text: str) -> str:
# Use chat tool to generate response
chat_result = await self._handle_chat({
"message": user_text,
"session_id": "voice_session",
"use_memory": True,
"use_rag": True
})
return chat_result["response"]
response = await self.voice_interface.voice_conversation(chat_callback)
return {"response": response}
else:
raise ValueError(f"Unknown voice mode: {mode}")
async def _handle_memory_management(self, args: Dict[str, Any]) -> Dict[str, Any]:
"""Handle memory management tool"""
action = args["action"]
if action == "get_context":
session_id = args.get("session_id", "default")
context = await self.memory_manager.get_conversation_context(session_id)
return {"context": context}
elif action == "get_preferences":
category = args.get("category")
preferences = self.memory_manager.get_user_preferences(category)
return {"preferences": preferences}
elif action == "learn_preference":
category = args["category"]
key = args["key"]
value = args["value"]
confidence = args.get("confidence", 0.5)
await self.memory_manager.learn_preference(category, key, value, confidence)
return {"status": "preference_learned"}
else:
raise ValueError(f"Unknown memory action: {action}")
async def _handle_knowledge_base(self, args: Dict[str, Any]) -> Dict[str, Any]:
"""Handle knowledge base tool"""
action = args["action"]
if action == "search":
query = args["query"]
collection = args.get("collection", "documents")
results = await self.vector_store.similarity_search(
query=query,
collection_name=collection
)
return {"results": results}
elif action == "add":
content = args["query"] # Using query field for content
collection = args.get("collection", "documents")
metadata = args.get("metadata", {})
doc_id = await self.vector_store.add_document(
content=content,
metadata=metadata,
collection_name=collection
)
return {"document_id": doc_id}
else:
raise ValueError(f"Unknown knowledge base action: {action}")
async def _read_resource(self, uri: str) -> str:
"""Read content from a resource URI"""
# Implement resource reading logic based on URI scheme
if uri.startswith("file://"):
file_path = uri[7:] # Remove file:// prefix
with open(file_path, 'r', encoding='utf-8') as f:
return f.read()
elif uri.startswith("memory://"):
# Read from memory system
# Implementation depends on specific memory resource format
return "Memory resource content"
else:
raise ValueError(f"Unsupported resource URI: {uri}")
# MCP Client for connecting to other services
class MCPClient:
def __init__(self, server_url: str, auth_token: Optional[str] = None):
self.server_url = server_url
self.auth_token = auth_token
self.session = None
async def initialize(self) -> Dict[str, Any]:
"""Initialize connection with MCP server"""
import aiohttp
headers = {}
if self.auth_token:
headers["Authorization"] = f"Bearer {self.auth_token}"
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.server_url}/mcp/initialize",
headers=headers
) as response:
result = await response.json()
return result
async def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Call a tool on the MCP server"""
import aiohttp
request_data = {
"method": "tools/call",
"params": {
"name": tool_name,
"arguments": arguments
}
}
headers = {}
if self.auth_token:
headers["Authorization"] = f"Bearer {self.auth_token}"
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.server_url}/mcp/tools/call",
json=request_data,
headers=headers
) as response:
result = await response.json()
return result
async def list_tools(self) -> List[Dict[str, Any]]:
"""List available tools"""
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.get(f"{self.server_url}/mcp/tools") as response:
result = await response.json()
return result.get("result", {}).get("tools", [])
# Usage example
async def demo_mcp_server():
"""Demonstrate MCP server functionality"""
# Start MCP server
server = MCPServer()
# Test chat tool
chat_result = await server._execute_tool("chat", {
"message": "Hello, how are you?",
"session_id": "demo_session"
})
print("Chat result:", chat_result)
# Test knowledge base
kb_result = await server._execute_tool("knowledge_base", {
"action": "add",
"query": "AlexAI is a privacy-focused personal assistant that runs locally.",
"metadata": {"category": "system_info"}
})
print("Knowledge base result:", kb_result)
# Search knowledge base
search_result = await server._execute_tool("knowledge_base", {
"action": "search",
"query": "privacy-focused assistant"
})
print("Search result:", search_result)
if __name__ == "__main__":
import uvicorn
# Create and run MCP server
server = MCPServer()
# Run with uvicorn
uvicorn.run(
server.app,
host="0.0.0.0",
port=8000,
log_level="info"
)Deployment and Build Configuration
<!-- public/index.html -->
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8" />
<link rel="icon" type="image/svg+xml" href="/favicon.ico" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<meta name="description" content="AlexAI - Privacy-focused AI Personal Assistant" />
<title>AlexAI Assistant</title>
<!-- PWA manifest -->
<link rel="manifest" href="/manifest.json" />
<meta name="theme-color" content="#2563eb" />
<!-- Apple touch icon -->
<link rel="apple-touch-icon" href="/apple-touch-icon.png" />
</head>
<body>
<div id="app"></div>
<script type="module" src="/src/main.js"></script>
</body>
</html>// public/manifest.json
{
"name": "AlexAI Assistant",
"short_name": "AlexAI",
"description": "Privacy-focused AI Personal Assistant",
"icons": [
{
"src": "/icon-192.png",
"sizes": "192x192",
"type": "image/png"
},
{
"src": "/icon-512.png",
"sizes": "512x512",
"type": "image/png"
}
],
"start_url": "/",
"display": "standalone",
"background_color": "#ffffff",
"theme_color": "#2563eb",
"categories": ["productivity", "utilities"]
}Production Build Script
#!/bin/bash
# build.sh - Production build script
echo "🚀 Building AlexAI Frontend..."
# Install dependencies
echo "📦 Installing dependencies..."
npm ci
# Run linting
echo "🔍 Running linter..."
npm run lint
# Build for production
echo "🏗️ Building for production..."
npm run build
# Copy to backend static directory
echo "📁 Copying to backend..."
if [ -d "../backend/static" ]; then
rm -rf ../backend/static/*
cp -r dist/* ../backend/static/
echo "✅ Static files copied to backend"
else
echo "⚠️ Backend static directory not found"
fi
echo "✨ Build complete!"Docker Configuration for Frontend
# Dockerfile.frontend
FROM node:18-alpine as build
WORKDIR /app
# Copy package files
COPY package*.json ./
# Install dependencies
RUN npm ci --only=production
# Copy source code
COPY . .
# Build application
RUN npm run build
# Production stage
FROM nginx:alpine
# Copy built files
COPY --from=build /app/dist /usr/share/nginx/html
# Copy nginx configuration
COPY nginx.conf /etc/nginx/nginx.conf
# Expose port
EXPOSE 80
# Health check
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
CMD curl -f http://localhost/ || exit 1
CMD ["nginx", "-g", "daemon off;"]# nginx.conf
events {
worker_connections 1024;
}
http {
include /etc/nginx/mime.types;
default_type application/octet-stream;
# Gzip compression
gzip on;
gzip_vary on;
gzip_min_length 1024;
gzip_types text/plain text/css application/json application/javascript text/xml application/xml application/xml+rss text/javascript;
server {
listen 80;
server_name localhost;
root /usr/share/nginx/html;
index index.html;
# Security headers
add_header X-Frame-Options "SAMEORIGIN" always;
add_header X-Content-Type-Options "nosniff" always;
add_header X-XSS-Protection "1; mode=block" always;
# Cache static assets
location ~* \.(js|css|png|jpg|jpeg|gif|ico|svg)$ {
expires 1y;
add_header Cache-Control "public, immutable";
}
# API proxy
location /api/ {
proxy_pass http://backend:8000/;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}
# SPA fallback
location / {
try_files $uri $uri/ /index.html;
}
}
}Complete Docker Compose with Frontend
# docker-compose.full.yml
version: '3.8'
services:
# Backend API
alexai-backend:
build:
context: ./backend
dockerfile: Dockerfile
container_name: alexai-backend
ports:
- "8000:8000"
volumes:
- ./data:/app/data
- ./documents:/app/documents
environment:
- OLLAMA_BASE_URL=http://ollama:11434
- MQTT_BROKER=mqtt
- CHROMA_PERSIST_DIR=/app/data/chroma_db
- MEMORY_DB_URL=sqlite:///./data/memory.db
depends_on:
- ollama
- mqtt
- chroma
restart: unless-stopped
networks:
- alexai-network
# Frontend
alexai-frontend:
build:
context: ./frontend
dockerfile: Dockerfile.frontend
container_name: alexai-frontend
ports:
- "80:80"
- "443:443"
volumes:
- ./ssl:/etc/nginx/ssl:ro
depends_on:
- alexai-backend
restart: unless-stopped
networks:
- alexai-network
# Ollama for local LLMs
ollama:
image: ollama/ollama:latest
container_name: alexai-ollama
ports:
- "11434:11434"
volumes:
- ollama_data:/root/.ollama
environment:
- OLLAMA_KEEP_ALIVE=24h
restart: unless-stopped
networks:
- alexai-network
# MQTT broker for A2A communication
mqtt:
image: eclipse-mosquitto:2
container_name: alexai-mqtt
ports:
- "1883:1883"
- "9001:9001"
volumes:
- ./mosquitto.conf:/mosquitto/config/mosquitto.conf
- mqtt_data:/mosquitto/data
- mqtt_logs:/mosquitto/log
restart: unless-stopped
networks:
- alexai-network
# ChromaDB for vector storage
chroma:
image: chromadb/chroma:latest
container_name: alexai-chroma
ports:
- "8001:8000"
volumes:
- chroma_data:/chroma/chroma
environment:
- CHROMA_SERVER_AUTH_CREDENTIALS_PROVIDER=chromadb.auth.token.TokenAuthCredentialsProvider
- CHROMA_SERVER_AUTH_CREDENTIALS=alexai-token
restart: unless-stopped
networks:
- alexai-network
volumes:
ollama_data:
chroma_data:
mqtt_data:
mqtt_logs:
networks:
alexai-network:
driver: bridgeEnvironment Configuration
# .env.production
VUE_APP_API_URL=http://localhost:8000
VUE_APP_WS_URL=ws://localhost:8000/ws
VUE_APP_VERSION=1.0.0
VUE_APP_SENTRY_DSN=
VUE_APP_ANALYTICS_ID=
# Feature flags
VUE_APP_ENABLE_VOICE=true
VUE_APP_ENABLE_DOCUMENT_PROCESSING=true
VUE_APP_ENABLE_A2A=true
VUE_APP_ENABLE_DEBUG=false
# Performance
VUE_APP_MAX_FILE_SIZE=10485760
VUE_APP_REQUEST_TIMEOUT=30000
VUE_APP_RETRY_ATTEMPTS=3Testing Configuration
// tests/unit/ChatInterface.spec.js
import { mount } from '@vue/test-utils'
import ChatInterface from '@/components/Chat/ChatInterface.vue'
describe('ChatInterface', () => {
it('renders properly', () => {
const wrapper = mount(ChatInterface, {
props: {
messages: [],
loading: false
}
})
expect(wrapper.find('h2').text()).toBe('Chat with AlexAI')
})
it('emits send-message when form is submitted', async () => {
const wrapper = mount(ChatInterface, {
props: {
messages: [],
loading: false
}
})
const input = wrapper.find('input[type="text"]')
const button = wrapper.find('button')
await input.setValue('Hello AlexAI')
await button.trigger('click')
expect(wrapper.emitted('send-message')).toBeTruthy()
expect(wrapper.emitted('send-message')[0]).toEqual(['Hello AlexAI'])
})
it('disables input when loading', () => {
const wrapper = mount(ChatInterface, {
props: {
messages: [],
loading: true
}
})
const input = wrapper.find('input[type="text"]')
const button = wrapper.find('button')
expect(input.attributes('disabled')).toBeDefined()
expect(button.attributes('disabled')).toBeDefined()
})
})Performance Optimizations
// src/utils/performance.js
export class PerformanceOptimizer {
constructor() {
this.debounceTimers = new Map()
this.throttleTimers = new Map()
}
// Debounce function calls
debounce(key, func, delay = 300) {
if (this.debounceTimers.has(key)) {
clearTimeout(this.debounceTimers.get(key))
}
const timer = setTimeout(() => {
func()
this.debounceTimers.delete(key)
}, delay)
this.debounceTimers.set(key, timer)
}
// Throttle function calls
throttle(key, func, delay = 100) {
if (this.throttleTimers.has(key)) {
return
}
func()
const timer = setTimeout(() => {
this.throttleTimers.delete(key)
}, delay)
this.throttleTimers.set(key, timer)
}
// Lazy load images
lazyLoadImages() {
const images = document.querySelectorAll('img[data-src]')
const imageObserver = new IntersectionObserver((entries, observer) => {
entries.forEach(entry => {
if (entry.isIntersecting) {
const img = entry.target
img.src = img.dataset.src
img.removeAttribute('data-src')
observer.unobserve(img)
}
})
})
images.forEach(img => imageObserver.observe(img))
}
// Virtual scrolling for large lists
virtualScroll(container, items, itemHeight, renderItem) {
const containerHeight = container.clientHeight
const scrollTop = container.scrollTop
const startIndex = Math.floor(scrollTop / itemHeight)
const endIndex = Math.min(
startIndex + Math.ceil(containerHeight / itemHeight) + 1,
items.length
)
// Clear container
container.innerHTML = ''
// Create spacer for items above viewport
if (startIndex > 0) {
const topSpacer = document.createElement('div')
topSpacer.style.height = `${startIndex * itemHeight}px`
container.appendChild(topSpacer)
}
// Render visible items
for (let i = startIndex; i < endIndex; i++) {
const element = renderItem(items[i], i)
container.appendChild(element)
}
// Create spacer for items below viewport
const remainingItems = items.length - endIndex
if (remainingItems > 0) {
const bottomSpacer = document.createElement('div')
bottomSpacer.style.height = `${remainingItems * itemHeight}px`
container.appendChild(bottomSpacer)
}
}
}
export const performanceOptimizer = new PerformanceOptimizer()Summary: Complete Vue.js Frontend
What We’ve Built
This comprehensive Vue.js frontend provides:
- Modern UI/UX: Clean, responsive interface with Tailwind CSS
- Real-time Chat: Interactive chat with message history and metadata
- Document Processing: Drag-and-drop file upload with OCR analysis
- Voice Interface: Voice interaction with visual feedback
- Knowledge Management: Search and manage personal knowledge base
- System Monitoring: Real-time status and performance metrics
- Type Safety: Proper data validation and error handling
- Production Ready: Docker, nginx, testing, and optimization
Key Features
- Component Architecture: Modular, reusable Vue 3 components
- Reactive State: Composition API for better state management
- API Integration: Seamless communication with FastAPI backend
- Error Handling: Comprehensive error boundaries and user feedback
- Performance: Debouncing, throttling, and virtual scrolling
- Accessibility: ARIA labels and keyboard navigation
- PWA Ready: Service worker and offline capabilities
- Testing: Unit tests and integration testing setup
Deployment Options
- Development:
npm run devfor hot-reload development - Production: Docker container with nginx reverse proxy
- Static Hosting: Build and deploy to CDN or static hosting
- Embedded: Serve from FastAPI backend as static files
This frontend completes our AlexAI ecosystem, providing users with an intuitive, powerful interface to interact with all the AI capabilities we’ve built throughout this series. The modular architecture makes it easy to extend with new features while maintaining performance and user experience.```
// src/main.js
import { createApp } from 'vue'
import App from './App.vue'
import './style.css'
const app = createApp(App)
// Global error handler
app.config.errorHandler = (err, vm, info) => {
console.error('Vue error:', err, info)
}
app.mount('#app')/* src/style.css */
@import 'tailwindcss/base';
@import 'tailwindcss/components';
@import 'tailwindcss/utilities';
@import url('https://fonts.googleapis.com/css2?family=Inter:wght@300;400;500;600;700&display=swap');
/* Custom components */
@layer components {
.btn-primary {
@apply bg-blue-600 text-white px-4 py-2 rounded-lg hover:bg-blue-700 transition-colors;
}
.btn-secondary {
@apply bg-gray-200 text-gray-800 px-4 py-2 rounded-lg hover:bg-gray-300 transition-colors;
}
.input-field {
@apply border border-gray-300 rounded-lg px-3 py-2 focus:ring-2 focus:ring-blue-500 focus:border-transparent;
}
.card {
@apply bg-white rounded-lg border shadow-sm p-4;
}
}
/* Custom animations */
@keyframes float {
0%, 100% { transform: translateY(0px); }
50% { transform: translateY(-10px); }
}
.float-animation {
animation: float 3s ease-in-out infinite;
}
/* Loading states */
.loading-dots::after {
content: '';
animation: dots 2s infinite;
}
@keyframes dots {
0%, 20% { content: ''; }
40% { content: '.'; }
60% { content: '..'; }
80%, 100% { content: '...'; }
}
/* Scrollbar styling */
.custom-scrollbar::-webkit-scrollbar {
width: 6px;
}
.custom-scrollbar::-webkit-scrollbar-track {
background: #f1f1f1;
border-radius: 3px;
}
.custom-scrollbar::-webkit-scrollbar-thumb {
background: #c1c1c1;
border-radius: 3px;
}
.custom-scrollbar::-webkit-scrollbar-thumb:hover {
background: #a8a8a8;
}Additional Components
Knowledge Base Component
<!-- components/Knowledge/KnowledgeBase.vue -->
<template>
<div class="flex flex-col h-full">
<!-- Header -->
<div class="bg-white border-b px-6 py-4">
<h2 class="text-lg font-semibold text-gray-800">Knowledge Base</h2>
<p class="text-sm text-gray-600 mt-1">Search and manage your personal knowledge</p>
</div>
<div class="flex-1 p-6 overflow-y-auto">
<!-- Search Section -->
<div class="mb-6">
<div class="flex space-x-2">
<input
v-model="searchQuery"
@keyup.enter="performSearch"
type="text"
placeholder="Search knowledge base..."
class="flex-1 input-field"
/>
<select v-model="selectedCollection" class="input-field w-40">
<option value="documents">Documents</option>
<option value="conversations">Conversations</option>
<option value="web_searches">Web Searches</option>
<option value="personal_notes">Personal Notes</option>
</select>
<button
@click="performSearch"
:disabled="!searchQuery.trim() || searching"
class="btn-primary"
>
{{ searching ? 'Searching...' : 'Search' }}
</button>
</div>
</div>
<!-- Add Knowledge Section -->
<div class="mb-6 card">
<h3 class="text-sm font-medium text-gray-800 mb-3">Add Knowledge</h3>
<div class="space-y-3">
<textarea
v-model="newKnowledgeContent"
placeholder="Enter content to add to knowledge base..."
class="w-full input-field h-24 resize-none"
></textarea>
<div class="grid grid-cols-2 gap-3">
<input
v-model="newKnowledgeCategory"
type="text"
placeholder="Category (optional)"
class="input-field"
/>
<select v-model="newKnowledgeCollection" class="input-field">
<option value="documents">Documents</option>
<option value="personal_notes">Personal Notes</option>
</select>
</div>
<button
@click="addNewKnowledge"
:disabled="!newKnowledgeContent.trim() || adding"
class="btn-primary w-full"
>
{{ adding ? 'Adding...' : 'Add to Knowledge Base' }}
</button>
</div>
</div>
<!-- Search Results -->
<div v-if="searchResults.length > 0">
<h3 class="text-sm font-medium text-gray-800 mb-3">
Search Results ({{ searchResults.length }} found)
</h3>
<div class="space-y-3">
<div
v-for="(result, index) in searchResults"
:key="index"
class="card hover:shadow-md transition-shadow"
>
<div class="flex justify-between items-start mb-2">
<div class="flex-1">
<div class="flex items-center space-x-2 mb-1">
<span class="text-xs bg-blue-100 text-blue-800 px-2 py-1 rounded">
{{ result.collection }}
</span>
<span class="text-xs bg-green-100 text-green-800 px-2 py-1 rounded">
{{ (result.similarity_score * 100).toFixed(1) }}% match
</span>
<span v-if="result.metadata?.category" class="text-xs bg-gray-100 text-gray-800 px-2 py-1 rounded">
{{ result.metadata.category }}
</span>
</div>
<div class="text-sm text-gray-900 mb-2">
{{ truncateText(result.content, 200) }}
</div>
<div v-if="result.metadata" class="text-xs text-gray-500">
<span v-if="result.metadata.timestamp">
Added: {{ formatDate(result.metadata.timestamp) }}
</span>
<span v-if="result.metadata.content_length" class="ml-3">
{{ result.metadata.content_length }} characters
</span>
</div>
</div>
<div class="flex space-x-1 ml-4">
<button
@click="copyToClipboard(result.content)"
class="text-xs bg-gray-100 text-gray-700 px-2 py-1 rounded hover:bg-gray-200 transition-colors"
title="Copy content"
>
📋
</button>
<button
@click="expandResult(result)"
class="text-xs bg-blue-100 text-blue-700 px-2 py-1 rounded hover:bg-blue-200 transition-colors"
title="View full content"
>
👁️
</button>
</div>
</div>
</div>
</div>
</div>
<!-- No Results -->
<div v-else-if="hasSearched && !searching" class="text-center py-8">
<svg class="mx-auto h-12 w-12 text-gray-400" fill="none" stroke="currentColor" viewBox="0 0 24 24">
<path stroke-linecap="round" stroke-linejoin="round" stroke-width="2" d="M9.172 16.172a4 4 0 015.656 0M9 12h6m-6-4h6m2 5.291A7.962 7.962 0 0112 15c-2.34 0-4.29-1.175-5.5-2.967L3.5 9.5m17 4.5L17.5 9.5C16.29 10.825 14.34 12 12 12c-2.34 0-4.29-1.175-5.5-2.967"/>
</svg>
<p class="mt-2 text-sm text-gray-500">No results found for "{{ lastSearchQuery }}"</p>
<p class="text-xs text-gray-400 mt-1">Try different keywords or check spelling</p>
</div>
<!-- Knowledge Stats -->
<div class="mt-8 grid grid-cols-2 gap-4">
<div class="card text-center">
<div class="text-2xl font-bold text-blue-600">{{ knowledgeStats.total_documents }}</div>
<div class="text-sm text-gray-600">Total Documents</div>
</div>
<div class="card text-center">
<div class="text-2xl font-bold text-green-600">{{ knowledgeStats.total_searches }}</div>
<div class="text-sm text-gray-600">Searches Performed</div>
</div>
</div>
</div>
<!-- Expanded Content Modal -->
<div
v-if="expandedResult"
class="fixed inset-0 bg-black bg-opacity-50 flex items-center justify-center z-50 p-4"
@click="expandedResult = null"
>
<div
class="bg-white rounded-lg max-w-4xl w-full max-h-[80vh] overflow-hidden"
@click.stop
>
<div class="bg-gray-50 px-6 py-4 border-b">
<div class="flex justify-between items-center">
<h3 class="text-lg font-medium text-gray-900">Knowledge Content</h3>
<button
@click="expandedResult = null"
class="text-gray-400 hover:text-gray-600 transition-colors"
>
<svg class="w-6 h-6" fill="currentColor" viewBox="0 0 20 20">
<path fill-rule="evenodd" d="M4.293 4.293a1 1 0 011.414 0L10 8.586l4.293-4.293a1 1 0 111.414 1.414L11.414 10l4.293 4.293a1 1 0 01-1.414 1.414L10 11.414l-4.293 4.293a1 1 0 01-1.414-1.414L8.586 10 4.293 5.707a1 1 0 010-1.414z" clip-rule="evenodd"/>
</svg>
</button>
</div>
</div>
<div class="p-6 overflow-y-auto max-h-[60vh]">
<div class="mb-4">
<div class="flex space-x-2 mb-2">
<span class="text-xs bg-blue-100 text-blue-800 px-2 py-1 rounded">
{{ expandedResult.collection }}
</span>
<span class="text-xs bg-green-100 text-green-800 px-2 py-1 rounded">
{{ (expandedResult.similarity_score * 100).toFixed(1) }}% match
</span>
</div>
</div>
<div class="prose max-w-none">
<pre class="whitespace-pre-wrap text-sm text-gray-900 bg-gray-50 p-4 rounded">{{ expandedResult.content }}</pre>
</div>
</div>
<div class="bg-gray-50 px-6 py-4 border-t flex justify-end space-x-3">
<button
@click="copyToClipboard(expandedResult.content)"
class="btn-secondary"
>
Copy Content
</button>
<button
@click="expandedResult = null"
class="btn-primary"
>
Close
</button>
</div>
</div>
</div>
</div>
</template>
<script>
import { ref, onMounted } from 'vue'
export default {
name: 'KnowledgeBase',
props: {
searchResults: {
type: Array,
default: () => []
}
},
emits: ['search-knowledge', 'add-knowledge'],
setup(props, { emit }) {
const searchQuery = ref('')
const selectedCollection = ref('documents')
const searching = ref(false)
const hasSearched = ref(false)
const lastSearchQuery = ref('')
const newKnowledgeContent = ref('')
const newKnowledgeCategory = ref('')
const newKnowledgeCollection = ref('documents')
const adding = ref(false)
const expandedResult = ref(null)
const knowledgeStats = ref({
total_documents: 0,
total_searches: 0
})
const performSearch = async () => {
if (!searchQuery.value.trim() || searching.value) return
searching.value = true
hasSearched.value = true
lastSearchQuery.value = searchQuery.value
try {
await emit('search-knowledge', searchQuery.value, selectedCollection.value)
knowledgeStats.value.total_searches++
} finally {
searching.value = false
}
}
const addNewKnowledge = async () => {
if (!newKnowledgeContent.value.trim() || adding.value) return
adding.value = true
try {
const metadata = {}
if (newKnowledgeCategory.value.trim()) {
metadata.category = newKnowledgeCategory.value.trim()
}
await emit('add-knowledge', newKnowledgeContent.value.trim(), metadata)
// Clear form
newKnowledgeContent.value = ''
newKnowledgeCategory.value = ''
knowledgeStats.value.total_documents++
} finally {
adding.value = false
}
}
const expandResult = (result) => {
expandedResult.value = result
}
const truncateText = (text, maxLength) => {
if (text.length <= maxLength) return text
return text.substring(0, maxLength) + '...'
}
const formatDate = (dateString) => {
return new Date(dateString).toLocaleDateString()
}
const copyToClipboard = async (text) => {
try {
await navigator.clipboard.writeText(text)
// Could add notification here
} catch (error) {
console.error('Failed to copy:', error)
}
}
onMounted(() => {
// Load initial stats
knowledgeStats.value = {
total_documents: 25, // Could be loaded from API
total_searches: 127
}
})
return {
searchQuery,
selectedCollection,
searching,
hasSearched,
lastSearchQuery,
newKnowledgeContent,
newKnowledgeCategory,
newKnowledgeCollection,
adding,
expandedResult,
knowledgeStats,
performSearch,
addNewKnowledge,
expandResult,
truncateText,
formatDate,
copyToClipboard
}
}
}
</script>System Monitor Component
<!-- components/System/SystemMonitor.vue -->
<template>
<div class="flex flex-col h-full">
<!-- Header -->
<div class="bg-white border-b px-6 py-4">
<div class="flex justify-between items-center">
<div>
<h2 class="text-lg font-semibold text-gray-800">System Monitor</h2>
<p class="text-sm text-gray-600 mt-1">AlexAI system status and performance</p>
</div>
<button
@click="$emit('refresh-status')"
:class="[
'px-4 py-2 rounded-lg transition-colors flex items-center space-x-2',
refreshing ? 'bg-gray-100 text-gray-500' : 'bg-blue-100 text-blue-700 hover:bg-blue-200'
]"
:disabled="refreshing"
>
<svg
:class="['w-4 h-4', refreshing ? 'animate-spin' : '']"
fill="currentColor"
viewBox="0 0 20 20"
>
<path fill-rule="evenodd" d="M4 2a1 1 0 011 1v2.101a7.002 7.002 0 0111.601 2.566 1 1 0 11-1.885.666A5.002 5.002 0 005.999 7H9a1 1 0 010 2H4a1 1 0 01-1-1V3a1 1 0 011-1zm.008 9.057a1 1 0 011.276.61A5.002 5.002 0 0014.001 13H11a1 1 0 110-2h5a1 1 0 011 1v5a1 1 0 11-2 0v-2.101a7.002 7.002 0 01-11.601-2.566 1 1 0 01.61-1.276z" clip-rule="evenodd"/>
</svg>
<span>{{ refreshing ? 'Refreshing...' : 'Refresh' }}</span>
</button>
</div>
</div>
<div class="flex-1 p-6 overflow-y-auto">
<!-- Status Overview -->
<div class="grid grid-cols-1 md:grid-cols-3 gap-6 mb-6">
<!-- Service Status -->
<div class="card">
<div class="flex items-center justify-between mb-2">
<h3 class="text-sm font-medium text-gray-700">Service Status</h3>
<div :class="[
'w-3 h-3 rounded-full',
status.status === 'running' ? 'bg-green-500' :
status.status === 'error' ? 'bg-red-500' :
'bg-yellow-500'
]"></div>
</div>
<div class="text-2xl font-bold text-gray-900 mb-1">
{{ status.status?.toUpperCase() }}
</div>
<div class="text-sm text-gray-600">
{{ status.service }} v{{ status.version }}
</div>
</div>
<!-- Uptime -->
<div class="card">
<h3 class="text-sm font-medium text-gray-700 mb-2">Uptime</h3>
<div class="text-2xl font-bold text-blue-600 mb-1">
{{ formatUptime(status.uptime) }}
</div>
<div class="text-sm text-gray-600">
Since {{ startTime }}
</div>
</div>
<!-- Active Sessions -->
<div class="card">
<h3 class="text-sm font-medium text-gray-700 mb-2">Activity</h3>
<div class="space-y-2">
<div class="flex justify-between">
<span class="text-sm text-gray-600">Sessions:</span>
<span class="text-sm font-medium">{{ status.active_sessions }}</span>
</div>
<div class="flex justify-between">
<span class="text-sm text-gray-600">Requests:</span>
<span class="text-sm font-medium">{{ status.total_requests }}</span>
</div>
</div>
</div>
</div>
<!-- Performance Metrics -->
<div class="grid grid-cols-1 lg:grid-cols-2 gap-6 mb-6">
<!-- Memory Usage -->
<div class="card">
<h3 class="text-sm font-medium text-gray-700 mb-4">Memory Usage</h3>
<div class="space-y-3">
<div v-for="(value, key) in status.memory_usage" :key="key">
<div class="flex justify-between text-sm mb-1">
<span class="text-gray-600">{{ formatMemoryKey(key) }}:</span>
<span class="font-medium">{{ formatMemoryValue(value) }}</span>
</div>
<div class="w-full bg-gray-200 rounded-full h-2">
<div
class="bg-blue-500 h-2 rounded-full transition-all duration-300"
:style="{ width: getMemoryPercentage(value) + '%' }"
></div>
</div>
</div>
</div>
</div>
<!-- System Health -->
<div class="card">
<h3 class="text-sm font-medium text-gray-700 mb-4">System Health</h3>
<div class="space-y-3">
<div class="flex items-center justify-between">
<span class="text-sm text-gray-600">Ollama Connection</span>
<span :class="[
'text-xs px-2 py-1 rounded',
ollamaStatus === 'connected' ? 'bg-green-100 text-green-800' :
'bg-red-100 text-red-800'
]">
{{ ollamaStatus }}
</span>
</div>
<div class="flex items-center justify-between">
<span class="text-sm text-gray-600">Vector Database</span>
<span :class="[
'text-xs px-2 py-1 rounded',
vectorDbStatus === 'connected' ? 'bg-green-100 text-green-800' :
'bg-red-100 text-red-800'
]">
{{ vectorDbStatus }}
</span>
</div>
<div class="flex items-center justify-between">
<span class="text-sm text-gray-600">Memory System</span>
<span :class="[
'text-xs px-2 py-1 rounded',
memoryStatus === 'connected' ? 'bg-green-100 text-green-800' :
'bg-red-100 text-red-800'
]">
{{ memoryStatus }}
</span>
</div>
<div class="flex items-center justify-between">
<span class="text-sm text-gray-600">Voice Interface</span>
<span :class="[
'text-xs px-2 py-1 rounded',
voiceStatus === 'available' ? 'bg-green-100 text-green-800' :
'bg-yellow-100 text-yellow-800'
]">
{{ voiceStatus }}
</span>
</div>
</div>
</div>
</div>
<!-- Recent Activity -->
<div class="card">
<div class="flex justify-between items-center mb-4">
<h3 class="text-sm font-medium text-gray-700">Recent Activity</h3>
<select v-model="activityFilter" class="text-xs border border-gray-300 rounded px-2 py-1">
<option value="all">All</option>
<option value="chat">Chat</option>
<option value="document">Document</option>
<option value="voice">Voice</option>
<option value="system">System</option>
</select>
</div>
<div class="space-y-2 max-h-64 overflow-y-auto custom-scrollbar">
<div
v-for="(log, index) in filteredLogs"
:key="index"
class="flex items-center space-x-3 text-sm p-2 rounded hover:bg-gray-50"
>
<div :class="[
'w-2 h-2 rounded-full flex-shrink-0',
log.level === 'error' ? 'bg-red-500' :
log.level === 'warning' ? 'bg-yellow-500' :
log.level === 'info' ? 'bg-blue-500' :
'bg-gray-500'
]"></div>
<div class="flex-1 min-w-0">
<div class="text-gray-900 truncate">{{ log.message }}</div>
<div class="text-xs text-gray-500">
{{ formatLogTime(log.timestamp) }} • {{ log.component }}
</div>
</div>
<div v-if="log.metadata" class="text-xs text-gray-400">
{{ log.metadata.duration || log.metadata.status }}
</div>
</div>
</div>
<div v-if="filteredLogs.length === 0" class="text-center py-4 text-gray-500 text-sm">
No activity logs available
</div>
</div>
<!-- Quick Actions -->
<div class="mt-6 grid grid-cols-2 md:grid-cols-4 gap-3">
<button
@click="performHealthCheck"
:disabled="performingHealthCheck"
class="btn-secondary text-sm"
>
{{ performingHealthCheck ? 'Checking...' : 'Health Check' }}
</button>
<button
@click="clearLogs"
class="btn-secondary text-sm"
>
Clear Logs
</button>
<button
@click="exportLogs"
class="btn-secondary text-sm"
>
Export Logs
</button>
<button
@click="showSystemInfo = true"
class="btn-secondary text-sm"
>
System Info
</button>
</div>
</div>
<!-- System Info Modal -->
<div
v-if="showSystemInfo"
class="fixed inset-0 bg-black bg-opacity-50 flex items-center justify-center z-50 p-4"
@click="showSystemInfo = false"
>
<div
class="bg-white rounded-lg max-w-2xl w-full max-h-[80vh] overflow-hidden"
@click.stop
>
<div class="bg-gray-50 px-6 py-4 border-b">
<div class="flex justify-between items-center">
<h3 class="text-lg font-medium text-gray-900">System Information</h3>
<button
@click="showSystemInfo = false"
class="text-gray-400 hover:text-gray-600"
>
<svg class="w-6 h-6" fill="currentColor" viewBox="0 0 20 20">
<path fill-rule="evenodd" d="M4.293 4.293a1 1 0 011.414 0L10 8.586l4.293-4.293a1 1 0 111.414 1.414L11.414 10l4.293 4.293a1 1 0 01-1.414 1.414L10 11.414l-4.293 4.293a1 1 0 01-1.414-1.414L8.586 10 4.293 5.707a1 1 0 010-1.414z" clip-rule="evenodd"/>
</svg>
</button>
</div>
</div>
<div class="p-6 overflow-y-auto max-h-[60vh]">
<div class="space-y-4">
<div>
<h4 class="font-medium text-gray-900 mb-2">Environment</h4>
<div class="bg-gray-50 rounded p-3 text-sm font-mono">
<div>Browser: {{ systemInfo.browser }}</div>
<div>Platform: {{ systemInfo.platform }}</div>
<div>User Agent: {{ systemInfo.userAgent }}</div>
</div>
</div>
<div>
<h4 class="font-medium text-gray-900 mb-2">API Configuration</h4>
<div class="bg-gray-50 rounded p-3 text-sm">
<div>Base URL: {{ apiBaseUrl }}</div>
<div>Timeout: {{ apiTimeout }}ms</div>
<div>Version: {{ status.version }}</div>
</div>
</div>
<div>
<h4 class="font-medium text-gray-900 mb-2">Features</h4>
<div class="grid grid-cols-2 gap-2 text-sm">
<div class="flex items-center space-x-2">
<div class="w-2 h-2 bg-green-500 rounded-full"></div>
<span>Chat Interface</span>
</div>
<div class="flex items-center space-x-2">
<div class="w-2 h-2 bg-green-500 rounded-full"></div>
<span>Document Processing</span>
</div>
<div class="flex items-center space-x-2">
<div :class="[
'w-2 h-2 rounded-full',
voiceStatus === 'available' ? 'bg-green-500' : 'bg-yellow-500'
]"></div>
<span>Voice Interface</span>
</div>
<div class="flex items-center space-x-2">
<div class="w-2 h-2 bg-green-500 rounded-full"></div>
<span>Knowledge Base</span>
</div>
</div>
</div>
</div>
</div>
</div>
</div>
</div>
</template>
<script>
import { ref, computed, onMounted } from 'vue'
export default {
name: 'SystemMonitor',
props: {
status: {
type: Object,
required: true
},
logs: {
type: Array,
default: () => []
}
},
emits: ['refresh-status'],
setup(props, { emit }) {
const refreshing = ref(false)
const activityFilter = ref('all')
const performingHealthCheck = ref(false)
const showSystemInfo = ref(false)
// System status states
const ollamaStatus = ref('connected')
const vectorDbStatus = ref('connected')
const memoryStatus = ref('connected')
const voiceStatus = ref('available')
// System info
const systemInfo = ref({
browser: navigator.userAgent.split(' ').slice(-1)[0],
platform: navigator.platform,
userAgent: navigator.userAgent
})
const apiBaseUrl = ref('http://localhost:8000')
const apiTimeout = ref(30000)
const startTime = computed(() => {
const now = new Date()
const uptime = props.status.uptime || 0
const startDate = new Date(now.getTime() - (uptime * 1000))
return startDate.toLocaleString()
})
const filteredLogs = computed(() => {
if (activityFilter.value === 'all') {
return props.logs.slice(-20) // Show last 20 logs
}
return props.logs
.filter(log => log.component === activityFilter.value)
.slice(-20)
})
const formatUptime = (seconds) => {
if (!seconds) return '0s'
const days = Math.floor(seconds / 86400)
const hours = Math.floor((seconds % 86400) / 3600)
const minutes = Math.floor((seconds % 3600) / 60)
const secs = Math.floor(seconds % 60)
if (days > 0) return `${days}d ${hours}h`
if (hours > 0) return `${hours}h ${minutes}m`
if (minutes > 0) return `${minutes}m ${secs}s`
return `${secs}s`
}
const formatMemoryKey = (key) => {
return key.replace(/_/g, ' ').replace(/\b\w/g, l => l.toUpperCase())
}
const formatMemoryValue = (value) => {
if (typeof value === 'number') {
if (value > 1024 * 1024) {
return `${(value / (1024 * 1024)).toFixed(1)} MB`
}
if (value > 1024) {
return `${(value / 1024).toFixed(1)} KB`
}
return `${value} B`
}
return String(value)
}
const getMemoryPercentage = (value) => {
// Simple percentage calculation for demo
if (typeof value === 'number') {
return Math.min((value / (100 * 1024 * 1024)) * 100, 100)
}
return 0
}
const formatLogTime = (timestamp) => {
return new Date(timestamp).toLocaleTimeString()
}
const performHealthCheck = async () => {
performingHealthCheck.value = true
try {
// Simulate health check
await new Promise(resolve => setTimeout(resolve, 2000))
// Update component statuses
ollamaStatus.value = 'connected'
vectorDbStatus.value = 'connected'
memoryStatus.value = 'connected'
voiceStatus.value = navigator.mediaDevices ? 'available' : 'unavailable'
// Emit refresh to get latest status
emit('refresh-status')
} catch (error) {
console.error('Health check failed:', error)
} finally {
performingHealthCheck.value = false
}
}
const clearLogs = () => {
// This would typically emit an event to clear logs
console.log('Clearing logs...')
}
const exportLogs = () => {
const logsText = props.logs
.map(log => `${log.timestamp} [${log.level.toUpperCase()}] ${log.component}: ${log.message}`)
.join('\n')
const blob = new Blob([logsText], { type: 'text/plain' })
const url = URL.createObjectURL(blob)
const a = document.createElement('a')
a.href = url
a.download = `alexai-logs-${new Date().toISOString().split('T')[0]}.txt`
document.body.appendChild(a)
a.click()
document.body.removeChild(a)
URL.revokeObjectURL(url)
}
onMounted(() => {
// Check voice capabilities
if (!navigator.mediaDevices || !navigator.mediaDevices.getUserMedia) {
voiceStatus.value = 'unavailable'
}
})
return {
refreshing,
activityFilter,
performingHealthCheck,
showSystemInfo,
ollamaStatus,
vectorDbStatus,
memoryStatus,
voiceStatus,
systemInfo,
apiBaseUrl,
apiTimeout,
startTime,
filteredLogs,
formatUptime,
formatMemoryKey,
formatMemoryValue,
getMemoryPercentage,
formatLogTime,
performHealthCheck,
clearLogs,
exportLogs
}
}
}
</script>