基于本地AI的电脑活动记忆系统：原理、实现与隐私优先实践

最近在尝试将 ChatGPT 更深度地融入日常工作流时，发现了一个痛点：每次对话都像是“重启”一次，需要反复向 AI 重复项目背景、代码片段或个人偏好，效率大打折扣。如果 AI 能记住我电脑上的活动，比如正在编辑的文档、浏览的网页或运行的代码，那交互体验将发生质变。这正是“电脑活动记忆”功能试图解决的问题。本文将围绕这一前沿概念，深入探讨其技术原理、实现思路，并提供一个基于本地化、隐私优先的桌面应用原型开发教程。无论你是想为现有应用添加智能记忆层，还是探索下一代人机交互方式，都能从中获得一套完整的、可落地的技术方案。

1. 背景与核心概念：什么是“电脑活动记忆”？

在传统的 ChatGPT 或类似大模型应用中，每一次对话都是独立的。模型不具备跨会话的记忆能力，更无法感知用户在其本地设备上的操作和上下文。所谓“电脑活动记忆”（Computer Activity Memory），指的是一个 AI 助手能够持续地、在用户授权和隐私保护的前提下，观察、记录并理解用户在电脑上的各种活动（如打开的应用程序、浏览的网页、编辑的文档、终端命令等），并利用这些信息来构建一个动态的、个性化的上下文知识库。当用户与 AI 交互时，AI 可以主动引用或关联这些记忆，提供更具连续性、相关性和深度的协助。

核心价值与场景：

• 编程开发：AI 能记住你当前项目的代码结构、最近修改的文件、遇到的编译错误，并在你提问时直接基于这些上下文给出建议。
• 研究与写作：AI 能追踪你阅读的文献、收集的网页资料、正在撰写的文档草稿，帮助你整合信息、生成大纲或润色文字。
• 工作流自动化：AI 通过学习你的日常操作习惯（如每天上午先检查邮件，然后打开某个数据分析工具），可以预测你的需求，甚至自动执行一些例行任务。
• 个性化助手：记住你的偏好设置、常用命令、项目术语，减少重复性描述，使对话更自然、高效。

重要前提：隐私与安全实现此功能的首要原则是“隐私优先，本地处理”。所有对用户活动的监控和数据记录都应在用户设备本地完成，敏感信息不应未经处理就上传至云端。核心的 AI 推理和记忆查询也尽可能在本地进行，或使用安全的、用户可控的 API 方式。本文的实践方案将严格遵守这一原则。

2. 环境准备与版本说明

我们将构建一个跨平台的桌面应用原型，演示如何捕获系统活动并与本地大模型（或经过代理的 API）结合。技术栈选择注重跨平台能力和开发效率。

开发环境：

• 操作系统：Windows 10/11, macOS 12+, 或 Linux (Ubuntu 22.04+)。本文示例命令以 macOS/Linux 为主，Windows 会有相应说明。
• 编程语言：Python 3.9+ (用于数据采集、处理和后端逻辑)
• 桌面应用框架：Flutter 3.13+ (使用dart:ffi或平台通道与 Python 后端通信) 或 PyQt6/PySide6 (纯 Python 方案)。本文将以PyQt6为例，便于快速集成 Python 生态。
• 模型/API：
  • 本地模型方案：使用ollama运行本地大模型（如llama3.2,qwen2.5等）。
  • API 方案：通过配置访问 OpenAI API (或兼容的国产大模型 API)。请注意，使用任何 API 都需自行解决网络访问和账号问题，本文不涉及相关教程。
• 关键 Python 库：
  • psutil: 跨平台获取系统进程信息。
  • pynput/pyautogui: 监控键盘、鼠标和屏幕（需谨慎使用，并明确告知用户）。
  • watchdog: 监控文件系统变化。
  • sqlite3: 本地存储活动记录和记忆向量。
  • chromadb/faiss: 用于记忆的向量存储与检索。
  • openai(可选): 调用 OpenAI API。
  • pyqt6: GUI 框架。

版本说明：本文示例代码基于 Python 3.10 和 PyQt6 编写。由于系统监控和 AI 模型领域更新较快，部分库的 API 可能有细微变化，请根据你的实际环境调整。核心思路具有普适性。

