Build a Personal AI Assistant: Complete Python Project with Memory and Tools

Personal AI Assistant:

Persistence Layer:
  - conversations.db (SQLite): full conversation history
  - memories.json: explicit user facts and preferences
  - notes/: directory of personal notes

Runtime Layer:
  - Current session messages list
  - Loaded system prompt (persona + memories + context)
  - Available tools

Tools:
  - web_search: search current information
  - save_note: create/update notes
  - read_note: read existing notes
  - list_notes: show all notes
  - calculate: precise math
  - get_current_time: current date/time

Flow:
  User input
    → Load relevant memories
    → Build context-aware system prompt
    → Tool call loop (agent decides which tools to use)
    → Stream response
    → Save to history
    → Extract and update memories

# pip install openai sqlite-utils chromadb requests

import sqlite3
import json
from pathlib import Path
from datetime import datetime

class AssistantStorage:
    def __init__(self, data_dir: str = "./assistant_data"):
        self.data_dir = Path(data_dir)
        self.data_dir.mkdir(exist_ok=True)
        self.notes_dir = self.data_dir / "notes"
        self.notes_dir.mkdir(exist_ok=True)
        
        self.db_path = self.data_dir / "conversations.db"
        self.memories_path = self.data_dir / "memories.json"
        self.preferences_path = self.data_dir / "preferences.json"
        
        self._init_db()
        self._init_memories()
    
    def _init_db(self):
        conn = sqlite3.connect(self.db_path)
        conn.execute("""
            CREATE TABLE IF NOT EXISTS conversations (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                session_id TEXT,
                role TEXT,
                content TEXT,
                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
            )
        """)
        conn.execute("CREATE INDEX IF NOT EXISTS idx_session ON conversations(session_id)")
        conn.commit()
        conn.close()
    
    def _init_memories(self):
        if not self.memories_path.exists():
            self.memories_path.write_text(json.dumps({"facts": [], "preferences": []}))
        if not self.preferences_path.exists():
            self.preferences_path.write_text(json.dumps({
                "response_style": "concise but complete",
                "name": "User",
                "timezone": "UTC"
            }))
    
    def save_message(self, session_id: str, role: str, content: str):
        conn = sqlite3.connect(self.db_path)
        conn.execute(
            "INSERT INTO conversations (session_id, role, content) VALUES (?, ?, ?)",
            (session_id, role, content)
        )
        conn.commit()
        conn.close()
    
    def get_recent_history(self, limit: int = 20) -> list[dict]:
        conn = sqlite3.connect(self.db_path)
        rows = conn.execute("""
            SELECT role, content, timestamp
            FROM conversations
            ORDER BY timestamp DESC
            LIMIT ?
        """, (limit,)).fetchall()
        conn.close()
        
        return [
            {"role": row[0], "content": row[1], "timestamp": row[2]}
            for row in reversed(rows)
        ]
    
    def get_memories(self) -> dict:
        return json.loads(self.memories_path.read_text())
    
    def add_memory(self, fact: str, category: str = "general"):
        memories = self.get_memories()
        memories["facts"].append({
            "fact": fact,
            "category": category,
            "added": datetime.now().isoformat()
        })
        self.memories_path.write_text(json.dumps(memories, indent=2))
    
    def get_preferences(self) -> dict:
        return json.loads(self.preferences_path.read_text())
    
    # Note management
    def save_note(self, title: str, content: str) -> str:
        filename = title.lower().replace(" ", "_").replace("/", "_") + ".md"
        note_path = self.notes_dir / filename
        note_path.write_text(f"# {title}\n\n{content}\n\nLast updated: {datetime.now().strftime('%Y-%m-%d %H:%M')}")
        return f"Note saved: {filename}"
    
    def read_note(self, title: str) -> str:
        filename = title.lower().replace(" ", "_") + ".md"
        note_path = self.notes_dir / filename
        if note_path.exists():
            return note_path.read_text()
        return f"Note not found: {title}"
    
    def list_notes(self) -> list[str]:
        return [p.stem.replace("_", " ") for p in self.notes_dir.glob("*.md")]

import requests
from datetime import datetime

def create_tools(storage: AssistantStorage) -> list[dict]:
    return [
        {
            "type": "function",
            "function": {
                "name": "web_search",
                "description": "Search the web for current information",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {"type": "string", "description": "Search query"}
                    },
                    "required": ["query"]
                }
            }
        },
        {
            "type": "function",
            "function": {
                "name": "save_note",
                "description": "Save a note for later reference",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "title": {"type": "string", "description": "Note title"},
                        "content": {"type": "string", "description": "Note content in markdown"}
                    },
                    "required": ["title", "content"]
                }
            }
        },
        {
            "type": "function",
            "function": {
                "name": "read_note",
                "description": "Read an existing note by title",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "title": {"type": "string", "description": "Note title to read"}
                    },
                    "required": ["title"]
                }
            }
        },
        {
            "type": "function",
            "function": {
                "name": "list_notes",
                "description": "List all saved notes",
                "parameters": {"type": "object", "properties": {}}
            }
        },
        {
            "type": "function",
            "function": {
                "name": "get_current_time",
                "description": "Get the current date and time",
                "parameters": {"type": "object", "properties": {}}
            }
        },
        {
            "type": "function",
            "function": {
                "name": "remember_fact",
                "description": "Store an important fact about the user for future reference",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "fact": {"type": "string", "description": "The fact to remember"},
                        "category": {
                            "type": "string",
                            "enum": ["preference", "project", "contact", "schedule", "general"]
                        }
                    },
                    "required": ["fact"]
                }
            }
        }
    ]

