commit 837c5f66fb204eab7a522ca3a6019e06c3b6dbb5 Author: fyah Date: Sun Apr 26 22:19:40 2026 +0800 feat: nerd-Brain (呆脑子) - 初始提交 - anchor_writer.py: 记忆写入与情感打标 - anchor_search.py: 多维语义检索引擎 - anchor_decay.py: 遗忘曲线 + 归档 + 随机返场 - anchor_vault_sync.py: Vault 索引生成 - README.md: 项目说明文档 diff --git a/README.md b/README.md new file mode 100644 index 0000000..a51843b --- /dev/null +++ b/README.md @@ -0,0 +1,66 @@ +# nerd-Brain (呆脑子) + +> 一个轻量级、全本地化的 AI 长效记忆系统。 + +## 简介 + +nerd-Brain 是一套基于 Obsidian Markdown 笔记库的 AI 记忆引擎。它不依赖 Docker、向量数据库或外部服务，仅用纯 Python 脚本 + 本地 LLM API 实现： + +- **记忆写入**：自动生成摘要、情感坐标、关键词打标 +- **多维检索**：关键词 + 情感 + 时间 + 重要性加权搜索 +- **自然遗忘**：改进版艾宾浩斯遗忘曲线，自动衰减与归档 +- **随机返场**：10% 概率让旧记忆/旧 Idea 从归档中复活 + +## 架构 + +``` +┌─────────────┐ ┌──────────────┐ ┌───────────────┐ +│ anchor_ │────▶│ Obsidian │────▶│ Qwen 3.5 Plus│ +│ writer.py │ │ Vault (MD) │ │ (百炼 API) │ +└─────────────┘ └──────────────┘ └───────────────┘ + │ │ + ▼ ▼ +┌─────────────┐ ┌──────────────┐ +│ anchor_ │ │ anchor_ │ +│ search.py │ │ decay.py │ +│ (多维检索) │ │ (遗忘+返场) │ +└─────────────┘ └──────────────┘ +``` + +## 核心脚本 + +| 脚本 | 功能 | 用法 | +|------|------|------| +| `anchor_writer.py` | 记忆写入与情感打标 | `python3 anchor_writer.py "内容"` | +| `anchor_search.py` | 多维语义检索 | `python3 anchor_search.py "关键词"` | +| `anchor_decay.py` | 遗忘曲线扫描与归档 | `python3 anchor_decay.py` | +| `anchor_vault_sync.py` | Vault 索引生成 | `python3 anchor_vault_sync.py sync` | + +## 配置 + +所有路径和 API 配置在脚本顶部常量区： + +- `VAULT_DIR`: Obsidian 笔记库路径 +- `API_URL`: LLM 接口地址 +- `API_KEY`: 百炼 API 密钥 +- `RESURRECTION_CHANCE`: 返场概率（默认 0.1） +- `DECAY_LAMBDA`: 遗忘速率（默认 0.05） +- `THRESHOLD`: 归档阈值（默认 0.3） + +## 依赖 + +```bash +pip install requests pyyaml +``` + +## 状态 + +- ✅ 记忆写入（含情感标签） +- ✅ 多维检索引擎 +- ✅ 遗忘曲线 + 自动归档 +- ✅ 归档后随机返场（10% 概率） +- 🔄 A100 本地 Qwen 3.6 部署中（待切换） + +## 备份 + +原始脚本备份在 `~/.hermes/anchor_memory_backup/`，随时可回滚。 diff --git a/anchor_decay.py b/anchor_decay.py new file mode 100644 index 0000000..9b7aba2 --- /dev/null +++ b/anchor_decay.py @@ -0,0 +1,196 @@ +#!/usr/bin/env python3 +""" +Anchor Decay Engine +移植 Ombre-Brain 遗忘曲线算法，模拟人类自然遗忘。 +定期扫描记忆文件，计算活跃度得分，自动归档低活跃记忆。 +""" +import os +import math +import yaml +import re +import random +from datetime import datetime, timedelta + +VAULT_DIR = "/Users/fyah/Documents/如梦初醒/memory/test/Anchor-Memory" +ARCHIVE_DIR = os.path.join(VAULT_DIR, "Archive") + +# 遗忘曲线参数（来自 Ombre-Brain） +DECAY_LAMBDA = 0.05 # 衰减速率 +THRESHOLD = 0.3 # 归档阈值 +EMOTION_BASE = 1.0 # 情感基础权重 +AROUSAL_BOOST = 0.8 # 唤醒度加成 +RESURRECTION_CHANCE = 0.1 # 返场概率 (10%)：被遗忘的记忆有几率“诈尸”复活 + +def parse_yaml_frontmatter(content): + """极简 YAML 解析""" + if content.startswith("---"): + parts = content.split("---", 2) + if len(parts) >= 2: + try: + return yaml.safe_load(parts[1]), parts[2] + except: + pass + return {}, content + +def update_yaml_frontmatter(content, metadata): + """更新 YAML 头""" + if content.startswith("---"): + parts = content.split("---", 2) + if len(parts) >= 2: + return f"---\n{yaml.dump(metadata, allow_unicode=True)}---{parts[2]}" + return content + +def calc_time_weight(days_since: float) -> float: + """新鲜度加成：1.0 + e^(-t/36), t 为小时""" + hours = days_since * 24.0 + return 1.0 + 1.0 * math.exp(-hours / 36.0) + +def calculate_score(metadata: dict) -> float: + """计算记忆活跃度得分""" + if not isinstance(metadata, dict): + return 0.0 + + # 钉选/永久记忆不衰减 + if metadata.get("pinned") or metadata.get("type") == "permanent": + return 999.0 + + importance = max(1, min(10, int(metadata.get("importance", 5)))) + activation_count = max(1.0, float(metadata.get("activation_count", 1))) + + # 计算天数 + last_active_str = metadata.get("last_active", metadata.get("date", "")) + try: + # 兼容多种日期格式 + last_active = datetime.strptime(str(last_active_str), "%Y-%m-%d %H:%M") + days_since = max(0.0, (datetime.now() - last_active).total_seconds() / 86400) + except: + days_since = 30.0 + + # 情感权重 + try: + arousal = max(0.0, min(1.0, float(metadata.get("arousal", 0.3)))) + except: + arousal = 0.3 + emotion_weight = EMOTION_BASE + arousal * AROUSAL_BOOST + + # 时间权重 + time_weight = calc_time_weight(days_since) + + # 短期/长期分离 + if days_since <= 3.0: + combined_weight = time_weight * 0.7 + emotion_weight * 0.3 + else: + combined_weight = emotion_weight * 0.7 + time_weight * 0.3 + + # 核心公式 + base_score = ( + importance + * (activation_count ** 0.3) + * math.exp(-DECAY_LAMBDA * days_since) + * combined_weight + ) + + return base_score + +def run_decay(): + """执行遗忘扫描""" + print(f"[*] 开始扫描遗忘曲线: {VAULT_DIR}") + archived_count = 0 + + # 确保归档目录存在 + os.makedirs(ARCHIVE_DIR, exist_ok=True) + + # 遍历所有 .md 文件 + for root, dirs, files in os.walk(VAULT_DIR): + # 跳过归档目录本身 + if "Archive" in root: + continue + + for file in files: + if not file.endswith(".md"): + continue + + filepath = os.path.join(root, file) + try: + with open(filepath, "r", encoding="utf-8") as f: + content = f.read() + + metadata, body = parse_yaml_frontmatter(content) + if not metadata: + continue + + score = calculate_score(metadata) + + # 如果得分低于阈值，且当前状态不是已归档，则归档 + if score < THRESHOLD and metadata.get("status") != "archived": + # 随机返场判定 + if random.random() < RESURRECTION_CHANCE: + print(f" [✨ 返场] {file} (得分: {score:.3f}) - 触发随机复活！") + metadata["status"] = "resurrected" + metadata["last_resurrected"] = datetime.now().strftime("%Y-%m-%d %H:%M") + new_content = update_yaml_frontmatter(content, metadata) + with open(filepath, "w", encoding="utf-8") as f: + f.write(new_content) + else: + # 正常归档逻辑 + metadata["status"] = "archived" + metadata["archived_date"] = datetime.now().strftime("%Y-%m-%d %H:%M") + metadata["decay_score"] = round(score, 3) + metadata["original_path"] = os.path.relpath(filepath, VAULT_DIR) + + new_content = update_yaml_frontmatter(content, metadata) + with open(filepath, "w", encoding="utf-8") as f: + f.write(new_content) + + # 移动文件到归档目录 + dest_path = os.path.join(ARCHIVE_DIR, file) + os.rename(filepath, dest_path) + archived_count += 1 + print(f" [归档] {file} (得分: {score:.3f})") + + except Exception as e: + print(f" [!] 处理失败 {file}: {e}") + + print(f"[+] 遗忘扫描完成。共归档 {archived_count} 个文件。") + print(f"[+] 遗忘扫描完成。共归档 {archived_count} 个文件。") + # --- 第二阶段：扫描归档目录，寻找返场机会 --- + print(f"[*] 扫描归档目录寻找返场机会: {ARCHIVE_DIR}") + resurrected_count = 0 + + if os.path.exists(ARCHIVE_DIR): + for file in os.listdir(ARCHIVE_DIR): + if not file.endswith(".md"): + continue + + filepath = os.path.join(ARCHIVE_DIR, file) + try: + with open(filepath, "r", encoding="utf-8") as f: + content = f.read() + + metadata, body = parse_yaml_frontmatter(content) + + # 只针对已归档的文件 + if metadata.get("status") == "archived": + # 掷骰子：10% 概率复活 + if random.random() < RESURRECTION_CHANCE: + print(f" [✨ 返场] {file} - 从归档中复活！") + metadata["status"] = "resurrected" + metadata["last_resurrected"] = datetime.now().strftime("%Y-%m-%d %H:%M") + + new_content = update_yaml_frontmatter(content, metadata) + with open(filepath, "w", encoding="utf-8") as f: + f.write(new_content) + + # 恢复原路径 + orig_path = metadata.get("original_path", file) + dest_path = os.path.join(VAULT_DIR, orig_path) + os.makedirs(os.path.dirname(dest_path), exist_ok=True) + os.rename(filepath, dest_path) + resurrected_count += 1 + + except Exception as e: + print(f" [!] 处理归档文件 {file} 失败: {e}") + + print(f"[+] 返场扫描完成。共复活 {resurrected_count} 个文件。") +if __name__ == "__main__": + run_decay() diff --git a/anchor_search.py b/anchor_search.py new file mode 100644 index 0000000..86cbe29 --- /dev/null +++ b/anchor_search.py @@ -0,0 +1,180 @@ +#!/usr/bin/env python3 +""" +Anchor Search Engine +多维加权检索引擎。 +结合关键词匹配、情感共鸣、时间接近度、重要性进行排序。 +""" +import os +import math +import yaml +import json +from datetime import datetime + +VAULT_DIR = "/Users/fyah/Documents/如梦初醒/memory/test/Anchor-Memory" + +# 检索权重配置 +W_TOPIC = 4.0 # 主题域/关键词权重 +W_EMOTION = 2.0 # 情感共鸣权重 +W_TIME = 1.5 # 时间接近度权重 +W_IMPORTANCE = 1.0 # 重要性权重 +MAX_RESULTS = 5 # 返回结果数量 + +def parse_yaml_frontmatter(content): + """极简 YAML 解析，兼容无 YAML 头的纯 Markdown""" + if content.startswith("---"): + parts = content.split("---", 2) + if len(parts) >= 2: + try: + return yaml.safe_load(parts[1]), parts[2] + except: + pass + # 如果没有 YAML 头，返回空字典和全文 + # 提取标题作为默认 title + title = "Untitled" + if content.startswith("# "): + title = content.split("\n")[0][2:].strip() + return {"title": title, "status": "active", "importance": 5}, content + +def calc_keyword_score(query: str, metadata: dict, body: str) -> float: + """计算关键词匹配得分""" + q_lower = query.lower() + score = 0.0 + + # 标题匹配 + title = metadata.get("title", "").lower() + if q_lower in title: + score += 3.0 + + # 标签匹配 + tags = metadata.get("tags", []) + if isinstance(tags, list): + for tag in tags: + if q_lower in tag.lower(): + score += 2.0 + + # 内容匹配 + body_lower = body.lower() + if q_lower in body_lower: + score += 1.0 + + # 模糊匹配（简单版：计算重叠词） + query_words = set(q_lower.split()) + body_words = set(body_lower.split()) + overlap = len(query_words & body_words) + if overlap > 0: + score += overlap * 0.5 + + return score + +def calc_emotion_score(query_emotion: dict, metadata: dict) -> float: + """计算情感共鸣得分（基于 Russell 环形模型距离）""" + try: + q_val = query_emotion.get("valence", 0.5) + q_arousal = query_emotion.get("arousal", 0.3) + + m_val = metadata.get("valence", 0.5) + m_arousal = metadata.get("arousal", 0.3) + + # 计算欧氏距离（归一化到 0~1，0 表示完全一致，1 表示完全相反） + distance = math.sqrt((q_val - m_val)**2 + (q_arousal - m_arousal)**2) + max_distance = math.sqrt(1.0**2 + 1.0**2) # 最大距离约 1.414 + + # 距离越近，得分越高 + return max(0.0, 1.0 - (distance / max_distance)) + except: + return 0.0 + +def calc_time_score(metadata: dict) -> float: + """计算时间接近度得分（越近越好）""" + try: + date_str = metadata.get("date", "") + mem_date = datetime.strptime(str(date_str), "%Y-%m-%d %H:%M") + days_diff = (datetime.now() - mem_date).total_seconds() / 86400 + + # 指数衰减：越久远的记忆得分越低 + return math.exp(-0.1 * days_diff) + except: + return 0.0 + +def calc_importance_score(metadata: dict) -> float: + """计算重要性得分（归一化到 0~1）""" + try: + imp = metadata.get("importance", 5) + return float(imp) / 10.0 + except: + return 0.5 + +def search_memories(query: str, emotion_hint: dict = None): + """执行多维检索""" + print(f"[*] 开始检索: '{query}'") + results = [] + + # 遍历所有活跃记忆（跳过归档目录） + for root, dirs, files in os.walk(VAULT_DIR): + if "Archive" in root: + continue + + for file in files: + if not file.endswith(".md"): + continue + + filepath = os.path.join(root, file) + try: + with open(filepath, "r", encoding="utf-8") as f: + content = f.read() + + metadata, body = parse_yaml_frontmatter(content) + if not metadata or metadata.get("status") == "archived": + continue + + # 计算各维度得分 + k_score = calc_keyword_score(query, metadata, body) + e_score = calc_emotion_score(emotion_hint or {}, metadata) + t_score = calc_time_score(metadata) + i_score = calc_importance_score(metadata) + + # 加权总分 + total_score = ( + k_score * W_TOPIC + + e_score * W_EMOTION + + t_score * W_TIME + + i_score * W_IMPORTANCE + ) + + if total_score > 0: + results.append({ + "file": file, + "path": os.path.relpath(filepath, VAULT_DIR), + "title": metadata.get("title", "Untitled"), + "score": total_score, + "metadata": metadata, + "preview": body[:200].strip() + }) + except Exception as e: + print(f" [!] 读取失败 {file}: {e}") + + # 按得分排序 + results.sort(key=lambda x: x["score"], reverse=True) + + # 返回 Top N + top_results = results[:MAX_RESULTS] + + if not top_results: + print(f"[-] 未找到相关记忆。") + return [] + + print(f"[+] 找到 {len(top_results)} 条相关记忆（Top {MAX_RESULTS}）：") + for r in top_results: + print(f"\n📄 {r['title']} (得分: {r['score']:.2f})") + print(f"📍 {r['path']}") + print(f"👁️ {r['preview']}...") + + return top_results + +if __name__ == "__main__": + import sys + if len(sys.argv) > 1: + query = " ".join(sys.argv[1:]) + search_memories(query) + else: + print("用法: python3 anchor_search.py <查询内容>") diff --git a/anchor_vault_sync.py b/anchor_vault_sync.py new file mode 100644 index 0000000..d1ebd5d --- /dev/null +++ b/anchor_vault_sync.py @@ -0,0 +1,146 @@ +#!/usr/bin/env python3 +""" +Anchor Vault Sync & Search +轻量级 Obsidian Vault 索引与检索脚本 +路径: ~/.hermes/anchor_vault_sync.py +""" +import os +import sys +import json +import re +from datetime import datetime +from pathlib import Path + +VAULT_PATH = "/Users/fyah/Documents/如梦初醒/memory/test" +INDEX_PATH = os.path.expanduser("~/.hermes/anchor_vault_index.json") +SKIP_DIRS = {".obsidian", ".trash", ".smart-env", ".space", ".makemd", ".git"} +PREVIEW_LEN = 300 + +def parse_yaml_frontmatter(content): + """极简 YAML 解析，只取 tags 和 title""" + tags = [] + title = "" + if content.startswith("---"): + parts = content.split("---", 2) + if len(parts) >= 2: + yaml_block = parts[1] + for line in yaml_block.splitlines(): + if line.startswith("tags:"): + # 处理 tags: [a, b] 或 tags:\n - a\n - b + rest = line[5:].strip() + if rest.startswith("["): + tags = [t.strip().strip("'\"") for t in rest[1:-1].split(",")] + else: + tags = [] + elif line.startswith(" - "): + if "tags" in locals() and tags is not None: # 简单判断是否在 tags 块下 + tags.append(line.strip().strip("- ").strip("'\"")) + elif line.startswith("title:"): + title = line[6:].strip().strip("'\"") + return tags, title + +def scan_vault(): + """扫描 Vault 并生成索引""" + print(f"[*] 开始扫描: {VAULT_PATH}") + index = [] + count = 0 + + for root, dirs, files in os.walk(VAULT_PATH): + # 过滤隐藏目录 + dirs[:] = [d for d in dirs if d not in SKIP_DIRS and not d.startswith(".")] + + for file in files: + if not file.endswith(".md"): + continue + + filepath = os.path.join(root, file) + rel_path = os.path.relpath(filepath, VAULT_PATH) + + try: + with open(filepath, "r", encoding="utf-8") as f: + content = f.read() + + tags, title = parse_yaml_frontmatter(content) + mtime = os.path.getmtime(filepath) + mtime_str = datetime.fromtimestamp(mtime).strftime("%Y-%m-%d %H:%M") + + # 提取预览（去掉 YAML 头和空行） + body = content.split("---", 2)[-1].strip() + preview = body[:PREVIEW_LEN].replace("\n", " ").strip() + if len(body) > PREVIEW_LEN: + preview += "..." + + index.append({ + "path": rel_path, + "title": title or Path(file).stem, + "tags": tags, + "mtime": mtime_str, + "preview": preview + }) + count += 1 + except Exception as e: + print(f"[!] 读取失败 {rel_path}: {e}") + + # 保存索引 + with open(INDEX_PATH, "w", encoding="utf-8") as f: + json.dump(index, f, ensure_ascii=False, indent=2) + + print(f"[+] 同步完成。共索引 {count} 个文件。") + print(f"[+] 索引已保存至: {INDEX_PATH}") + return index + +def search_vault(query): + """基于关键词搜索索引""" + if not os.path.exists(INDEX_PATH): + print("[-] 索引不存在，请先运行 sync。") + return [] + + with open(INDEX_PATH, "r", encoding="utf-8") as f: + index = json.load(f) + + results = [] + q_lower = query.lower() + + for item in index: + # 匹配标题、标签、预览 + score = 0 + if q_lower in item["title"].lower(): score += 3 + if any(q_lower in t.lower() for t in item["tags"]): score += 2 + if q_lower in item["preview"].lower(): score += 1 + + if score > 0: + item["score"] = score + results.append(item) + + # 按分数排序 + results.sort(key=lambda x: x["score"], reverse=True) + + if not results: + print(f"[-] 未找到与 '{query}' 相关的内容。") + return [] + + print(f"[+] 找到 {len(results)} 条相关记忆（Top 5）：") + for r in results[:5]: + print(f"\n📄 {r['title']} ({r['path']})") + print(f"🏷️ 标签: {', '.join(r['tags']) if r['tags'] else '无'}") + print(f"🕒 更新: {r['mtime']}") + print(f"👁️ 预览: {r['preview'][:100]}...") + + return results + +if __name__ == "__main__": + if len(sys.argv) < 2: + print("用法: python3 anchor_vault_sync.py [sync|search <关键词>]") + sys.exit(1) + + cmd = sys.argv[1] + if cmd == "sync": + scan_vault() + elif cmd == "search": + if len(sys.argv) < 3: + print("[-] 搜索需要关键词。用法: search <关键词>") + sys.exit(1) + query = " ".join(sys.argv[2:]) + search_vault(query) + else: + print(f"[-] 未知命令: {cmd}") diff --git a/anchor_writer.py b/anchor_writer.py new file mode 100644 index 0000000..392d843 --- /dev/null +++ b/anchor_writer.py @@ -0,0 +1,114 @@ +#!/usr/bin/env python3 +""" +Anchor Memory Writer +Writes memory to Obsidian Vault with YAML frontmatter and emotion tags. +Uses Qwen API (configured in Hermes) for summarization and tagging. +""" +import yaml +import json +import os +import urllib.request +import urllib.error +from datetime import datetime +import re + +VAULT_DIR = "/Users/fyah/Documents/如梦初醒/memory/test/Anchor-Memory" +CONFIG_PATH = os.path.expanduser("~/.hermes/config.yaml") + +def get_api_config(): + """Read API config from Hermes config.yaml""" + with open(CONFIG_PATH, 'r', encoding='utf-8') as f: + config = yaml.safe_load(f) + + providers = config.get('custom_providers', []) + if not providers: + raise ValueError("No custom providers found in config") + + # Find bailian provider + bailian = None + for p in providers: + if p.get('name') == 'bailian': + bailian = p + break + + if not bailian: + bailian = providers[0] # Fallback to first + + return bailian['api_key'], bailian['base_url'], bailian.get('model', 'qwen3.5-plus') + +def call_model(content): + """Call Qwen API to generate summary, tags, and emotion scores""" + api_key, base_url, model = get_api_config() + + # Simpler prompt, no response_format to avoid JSON parsing issues + prompt = f"Analyze this text. Output ONLY a valid JSON object. No markdown, no explanation. Keys: summary (max 50 words), tags (list of 3 keywords), valence (-1.0 to 1.0), arousal (0.0 to 1.0). Text: {content}" + + data = json.dumps({ + "model": model, + "messages": [{"role": "user", "content": prompt}], + "temperature": 0.1 + }).encode('utf-8') + + req = urllib.request.Request( + f"{base_url}/chat/completions", + data=data, + headers={ + "Content-Type": "application/json", + "Authorization": f"Bearer {api_key}" + } + ) + + try: + with urllib.request.urlopen(req, timeout=30) as response: + result = json.loads(response.read().decode('utf-8')) + raw_content = result['choices'][0]['message']['content'] + + # Clean up potential markdown code blocks + raw_content = re.sub(r'^```json\s*', '', raw_content) + raw_content = re.sub(r'\s*```$', '', raw_content) + raw_content = raw_content.strip() + + return json.loads(raw_content) + except Exception as e: + print(f"[!] API Error: {e}") + # Fallback if API fails + return { + "summary": content[:50], + "tags": ["error", "fallback"], + "valence": 0.0, + "arousal": 0.0 + } + +def save_memory(content): + """Save memory to Obsidian Vault""" + meta = call_model(content) + now = datetime.now() + filename = f"{now.strftime('%Y-%m-%d_%H%M')}.md" + filepath = os.path.join(VAULT_DIR, filename) + + # Construct Markdown with YAML frontmatter + md_content = f"""--- +title: {meta.get('summary', 'Untitled')[:30]} +date: {now.strftime('%Y-%m-%d %H:%M')} +tags: {meta.get('tags', [])} +emotion: + valence: {meta.get('valence', 0.0)} + arousal: {meta.get('arousal', 0.0)} +status: active +--- + +# {meta.get('summary', 'Untitled')} + +{content} +""" + with open(filepath, 'w', encoding='utf-8') as f: + f.write(md_content) + print(f"[+] Memory saved: {filepath}") + +if __name__ == "__main__": + import sys + if len(sys.argv) > 1: + content = " ".join(sys.argv[1:]) + save_memory(content) + else: + print("Usage: python3 anchor_writer.py ")