3. 核心原理与技术拆解

实现“电脑活动记忆”功能，可以拆解为四个核心模块：

3.1 活动捕获模块

负责以非侵入、低功耗的方式收集用户活动数据。

• 窗口焦点与标题：获取当前最前端的应用程序窗口及其标题（如 “Visual Studio Code - my_project”）。这能知道用户正在“做什么”。
• 文件访问与编辑：监控特定工作目录的文件变化（创建、修改、删除）。结合编辑器或 IDE 的集成，可以获取更精确的代码上下文。
• 浏览器活动：通过浏览器扩展（如 Chrome/Firefox）或监听特定端口（如 DevTools Protocol）获取当前标签页的 URL 和标题。必须获得用户明确授权。
• 终端命令：在用户使用的 Shell（如 zsh, bash）中通过 hook 方式（如$PROMPT_COMMAND）记录执行的命令及其输出。
• 剪贴板内容：监控剪贴板变化，捕获复制的文本、代码片段。

技术要点：所有捕获操作都应提供开关，允许用户选择监控哪些类型的数据。数据应立即进行匿名化或去敏感化处理（如移除密码、密钥）。

3.2 记忆存储与向量化模块

原始活动日志是时序文本流，需要转化为可查询的“记忆”。

• 结构化存储：使用 SQLite 存储原始活动日志，包含时间戳、活动类型、内容摘要等字段。
• 文本向量化：将活动内容（如“在main.py第 30 行添加了函数calculate_score”）通过嵌入模型（Embedding Model）转换为高维向量。本地可使用sentence-transformers库或ollama的嵌入端点。
• 向量数据库：将向量存入本地向量数据库（如 ChromaDB）。每条向量关联原始日志的 ID，便于回溯。

3.3 记忆检索与上下文构建模块

当用户与 AI 对话时，需要从海量记忆中找出最相关的部分。

• 查询向量化：将用户的当前问题或对话历史转换为向量。
• 相似性检索：在向量数据库中执行相似性搜索，找出与当前查询最相关的 K 条历史记忆。
• 上下文组装：将检索到的记忆条目，按时间或相关性排序，组装成一段连贯的“背景故事”，作为系统提示词的一部分提交给大模型。

3.4 AI 集成与对话模块

将构建好的上下文与用户问题一起发送给 AI，并管理对话历史。

• 提示词工程：设计系统提示词，明确告知 AI 其具备“记忆”能力，并指导它如何利用提供的上下文。例如：“你是一个拥有记忆的 AI 助手。以下是你观察到的用户近期电脑活动摘要：[记忆上下文]。请基于这些信息回答用户的问题。”
• 模型调用：连接本地ollama服务或配置好的 API 端点。
• 对话管理：维护一个有限长度的对话历史，实现多轮对话的连贯性。

4. 完整实战案例：构建一个简易记忆桌面助手

我们将创建一个名为 “MemoAssistant” 的简易桌面应用，它能在后台安静记录活跃窗口标题，并在你提问时，尝试从记忆中寻找相关线索来回答。

4.1 项目结构与依赖安装

创建项目目录并初始化虚拟环境：

mkdir memo_assistant && cd memo_assistant python -m venv venv # Windows: venv\Scripts\activate source venv/bin/activate

安装核心依赖：

pip install psutil pyqt6 chromadb sentence-transformers # 如果使用本地模型，安装 ollama 并拉取模型 # curl -fsSL https://ollama.com/install.sh | sh # ollama pull llama3.2:3b # 拉取一个较小模型

项目结构：

memo_assistant/ ├── main.py # 应用主入口 ├── activity_monitor.py # 活动监控模块 ├── memory_store.py # 记忆存储与检索模块 ├── ai_client.py # AI 客户端模块 ├── ui_main_window.py # 主窗口 UI └── database/ └── memories.db # SQLite 数据库文件

4.2 核心模块实现

1. 活动监控模块 (activity_monitor.py)此模块负责周期性获取当前活动窗口信息。

