"""
Pattern 233: Virtual Assistant
Purpose: Personalized AI assistant that executes tasks, manages schedules, and provides proactive help.
Handles calendar, reminders, information retrieval, and task management with user preferences.
"""
from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
class TaskType(Enum):
"""Types of tasks the assistant can handle."""
SCHEDULE = "schedule" # Calendar events
REMINDER = "reminder" # Set reminders
SEARCH = "search" # Information retrieval
EMAIL = "email" # Email management
TODO = "todo" # Task management
WEATHER = "weather" # Weather info
NEWS = "news" # News updates
CALCULATION = "calculation" # Math calculations
TRANSLATION = "translation" # Language translation
GENERAL = "general" # General assistance
class Priority(Enum):
"""Task priority levels."""
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
URGENT = "urgent"
@dataclass
class CalendarEvent:
"""Calendar event."""
title: str
start_time: datetime
end_time: datetime
description: str = ""
location: str = ""
attendees: List[str] = field(default_factory=list)
reminder_minutes: int = 15
@dataclass
class Reminder:
"""Reminder for user."""
message: str
remind_at: datetime
priority: Priority = Priority.MEDIUM
recurring: bool = False
completed: bool = False
@dataclass
class TodoItem:
"""Todo list item."""
task: str
priority: Priority = Priority.MEDIUM
due_date: Optional[datetime] = None
completed: bool = False
tags: List[str] = field(default_factory=list)
@dataclass
class UserPreferences:
"""User preferences and profile."""
name: str
timezone: str = "UTC"
preferred_language: str = "English"
work_hours: tuple = (9, 17) # 9 AM to 5 PM
interests: List[str] = field(default_factory=list)
notification_style: str = "normal" # normal, minimal, detailed
class VirtualAssistant:
"""
Virtual Assistant - Personalized AI assistant for task execution and scheduling.
Features:
- Calendar management with scheduling
- Reminder system with priorities
- Todo list management
- Information retrieval and search
- Proactive suggestions
- Personalized responses based on user preferences
"""
def __init__(self,
name: str,
user_preferences: Optional[UserPreferences] = None):
self.name = name
self.user_preferences = user_preferences or UserPreferences(name="User")
self.llm = ChatOpenAI(model="gpt-4", temperature=0.7)
# State management
self.calendar: List[CalendarEvent] = []
self.reminders: List[Reminder] = []
self.todo_list: List[TodoItem] = []
self.conversation_context: List[str] = []
# Task handlers
self.task_handlers: Dict[TaskType, Callable] = {
TaskType.SCHEDULE: self._handle_schedule,
TaskType.REMINDER: self._handle_reminder,
TaskType.SEARCH: self._handle_search,
TaskType.TODO: self._handle_todo,
TaskType.WEATHER: self._handle_weather,
TaskType.CALCULATION: self._handle_calculation,
TaskType.GENERAL: self._handle_general
}
# Statistics
self.statistics: Dict[str, Any] = {
"tasks_completed": 0,
"events_scheduled": 0,
"reminders_set": 0,
"proactive_suggestions": 0
}
def assist(self, user_request: str) -> str:
"""Process user request and provide assistance."""
# Add to conversation context
self.conversation_context.append(user_request)
# Determine task type
task_type = self._classify_task(user_request)
# Execute appropriate handler
handler = self.task_handlers.get(task_type, self._handle_general)
response = handler(user_request)
# Check for proactive suggestions
suggestions = self._generate_proactive_suggestions()
if suggestions:
response += f"\n\n💡 Proactive suggestion: {suggestions}"
self.statistics["proactive_suggestions"] += 1
self.statistics["tasks_completed"] += 1
return response
def _classify_task(self, user_request: str) -> TaskType:
"""Classify the type of task from user request."""
request_lower = user_request.lower()
# Pattern matching for task classification
if any(word in request_lower for word in ["schedule", "meeting", "appointment", "calendar", "book"]):
return TaskType.SCHEDULE
elif any(word in request_lower for word in ["remind", "reminder", "alert", "notify"]):
return TaskType.REMINDER
elif any(word in request_lower for word in ["search", "find", "look up", "google", "what is"]):
return TaskType.SEARCH
elif any(word in request_lower for word in ["todo", "task", "add to list", "need to do"]):
return TaskType.TODO
elif any(word in request_lower for word in ["weather", "temperature", "forecast", "rain"]):
return TaskType.WEATHER
elif any(word in request_lower for word in ["calculate", "compute", "math", "+", "-", "*", "/"]):
return TaskType.CALCULATION
else:
return TaskType.GENERAL
# Task handlers
def _handle_schedule(self, request: str) -> str:
"""Handle calendar scheduling."""
# Extract event details using LLM
prompt = ChatPromptTemplate.from_messages([
("system", "Extract event details from the user's scheduling request. "
"Provide: title, date, time, and any other relevant details."),
("human", "{request}")
])
chain = prompt | self.llm
response = chain.invoke({"request": request})
# Create event (simplified - in production would parse LLM response)
event = CalendarEvent(
title="Meeting",
start_time=datetime.now() + timedelta(days=1, hours=2),
end_time=datetime.now() + timedelta(days=1, hours=3),
description=request
)
self.calendar.append(event)
self.statistics["events_scheduled"] += 1
return f"✓ Scheduled: {event.title} for {event.start_time.strftime('%Y-%m-%d at %H:%M')}. " \
f"I'll remind you 15 minutes before."
def _handle_reminder(self, request: str) -> str:
"""Handle reminder setting."""
# Determine reminder time and message
reminder = Reminder(
message=request,
remind_at=datetime.now() + timedelta(hours=1), # Default 1 hour
priority=Priority.MEDIUM
)
self.reminders.append(reminder)
self.statistics["reminders_set"] += 1
return f"✓ Reminder set: '{reminder.message}' for {reminder.remind_at.strftime('%Y-%m-%d at %H:%M')}"
def _handle_search(self, request: str) -> str:
"""Handle information search."""
# Use LLM for information retrieval
prompt = ChatPromptTemplate.from_messages([
("system", f"You are {self.name}, a helpful virtual assistant. "
f"Provide concise, accurate information in response to user queries. "
f"User's interests: {', '.join(self.user_preferences.interests or ['general topics'])}"),
("human", "{query}")
])
chain = prompt | self.llm
response = chain.invoke({"query": request})
return f"🔍 Search result: {response.content}"
def _handle_todo(self, request: str) -> str:
"""Handle todo list management."""
# Extract task from request
task_text = request.replace("add", "").replace("to do", "").replace("todo", "").strip()
todo = TodoItem(
task=task_text,
priority=Priority.MEDIUM,
due_date=None
)
self.todo_list.append(todo)
incomplete_count = sum(1 for t in self.todo_list if not t.completed)
return f"✓ Added to todo list: '{todo.task}'. You now have {incomplete_count} pending tasks."
def _handle_weather(self, request: str) -> str:
"""Handle weather queries."""
# Simulate weather response (in production, would call weather API)
return f"☀️ Weather forecast: Currently 72°F and sunny. " \
f"Perfect weather for outdoor activities!"
def _handle_calculation(self, request: str) -> str:
"""Handle mathematical calculations."""
import re
# Extract numbers and operators
numbers = re.findall(r'\d+\.?\d*', request)
if len(numbers) >= 2:
try:
# Simple calculation (in production, use safer evaluation)
if '+' in request:
result = sum(float(n) for n in numbers)
return f"🔢 Calculation: {' + '.join(numbers)} = {result}"
elif '-' in request:
result = float(numbers[0]) - float(numbers[1])
return f"🔢 Calculation: {numbers[0]} - {numbers[1]} = {result}"
elif '*' in request or 'times' in request:
result = float(numbers[0]) * float(numbers[1])
return f"🔢 Calculation: {numbers[0]} × {numbers[1]} = {result}"
elif '/' in request or 'divided' in request:
result = float(numbers[0]) / float(numbers[1])
return f"🔢 Calculation: {numbers[0]} ÷ {numbers[1]} = {result:.2f}"
except:
pass
return "I can help with calculations. Please provide a clear mathematical expression."
def _handle_general(self, request: str) -> str:
"""Handle general assistance requests."""
prompt = ChatPromptTemplate.from_messages([
("system", f"You are {self.name}, a personalized virtual assistant for {self.user_preferences.name}. "
f"Be helpful, friendly, and proactive. Provide clear, actionable responses. "
f"User interests: {', '.join(self.user_preferences.interests or ['general topics'])}. "
f"Preferred language: {self.user_preferences.preferred_language}."),
("human", "{request}")
])
chain = prompt | self.llm
response = chain.invoke({"request": request})
return response.content
def _generate_proactive_suggestions(self) -> Optional[str]:
"""Generate proactive suggestions based on context."""
current_hour = datetime.now().hour
# Morning suggestion
if 6 <= current_hour 0:
next_event = min(self.calendar, key=lambda e: e.start_time)
if next_event.start_time.date() == datetime.now().date():
return f"You have '{next_event.title}' at {next_event.start_time.strftime('%H:%M')} today."
# Pending tasks suggestion
if len(self.todo_list) > 5:
incomplete = [t for t in self.todo_list if not t.completed]
if len(incomplete) > 3:
return f"You have {len(incomplete)} pending tasks. Would you like me to prioritize them?"
# Reminder check
upcoming_reminders = [r for r in self.reminders
if not r.completed and r.remind_at List[CalendarEvent]:
"""Get schedule for a specific date."""
target_date = date or datetime.now()
return [e for e in self.calendar if e.start_time.date() == target_date.date()]
def get_pending_tasks(self) -> List[TodoItem]:
"""Get all pending todo items."""
return [t for t in self.todo_list if not t.completed]
def complete_task(self, task_index: int) -> str:
"""Mark a todo task as completed."""
if 0 <= task_index str:
"""Get daily summary."""
today_events = self.get_schedule()
pending_tasks = self.get_pending_tasks()
active_reminders = [r for r in self.reminders if not r.completed]
summary = f"📊 Daily Summary for {self.user_preferences.name}:\n"
summary += f" • Events today: {len(today_events)}\n"
summary += f" • Pending tasks: {len(pending_tasks)}\n"
summary += f" • Active reminders: {len(active_reminders)}\n"
return summary
def get_statistics(self) -> Dict[str, Any]:
"""Get assistant statistics."""
return {
"name": self.name,
"user": self.user_preferences.name,
"total_events": len(self.calendar),
"total_reminders": len(self.reminders),
"pending_tasks": len(self.get_pending_tasks()),
**self.statistics
}
class VirtualAssistantPattern:
"""
Virtual Assistant Pattern - Manage personalized AI assistants.
Coordinates multiple virtual assistants for different users.
"""
def __init__(self):
self.assistants: Dict[str, VirtualAssistant] = {}
def create_assistant(self, name: str, user_preferences: UserPreferences) -> VirtualAssistant:
"""Create a new virtual assistant."""
assistant = VirtualAssistant(name, user_preferences)
self.assistants[name] = assistant
print(f" Created assistant: {name} for {user_preferences.name}")
return assistant
def run(self, assistant_name: str, requests: List[str]) -> Dict[str, Any]:
"""Run series of requests with assistant."""
if assistant_name not in self.assistants:
return {"error": f"Assistant {assistant_name} not found"}
assistant = self.assistants[assistant_name]
responses = []
print(f"\n=== {assistant_name} Assistance Session ===\n")
for request in requests:
print(f"User: {request}")
response = assistant.assist(request)
print(f"{assistant_name}: {response}\n")
responses.append(response)
# Show summary
print(assistant.get_summary())
return {
"assistant": assistant_name,
"requests_handled": len(requests),
"responses": responses,
"statistics": assistant.get_statistics()
}
# Example usage
if __name__ == "__main__":
print("=== Virtual Assistant Pattern Demo ===\n")
# Create pattern
pattern = VirtualAssistantPattern()
# Create user preferences
print("1. Setting up virtual assistants...")
user_prefs = UserPreferences(
name="Alice",
timezone="America/New_York",
preferred_language="English",
interests=["technology", "productivity", "health"]
)
assistant = pattern.create_assistant("Jarvis", user_prefs)
# Add some initial data
assistant.todo_list.append(TodoItem(task="Review project proposal", priority=Priority.HIGH))
assistant.todo_list.append(TodoItem(task="Call dentist", priority=Priority.MEDIUM))
# Run assistance session
print("\n2. Running assistance session...")
requests = [
"What's on my schedule today?",
"Add a reminder to buy groceries",
"Schedule a meeting for tomorrow at 2 PM",
"What's 25 times 4?",
"Add 'Prepare presentation' to my todo list",
"Search for information about artificial intelligence",
"What's the weather like?",
"Show me my pending tasks"
]
result = pattern.run("Jarvis", requests)
# Print statistics
print("\n3. Assistant Statistics:")
stats = result["statistics"]
for key, value in stats.items():
print(f" {key}: {value}")
print("\n✓ Virtual assistant provided personalized help and task management!")
self.state.current_step += 1
prompt = ChatPromptTemplate.from_messages([
("system", f"You are {self.name}, implementing Virtual Assistant. "
f"Provide clear, structured responses."),
("human", "Process this input: {input_data}")
])
chain = prompt | self.llm
response = chain.invoke({"input_data": str(input_data)})
result = {
"agent": self.name,
"step": self.state.current_step,
"input": input_data,
"output": str(response.content),
"timestamp": datetime.now()
}
self.execution_history.append(result)
self.state.results.append(result)
return result
def get_statistics(self) -> Dict[str, Any]:
"""Get agent statistics."""
return {
"name": self.name,
"total_executions": len(self.execution_history),
"current_step": self.state.current_step,
"results_count": len(self.state.results)
}
class VirtualAssistantPattern:
"""
Virtual Assistant - Enhanced implementation
This pattern provides comprehensive functionality for virtual assistant
with full LangChain/LangGraph integration.
Features:
- Multi-agent coordination
- State management
- Execution tracking
- LangGraph workflow integration
"""
def __init__(self, config: VirtualAssistantConfig = None):
self.config = config or VirtualAssistantConfig()
self.agents: Dict[str, VirtualAssistantAgent] = {}
self.workflow = self._build_workflow()
self.execution_log: List[Dict[str, Any]] = []
def _build_workflow(self) -> StateGraph:
"""Build LangGraph workflow for pattern execution."""
class WorkflowState(TypedDict):
input: str
processed: str
output: str
metadata: Dict[str, Any]
def process_input(state: WorkflowState) -> WorkflowState:
"""Process input node."""
return {
**state,
"processed": f"Processed: {state['input']}",
"metadata": {"step": "process_input"}
}
def generate_output(state: WorkflowState) -> WorkflowState:
"""Generate output node."""
llm = ChatOpenAI(model=self.config.model_name, temperature=self.config.temperature)
prompt = ChatPromptTemplate.from_messages([
("system", "You are implementing Virtual Assistant. Generate comprehensive output."),
("human", "{input}")
])
chain = prompt | llm
response = chain.invoke({"input": state["processed"]})
return {
**state,
"output": str(response.content),
"metadata": {**state.get("metadata", {}), "step": "generate_output"}
}
# Build workflow graph
workflow = StateGraph(WorkflowState)
workflow.add_node("process", process_input)
workflow.add_node("generate", generate_output)
workflow.set_entry_point("process")
workflow.add_edge("process", "generate")
workflow.add_edge("generate", END)
return workflow.compile()
def register_agent(self, agent: VirtualAssistantAgent):
"""Register an agent with the pattern."""
self.agents[agent.name] = agent
if self.config.enable_logging:
print(f"Registered agent: {agent.name}")
def execute(self, input_data: Any, agent_name: str = None) -> Dict[str, Any]:
"""Execute the pattern with given input."""
# Use workflow if no specific agent requested
if agent_name is None:
result = self.workflow.invoke({
"input": str(input_data),
"processed": "",
"output": "",
"metadata": {}
})
execution_record = {
"pattern": "Virtual Assistant",
"input": input_data,
"output": result.get("output", ""),
"metadata": result.get("metadata", {}),
"timestamp": datetime.now()
}
self.execution_log.append(execution_record)
return execution_record
# Execute with specific agent
agent = self.agents.get(agent_name)
if not agent:
raise ValueError(f"Agent {agent_name} not found")
result = agent.process(input_data)
self.execution_log.append(result)
return result
def run(self, input_data: Any) -> Dict[str, Any]:
"""Main execution method."""
return self.execute(input_data)
def get_statistics(self) -> Dict[str, Any]:
"""Get pattern statistics."""
return {
"pattern": "Virtual Assistant",
"total_executions": len(self.execution_log),
"registered_agents": len(self.agents),
"agents": [agent.get_statistics() for agent in self.agents.values()]
}
# Example usage
if __name__ == "__main__":
print("=== Virtual Assistant Demo ===\n")
# Create configuration
config = VirtualAssistantConfig(
model_name="gpt-4",
temperature=0.7,
enable_logging=True
)
# Create pattern
pattern = VirtualAssistantPattern(config)
# Create and register agents
agent1 = VirtualAssistantAgent("Agent1", config)
agent2 = VirtualAssistantAgent("Agent2", config)
pattern.register_agent(agent1)
pattern.register_agent(agent2)
print("\nExecuting pattern workflow...")
# Execute with workflow
result1 = pattern.run("Demonstrate virtual assistant functionality")
print(f"\nWorkflow Result:")
print(f" Output: {result1['output'][:150]}...")
# Execute with specific agents
result2 = pattern.execute("Specific task for Agent1", agent_name="Agent1")
print(f"\nAgent1 Result:")
print(f" Output: {result2['output'][:150]}...")
result3 = pattern.execute("Specific task for Agent2", agent_name="Agent2")
print(f"\nAgent2 Result:")
print(f" Output: {result3['output'][:150]}...")
# Show statistics
print("\n=== Statistics ===")
stats = pattern.get_statistics()
print(f"Pattern: {stats['pattern']}")
print(f"Total Executions: {stats['total_executions']}")
print(f"Registered Agents: {stats['registered_agents']}")
print(f"\nAgent Statistics:")
for agent_stats in stats['agents']:
print(f" {agent_stats['name']}: {agent_stats['total_executions']} executions")
Agentic Architecture Pattern: Virtual Assistant
Pattern 233: 233 Virtual Assistant – Line-by-Line Explanation
Overview
Pattern 233: Virtual Assistant
Purpose: Personalized AI assistant that executes tasks, manages schedules, and provides proactive help. Handles calendar, reminders, information retrieval, and task management with user preferences.
Code Structure Analysis
File Information
- Pattern Number: 233
- Pattern Name: 233 Virtual Assistant
- File Path:
patterns\domain_specific\233_virtual_assistant.py - Total Lines: 640
Detailed Code Explanation
Module Docstring
"""
Pattern 233: Virtual Assistant
Purpose: Personalized AI assistant that executes tasks, manages schedules, and provides proactive help.
Handles calendar, reminders, information retrieval, and task management with user preferences.
"""
Explanation: This module docstring describes the purpose and key features of the 233 Virtual Assistant pattern.
Imports Section
from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import re # First 20 imports shown
Explanation: The module imports necessary dependencies:
– from typing import Dict, Any, List, Optional, Callable: Type hints for better code documentation
– from dataclasses import dataclass, field: Data class decorators for clean data structures
– from datetime import datetime, timedelta: Date and time utilities
– from enum import Enum: Enumeration support for fixed sets of values
– from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration
– from langchain_openai import ChatOpenAI: LangChain framework integration
Design Patterns Used
- Strategy Pattern: Different execution strategies for various scenarios
- State Pattern: Managing different states in the workflow
- Chain of Responsibility: Sequential processing through workflow nodes
- Observer Pattern: Tracking and monitoring pattern execution
- Factory Pattern: Creating instances with proper configuration
Usage Example
# Usage example not available
Key Concepts
- LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
- Workflow Orchestration: LangGraph manages execution flow through nodes
- State Management: TypedDict defines workflow state structure
- Error Handling: Comprehensive error handling and fallback mechanisms
- Extensibility: Designed for easy customization and extension
Extension Points
This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities