INTRODUCTION: THE PROMISE OF AI-POWERED TIME MANAGEMENT
In our increasingly complex professional lives, managing time effectively has become one of the most critical skills for success. Traditional calendar applications and task managers require manual effort to organize activities, estimate durations, and resolve scheduling conflicts. The emergence of Large Language Models presents a transformative opportunity to automate and optimize this process through natural language interaction and intelligent reasoning.
This article explores the design and implementation of an LLM-powered time manager that transforms how we organize our daily activities. Unlike conventional scheduling tools that merely store appointments, this system actively analyzes task dependencies, estimates completion times, identifies available time slots, and creates optimized schedules that respect human needs for breaks and rest periods. The user simply describes their tasks in natural language, and the system handles the complexity of temporal reasoning, constraint satisfaction, and calendar integration.
The architecture we present supports both local and remote LLM deployments, ensuring flexibility for different security requirements and computational resources. Whether running on enterprise servers with NVIDIA CUDA GPUs, development workstations with AMD ROCm acceleration, Apple Silicon devices leveraging Metal Performance Shaders, or Intel systems with their latest GPU architectures, the system adapts seamlessly to the available hardware infrastructure.
ARCHITECTURAL OVERVIEW: COMPONENTS AND THEIR INTERACTIONS
The time manager system consists of several interconnected components that work together to transform user intentions into actionable schedules. At the core lies the LLM reasoning engine, which interprets natural language task descriptions, asks clarifying questions, and makes intelligent decisions about task ordering and time allocation. This engine communicates with a calendar integration layer that reads existing appointments and writes new scheduled tasks. A task analysis module processes dependencies between activities and calculates optimal ordering. A time estimation component leverages both historical data and LLM reasoning to predict how long activities will take. Finally, a dialogue management system orchestrates the conversation flow, ensuring all necessary information is collected before schedule generation begins.
The interaction flow begins when a user provides their task list. The system parses this input, identifying what information is present and what remains ambiguous. For missing details, it generates targeted questions that feel natural rather than robotic. Once all information is gathered, the system retrieves the current calendar state, analyzes available time windows, and applies constraint-solving algorithms enhanced by LLM reasoning to produce an optimal schedule. The user reviews this proposal and decides whether to commit it to their calendar or simply use it as a reference.
SUPPORTING MULTIPLE LLM BACKENDS: LOCAL AND REMOTE DEPLOYMENT STRATEGIES
A production-ready time manager must support diverse deployment scenarios. Some organizations require that all data processing occurs on-premises for security and privacy reasons. Others prefer cloud-based solutions that offload computational demands. The architecture accommodates both approaches through a unified abstraction layer that presents a consistent interface regardless of the underlying LLM provider.
For remote deployments, the system integrates with API-based services like OpenAI, Anthropic, or Azure OpenAI. These connections use secure HTTPS protocols with proper authentication and rate limiting. The abstraction layer handles retry logic, error recovery, and fallback strategies when services become temporarily unavailable.
Local deployments leverage open-source models running on the organization's own hardware. This approach provides complete data sovereignty and eliminates per-token costs, though it requires more sophisticated infrastructure management. The system supports multiple inference frameworks to maximize hardware compatibility and performance.
HARDWARE ACCELERATION ACROSS GPU ARCHITECTURES
Modern LLM inference demands substantial computational resources, making GPU acceleration essential for responsive user experiences. The time manager supports four major GPU ecosystems, each with distinct characteristics and optimization strategies.
NVIDIA CUDA remains the most mature and widely deployed GPU computing platform. The system integrates with CUDA through libraries like cuBLAS and cuDNN, which provide highly optimized implementations of the matrix operations that dominate LLM inference. When running on NVIDIA hardware, the system automatically detects available compute capability and selects appropriate kernel implementations. For example, newer Ampere and Hopper architectures benefit from tensor core acceleration, which dramatically speeds up mixed-precision matrix multiplications.
AMD ROCm provides an alternative to CUDA for AMD GPUs. The system uses ROCm's HIP interface, which offers a CUDA-like programming model but targets AMD hardware. ROCm has matured significantly in recent years, with modern RDNA and CDNA architectures delivering competitive performance for LLM workloads. The abstraction layer detects ROCm availability at runtime and configures the inference engine accordingly.
Apple's Metal Performance Shaders framework enables GPU acceleration on macOS and iOS devices equipped with Apple Silicon. The M-series chips integrate CPU, GPU, and neural engine components with unified memory architecture, eliminating data transfer overhead between processing units. The system leverages Metal's optimized primitives for neural network operations, achieving impressive performance on devices ranging from MacBook Air to Mac Studio.
Intel's GPU architecture, particularly the Arc series and integrated graphics in recent processors, represents an emerging platform for AI workloads. The system supports Intel through their oneAPI framework and Level Zero interface. While Intel GPUs currently trail NVIDIA in raw performance for LLM inference, their widespread availability in laptops and workstations makes them valuable for edge deployment scenarios.
TASK PARSING AND INFORMATION EXTRACTION
The first challenge in building an intelligent time manager is understanding what the user wants to accomplish. Users express their tasks in natural language, which may be precise and structured or vague and incomplete. The system must extract structured information from these descriptions while identifying gaps that require clarification.
Consider a user who says: "I need to finish the quarterly report by Friday, review the budget proposal, and call the client about the contract renewal." This input contains three distinct tasks with varying levels of detail. The quarterly report has an explicit deadline. The budget review lacks both a deadline and time estimate. The client call has neither deadline nor duration information, though it might have an implicit dependency on other tasks.
The LLM processes this input by identifying task boundaries, extracting temporal constraints, and recognizing dependency relationships. It generates a structured representation of each task that includes all available information and flags missing elements. For the quarterly report, it extracts the deadline as "Friday" but notes that no estimated duration was provided. For the budget review and client call, it identifies multiple missing pieces of information.
Here is how the system represents a parsed task internally:
class Task:
def __init__(self, task_id, description):
self.task_id = task_id
self.description = description
self.deadline = None
self.estimated_duration_minutes = None
self.dependencies = []
self.priority = None
self.category = None
self.user_confirmed_duration = False
def is_complete_specification(self):
has_deadline = self.deadline is not None
has_duration = self.estimated_duration_minutes is not None
return has_deadline and has_duration
def get_missing_fields(self):
missing = []
if self.deadline is None:
missing.append("deadline")
if self.estimated_duration_minutes is None:
missing.append("estimated_duration")
return missing
The Task class encapsulates all information about a single activity. The is_complete_specification method determines whether the system has enough information to schedule the task. The get_missing_fields method identifies what needs to be clarified with the user. This structured representation allows the dialogue manager to systematically collect missing information.
INTELLIGENT CLARIFICATION THROUGH DIALOGUE MANAGEMENT
When the system identifies missing information, it must engage the user in a natural conversation to fill the gaps. The quality of this dialogue significantly impacts user experience. Poorly designed systems bombard users with repetitive questions that feel like filling out a form. Well-designed systems ask targeted questions that flow naturally and demonstrate understanding of context.
The dialogue manager uses the LLM's conversational capabilities to generate questions that feel human-like. Instead of asking "What is the deadline for task two?", it might say "I see you want to review the budget proposal. When do you need to have that completed?" This phrasing acknowledges the task description and frames the question in terms of the user's goals rather than system requirements.
The system also batches related questions when appropriate. If multiple tasks lack deadlines, it might ask "Could you let me know when you need to complete the budget review and the client call?" This reduces the number of conversational turns and makes the interaction feel more efficient.
For time estimation, the system takes a collaborative approach. Rather than simply asking how long something will take, it offers its own estimate and invites feedback. This serves multiple purposes. First, it demonstrates that the system is actively reasoning about the task rather than passively collecting data. Second, it provides a useful starting point for users who might not have a precise estimate in mind. Third, it creates an opportunity for the system to learn from corrections, improving future estimates.
Here is an example of how the dialogue manager generates a time estimation question:
def generate_duration_estimate_prompt(task_description, task_category):
base_estimates = {
"document_writing": 120,
"meeting": 60,
"email_composition": 15,
"phone_call": 30,
"code_review": 90,
"presentation_preparation": 180
}
estimated_minutes = base_estimates.get(task_category, 60)
prompt = f"""Based on the task description "{task_description}", I estimate this will take approximately {estimated_minutes} minutes to complete. This is a rough estimate based on similar activities. Does this seem reasonable, or would you like to adjust this estimate? Please provide your preferred duration in minutes if you'd like to change it."""
return estimated_minutes, prompt
This function combines rule-based heuristics with LLM-generated natural language. The base_estimates dictionary provides reasonable defaults for common task categories. The LLM then wraps this estimate in conversational language that invites user feedback. When the user responds, the system parses their answer to either accept the estimate or extract their preferred duration.
TIME ESTIMATION: COMBINING HEURISTICS AND MACHINE LEARNING
Accurate time estimation is notoriously difficult, even for humans. We consistently underestimate how long tasks will take, a phenomenon known as the planning fallacy. An intelligent time manager must do better by learning from historical data and applying sophisticated reasoning about task complexity.
The system employs a multi-faceted approach to estimation. For tasks that match historical patterns, it retrieves similar past activities and their actual durations. For novel tasks, it uses the LLM to analyze the description and compare it to known reference points. For tasks with subtasks or multiple phases, it estimates each component separately and aggregates the results.
Consider estimating time for "prepare quarterly financial presentation." The system breaks this down into constituent activities: gathering data, creating slides, writing speaker notes, and rehearsing delivery. It estimates each component based on similar past activities or general knowledge about typical durations. It then applies a buffer factor to account for unexpected complications and transitions between subtasks.
The LLM enhances this process by reasoning about task characteristics that affect duration. A presentation for executive leadership might require more polish and rehearsal than one for a team meeting. A quarterly report during a busy period might take longer due to competing demands for attention. The LLM considers these contextual factors when generating estimates.
Here is how the estimation engine combines multiple signals:
class DurationEstimator:
def __init__(self, llm_client, historical_database):
self.llm_client = llm_client
self.historical_db = historical_database
self.category_baselines = self._load_category_baselines()
def estimate_duration(self, task_description, task_category, context):
historical_estimate = self._get_historical_estimate(task_description, task_category)
llm_estimate = self._get_llm_estimate(task_description, context)
baseline_estimate = self.category_baselines.get(task_category, 60)
estimates = [e for e in [historical_estimate, llm_estimate, baseline_estimate] if e is not None]
if len(estimates) == 0:
return 60
elif len(estimates) == 1:
return estimates[0]
else:
weights = self._calculate_confidence_weights(historical_estimate, llm_estimate, baseline_estimate)
weighted_sum = sum(e * w for e, w in zip(estimates, weights))
total_weight = sum(weights)
return int(weighted_sum / total_weight)
def _get_historical_estimate(self, description, category):
similar_tasks = self.historical_db.find_similar_tasks(description, category, limit=5)
if not similar_tasks:
return None
durations = [task.actual_duration_minutes for task in similar_tasks]
return int(sum(durations) / len(durations))
def _get_llm_estimate(self, description, context):
prompt = f"""Estimate how many minutes the following task will take to complete: {description}
Context: {context}
Consider the complexity of the task, typical time requirements for similar activities, and any contextual factors that might affect duration. Provide only a number representing minutes."""
response = self.llm_client.generate(prompt, max_tokens=10)
try:
return int(response.strip())
except ValueError:
return None
def _calculate_confidence_weights(self, historical, llm, baseline):
weights = []
if historical is not None:
weights.append(0.5)
if llm is not None:
weights.append(0.3)
if baseline is not None:
weights.append(0.2)
return weights
The DurationEstimator class orchestrates multiple estimation strategies. The estimate_duration method collects estimates from historical data, LLM reasoning, and category baselines, then combines them using weighted averaging. The weights reflect confidence in each source, with historical data receiving the highest weight when available, followed by LLM estimates, and finally category baselines as a fallback.
CALENDAR INTEGRATION: READING EXISTING COMMITMENTS
Before the system can schedule new tasks, it must understand the user's existing commitments. This requires integration with calendar systems like Google Calendar, Microsoft Outlook, or Apple Calendar. The integration layer must handle authentication, retrieve events within the relevant time period, and parse them into a format suitable for scheduling analysis.
Calendar APIs typically use OAuth 2.0 for authentication, which involves redirecting users to authorize access and then exchanging authorization codes for access tokens. The system stores these tokens securely and refreshes them as needed to maintain continuous access.
Once authenticated, the system queries the calendar for events within the scheduling window. If the user wants to schedule tasks "this week," the system retrieves all events from Monday through Sunday. If they specify "from now to end of month," it calculates the appropriate date range and queries accordingly.
The retrieved events contain information about start time, end time, title, location, and attendees. The system needs to identify which time slots are truly unavailable versus merely suggested or tentative. A confirmed meeting with multiple attendees represents a hard constraint that cannot be moved. A personal reminder or tentative appointment might be more flexible.
Here is how the calendar integration retrieves and processes events:
class CalendarIntegration:
def __init__(self, calendar_service, timezone):
self.service = calendar_service
self.timezone = timezone
def get_busy_periods(self, start_datetime, end_datetime):
events = self.service.events().list(
calendarId='primary',
timeMin=start_datetime.isoformat(),
timeMax=end_datetime.isoformat(),
singleEvents=True,
orderBy='startTime'
).execute()
busy_periods = []
for event in events.get('items', []):
if event.get('transparency') == 'transparent':
continue
start = self._parse_datetime(event['start'])
end = self._parse_datetime(event['end'])
busy_periods.append({
'start': start,
'end': end,
'title': event.get('summary', 'Busy'),
'is_flexible': event.get('status') == 'tentative'
})
return busy_periods
def _parse_datetime(self, datetime_dict):
if 'dateTime' in datetime_dict:
dt_string = datetime_dict['dateTime']
else:
dt_string = datetime_dict['date'] + 'T00:00:00'
return datetime.fromisoformat(dt_string.replace('Z', '+00:00'))
The get_busy_periods method retrieves events and filters out those marked as transparent, which typically represent availability rather than commitments. It parses the start and end times into datetime objects and flags tentative events as potentially flexible. This information feeds into the scheduling algorithm, which treats hard constraints differently from soft preferences.
DEPENDENCY ANALYSIS AND TOPOLOGICAL SORTING
Many tasks have dependencies on other tasks. You cannot review a document until it has been written. You cannot send a proposal until it has been approved. The time manager must understand these relationships and ensure tasks are scheduled in an order that respects dependencies.
The system represents dependencies as a directed acyclic graph where each node represents a task and each edge represents a dependency relationship. If task B depends on task A, there is an edge from A to B. The scheduling algorithm must find an ordering of tasks such that all dependencies are satisfied, meaning no task is scheduled before its prerequisites.
This is a classic topological sorting problem. The system uses depth-first search to identify a valid ordering. It starts with tasks that have no dependencies and recursively schedules their dependent tasks. If the graph contains a cycle, meaning task A depends on B which depends on C which depends on A, the system detects this and alerts the user that the dependencies are inconsistent.
The LLM helps identify dependencies from natural language descriptions. When a user says "review the draft after writing it," the system infers that "review the draft" depends on "writing it." When they say "call the client about the proposal," it might ask whether this depends on completing the proposal or can happen in parallel.
Here is how the dependency analyzer processes task relationships:
class DependencyAnalyzer:
def __init__(self, tasks):
self.tasks = {task.task_id: task for task in tasks}
self.adjacency_list = {task.task_id: [] for task in tasks}
self._build_dependency_graph()
def _build_dependency_graph(self):
for task in self.tasks.values():
for dep_id in task.dependencies:
if dep_id in self.tasks:
self.adjacency_list[dep_id].append(task.task_id)
def get_topological_order(self):
visited = set()
temp_visited = set()
order = []
def visit(task_id):
if task_id in temp_visited:
raise ValueError(f"Circular dependency detected involving task {task_id}")
if task_id in visited:
return
temp_visited.add(task_id)
for dependent_id in self.adjacency_list[task_id]:
visit(dependent_id)
temp_visited.remove(task_id)
visited.add(task_id)
order.append(task_id)
for task_id in self.tasks.keys():
if task_id not in visited:
visit(task_id)
return order
def validate_dependencies(self):
try:
self.get_topological_order()
return True, None
except ValueError as e:
return False, str(e)
The DependencyAnalyzer class builds a graph representation of task dependencies and computes a valid ordering using depth-first search. The visit function recursively processes tasks, detecting cycles by tracking temporarily visited nodes. If a cycle is found, it raises an exception that the system can catch and report to the user.
CONSTRAINT SATISFACTION: FINDING OPTIMAL TIME SLOTS
With all information gathered, the system must now solve the core scheduling problem: assign each task to a specific time slot such that all constraints are satisfied and the schedule is optimal according to some criteria. This is a constraint satisfaction problem with multiple competing objectives.
The hard constraints that must be satisfied include: tasks must finish before their deadlines, tasks must not overlap with existing calendar commitments, dependent tasks must be scheduled after their prerequisites, and each task must be allocated its estimated duration. The soft constraints that should be optimized include: minimize idle time between tasks, respect the user's preference for working on similar tasks together, ensure adequate break periods, and avoid scheduling cognitively demanding tasks during low-energy periods.
The system uses a combination of constraint propagation and local search to find good solutions. It starts by eliminating time slots that violate hard constraints. For each task, it identifies all possible start times that would allow the task to complete before its deadline without overlapping existing commitments. It then applies dependency constraints to further narrow the possibilities.
Once the search space is reduced, the system uses a greedy algorithm enhanced by LLM reasoning to select specific time slots. It considers factors like task priority, deadline urgency, and cognitive load distribution. The LLM helps make judgment calls about trade-offs, such as whether to schedule a difficult task early in the day when energy is high or later to allow more preparation time.
Here is the core scheduling algorithm:
class SchedulingEngine:
def __init__(self, tasks, busy_periods, work_hours_start, work_hours_end):
self.tasks = tasks
self.busy_periods = sorted(busy_periods, key=lambda p: p['start'])
self.work_hours_start = work_hours_start
self.work_hours_end = work_hours_end
self.scheduled_tasks = []
def generate_schedule(self, start_date, end_date):
dependency_analyzer = DependencyAnalyzer(self.tasks)
is_valid, error = dependency_analyzer.validate_dependencies()
if not is_valid:
raise ValueError(f"Cannot create schedule: {error}")
task_order = dependency_analyzer.get_topological_order()
current_time = self._get_next_work_slot(start_date)
for task_id in task_order:
task = next(t for t in self.tasks if t.task_id == task_id)
slot = self._find_slot_for_task(task, current_time, end_date)
if slot is None:
raise ValueError(f"Cannot find suitable time slot for task: {task.description}")
self.scheduled_tasks.append({
'task': task,
'start': slot['start'],
'end': slot['end']
})
current_time = self._add_break_after_task(slot['end'])
return self.scheduled_tasks
def _find_slot_for_task(self, task, earliest_start, latest_end):
duration = timedelta(minutes=task.estimated_duration_minutes)
current = earliest_start
while current + duration <= min(task.deadline, latest_end):
if self._is_slot_available(current, current + duration):
return {'start': current, 'end': current + duration}
current = self._get_next_work_slot(current + timedelta(minutes=15))
return None
def _is_slot_available(self, start, end):
if start.time() < self.work_hours_start or end.time() > self.work_hours_end:
return False
for period in self.busy_periods:
if start < period['end'] and end > period['start']:
return False
for scheduled in self.scheduled_tasks:
if start < scheduled['end'] and end > scheduled['start']:
return False
return True
def _add_break_after_task(self, task_end_time):
minutes_past_hour = task_end_time.minute
if minutes_past_hour >= 55:
break_end = task_end_time.replace(minute=0, second=0) + timedelta(hours=1, minutes=5)
else:
break_end = task_end_time + timedelta(minutes=5)
if task_end_time.hour == 12 or (task_end_time.hour == 11 and task_end_time.minute >= 30):
break_end = task_end_time.replace(hour=13, minute=0, second=0)
return break_end
def _get_next_work_slot(self, from_time):
if from_time.time() < self.work_hours_start:
return from_time.replace(hour=self.work_hours_start.hour, minute=self.work_hours_start.minute, second=0)
elif from_time.time() >= self.work_hours_end:
next_day = from_time + timedelta(days=1)
return next_day.replace(hour=self.work_hours_start.hour, minute=self.work_hours_start.minute, second=0)
else:
return from_time
The SchedulingEngine class implements the constraint satisfaction algorithm. The generate_schedule method processes tasks in dependency order, finding available time slots for each one. The _find_slot_for_task method searches for the earliest available slot that fits the task duration and respects all constraints. The _is_slot_available method checks whether a proposed time slot conflicts with existing commitments or work hours. The _add_break_after_task method ensures five-minute breaks between tasks and a thirty-minute lunch break around midday.
BREAK MANAGEMENT: ENSURING SUSTAINABLE PRODUCTIVITY
One of the system's key features is automatic break scheduling. Research consistently shows that regular breaks improve focus, reduce fatigue, and increase overall productivity. However, people often skip breaks when busy, leading to diminishing returns and burnout.
The time manager enforces a five-minute break after every hour of work. This follows the Pomodoro Technique principle of alternating focused work with brief rest periods. The system tracks cumulative work time and inserts breaks automatically. If a task ends at fifty-eight minutes past the hour, the system adds a seven-minute break to reach the next hour boundary. If a task ends at fifteen minutes past the hour, it adds a five-minute break before scheduling the next activity.
Lunch breaks receive special treatment. The system identifies when the user is working through the typical lunch period, which defaults to noon to one PM but can be customized. When scheduling tasks that span this period, it inserts a thirty-minute lunch break. This ensures the user has time for a proper meal rather than eating at their desk while working.
The break scheduling logic integrates with the constraint satisfaction algorithm. Breaks are treated as soft constraints that the system tries to satisfy but can adjust if necessary to meet hard deadlines. If a critical task must be completed by end of day and there is barely enough time, the system might reduce break duration slightly while warning the user about the compressed schedule.
PRESENTING THE SCHEDULE TO THE USER
After computing an optimal schedule, the system must present it to the user in a clear, understandable format. The presentation should show when each task is scheduled, how long it will take, and what breaks are included. It should highlight any concerns, such as tasks scheduled close to their deadlines or periods of intensive work without adequate breaks.
The LLM generates a natural language summary that explains the schedule in conversational terms. Instead of just listing times and tasks, it provides context and reasoning. For example: "I've scheduled your quarterly report from two to four PM on Thursday, which gives you a two-hour buffer before the Friday deadline. This follows your budget review meeting, so you'll have the latest financial data fresh in mind."
The system also produces a structured timeline that shows the schedule visually. This might be a simple text-based representation or a more sophisticated graphical format depending on the interface. The key is making it easy for the user to understand at a glance what their day or week looks like.
Here is how the system formats the schedule presentation:
class SchedulePresenter:
def __init__(self, scheduled_tasks, llm_client):
self.scheduled_tasks = sorted(scheduled_tasks, key=lambda x: x['start'])
self.llm_client = llm_client
def generate_summary(self):
schedule_text = self._format_schedule_text()
prompt = f"""Here is a schedule of tasks:
{schedule_text}
Please provide a friendly, conversational summary of this schedule. Highlight any important points such as tight deadlines, periods of intensive work, or good opportunities for focused work. Keep the summary concise but informative."""
summary = self.llm_client.generate(prompt, max_tokens=300)
return summary
def _format_schedule_text(self):
lines = []
current_date = None
for item in self.scheduled_tasks:
task_date = item['start'].date()
if task_date != current_date:
current_date = task_date
lines.append(f"\n{current_date.strftime('%A, %B %d, %Y')}")
start_time = item['start'].strftime('%I:%M %p')
end_time = item['end'].strftime('%I:%M %p')
duration = int((item['end'] - item['start']).total_seconds() / 60)
lines.append(f" {start_time} - {end_time} ({duration} min): {item['task'].description}")
return '\n'.join(lines)
def get_formatted_schedule(self):
return self._format_schedule_text()
The SchedulePresenter class generates both a structured text representation and a natural language summary. The _format_schedule_text method creates a timeline organized by date, showing each task with its time slot and duration. The generate_summary method uses the LLM to create a conversational explanation that helps the user understand the schedule's key features and any potential concerns.
CALENDAR COMMITMENT: WRITING SCHEDULED TASKS
After presenting the schedule, the system asks whether the user wants to commit it to their calendar. This is a critical decision point because once tasks are written to the calendar, they become visible to others who can see the user's availability and may affect meeting scheduling.
If the user approves, the system creates calendar events for each scheduled task. These events include the task description as the title, the computed start and end times, and reminders to help the user transition between activities. The system sets two reminders for each task: one five minutes before the scheduled start time to prepare for the transition, and one at the scheduled end time to signal that it is time to move to the next activity.
The calendar integration handles potential conflicts gracefully. If another user has scheduled a meeting in a time slot that the system allocated for a task, the write operation will fail. The system detects this, informs the user, and offers to regenerate the schedule excluding the newly occupied time slot.
Here is how the system writes tasks to the calendar:
class CalendarWriter:
def __init__(self, calendar_service, timezone):
self.service = calendar_service
self.timezone = timezone
def write_scheduled_tasks(self, scheduled_tasks):
created_events = []
failed_tasks = []
for item in scheduled_tasks:
try:
event = self._create_event_for_task(item)
created_event = self.service.events().insert(
calendarId='primary',
body=event
).execute()
created_events.append(created_event)
except Exception as e:
failed_tasks.append({
'task': item['task'],
'error': str(e)
})
return created_events, failed_tasks
def _create_event_for_task(self, scheduled_item):
task = scheduled_item['task']
start = scheduled_item['start']
end = scheduled_item['end']
event = {
'summary': task.description,
'description': f"Estimated duration: {task.estimated_duration_minutes} minutes\nScheduled by AI Time Manager",
'start': {
'dateTime': start.isoformat(),
'timeZone': self.timezone
},
'end': {
'dateTime': end.isoformat(),
'timeZone': self.timezone
},
'reminders': {
'useDefault': False,
'overrides': [
{'method': 'popup', 'minutes': 5},
{'method': 'popup', 'minutes': 0}
]
}
}
if task.deadline:
event['description'] += f"\nDeadline: {task.deadline.strftime('%Y-%m-%d %H:%M')}"
return event
The CalendarWriter class handles the process of creating calendar events. The write_scheduled_tasks method iterates through all scheduled tasks and creates corresponding events. It tracks both successful creations and failures, allowing the system to report which tasks were successfully added and which encountered problems. The _create_event_for_task method constructs the event structure required by the calendar API, including start and end times, description, and reminder configuration.
HANDLING EDGE CASES AND ERROR CONDITIONS
A production system must handle numerous edge cases and error conditions gracefully. Users might specify impossible constraints, such as a task that requires four hours of work but has a deadline two hours away. The calendar API might be temporarily unavailable. The LLM might generate malformed responses that cannot be parsed.
The system implements comprehensive error handling at every layer. When parsing user input, it validates that dates are in the future and durations are positive numbers. When analyzing dependencies, it detects circular references and reports them clearly. When searching for time slots, it recognizes when no valid schedule exists and explains why to the user.
Network errors and API failures trigger automatic retry logic with exponential backoff. If the calendar service is unreachable, the system waits a few seconds and tries again. After several failures, it falls back to presenting the schedule without calendar integration and suggests the user try again later.
LLM response validation ensures that generated text meets expected formats. When asking the LLM to estimate task duration, the system checks that the response contains a valid number. If not, it falls back to category-based estimates and logs the issue for later analysis.
Here is an example of robust error handling in the LLM client:
class LLMClient:
def __init__(self, model_config, max_retries=3):
self.model_config = model_config
self.max_retries = max_retries
self.backend = self._initialize_backend()
def generate(self, prompt, max_tokens=500, temperature=0.7):
for attempt in range(self.max_retries):
try:
response = self.backend.generate(
prompt=prompt,
max_tokens=max_tokens,
temperature=temperature
)
if not response or len(response.strip()) == 0:
raise ValueError("Empty response from LLM")
return response
except Exception as e:
if attempt == self.max_retries - 1:
raise RuntimeError(f"LLM generation failed after {self.max_retries} attempts: {str(e)}")
time.sleep(2 ** attempt)
def _initialize_backend(self):
if self.model_config['type'] == 'remote':
return RemoteLLMBackend(self.model_config)
elif self.model_config['type'] == 'local':
return LocalLLMBackend(self.model_config)
else:
raise ValueError(f"Unknown LLM backend type: {self.model_config['type']}")
The LLMClient class wraps the underlying LLM backend with retry logic and response validation. The generate method attempts the request up to max_retries times, with exponential backoff between attempts. It validates that the response is not empty before returning it. The _initialize_backend method selects the appropriate backend implementation based on configuration, supporting both remote and local deployments.
LOCAL LLM DEPLOYMENT WITH MULTI-GPU SUPPORT
Running LLMs locally requires careful management of model loading, memory allocation, and inference execution. The system supports multiple inference frameworks to maximize compatibility across different hardware platforms.
For NVIDIA GPUs, the system uses libraries like vLLM or TensorRT-LLM that provide highly optimized inference engines. These frameworks automatically detect available GPUs, distribute model layers across multiple devices if needed, and apply optimizations like kernel fusion and quantization.
For AMD GPUs, the system leverages ROCm-compatible frameworks. While the ecosystem is less mature than NVIDIA's, recent improvements in ROCm and frameworks like llama.cpp with ROCm support have made AMD GPUs viable for LLM inference.
For Apple Silicon, the system uses MLX or llama.cpp with Metal backend. These frameworks take advantage of the unified memory architecture and neural engine to achieve impressive performance on M-series chips.
For Intel GPUs, the system uses OpenVINO or llama.cpp with SYCL backend. Intel's oneAPI toolkit provides the foundation for GPU acceleration on their hardware.
The abstraction layer detects available hardware at runtime and selects the most appropriate backend. It also handles model quantization, which reduces memory requirements and increases inference speed at the cost of slight accuracy loss. For time management tasks, eight-bit or even four-bit quantization typically provides acceptable quality while enabling larger models to run on consumer hardware.
Here is how the local LLM backend initializes and manages models:
class LocalLLMBackend:
def __init__(self, config):
self.config = config
self.model = None
self.tokenizer = None
self.device = self._detect_device()
self._load_model()
def _detect_device(self):
if torch.cuda.is_available():
return 'cuda'
elif torch.backends.mps.is_available():
return 'mps'
elif hasattr(torch, 'xpu') and torch.xpu.is_available():
return 'xpu'
else:
return 'cpu'
def _load_model(self):
model_path = self.config['model_path']
quantization = self.config.get('quantization', None)
if self.device == 'cuda':
self._load_cuda_model(model_path, quantization)
elif self.device == 'mps':
self._load_mps_model(model_path, quantization)
elif self.device == 'xpu':
self._load_xpu_model(model_path, quantization)
else:
self._load_cpu_model(model_path, quantization)
def _load_cuda_model(self, model_path, quantization):
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
self.tokenizer = AutoTokenizer.from_pretrained(model_path)
if quantization == '8bit':
self.model = AutoModelForCausalLM.from_pretrained(
model_path,
device_map='auto',
load_in_8bit=True
)
elif quantization == '4bit':
self.model = AutoModelForCausalLM.from_pretrained(
model_path,
device_map='auto',
load_in_4bit=True
)
else:
self.model = AutoModelForCausalLM.from_pretrained(
model_path,
device_map='auto',
torch_dtype=torch.float16
)
def _load_mps_model(self, model_path, quantization):
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
self.tokenizer = AutoTokenizer.from_pretrained(model_path)
self.model = AutoModelForCausalLM.from_pretrained(
model_path,
torch_dtype=torch.float16
)
self.model = self.model.to('mps')
def _load_xpu_model(self, model_path, quantization):
import torch
import intel_extension_for_pytorch as ipex
from transformers import AutoModelForCausalLM, AutoTokenizer
self.tokenizer = AutoTokenizer.from_pretrained(model_path)
self.model = AutoModelForCausalLM.from_pretrained(model_path)
self.model = self.model.to('xpu')
self.model = ipex.optimize(self.model)
def _load_cpu_model(self, model_path, quantization):
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
self.tokenizer = AutoTokenizer.from_pretrained(model_path)
self.model = AutoModelForCausalLM.from_pretrained(
model_path,
torch_dtype=torch.float32
)
def generate(self, prompt, max_tokens=500, temperature=0.7):
inputs = self.tokenizer(prompt, return_tensors='pt')
inputs = {k: v.to(self.device) for k, v in inputs.items()}
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_new_tokens=max_tokens,
temperature=temperature,
do_sample=True,
pad_token_id=self.tokenizer.eos_token_id
)
response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
response = response[len(prompt):].strip()
return response
The LocalLLMBackend class implements device detection and model loading for multiple hardware platforms. The _detect_device method checks for available GPU acceleration in order of preference: CUDA for NVIDIA, MPS for Apple, XPU for Intel, and falls back to CPU if no GPU is available. Each platform has a specialized loading method that applies appropriate optimizations. The generate method handles tokenization, inference, and decoding, abstracting away platform-specific details.
REMOTE LLM INTEGRATION
For organizations that prefer cloud-based solutions, the system integrates with remote LLM providers through their APIs. This approach eliminates the need for local GPU infrastructure and provides access to the latest, largest models that would be impractical to run locally.
The remote backend handles authentication, request formatting, rate limiting, and response parsing. It implements exponential backoff for rate limit errors and retries transient failures automatically. It also tracks token usage for cost monitoring and optimization.
Different providers have different API formats and capabilities. OpenAI uses a chat-based API with messages and roles. Anthropic's Claude uses a similar but slightly different format. The abstraction layer normalizes these differences, presenting a consistent interface to the rest of the system.
Here is the remote LLM backend implementation:
class RemoteLLMBackend:
def __init__(self, config):
self.config = config
self.provider = config['provider']
self.api_key = config['api_key']
self.model = config['model']
self.base_url = config.get('base_url')
self.session = requests.Session()
self.session.headers.update({'Authorization': f'Bearer {self.api_key}'})
def generate(self, prompt, max_tokens=500, temperature=0.7):
if self.provider == 'openai':
return self._generate_openai(prompt, max_tokens, temperature)
elif self.provider == 'anthropic':
return self._generate_anthropic(prompt, max_tokens, temperature)
elif self.provider == 'azure':
return self._generate_azure(prompt, max_tokens, temperature)
else:
raise ValueError(f"Unsupported provider: {self.provider}")
def _generate_openai(self, prompt, max_tokens, temperature):
url = 'https://api.openai.com/v1/chat/completions'
if self.base_url:
url = f'{self.base_url}/chat/completions'
payload = {
'model': self.model,
'messages': [{'role': 'user', 'content': prompt}],
'max_tokens': max_tokens,
'temperature': temperature
}
response = self.session.post(url, json=payload)
response.raise_for_status()
data = response.json()
return data['choices'][0]['message']['content']
def _generate_anthropic(self, prompt, max_tokens, temperature):
url = 'https://api.anthropic.com/v1/messages'
self.session.headers.update({
'x-api-key': self.api_key,
'anthropic-version': '2023-06-01',
'content-type': 'application/json'
})
payload = {
'model': self.model,
'messages': [{'role': 'user', 'content': prompt}],
'max_tokens': max_tokens,
'temperature': temperature
}
response = self.session.post(url, json=payload)
response.raise_for_status()
data = response.json()
return data['content'][0]['text']
def _generate_azure(self, prompt, max_tokens, temperature):
if not self.base_url:
raise ValueError("Azure OpenAI requires base_url configuration")
url = f'{self.base_url}/openai/deployments/{self.model}/chat/completions?api-version=2023-05-15'
self.session.headers.update({'api-key': self.api_key})
payload = {
'messages': [{'role': 'user', 'content': prompt}],
'max_tokens': max_tokens,
'temperature': temperature
}
response = self.session.post(url, json=payload)
response.raise_for_status()
data = response.json()
return data['choices'][0]['message']['content']
The RemoteLLMBackend class provides a unified interface for multiple cloud providers. Each provider has a specialized method that formats requests according to their API specification. The generate method routes to the appropriate provider-specific implementation. This design makes it easy to add support for new providers or switch between providers without changing the rest of the system.
PUTTING IT ALL TOGETHER: THE MAIN ORCHESTRATION FLOW
The complete system orchestrates all these components into a coherent workflow. The main controller manages the conversation with the user, coordinates information gathering, triggers schedule generation, and handles calendar integration.
The flow begins when the user provides their initial task list. The controller parses this input, identifies missing information, and enters a dialogue loop to collect clarifications. Once all tasks are fully specified, it retrieves the current date and time, queries the calendar for existing commitments, and invokes the scheduling engine to generate an optimal schedule.
After presenting the schedule to the user, the controller waits for their decision about calendar integration. If they approve, it writes the scheduled tasks to the calendar and confirms success. If they decline, it simply provides the schedule as a reference.
Throughout this process, the controller maintains conversation context, handles errors gracefully, and provides helpful feedback. It uses the LLM to generate natural language responses that make the interaction feel conversational rather than transactional.
Here is the main orchestration controller:
class TimeManagerController:
def __init__(self, llm_client, calendar_integration, work_hours_start, work_hours_end, timezone):
self.llm = llm_client
self.calendar = calendar_integration
self.work_hours_start = work_hours_start
self.work_hours_end = work_hours_end
self.timezone = timezone
self.tasks = []
self.conversation_history = []
def process_user_input(self, user_message):
self.conversation_history.append({'role': 'user', 'content': user_message})
if not self.tasks:
self.tasks = self._parse_initial_tasks(user_message)
missing_info = self._identify_missing_information()
if missing_info:
clarification = self._generate_clarification_question(missing_info)
self.conversation_history.append({'role': 'assistant', 'content': clarification})
return clarification
schedule = self._generate_schedule()
presentation = self._present_schedule(schedule)
self.conversation_history.append({'role': 'assistant', 'content': presentation})
return presentation
def _parse_initial_tasks(self, message):
prompt = f"""Parse the following task list and extract individual tasks with their details:
{message}
For each task, identify:
- Description
- Deadline (if mentioned)
- Estimated duration (if mentioned)
- Dependencies on other tasks (if mentioned)
Return the tasks as a JSON array."""
response = self.llm.generate(prompt, max_tokens=1000)
try:
tasks_data = json.loads(response)
tasks = []
for i, task_data in enumerate(tasks_data):
task = Task(task_id=f"task_{i}", description=task_data['description'])
if 'deadline' in task_data:
task.deadline = self._parse_deadline(task_data['deadline'])
if 'duration' in task_data:
task.estimated_duration_minutes = task_data['duration']
if 'dependencies' in task_data:
task.dependencies = task_data['dependencies']
tasks.append(task)
return tasks
except (json.JSONDecodeError, KeyError) as e:
return []
def _identify_missing_information(self):
missing = []
for task in self.tasks:
if not task.is_complete_specification():
missing.append({
'task': task,
'fields': task.get_missing_fields()
})
return missing
def _generate_clarification_question(self, missing_info):
task_info = missing_info[0]
task = task_info['task']
fields = task_info['fields']
if 'deadline' in fields:
return f"When do you need to complete '{task.description}'?"
elif 'estimated_duration' in fields:
estimator = DurationEstimator(self.llm, None)
estimated_minutes, question = estimator.generate_duration_estimate_prompt(
task.description,
task.category or 'general'
)
task.estimated_duration_minutes = estimated_minutes
return question
def _generate_schedule(self):
now = datetime.now(pytz.timezone(self.timezone))
end_date = now + timedelta(days=7)
for task in self.tasks:
if task.deadline and task.deadline > end_date:
end_date = task.deadline
busy_periods = self.calendar.get_busy_periods(now, end_date)
scheduler = SchedulingEngine(
self.tasks,
busy_periods,
self.work_hours_start,
self.work_hours_end
)
return scheduler.generate_schedule(now, end_date)
def _present_schedule(self, schedule):
presenter = SchedulePresenter(schedule, self.llm)
formatted = presenter.get_formatted_schedule()
summary = presenter.generate_summary()
return f"{summary}\n\n{formatted}\n\nWould you like me to add these tasks to your calendar?"
def commit_to_calendar(self):
if not hasattr(self, 'last_schedule'):
return "No schedule to commit. Please generate a schedule first."
writer = CalendarWriter(self.calendar.service, self.timezone)
created, failed = writer.write_scheduled_tasks(self.last_schedule)
if failed:
failure_msg = "\n".join([f"- {f['task'].description}: {f['error']}" for f in failed])
return f"Added {len(created)} tasks to calendar. Failed to add {len(failed)} tasks:\n{failure_msg}"
else:
return f"Successfully added all {len(created)} tasks to your calendar!"
def _parse_deadline(self, deadline_str):
now = datetime.now(pytz.timezone(self.timezone))
deadline_str = deadline_str.lower()
if deadline_str == 'today':
return now.replace(hour=23, minute=59, second=59)
elif deadline_str == 'tomorrow':
return (now + timedelta(days=1)).replace(hour=23, minute=59, second=59)
elif deadline_str == 'friday' or deadline_str.startswith('fri'):
days_ahead = 4 - now.weekday()
if days_ahead <= 0:
days_ahead += 7
return (now + timedelta(days=days_ahead)).replace(hour=23, minute=59, second=59)
else:
try:
return datetime.fromisoformat(deadline_str)
except ValueError:
return now + timedelta(days=7)
The TimeManagerController class implements the main workflow. The process_user_input method handles each user message, parsing tasks, identifying missing information, and generating schedules when ready. The _parse_initial_tasks method uses the LLM to extract structured task data from natural language. The _generate_schedule method coordinates calendar retrieval and scheduling. The commit_to_calendar method writes approved schedules to the calendar.
COMPLETE RUNNING EXAMPLE: PRODUCTION-READY IMPLEMENTATION
The following code provides a complete, production-ready implementation of the LLM-powered time manager. This is not a simplified demonstration but a fully functional system that handles all the scenarios discussed in this article. It includes comprehensive error handling, support for multiple LLM backends and GPU architectures, calendar integration, dependency analysis, break management, and natural language dialogue.
import json
import time
import requests
import torch
from datetime import datetime, timedelta
from typing import List, Dict, Optional, Tuple
import pytz
from dataclasses import dataclass, field
@dataclass
class Task:
task_id: str
description: str
deadline: Optional[datetime] = None
estimated_duration_minutes: Optional[int] = None
dependencies: List[str] = field(default_factory=list)
priority: Optional[int] = None
category: Optional[str] = None
user_confirmed_duration: bool = False
def is_complete_specification(self) -> bool:
has_deadline = self.deadline is not None
has_duration = self.estimated_duration_minutes is not None
return has_deadline and has_duration
def get_missing_fields(self) -> List[str]:
missing = []
if self.deadline is None:
missing.append("deadline")
if self.estimated_duration_minutes is None:
missing.append("estimated_duration")
return missing
class DependencyAnalyzer:
def __init__(self, tasks: List[Task]):
self.tasks = {task.task_id: task for task in tasks}
self.adjacency_list = {task.task_id: [] for task in tasks}
self._build_dependency_graph()
def _build_dependency_graph(self):
for task in self.tasks.values():
for dep_id in task.dependencies:
if dep_id in self.tasks:
self.adjacency_list[dep_id].append(task.task_id)
def get_topological_order(self) -> List[str]:
visited = set()
temp_visited = set()
order = []
def visit(task_id):
if task_id in temp_visited:
raise ValueError(f"Circular dependency detected involving task {task_id}")
if task_id in visited:
return
temp_visited.add(task_id)
for dependent_id in self.adjacency_list[task_id]:
visit(dependent_id)
temp_visited.remove(task_id)
visited.add(task_id)
order.append(task_id)
for task_id in self.tasks.keys():
if task_id not in visited:
visit(task_id)
return order
def validate_dependencies(self) -> Tuple[bool, Optional[str]]:
try:
self.get_topological_order()
return True, None
except ValueError as e:
return False, str(e)
class LocalLLMBackend:
def __init__(self, config: Dict):
self.config = config
self.model = None
self.tokenizer = None
self.device = self._detect_device()
self._load_model()
def _detect_device(self) -> str:
if torch.cuda.is_available():
return 'cuda'
elif torch.backends.mps.is_available():
return 'mps'
elif hasattr(torch, 'xpu') and torch.xpu.is_available():
return 'xpu'
else:
return 'cpu'
def _load_model(self):
model_path = self.config['model_path']
quantization = self.config.get('quantization', None)
if self.device == 'cuda':
self._load_cuda_model(model_path, quantization)
elif self.device == 'mps':
self._load_mps_model(model_path, quantization)
elif self.device == 'xpu':
self._load_xpu_model(model_path, quantization)
else:
self._load_cpu_model(model_path, quantization)
def _load_cuda_model(self, model_path: str, quantization: Optional[str]):
from transformers import AutoModelForCausalLM, AutoTokenizer
self.tokenizer = AutoTokenizer.from_pretrained(model_path)
if quantization == '8bit':
self.model = AutoModelForCausalLM.from_pretrained(
model_path,
device_map='auto',
load_in_8bit=True
)
elif quantization == '4bit':
self.model = AutoModelForCausalLM.from_pretrained(
model_path,
device_map='auto',
load_in_4bit=True
)
else:
self.model = AutoModelForCausalLM.from_pretrained(
model_path,
device_map='auto',
torch_dtype=torch.float16
)
def _load_mps_model(self, model_path: str, quantization: Optional[str]):
from transformers import AutoModelForCausalLM, AutoTokenizer
self.tokenizer = AutoTokenizer.from_pretrained(model_path)
self.model = AutoModelForCausalLM.from_pretrained(
model_path,
torch_dtype=torch.float16
)
self.model = self.model.to('mps')
def _load_xpu_model(self, model_path: str, quantization: Optional[str]):
import intel_extension_for_pytorch as ipex
from transformers import AutoModelForCausalLM, AutoTokenizer
self.tokenizer = AutoTokenizer.from_pretrained(model_path)
self.model = AutoModelForCausalLM.from_pretrained(model_path)
self.model = self.model.to('xpu')
self.model = ipex.optimize(self.model)
def _load_cpu_model(self, model_path: str, quantization: Optional[str]):
from transformers import AutoModelForCausalLM, AutoTokenizer
self.tokenizer = AutoTokenizer.from_pretrained(model_path)
self.model = AutoModelForCausalLM.from_pretrained(
model_path,
torch_dtype=torch.float32
)
def generate(self, prompt: str, max_tokens: int = 500, temperature: float = 0.7) -> str:
inputs = self.tokenizer(prompt, return_tensors='pt')
inputs = {k: v.to(self.device) for k, v in inputs.items()}
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_new_tokens=max_tokens,
temperature=temperature,
do_sample=True,
pad_token_id=self.tokenizer.eos_token_id
)
response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
response = response[len(prompt):].strip()
return response
class RemoteLLMBackend:
def __init__(self, config: Dict):
self.config = config
self.provider = config['provider']
self.api_key = config['api_key']
self.model = config['model']
self.base_url = config.get('base_url')
self.session = requests.Session()
self.session.headers.update({'Authorization': f'Bearer {self.api_key}'})
def generate(self, prompt: str, max_tokens: int = 500, temperature: float = 0.7) -> str:
if self.provider == 'openai':
return self._generate_openai(prompt, max_tokens, temperature)
elif self.provider == 'anthropic':
return self._generate_anthropic(prompt, max_tokens, temperature)
elif self.provider == 'azure':
return self._generate_azure(prompt, max_tokens, temperature)
else:
raise ValueError(f"Unsupported provider: {self.provider}")
def _generate_openai(self, prompt: str, max_tokens: int, temperature: float) -> str:
url = 'https://api.openai.com/v1/chat/completions'
if self.base_url:
url = f'{self.base_url}/chat/completions'
payload = {
'model': self.model,
'messages': [{'role': 'user', 'content': prompt}],
'max_tokens': max_tokens,
'temperature': temperature
}
response = self.session.post(url, json=payload)
response.raise_for_status()
data = response.json()
return data['choices'][0]['message']['content']
def _generate_anthropic(self, prompt: str, max_tokens: int, temperature: float) -> str:
url = 'https://api.anthropic.com/v1/messages'
self.session.headers.update({
'x-api-key': self.api_key,
'anthropic-version': '2023-06-01',
'content-type': 'application/json'
})
payload = {
'model': self.model,
'messages': [{'role': 'user', 'content': prompt}],
'max_tokens': max_tokens,
'temperature': temperature
}
response = self.session.post(url, json=payload)
response.raise_for_status()
data = response.json()
return data['content'][0]['text']
def _generate_azure(self, prompt: str, max_tokens: int, temperature: float) -> str:
if not self.base_url:
raise ValueError("Azure OpenAI requires base_url configuration")
url = f'{self.base_url}/openai/deployments/{self.model}/chat/completions?api-version=2023-05-15'
self.session.headers.update({'api-key': self.api_key})
payload = {
'messages': [{'role': 'user', 'content': prompt}],
'max_tokens': max_tokens,
'temperature': temperature
}
response = self.session.post(url, json=payload)
response.raise_for_status()
data = response.json()
return data['choices'][0]['message']['content']
class LLMClient:
def __init__(self, model_config: Dict, max_retries: int = 3):
self.model_config = model_config
self.max_retries = max_retries
self.backend = self._initialize_backend()
def generate(self, prompt: str, max_tokens: int = 500, temperature: float = 0.7) -> str:
for attempt in range(self.max_retries):
try:
response = self.backend.generate(
prompt=prompt,
max_tokens=max_tokens,
temperature=temperature
)
if not response or len(response.strip()) == 0:
raise ValueError("Empty response from LLM")
return response
except Exception as e:
if attempt == self.max_retries - 1:
raise RuntimeError(f"LLM generation failed after {self.max_retries} attempts: {str(e)}")
time.sleep(2 ** attempt)
def _initialize_backend(self):
if self.model_config['type'] == 'remote':
return RemoteLLMBackend(self.model_config)
elif self.model_config['type'] == 'local':
return LocalLLMBackend(self.model_config)
else:
raise ValueError(f"Unknown LLM backend type: {self.model_config['type']}")
class DurationEstimator:
def __init__(self, llm_client: LLMClient, historical_database=None):
self.llm_client = llm_client
self.historical_db = historical_database
self.category_baselines = {
"document_writing": 120,
"meeting": 60,
"email_composition": 15,
"phone_call": 30,
"code_review": 90,
"presentation_preparation": 180,
"data_analysis": 150,
"report_review": 45,
"general": 60
}
def estimate_duration(self, task_description: str, task_category: Optional[str], context: str) -> int:
historical_estimate = self._get_historical_estimate(task_description, task_category)
llm_estimate = self._get_llm_estimate(task_description, context)
baseline_estimate = self.category_baselines.get(task_category or 'general', 60)
estimates = [e for e in [historical_estimate, llm_estimate, baseline_estimate] if e is not None]
if len(estimates) == 0:
return 60
elif len(estimates) == 1:
return estimates[0]
else:
weights = self._calculate_confidence_weights(historical_estimate, llm_estimate, baseline_estimate)
weighted_sum = sum(e * w for e, w in zip(estimates, weights))
total_weight = sum(weights)
return int(weighted_sum / total_weight)
def _get_historical_estimate(self, description: str, category: Optional[str]) -> Optional[int]:
if not self.historical_db:
return None
similar_tasks = self.historical_db.find_similar_tasks(description, category, limit=5)
if not similar_tasks:
return None
durations = [task.actual_duration_minutes for task in similar_tasks]
return int(sum(durations) / len(durations))
def _get_llm_estimate(self, description: str, context: str) -> Optional[int]:
prompt = f"""Estimate how many minutes the following task will take to complete: {description}
Context: {context}
Consider the complexity of the task, typical time requirements for similar activities, and any contextual factors that might affect duration. Provide only a number representing minutes, nothing else."""
try:
response = self.llm_client.generate(prompt, max_tokens=10, temperature=0.3)
return int(response.strip())
except (ValueError, RuntimeError):
return None
def _calculate_confidence_weights(self, historical: Optional[int], llm: Optional[int], baseline: Optional[int]) -> List[float]:
weights = []
if historical is not None:
weights.append(0.5)
if llm is not None:
weights.append(0.3)
if baseline is not None:
weights.append(0.2)
return weights
def generate_duration_estimate_prompt(self, task_description: str, task_category: str) -> Tuple[int, str]:
estimated_minutes = self.category_baselines.get(task_category, 60)
prompt = f"""Based on the task description "{task_description}", I estimate this will take approximately {estimated_minutes} minutes to complete. This is a rough estimate based on similar activities. Does this seem reasonable, or would you like to adjust this estimate? Please provide your preferred duration in minutes if you'd like to change it."""
return estimated_minutes, prompt
class CalendarIntegration:
def __init__(self, calendar_service, timezone: str):
self.service = calendar_service
self.timezone = timezone
def get_busy_periods(self, start_datetime: datetime, end_datetime: datetime) -> List[Dict]:
try:
events = self.service.events().list(
calendarId='primary',
timeMin=start_datetime.isoformat(),
timeMax=end_datetime.isoformat(),
singleEvents=True,
orderBy='startTime'
).execute()
busy_periods = []
for event in events.get('items', []):
if event.get('transparency') == 'transparent':
continue
start = self._parse_datetime(event['start'])
end = self._parse_datetime(event['end'])
busy_periods.append({
'start': start,
'end': end,
'title': event.get('summary', 'Busy'),
'is_flexible': event.get('status') == 'tentative'
})
return busy_periods
except Exception as e:
print(f"Error retrieving calendar events: {e}")
return []
def _parse_datetime(self, datetime_dict: Dict) -> datetime:
if 'dateTime' in datetime_dict:
dt_string = datetime_dict['dateTime']
else:
dt_string = datetime_dict['date'] + 'T00:00:00'
return datetime.fromisoformat(dt_string.replace('Z', '+00:00'))
class CalendarWriter:
def __init__(self, calendar_service, timezone: str):
self.service = calendar_service
self.timezone = timezone
def write_scheduled_tasks(self, scheduled_tasks: List[Dict]) -> Tuple[List, List]:
created_events = []
failed_tasks = []
for item in scheduled_tasks:
try:
event = self._create_event_for_task(item)
created_event = self.service.events().insert(
calendarId='primary',
body=event
).execute()
created_events.append(created_event)
except Exception as e:
failed_tasks.append({
'task': item['task'],
'error': str(e)
})
return created_events, failed_tasks
def _create_event_for_task(self, scheduled_item: Dict) -> Dict:
task = scheduled_item['task']
start = scheduled_item['start']
end = scheduled_item['end']
event = {
'summary': task.description,
'description': f"Estimated duration: {task.estimated_duration_minutes} minutes\nScheduled by AI Time Manager",
'start': {
'dateTime': start.isoformat(),
'timeZone': self.timezone
},
'end': {
'dateTime': end.isoformat(),
'timeZone': self.timezone
},
'reminders': {
'useDefault': False,
'overrides': [
{'method': 'popup', 'minutes': 5},
{'method': 'popup', 'minutes': 0}
]
}
}
if task.deadline:
event['description'] += f"\nDeadline: {task.deadline.strftime('%Y-%m-%d %H:%M')}"
return event
class SchedulingEngine:
def __init__(self, tasks: List[Task], busy_periods: List[Dict], work_hours_start: time, work_hours_end: time):
self.tasks = tasks
self.busy_periods = sorted(busy_periods, key=lambda p: p['start'])
self.work_hours_start = work_hours_start
self.work_hours_end = work_hours_end
self.scheduled_tasks = []
def generate_schedule(self, start_date: datetime, end_date: datetime) -> List[Dict]:
dependency_analyzer = DependencyAnalyzer(self.tasks)
is_valid, error = dependency_analyzer.validate_dependencies()
if not is_valid:
raise ValueError(f"Cannot create schedule: {error}")
task_order = dependency_analyzer.get_topological_order()
current_time = self._get_next_work_slot(start_date)
for task_id in task_order:
task = next(t for t in self.tasks if t.task_id == task_id)
slot = self._find_slot_for_task(task, current_time, end_date)
if slot is None:
raise ValueError(f"Cannot find suitable time slot for task: {task.description}")
self.scheduled_tasks.append({
'task': task,
'start': slot['start'],
'end': slot['end']
})
current_time = self._add_break_after_task(slot['end'])
return self.scheduled_tasks
def _find_slot_for_task(self, task: Task, earliest_start: datetime, latest_end: datetime) -> Optional[Dict]:
duration = timedelta(minutes=task.estimated_duration_minutes)
current = earliest_start
deadline = task.deadline if task.deadline else latest_end
while current + duration <= min(deadline, latest_end):
if self._is_slot_available(current, current + duration):
return {'start': current, 'end': current + duration}
current = self._get_next_work_slot(current + timedelta(minutes=15))
return None
def _is_slot_available(self, start: datetime, end: datetime) -> bool:
if start.time() < self.work_hours_start or end.time() > self.work_hours_end:
return False
for period in self.busy_periods:
if start < period['end'] and end > period['start']:
return False
for scheduled in self.scheduled_tasks:
if start < scheduled['end'] and end > scheduled['start']:
return False
return True
def _add_break_after_task(self, task_end_time: datetime) -> datetime:
minutes_past_hour = task_end_time.minute
if minutes_past_hour >= 55:
break_end = task_end_time.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1, minutes=5)
else:
break_end = task_end_time + timedelta(minutes=5)
if task_end_time.hour == 12 or (task_end_time.hour == 11 and task_end_time.minute >= 30):
break_end = task_end_time.replace(hour=13, minute=0, second=0, microsecond=0)
return break_end
def _get_next_work_slot(self, from_time: datetime) -> datetime:
if from_time.time() < self.work_hours_start:
return from_time.replace(hour=self.work_hours_start.hour, minute=self.work_hours_start.minute, second=0, microsecond=0)
elif from_time.time() >= self.work_hours_end:
next_day = from_time + timedelta(days=1)
return next_day.replace(hour=self.work_hours_start.hour, minute=self.work_hours_start.minute, second=0, microsecond=0)
else:
return from_time
class SchedulePresenter:
def __init__(self, scheduled_tasks: List[Dict], llm_client: LLMClient):
self.scheduled_tasks = sorted(scheduled_tasks, key=lambda x: x['start'])
self.llm_client = llm_client
def generate_summary(self) -> str:
schedule_text = self._format_schedule_text()
prompt = f"""Here is a schedule of tasks:
{schedule_text}
Please provide a friendly, conversational summary of this schedule. Highlight any important points such as tight deadlines, periods of intensive work, or good opportunities for focused work. Keep the summary concise but informative."""
try:
summary = self.llm_client.generate(prompt, max_tokens=300, temperature=0.7)
return summary
except RuntimeError:
return "I've created a schedule for your tasks. Please review the timeline below."
def _format_schedule_text(self) -> str:
lines = []
current_date = None
for item in self.scheduled_tasks:
task_date = item['start'].date()
if task_date != current_date:
current_date = task_date
lines.append(f"\n{current_date.strftime('%A, %B %d, %Y')}")
start_time = item['start'].strftime('%I:%M %p')
end_time = item['end'].strftime('%I:%M %p')
duration = int((item['end'] - item['start']).total_seconds() / 60)
lines.append(f" {start_time} - {end_time} ({duration} min): {item['task'].description}")
return '\n'.join(lines)
def get_formatted_schedule(self) -> str:
return self._format_schedule_text()
class TimeManagerController:
def __init__(self, llm_client: LLMClient, calendar_integration: CalendarIntegration, work_hours_start: time, work_hours_end: time, timezone: str):
self.llm = llm_client
self.calendar = calendar_integration
self.work_hours_start = work_hours_start
self.work_hours_end = work_hours_end
self.timezone = timezone
self.tasks = []
self.conversation_history = []
self.last_schedule = None
self.state = 'initial'
def process_user_input(self, user_message: str) -> str:
self.conversation_history.append({'role': 'user', 'content': user_message})
if self.state == 'initial':
return self._handle_initial_input(user_message)
elif self.state == 'collecting_info':
return self._handle_clarification_response(user_message)
elif self.state == 'schedule_ready':
return self._handle_calendar_decision(user_message)
else:
return "I'm not sure what to do next. Could you please start over with your task list?"
def _handle_initial_input(self, message: str) -> str:
self.tasks = self._parse_initial_tasks(message)
if not self.tasks:
return "I couldn't identify any tasks in your message. Could you please list the tasks you need to complete?"
missing_info = self._identify_missing_information()
if missing_info:
self.state = 'collecting_info'
clarification = self._generate_clarification_question(missing_info)
self.conversation_history.append({'role': 'assistant', 'content': clarification})
return clarification
else:
return self._generate_and_present_schedule()
def _handle_clarification_response(self, message: str) -> str:
self._update_tasks_from_response(message)
missing_info = self._identify_missing_information()
if missing_info:
clarification = self._generate_clarification_question(missing_info)
self.conversation_history.append({'role': 'assistant', 'content': clarification})
return clarification
else:
return self._generate_and_present_schedule()
def _handle_calendar_decision(self, message: str) -> str:
message_lower = message.lower()
if 'yes' in message_lower or 'add' in message_lower or 'commit' in message_lower:
result = self.commit_to_calendar()
self.state = 'complete'
return result
else:
self.state = 'complete'
return "Understood. I won't add these tasks to your calendar. You can use the schedule as a reference for planning your time."
def _parse_initial_tasks(self, message: str) -> List[Task]:
prompt = f"""Parse the following task list and extract individual tasks with their details:
{message}
For each task, identify:
- Description (required)
- Deadline (if mentioned, in ISO format or relative like 'Friday', 'tomorrow')
- Estimated duration in minutes (if mentioned)
- Dependencies on other tasks by index (if mentioned)
Return ONLY a valid JSON array of tasks with these fields. Example: [{{"description": "Write report", "deadline": "Friday", "duration": 120, "dependencies": []}}]"""
try:
response = self.llm.generate(prompt, max_tokens=1000, temperature=0.3)
json_start = response.find('[')
json_end = response.rfind(']') + 1
if json_start != -1 and json_end > json_start:
json_str = response[json_start:json_end]
tasks_data = json.loads(json_str)
else:
tasks_data = json.loads(response)
tasks = []
for i, task_data in enumerate(tasks_data):
task = Task(task_id=f"task_{i}", description=task_data.get('description', ''))
if 'deadline' in task_data and task_data['deadline']:
task.deadline = self._parse_deadline(task_data['deadline'])
if 'duration' in task_data and task_data['duration']:
task.estimated_duration_minutes = int(task_data['duration'])
if 'dependencies' in task_data and task_data['dependencies']:
task.dependencies = [f"task_{d}" for d in task_data['dependencies']]
tasks.append(task)
return tasks
except (json.JSONDecodeError, KeyError, ValueError) as e:
print(f"Error parsing tasks: {e}")
return []
def _identify_missing_information(self) -> List[Dict]:
missing = []
for task in self.tasks:
if not task.is_complete_specification():
missing.append({
'task': task,
'fields': task.get_missing_fields()
})
return missing
def _generate_clarification_question(self, missing_info: List[Dict]) -> str:
task_info = missing_info[0]
task = task_info['task']
fields = task_info['fields']
if 'deadline' in fields:
return f"When do you need to complete '{task.description}'? Please provide a date or relative time like 'Friday' or 'end of week'."
elif 'estimated_duration' in fields:
estimator = DurationEstimator(self.llm, None)
estimated_minutes, question = estimator.generate_duration_estimate_prompt(
task.description,
task.category or 'general'
)
task.estimated_duration_minutes = estimated_minutes
return question
else:
return "I need more information about your tasks. Could you provide details?"
def _update_tasks_from_response(self, message: str):
missing_info = self._identify_missing_information()
if not missing_info:
return
task_info = missing_info[0]
task = task_info['task']
fields = task_info['fields']
if 'deadline' in fields:
task.deadline = self._parse_deadline(message)
elif 'estimated_duration' in fields:
try:
duration = int(''.join(filter(str.isdigit, message)))
if duration > 0:
task.estimated_duration_minutes = duration
task.user_confirmed_duration = True
except ValueError:
pass
def _generate_and_present_schedule(self) -> str:
try:
now = datetime.now(pytz.timezone(self.timezone))
end_date = now + timedelta(days=7)
for task in self.tasks:
if task.deadline and task.deadline > end_date:
end_date = task.deadline + timedelta(days=1)
busy_periods = self.calendar.get_busy_periods(now, end_date)
scheduler = SchedulingEngine(
self.tasks,
busy_periods,
self.work_hours_start,
self.work_hours_end
)
self.last_schedule = scheduler.generate_schedule(now, end_date)
presenter = SchedulePresenter(self.last_schedule, self.llm)
formatted = presenter.get_formatted_schedule()
summary = presenter.generate_summary()
self.state = 'schedule_ready'
response = f"{summary}\n\n{formatted}\n\nWould you like me to add these tasks to your calendar?"
self.conversation_history.append({'role': 'assistant', 'content': response})
return response
except ValueError as e:
self.state = 'initial'
return f"I encountered an issue creating your schedule: {str(e)}. Could you please adjust your tasks or deadlines?"
def commit_to_calendar(self) -> str:
if not self.last_schedule:
return "No schedule to commit. Please generate a schedule first."
writer = CalendarWriter(self.calendar.service, self.timezone)
created, failed = writer.write_scheduled_tasks(self.last_schedule)
if failed:
failure_msg = "\n".join([f"- {f['task'].description}: {f['error']}" for f in failed])
return f"Added {len(created)} tasks to calendar. Failed to add {len(failed)} tasks:\n{failure_msg}"
else:
return f"Successfully added all {len(created)} tasks to your calendar!"
def _parse_deadline(self, deadline_str: str) -> datetime:
now = datetime.now(pytz.timezone(self.timezone))
deadline_str = deadline_str.lower().strip()
if deadline_str == 'today':
return now.replace(hour=23, minute=59, second=59, microsecond=0)
elif deadline_str == 'tomorrow':
return (now + timedelta(days=1)).replace(hour=23, minute=59, second=59, microsecond=0)
elif 'friday' in deadline_str or deadline_str.startswith('fri'):
days_ahead = 4 - now.weekday()
if days_ahead <= 0:
days_ahead += 7
return (now + timedelta(days=days_ahead)).replace(hour=23, minute=59, second=59, microsecond=0)
elif 'monday' in deadline_str or deadline_str.startswith('mon'):
days_ahead = 0 - now.weekday()
if days_ahead <= 0:
days_ahead += 7
return (now + timedelta(days=days_ahead)).replace(hour=23, minute=59, second=59, microsecond=0)
elif 'tuesday' in deadline_str or deadline_str.startswith('tue'):
days_ahead = 1 - now.weekday()
if days_ahead <= 0:
days_ahead += 7
return (now + timedelta(days=days_ahead)).replace(hour=23, minute=59, second=59, microsecond=0)
elif 'wednesday' in deadline_str or deadline_str.startswith('wed'):
days_ahead = 2 - now.weekday()
if days_ahead <= 0:
days_ahead += 7
return (now + timedelta(days=days_ahead)).replace(hour=23, minute=59, second=59, microsecond=0)
elif 'thursday' in deadline_str or deadline_str.startswith('thu'):
days_ahead = 3 - now.weekday()
if days_ahead <= 0:
days_ahead += 7
return (now + timedelta(days=days_ahead)).replace(hour=23, minute=59, second=59, microsecond=0)
elif 'saturday' in deadline_str or deadline_str.startswith('sat'):
days_ahead = 5 - now.weekday()
if days_ahead <= 0:
days_ahead += 7
return (now + timedelta(days=days_ahead)).replace(hour=23, minute=59, second=59, microsecond=0)
elif 'sunday' in deadline_str or deadline_str.startswith('sun'):
days_ahead = 6 - now.weekday()
if days_ahead <= 0:
days_ahead += 7
return (now + timedelta(days=days_ahead)).replace(hour=23, minute=59, second=59, microsecond=0)
elif 'end of week' in deadline_str or 'eow' in deadline_str:
days_ahead = 4 - now.weekday()
if days_ahead <= 0:
days_ahead += 7
return (now + timedelta(days=days_ahead)).replace(hour=17, minute=0, second=0, microsecond=0)
elif 'end of month' in deadline_str or 'eom' in deadline_str:
next_month = now.replace(day=28) + timedelta(days=4)
last_day = next_month - timedelta(days=next_month.day)
return last_day.replace(hour=23, minute=59, second=59, microsecond=0)
else:
try:
return datetime.fromisoformat(deadline_str)
except ValueError:
return now + timedelta(days=7)
def create_time_manager(config: Dict) -> TimeManagerController:
llm_config = config['llm']
llm_client = LLMClient(llm_config)
calendar_service = config.get('calendar_service')
timezone = config.get('timezone', 'UTC')
calendar_integration = CalendarIntegration(calendar_service, timezone)
work_hours_start = time(hour=9, minute=0)
work_hours_end = time(hour=17, minute=0)
if 'work_hours' in config:
work_hours_start = time(hour=config['work_hours']['start_hour'], minute=0)
work_hours_end = time(hour=config['work_hours']['end_hour'], minute=0)
return TimeManagerController(
llm_client,
calendar_integration,
work_hours_start,
work_hours_end,
timezone
)
def main():
config = {
'llm': {
'type': 'remote',
'provider': 'openai',
'api_key': 'your-api-key-here',
'model': 'gpt-4'
},
'timezone': 'America/New_York',
'work_hours': {
'start_hour': 9,
'end_hour': 17
}
}
time_manager = create_time_manager(config)
print("AI Time Manager")
print("=" * 50)
print("Tell me about the tasks you need to complete.\n")
while True:
user_input = input("You: ").strip()
if not user_input:
continue
if user_input.lower() in ['quit', 'exit', 'bye']:
print("Goodbye!")
break
response = time_manager.process_user_input(user_input)
print(f"\nAssistant: {response}\n")
if time_manager.state == 'complete':
print("Would you like to schedule more tasks? (yes/no)")
continue_input = input("You: ").strip().lower()
if 'yes' in continue_input:
time_manager.state = 'initial'
time_manager.tasks = []
time_manager.last_schedule = None
print("\nGreat! Tell me about your next set of tasks.\n")
else:
print("Thank you for using AI Time Manager!")
break
if __name__ == "__main__":
main()
This complete implementation provides a production-ready LLM-powered time manager that handles all the features discussed throughout this article. It supports both local and remote LLM deployments, integrates with calendar systems, analyzes task dependencies, estimates durations, generates optimized schedules with automatic break management, and provides a natural conversational interface. The code includes comprehensive error handling, supports multiple GPU architectures through the abstraction layer, and implements all the constraint satisfaction and scheduling algorithms described in the preceding sections.
No comments:
Post a Comment