import psutil import time import logging from datetime import datetime # 注意：跨平台获取窗口标题需要额外库，此处以 macOS 的 `AppKit` 和 Linux 的 `ewmh` 为例，Windows 可用 `pygetwindow` # 为简化，本例仅捕获进程名。实际项目需根据平台实现。 class ActivityMonitor: def __init__(self, callback): """ 初始化监控器。 :param callback: 当检测到活动变化时的回调函数，接收活动数据字典。 """ self.callback = callback self.last_activity = None self.is_monitoring = False logging.basicConfig(level=logging.INFO) def get_current_activity(self): """获取当前活动摘要（简化版：仅进程名）""" try: # 获取前台进程（跨平台精确获取需平台特定代码） # 这里使用 psutil 遍历进程，找出可能的前台应用（启发式） for proc in psutil.process_iter(['pid', 'name', 'create_time']): try: # 一个非常简单的判断：最近有用户输入的进程 # 真实场景应使用平台API（如Windows的GetForegroundWindow） if proc.info['name'] and 'bash' not in proc.info['name'].lower(): activity = { 'timestamp': datetime.now().isoformat(), 'type': 'window_focus', 'process_name': proc.info['name'], 'title': 'N/A' # 实际需要平台API获取窗口标题 } return activity except (psutil.NoSuchProcess, psutil.AccessDenied): continue except Exception as e: logging.error(f"获取活动失败: {e}") return None def start_monitoring(self, interval=5): """开始监控，每 interval 秒检查一次""" self.is_monitoring = True logging.info("活动监控已启动") while self.is_monitoring: current_act = self.get_current_activity() if current_act and current_act != self.last_activity: self.last_activity = current_act self.callback(current_act) # 将活动数据传递给回调函数 logging.info(f"活动变更: {current_act['process_name']}") time.sleep(interval) def stop_monitoring(self): """停止监控""" self.is_monitoring = False logging.info("活动监控已停止") # 示例回调函数 def on_activity_detected(activity_data): print(f"[活动记录] {activity_data['timestamp']} - {activity_data['process_name']}") # 这里可以将 activity_data 传递给 memory_store 进行存储

2. 记忆存储与检索模块 (memory_store.py)此模块负责将活动存储到 SQLite，并利用向量数据库进行检索。

import sqlite3 import json from datetime import datetime import chromadb from chromadb.config import Settings from sentence_transformers import SentenceTransformer import logging logging.basicConfig(level=logging.INFO) class MemoryStore: def __init__(self, db_path='database/memories.db', chroma_persist_dir='./chroma_db'): self.db_path = db_path self._init_sqlite() # 初始化向量数据库客户端 self.chroma_client = chromadb.PersistentClient(path=chroma_persist_dir, settings=Settings(anonymized_telemetry=False)) self.collection = self.chroma_client.get_or_create_collection(name="activity_memories") # 加载嵌入模型（本地，首次运行会下载） self.embedder = SentenceTransformer('all-MiniLM-L6-v2') # 轻量级句子转换模型 logging.info("记忆存储模块初始化完成") def _init_sqlite(self): """初始化 SQLite 数据库表""" conn = sqlite3.connect(self.db_path) c = conn.cursor() c.execute(''' CREATE TABLE IF NOT EXISTS activities (id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT NOT NULL, type TEXT NOT NULL, content TEXT NOT NULL, raw_data TEXT) ''') conn.commit() conn.close() def add_activity(self, activity_type, content, raw_data=None): """添加一条活动记录到 SQLite 和向量数据库""" timestamp = datetime.now().isoformat() # 1. 存入 SQLite conn = sqlite3.connect(self.db_path) c = conn.cursor() c.execute("INSERT INTO activities (timestamp, type, content, raw_data) VALUES (?, ?, ?, ?)", (timestamp, activity_type, content, json.dumps(raw_data) if raw_data else None)) activity_id = c.lastrowid conn.commit() conn.close() # 2. 生成向量并存入 ChromaDB # 将内容和类型组合成文本进行向量化 text_to_embed = f"{activity_type}: {content}" embedding = self.embedder.encode(text_to_embed).tolist() # 使用 SQLite 的 rowid 作为 ChromaDB 的 id，确保关联 self.collection.add( embeddings=[embedding], metadatas=[{"type": activity_type, "timestamp": timestamp, "sqlite_id": activity_id}], documents=[content], # 存储原始文本用于返回 ids=[str(activity_id)] ) logging.info(f"记忆已存储: ID={activity_id}, 内容={content[:50]}...") return activity_id def search_memories(self, query_text, n_results=3): """根据查询文本搜索相关记忆""" # 将查询文本向量化 query_embedding = self.embedder.encode(query_text).tolist() # 在向量数据库中搜索 results = self.collection.query( query_embeddings=[query_embedding], n_results=n_results ) # 组织返回结果 memories = [] if results['documents']: for i in range(len(results['documents'][0])): memory = { 'content': results['documents'][0][i], 'metadata': results['metadatas'][0][i], 'distance': results['distances'][0][i] } memories.append(memory) return memories def get_recent_activities(self, limit=10): """从 SQLite 获取最近的活动记录""" conn = sqlite3.connect(self.db_path) c = conn.cursor() c.execute("SELECT timestamp, type, content FROM activities ORDER BY timestamp DESC LIMIT ?", (limit,)) rows = c.fetchall() conn.close() return [{'timestamp': r[0], 'type': r[1], 'content': r[2]} for r in rows]