def execute_tool(tool_name: str, arguments: dict, storage: AssistantStorage) -> str:
    """Execute a tool and return the result."""
    
    if tool_name == "web_search":
        query = arguments["query"]
        # Using DuckDuckGo's instant answer API (no key required)
        try:
            response = requests.get(
                "https://api.duckduckgo.com/",
                params={"q": query, "format": "json", "no_html": "1"},
                timeout=5
            )
            data = response.json()
            if data.get("Abstract"):
                return f"Search result: {data['Abstract']}\nSource: {data['AbstractURL']}"
            return f"Search for '{query}' completed. No instant answer available — please provide context from your knowledge."
        except Exception as e:
            return f"Search error: {e}. Try again or answer from your training data."
    
    elif tool_name == "save_note":
        return storage.save_note(arguments["title"], arguments["content"])
    
    elif tool_name == "read_note":
        return storage.read_note(arguments["title"])
    
    elif tool_name == "list_notes":
        notes = storage.list_notes()
        if notes:
            return f"Your notes: {', '.join(notes)}"
        return "No notes saved yet."
    
    elif tool_name == "get_current_time":
        return f"Current date/time: {datetime.now().strftime('%A, %B %d, %Y at %I:%M %p')}"
    
    elif tool_name == "remember_fact":
        storage.add_memory(arguments["fact"], arguments.get("category", "general"))
        return f"Remembered: {arguments['fact']}"
    
    return f"Unknown tool: {tool_name}"

import json
import uuid
from openai import OpenAI

def build_system_prompt(storage: AssistantStorage) -> str:
    prefs = storage.get_preferences()
    memories = storage.get_memories()
    
    memory_text = ""
    if memories["facts"]:
        recent_facts = memories["facts"][-10:]  # Last 10 facts
        memory_text = "\n\nWhat I know about you:\n" + \
            "\n".join(f"- {m['fact']}" for m in recent_facts)
    
    return f"""You are {prefs.get('name', 'User')}'s personal AI assistant.

Your style: {prefs.get('response_style', 'concise but helpful')}
Timezone: {prefs.get('timezone', 'UTC')}
{memory_text}

You have access to tools: web search, note-taking, and memory.
When you learn important things about the user (their projects, preferences, working style),
use the remember_fact tool to store them for future reference.

Be personal, helpful, and remember you're building a relationship with this user over time.
If you need to search for current information, use the web_search tool."""

class PersonalAssistant:
    def __init__(self):
        self.client = OpenAI()
        self.storage = AssistantStorage()
        self.session_id = str(uuid.uuid4())
        self.messages = []
        
    def _load_session(self):
        """Load recent history and build initial context."""
        recent = self.storage.get_recent_history(limit=10)
        for msg in recent:
            if msg["role"] in ["user", "assistant"]:
                self.messages.append({"role": msg["role"], "content": msg["content"]})
    
    def chat(self, user_input: str) -> str:
        self.messages.append({"role": "user", "content": user_input})
        self.storage.save_message(self.session_id, "user", user_input)
        
        tools = create_tools(self.storage)
        system = build_system_prompt(self.storage)
        
        # Agent loop
        while True:
            response = self.client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "system", "content": system}] + self.messages,
                tools=tools,
                tool_choice="auto"
            )
            
            choice = response.choices[0]
            
            if choice.finish_reason == "stop":
                answer = choice.message.content
                self.messages.append({"role": "assistant", "content": answer})
                self.storage.save_message(self.session_id, "assistant", answer)
                return answer
            
            elif choice.finish_reason == "tool_calls":
                self.messages.append(choice.message)
                
                for tool_call in choice.message.tool_calls:
                    args = json.loads(tool_call.function.arguments)
                    result = execute_tool(tool_call.function.name, args, self.storage)
                    
                    self.messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": result
                    })
    
    def run(self):
        self._load_session()
        print("Personal Assistant ready. Type 'quit' to exit.\n")
        print("I can: search the web, take notes, remember things about you.")
        print("-" * 50)
        
        while True:
            user_input = input("\nYou: ").strip()
            if user_input.lower() in ["quit", "exit", "bye"]:
                print("Goodbye!")
                break
            if not user_input:
                continue
            
            print("\nAssistant: ", end="", flush=True)
            response = self.chat(user_input)
            print(response)

if __name__ == "__main__":
    assistant = PersonalAssistant()
    assistant.run()