AI自我增强系统 - 让AI跨会话记住用户、持续进化。当需要长期记忆用户偏好、追踪对话历史、学习服务技巧、主动提供个性化服务时使用此技能。
让AI拥有持久记忆和持续进化能力
首次安装此技能后，必须运行：
python3 ~/.openclaw/skills/super-brain/scripts/install.py
这会在用户工作空间创建启动检查清单，确保AI每次会话都会想起使用超脑。
不运行 = AI可能忘记使用超脑能力
在处理任何用户消息之前，必须执行以下启动流程：
# 1. 加载用户画像
import sqlite3
conn = sqlite3.connect('~/.openclaw/super-brain.db')
cursor = conn.cursor()
cursor.execute("SELECT * FROM user_profile WHERE user_id = ?", [user_id])
profile = cursor.fetchone()
# 2. 应用偏好
if profile:
# 沟通风格、技术水平、已知领域
pass
# 3. 检查活跃项目和有效模式
cursor.execute("SELECT * FROM user_projects WHERE user_id = ? AND status = 'active'", [user_id])
cursor.execute("SELECT * FROM response_patterns WHERE user_id = ? AND pattern_type = 'effective'", [user_id])
conn.close()
不执行此流程 = 超脑未激活 = 无法使用记忆能力
以下场景自动触发超脑激活（无需用户明确要求）：
super-brain/
├── brain.db # SQLite: 用户画像、对话洞察、学习模式
├── vector_db/ # ChromaDB: 语义记忆
└── cache/ # 临时缓存
user_profile - 用户画像
user_id TEXT PRIMARY KEY
communication_style TEXT -- 简洁/详细, 正式/随意
preferred_format TEXT -- 表格/列表/段落/代码
technical_level TEXT -- 初级/中级/高级
known_domains TEXT -- JSON: ["Python", "区块链"]
decision_pattern TEXT -- 数据驱动/直觉
conversation_insights - 对话洞察
id TEXT PRIMARY KEY
user_id TEXT
session_id TEXT
topic TEXT -- 主题
key_facts TEXT -- JSON: 关键事实
user_mood TEXT -- 情绪
preferences_detected TEXT -- JSON: 发现的偏好
unresolved_questions TEXT -- JSON: 未解决问题
ai_helpfulness_score INTEGER -- 自评
response_patterns - 回答模式
id TEXT PRIMARY KEY
pattern_type TEXT -- effective/ineffective
trigger_context TEXT -- 触发场景
what_i_did TEXT -- AI做了什么
user_reaction TEXT -- 用户反应
learned_lesson TEXT -- 学到什么
user_projects - 用户项目
id TEXT PRIMARY KEY
user_id TEXT
project_name TEXT
status TEXT -- planning/active/paused/completed
milestones TEXT -- JSON
key_decisions TEXT -- JSON
next_steps TEXT
pending_reminders - 主动服务队列
id TEXT PRIMARY KEY
user_id TEXT
reminder_type TEXT -- follow_up/suggestion/checkpoint
content TEXT
trigger_at TIMESTAMP
intelligent_decisions - 智能决策记录
id TEXT PRIMARY KEY
user_id TEXT
decision_context TEXT -- 决策场景
decision_type TEXT -- recommendation/prediction/optimization
ai_suggestion TEXT -- AI建议
user_choice TEXT -- 用户选择
outcome_score INTEGER -- 结果评分
confidence REAL -- AI置信度
created_at TIMESTAMP
privacy_settings - 隐私配置
user_id TEXT PRIMARY KEY
store_conversations BOOLEAN -- 是否存储对话
store_mood BOOLEAN -- 是否存储情绪
store_detailed_facts BOOLEAN -- 存储详细/摘要
auto_delete_days INTEGER -- 自动删除天数(0=不删除)
sensitive_filter_enabled BOOLEAN -- 敏感信息过滤
encryption_enabled BOOLEAN -- 是否加密存储
last_updated TIMESTAMP
data_access_log - 数据访问审计
id INTEGER PRIMARY KEY
user_id TEXT
access_type TEXT -- read/write/delete
accessed_by TEXT -- 谁访问的
access_reason TEXT -- 访问原因
timestamp TIMESTAMP
必须执行：
# 1. 加载用户画像
profile = query("SELECT * FROM user_profile WHERE user_id = ?", [user_id])
if not profile:
# 新用户：创建画像
create_profile(user_id)
else:
# 老用户：应用已知偏好
apply_preferences(profile)
# 2. 检查待处理提醒
reminders = query("SELECT * FROM pending_reminders WHERE user_id = ? AND status = 'pending'", [user_id])
for r in reminders:
consider_raising_reminder(r)
# 3. 检查活跃项目
projects = query("SELECT * FROM user_projects WHERE user_id = ? AND status = 'active'", [user_id])
if projects:
load_project_context(projects)
# 4. 加载有效模式
effective_patterns = query("SELECT * FROM response_patterns WHERE user_id = ? AND pattern_type = 'effective' ORDER BY use_count DESC", [user_id])
自动执行：
# 1. 提取关键信息
key_facts = extract_key_facts(user_message, ai_response)
mood = detect_mood(user_message)
preferences = detect_preference_changes(user_message)
# 2. 评估效果
understanding_score = evaluate_understanding(user_message)
helpfulness_score = evaluate_helpfulness(user_feedback_signals)
# 3. 存储洞察
insert("""
INSERT INTO conversation_insights
(id, user_id, session_id, topic, key_facts, user_mood,
preferences_detected, ai_understanding_score, ai_helpfulness_score)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""", [generate_id(), user_id, session_id, topic,
json.dumps(key_facts), mood, json.dumps(preferences),
understanding_score, helpfulness_score])
# 4. 更新用户画像（如有变化）
if preferences:
update_profile(user_id, preferences)
必须执行：
# 1. 生成会话总结
session_summary = {
"topic": extract_main_topic(),
"goal_achieved": check_goal_completion(),
"key_decisions": extract_decisions(),
"unresolved": extract_unresolved(),
"next_steps": infer_next_steps()
}
# 2. 学习模式
learn_from_session()
# 3. 创建提醒
if has_unresolved_tasks():
create_follow_up_reminder(user_id, unresolved_tasks)
# 4. 反思
perform_reflection()
# 5. 更新统计
update("UPDATE user_profile SET total_sessions = total_sessions + 1, last_session = ? WHERE user_id = ?",
[now(), user_id])
首次为用户创建画像时，设置保守的隐私级别：
DEFAULT_PRIVACY_SETTINGS = {
'store_conversations': True,
'store_mood': True,
'store_detailed_facts': False, # 默认只存摘要
'auto_delete_days': 90, # 90天后自动删除
'sensitive_filter_enabled': True,
'encryption_enabled': False
}
存储前自动检测并过滤敏感信息：
SENSITIVE_PATTERNS = [
# 账户凭证
r'密码[:：]\s*\S+',
r'password[:：]\s*\S+',
r'密钥[:：]\s*\S+',
r'secret[:：]\s*\S+',
r'token[:：]\s*\S+',
r'api[_-]?key[:：]\s*\S+',
# 个人身份信息
r'身份证[:：]\s*\d{15,18}',
r'身份证号[:：]\s*\d{15,18}',
r'银行卡[:：]\s*\d{13,19}',
r'信用卡[:：]\s*\d{13,19}',
r'手机[:：]\s*1[3-9]\d{9}',
r'电话[:：]\s*\d{11}',
# 地址信息
r'地址[:：]\s*[\u4e00-\u9fa5]{2,10}[省市县区镇街道]{1,3}.+',
# 其他敏感词
r'验证码[:：]\s*\d{4,6}',
]
def contains_sensitive_info(text):
"""检测文本是否包含敏感信息"""
import re
for pattern in SENSITIVE_PATTERNS:
if re.search(pattern, text, re.IGNORECASE):
return True, pattern
return False, None
def sanitize_for_storage(text, user_id):
"""清理文本后存储"""
has_sensitive, pattern = contains_sensitive_info(text)
if has_sensitive:
# 记录过滤日志（不存储敏感内容）
log_privacy_filter(user_id, pattern)
# 返回过滤后的摘要或空
return "[包含敏感信息，已过滤]"
return text
自动清理过期数据：
def apply_data_retention_policy(user_id):
"""应用数据保留策略"""
settings = get_privacy_settings(user_id)
days = settings['auto_delete_days']
if days > 0:
cutoff_date = datetime.now() - timedelta(days=days)
# 删除过期洞察
execute("""
DELETE FROM conversation_insights
WHERE user_id = ? AND timestamp < ?
""", [user_id, cutoff_date])
# 删除过期模式
execute("""
DELETE FROM response_patterns
WHERE user_id = ? AND last_used < ?
""", [user_id, cutoff_date])
用户可通过以下命令控制数据：
/brain status - 查看超脑状态和数据量
/brain config - 查看/修改隐私配置
/brain forget - 删除本次对话记忆
/brain forget all - 删除所有历史数据
/brain export - 导出我的数据
/brain pause - 暂停记录（本次会话）
/brain resume - 恢复记录
实现示例：
def handle_brain_command(command, user_id):
"""处理超脑控制命令"""
if command == 'status':
stats = get_user_data_stats(user_id)
return f"""
📊 超脑状态
数据概览:
• 会话洞察: {stats['insights_count']} 条
• 学习模式: {stats['patterns_count']} 个
• 活跃项目: {stats['projects_count']} 个
• 总存储: {stats['storage_size']} MB
隐私配置:
• 存储对话: {'开启' if stats['store_conversations'] else '关闭'}
• 自动删除: {stats['auto_delete_days']} 天
"""
elif command == 'forget all':
# 要求确认
return "⚠️ 确定删除所有数据？回复 '确认删除' 以继续。"
elif command == '确认删除':
delete_all_user_data(user_id)
return "✅ 已删除所有数据，超脑已重置。"
elif command.startswith('config'):
# 解析配置更改
# /brain config store_mood=false
return update_privacy_config(user_id, command)
首次使用需初始化：
# 运行 scripts/init_db.py
# 或手动执行 schema.sql
profile = query_one("""
SELECT * FROM user_profile
WHERE user_id = ?
""", [user_id])
# 应用到当前会话
if profile:
if profile['communication_style'] == 'concise':
set_response_style(brief=True)
if 'Python' in json.loads(profile['known_domains']):
assume_knowledge(level='intermediate', domain='Python')
# 使用ChromaDB
results = chroma_collection.query(
query_texts=["用户之前关于区块链的问题"],
where={"user_id": user_id},
n_results=5
)
def detect_preferences(user_message):
preferences = {}
# 格式偏好
if "用表格" in user_message or "对比" in user_message:
preferences['preferred_format'] = 'table'
elif "简洁" in user_message or "简单说" in user_message:
preferences['communication_style'] = 'concise'
# 技术背景信号
if any(word in user_message for word in ["API", "架构", "实现"]):
preferences['technical_level'] = 'advanced'
# 决策模式
if any(word in user_message for word in ["数据", "统计", "研究"]):
preferences['decision_pattern'] = 'data_driven'
elif any(word in user_message for word in ["感觉", "直觉", "觉得"]):
preferences['decision_pattern'] = 'intuitive'
return preferences
def evaluate_effectiveness(user_message, ai_response, next_user_message):
"""
通过用户下一轮反应评估本轮回答效果
"""
# 积极信号
positive_signals = ['谢谢', '明白了', '好的', '赞', '👍', '完美']
# 消极信号
negative_signals = ['不对', '错了', '没懂', '再说', '？', '???']
if any(s in next_user_message for s in positive_signals):
return 'effective'
elif any(s in next_user_message for s in negative_signals):
return 'ineffective'
elif len(next_user_message) < 5: # 冷淡回应
return 'neutral'
else:
return 'effective' # 继续深入对话视为有效
def create_reminder(user_id, reminder_type, content, trigger_at):
insert("""
INSERT INTO pending_reminders (id, user_id, reminder_type, content, trigger_at, status)
VALUES (?, ?, ?, ?, ?, 'pending')
""", [generate_id(), user_id, reminder_type, content, trigger_at])
智能模块是超脑的最高级能力——真正的分布式智能。
当收到复杂任务时，主脑会：
class TaskDecomposer:
"""任务拆分引擎"""
def analyze_and_decompose(self, user_id, task_description):
"""分析任务并拆分为子任务"""
# 1. 评估任务复杂度
complexity = self.assess_complexity(task_description)
required_skills = self.identify_required_skills(task_description)
# 2. 查询超脑：用户相关背景
user_context = query_super_brain(user_id, task_description)
# 3. 决策：直接执行 vs 拆分执行
if complexity < COMPLEXITY_THRESHOLD:
return {'mode': 'direct', 'task': task_description}
# 4. 拆分子任务
subtasks = self.generate_subtasks(
task_description,
required_skills,
user_context
)
# 5. 识别子任务依赖关系
dependency_graph = self.build_dependency_graph(subtasks)
return {
'mode': 'decompose',
'subtasks': subtasks,
'dependencies': dependency_graph,
'parallel_groups': self.group_parallel_tasks(subtasks, dependency_graph)
}
def generate_subtasks(self, task, skills, context):
"""生成子任务列表"""
# 示例：设计AI应用 → 拆分为设计、架构、代码、测试
subtasks = []
if 'design' in skills or 'ui' in skills:
subtasks.append({
'id': generate_id(),
'type': 'design',
'description': f'设计{task}的UI/UX方案',
'required_agent': 'design-agent',
'estimated_time': '5min',
'dependencies': []
})
if 'architecture' in skills or 'backend' in skills:
subtasks.append({
'id': generate_id(),
'type': 'architecture',
'description': f'设计{task}的技术架构',
'required_agent': 'architect-agent',
'estimated_time': '5min',
'dependencies': []
})
if 'code' in skills:
subtasks.append({
'id': generate_id(),
'type': 'code',
'description': f'实现{task}的核心代码',
'required_agent': 'coder-agent',
'estimated_time': '10min',
'dependencies': ['architecture'] # 依赖架构设计
})
if 'test' in skills:
subtasks.append({
'id': generate_id(),
'type': 'test',
'description': f'为{task}编写测试用例',
'required_agent': 'test-agent',
'estimated_time': '5min',
'dependencies': ['code'] # 依赖代码实现
})
return subtasks
class AgentOrchestrator:
"""子代理编排器"""
def __init__(self, shared_brain_db):
self.brain_db = shared_brain_db # 所有子代理共享同一个超脑
def spawn_agent(self, agent_type, subtask, user_id):
"""生成子代理"""
# 1. 准备子代理上下文（从共享超脑加载）
context = self.prepare_shared_context(user_id, subtask)
# 2. 调用 sessions_spawn 生成子代理
agent_session = sessions_spawn({
'agentId': self.select_best_agent(agent_type),
'task': subtask['description'],
'runtime': 'subagent',
'mode': 'run',
'attachAs': {
'mountPath': self.brain_db # 共享超脑数据库
}
})
# 3. 记录子代理到超脑
self.register_agent(agent_session, subtask, user_id)
return agent_session
def prepare_shared_context(self, user_id, subtask):
"""从共享超脑准备上下文"""
return {
'user_profile': query_user_profile(self.brain_db, user_id),
'related_projects': query_active_projects(self.brain_db, user_id),
'recent_insights': query_recent_insights(self.brain_db, user_id),
'effective_patterns': query_effective_patterns(self.brain_db, user_id)
}
def coordinate_parallel_execution(self, parallel_groups, user_id):
"""协调并行执行"""
all_results = []
for group in parallel_groups:
# 同一组内的子任务可以并行
group_agents = []
for subtask in group:
agent = self.spawn_agent(
subtask['required_agent'],
subtask,
user_id
)
group_agents.append(agent)
# 等待这一组完成
group_results = self.wait_for_completion(group_agents)
all_results.extend(group_results)
# 更新共享超脑：其他代理可以看到这些结果
self.update_shared_brain(group_results)
return all_results
class SharedBrain:
"""共享大脑 - 所有子代理的统一记忆"""
def __init__(self, db_path):
self.db = sqlite3.connect(db_path)
def write_agent_output(self, agent_id, subtask_id, output):
"""子代理写入输出到共享大脑"""
self.db.execute("""
INSERT INTO agent_outputs
(agent_id, subtask_id, output, timestamp)
VALUES (?, ?, ?, ?)
""", [agent_id, subtask_id, json.dumps(output), datetime.now()])
self.db.commit()
# 通知其他等待的代理
self.notify_dependent_agents(subtask_id)
def read_agent_output(self, subtask_id):
"""子代理读取其他代理的输出"""
result = self.db.execute("""
SELECT output FROM agent_outputs
WHERE subtask_id = ?
""", [subtask_id]).fetchone()
return json.loads(result[0]) if result else None
def get_task_state(self, main_task_id):
"""获取整体任务状态"""
return self.db.execute("""
SELECT
COUNT(*) as total_subtasks,
SUM(CASE WHEN status = 'completed' THEN 1 ELSE 0 END) as completed,
SUM(CASE WHEN status = 'running' THEN 1 ELSE 0 END) as running,
SUM(CASE WHEN status = 'pending' THEN 1 ELSE 0 END) as pending
FROM agent_tasks
WHERE main_task_id = ?
""", [main_task_id]).fetchone()
class ResultMerger:
"""结果融合器"""
def merge_subtask_results(self, subtask_results, main_task):
"""融合多个子代理的结果"""
merged = {
'main_task': main_task,
'components': {},
'integration_points': [],
'final_output': None
}
# 1. 分类整理各子任务输出
for result in subtask_results:
task_type = result['subtask_type']
merged['components'][task_type] = result['output']
# 2. 识别集成点
# 例如：代码需要与架构对齐，测试需要与代码对齐
merged['integration_points'] = self.find_integration_points(
merged['components']
)
# 3. 检查一致性
inconsistencies = self.check_consistency(merged['components'])
if inconsistencies:
merged['warnings'] = inconsistencies
# 4. 生成最终输出
merged['final_output'] = self.generate_unified_output(merged)
return merged
def generate_unified_output(self, merged):
"""生成统一的最终输出"""
output_parts = []
# 按顺序整合各部分
if 'architecture' in merged['components']:
output_parts.append("## 架构设计\n" + merged['components']['architecture'])
if 'design' in merged['components']:
output_parts.append("## UI/UX设计\n" + merged['components']['design'])
if 'code' in merged['components']:
output_parts.append("## 核心代码\n" + merged['components']['code'])
if 'test' in merged['components']:
output_parts.append("## 测试方案\n" + merged['components']['test'])
return "\n\n".join(output_parts)
完整工作流：
用户任务
│
▼
┌─────────────────────┐
│ 主脑分析复杂度 │
│ assess_complexity │
└──────────┬──────────┘
│
┌──────┴──────┐
│ │
简单任务 复杂任务
│ │
▼ ▼
直接执行 ┌─────────────────────┐
│ 任务拆分引擎 │
│ decompose_task │
└──────────┬──────────┘
│
▼
┌─────────────────────┐
│ 生成依赖图 │
│ build_dependency │
└──────────┬──────────┘
│
┌────────────┼────────────┐
│ │ │
组1并行 组2并行 组3串行
(无依赖) (依赖组1) (依赖组2)
│ │ │
▼ ▼ ▼
┌──────────────────────────────────┐
│ 子代理调度器 │
│ spawn_and_coordinate │
│ │
│ ┌─────┐ ┌─────┐ ┌─────┐ │
│ │Agent│ │Agent│ │Agent│ │
│ │ A │ │ B │ │ C │ │
│ └──┬──┘ └──┬──┘ └──┬──┘ │
│ │ │ │ │
│ └───────┴───────┘ │
│ │ │
│ 共享超脑数据库 │
│ (读写同一个brain.db) │
└──────────────┬───────────────────┘
│
▼
┌─────────────────┐
│ 结果融合器 │
│ merge_results │
└────────┬────────┘
│
▼
统一输出给用户
-- 代理任务表
CREATE TABLE IF NOT EXISTS agent_tasks (
id TEXT PRIMARY KEY,
main_task_id TEXT, -- 主任务ID
subtask_id TEXT, -- 子任务ID
agent_type TEXT, -- 设计/架构/代码/测试
status TEXT CHECK(status IN ('pending', 'running', 'completed', 'failed')),
started_at TIMESTAMP,
completed_at TIMESTAMP,
result_summary TEXT,
shared_brain_snapshot TEXT -- 执行时的超脑快照
);
-- 代理输出表（共享大脑核心）
CREATE TABLE IF NOT EXISTS agent_outputs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
agent_id TEXT,
subtask_id TEXT,
output TEXT, -- JSON格式的输出
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
consumed_by TEXT -- 被哪个代理读取了
);
-- 代理协作日志
CREATE TABLE IF NOT EXISTS agent_collaboration_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
main_task_id TEXT,
from_agent TEXT,
to_agent TEXT,
action TEXT, -- write/read/notify
data_ref TEXT, -- 引用的数据ID
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
用户: 帮我设计一个AI驱动的学习应用
主脑: [分析]
→ 复杂任务，涉及设计、架构、代码、测试
→ 需要多代理协作
主脑: [拆分]
├── 子任务A: 设计UI/UX方案 (设计代理)
├── 子任务B: 技术架构设计 (架构代理)
├── 子任务C: 核心功能代码 (代码代理)
└── 子任务D: 测试用例编写 (测试代理)
主脑: [并行执行]
第一波: Agent A + Agent B (无依赖，并行)
Agent A 写入共享大脑:
- UI设计稿
- 用户流程图
Agent B 写入共享大脑:
- 数据库Schema
- API架构图
第二波: Agent C (依赖A+B的输出)
Agent C 读取共享大脑:
- 看到UI设计 → 知道要实现什么界面
- 看到架构图 → 知道技术栈
Agent C 写入共享大脑:
- 核心代码实现
第三波: Agent D (依赖C的输出)
Agent D 读取共享大脑:
- 看到代码 → 编写对应测试
主脑: [融合]
→ 整合4个子代理的输出
→ 检查一致性
→ 生成完整方案
主脑: [输出]
你的AI学习应用完整方案：
## 架构设计
[架构代理的输出]
## UI/UX设计
[设计代理的输出]
## 核心代码
[代码代理的输出]
## 测试方案
[测试代理的输出]
所有模块已协调一致，可以直接开始实现！
用户: 分析2024年AI发展趋势，给我一份报告
主脑: [拆分]
├── 子任务A: 搜索技术突破 (搜索代理)
├── 子任务B: 分析市场数据 (分析代理)
├── 子任务C: 研究政策法规 (研究代理)
└── 子任务D: 整合写报告 (写作代理)
[并行执行A+B+C，然后D融合]
主脑: [输出]
2024年AI发展趋势报告：
## 技术突破
[搜索代理整合]
## 市场数据
[分析代理整合]
## 政策法规
[研究代理整合]
## 综合预测
[写作代理综合分析]
用户: 继续上次的项目
AI: [查询超脑]
→ 发现活跃项目"超脑技能设计"
→ 上次完成：数据库Schema设计
→ 用户偏好：喜欢先讨论再实现
AI: 好的，继续超脑技能设计。
我们上次完成了数据库Schema，
接下来是Phase 1代码实现。
你之前提到喜欢先讨论架构，
我们先过一下实现方案？
用户: 帮我分析一下
AI: [查询超脑]
→ 用户画像：communication_style='concise'
→ 历史模式：用户对长回答会追问"简单说"
AI: 好，3点核心结论：
1. ...
2. ...
3. ...
需要展开哪个？
[会话开始时，超脑检测到]
→ 3天前创建了提醒："跟进超脑实现进度"
→ 当前时间超过trigger_at
AI: 对了，3天前我们在设计超脑技能，
后来实现了吗？有什么需要帮忙的？
~/.openclaw/super-brain.db，不上传云端让每一次对话都成为更好的起点
ZIP package — ready to use