3. AI 客户端模块 (ai_client.py)此模块封装与 AI 模型的交互。这里提供本地ollama和 OpenAI API 两种示例。

import requests import json import logging from typing import List, Dict logging.basicConfig(level=logging.INFO) class AIClient: def __init__(self, mode='local', base_url='http://localhost:11434', api_key=None, model='llama3.2'): """ 初始化 AI 客户端。 :param mode: 'local' 或 'openai' :param base_url: ollama 服务地址或 OpenAI 兼容 API 地址 :param api_key: OpenAI API Key (mode='openai' 时必需) :param model: 模型名称 """ self.mode = mode self.base_url = base_url self.api_key = api_key self.model = model self.headers = {"Content-Type": "application/json"} if mode == 'openai' and api_key: self.headers["Authorization"] = f"Bearer {api_key}" def generate_response(self, system_prompt: str, user_prompt: str, conversation_history: List[Dict] = None) -> str: """生成 AI 回复""" messages = [{"role": "system", "content": system_prompt}] if conversation_history: messages.extend(conversation_history) messages.append({"role": "user", "content": user_prompt}) if self.mode == 'local': # 调用本地 ollama payload = { "model": self.model, "messages": messages, "stream": False } try: response = requests.post(f"{self.base_url}/api/chat", json=payload, headers=self.headers) response.raise_for_status() return response.json()['message']['content'] except Exception as e: logging.error(f"调用本地模型失败: {e}") return f"错误: 无法连接本地AI服务。请确保 ollama 已运行且模型 {self.model} 已安装。" elif self.mode == 'openai': # 调用 OpenAI 兼容 API payload = { "model": self.model, "messages": messages, "temperature": 0.7 } try: # 注意：实际地址可能是 https://api.openai.com/v1/chat/completions 或其他代理地址 api_endpoint = f"{self.base_url}/v1/chat/completions" response = requests.post(api_endpoint, json=payload, headers=self.headers) response.raise_for_status() return response.json()['choices'][0]['message']['content'] except Exception as e: logging.error(f"调用 API 失败: {e}") return "错误: AI 服务调用失败，请检查网络和配置。" else: return "错误: AI 客户端模式配置不正确。" # 示例：使用本地 ollama # ai = AIClient(mode='local', model='llama3.2') # 示例：使用 OpenAI API (需自行解决访问问题) # ai = AIClient(mode='openai', base_url='https://api.openai.com/v1', api_key='your-sk-...', model='gpt-3.5-turbo')

4.3 用户界面与主程序集成 (ui_main_window.py和main.py)

主窗口 UI (ui_main_window.py):

