COURSE · OPS · A018
OpenCLaw: Building a Personal AI Agent
Learn openclaw: building a personal ai agent through RunLocalAI's practical lens: personal agent, memory, tools and scheduling, hardware fit, runtime settings, verification habits and local-vs-cloud tradeoffs.
PREREQUISITES
- I002
- B016
Why this course matters
OpenCLaw: Building a Personal AI Agent is for operators making local AI reliable, measurable and cheaper to run. It connects personal agent, memory, tools, scheduling and plugins to the questions RunLocalAI wants every reader to answer before they install, upgrade or scale a model: will it run, what will it cost in memory, what setting changes the result, and how do you verify the answer instead of trusting a demo?
What you will be able to do
By the end, you should be able to explain the main tradeoffs in plain language, choose a safe next experiment, and use the chapter exercises as a repeatable operator checklist. The course favors local evidence, hardware fit, context limits, latency and failure modes over generic AI vocabulary.
How to use this course
Start at chapter one if the topic is new. If you already have a working stack, scan for chapters such as OpenCLaw Vision, Always-On Architecture, Background Service and Persistent Memory and use those lessons as a quality-control pass before changing a workstation, team workflow or production-like local deployment.
CHAPTERS
- 01OpenCLaw VisionOpenCLaw provides a framework for building personal AI agents that operate continuously, maintaining context across interactions through layered memory systems. OpenCLaw represents an architectural approach to building autonomous AI agents that run persistently on local hardware. Unlike session-based AI interactions, an OpenCLaw agent maintains continuous operation, accumulating knowledge and context over time. This persistent operation enables the agent to build a coherent understanding of the user's preferences, projects, and workflows. The core philosophy centers on three pillars: continuous availability, persistent memory, and extensible tool access. A personal AI agent should be available whenever needed, remember previous interactions and learned information, and have the ability to interact with external systems through tools. Architecture Overview OpenCLaw agents consist of several interconnected components working together. The agent core processes inputs and generates responses. Memory systems store and retrieve information across different time scales. Tool runners execute external actions. The scheduler manages time-based tasks. A plugin system allows extensibility without core modifications. The agent loop follows a predictable pattern: receive input, retrieve relevant context from memory, reason about appropriate actions, execute necessary tools, update memory with results, and generate output. This loop repeats continuously while the agent runs. Typical Use Cases Personal AI agents built with OpenCLaw serve various purposes. They can monitor email and calendar events, automatically categorizing and summarizing incoming information. Development assistance includes code review, documentation generation, and bug investigation. Research support involves gathering information, organizing findings, and maintaining project notes. Personal organization encompasses task management, habit tracking, and schedule optimization. The agent maintains a user profile that evolves over time. Initial setup captures basic preferences and priorities. Ongoing interaction refines understanding of communication style, work patterns, and contextual requirements.10 min
- 02Always-On ArchitectureAlways-on operation requires careful handling of resource usage, state management, and graceful degradation under varying system conditions. An always-on agent must coexist with other system operations without consuming excessive resources. CPU usage should remain minimal during idle periods. Memory consumption must stay bounded as the agent accumulates information over months or years of operation. Disk usage requires monitoring to prevent unbounded growth. The architecture separates concerns into distinct processes. The agent process handles reasoning and response generation. Worker processes manage tool execution in isolation. The memory service provides data access without blocking the main loop. This separation prevents tool execution failures from affecting agent stability. Graceful Degradation System resources are finite, and an always-on agent must handle resource constraints gracefully. Under high load, the agent prioritizes essential functions over background tasks. Memory pressure triggers aggressive cleanup of transient caches. Extended inactivity activates power-saving modes. The agent maintains health metrics and logs operational status. Monitoring these metrics helps identify degradation before failures occur. Automatic restart mechanisms recover from unexpected crashes without manual intervention. State Persistence Continuous operation means the agent must persist state across process restarts and system reboots. Critical state includes conversation history, learned preferences, scheduled tasks, and tool configurations. The persistence layer writes state to durable storage at regular intervals and during significant events. Crash recovery reconstructs the agent's operational state from persisted data. The reconstruction process validates data integrity and recovers to a consistent point. Some recent state may be lost during recovery, requiring graceful acknowledgment of incomplete information.10 min
- 03Background ServiceRunning an agent as a background service requires systemd integration, proper user context, and secure logging infrastructure. The agent runs as a systemd service to ensure reliable startup, automatic restart on failure, and proper system integration. Service configuration defines resource limits, user context, and dependency relationships with other system services. Service Configuration The systemd unit file specifies how the service runs and how systemd manages it. The `[Unit]` section declares dependencies and documentation. The `[Service]` section defines the execution environment, restart behavior, and resource constraints. The `[Install]` section specifies how the service connects to system boot targets. ```ini [Unit] Description=OpenCLaw Personal AI Agent After=network-online.target Wants=network-online.target Documentation=https://runlocalai.co/learn [Service] Type=simple User=openclaw Group=openclaw WorkingDirectory=/var/lib/openclaw Environment=OPENCLAW_HOME=/var/lib/openclaw ExecStart=/usr/local/bin/openclaw-agent serve Restart=on-failure RestartSec=10 StandardOutput=journal StandardError=journal WatchdogSec=60 MemoryMax=512M CPUQuota=50% [Install] WantedBy=multi-user.target ``` User Context and Security The agent runs under a dedicated user account with limited permissions. This isolation prevents security vulnerabilities in the agent from affecting other system components. File permissions restrict access to sensitive data. Network access follows the principle of least privilege. Logging goes to the systemd journal rather than plain files. Journalctl provides filtering by severity, time range, and other criteria. Log rotation happens automatically through journald configuration.15 min
- 04Persistent MemoryMemory systems must balance retrieval accuracy, storage efficiency, and access latency across multiple tiers of storage with different characteristics. Personal agents accumulate information over extended periods. Without organized memory, the agent cannot effectively leverage past interactions. The memory system solves this by providing structured storage with retrieval mechanisms optimized for different access patterns. Memory architecture follows a tiered approach. Each tier offers different tradeoffs between capacity, speed, and durability. The agent chooses the appropriate tier based on the information type and expected access patterns. Memory Tiers Overview Short-term memory stores recent conversations and temporary working data. This tier prioritizes low-latency access over long-term durability. Information in short-term memory expires after a configurable period or when capacity limits are reached. Medium-term memory stores important facts learned during interactions. This tier persists across sessions but may be compressed or archived during low-activity periods. Retrieval targets higher accuracy than short-term memory. Long-term memory stores the complete history of the agent's operation. This tier prioritizes capacity and durability over access speed. Retrieval may require longer response times but supports complex queries across large datasets. The tiers work together through a memory consolidation process. Important information graduates from short-term to medium-term to long-term storage. Unimportant information is discarded at appropriate intervals.10 min
- 05SQLite for Short-TermSQLite provides ACID-compliant storage with minimal overhead, making it ideal for short-term memory where reliability matters more than concurrent write performance. SQLite handles short-term memory storage efficiently for single-process agents. The database provides transactional guarantees without requiring a separate server process. Embedded operation eliminates network latency and simplifies deployment. Schema Design Short-term memory stores conversation turns, working context, and transient state. The schema reflects these use cases with tables optimized for recent data access patterns. ```sql CREATE TABLE conversations ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT NOT NULL, started_at DATETIME DEFAULT CURRENT_TIMESTAMP, last_activity DATETIME DEFAULT CURRENT_TIMESTAMP, context_summary TEXT, is_active BOOLEAN DEFAULT TRUE ); CREATE TABLE messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, conversation_id INTEGER NOT NULL, role TEXT NOT NULL CHECK(role IN ('user', 'assistant', 'system')), content TEXT NOT NULL, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, metadata JSON, FOREIGN KEY (conversation_id) REFERENCES conversations(id) ); CREATE INDEX idx_messages_conversation_time ON messages(conversation_id, created_at DESC); CREATE TABLE working_memory ( key TEXT PRIMARY KEY, value TEXT NOT NULL, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, expires_at DATETIME, access_count INTEGER DEFAULT 0 ); ``` Failure Handling SQLite handles many failure scenarios automatically through its journal mechanism. Incomplete transactions roll back cleanly on restart. Corrupted databases can often be recovered from WAL (Write-Ahead Logging) archives. However, SQLite is not immune to all failures. Disk full conditions require proactive monitoring. Database locks can block operations during high-concurrency scenarios. Large databases may experience performance degradation without regular maintenance. Maintenance Operations Regular maintenance keeps the database healthy. The VACUUM command reclaims deleted space and rebuilds indexes. The ANALYZE command updates query planning statistics. These operations should run during low-activity periods. ```sql -- Clean up expired working memory DELETE FROM working_memory WHERE expires_at IS NOT NULL AND expires_at < CURRENT_TIMESTAMP; -- Archive old conversations INSERT INTO conversations_archive SELECT * FROM conversations WHERE last_activity < datetime('now', '-30 days'); -- Reclaim space VACUUM; ```15 min
- 06Vector Store for Long-TermVector embeddings enable semantic similarity search across large memory stores, allowing the agent to retrieve relevant past information without exact keyword matches. Long-term memory must support complex queries over accumulated knowledge. A user asking about "the Python project we worked on last month" expects the agent to identify relevant conversations without explicit project naming. Vector similarity search solves this problem by representing memories as points in a high-dimensional space. Embedding Generation Each memory entry receives a vector embedding that captures its semantic meaning. The embedding model converts text into a fixed-length numeric vector. Similar concepts produce vectors with small angular distances. This property enables retrieval through cosine similarity or dot product comparisons. ```python import numpy as np from openclaw.embeddings import get_embedding_model class VectorStore: def __init__(self, dimension: int, metric: str = "cosine"): self.dimension = dimension self.metric = metric self.vectors: dict[str, np.ndarray] = {} self.embedding_model = get_embedding_model() def add(self, memory_id: str, text: str, metadata: dict) -> None: """Add a memory with its embedding.""" embedding = self.embedding_model.encode(text) self.vectors[memory_id] = embedding # Store metadata and text in SQLite for retrieval self._store_record(memory_id, text, metadata) def search(self, query: str, top_k: int = 10) -> list[dict]: """Find the most similar memories to the query.""" query_embedding = self.embedding_model.encode(query) similarities = [] for memory_id, vector in self.vectors.items(): similarity = self._compute_similarity(query_embedding, vector) similarities.append((memory_id, similarity)) # Return top-k results similarities.sort(key=lambda x: x[1], reverse=True) return [ {**self._get_record(memory_id), "score": score} for memory_id, score in similarities[:top_k] ] def _compute_similarity( self, a: np.ndarray, b: np.ndarray ) -> float: if self.metric == "cosine": return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)) return -np.sum((a - b) ** 2) # Euclidean distance ``` Indexing for Scale Naive vector comparison scales poorly with memory size. Production systems use approximate nearest neighbor (ANN) indexes that sacrifice some accuracy for dramatically improved performance. Popular options include FAISS, Annoy, and HNSW. The index must be rebuilt when memories are added or deleted. Incremental updates avoid full rebuilds when possible. Memory constraints may require index partitioning across multiple files.15 min
- 07Memory ConsolidationMemory consolidation transfers important information between storage tiers while pruning irrelevant data, maintaining a memory system that stays useful without unbounded growth. Consolidation is the process that keeps the memory system healthy over time. Without consolidation, short-term memory grows unbounded while long-term memory accumulates noise. The consolidation strategy defines when and how memories move between tiers. Consolidation Triggers Consolidation runs on multiple schedules. Immediate consolidation happens when the agent explicitly learns something important. Periodic consolidation runs on a schedule to handle routine memory movement. Emergency consolidation triggers when storage approaches capacity limits. ```python import asyncio from datetime import datetime, timedelta from dataclasses import dataclass from enum import Enum class MemoryTier(Enum): SHORT_TERM = "short_term" MEDIUM_TERM = "medium_term" LONG_TERM = "long_term" @dataclass class ConsolidationTask: memory_id: str source_tier: MemoryTier target_tier: MemoryTier reason: str scheduled_at: datetime async def consolidation_loop(agent: "OpenCLawAgent"): """Background task that manages memory consolidation.""" while True: try: # Check for immediate consolidation requests await process_immediate_tasks(agent) # Run periodic consolidation await run_periodic_consolidation(agent) # Check capacity and consolidate if needed await check_and_emergency_consolidate(agent) except Exception as e: agent.logger.error(f"Consolidation error: {e}") await asyncio.sleep(300) # Check every 5 minutes async def run_periodic_consolidation(agent: "OpenCLawAgent"): """Consolidate memories on a scheduled basis.""" cutoff = datetime.now() - timedelta(days=7) # Get short-term memories older than cutoff candidates = await agent.memory.get_short_term_older_than(cutoff) for memory in candidates: importance = await assess_importance(agent, memory) if importance > 0.7: await promote_memory(agent, memory, MemoryTier.MEDIUM_TERM) elif importance > 0.3: await promote_memory(agent, memory, MemoryTier.LONG_TERM) else: await prune_memory(agent, memory) ``` Importance Assessment Determining which memories to keep requires assessing their importance. The assessment considers multiple factors: frequency of reference, recency of access, explicit user feedback, and relationship to other important memories. User feedback provides the strongest signal. When users indicate a memory was useful, its importance increases. When they correct the agent's understanding, the old incorrect memory may need revision or removal.15 min
- 08Tool IntegrationTools extend the agent's capabilities beyond reasoning, enabling it to interact with external systems through a well-defined interface with error handling and result processing. The agent's core capability is reasoning, but reasoning alone cannot accomplish practical tasks. Tool integration provides the means to take action in the world. Each tool implements a specific capability with standardized inputs, outputs, and error handling. Tool Interface Design All tools follow the same interface pattern. This consistency simplifies tool development and enables generic tool handling in the agent core. The interface specifies input validation, execution, and result formatting. ```python from abc import ABC, abstractmethod from typing import Any, TypedDict from datetime import datetime class ToolResult(TypedDict): success: bool data: Any | None error: str | None execution_time_ms: int class BaseTool(ABC): name: str description: str parameters: dict # JSON Schema for parameters @abstractmethod async def execute(self, **kwargs) -> ToolResult: """Execute the tool with given parameters.""" pass def validate_parameters(self, params: dict) -> tuple[bool, str | None]: """Validate parameters against the schema.""" # Implementation uses jsonschema or similar pass async def run(self, params: dict) -> ToolResult: """Run with timing and error handling wrapper.""" start = datetime.now() valid, error = self.validate_parameters(params) if not valid: return ToolResult( success=False, data=None, error=f"Invalid parameters: {error}", execution_time_ms=0 ) try: data = await self.execute(**params) elapsed = (datetime.now() - start).total_seconds() * 1000 return ToolResult( success=True, data=data, error=None, execution_time_ms=int(elapsed) ) except Exception as e: elapsed = (datetime.now() - start).total_seconds() * 1000 return ToolResult( success=False, data=None, error=str(e), execution_time_ms=int(elapsed) ) ``` Tool Categories Tools fall into several categories based on their purpose. Data retrieval tools query external information sources. Modification tools change state in external systems. Notification tools send information to users or other systems. Utility tools perform computations or transformations. A typical agent deployment includes tools from multiple categories. The specific selection depends on the agent's intended use cases and the systems it needs to access.15 min
- 09Plugin SystemA plugin system allows extending the agent without modifying core code, providing isolation between custom functionality and the agent framework. As requirements evolve, the agent needs new capabilities. Modifying core code for every enhancement creates maintenance burden and risks breaking existing functionality. A plugin system provides a standardized extension mechanism that isolates custom code from the core framework. Plugin Architecture Plugins are self-contained packages that provide tools, memory handlers, or other extensions. The plugin system loads plugins at startup, validates their interfaces, and integrates them into the agent's operation. Plugins can declare dependencies on other plugins or system features. ```python from pathlib import Path import importlib import sys from typing import Protocol, runtime_checkable @runtime_checkable class Plugin(Protocol): """Protocol defining the plugin interface.""" name: str version: str async def initialize(self, agent: "OpenCLawAgent") -> None: """Initialize the plugin with agent context.""" ... async def shutdown(self) -> None: """Clean up resources on shutdown.""" ... class PluginManager: def __init__(self, plugin_dir: Path): self.plugin_dir = plugin_dir self.plugins: dict[str, Plugin] = {} self._load_plugins() def _load_plugins(self) -> None: """Discover and load plugins from the plugin directory.""" for item in self.plugin_dir.iterdir(): if not item.is_dir(): continue # Check for plugin marker file if not (item / "plugin.json").exists(): continue try: # Import the plugin module sys.path.insert(0, str(item)) module = importlib.import_module("plugin") # Verify it implements the Plugin protocol if isinstance(module, Plugin): self.plugins[module.name] = module else: raise TypeError("Plugin does not implement Plugin protocol") except Exception as e: print(f"Failed to load plugin from {item}: {e}") async def initialize_all(self, agent: "OpenCLawAgent") -> None: """Initialize all loaded plugins.""" for name, plugin in self.plugins.items(): try: await plugin.initialize(agent) except Exception as e: raise RuntimeError(f"Plugin {name} initialization failed: {e}") ``` Plugin Lifecycle Plugins move through a defined lifecycle. Discovery happens at startup when the plugin manager scans directories. Loading imports the plugin code and validates the interface. Initialization provides the agent context and allows plugins to register their capabilities. Active operation continues until shutdown. Cleanup releases resources and saves state.15 min
- 10Plugin APIThe Plugin API exposes agent capabilities to plugins through a well-defined interface, enabling rich integration without coupling plugins to internal implementation details. Plugins need access to agent capabilities to be useful. The Plugin API provides this access through a stable interface. Changes to internal implementation do not break plugins as long as the API remains compatible. Core API Components The API provides access to several agent subsystems. The tool registry allows plugins to register new tools. The memory system enables plugins to store and retrieve information. The event bus lets plugins subscribe to agent events. The configuration interface provides access to settings. ```python class AgentAPI: """Public API available to plugins.""" def __init__(self, agent: "OpenCLawAgent"): self._agent = agent @property def tools(self) -> "ToolRegistry": """Access the tool registry to register or lookup tools.""" return self._agent._tool_registry @property def memory(self) -> "MemorySystem": """Access the memory system for storage and retrieval.""" return self._agent._memory @property def events(self) -> "EventBus": """Subscribe to agent events.""" return self._agent._event_bus @property def config(self) -> "Config": """Access agent configuration.""" return self._agent._config async def send_message(self, content: str, role: str = "user") -> str: """Send a message to the agent and get the response.""" response = await self._agent.process_message(content, role) return response.content def get_user_context(self) -> dict: """Retrieve known information about the user.""" return self._agent._user_profile.snapshot() ``` Event System Plugins often need to react to agent events. The event system provides publish-subscribe messaging. Plugins subscribe to specific event types and receive notifications when those events occur. ```python class EventBus: def __init__(self): self._subscribers: dict[str, list[callable]] = {} def subscribe(self, event_type: str, handler: callable) -> None: """Subscribe to an event type.""" if event_type not in self._subscribers: self._subscribers[event_type] = [] self._subscribers[event_type].append(handler) async def publish(self, event_type: str, data: dict) -> None: """Publish an event to all subscribers.""" handlers = self._subscribers.get(event_type, []) for handler in handlers: try: if asyncio.iscoroutinefunction(handler): await handler(data) else: handler(data) except Exception as e: logging.error(f"Event handler error: {e}") # Example event types EVENT_MESSAGE_RECEIVED = "message.received" EVENT_TOOL_EXECUTED = "tool.executed" EVENT_MEMORY_CONSOLIDATED = "memory.consolidated" EVENT_USER_PRESENT = "user.present" ```15 min
- 11Cron SchedulingCron expressions provide a human-readable way to define recurring schedules, enabling the agent to run tasks at specific times without continuous polling. Scheduled tasks require a mechanism to determine when they should run. Cron expressions provide a standard format for describing time-based schedules. The agent's scheduler evaluates cron expressions to determine the next run time for each task. Cron Expression Format A cron expression consists of five fields separated by spaces: minute, hour, day of month, month, and day of week. Each field accepts specific values, ranges, and special characters. ``` ┌───────────── minute (0-59) │ ┌───────────── hour (0-23) │ │ ┌───────────── day of month (1-31) │ │ │ ┌───────────── month (1-12) │ │ │ │ ┌───────────── day of week (0-6, Sunday=0) │ │ │ │ │ * * * * * ``` Common patterns include: `0 9 * * 1-5` for 9 AM on weekdays, `*/15 * * * *` for every 15 minutes, and `0 0 1 * *` for midnight on the first of each month. Cron Parser Implementation ```python import re from datetime import datetime, timedelta from typing import Callable, Awaitable class CronField: """Parse and match a single cron field.""" def __init__(self, value: str, min_val: int, max_val: int): self.value = value self.min_val = min_val self.max_val = max_val self._parse() def _parse(self) -> None: """Parse the field into a set of matching values.""" self.values: set[int] = set() self._ranges: list[tuple[int, int]] = [] for part in self.value.split(","): if "/" in part: # Handle step values: */15 or 1-30/5 range_part, step = part.split("/") step = int(step) if range_part == "*": start, end = self.min_val, self.max_val else: start, end = self._parse_range(range_part) for v in range(start, end + 1, step): self.values.add(v) elif "-" in part: # Handle ranges: 1-5 start, end = self._parse_range(part) self._ranges.append((start, end)) for v in range(start, end + 1): self.values.add(v) elif part == "*": # All values for v in range(self.min_val, self.max_val + 1): self.values.add(v) else: # Single value self.values.add(int(part)) def matches(self, value: int) -> bool: """Check if a value matches this field.""" if value in self.values: return True return any(start <= value <= end for start, end in self._ranges) class CronSchedule: """Parse and evaluate cron expressions.""" def __init__(self, expression: str): parts = expression.split() if len(parts) != 5: raise ValueError(f"Invalid cron expression: {expression}") self.minute = CronField(parts[0], 0, 59) self.hour = CronField(parts[1], 0, 23) self.day_of_month = CronField(parts[2], 1, 31) self.month = CronField(parts[3], 1, 12) self.day_of_week = CronField(parts[4], 0, 6) def next_run(self, after: datetime) -> datetime: """Calculate the next run time after the given datetime.""" # Implementation iterates forward minute by minute # until all fields match current = after.replace(second=0, microsecond=0) + timedelta(minutes=1) max_attempts = 366 * 24 * 60 # Max 1 year ahead for _ in range(max_attempts): if self._matches(current): return current current += timedelta(minutes=1) raise ValueError("No matching time found within one year") def _matches(self, dt: datetime) -> bool: """Check if a datetime matches the schedule.""" return ( self.minute.matches(dt.minute) and self.hour.matches(dt.hour) and self.day_of_month.matches(dt.day) and self.month.matches(dt.month) and self.day_of_week.matches(dt.weekday()) ) ```20 min
- 12Scheduled TasksScheduled tasks integrate with the agent's task system, enabling background execution with proper state management, error handling, and result reporting. The scheduler manages tasks that run at specified times or intervals. Each scheduled task includes the task logic, schedule definition, and configuration for retries, timeouts, and notifications. Scheduled Task Structure ```python from dataclasses import dataclass, field from enum import Enum from typing import Callable, Awaitable, Any import asyncio import uuid class TaskStatus(Enum): PENDING = "pending" RUNNING = "running" COMPLETED = "completed" FAILED = "failed" CANCELLED = "cancelled" @dataclass class ScheduledTask: """Definition of a scheduled task.""" id: str = field(default_factory=lambda: str(uuid.uuid4())) name: str = "" description: str = "" schedule: CronSchedule handler: Callable[..., Awaitable[Any]] args: tuple = field(default_factory=tuple) kwargs: dict = field(default_factory=dict) # Execution configuration max_retries: int = 3 retry_delay_seconds: int = 60 timeout_seconds: int = 300 # State status: TaskStatus = TaskStatus.PENDING last_run: datetime | None = None next_run: datetime | None = None last_error: str | None = None run_count: int = 0 class TaskScheduler: """Manages scheduled task execution.""" def __init__(self, agent: "OpenCLawAgent"): self.agent = agent self.tasks: dict[str, ScheduledTask] = {} self._running_tasks: set[str] = set() self._should_stop = False def schedule(self, task: ScheduledTask) -> str: """Add a task to the schedule.""" task.next_run = task.schedule.next_run(datetime.now()) self.tasks[task.id] = task return task.id async def run(self) -> None: """Main scheduler loop.""" while not self._should_stop: try: await self._check_and_run_tasks() except Exception as e: self.agent.logger.error(f"Scheduler error: {e}") await asyncio.sleep(10) # Check every 10 seconds async def _check_and_run_tasks(self) -> None: """Check for tasks that need to run and execute them.""" now = datetime.now() for task_id, task in self.tasks.items(): if task.status == TaskStatus.RUNNING: continue if task.next_run and task.next_run <= now: await self._execute_task(task) async def _execute_task(self, task: ScheduledTask) -> None: """Execute a task with error handling and retries.""" task.status = TaskStatus.RUNNING self._running_tasks.add(task.id) for attempt in range(task.max_retries + 1): try: result = await asyncio.wait_for( task.handler(*task.args, **task.kwargs), timeout=task.timeout_seconds ) task.status = TaskStatus.COMPLETED task.last_run = datetime.now() task.next_run = task.schedule.next_run(task.last_run) task.last_error = None task.run_count += 1 break except asyncio.TimeoutError: task.last_error = f"Timeout after {task.timeout_seconds}s" except Exception as e: task.last_error = str(e) if attempt < task.max_retries: await asyncio.sleep(task.retry_delay_seconds) else: task.status = TaskStatus.FAILED self._running_tasks.discard(task.id) ``` Task Persistence Scheduled tasks and their state must persist across agent restarts. The scheduler saves task definitions and state to the database. On restart, it loads tasks and recalculates next run times based on the current time. ```python async def save_state(self) -> None: """Persist scheduler state to database.""" for task in self.tasks.values(): await self.agent.db.execute(""" INSERT OR REPLACE INTO scheduled_tasks (id, name, status, last_run, next_run, last_error, run_count) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( task.id, task.name, task.status.value, task.last_run, task.next_run, task.last_error, task.run_count )) async def load_state(self) -> None: """Restore scheduler state from database.""" rows = await self.agent.db.execute( "SELECT * FROM scheduled_tasks" ) for row in rows: # Reconstruct task and recalculate next_run task = self._reconstruct_task(row) task.next_run = task.schedule.next_run(datetime.now()) self.tasks[task.id] = task ```20 min
- 13Learning from FeedbackFeedback learning requires explicit and implicit channels. Explicit corrections provide clear teaching signals, while implicit acceptance/rejection patterns capture subtler user preferences. Combining both yields reliable adaptation without overfitting to noise.15 min
- 14User Preference LearningUser preferences should be learned probabilistically rather than stored as fixed values. Bayesian estimation handles sparse observations gracefully and provides confidence metrics that prevent over-application of uncertain preferences.20 min
- 15Conversation SummariesConversation summarization must balance compression ratio against information preservation. Hierarchical summarization provides predictable compression while maintaining access to entities, decisions, and user preferences across long interactions.15 min
- 16Notification SystemNotification systems must balance awareness with intrusion. Priority classification enables appropriate channel selection, while batching reduces interruption frequency for non-urgent information without suppressing important updates.20 min
- 17Web InterfaceA web interface for a personal AI agent must balance feature density with usability. Real-time WebSocket updates provide live awareness, while a clean separation between monitoring and configuration views keeps the interface navigable.15 min
- 18Security ModelSecurity models for personal AI agents must handle multiple trust levels. Local users may warrant full access, while network-accessible endpoints require stricter controls. Capability-based authorization enables fine-grained permission management.20 min
- 19Privacy by DesignPrivacy by design requires technical enforcement, not just policy documentation. Automated retention enforcement, purpose tagging, and user-facing controls provide enforceable privacy guarantees rather than theoretical commitments.20 min
- 20Performance OptimizationPerformance optimization on constrained hardware requires measurement-driven approaches. Profiling reveals actual bottlenecks, while caching provides predictable speedups for repeated operations.20 min
- 21Power ManagementPower management for always-on agents must balance availability with energy constraints. Activity scheduling reduces consumption during predictable idle periods, while battery integration provides adaptive behavior on portable devices.20 min
- 22Plugin Development SDKPlugin systems must balance extensibility with security and stability. Sandboxed execution, capability-based permissions, and initialization validation prevent malicious or broken plugins from compromising the host system.20 min
- 23Testing and DebuggingTesting personal AI agents requires special attention to non-deterministic behavior and context-dependent outputs. Structured test fixtures, integration harnesses, and debugging tools help verify correctness and diagnose issues.20 min
- 24OpenCLaw Implementation ProjectBuilding a personal AI agent requires integrating diverse components: LLM interfaces, context management, learning systems, security, and privacy. Each component must work reliably in isolation and cooperate smoothly in production. The project structure and incremental implementation approach ensure a maintainable, extensible system.25 min