INTRODUCTION TO AGENTIC AI FOR PERSONALIZED LEARNING
The landscape of artificial intelligence has evolved dramatically from simple question-answering systems to sophisticated multi-agent architectures that can autonomously plan, execute, and coordinate complex tasks. An Agentic AI system represents a paradigm where multiple specialized agents work together, each with distinct responsibilities, to accomplish goals that would be difficult or impossible for a single monolithic system. In this comprehensive tutorial, we will construct a Learning Assistant that leverages this agentic architecture to create personalized learning paths for users based on their interests and current knowledge levels.
The Learning Assistant we will build consists of five primary agents working in concert. The User Interface Agent serves as the orchestrator, managing all interactions with the user and coordinating the other specialized agents. Four research agents work in parallel to gather learning resources: the Book Research Agent searches for relevant books and textbooks, the Tutorial Research Agent finds online tutorials and written guides, the Course Research Agent discovers structured courses on platforms like Coursera and edX, and the Video Research Agent locates educational YouTube videos. Finally, a Report Generation Agent synthesizes all gathered information into a coherent, personalized learning plan.
The architecture we will implement uses LlamaIndex as the primary framework because it provides excellent support for agentic workflows, retrieval-augmented generation, and integration with various language models. LlamaIndex offers native support for both local and remote LLMs, handles multiple GPU architectures including Apple MPS and NVIDIA CUDA, and provides robust tools for web searching and information retrieval. The system will be built entirely with open-source components, ensuring transparency, customizability, and cost-effectiveness. Critically, when using local models through Ollama, the system will leverage GPU acceleration to ensure fast response times and efficient resource utilization.
UNDERSTANDING GPU ACCELERATION FOR LOCAL LANGUAGE MODELS
Before diving into the implementation, it is essential to understand how GPU acceleration works with local language models and why it matters for our Learning Assistant. When running large language models locally through Ollama, the computational workload can be substantial. Without GPU acceleration, inference times can be prohibitively slow, making the interactive experience frustrating for users.
Modern GPUs provide massive parallel processing capabilities that are ideal for the matrix operations underlying neural network inference. NVIDIA GPUs use the CUDA framework, while Apple Silicon Macs use Metal Performance Shaders. Both provide significant speedups compared to CPU-only inference, often reducing response times from minutes to seconds for large models.
Our implementation will automatically detect available GPU hardware and configure the system appropriately. On systems with NVIDIA GPUs, we will verify CUDA availability and configure PyTorch to use CUDA devices. On Apple Silicon, we will detect and utilize the MPS backend. The system will also gracefully fall back to CPU operation when no GPU is available, ensuring compatibility across all hardware configurations.
SETTING UP THE DEVELOPMENT ENVIRONMENT WITH GPU SUPPORT
To build our Learning Assistant with full GPU support, we need to establish a robust development environment with all necessary dependencies including GPU-enabled libraries. The installation process varies slightly depending on your hardware configuration.
For systems with NVIDIA GPUs, you must first install CUDA Toolkit from NVIDIA. Then install PyTorch with CUDA support. For Apple Silicon Macs, PyTorch automatically includes MPS support in recent versions. The core dependencies remain the same across platforms.
Here is the complete setup for our development environment:
pip install torch torchvision torchaudio
pip install llama-index
pip install llama-index-llms-openai
pip install llama-index-llms-ollama
pip install llama-index-embeddings-huggingface
pip install duckduckgo-search
pip install nest-asyncio
pip install python-dotenv
pip install pydantic
pip install psutil
pip install aiohttp
The psutil library is added to monitor system resources including GPU memory usage. The aiohttp library provides robust async HTTP capabilities. For NVIDIA systems, you may also want to install nvidia-ml-py3 for detailed GPU monitoring. The torch installation will automatically detect and use available GPU acceleration.
To verify your GPU setup, you can run a simple test. On NVIDIA systems, torch.cuda.is_available should return True. On Apple Silicon, torch.backends.mps.is_available should return True. Our system will perform these checks automatically during initialization and log detailed information about the detected hardware.
IMPLEMENTING GPU DETECTION AND CONFIGURATION
The foundation of GPU-accelerated inference is proper device detection and configuration. We will create a comprehensive GPU management system that detects available hardware, selects the optimal device, monitors resource usage, provides fallback mechanisms when GPU acceleration is unavailable, and properly cleans up resources on shutdown.
The GPU configuration system must handle three scenarios: NVIDIA CUDA devices, Apple MPS devices, and CPU fallback. Each scenario requires different initialization code and has different performance characteristics. The system must also handle edge cases such as multiple GPUs, insufficient GPU memory, driver issues, and concurrent access patterns.
Let‘s start the implementation of our GPU detection and configuration system with proper resource management:
import torch
import logging
import atexit
from typing import Optional, Dict, Any
from enum import Enum
from contextlib import contextmanager
class DeviceType(str, Enum):
"""Enumeration of supported compute devices."""
CUDA = "cuda"
MPS = "mps"
CPU = "cpu"
class GPUConfig:
"""
Manages GPU detection, configuration, and monitoring for the Learning Assistant.
Supports NVIDIA CUDA, Apple MPS, and CPU fallback with proper resource cleanup.
"""
def __init__(self):
"""Initialize GPU configuration with automatic device detection."""
self.logger = logging.getLogger(self.__class__.__name__)
self.device_type = self._detect_device()
self.device = self._initialize_device()
self._log_device_info()
atexit.register(self.cleanup)
def _detect_device(self) -> DeviceType:
"""
Detects the best available compute device.
Priority: CUDA > MPS > CPU
Returns:
DeviceType enum indicating the detected device type
"""
try:
if torch.cuda.is_available():
self.logger.info("CUDA GPU detected")
return DeviceType.CUDA
elif hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
try:
test_tensor = torch.zeros(1, device='mps')
del test_tensor
self.logger.info("Apple MPS GPU detected and verified")
return DeviceType.MPS
except Exception as e:
self.logger.warning(f"MPS detected but not functional: {str(e)}")
return DeviceType.CPU
else:
self.logger.warning("No GPU detected, falling back to CPU")
return DeviceType.CPU
except Exception as e:
self.logger.error(f"Error during device detection: {str(e)}")
return DeviceType.CPU
def _initialize_device(self) -> torch.device:
"""
Initializes the torch device based on detected hardware.
Returns:
Configured torch.device object
"""
try:
if self.device_type == DeviceType.CUDA:
device_id = self._select_cuda_device()
return torch.device(f"cuda:{device_id}")
elif self.device_type == DeviceType.MPS:
return torch.device("mps")
else:
return torch.device("cpu")
except Exception as e:
self.logger.error(f"Error initializing device: {str(e)}")
return torch.device("cpu")
def _select_cuda_device(self) -> int:
"""
Selects the best CUDA device when multiple GPUs are available.
Chooses the GPU with the most free memory.
Returns:
Device ID of the selected GPU
"""
if not torch.cuda.is_available():
return 0
num_devices = torch.cuda.device_count()
if num_devices == 1:
return 0
max_free_memory = 0
best_device = 0
for device_id in range(num_devices):
try:
props = torch.cuda.get_device_properties(device_id)
total_memory = props.total_memory
allocated_memory = torch.cuda.memory_allocated(device_id)
free_memory = total_memory - allocated_memory
if free_memory > max_free_memory:
max_free_memory = free_memory
best_device = device_id
except Exception as e:
self.logger.warning(f"Error checking device {device_id}: {str(e)}")
self.logger.info(f"Selected CUDA device {best_device} with {max_free_memory / 1e9:.2f} GB free memory")
return best_device
def _log_device_info(self):
"""Logs detailed information about the selected device."""
try:
if self.device_type == DeviceType.CUDA:
device_id = self.device.index if self.device.index is not None else 0
props = torch.cuda.get_device_properties(device_id)
self.logger.info(f"Using CUDA device: {props.name}")
self.logger.info(f"Total memory: {props.total_memory / 1e9:.2f} GB")
self.logger.info(f"Compute capability: {props.major}.{props.minor}")
elif self.device_type == DeviceType.MPS:
self.logger.info("Using Apple Metal Performance Shaders (MPS)")
self.logger.info("MPS provides GPU acceleration on Apple Silicon")
else:
self.logger.info("Using CPU (no GPU acceleration)")
self.logger.info("For better performance, consider using a GPU-enabled system")
except Exception as e:
self.logger.error(f"Error logging device info: {str(e)}")
def get_device_info(self) -> Dict[str, Any]:
"""
Returns detailed information about the current device.
Returns:
Dictionary containing device information
"""
info = {
"device_type": self.device_type.value,
"device_str": str(self.device)
}
try:
if self.device_type == DeviceType.CUDA:
device_id = self.device.index if self.device.index is not None else 0
props = torch.cuda.get_device_properties(device_id)
info.update({
"name": props.name,
"total_memory_gb": props.total_memory / 1e9,
"allocated_memory_gb": torch.cuda.memory_allocated(device_id) / 1e9,
"cached_memory_gb": torch.cuda.memory_reserved(device_id) / 1e9,
"compute_capability": f"{props.major}.{props.minor}"
})
except Exception as e:
self.logger.error(f"Error getting device info: {str(e)}")
return info
def clear_cache(self):
"""Clears GPU cache to free memory."""
try:
if self.device_type == DeviceType.CUDA:
torch.cuda.empty_cache()
self.logger.info("Cleared CUDA cache")
elif self.device_type == DeviceType.MPS:
if hasattr(torch.mps, 'empty_cache'):
torch.mps.empty_cache()
self.logger.info("Cleared MPS cache")
except Exception as e:
self.logger.error(f"Error clearing cache: {str(e)}")
def get_memory_usage(self) -> Dict[str, float]:
"""
Returns current GPU memory usage in GB.
Returns:
Dictionary with memory usage statistics
"""
try:
if self.device_type == DeviceType.CUDA:
device_id = self.device.index if self.device.index is not None else 0
return {
"allocated_gb": torch.cuda.memory_allocated(device_id) / 1e9,
"reserved_gb": torch.cuda.memory_reserved(device_id) / 1e9,
"total_gb": torch.cuda.get_device_properties(device_id).total_memory / 1e9
}
except Exception as e:
self.logger.error(f"Error getting memory usage: {str(e)}")
return {"allocated_gb": 0, "reserved_gb": 0, "total_gb": 0}
def cleanup(self):
"""Cleanup GPU resources on shutdown."""
try:
self.logger.info("Cleaning up GPU resources")
self.clear_cache()
if self.device_type == DeviceType.CUDA:
torch.cuda.synchronize()
except Exception as e:
self.logger.error(f"Error during cleanup: {str(e)}")
@contextmanager
def managed_memory(self):
"""
Context manager for automatic memory cleanup.
Usage:
with gpu_config.managed_memory():
# GPU operations here
pass
"""
try:
yield
finally:
self.clear_cache()
This GPU configuration system provides comprehensive device management with automatic detection, optimal device selection, detailed logging, memory monitoring, and proper resource cleanup.
DESIGNING THE AGENT COMMUNICATION PROTOCOL
Effective communication between agents is crucial for system reliability and maintainability. We will design a structured communication protocol using Pydantic models to ensure type safety and validation. Each message passed between agents will be a well-defined data structure with clear semantics and proper input sanitization.
Here is the foundational code for our communication protocol with enhanced validation:
from pydantic import BaseModel, Field, validator
from typing import Optional, List, Dict, Any
from datetime import datetime
from enum import Enum
import re
class KnowledgeLevel(str, Enum):
"""Enumeration of possible knowledge levels for learners."""
BEGINNER = "beginner"
INTERMEDIATE = "intermediate"
ADVANCED = "advanced"
EXPERT = "expert"
class ResourceType(str, Enum):
"""Enumeration of resource types that can be searched."""
BOOK = "book"
TUTORIAL = "tutorial"
COURSE = "course"
VIDEO = "video"
class AgentMessage(BaseModel):
"""Base message structure for inter-agent communication."""
sender: str
recipient: str
timestamp: datetime = Field(default_factory=datetime.now)
message_type: str
correlation_id: str
@validator('sender', 'recipient')
def validate_not_empty(cls, v):
"""Validate that sender and recipient are not empty."""
if not v or not v.strip():
raise ValueError('Sender and recipient must not be empty')
return v.strip()
@validator('correlation_id')
def validate_correlation_id(cls, v):
"""Validate correlation ID format."""
if not v or not v.strip():
raise ValueError('Correlation ID must not be empty')
if not re.match(r'^[a-f0-9\-]{36}$', v):
raise ValueError('Correlation ID must be a valid UUID')
return v
class ResearchRequest(AgentMessage):
"""Request message for research agents."""
topic: str
knowledge_level: KnowledgeLevel
resource_type: ResourceType
specific_requirements: Optional[str] = None
max_results: int = 5
@validator('topic')
def validate_topic(cls, v):
"""Validate and sanitize topic."""
if not v or not v.strip():
raise ValueError('Topic must not be empty')
sanitized = v.strip()
if len(sanitized) > 200:
raise ValueError('Topic must be 200 characters or less')
if re.search(r'[<>{}]', sanitized):
raise ValueError('Topic contains invalid characters')
return sanitized
@validator('max_results')
def validate_max_results(cls, v):
"""Validate max_results range."""
if v < 1 or v > 20:
raise ValueError('max_results must be between 1 and 20')
return v
@validator('specific_requirements')
def validate_requirements(cls, v):
"""Validate specific requirements."""
if v is not None:
sanitized = v.strip()
if len(sanitized) > 500:
raise ValueError('Specific requirements must be 500 characters or less')
return sanitized
return v
class ResourceItem(BaseModel):
"""Represents a single learning resource."""
title: str
url: str
description: str
difficulty_level: KnowledgeLevel
estimated_time: Optional[str] = None
rating: Optional[float] = None
source: str
@validator('url')
def validate_url(cls, v):
"""Validate URL format and safety."""
if not v or not v.strip():
raise ValueError('URL must not be empty')
url = v.strip()
if not url.startswith(('http://', 'https://')):
raise ValueError('URL must be a valid HTTP/HTTPS URL')
if len(url) > 2048:
raise ValueError('URL is too long')
return url
@validator('title')
def validate_title(cls, v):
"""Validate title."""
if not v or not v.strip():
raise ValueError('Title must not be empty')
title = v.strip()
if len(title) > 500:
return title[:500]
return title
@validator('description')
def validate_description(cls, v):
"""Validate description."""
if not v:
return "No description available"
desc = v.strip()
if len(desc) > 1000:
return desc[:1000] + "..."
return desc
class ResearchResponse(AgentMessage):
"""Response message from research agents."""
resources: List[ResourceItem]
search_query_used: str
total_results_found: int
errors: List[str] = Field(default_factory=list)
@validator('total_results_found')
def validate_total_results(cls, v):
"""Validate total results count."""
if v < 0:
raise ValueError('Total results cannot be negative')
return v
This enhanced protocol includes comprehensive validation, input sanitization, and protection against common security issues.
IMPLEMENTING THE WEB SEARCH CAPABILITY WITH RATE LIMITING
The foundation of our research agents is the ability to search the web for relevant learning resources. We will create a robust search utility class with rate limiting, proper timeout handling, and comprehensive error recovery.
Let’s view the implementation:
from duckduckgo_search import DDGS
from typing import List, Set
import time
import asyncio
from datetime import datetime, timedelta
class RateLimiter:
"""Simple rate limiter for API calls."""
def __init__(self, max_calls: int = 10, time_window: int = 60):
"""
Initialize rate limiter.
Args:
max_calls: Maximum number of calls allowed in time window
time_window: Time window in seconds
"""
self.max_calls = max_calls
self.time_window = time_window
self.calls = []
self.logger = logging.getLogger(self.__class__.__name__)
async def acquire(self):
"""Wait if necessary to respect rate limits."""
now = datetime.now()
self.calls = [call_time for call_time in self.calls
if now - call_time < timedelta(seconds=self.time_window)]
if len(self.calls) >= self.max_calls:
sleep_time = (self.calls[0] + timedelta(seconds=self.time_window) - now).total_seconds()
if sleep_time > 0:
self.logger.info(f"Rate limit reached, waiting {sleep_time:.2f} seconds")
await asyncio.sleep(sleep_time)
return await self.acquire()
self.calls.append(now)
class SearchUtility:
"""
Utility class for web searching using DuckDuckGo with rate limiting
and comprehensive error handling.
"""
def __init__(self, max_retries: int = 3):
"""
Initialize search utility.
Args:
max_retries: Maximum number of retry attempts for failed searches
"""
self.ddgs = DDGS()
self.max_retries = max_retries
self.rate_limiter = RateLimiter(max_calls=10, time_window=60)
self.logger = logging.getLogger(self.__class__.__name__)
def construct_search_query(self, topic: str, resource_type: ResourceType,
knowledge_level: KnowledgeLevel) -> str:
"""
Constructs an optimized search query based on the learning parameters.
Args:
topic: The learning topic to search for
resource_type: Type of resource to find
knowledge_level: User's current knowledge level
Returns:
Optimized search query string
"""
level_terms = {
KnowledgeLevel.BEGINNER: "beginner introduction basics fundamentals",
KnowledgeLevel.INTERMEDIATE: "intermediate practical applied",
KnowledgeLevel.ADVANCED: "advanced in-depth comprehensive",
KnowledgeLevel.EXPERT: "expert master advanced professional"
}
resource_terms = {
ResourceType.BOOK: "book textbook guide",
ResourceType.TUTORIAL: "tutorial how-to guide walkthrough",
ResourceType.COURSE: "course online course training program",
ResourceType.VIDEO: "video lecture youtube tutorial"
}
query_parts = [topic.strip()]
query_parts.append(resource_terms[resource_type])
query_parts.append(level_terms[knowledge_level].split()[0])
return " ".join(query_parts)
def infer_difficulty_level(self, text: str) -> KnowledgeLevel:
"""
Attempts to infer the difficulty level of a resource based on keywords.
Args:
text: Combined title and description text
Returns:
Inferred knowledge level
"""
if not text:
return KnowledgeLevel.INTERMEDIATE
text_lower = text.lower()
beginner_keywords = ["beginner", "introduction", "basics", "fundamentals",
"getting started", "101", "for beginners", "start here"]
advanced_keywords = ["advanced", "expert", "master", "professional",
"in-depth", "comprehensive", "deep dive"]
intermediate_keywords = ["intermediate", "practical", "applied",
"hands-on", "real-world", "beyond basics"]
beginner_score = sum(1 for kw in beginner_keywords if kw in text_lower)
advanced_score = sum(1 for kw in advanced_keywords if kw in text_lower)
intermediate_score = sum(1 for kw in intermediate_keywords if kw in text_lower)
if beginner_score > advanced_score and beginner_score > intermediate_score:
return KnowledgeLevel.BEGINNER
elif advanced_score > beginner_score and advanced_score > intermediate_score:
return KnowledgeLevel.ADVANCED
elif intermediate_score > 0:
return KnowledgeLevel.INTERMEDIATE
else:
return KnowledgeLevel.INTERMEDIATE
def deduplicate_resources(self, resources: List[ResourceItem]) -> List[ResourceItem]:
"""
Removes duplicate resources based on URL.
Args:
resources: List of resource items
Returns:
Deduplicated list of resources
"""
seen_urls: Set[str] = set()
deduplicated = []
for resource in resources:
try:
normalized_url = resource.url.lower().rstrip('/')
if normalized_url not in seen_urls:
seen_urls.add(normalized_url)
deduplicated.append(resource)
except Exception as e:
self.logger.warning(f"Error deduplicating resource: {str(e)}")
return deduplicated
async def search_resources(self, request: ResearchRequest) -> List[ResourceItem]:
"""
Executes a web search with retry logic and rate limiting.
Args:
request: Research request containing search parameters
Returns:
List of structured resource items
"""
query = self.construct_search_query(
request.topic,
request.resource_type,
request.knowledge_level
)
for attempt in range(self.max_retries):
try:
await self.rate_limiter.acquire()
loop = asyncio.get_event_loop()
results = await asyncio.wait_for(
loop.run_in_executor(None, lambda: self.ddgs.text(query, max_results=request.max_results * 3)),
timeout=30.0
)
resources = []
if not results:
self.logger.warning(f"No results found for query: {query}")
return []
for result in results:
try:
title = result.get('title', '').strip()
url = result.get('href', '').strip()
description = result.get('body', '').strip()
if not title or not url:
continue
if not url.startswith(('http://', 'https://')):
continue
inferred_level = self.infer_difficulty_level(title + " " + description)
resource = ResourceItem(
title=title,
url=url,
description=description if description else "No description available",
difficulty_level=inferred_level,
source="DuckDuckGo Search"
)
resources.append(resource)
if len(resources) >= request.max_results * 2:
break
except Exception as e:
self.logger.warning(f"Error processing search result: {str(e)}")
continue
deduplicated = self.deduplicate_resources(resources)
return deduplicated[:request.max_results]
except asyncio.TimeoutError:
self.logger.error(f"Search timeout on attempt {attempt + 1}")
if attempt < self.max_retries - 1:
wait_time = 2 ** attempt
await asyncio.sleep(wait_time)
else:
self.logger.error("All search attempts timed out")
return []
except Exception as e:
self.logger.error(f"Search attempt {attempt + 1} failed: {str(e)}")
if attempt < self.max_retries - 1:
wait_time = 2 ** attempt
await asyncio.sleep(wait_time)
else:
self.logger.error("All search attempts failed")
return []
return []
This enhanced search utility includes rate limiting, proper async timeout handling, and comprehensive error recovery.
FULL RUNNING EXAMPLE WITH ALL CORRECTIONS
Now we have completed the full, production-ready implementation:
# learning_assistant.py
# Complete GPU-accelerated Agentic AI Learning Assistant
import os
import sys
import uuid
import asyncio
import logging
import time
import signal
import torch
import re
import atexit
from datetime import datetime, timedelta
from typing import Optional, List, Dict, Any, Tuple, Set
from enum import Enum
from contextlib import contextmanager
from pydantic import BaseModel, Field, validator
from dotenv import load_dotenv
from llama_index.core.llms import ChatMessage, MessageRole
from llama_index.llms.openai import OpenAI
from llama_index.llms.ollama import Ollama
from duckduckgo_search import DDGS
import nest_asyncio
nest_asyncio.apply()
# ============================================================================
# SIGNAL HANDLING FOR GRACEFUL SHUTDOWN
# ============================================================================
class GracefulShutdown:
"""Handles graceful shutdown on SIGINT and SIGTERM."""
def __init__(self):
self.shutdown_requested = False
signal.signal(signal.SIGINT, self._signal_handler)
signal.signal(signal.SIGTERM, self._signal_handler)
def _signal_handler(self, signum, frame):
"""Handle shutdown signals."""
self.shutdown_requested = True
logging.info(f"Shutdown signal {signum} received")
def should_shutdown(self) -> bool:
"""Check if shutdown was requested."""
return self.shutdown_requested
# ============================================================================
# GPU CONFIGURATION AND MANAGEMENT
# ============================================================================
class DeviceType(str, Enum):
"""Enumeration of supported compute devices."""
CUDA = "cuda"
MPS = "mps"
CPU = "cpu"
class GPUConfig:
"""
Manages GPU detection, configuration, and monitoring.
Supports NVIDIA CUDA, Apple MPS, and CPU fallback.
"""
def __init__(self):
"""Initialize GPU configuration with automatic device detection."""
self.logger = logging.getLogger(self.__class__.__name__)
self.device_type = self._detect_device()
self.device = self._initialize_device()
self._log_device_info()
atexit.register(self.cleanup)
def _detect_device(self) -> DeviceType:
"""
Detects the best available compute device.
Returns:
DeviceType enum indicating the detected device type
"""
try:
if torch.cuda.is_available():
self.logger.info("CUDA GPU detected")
return DeviceType.CUDA
elif hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
try:
test_tensor = torch.zeros(1, device='mps')
del test_tensor
self.logger.info("Apple MPS GPU detected and verified")
return DeviceType.MPS
except Exception as e:
self.logger.warning(f"MPS detected but not functional: {str(e)}")
return DeviceType.CPU
else:
self.logger.warning("No GPU detected, falling back to CPU")
return DeviceType.CPU
except Exception as e:
self.logger.error(f"Error during device detection: {str(e)}")
return DeviceType.CPU
def _initialize_device(self) -> torch.device:
"""
Initializes the torch device based on detected hardware.
Returns:
Configured torch.device object
"""
try:
if self.device_type == DeviceType.CUDA:
device_id = self._select_cuda_device()
return torch.device(f"cuda:{device_id}")
elif self.device_type == DeviceType.MPS:
return torch.device("mps")
else:
return torch.device("cpu")
except Exception as e:
self.logger.error(f"Error initializing device: {str(e)}")
return torch.device("cpu")
def _select_cuda_device(self) -> int:
"""
Selects the best CUDA device when multiple GPUs are available.
Returns:
Device ID of the selected GPU
"""
if not torch.cuda.is_available():
return 0
num_devices = torch.cuda.device_count()
if num_devices == 1:
return 0
max_free_memory = 0
best_device = 0
for device_id in range(num_devices):
try:
props = torch.cuda.get_device_properties(device_id)
total_memory = props.total_memory
allocated_memory = torch.cuda.memory_allocated(device_id)
free_memory = total_memory - allocated_memory
if free_memory > max_free_memory:
max_free_memory = free_memory
best_device = device_id
except Exception as e:
self.logger.warning(f"Error checking device {device_id}: {str(e)}")
self.logger.info(f"Selected CUDA device {best_device} with {max_free_memory / 1e9:.2f} GB free")
return best_device
def _log_device_info(self):
"""Logs detailed information about the selected device."""
try:
if self.device_type == DeviceType.CUDA:
device_id = self.device.index if self.device.index is not None else 0
props = torch.cuda.get_device_properties(device_id)
self.logger.info(f"Using CUDA device: {props.name}")
self.logger.info(f"Total memory: {props.total_memory / 1e9:.2f} GB")
self.logger.info(f"Compute capability: {props.major}.{props.minor}")
elif self.device_type == DeviceType.MPS:
self.logger.info("Using Apple Metal Performance Shaders (MPS)")
else:
self.logger.info("Using CPU (no GPU acceleration)")
except Exception as e:
self.logger.error(f"Error logging device info: {str(e)}")
def get_device_info(self) -> Dict[str, Any]:
"""
Returns detailed information about the current device.
Returns:
Dictionary containing device information
"""
info = {
"device_type": self.device_type.value,
"device_str": str(self.device)
}
try:
if self.device_type == DeviceType.CUDA:
device_id = self.device.index if self.device.index is not None else 0
props = torch.cuda.get_device_properties(device_id)
info.update({
"name": props.name,
"total_memory_gb": props.total_memory / 1e9,
"allocated_memory_gb": torch.cuda.memory_allocated(device_id) / 1e9,
"cached_memory_gb": torch.cuda.memory_reserved(device_id) / 1e9,
"compute_capability": f"{props.major}.{props.minor}"
})
except Exception as e:
self.logger.error(f"Error getting device info: {str(e)}")
return info
def clear_cache(self):
"""Clears GPU cache to free memory."""
try:
if self.device_type == DeviceType.CUDA:
torch.cuda.empty_cache()
self.logger.debug("Cleared CUDA cache")
elif self.device_type == DeviceType.MPS:
if hasattr(torch.mps, 'empty_cache'):
torch.mps.empty_cache()
self.logger.debug("Cleared MPS cache")
except Exception as e:
self.logger.error(f"Error clearing cache: {str(e)}")
def get_memory_usage(self) -> Dict[str, float]:
"""
Returns current GPU memory usage in GB.
Returns:
Dictionary with memory usage statistics
"""
try:
if self.device_type == DeviceType.CUDA:
device_id = self.device.index if self.device.index is not None else 0
return {
"allocated_gb": torch.cuda.memory_allocated(device_id) / 1e9,
"reserved_gb": torch.cuda.memory_reserved(device_id) / 1e9,
"total_gb": torch.cuda.get_device_properties(device_id).total_memory / 1e9
}
except Exception as e:
self.logger.error(f"Error getting memory usage: {str(e)}")
return {"allocated_gb": 0, "reserved_gb": 0, "total_gb": 0}
def cleanup(self):
"""Cleanup GPU resources on shutdown."""
try:
self.logger.info("Cleaning up GPU resources")
self.clear_cache()
if self.device_type == DeviceType.CUDA:
torch.cuda.synchronize()
except Exception as e:
self.logger.error(f"Error during cleanup: {str(e)}")
@contextmanager
def managed_memory(self):
"""Context manager for automatic memory cleanup."""
try:
yield
finally:
self.clear_cache()
# ============================================================================
# DATA MODELS AND ENUMERATIONS
# ============================================================================
class KnowledgeLevel(str, Enum):
"""Enumeration of possible knowledge levels."""
BEGINNER = "beginner"
INTERMEDIATE = "intermediate"
ADVANCED = "advanced"
EXPERT = "expert"
class ResourceType(str, Enum):
"""Enumeration of resource types."""
BOOK = "book"
TUTORIAL = "tutorial"
COURSE = "course"
VIDEO = "video"
class ConversationState(str, Enum):
"""States in the conversation flow."""
INITIAL = "initial"
GATHERING_INFO = "gathering_info"
RESEARCHING = "researching"
GENERATING_REPORT = "generating_report"
COMPLETE = "complete"
ERROR = "error"
class AgentMessage(BaseModel):
"""Base message structure for inter-agent communication."""
sender: str
recipient: str
timestamp: datetime = Field(default_factory=datetime.now)
message_type: str
correlation_id: str
@validator('sender', 'recipient')
def validate_not_empty(cls, v):
if not v or not v.strip():
raise ValueError('Sender and recipient must not be empty')
return v.strip()
@validator('correlation_id')
def validate_correlation_id(cls, v):
if not v or not v.strip():
raise ValueError('Correlation ID must not be empty')
if not re.match(r'^[a-f0-9\-]{36}$', v):
raise ValueError('Correlation ID must be a valid UUID')
return v
class ResearchRequest(AgentMessage):
"""Request message for research agents."""
topic: str
knowledge_level: KnowledgeLevel
resource_type: ResourceType
specific_requirements: Optional[str] = None
max_results: int = 5
@validator('topic')
def validate_topic(cls, v):
if not v or not v.strip():
raise ValueError('Topic must not be empty')
sanitized = v.strip()
if len(sanitized) > 200:
raise ValueError('Topic must be 200 characters or less')
if re.search(r'[<>{}]', sanitized):
raise ValueError('Topic contains invalid characters')
return sanitized
@validator('max_results')
def validate_max_results(cls, v):
if v < 1 or v > 20:
raise ValueError('max_results must be between 1 and 20')
return v
@validator('specific_requirements')
def validate_requirements(cls, v):
if v is not None:
sanitized = v.strip()
if len(sanitized) > 500:
raise ValueError('Specific requirements must be 500 characters or less')
return sanitized
return v
class ResourceItem(BaseModel):
"""Represents a single learning resource."""
title: str
url: str
description: str
difficulty_level: KnowledgeLevel
estimated_time: Optional[str] = None
rating: Optional[float] = None
source: str
@validator('url')
def validate_url(cls, v):
if not v or not v.strip():
raise ValueError('URL must not be empty')
url = v.strip()
if not url.startswith(('http://', 'https://')):
raise ValueError('URL must be a valid HTTP/HTTPS URL')
if len(url) > 2048:
raise ValueError('URL is too long')
return url
@validator('title')
def validate_title(cls, v):
if not v or not v.strip():
raise ValueError('Title must not be empty')
title = v.strip()
if len(title) > 500:
return title[:500]
return title
@validator('description')
def validate_description(cls, v):
if not v:
return "No description available"
desc = v.strip()
if len(desc) > 1000:
return desc[:1000] + "..."
return desc
class ResearchResponse(AgentMessage):
"""Response message from research agents."""
resources: List[ResourceItem]
search_query_used: str
total_results_found: int
errors: List[str] = Field(default_factory=list)
@validator('total_results_found')
def validate_total_results(cls, v):
if v < 0:
raise ValueError('Total results cannot be negative')
return v
class LearningPlan(BaseModel):
"""Complete learning plan structure."""
topic: str
knowledge_level: KnowledgeLevel
introduction: str
resource_sections: List[Dict[str, Any]]
estimated_total_time: str
learning_tips: List[str]
next_steps: str
# ============================================================================
# RATE LIMITER
# ============================================================================
class RateLimiter:
"""Simple rate limiter for API calls."""
def __init__(self, max_calls: int = 10, time_window: int = 60):
self.max_calls = max_calls
self.time_window = time_window
self.calls = []
self.logger = logging.getLogger(self.__class__.__name__)
async def acquire(self):
"""Wait if necessary to respect rate limits."""
now = datetime.now()
self.calls = [call_time for call_time in self.calls
if now - call_time < timedelta(seconds=self.time_window)]
if len(self.calls) >= self.max_calls:
sleep_time = (self.calls[0] + timedelta(seconds=self.time_window) - now).total_seconds()
if sleep_time > 0:
self.logger.info(f"Rate limit reached, waiting {sleep_time:.2f}s")
await asyncio.sleep(sleep_time)
return await self.acquire()
self.calls.append(now)
# ============================================================================
# SEARCH UTILITY
# ============================================================================
class SearchUtility:
"""Web searching utility with rate limiting and error handling."""
def __init__(self, max_retries: int = 3):
self.ddgs = DDGS()
self.max_retries = max_retries
self.rate_limiter = RateLimiter(max_calls=10, time_window=60)
self.logger = logging.getLogger(self.__class__.__name__)
def construct_search_query(self, topic: str, resource_type: ResourceType,
knowledge_level: KnowledgeLevel) -> str:
"""Constructs an optimized search query."""
level_terms = {
KnowledgeLevel.BEGINNER: "beginner introduction basics fundamentals",
KnowledgeLevel.INTERMEDIATE: "intermediate practical applied",
KnowledgeLevel.ADVANCED: "advanced in-depth comprehensive",
KnowledgeLevel.EXPERT: "expert master advanced professional"
}
resource_terms = {
ResourceType.BOOK: "book textbook guide",
ResourceType.TUTORIAL: "tutorial how-to guide walkthrough",
ResourceType.COURSE: "course online course training program",
ResourceType.VIDEO: "video lecture youtube tutorial"
}
query_parts = [topic.strip()]
query_parts.append(resource_terms[resource_type])
query_parts.append(level_terms[knowledge_level].split()[0])
return " ".join(query_parts)
def infer_difficulty_level(self, text: str) -> KnowledgeLevel:
"""Infers difficulty level from text."""
if not text:
return KnowledgeLevel.INTERMEDIATE
text_lower = text.lower()
beginner_keywords = ["beginner", "introduction", "basics", "fundamentals",
"getting started", "101", "for beginners"]
advanced_keywords = ["advanced", "expert", "master", "professional",
"in-depth", "comprehensive"]
intermediate_keywords = ["intermediate", "practical", "applied",
"hands-on", "real-world"]
beginner_score = sum(1 for kw in beginner_keywords if kw in text_lower)
advanced_score = sum(1 for kw in advanced_keywords if kw in text_lower)
intermediate_score = sum(1 for kw in intermediate_keywords if kw in text_lower)
if beginner_score > max(advanced_score, intermediate_score):
return KnowledgeLevel.BEGINNER
elif advanced_score > max(beginner_score, intermediate_score):
return KnowledgeLevel.ADVANCED
elif intermediate_score > 0:
return KnowledgeLevel.INTERMEDIATE
else:
return KnowledgeLevel.INTERMEDIATE
def deduplicate_resources(self, resources: List[ResourceItem]) -> List[ResourceItem]:
"""Removes duplicate resources."""
seen_urls: Set[str] = set()
deduplicated = []
for resource in resources:
try:
normalized_url = resource.url.lower().rstrip('/')
if normalized_url not in seen_urls:
seen_urls.add(normalized_url)
deduplicated.append(resource)
except Exception as e:
self.logger.warning(f"Error deduplicating: {str(e)}")
return deduplicated
async def search_resources(self, request: ResearchRequest) -> List[ResourceItem]:
"""Executes web search with retry logic and rate limiting."""
query = self.construct_search_query(
request.topic,
request.resource_type,
request.knowledge_level
)
for attempt in range(self.max_retries):
try:
await self.rate_limiter.acquire()
loop = asyncio.get_event_loop()
results = await asyncio.wait_for(
loop.run_in_executor(None, lambda: self.ddgs.text(query, max_results=request.max_results * 3)),
timeout=30.0
)
resources = []
if not results:
self.logger.warning(f"No results for query: {query}")
return []
for result in results:
try:
title = result.get('title', '').strip()
url = result.get('href', '').strip()
description = result.get('body', '').strip()
if not title or not url:
continue
if not url.startswith(('http://', 'https://')):
continue
inferred_level = self.infer_difficulty_level(title + " " + description)
resource = ResourceItem(
title=title,
url=url,
description=description if description else "No description available",
difficulty_level=inferred_level,
source="DuckDuckGo Search"
)
resources.append(resource)
if len(resources) >= request.max_results * 2:
break
except Exception as e:
self.logger.warning(f"Error processing result: {str(e)}")
continue
deduplicated = self.deduplicate_resources(resources)
return deduplicated[:request.max_results]
except asyncio.TimeoutError:
self.logger.error(f"Search timeout on attempt {attempt + 1}")
if attempt < self.max_retries - 1:
await asyncio.sleep(2 ** attempt)
else:
return []
except Exception as e:
self.logger.error(f"Search failed attempt {attempt + 1}: {str(e)}")
if attempt < self.max_retries - 1:
await asyncio.sleep(2 ** attempt)
else:
return []
return []
# ============================================================================
# RESEARCH AGENTS
# ============================================================================
class BookResearchAgent:
"""Specialized agent for finding books."""
def __init__(self):
self.name = "BookResearchAgent"
self.search_utility = SearchUtility()
self.logger = logging.getLogger(self.__class__.__name__)
def enhance_book_search(self, request: ResearchRequest) -> ResearchRequest:
"""Enhances search request with book-specific parameters."""
enhanced_request = request.model_copy()
if request.specific_requirements:
enhanced_request.specific_requirements += " book textbook"
else:
enhanced_request.specific_requirements = "book textbook"
return enhanced_request
def filter_book_results(self, resources: List[ResourceItem]) -> List[ResourceItem]:
"""Filters results to prioritize books."""
if not resources:
return []
book_indicators = ["book", "textbook", "isbn", "publisher", "edition",
"author", "pdf", "ebook"]
scored_resources = []
for resource in resources:
text = (resource.title + " " + resource.description).lower()
score = sum(1 for indicator in book_indicators if indicator in text)
scored_resources.append((score, resource))
scored_resources.sort(key=lambda x: x[0], reverse=True)
return [resource for score, resource in scored_resources]
async def research(self, request: ResearchRequest) -> ResearchResponse:
"""Executes book research."""
errors = []
try:
enhanced_request = self.enhance_book_search(request)
resources = await self.search_utility.search_resources(enhanced_request)
if not resources:
errors.append("No book resources found")
filtered_resources = self.filter_book_results(resources)
return ResearchResponse(
sender=self.name,
recipient=request.sender,
message_type="research_response",
correlation_id=request.correlation_id,
resources=filtered_resources[:request.max_results],
search_query_used=self.search_utility.construct_search_query(
request.topic, request.resource_type, request.knowledge_level
),
total_results_found=len(resources),
errors=errors
)
except Exception as e:
self.logger.error(f"Research failed: {str(e)}")
return ResearchResponse(
sender=self.name,
recipient=request.sender,
message_type="research_response",
correlation_id=request.correlation_id,
resources=[],
search_query_used="",
total_results_found=0,
errors=[f"Research error: {str(e)}"]
)
class TutorialResearchAgent:
"""Specialized agent for finding tutorials."""
def __init__(self):
self.name = "TutorialResearchAgent"
self.search_utility = SearchUtility()
self.logger = logging.getLogger(self.__class__.__name__)
def enhance_tutorial_search(self, request: ResearchRequest) -> ResearchRequest:
enhanced_request = request.model_copy()
if request.specific_requirements:
enhanced_request.specific_requirements += " tutorial guide"
else:
enhanced_request.specific_requirements = "tutorial guide"
return enhanced_request
def filter_tutorial_results(self, resources: List[ResourceItem]) -> List[ResourceItem]:
if not resources:
return []
tutorial_indicators = ["tutorial", "guide", "how-to", "walkthrough",
"step-by-step", "documentation"]
scored_resources = []
for resource in resources:
text = (resource.title + " " + resource.description).lower()
score = sum(1 for indicator in tutorial_indicators if indicator in text)
scored_resources.append((score, resource))
scored_resources.sort(key=lambda x: x[0], reverse=True)
return [resource for score, resource in scored_resources]
async def research(self, request: ResearchRequest) -> ResearchResponse:
errors = []
try:
enhanced_request = self.enhance_tutorial_search(request)
resources = await self.search_utility.search_resources(enhanced_request)
if not resources:
errors.append("No tutorial resources found")
filtered_resources = self.filter_tutorial_results(resources)
return ResearchResponse(
sender=self.name,
recipient=request.sender,
message_type="research_response",
correlation_id=request.correlation_id,
resources=filtered_resources[:request.max_results],
search_query_used=self.search_utility.construct_search_query(
request.topic, request.resource_type, request.knowledge_level
),
total_results_found=len(resources),
errors=errors
)
except Exception as e:
self.logger.error(f"Research failed: {str(e)}")
return ResearchResponse(
sender=self.name,
recipient=request.sender,
message_type="research_response",
correlation_id=request.correlation_id,
resources=[],
search_query_used="",
total_results_found=0,
errors=[f"Research error: {str(e)}"]
)
class CourseResearchAgent:
"""Specialized agent for finding courses."""
def __init__(self):
self.name = "CourseResearchAgent"
self.search_utility = SearchUtility()
self.logger = logging.getLogger(self.__class__.__name__)
def enhance_course_search(self, request: ResearchRequest) -> ResearchRequest:
enhanced_request = request.model_copy()
course_platforms = "coursera edx udacity"
if request.specific_requirements:
enhanced_request.specific_requirements += f" course {course_platforms}"
else:
enhanced_request.specific_requirements = f"course {course_platforms}"
return enhanced_request
def filter_course_results(self, resources: List[ResourceItem]) -> List[ResourceItem]:
if not resources:
return []
course_indicators = ["course", "coursera", "edx", "udacity",
"certificate", "enroll"]
platform_domains = ["coursera.org", "edx.org", "udacity.com"]
scored_resources = []
for resource in resources:
text = (resource.title + " " + resource.description).lower()
score = sum(1 for indicator in course_indicators if indicator in text)
if any(domain in resource.url.lower() for domain in platform_domains):
score += 5
scored_resources.append((score, resource))
scored_resources.sort(key=lambda x: x[0], reverse=True)
return [resource for score, resource in scored_resources]
async def research(self, request: ResearchRequest) -> ResearchResponse:
errors = []
try:
enhanced_request = self.enhance_course_search(request)
resources = await self.search_utility.search_resources(enhanced_request)
if not resources:
errors.append("No course resources found")
filtered_resources = self.filter_course_results(resources)
return ResearchResponse(
sender=self.name,
recipient=request.sender,
message_type="research_response",
correlation_id=request.correlation_id,
resources=filtered_resources[:request.max_results],
search_query_used=self.search_utility.construct_search_query(
request.topic, request.resource_type, request.knowledge_level
),
total_results_found=len(resources),
errors=errors
)
except Exception as e:
self.logger.error(f"Research failed: {str(e)}")
return ResearchResponse(
sender=self.name,
recipient=request.sender,
message_type="research_response",
correlation_id=request.correlation_id,
resources=[],
search_query_used="",
total_results_found=0,
errors=[f"Research error: {str(e)}"]
)
class VideoResearchAgent:
"""Specialized agent for finding videos."""
def __init__(self):
self.name = "VideoResearchAgent"
self.search_utility = SearchUtility()
self.logger = logging.getLogger(self.__class__.__name__)
def enhance_video_search(self, request: ResearchRequest) -> ResearchRequest:
enhanced_request = request.model_copy()
if request.specific_requirements:
enhanced_request.specific_requirements += " video youtube"
else:
enhanced_request.specific_requirements = "video youtube"
return enhanced_request
def filter_video_results(self, resources: List[ResourceItem]) -> List[ResourceItem]:
if not resources:
return []
video_indicators = ["video", "youtube", "lecture", "watch"]
video_domains = ["youtube.com", "youtu.be", "vimeo.com"]
scored_resources = []
for resource in resources:
text = (resource.title + " " + resource.description).lower()
score = sum(1 for indicator in video_indicators if indicator in text)
if any(domain in resource.url.lower() for domain in video_domains):
score += 5
scored_resources.append((score, resource))
scored_resources.sort(key=lambda x: x[0], reverse=True)
return [resource for score, resource in scored_resources]
async def research(self, request: ResearchRequest) -> ResearchResponse:
errors = []
try:
enhanced_request = self.enhance_video_search(request)
resources = await self.search_utility.search_resources(enhanced_request)
if not resources:
errors.append("No video resources found")
filtered_resources = self.filter_video_results(resources)
return ResearchResponse(
sender=self.name,
recipient=request.sender,
message_type="research_response",
correlation_id=request.correlation_id,
resources=filtered_resources[:request.max_results],
search_query_used=self.search_utility.construct_search_query(
request.topic, request.resource_type, request.knowledge_level
),
total_results_found=len(resources),
errors=errors
)
except Exception as e:
self.logger.error(f"Research failed: {str(e)}")
return ResearchResponse(
sender=self.name,
recipient=request.sender,
message_type="research_response",
correlation_id=request.correlation_id,
resources=[],
search_query_used="",
total_results_found=0,
errors=[f"Research error: {str(e)}"]
)
# ============================================================================
# REPORT GENERATION AGENT
# ============================================================================
class ReportGenerationAgent:
"""Agent for synthesizing research into learning plans."""
def __init__(self, llm, gpu_config: GPUConfig, timeout: int = 60):
self.name = "ReportGenerationAgent"
self.llm = llm
self.gpu_config = gpu_config
self.timeout = timeout
self.logger = logging.getLogger(self.__class__.__name__)
def organize_resources(self, responses: List[ResearchResponse]) -> Dict[str, List[ResourceItem]]:
"""Organizes resources by type."""
organized = {
"books": [],
"tutorials": [],
"courses": [],
"videos": []
}
for response in responses:
if not response.resources:
continue
for resource in response.resources:
if "book" in response.sender.lower():
organized["books"].append(resource)
elif "tutorial" in response.sender.lower():
organized["tutorials"].append(resource)
elif "course" in response.sender.lower():
organized["courses"].append(resource)
elif "video" in response.sender.lower():
organized["videos"].append(resource)
return organized
def create_resource_summary(self, organized_resources: Dict[str, List[ResourceItem]]) -> str:
"""Creates text summary of resources."""
summary_parts = []
for resource_type, resources in organized_resources.items():
if resources:
summary_parts.append(f"\n{resource_type.upper()}:")
for idx, resource in enumerate(resources, 1):
summary_parts.append(
f"{idx}. {resource.title}\n"
f" URL: {resource.url}\n"
f" Description: {resource.description}\n"
f" Difficulty: {resource.difficulty_level.value}\n"
)
if not summary_parts:
return "No resources were found."
return "\n".join(summary_parts)
def create_fallback_plan(self, topic: str, knowledge_level: KnowledgeLevel,
organized_resources: Dict[str, List[ResourceItem]]) -> str:
"""Creates basic plan when LLM fails."""
plan_parts = []
plan_parts.append(f"Learning plan for {topic} at {knowledge_level.value} level.")
plan_parts.append("\nI have gathered resources to support your learning.")
total_resources = sum(len(resources) for resources in organized_resources.values())
if total_resources == 0:
plan_parts.append("\nNo specific resources found. Try refining your topic.")
return "\n".join(plan_parts)
async def generate_learning_plan(self, topic: str, knowledge_level: KnowledgeLevel,
responses: List[ResearchResponse]) -> LearningPlan:
"""Generates comprehensive learning plan."""
self.logger.info(f"Generating plan using {self.gpu_config.device_type.value}")
if self.gpu_config.device_type != DeviceType.CPU:
memory_before = self.gpu_config.get_memory_usage()
self.logger.debug(f"GPU memory before: {memory_before.get('allocated_gb', 0):.2f} GB")
organized_resources = self.organize_resources(responses)
resource_summary = self.create_resource_summary(organized_resources)
prompt = f"""Create a personalized learning plan.
Topic: {topic} Level: {knowledge_level.value}
Resources: {resource_summary}
Include:
- Welcoming introduction
- How to use resources effectively
- Recommended order
- Tips for {knowledge_level.value} level
- Encouragement
Be practical and encouraging."""
messages = [
ChatMessage(role=MessageRole.SYSTEM, content="You are an expert educational advisor."),
ChatMessage(role=MessageRole.USER, content=prompt)
]
try:
response = await asyncio.wait_for(
self.llm.achat(messages),
timeout=self.timeout
)
plan_text = response.message.content
if not plan_text or len(plan_text.strip()) < 50:
plan_text = self.create_fallback_plan(topic, knowledge_level, organized_resources)
except asyncio.TimeoutError:
self.logger.error(f"LLM timeout after {self.timeout}s")
plan_text = self.create_fallback_plan(topic, knowledge_level, organized_resources)
except Exception as e:
self.logger.error(f"Error generating plan: {str(e)}")
plan_text = self.create_fallback_plan(topic, knowledge_level, organized_resources)
if self.gpu_config.device_type != DeviceType.CPU:
memory_after = self.gpu_config.get_memory_usage()
self.logger.debug(f"GPU memory after: {memory_after.get('allocated_gb', 0):.2f} GB")
self.gpu_config.clear_cache()
resource_sections = []
for resource_type, resources in organized_resources.items():
if resources:
resource_sections.append({
"type": resource_type,
"resources": [r.model_dump() for r in resources]
})
return LearningPlan(
topic=topic,
knowledge_level=knowledge_level,
introduction=plan_text,
resource_sections=resource_sections,
estimated_total_time="Varies based on pace",
learning_tips=[
"Set aside dedicated time daily",
"Take notes and practice",
"Understanding over speed",
"Join communities for support",
"Build projects to apply knowledge"
],
next_steps="After completing resources, assess understanding and identify areas for deeper study"
)
def format_plan_for_display(self, plan: LearningPlan) -> str:
"""Formats plan for display."""
output = []
output.append("=" * 80)
output.append(f"LEARNING PLAN: {plan.topic.upper()}")
output.append(f"Level: {plan.knowledge_level.value.title()}")
output.append("=" * 80)
output.append("")
output.append(plan.introduction)
output.append("")
if plan.resource_sections:
output.append("-" * 80)
output.append("RESOURCES")
output.append("-" * 80)
for section in plan.resource_sections:
output.append("")
output.append(f"{section['type'].upper()}:")
output.append("")
for idx, resource in enumerate(section['resources'], 1):
output.append(f"{idx}. {resource['title']}")
output.append(f" Difficulty: {resource['difficulty_level']}")
output.append(f" URL: {resource['url']}")
output.append(f" {resource['description']}")
output.append("")
output.append("-" * 80)
output.append("TIPS")
output.append("-" * 80)
for tip in plan.learning_tips:
output.append(f"- {tip}")
output.append("")
output.append("-" * 80)
output.append("NEXT STEPS")
output.append("-" * 80)
output.append(plan.next_steps)
output.append("")
output.append("=" * 80)
return "\n".join(output)
# ============================================================================
# USER INTERFACE AGENT
# ============================================================================
class UserInterfaceAgent:
"""Main orchestration agent."""
def __init__(self, llm, gpu_config: GPUConfig, extraction_timeout: int = 30):
self.name = "UserInterfaceAgent"
self.llm = llm
self.gpu_config = gpu_config
self.extraction_timeout = extraction_timeout
self.state = ConversationState.INITIAL
self.topic: Optional[str] = None
self.knowledge_level: Optional[KnowledgeLevel] = None
self.conversation_history = []
self.logger = logging.getLogger(self.__class__.__name__)
self.book_agent = BookResearchAgent()
self.tutorial_agent = TutorialResearchAgent()
self.course_agent = CourseResearchAgent()
self.video_agent = VideoResearchAgent()
self.report_agent = ReportGenerationAgent(llm, gpu_config)
async def extract_learning_info(self, user_message: str) -> Tuple[Optional[str], Optional[KnowledgeLevel]]:
"""Extracts topic and level from user input."""
if not user_message or not user_message.strip():
return None, None
prompt = f"""Extract topic and level from: "{user_message}"
Format: TOPIC: [topic or "UNKNOWN"] LEVEL: [beginner/intermediate/advanced/expert or "UNKNOWN"]"""
messages = [
ChatMessage(role=MessageRole.SYSTEM, content="Extract information accurately."),
ChatMessage(role=MessageRole.USER, content=prompt)
]
try:
response = await asyncio.wait_for(
self.llm.achat(messages),
timeout=self.extraction_timeout
)
content = response.message.content
if not content:
return None, None
topic = None
level = None
for line in content.split('\n'):
line = line.strip()
if line.startswith('TOPIC:'):
extracted = line.replace('TOPIC:', '').strip()
if extracted and extracted != "UNKNOWN":
topic = extracted
elif line.startswith('LEVEL:'):
extracted = line.replace('LEVEL:', '').strip().lower()
if extracted in [lev.value for lev in KnowledgeLevel]:
level = KnowledgeLevel(extracted)
return topic, level
except Exception as e:
self.logger.error(f"Extraction error: {str(e)}")
return None, None
async def generate_clarifying_question(self) -> str:
"""Generates clarifying question."""
missing = []
if self.topic is None:
missing.append("topic")
if self.knowledge_level is None:
missing.append("knowledge level")
if not missing:
return ""
try:
prompt = f"Ask friendly question for: {' and '.join(missing)}"
messages = [
ChatMessage(role=MessageRole.SYSTEM, content="Be helpful."),
ChatMessage(role=MessageRole.USER, content=prompt)
]
response = await asyncio.wait_for(
self.llm.achat(messages),
timeout=self.extraction_timeout
)
return response.message.content.strip()
except Exception:
if self.topic is None and self.knowledge_level is None:
return "What would you like to learn and what is your level?"
elif self.topic is None:
return "What topic interests you?"
else:
return "What is your knowledge level (beginner/intermediate/advanced/expert)?"
async def dispatch_research_tasks(self) -> List[ResearchResponse]:
"""Dispatches parallel research tasks."""
if not self.topic or not self.knowledge_level:
return []
correlation_id = str(uuid.uuid4())
try:
requests = [
ResearchRequest(
sender=self.name,
recipient="BookResearchAgent",
message_type="research_request",
correlation_id=correlation_id,
topic=self.topic,
knowledge_level=self.knowledge_level,
resource_type=ResourceType.BOOK,
max_results=3
),
ResearchRequest(
sender=self.name,
recipient="TutorialResearchAgent",
message_type="research_request",
correlation_id=correlation_id,
topic=self.topic,
knowledge_level=self.knowledge_level,
resource_type=ResourceType.TUTORIAL,
max_results=3
),
ResearchRequest(
sender=self.name,
recipient="CourseResearchAgent",
message_type="research_request",
correlation_id=correlation_id,
topic=self.topic,
knowledge_level=self.knowledge_level,
resource_type=ResourceType.COURSE,
max_results=3
),
ResearchRequest(
sender=self.name,
recipient="VideoResearchAgent",
message_type="research_request",
correlation_id=correlation_id,
topic=self.topic,
knowledge_level=self.knowledge_level,
resource_type=ResourceType.VIDEO,
max_results=3
)
]
except Exception as e:
self.logger.error(f"Error creating requests: {str(e)}")
return []
tasks = [
self.book_agent.research(requests[0]),
self.tutorial_agent.research(requests[1]),
self.course_agent.research(requests[2]),
self.video_agent.research(requests[3])
]
try:
responses = await asyncio.wait_for(
asyncio.gather(*tasks, return_exceptions=True),
timeout=120.0
)
valid_responses = []
for response in responses:
if isinstance(response, ResearchResponse):
valid_responses.append(response)
return valid_responses
except asyncio.TimeoutError:
self.logger.error("Research tasks timed out")
return []
except Exception as e:
self.logger.error(f"Error gathering responses: {str(e)}")
return []
async def process_user_message(self, user_message: str) -> str:
"""Processes user input and orchestrates system."""
if not user_message or not user_message.strip():
return "Please tell me what you would like to learn."
self.conversation_history.append({"role": "user", "content": user_message})
try:
if self.state in [ConversationState.INITIAL, ConversationState.GATHERING_INFO]:
extracted_topic, extracted_level = await self.extract_learning_info(user_message)
if extracted_topic and self.topic is None:
self.topic = extracted_topic
if extracted_level and self.knowledge_level is None:
self.knowledge_level = extracted_level
if self.topic and self.knowledge_level:
self.state = ConversationState.RESEARCHING
response = f"Creating plan for {self.topic} at {self.knowledge_level.value} level...\n\n"
research_responses = await self.dispatch_research_tasks()
if not research_responses:
self.state = ConversationState.ERROR
return "Error searching. Please try again."
self.state = ConversationState.GENERATING_REPORT
learning_plan = await self.report_agent.generate_learning_plan(
self.topic, self.knowledge_level, research_responses
)
formatted = self.report_agent.format_plan_for_display(learning_plan)
response += formatted
self.state = ConversationState.COMPLETE
return response
else:
self.state = ConversationState.GATHERING_INFO
return await self.generate_clarifying_question()
elif self.state == ConversationState.COMPLETE:
self.topic = None
self.knowledge_level = None
self.state = ConversationState.INITIAL
self.conversation_history = []
return "Reset. What would you like to learn now?"
elif self.state == ConversationState.ERROR:
self.topic = None
self.knowledge_level = None
self.state = ConversationState.INITIAL
return "Let's start fresh. What interests you?"
return "Processing..."
except Exception as e:
self.logger.error(f"Error: {str(e)}")
self.state = ConversationState.ERROR
return "An error occurred. Please try again."
# ============================================================================
# LLM CONFIGURATION
# ============================================================================
class LLMConfig:
"""LLM configuration with GPU support."""
@staticmethod
def validate_ollama():
"""Validates Ollama availability."""
try:
import subprocess
result = subprocess.run(['ollama', 'list'], capture_output=True, text=True, timeout=5)
if result.returncode == 0:
logging.info("Ollama is running")
return True
logging.warning("Ollama not responding")
return False
except Exception as e:
logging.warning(f"Could not verify Ollama: {str(e)}")
return False
@staticmethod
def initialize_llm(provider: str = "ollama", model: str = None, gpu_config: Optional[GPUConfig] = None):
"""Initializes LLM with GPU configuration."""
load_dotenv()
provider = provider.lower().strip()
if provider == "openai":
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
raise ValueError("OPENAI_API_KEY not found")
model_name = model or "gpt-3.5-turbo"
logging.info(f"Initializing OpenAI: {model_name}")
return OpenAI(model=model_name, api_key=api_key, temperature=0.7, timeout=60.0)
elif provider == "ollama":
if not LLMConfig.validate_ollama():
raise ValueError("Ollama is not available. Please start Ollama server.")
model_name = model or "llama2"
logging.info(f"Initializing Ollama: {model_name}")
if gpu_config:
logging.info(f"GPU: {gpu_config.device_type.value}")
logging.info("Ollama manages GPU internally")
return Ollama(model=model_name, request_timeout=120.0, temperature=0.7)
else:
raise ValueError(f"Unsupported provider: {provider}")
# ============================================================================
# MAIN APPLICATION
# ============================================================================
class LearningAssistantApp:
"""Main application with GPU support."""
def __init__(self, llm_provider: str = "ollama", llm_model: str = None):
self.setup_logging()
self.logger.info(f"Initializing with {llm_provider}")
self.shutdown_handler = GracefulShutdown()
self.gpu_config = GPUConfig()
self.logger.info(f"GPU: {self.gpu_config.get_device_info()}")
try:
self.llm = LLMConfig.initialize_llm(llm_provider, llm_model, self.gpu_config)
self.ui_agent = UserInterfaceAgent(self.llm, self.gpu_config)
self.logger.info("Initialized successfully")
except Exception as e:
self.logger.error(f"Initialization failed: {str(e)}")
raise
def setup_logging(self):
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('learning_assistant.log'),
logging.StreamHandler(sys.stdout)
]
)
self.logger = logging.getLogger('LearningAssistant')
def print_welcome(self):
device_info = self.gpu_config.get_device_info()
gpu_status = f"Running on {device_info['device_type'].upper()}"
if device_info['device_type'] == 'cuda':
gpu_status += f" ({device_info.get('name', 'GPU')})"
print(f"""
================================================================================ LEARNING ASSISTANT ================================================================================
{gpu_status}
Tell me what you want to learn and your knowledge level.
Commands: 'quit', 'reset', 'gpu'
""")
async def interactive_mode(self):
self.print_welcome()
while not self.shutdown_handler.should_shutdown():
try:
user_input = input("\nYou: ").strip()
if not user_input:
continue
if user_input.lower() in ['quit', 'exit']:
print("\nThank you!")
break
if user_input.lower() == 'gpu':
info = self.gpu_config.get_device_info()
print(f"\nGPU Status:")
for key, value in info.items():
print(f" {key}: {value}")
continue
if user_input.lower() == 'reset':
self.ui_agent.topic = None
self.ui_agent.knowledge_level = None
self.ui_agent.state = ConversationState.INITIAL
self.gpu_config.clear_cache()
print("\nReset complete.")
continue
response = await self.ui_agent.process_user_message(user_input)
print(f"\nAssistant: {response}")
except KeyboardInterrupt:
print("\n\nGoodbye!")
break
except Exception as e:
self.logger.error(f"Error: {str(e)}")
print(f"\nError: {str(e)}")
async def generate_plan(self, topic: str, knowledge_level: str) -> str:
"""Generates plan programmatically."""
if not topic or not topic.strip():
raise ValueError("Topic cannot be empty")
try:
level_enum = KnowledgeLevel(knowledge_level.lower().strip())
except ValueError:
raise ValueError(f"Invalid knowledge level: {knowledge_level}")
self.ui_agent.topic = topic.strip()
self.ui_agent.knowledge_level = level_enum
self.ui_agent.state = ConversationState.RESEARCHING
research_responses = await self.ui_agent.dispatch_research_tasks()
if not research_responses:
raise RuntimeError("Failed to gather research")
learning_plan = await self.ui_agent.report_agent.generate_learning_plan(
topic.strip(), level_enum, research_responses
)
return self.ui_agent.report_agent.format_plan_for_display(learning_plan)
# ============================================================================
# MAIN ENTRY POINT
# ============================================================================
async def main():
import argparse
parser = argparse.ArgumentParser(description="GPU-Accelerated Learning Assistant")
parser.add_argument("--provider", type=str, default="ollama", choices=["openai", "ollama"])
parser.add_argument("--model", type=str, default=None)
parser.add_argument("--topic", type=str, default=None)
parser.add_argument("--level", type=str, default=None,
choices=["beginner", "intermediate", "advanced", "expert"])
args = parser.parse_args()
try:
app = LearningAssistantApp(llm_provider=args.provider, llm_model=args.model)
if args.topic and args.level:
print(f"Generating plan for: {args.topic} ({args.level})")
plan = await app.generate_plan(args.topic, args.level)
print(plan)
else:
await app.interactive_mode()
except Exception as e:
logging.error(f"Application error: {str(e)}")
print(f"Error: {str(e)}")
sys.exit(1)
if __name__ == "__main__":
asyncio.run(main())
The system now includes comprehensive GPU support with automatic detection, proper resource cleanup, rate limiting, input validation, graceful shutdown handling, timeout management for all async operations, and production-ready error handling throughout.
No comments:
Post a Comment