import sys from PyQt6.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, QTextEdit, QLineEdit, QPushButton, QLabel, QListWidget) from PyQt6.QtCore import QThread, pyqtSignal from activity_monitor import ActivityMonitor from memory_store import MemoryStore from ai_client import AIClient class MonitorThread(QThread): """将活动监控放在单独线程，避免阻塞UI""" activity_signal = pyqtSignal(dict) def __init__(self, memory_store): super().__init__() self.memory_store = memory_store self.monitor = ActivityMonitor(self.on_activity) def on_activity(self, activity_data): # 当监控器检测到活动时，存储并发射信号更新UI content = f"{activity_data['process_name']} (标题: {activity_data.get('title', 'N/A')})" self.memory_store.add_activity(activity_data['type'], content, activity_data) self.activity_signal.emit(activity_data) def run(self): self.monitor.start_monitoring(interval=10) # 每10秒检查一次 def stop(self): self.monitor.stop_monitoring() class MainWindow(QMainWindow): def __init__(self): super().__init__() self.memory_store = MemoryStore() self.ai_client = AIClient(mode='local') # 默认使用本地 ollama self.conversation_history = [] self.init_ui() self.start_monitoring() def init_ui(self): self.setWindowTitle('MemoAssistant - 记忆助手') self.setGeometry(300, 300, 800, 600) central_widget = QWidget() self.setCentralWidget(central_widget) layout = QVBoxLayout() # 记忆显示区域 self.memory_list = QListWidget() self.memory_list.setMaximumHeight(150) layout.addWidget(QLabel("近期活动记忆:")) layout.addWidget(self.memory_list) # 对话显示区域 self.chat_display = QTextEdit() self.chat_display.setReadOnly(True) layout.addWidget(QLabel("与 AI 对话:")) layout.addWidget(self.chat_display) # 输入区域 self.input_field = QLineEdit() self.input_field.setPlaceholderText("输入你的问题... (例如：我刚才在用哪个软件？)") self.input_field.returnPressed.connect(self.send_message) layout.addWidget(self.input_field) # 按钮 btn_send = QPushButton("发送") btn_send.clicked.connect(self.send_message) btn_clear = QPushButton("清空对话") btn_clear.clicked.connect(self.clear_chat) layout.addWidget(btn_send) layout.addWidget(btn_clear) central_widget.setLayout(layout) self.update_memory_list() def start_monitoring(self): self.monitor_thread = MonitorThread(self.memory_store) self.monitor_thread.activity_signal.connect(self.on_new_activity) self.monitor_thread.start() def on_new_activity(self, activity_data): # 在UI线程中更新记忆列表 self.update_memory_list() def update_memory_list(self): self.memory_list.clear() recent = self.memory_store.get_recent_activities(limit=5) for act in recent: self.memory_list.addItem(f"[{act['timestamp'][11:19]}] {act['content']}") def send_message(self): user_input = self.input_field.text().strip() if not user_input: return # 显示用户输入 self.chat_display.append(f"\n**你:** {user_input}") self.input_field.clear() # 1. 检索相关记忆 relevant_memories = self.memory_store.search_memories(user_input, n_results=2) memory_context = "无相关记忆。" if relevant_memories: memory_context = "基于你的电脑活动，我注意到：\n" for mem in relevant_memories: memory_context += f"- {mem['content']} (时间: {mem['metadata']['timestamp'][11:19]})\n" # 2. 构建系统提示词 system_prompt = f"""你是一个智能桌面助手，能够访问用户电脑活动的记忆。 以下是与当前问题可能相关的近期活动记录： {memory_context} 请基于这些记忆（如果相关）和你的通用知识来回答用户的问题。如果记忆不相关，请忽略它，直接回答问题。 回答要简洁、有帮助。""" # 3. 调用 AI self.chat_display.append("**AI:** 思考中...") QApplication.processEvents() # 让UI更新“思考中”提示 response = self.ai_client.generate_response(system_prompt, user_input, self.conversation_history) # 4. 更新对话历史并显示回复 self.conversation_history.append({"role": "user", "content": user_input}) self.conversation_history.append({"role": "assistant", "content": response}) # 保持历史长度，避免过长 if len(self.conversation_history) > 10: self.conversation_history = self.conversation_history[-10:] self.chat_display.append(f"**助手:** {response}") def clear_chat(self): self.chat_display.clear() self.conversation_history = [] self.chat_display.append("对话已清空。") def closeEvent(self, event): self.monitor_thread.stop() self.monitor_thread.wait() event.accept() def main(): app = QApplication(sys.argv) window = MainWindow() window.show() sys.exit(app.exec()) if __name__ == '__main__': main()

应用主入口 (main.py):

from ui_main_window import main if __name__ == "__main__": main()

4.4 运行与验证

1. 确保环境就绪：已安装所有 Python 依赖，如果使用本地ollama，请确保已安装并运行了ollama serve，且已拉取模型（如ollama pull llama3.2:3b）。
2. 启动应用：在项目根目录下运行：

   python main.py

3. 界面操作：
   • 应用启动后，后台线程会自动开始记录你的活动窗口（进程名）。
   • “近期活动记忆”列表会显示捕获到的活动。
   • 在下方输入框提问，例如：“我刚才在用什么软件？” 或 “我今天下午都做了什么？”。AI 会尝试从记忆库中检索相关信息并回答。
4. 观察日志：控制台会输出活动捕获和记忆存储的日志信息。

4.5 结果说明

运行成功后，你将看到一个简单的桌面窗口。它演示了“电脑活动记忆”的核心闭环：

1. 捕获：后台默默记录你的前台应用。
2. 存储与索引：活动被存入 SQLite 并向量化后存入 ChromaDB。
3. 检索：当你提问时，问题被向量化，并从记忆中找出最相关的几条。
4. 增强推理：相关记忆作为上下文被注入给 AI 模型，使回答更具个性化。
5. 交互：通过图形界面完成对话。

这只是一个原型，实际产品级应用需要更精细的活动捕获、强大的隐私控制、更复杂的记忆组织（如时间线、主题聚类）以及更稳定的本地模型集成。

5. 常见问题与排查思路

6. 最佳实践与工程建议

将“电脑活动记忆”功能产品化，需要远超原型的工程化考量。

1. 隐私与安全第一

   • 明确告知与授权：应用首次启动必须有清晰的隐私协议和权限申请，让用户选择监控哪些活动（如文件、浏览器、键盘）。
   • 数据本地化：所有原始数据必须加密存储在用户本地。向量化过程也应在本地完成。
   • 敏感信息过滤：在数据捕获层实现实时过滤器，自动剔除密码、密钥、身份证号等敏感模式的内容。
   • 数据生命周期：提供设置允许用户自动删除超过一定时间的记忆，或一键清除所有数据。

2. 系统活动捕获的健壮性

   • 跨平台抽象层：为 Windows、macOS、Linux 分别实现原生、高效的活动捕获接口，并封装成统一的 API。
   • 性能与功耗：监控频率要合理（如非活跃时降低频率），使用高效的事件驱动机制（如文件系统事件watchdog）而非轮询。
   • 错误恢复：监控进程崩溃后应能自动重启，并记录错误日志。

3. 记忆系统的智能性

   • 记忆分层：区分短期工作记忆（当前会话）和长期知识记忆（跨会话），采用不同的存储和检索策略。
   • 记忆摘要与压缩：定期对高频、重复的活动进行自动摘要，避免向量数据库膨胀。
   • 关联与图谱：不仅存储孤立事件，还尝试建立事件之间的关联（如“编辑文件A”后“运行了测试B”），构建知识图谱。
   • 主动学习：允许用户对 AI 的回答进行反馈（“这个记忆有用/无用”），利用反馈优化检索和提示词。

4. AI 集成的稳定性与成本

   • 降级策略：当本地大模型不可用或 API 调用失败时，应有优雅降级方案（如仅展示原始记忆，不进行智能回答）。
   • 上下文长度管理：大模型有上下文窗口限制，需要智能截断或总结过长的记忆上下文。
   • 本地模型优化：探索量化、硬件加速（如 GPU， MPS）来提升本地模型的推理速度。
   • 成本控制：如果使用按 token 计费的云 API，需要对输入上下文进行压缩和优化，避免不必要的 token 消耗。

5. 用户体验与可控性

   • 记忆可视化：提供时间线或图谱视图，让用户直观查看、搜索和管理自己的记忆。
   • 记忆编辑与删除：用户必须能查看、修正或删除任何一条错误的或不愿保留的记忆。
   • 情景模式：允许用户创建不同的“情景”（如“工作模式”、“学习模式”），在不同模式下关注不同类型的活动。

开发此类功能是一个持续迭代的过程，应从最小可行原型（MVP）开始，紧密围绕用户真实需求，逐步增加功能和优化体验，同时始终将用户隐私和数据主权放在核心位置。