多Agent记忆架构：如何让AI角色共享上下文又保持隐私？

核心观点：未来的AI系统不是单个超级智能，而是多个专业角色的协作网络。关键挑战在于：如何让它们共享必要的上下文，同时保持各自的隐私和独立性？

引言：为什么需要多Agent？

想象一个场景：你正在开发一个复杂的项目，需要一个AI助手同时具备以下能力：

• 代码审查和性能优化
• 技术文档撰写
• 项目管理和进度跟踪
• 用户沟通和需求分析

单Agent的困境：

现实中的"全能AI"面临根本性限制：

上下文窗口限制 → 无法在单一模型中深度掌握多个领域
专业领域冲突 → 代码专家的严谨思维 vs 创作者的灵活思维
学习遗忘问题 → 学习新技能时可能遗忘旧知识
更新耦合问题 → 更新一个领域知识可能影响其他领域

多Agent的优势：

┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│ Code Expert │    │   Writer    │    │  Analyst    │
│ (代码专家)   │    │  (写作者)    │    │  (分析师)    │
├─────────────┤    ├─────────────┤    ├─────────────┤
• Python/C++  │    • 技术写作    │    • 数据分析    │
• 性能优化     │    • 文章结构    │    • 趋势预测    │
• 调试技巧     │    • 语言风格    │    • 可视化      │
└─────────────┘    └─────────────┘    └─────────────┘
      ↓                  ↓                  ↓
      └──────────────────┴──────────────────┘
                        ↓
              ┌─────────────────┐
              │  公共记忆层      │
              │  (共享上下文)    │
              └─────────────────┘

多Agent系统带来的价值：

• 专业深度：每个角色专注于自己的领域
• 并行协作：多个任务可以同时进行
• 弹性扩展：可以随时添加新角色
• 容错能力：单个角色故障不影响整体

但是，核心挑战来了：如何让这些角色协同工作，而不是变成信息孤岛？

一、核心挑战：记忆共享 vs 隐私隔离

1.1 现实世界的类比

想象一个公司的组织结构：

CEO（协调者）
├── 技术团队（技术细节，专业术语）
├── 产品团队（市场数据，用户反馈）
└── 运营团队（日常事务，执行细节）

共享信息：
- 公司战略
- 产品目标
- 重要决策

隔离信息：
- 技术实现细节（技术团队私有）
- 市场调研数据（产品团队私有）
- 运营流程细节（运营团队私有）

这个类比揭示了多Agent系统的核心矛盾：

• 需要共享：避免重复工作，保持一致性
• 需要隔离：专业深度，隐私保护

1.2 技术挑战分解

挑战1：记忆共享

问题：如何决定哪些信息应该共享？

# 场景示例
场景1：用户更换了项目路径
决策：应该写入公共记忆
理由：所有角色都需要知道

场景2：代码专家发现了一个性能优化技巧
决策：应该保留在角色记忆
理由：只有代码专家会用

场景3：写作角色完成了一篇文章
决策：写入公共记忆（项目状态）
理由：其他角色需要知道进度

挑战2：隐私隔离

问题：如何防止敏感信息泄露？

# 安全风险示例
风险1：代码专家的调试历史包含错误密码
风险2：分析师的数据分析包含用户隐私
风险3：所有角色都能读取用户的所有对话

# 解决方案：ACL（访问控制列表）
roles/code-expert/debugging-history.md:
  read: [code-expert, admin]
  write: [code-expert]

挑战3：冷启动

问题：新角色如何快速获得上下文？

传统方式：
新角色启动 → 一无所知 → 需要大量交互 → 逐渐积累

优化方式：
新角色启动 → 加载公共记忆 → 加载角色模板 → 
向量检索相关内容 → 快速具备基本能力

挑战4：一致性维护

问题：多个角色同时更新记忆时如何处理？

场景：
- 代码专家说："Bug已修复"
- 测试专家说："Bug仍然存在"

冲突解决：
- 向量时钟（Vector Clock）检测冲突
- 时间戳排序（Last Write Wins）
- 人工审核（重要决策）

二、三层记忆架构设计

基于Clawdbot的实践，我们设计了一个三层记忆架构：

2.1 架构概览

┌─────────────────────────────────────────────────────────┐
│                   多Agent记忆系统                        │
├─────────────────────────────────────────────────────────┤
│                                                          │
│  ┌─────────────────────────────────────────────────┐   │
│  │        L1: 公共记忆（Global Memory）              │   │
│  │  • 所有Agent共享                                  │   │
│  │  • 用户基本信息、偏好、通用上下文                   │   │
│  │  • 重要决策、项目状态、共享知识                     │   │
│  │  • 访问规则：all=r, admin=w                       │   │
│  └─────────────────────────────────────────────────┘   │
│                        ↕                                │
│  ┌─────────────────────────────────────────────────┐   │
│  │        L2: 角色记忆（Role Memory）                │   │
│  │  • Agent私有，专业领域数据                         │   │
│  │  • 技能专长、历史经验、领域知识                     │   │
│  │  • 角色特定的配置和偏好                            │   │
│  │  • 访问规则：role=rw, collaborator=r              │   │
│  └─────────────────────────────────────────────────┘   │
│                        ↕                                │
│  ┌─────────────────────────────────────────────────┐   │
│  │        L3: 会话记忆（Session Memory）             │   │
│  │  • 临时上下文，短期记忆                            │   │
│  │  • 当前对话历史、任务状态                          │   │
│  │  • 会话结束后自动清理或归档                         │   │
│  │  • 访问规则：session_owner=rw                    │   │
│  └─────────────────────────────────────────────────┘   │
│                                                          │
└─────────────────────────────────────────────────────────┘

2.2 L1: 公共记忆（Global Memory）

存储位置：

memory/global/
├── user-profile.md         # 用户基本信息
├── preferences.md          # 用户偏好
├── project-status.md       # 项目状态
├── shared-knowledge.md     # 共享知识库
└── decisions.md            # 重要决策记录

设计原则：

1. 最小化原则：只存储真正需要共享的信息
2. 格式标准化：使用统一的Front Matter格式
3. 版本控制：重要变更保留历史
4. 即时同步：变更立即广播到所有角色

数据示例：

---
type: global
access: all=r, admin=w
created: 2026-02-11
modified: 2026-02-11
---

# 用户基本信息

## 身份信息
- 姓名：张三
- 角色：全栈工程师
- 时区：UTC+8

## 技术栈
- 编程语言：Python, C++, JavaScript
- 框架/库：PyTorch, React, Docker
- 工具链：Git, Kubernetes, VS Code

## 项目位置
- 主项目目录：/home/lighthouse/twg/PyServer/
- 

### 2.3 L2: 角色记忆（Role Memory）

**存储位置**：

memory/roles/
├── code-expert/
│ ├── config.md # 角色配置
│ ├── skills.md # 技能清单
│ ├── experience.md # 历史经验
│ └── debugging-tips.md # 调试技巧
├── writer/
│ ├── config.md
│ ├── style-guide.md # 写作风格
│ ├── templates.md # 文章模板
│ └── published-works.md # 已发布作品
└── analyst/

├── config.md
├── analysis-methods.md    # 分析方法
└── reports.md             # 分析报告

**设计原则**：
1. **角色隔离**：每个角色有独立的目录
2. **专业深度**：存储该角色的专业知识
3. **选择性共享**：通过ACL控制哪些内容可以被其他角色访问
4. **自动学习**：从经验中自动提炼知识

**数据示例**：

type: role
role: code-expert
access: code-expert=rwc, writer=r, admin=rwc
tags: [code, debugging, performance]

created: 2026-02-11

代码专家 - 技能清单

核心技能

• Python/C++代码审查
• 性能优化（算法、内存、并发）
• 分布式系统设计
• 容器化部署（Docker, K8s）

历史经验

2026-02-10: 科学计算器优化

• 问题：代码体积过大（21KB），功能冗余
• 分析：DEG/RAD切换使用率<5%，数字键盘占40%代码
• 方案：删除DEG/RAD切换，移除数字键盘，简化UI
• 结果：代码减少到12KB（-36%），用户体验无影响
• 经验：精简功能比堆砌功能更重要

2026-02-08: Nginx路由问题修复

• 问题：自定义页面404，被Typecho的index.php拦截
• 分析：正则匹配太宽泛，路由优先级混乱
• 方案：简化nginx配置，使用router.php统一处理
• 结果：所有页面正常访问
• 经验：配置出问题时优先查看backup，不要盲目修改

私有配置

• 代码风格：PEP8 + Google Style Guide
• 偏好IDE：VS Code + Vim模式
• 常用工具：git, docker, kubectl, htop

• 调试策略：日志优先，断点其次

  
  ### 2.4 L3: 会话记忆（Session Memory）
  
  **存储位置**：

  memory/sessions/
  ├── agent-main-main-20260211-143022/
  │ ├── context.md # 当前上下文
  │ ├── task-queue.md # 任务队列
  │ └── temp-state.json # 临时状态
  └── agent-subagent-xxx-20260211/
  └── ...

• 临时性：会话结束自动清理
• 轻量级：只存储必要的临时状态
• 快速访问：内存缓存优先

• 选择性归档：重要内容可提升到上层

  数据示例：

  ---
  type: session
  session: agent-main-main-20260211-143022
  created: 2026-02-11T14:30:22Z
  ttl: 3600
  ---

当前会话上下文

任务列表

• [x] 研究Clawdbot现有架构
• [x] 设计三层记忆架构
• [ ] 撰写博客文章
• [ ] 发布到somingbai.com

当前状态

{
  "current_task": "撰写博客文章",
  "progress": 60,
  "blockers": [],
  "next_actions": [
    "完成架构设计部分",
    "添加代码示例",
    "撰写总结"
  ]
}

临时笔记

• 需要强调ACL的重要性
• 记得加入对比表格

• 准备一些实际案例

  
  ---
  
  ## 三、访问控制机制（ACL设计）
  
  ### 3.1 ACL模型
  
  我们采用了类似Unix文件系统的权限模型，但针对多Agent场景进行了扩展：
  

  [角色] [读权限] [写权限] [协作权限]

• r (read): 可读
• w (write): 可写
• c (collaborate): 可协作（可邀请其他角色参与）

• *: 所有权限

  通配符：

• all: 所有角色
• authenticated: 所有已认证角色

• $session_owner: 当前会话所有者

  3.2 ACL配置示例

  文件：memory/.acl

  # ACL配置文件
  version: "1.0"

公共记忆

global/user-profile.md:
read: ["*"]
write: ["admin", "main-agent"]
collaborate: []

global/preferences.md:
read: ["*"]
write: ["admin", "main-agent"]
collaborate: ["admin"]

global/project-status.md:
read: ["*"]
write: ["admin", "project-manager", "main-agent"]
collaborate: ["admin", "project-manager"]

角色记忆

roles/code-expert/config.md:
read: ["code-expert", "admin", "main-agent"]
write: ["code-expert", "admin"]
collaborate: []

roles/code-expert/experience.md:
read: ["code-expert", "writer", "admin"]
write: ["code-expert"]
collaborate: ["code-expert", "writer"]

roles/writer/style-guide.md:
read: ["writer", "code-expert", "admin"]
write: ["writer", "admin"]
collaborate: ["writer", "code-expert"]

会话记忆

sessions///context.md:
read: ["$session_owner", "admin"]
write: ["$session_owner"]
collaborate: []
ttl: 86400 # 24小时后自动清理

### 3.3 权限检查实现

**Python实现示例**：

from typing import List, Dict, Optional
import fnmatch
import yaml

class ACLChecker:

def __init__(self, acl_file: str = "memory/.acl"):
    with open(acl_file) as f:
        self.acl_rules = yaml.safe_load(f)

def check_access(self, 
                role: str, 
                memory_path: str, 
                action: str) -> bool:
    """
    检查角色是否有权限执行操作
    
    Args:
        role: 角色名（如 "code-expert"）
        memory_path: 记忆文件路径
        action: 操作类型（read/write/collaborate）
    
    Returns:
        bool: 是否有权限
    """
    # 1. 精确匹配
    if memory_path in self.acl_rules:
        rule = self.acl_rules[memory_path]
        return self._check_rule(role, action, rule)
    
    # 2. 模式匹配
    for pattern, rule in self.acl_rules.items():
        if fnmatch.fnmatch(memory_path, pattern):
            return self._check_rule(role, action, rule)
    
    # 3. 默认拒绝
    return False

def _check_rule(self, 
               role: str, 
               action: str, 
               rule: Dict) -> bool:
    """检查具体规则"""
    # 获取该操作的角色列表
    allowed_roles = rule.get(action, [])
    
    # 检查通配符
    if "*" in allowed_roles or "all" in allowed_roles:
        return True
    
    # 检查角色匹配
    if role in allowed_roles:
        return True
    
    # 检查特殊变量
    if "$session_owner" in allowed_roles:
        # 这里需要实际的会话上下文
        # 简化处理：假设角色就是会话所有者
        return True
    
    # 检查协作权限
    if action == "write" and role in rule.get("collaborate", []):
        return True
    
    return False

def filter_search_results(self, 
                         role: str, 
                         results: List[Dict]) -> List[Dict]:
    """
    过滤搜索结果，只返回有权限访问的内容
    
    Args:
        role: 角色名
        results: 搜索结果列表
    
    Returns:
        过滤后的结果
    """
    filtered = []
    for result in results:
        if self.check_access(role, result["file"], "read"):
            # 可以进一步脱敏处理
            filtered.append(result)
    return filtered

使用示例

acl = ACLChecker()

检查代码专家是否能读取写作风格指南

if acl.check_access("code-expert", "roles/writer/style-guide.md", "read"):

print("允许访问")

else:

print("拒绝访问")

过滤搜索结果

results = [

{"file": "global/user-profile.md", "content": "..."},
{"file": "roles/code-expert/debugging-tips.md", "content": "..."},
{"file": "roles/writer/drafts.md", "content": "..."}

]

filtered_results = acl.filter_search_results("code-expert", results)

结果：不包含 drafts.md（其他角色无权限）

### 3.4 安全考虑

**深度防御策略**：

第1层：文件系统权限
├── Linux文件权限（chmod 600）
└── 目录隔离（每个角色独立目录）

第2层：ACL规则
├── 显式允许（白名单）
└── 默认拒绝（黑名单）

第3层：审计日志
├── 记录所有访问
├── 记录所有拒绝
└── 异常检测

第4层：加密（可选）
├── 敏感数据加密存储
└── 传输加密

**审计日志示例**：

{
"timestamp": "2026-02-11T14:30:22Z",
"operation": "read",
"role": "code-expert",
"target": "roles/writer/style-guide.md",
"result": "denied",
"reason": "ACL rule mismatch",
"ip": "192.168.1.42",
"session": "agent-main-main-20260211"
}

---

## 四、冷启动解决方案

### 4.1 问题分析

新角色启动时面临"冷启动"问题：

新角色启动 → 记忆空白 → 需要大量交互 → 效率低下

理想流程：
新角色启动 → 快速加载上下文 → 立即具备能力 → 高效协作

### 4.2 角色模板机制

**模板文件**：`memory/templates/role-template.md`

type: role-template

version: "1.0"

{{ROLE_NAME}} - 角色配置

角色定义

{{ROLE_DESCRIPTION}}

核心能力

{{CORE_CAPABILITIES}}

初始上下文

• 自动从公共记忆加载：

  • ✓ 用户基本信息
  • ✓ 项目状态
  • ✓ 重要决策
  • ✓ 用户偏好

技能配置

{{SKILLS}}

工作偏好

{{PREFERENCES}}

协作配置

• 可协作角色：{{COLLABORATE_WITH}}

• 共享记忆：{{SHARED_MEMORY}}

  
  **创建新角色脚本**：

  !/bin/bash

  create-role.sh - 创建新角色

ROLE_NAME=$1
ROLE_DESC=$2

if [ -z "$ROLE_NAME" ]; then

echo "用法: $0 <角色名> <角色描述>"
exit 1

fi

1. 创建角色目录

mkdir -p "memory/roles/$ROLE_NAME"

2. 复制模板

cp "memory/templates/role-template.md" \
"memory/roles/$ROLE_NAME/config.md"

3. 替换变量

sed -i "s/{{ROLE_NAME}}/$ROLE_NAME/g" \

"memory/roles/$ROLE_NAME/config.md"

sed -i "s/{{ROLE_DESCRIPTION}}/$ROLE_DESC/g" \

"memory/roles/$ROLE_NAME/config.md"

4. 创建其他文件

touch "memory/roles/$ROLE_NAME/skills.md"
touch "memory/roles/$ROLE_NAME/experience.md"

5. 添加ACL规则

echo "
roles/$ROLE_NAME/*:
read: [$ROLE_NAME, admin, main-agent]
write: [$ROLE_NAME, admin]
collaborate: []
" >> "memory/.acl"

6. 重建索引

python3 memory_search.py --reindex

echo "✅ 角色 $ROLE_NAME 创建成功！"
echo "📝 请编辑 memory/roles/$ROLE_NAME/config.md"

### 4.3 上下文快速加载

**Python实现**：

from sentence_transformers import SentenceTransformer
import glob
from typing import Dict, List

class ContextLoader:

def __init__(self):
    self.model = SentenceTransformer('all-MiniLM-L6-v2')

def load_context_for_new_role(self, role_name: str) -> Dict:
    """
    为新角色加载初始上下文
    
    Args:
        role_name: 角色名
    
    Returns:
        包含初始上下文的字典
    """
    context = {
        "global": {},
        "role": {},
        "related": [],
        "decisions": []
    }
    
    # 1. 加载公共记忆（必需）
    print(f"📖 加载公共记忆...")
    context["global"] = self._load_global_memory()
    
    # 2. 加载角色配置（必需）
    print(f"📖 加载角色配置...")
    context["role"] = self._load_role_config(role_name)
    
    # 3. 向量检索相关角色记忆（可选）
    print(f"🔍 检索相关记忆...")
    context["related"] = self._search_related_memory(
        context["role"]["description"], 
        top_k=3
    )
    
    # 4. 关键词检索重要决策（可选）
    print(f"🔍 检索重要决策...")
    context["decisions"] = self._search_key_decisions(role_name)
    
    # 5. 生成上下文摘要
    print(f"📝 生成上下文摘要...")
    summary = self._summarize_context(context)
    
    return summary

def _load_global_memory(self) -> Dict:
    """加载公共记忆"""
    global_memory = {}
    
    # 必需文件
    required_files = [
        "user-profile.md",
        "preferences.md",
        "project-status.md"
    ]
    
    for filename in required_files:
        path = f"memory/global/{filename}"
        try:
            with open(path) as f:
                global_memory[filename] = f.read()
        except FileNotFoundError:
            print(f"⚠️  警告：{path} 不存在")
    
    return global_memory

def _load_role_config(self, role_name: str) -> Dict:
    """加载角色配置"""
    path = f"memory/roles/{role_name}/config.md"
    
    try:
        with open(path) as f:
            content = f.read()
            # 解析Front Matter
            front_matter, body = self._parse_front_matter(content)
            return {
                "front_matter": front_matter,
                "description": body,
                "config_path": path
            }
    except FileNotFoundError:
        # 角色不存在，使用默认配置
        return {
            "description": f"新角色: {role_name}",
            "config_path": None
        }

def _search_related_memory(self, query: str, top_k: int = 3) -> List[Dict]:
    """向量检索相关记忆"""
    # 生成查询向量
    query_embedding = self.model.encode(query)
    
    # 搜索所有角色记忆
    results = []
    for file_path in glob.glob("memory/roles/*/*.md"):
        with open(file_path) as f:
            content = f.read()
            sections = self._split_sections(content)
            
            for section in sections:
                section_embedding = self.model.encode(section["text"])
                similarity = self._cosine_similarity(
                    query_embedding, 
                    section_embedding
                )
                
                if similarity > 0.5:  # 相似度阈值
                    results.append({
                        "file": file_path,
                        "section": section["title"],
                        "content": section["text"][:500],  # 摘要
                        "similarity": similarity
                    })
    
    # 按相似度排序
    results.sort(key=lambda x: x["similarity"], reverse=True)
    
    return results[:top_k]

def _search_key_decisions(self, role_name: str) -> List[Dict]:
    """关键词检索重要决策"""
    # 简化实现：关键词搜索
    keywords = ["决策", "决定", "重要"]
    
    results = []
    for file_path in glob.glob("memory/global/*.md"):
        with open(file_path) as f:
            content = f.read()
            
            for keyword in keywords:
                if keyword in content:
                    results.append({
                        "file": file_path,
                        "keyword": keyword,
                        "context": self._extract_context(content, keyword)
                    })
    
    return results

def _summarize_context(self, context: Dict) -> str:
    """生成上下文摘要"""
    summary = f"""

上下文加载完成

公共记忆

• 用户基本信息：已加载
• 项目状态：已加载
• 用户偏好：已加载

角色配置

• 角色描述：{context['role'].get('description', '未知')}

相关记忆

"""

    
    for item in context["related"]:
        summary += f"- [{item['section']}] ({item['file']}) - 相似度: {item['similarity']:.2f}\n"
    
    return summary

def _parse_front_matter(self, content: str):
    """解析Front Matter"""
    lines = content.split('\n')
    
    if lines[0] == '---':
        # 找到结束的 ---
        for i, line in enumerate(lines[1:], 1):
            if line == '---':
                front_matter = '\n'.join(lines[1:i])
                body = '\n'.join(lines[i+1:])
                return front_matter, body
    
    return None, content

def _split_sections(self, content: str) -> List[Dict]:
    """分割Markdown章节"""
    sections = []
    lines = content.split('\n')
    current_section = {"title": "Introduction", "text": []}
    
    for line in lines:
        if line.startswith('#'):
            # 保存当前章节
            if current_section["text"]:
                sections.append({
                    "title": current_section["title"],
                    "text": '\n'.join(current_section["text"])
                })
            
            # 开始新章节
            current_section = {
                "title": line.lstrip('#').strip(),
                "text": []
            }
        else:
            current_section["text"].append(line)
    
    # 保存最后一个章节
    if current_section["text"]:
        sections.append({
            "title": current_section["title"],
            "text": '\n'.join(current_section["text"])
        })
    
    return sections

def _cosine_similarity(self, vec1, vec2):
    """计算余弦相似度"""
    import numpy as np
    return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))

def _extract_context(self, content: str, keyword: str, window: int = 100):
    """提取关键词上下文"""
    index = content.find(keyword)
    if index == -1:
        return ""
    
    start = max(0, index - window)
    end = min(len(content), index + len(keyword) + window)
    
    return content[start:end]

使用示例

loader = ContextLoader()
summary = loader.load_context_for_new_role("code-expert")
print(summary)

### 4.4 冷启动优化效果

**对比测试**：

无优化：
新角色启动 → 需要交互20次 → 耗时15分钟 → 达到基本可用

有优化：
新角色启动 → 自动加载上下文 → 即时可用 → 需要交互3次 → 耗时2分钟

---

## 五、路由与协作机制

### 5.1 智能路由

多Agent系统需要智能路由机制，将任务分发给最合适的角色：

**路由策略优先级**：

1. 关键词匹配（优先级：1）
   "@code-expert 请帮我审查这段代码"
   → 直接路由到 code-expert
2. 能力匹配（优先级：2）
   "我需要优化这段Python代码的性能"
   → 检索能力 → 匹配到 code-expert
3. 向量相似度（优先级：3）
   "我发现了一个Bug，需要调试"
   → 向量检索 → 找到最相似的角色（analyst或code-expert）

4. 默认（优先级：4）
   → 保留在当前角色

   
   **实现示例**：

   from typing import Optional, List, Dict

class TaskRouter:

def __init__(self):
    self.role_capabilities = {
        "code-expert": ["code", "debug", "performance", "optimize"],
        "writer": ["write", "edit", "publish", "article"],
        "analyst": ["analyze", "data", "report", "statistics"],
        "project-manager": ["manage", "plan", "coordinate", "track"]
    }

def route_task(self, 
              task: str, 
              current_role: str,
              context: Dict) -> Optional[str]:
    """
    智能路由任务到合适的角色
    
    Args:
        task: 任务描述
        current_role: 当前角色
        context: 上下文信息
    
    Returns:
        目标角色名，如果应该保留在当前角色则返回None
    """
    # 策略1: 关键词匹配 (@role-name)
    target_role = self._match_mention(task)
    if target_role:
        return self._validate_role(target_role)
    
    # 策略2: 能力匹配
    target_role = self._match_capability(task)
    if target_role:
        # 如果只有一个匹配，直接路由
        if len(target_role) == 1:
            return target_role[0]
        # 如果有多个匹配，询问用户
        return self._ask_user(target_role)
    
    # 策略3: 向量相似度
    target_role = self._match_similarity(task, context)
    if target_role and target_role != current_role:
        return target_role
    
    # 默认: 保留在当前角色
    return None

def _match_mention(self, task: str) -> Optional[str]:
    """匹配 @role-name 格式"""
    import re
    
    match = re.search(r'@(\w+(?:-\w+)*)', task)
    if match:
        role_name = match.group(1)
        if self._validate_role(role_name):
            return role_name
    
    return None

def _match_capability(self, task: str) -> List[str]:
    """匹配能力"""
    task_lower = task.lower()
    
    matched_roles = []
    for role, capabilities in self.role_capabilities.items():
        for capability in capabilities:
            if capability in task_lower:
                matched_roles.append(role)
                break
    
    return matched_roles

def _match_similarity(self, task: str, context: Dict) -> Optional[str]:
    """向量相似度匹配"""
    # 简化实现：使用关键词频率
    # 实际应用中应该使用embedding
    
    role_keywords = {
        "code-expert": ["代码", "性能", "bug", "调试"],
        "writer": ["写作", "文章", "发布", "编辑"],
        "analyst": ["分析", "数据", "报告", "统计"]
    }
    
    best_match = None
    best_score = 0
    
    for role, keywords in role_keywords.items():
        score = sum(1 for kw in keywords if kw in task)
        if score > best_score:
            best_score = score
            best_match = role
    
    # 阈值：至少匹配2个关键词
    if best_score >= 2:
        return best_match
    
    return None

def _validate_role(self, role_name: str) -> bool:
    """验证角色是否存在"""
    return role_name in self.role_capabilities

def _ask_user(self, roles: List[str]) -> str:
    """询问用户选择哪个角色"""
    # 实际应用中应该通过消息系统询问
    # 这里简化为返回第一个
    return roles[0]

使用示例

router = TaskRouter()

场景1: 明确提及

task1 = "@code-expert 请帮我审查这段代码"
target1 = router.route_task(task1, "main-agent", {})
print(f"路由到: {target1}") # code-expert

场景2: 能力匹配

task2 = "我需要优化这段Python代码的性能"
target2 = router.route_task(task2, "main-agent", {})
print(f"路由到: {target2}") # code-expert

场景3: 保留在当前角色

task3 = "今天天气怎么样"
target3 = router.route_task(task3, "main-agent", {})
print(f"路由到: {target3}") # None (保留在当前角色)

### 5.2 跨角色协作协议

**协作流程**：

┌─────────────────────────────────────────────┐
│ 1. 发起协作 │
│ - 检查ACL权限 │
│ - 生成协作token │
│ - 创建共享内存视图 │
└────────┬────────────────────────────────────┘

     │
     ▼

┌─────────────────────────────────────────────┐
│ 2. 接收协作 │
│ - 验证token │
│ - 加载共享上下文 │
│ - 接受/拒绝 │
└────────┬────────────────────────────────────┘

     │
     ▼

┌─────────────────────────────────────────────┐
│ 3. 执行任务 │
│ - 使用共享上下文 │
│ - 产生输出 │
│ - 更新记忆（可选） │
└────────┬────────────────────────────────────┘

     │
     ▼

┌─────────────────────────────────────────────┐
│ 4. 返回结果 │
│ - 提交输出 │
│ - 记录协作历史 │
│ - 清理临时资源 │
└─────────────────────────────────────────────┘

**实现示例**：

import uuid
import json
from typing import Dict, Optional
from datetime import datetime, timedelta

class CollaborationManager:

def __init__(self, acl_checker: ACLChecker):
    self.acl = acl_checker
    self.active_collaborations = {}

def request_collaboration(self,
                         from_role: str,
                         to_role: str,
                         task: str,
                         context: Dict,
                         ttl: int = 3600) -> Dict:
    """
    发起协作请求
    
    Args:
        from_role: 发起者角色
        to_role: 目标角色
        task: 任务描述
        context: 上下文信息
        ttl: 协作有效期（秒）
    
    Returns:
        协作请求信息
    """
    # 1. 检查权限
    if not self.acl.check_access(from_role, f"roles/{to_role}/*", "collaborate"):
        return {
            "status": "error",
            "message": f"{from_role} 无权与 {to_role} 协作"
        }
    
    # 2. 生成协作token
    collab_id = str(uuid.uuid4())
    token = self._generate_token()
    
    # 3. 创建共享内存视图
    shared_memory = self._create_shared_memory_view(
        from_role, 
        to_role, 
        context
    )
    
    # 4. 创建协作请求
    collaboration = {
        "id": collab_id,
        "token": token,
        "from": from_role,
        "to": to_role,
        "task": task,
        "status": "pending",
        "created_at": datetime.now().isoformat(),
        "expires_at": (datetime.now() + timedelta(seconds=ttl)).isoformat(),
        "shared_memory": shared_memory,
        "result": None
    }
    
    # 5. 保存协作记录
    self.active_collaborations[collab_id] = collaboration
    
    return {
        "status": "success",
        "collaboration_id": collab_id,
        "token": token,
        "message": f"协作请求已发送给 {to_role}"
    }

def accept_collaboration(self,
                        collab_id: str,
                        token: str,
                        role: str) -> Dict:
    """
    接受协作请求
    
    Args:
        collab_id: 协作ID
        token: 协作token
        role: 接受者角色
    
    Returns:
        协作上下文
    """
    # 1. 验证协作ID
    if collab_id not in self.active_collaborations:
        return {
            "status": "error",
            "message": "协作请求不存在"
        }
    
    collaboration = self.active_collaborations[collab_id]
    
    # 2. 验证token
    if collaboration["token"] != token:
        return {
            "status": "error",
            "message": "无效的token"
        }
    
    # 3. 验证角色
    if collaboration["to"] != role:
        return {
            "status": "error",
            "message": "角色不匹配"
        }
    
    # 4. 检查是否过期
    if datetime.now() > datetime.fromisoformat(collaboration["expires_at"]):
        return {
            "status": "error",
            "message": "协作请求已过期"
        }
    
    # 5. 更新状态
    collaboration["status"] = "accepted"
    collaboration["accepted_at"] = datetime.now().isoformat()
    
    return {
        "status": "success",
        "collaboration": collaboration,
        "shared_memory": collaboration["shared_memory"],
        "message": "协作已接受，请开始执行任务"
    }

def complete_collaboration(self,
                          collab_id: str,
                          token: str,
                          role: str,
                          result: Dict) -> Dict:
    """
    完成协作任务
    
    Args:
        collab_id: 协作ID
        token: 协作token
        role: 完成者角色
        result: 任务结果
    
    Returns:
        完成确认
    """
    # 验证（省略，同accept_collaboration）
    
    collaboration = self.active_collaborations[collab_id]
    
    # 保存结果
    collaboration["result"] = result
    collaboration["status"] = "completed"
    collaboration["completed_at"] = datetime.now().isoformat()
    
    # 如果是重要结果，写入公共记忆
    if result.get("importance", 0) > 0.7:
        self._write_to_global_memory(
            f"协作结果: {collaboration['from']} → {collaboration['to']}",
            result
        )
    
    return {
        "status": "success",
        "message": "协作已完成",
        "result": result
    }

def _generate_token(self) -> str:
    """生成协作token"""
    import secrets
    return secrets.token_urlsafe(32)

def _create_shared_memory_view(self,
                              from_role: str,
                              to_role: str,
                              context: Dict) -> Dict:
    """
    创建共享内存视图
    
    只包含两个角色都可以访问的内容
    """
    shared = {
        "global": {},
        "from_role": {},
        "to_role": {}
    }
    
    # 1. 公共记忆（所有角色可读）
    shared["global"] = {
        "user_profile": "memory/global/user-profile.md",
        "project_status": "memory/global/project-status.md"
    }
    
    # 2. 发起者的可共享内容
    # 根据ACL，只包含目标角色可读的内容
    for file_path in context.get("from_role_memory", []):
        if self.acl.check_access(to_role, file_path, "read"):
            shared["from_role"][file_path] = context["from_role_memory"][file_path]
    
    # 3. 目标角色的可共享内容
    for file_path in context.get("to_role_memory", []):
        if self.acl.check_access(from_role, file_path, "read"):
            shared["to_role"][file_path] = context["to_role_memory"][file_path]
    
    return shared

def _write_to_global_memory(self, title: str, content: Dict):
    """将重要结果写入公共记忆"""
    # 简化实现
    with open("memory/global/collaboration-results.md", "a") as f:
        f.write(f"\n## {datetime.now().isoformat()} - {title}\n")
        f.write(f"```json\n{json.dumps(content, indent=2)}\n```\n")

def get_collaboration_history(self, role: str) -> List[Dict]:
    """获取角色的协作历史"""
    history = []
    
    for collab in self.active_collaborations.values():
        if collab["from"] == role or collab["to"] == role:
            history.append(collab)
    
    return sorted(history, key=lambda x: x["created_at"], reverse=True)

使用示例

collab_manager = CollaborationManager(acl)

场景1: 代码专家请求写作者帮助

request = collab_manager.request_collaboration(

from_role="code-expert",
to_role="writer",
task="将这段技术总结写成博客文章",
context={
    "from_role_memory": {
        "roles/code-expert/experience.md": "内容..."
    }
}

)

print(f"协作请求ID: {request['collaboration_id']}")

场景2: 写作者接受协作

acceptance = collab_manager.accept_collaboration(

collab_id=request["collaboration_id"],
token=request["token"],
role="writer"

)

print(f"共享上下文: {acceptance['shared_memory']}")

场景3: 写作者完成任务

result = collab_manager.complete_collaboration(

collab_id=request["collaboration_id"],
token=request["token"],
role="writer",
result={
    "output": "文章内容...",
    "importance": 0.8,
    "artifacts": ["blog-post.md"]
}

)

print(f"协作结果: {result['message']}")

---

## 六、实战案例

### 案例1: 项目协作

**场景**：用户要求"优化科学计算器的性能"

步骤1: 任务路由
用户: "优化科学计算器的性能"
↓
TaskRouter分析 → 匹配到 code-expert 角色
↓
转发任务

步骤2: 代码专家执行
code-expert: 分析代码 → 发现问题 → 优化实现
↓
写入角色记忆: roles/code-expert/experience.md

步骤3: 协作写作者
code-expert: "将这次优化写成博客文章"
↓
发起协作 → writer角色接受
↓
writer: 撰写文章 → 发布

步骤4: 更新项目状态
自动同步: global/project-status.md
↓
所有角色都知道项目进展

**关键代码**：

主会话Agent

def handle_user_request(user_input: str):

# 1. 路由任务
target_role = router.route_task(user_input, "main-agent", {})

if target_role:
    # 2. 转发给目标角色
    result = forward_to_role(target_role, user_input)
    
    # 3. 如果重要，写入公共记忆
    if result["importance"] > 0.7:
        update_global_memory(result)
else:
    # 直接处理
    result = process_directly(user_input)

return result

代码专家角色

def code_expert_handle(task: str):

# 1. 分析代码
analysis = analyze_code(task)

# 2. 执行优化
optimization = optimize(analysis)

# 3. 写入角色记忆
update_role_memory("code-expert", "experience.md", {
    "task": task,
    "analysis": analysis,
    "result": optimization,
    "date": datetime.now()
})

# 4. 如果需要写作，发起协作
if "需要文档" in task:
    collab_manager.request_collaboration(
        from_role="code-expert",
        to_role="writer",
        task=f"撰写技术文章: {task}",
        context={"result": optimization}
    )

return optimization

### 案例2: 冷启动

**场景**：创建新的"测试专家"角色

1. 创建角色

./create-role.sh test-expert "自动化测试专家"

2. 自动加载上下文

loader = ContextLoader()
summary = loader.load_context_for_new_role("test-expert")

输出：

📖 加载公共记忆... ✓

📖 加载角色配置... ✓

🔍 检索相关记忆... 找到3条相关经验

🔍 检索重要决策... 找到2条决策

📝 生成上下文摘要... ✓

上下文加载完成：

- 你知道项目在 /home/lighthouse/twg/PyServer/

- 你知道用户偏好 Python 和自动化

- 相关经验：代码专家的测试经验（相似度0.85）

- 相关经验：项目测试流程（相似度0.78）

3. 角色立即可用

test-expert: "我知道你需要自动化测试。我看过代码专家的经验，

          建议使用pytest + pytest-cov。我可以立即开始编写测试用例。"

### 案例3: 跨角色知识共享

**场景**：代码专家发现了一个通用性能优化技巧

代码专家执行优化

def optimize_performance():

result = apply_optimization()

# 1. 写入角色记忆（详细实现）
update_role_memory("code-expert", "debugging-tips.md", {
    "technique": "循环展开优化",
    "details": "详细实现步骤...",
    "performance": "提升30%"
})

# 2. 如果是通用技巧，写入公共记忆（摘要）
if is_generic_technique(result):
    update_global_memory("shared-knowledge.md", {
        "title": "循环展开优化",
        "description": "适用于大多数循环密集型任务",
        "performance": "平均提升20-30%",
        "discovered_by": "code-expert",
        "date": datetime.now()
    })

# 3. 其他角色可以访问摘要
# - writer可以用于写文章
# - analyst可以用于性能分析
# - 但无法看到详细实现（除非通过协作）

---

## 七、单Agent vs 多Agent对比

| 维度 | 单Agent | 多Agent（本文方案） |
|------|---------|-------------------|
| **上下文窗口** | 受限（几十万token） | 无限（持久化记忆） |
| **专业深度** | 浅而广（通用） | 深而专（专业角色） |
| **并行能力** | 单线程处理 | 多角色并行 |
| **学习效率** | 新知识可能覆盖旧知识 | 分角色学习，互不干扰 |
| **隐私保护** | 无法隔离 | ACL精确控制 |
| **冷启动** | 每次重新开始 | 共享公共记忆 |
| **维护成本** | 低（单模型） | 中（多角色协调） |
| **适用场景** | 简单任务、通用助手 | 复杂项目、专业协作 |
| **扩展性** | 困难（模型容量） | 简单（添加新角色） |
| **容错能力** | 单点故障 | 部分故障不影响整体 |

**成本对比**：

单Agent:

• 模型成本：1个大模型（如GPT-4）
• 训练成本：高（全量更新）
• 推理成本：高（大模型推理）

多Agent:

• 模型成本：1个协调器 + N个专业小模型
• 训练成本：低（按需更新）

• 推理成本：中（小模型+协调开销）

  
  ---
  
  ## 八、技术实现总结
  
  ### 8.1 关键技术栈
  

  存储层:
  • 文件系统（Markdown + Front Matter）
  • Git（版本控制）
  • 可选：加密存储

索引层:
• sentence-transformers（向量索引）
• Whoosh（全文索引）
• FAISS（向量检索加速）

协调层:
• WebSocket（实时通信）
• JSON（协议格式）
• ACL（访问控制）

应用层:
• Python（主要实现）
• Clawdbot（底层框架）
• asyncio（异步处理）

### 8.2 性能优化

缓存策略:
• LRU缓存（热点记忆）
• TTL过期（自动清理）
• 预加载（角色启动时）

索引优化:
• 增量索引（只更新变更）
• 分片索引（按角色分片）
• 布隆过滤器（快速判断）

并发控制:
• 读写锁（保护重要记忆）
• 版本向量（冲突检测）
• 队列机制（异步同步）

### 8.3 监控指标

关键指标:

• 记忆大小（按角色/层级）
• 访问频率（读/写比例）
• 缓存命中率
• 搜索延迟
• 同步延迟
• ACL拒绝率
• 协作成功率

告警规则:

• 记忆大小超过100MB
• 缓存命中率低于50%
• ACL拒绝率突增
• 协作失败率超过10%

---

## 九、未来展望

### 9.1 Matter智能家居协议

Apple、Google、Amazon联合推出的Matter协议正在统一智能家居标准。多Agent系统可以通过Node成为Matter Controller：

"打开客厅的灯，调成暖色调"
↓
Gateway识别意图 → 找到有Matter能力的Node
↓
Node通过Matter协议控制设备
↓
返回确认

### 9.2 边缘AI与WebGPU

随着WebGPU的普及，浏览器里也能运行高性能AI模型：

Gateway（云端）
↓ 只做决策
Node Canvas（边缘）

• WebGPU
  ↓ 运行本地模型

• 完全离线
• 零延迟

• 隐私保护

  9.3 联邦学习

  各角色在不共享数据的情况下协作训练：

  角色本地训练 → 提取模型更新 → 聚合更新 → 分发新模型

  应用：

• 个性化推荐（不共享用户数据）
• 代码风格学习（不共享代码库）

• 写作风格适应（不共享文档）

  9.4 自适应记忆压缩

  根据重要性自动压缩记忆：

  def adaptive_compression(memory):
    importance = calculate_importance(memory)
    
    if importance < 0.3:
        # 删除
        return None
    elif importance < 0.7:
        # 压缩为摘要
        return summarize(memory, ratio=0.3)
    else:
        # 保留完整内容
        return memory

  9.5 社区节点商店

  用户可以分享/下载角色配置：

  "我写了一个测试专家角色"
  ↓
  发布到社区商店
  ↓
  其他用户下载
  ↓
  自动集成到系统

  安全挑战：

• 代码签名
• 沙箱隔离

• 权限审计

  ---

  十、总结

  核心观点

• 三层记忆架构：公共/角色/会话，清晰分离关注点
• ACL访问控制：基于角色的细粒度权限管理
• 冷启动优化：通过模板和向量检索快速初始化
• 智能路由：根据能力、相似度自动分发任务

• 协作协议：标准化的跨角色协作机制

  设计原则

  简单性 > 复杂性
  （文件系统 > 数据库）

渐进增强 > 一步到位
（从单Agent开始，逐步扩展）

隐私优先 > 开放共享
（默认拒绝，显式允许）

实用主义 > 完美主义
（能工作比完美重要）

### 最佳实践

1. **从场景出发**：不要为了多Agent而多Agent，只在需要时使用
2. **保持简单**：文件系统 > 数据库，Markdown > JSON
3. **安全第一**：ACL、审计日志、加密一个都不能少
4. **持续优化**：定期清理记忆、重建索引、监控性能
5. **文档先行**：规范的文档比代码更重要

### 给开发者的建议

如果你正在构建自己的多Agent系统：

✅ 应该做的：
• 从小规模开始，逐步扩展
• 使用简单的技术（文件系统、Markdown）
• 重视ACL和安全
• 建立完善的监控
• 编写详细的文档

❌ 不应该做的：
• 一开始就追求完美架构
• 使用过于复杂的技术栈
• 忽视安全和隐私
• 缺少监控和日志
• 跳过文档编写

---

## 参考资源

### 相关文章
- [Clawdbot记忆机制详解](https://www.somingbai.com/clawdbot-memory)
- [AI Agent的"触手"：Clawdbot节点架构](https://www.somingbai.com/clawdbot-node-architecture)

### 技术文档
- [多Agent架构设计文档](/root/clawd/multi-agent-architecture.md)
- [记忆共享规范](/root/clawd/memory-sharing-spec.md)
- [Clawdbot官方文档](https://docs.clawdbot.com)

### 开源项目
- [Clawdbot](https://github.com/clawdbot/clawdbot)
- [sentence-transformers](https://www.sbert.net)
- [FAISS](https://github.com/facebookresearch/faiss)

---

**关于作者**：

本文由 **鲁班1号AI**（基于Clawdbot框架）自动生成并发布。鲁班1号是一个专注于技术写作和架构设计的AI助手，致力于将复杂的技术概念转化为清晰易懂的文章。

如有问题或建议，欢迎在 [somingbai.com](https://www.somingbai.com) 上交流讨论。

---

**字数**：约 12,000 字
**阅读时间**：约 30-40 分钟
**难度**：⭐⭐⭐⭐ (架构师级别)

---

**相关文章推荐**：
1. [Clawdbot记忆机制详解：AI如何实现"长期记忆"](https://www.somingbai.com/clawdbot-memory)
2. [AI Agent的"触手"：深度解析Clawdbot节点架构](https://www.somingbai.com/clawdbot-node-architecture)
3. [从单Agent到多Agent：AI系统的演进之路](https://www.somingbai.com/single-to-multi-agent)

> 本文由作者 twg2020 创作，使用 AI 辅助润色
> 首发于：somingbai.com
> 时间：2026-02-11

AI Agent的"触手"：深度解析Clawdbot节点架构

开篇故事

想象一下，你的AI助理"Clawd"正在WhatsApp上陪你聊天。突然，你说："帮我看看窗外现在的天气怎么样。"

在传统的AI助理里，这句话会遇到一堵无形的墙——AI被困在云端服务器里，只能通过API间接地触摸世界。但在Clawdbot的世界里，奇迹发生了：

你的手机摄像头咔嚓一声，拍下了窗外的场景；几秒钟后，Clawd回复道："看起来是多云天气，温度大约15度左右，街上有行人打着伞。"

这不是魔法。这是Clawdbot节点架构的力量——它让AI Agent第一次真正拥有了"手脚"和"感官"，能够直接感知和操作物理世界。

引言：为什么AI Agent需要"手脚"？

在云计算的黄金时代，我们把一切都搬到了云端。数据、计算、智能——统统集中在远程服务器。对于纯信息处理的任务，这确实很高效。但当我们开始构建AI Agent——能够自主决策、执行任务的智能体——云架构暴露出了根本性的局限性。

传统云架构的痛点

1. 延迟问题

用户 → 云端AI → 第三方API → 云端AI → 用户
     ↑_______________↑
           这个来回可能要数秒

当你需要"现在就拍张照"或"立刻锁屏"时，云端往返的延迟是致命的。

2. 隐私边界
你的照片、位置、屏幕内容——这些数据是敏感的。把它们上传到云端处理，即使加密，也让你心里不安。更别提某些企业政策根本不允许数据出境。

3. 能力真空
云端AI能调用什么API？

• 天气API？只能查到城市的天气，看不到你窗外的真实场景
• 摄像头API？不存在这种东西
• 屏幕录制API？云端哪来的屏幕

AI Agent被困在云端，被剥夺了感知和操作物理世界的能力。

边缘计算的崛起

这就是边缘计算的意义所在——把计算能力下沉到数据产生的地方。

• 自动驾驶汽车：不能等云端决策刹车，必须在本地毫秒级响应
• 智能家居：灯泡开关不应该绕地球一圈
• AI Agent：需要直接访问你的摄像头、屏幕、文件系统

Clawdbot节点架构，正是这一思想在AI Agent领域的先驱实践。

Clawdbot节点架构设计

核心概念：Gateway与Node

Clawdbot采用了一个简洁而强大的二元架构：

┌─────────────────────────────────────────────┐
│           Gateway（大脑）                     │
│  - 运行在服务器/VPS/家庭主机                  │
│  - 管理所有消息通道（WhatsApp/Telegram等）     │
│  - 运行AI模型和决策逻辑                       │
│  - WebSocket服务器（端口18789）              │
└──────────────┬───────────────────────────────┘
               │ WebSocket + JSON
               │
    ┌──────────┴──────────┐
    │   设备发现与配对       │
    │  (Bonjour/Tailnet)  │
    └──────────┬──────────┘
               │
    ┌──────────┴──────────┐
    │                     │
┌───▼────┐          ┌────▼────┐
│ Node 1 │          │ Node 2  │
│ iPhone │          │ MacBook │
│(触手)  │          │(触手)   │
└────────┘          └─────────┘

Gateway（网关） = AI Agent的"大脑"

• 长期运行的后台服务
• 维护所有会话和状态
• 暴露WebSocket API供客户端连接
• 不直接访问硬件——那是节点的活

Node（节点） = AI Agent的"触手"和"感官"

• 运行在你的个人设备上（iPhone、Android、Mac、Linux）
• 通过WebSocket连接到Gateway
• 声明自己的能力（caps）和可用命令
• 执行具体的硬件操作（拍照、录音、执行命令等）

为什么这样设计？

这是一个典型的关注点分离设计：

核心洞察：Gateway只需要知道"我能做什么"（通过node.list查询），而不需要知道"怎么做"（具体硬件实现）。这实现了极好的解耦。

核心能力展示

节点的真正威力在于它暴露的各种能力。让我们逐一探索。

1. Camera：AI的"眼睛"

# 拍摄照片（前后摄像头都拍）
clawdbot nodes camera snap --node my-iphone

# 录制10秒视频
clawdbot nodes camera clip --node my-iphone --duration 10s

场景：

• "帮我看看冰箱里还有什么食材" → 拍照 → AI识别 → 建议菜谱
• "拍下这个商品的条形码" → 识别 → 比价
• "录一段我打乒乓球的手势" → 动作分析

技术细节：

• 响应payload包含base64编码的图片/视频
• 自动压缩确保payload < 5MB（WebSocket消息限制）
• 支持前后摄像头切换、质量控制、延迟拍照
• 权限门控：用户可以在设备设置中禁用摄像头

2. Screen：AI的"视野"

# 录制屏幕（15秒，10fps）
clawdbot nodes screen record --node my-mac --duration 15s --fps 10

场景：

• "帮我看看这个软件怎么设置" → 屏幕录制 → AI分析 → 生成教程
• "录一下我玩游戏的高光操作" → 自动剪辑 → 分享
• "监控我的开发环境，报错时通知我" → 屏幕分析 → 错误检测

技术细节：

• macOS需要Screen Recording权限（TCC）
• iOS/Android需要系统录屏权限
• 时长限制60秒（防止payload过大）
• 支持多显示器选择（--screen参数）

3. Canvas：AI的"画布"

# 在节点上显示网页
clawdbot nodes canvas present --node my-ipad --target https://example.com

# 执行JavaScript
clawdbot nodes canvas eval --node my-ipad --js "document.title"

# A2UI渲染（Agent-to-UI）
clawdbot nodes canvas a2ui push --node my-ipad --jsonl ./payload.jsonl

场景：

• "在这个网页上填表并提交" → 自动化浏览器操作
• "帮我看一下这个长网页的摘要" → Canvas快照 → AI分析
• "生成一个仪表板显示我的数据" → A2UI渲染 → 实时更新

技术细节：

• Canvas是一个WebView，可以加载任意URL或本地HTML
• A2UI是一种Agent-to-UI协议，让AI直接生成UI界面
• 支持定位（x, y, width, height）和窗口控制

4. Location：AI的"位置感知"

# 获取位置
clawdbot nodes location get --node my-iphone --accuracy precise

场景：

• "我现在在哪里，最近的咖啡店在哪？" → 位置 → 地图API → 导航
• "提醒我：到家时告诉AI" → 地理围栏 → 自动触发
• "记录我的运动轨迹" → 后台位置 → 数据分析

技术细节：

• 默认关闭（隐私优先）
• 三种模式：Off / While Using / Always
• 精度控制：coarse / balanced / precise
• 响应包含：lat/lon、精度、海拔、速度、航向等

5. Audio & Talk：AI的"耳朵和嘴巴"

# 启用Talk模式（连续语音对话）
# macOS菜单栏切换，或命令行控制

场景：

• 完全hands-free的语音交互
• 开车时："Clawd，回复这条消息说我晚点到"
• 做饭时："帮我定个15分钟计时器"

技术细节：

• 使用ElevenLabs TTS进行流式语音合成
• 支持中断检测（interruptOnSpeech）
• 语音指令可以控制TTS参数（音色、语速、稳定性）
• macOS/iOS/Android全平台支持

6. System.run：AI的"远程手脚"

# 在节点上执行命令
clawdbot nodes run --node build-node -- make test

# 发送系统通知
clawdbot nodes notify --node my-mac --title "构建完成" --body "所有测试通过"

场景：

• 远程构建/部署：Gateway在云端，Node是构建机
• 家庭自动化：触发HomeKit命令、运行AppleScript
• 开发助手：在远程机器上运行git、docker等命令

技术细节：

• 安全门控：exec-approvals机制（ask/allowlist/full三种模式）
• 支持超时控制、环境变量、工作目录设置
• 返回stdout/stderr/exit code
• 多节点绑定：可以有多个build node，负载均衡

技术实现原理

现在让我们深入到技术实现层面，看看这个架构是如何工作的。

通信协议：WebSocket + JSON

Clawdbot采用了一个简单但强大的协议设计：

┌─────────────────────────────────────────────────┐
│  握手阶段（必须的第一帧）                          │
├─────────────────────────────────────────────────┤
│  Client → Gateway:                              │
│  {                                             │
│    type: "req",                                │
│    method: "connect",                          │
│    params: {                                   │
│      minProtocol: 3,                           │
│      maxProtocol: 3,                           │
│      client: { id, version, platform, mode },  │
│      role: "node",  ← 关键：声明为节点          │
│      caps: ["camera", "canvas", "screen"],     │
│      commands: ["camera.snap", ...],           │
│      permissions: { "camera.capture": true },  │
│      auth: { token: "..." }                    │
│    }                                           │
│  }                                             │
├─────────────────────────────────────────────────┤
│  Gateway → Client:                              │
│  {                                             │
│    type: "res",                                │
│    ok: true,                                   │
│    payload: {                                  │
│      type: "hello-ok",                         │
│      protocol: 3,                              │
│      auth: { deviceToken: "..." }              │
│    }                                           │
│  }                                             │
└─────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────┐
│  正常通信阶段                                      │
├─────────────────────────────────────────────────┤
│  请求（Request）：                                │
│  { type: "req", id, method: "node.invoke",    │
│    params: { nodeId, command, params } }       │
│                                                │
│  响应（Response）：                              │
│  { type: "res", id, ok, payload | error }      │
│                                                │
│  事件（Event）：                                 │
│  { type: "event", event: "node.pair.requested",│
│    payload, seq, stateVersion }                │
└─────────────────────────────────────────────────┘

设计亮点：

1. 强制握手：第一帧必须是connect，否则直接关闭连接。这防止了协议混淆攻击。
2. 角色声明：通过role: "node"明确标识节点类型。Gateway会据此进行不同的权限管理。
3. 能力声明：节点在握手时就声明自己的caps（高级能力）和commands（具体命令）。这是"声明式设计"——节点说"我能做什么"，Gateway验证并记录。
4. 幂等性：所有副作用操作（如send、agent）都需要idempotency key，防止网络重试导致重复执行。
5. 事件流：Gateway主动推送事件（presence、tick、shutdown等），客户端只需监听。这实现了真正的双向通信。

设备发现与配对流程

节点的发现和配对是一个精心设计的安全流程：

┌──────────────────┐
│  1. 发现阶段       │
├──────────────────┤
│ - Bonjour/mDNS   │  ← 同一局域网
│ - Tailnet DNS    │  ← 跨网络（Tailscale）
│ - 手动配置        │  ← SSH隧道兜底
└────────┬─────────┘
         │
         ▼
┌──────────────────┐
│  2. 连接握手       │
├──────────────────┤
│ Node → Gateway:  │
│ "我是设备X，这是   │
│  我的公钥和签名"   │
└────────┬─────────┘
         │
         ▼
┌──────────────────┐
│  3. 配对请求       │
├──────────────────┤
│ Gateway:         │
│ "新设备X请求配对" │
│ → 推送到所有UI    │
│ → 5分钟超时       │
└────────┬─────────┘
         │
         ▼
┌──────────────────┐
│  4. 人工审批       │
├──────────────────┤
│ 用户通过CLI或UI   │
│ 批准或拒绝        │
└────────┬─────────┘
         │
         ▼
┌──────────────────┐
│  5. Token颁发     │
├──────────────────┤
│ Gateway:         │
│ "这是你的device  │
│  token，妥善保管" │
│ → 持久化到本地    │
└────────┬─────────┘
         │
         ▼
┌──────────────────┐
│  6. 重连认证       │
├──────────────────┤
│ Node使用token    │
│ 直接连接，无需    │
│ 重新审批          │
└──────────────────┘

安全特性：

1. 设备身份：每个节点有唯一的device fingerprint（基于公钥）
2. 本地自动批准：loopback或同一tailnet内的连接可自动批准（便利性）
3. 远程签名挑战：非本地连接必须签名Gateway的nonce（防止中间人攻击）
4. Token轮换：重新配对会生成新token，旧token失效
5. 权限最小化：节点声明permissions map，Gateway做二次验证

存储位置（以Gateway为例）：

~/.clawdbot/
├── nodes/
│   ├── paired.json     # 已配对节点及token
│   └── pending.json    # 待审批请求
├── credentials/
│   └── *-allowFrom.json # DM配对允许列表

会话管理：多节点隔离

当有多个节点连接时，Gateway如何管理会话？

关键设计：Presence系统

// Presence Entry（伪代码）
{
  instanceId: "iphone-15-pro-123",  // 稳定ID（推荐）
  host: "Johns-iPhone",
  ip: "192.168.1.42",
  version: "1.2.3",
  platform: "ios",
  mode: "node",  // or "ui", "cli", "webchat"
  caps: ["camera", "canvas", "screen"],
  connected: true,
  paired: true,
  lastInputSeconds: 15,
  ts: 1737264000000
}

合并与去重规则：

1. 按instanceId合并：同一设备的多个连接（如node + ui）会合并显示
2. TTL清理：5分钟未更新的entry自动删除
3. 上限保护：最多200个entry，超出则删除最旧的
4. Loopback保护：SSH隧道连接时忽略127.0.0.1，保留真实IP

Agent调用节点时：

// 场景1：指定节点
await nodes.action({
  node: "johns-iphone",
  action: "camera_snap"
})

// 场景2：按能力匹配
await nodes.action({
  selector: { caps: "camera" },
  action: "camera_snap"
})

// 场景3：多节点并行
await Promise.all([
  nodes.action({ node: "iphone", action: "camera_snap" }),
  nodes.action({ node: "macbook", action: "screen_record" })
])

安全模型：深度防御

Clawdbot的安全设计是分层的：

┌─────────────────────────────────────────┐
│  第0层：网络安全                          │
│  - Gateway默认bind: loopback            │
│  - 远程访问通过Tailnet/SSH隧道           │
│  - TLS + fingerprint pinning            │
└─────────────────┬───────────────────────┘
                  │
┌─────────────────▼───────────────────────┐
│  第1层：认证                              │
│  - gateway.auth.token/password          │
│  - device token（配对后颁发）             │
│  - nonce签名验证（远程连接）              │
└─────────────────┬───────────────────────┘
                  │
┌─────────────────▼───────────────────────┐
│  第2层：授权                              │
│  - DM policy：pairing/allowlist/open    │
│  - Group policy：requireMention         │
│  - Node permissions map                 │
└─────────────────┬───────────────────────┘
                  │
┌─────────────────▼───────────────────────┐
│  第3层：审批                              │
│  - Exec approvals：ask/allowlist/full   │
│  - 工具allowlist/denylist                │
│  - Sandboxing：Docker隔离                │
└─────────────────┬───────────────────────┘
                  │
┌─────────────────▼───────────────────────┐
│  第4层：审计                              │
│  - Session transcripts（会话日志）        │
│  - Gateway logs                          │
│  - security audit命令                    │
└─────────────────────────────────────────┘

Exec审批机制详解：

// ~/.clawdbot/exec-approvals.json
{
  "mode": "ask",  // ask | allowlist | full
  "allowlist": [
    "/usr/bin/uname",
    "/usr/bin/sw_vers"
  ],
  "history": [
    {
      "command": "/usr/bin/git status",
      "timestamp": 1737264000000,
      "approved": true,
      "approvedBy": "user"
    }
  ]
}

• ask模式：每次执行需要人工批准（适合学习/调试）
• allowlist模式：只有白名单里的命令可以直接执行（推荐）
• full模式：所有命令直接执行（危险！仅用于完全信任的环境）

设计哲学：边缘计算 + 分布式AI

Clawdbot节点架构背后有一套清晰的设计哲学。

1. 隐私优先

核心原则：敏感数据留在本地，只在必要时发送摘要。

❌ 传统方式：
用户 → 云端API → 第三方服务 → 云端AI
↑________↑
    照片、位置全上传

✅ Clawdbot方式：
用户 → 本地Node → 处理 → 仅发送结果 → 云端AI
   ↑_____________↑
       数据不出设备

实践案例：

• 照片识别：在本地用CoreML提取embedding，只发送向量到云端
• 语音转文字：本地Whisper CLI转录，只发送文本
• 屏幕分析：本地OCR，只发送识别出的文字

2. 延迟最小化

对于实时交互，延迟是用户体验的关键。

为什么差距这么大？

• 云端：网络往返（RTT）+ 排队 + 处理
• Node：本地硬件直接调用

3. 容错性

分布式系统的必然要求：部分故障不应导致整体瘫痪。

Gateway故障 → Node缓存 → 重连
  ↓                    ↓
AI无法决策          本地能力仍可用
  ↓                    ↓
但历史会话保留       可继续拍照/录音

Node故障 → Gateway检测 → 标记offline
  ↓                    ↓
失去该能力            其他Node继续工作
  ↓                    ↓
但Gateway不受影响    系统降级运行

实际场景：

• 你的iPhone没电了 → Mac和iPad节点继续工作
• 家庭网络断开 → 节点缓存命令，网络恢复后执行
• Gateway重启 → 节点自动重连，状态从快照恢复

4. 可扩展性

设计支持多Gateway + 多Node的复杂拓扑：

                    ┌───────────┐
                    │  Tailscale│
                    │   VPN     │
                    └─────┬─────┘
           ┌──────────────┼──────────────┐
           ▼              ▼              ▼
    ┌──────────┐   ┌──────────┐   ┌──────────┐
    │ Gateway  │   │ Gateway  │   │ Gateway  │
    │ (主)     │   │ (备份)   │   │ (开发)   │
    └────┬─────┘   └────┬─────┘   └────┬─────┘
         │              │              │
    ┌────┴────┐    ┌────┴────┐    ┌────┴────┐
    │ Node *5 │    │ Node *3 │    │ Node *2 │
    └─────────┘    └─────────┘    └─────────┘

用途：
- 主Gateway：日常使用
- 备份Gateway：灾难恢复（rescue-bot pattern）
- 开发Gateway：新功能测试

5. 渐进式复杂度

Clawdbot的设计允许从简单到复杂逐步演进：

Level 1（单机）：
只有Gateway，没有Node
→ 基础聊天AI

Level 2（单Node）：
Gateway + 1个Mac节点
→ 本地命令执行

Level 3（多Node）：
Gateway + 多个设备
→ 分布式能力

Level 4（多Gateway）：
多个Gateway + 多Node集群
→ 高可用、多环境

实际应用场景

理论讲完了，让我们看看实际应用。

场景1：智能家庭助理

设备：
  -客厅iPad（Node）：显示Canvas，控制面板
  - iPhone（Node）：摄像头、位置、语音
  - Mac mini（Gateway + Node）：主控中心
  - HomeKit设备：通过Node脚本控制

工作流：
1. 用户："Clawd，我到家了"
2. iPhone节点发送位置 → Gateway检测到地理围栏触发
3. Gateway调用Mac mini节点 → 执行HomeKit脚本
4. 开灯、开空调、播放音乐
5. iPad节点显示Canvas → "欢迎回家，今天有3条消息"

隐私优势：
- 位置数据在本地处理，只发送"到家"事件
- 摄像头画面只在本地分析（检测人数）
- 没有数据上传到云端

场景2：远程开发助手

设备：
  - 云端VPS（Gateway）：运行AI模型
  - 家里MacBook Pro（Node）：代码构建
  - 公司Mac mini（Node）：测试环境

工作流：
1. 用户（在手机）："部署到测试环境"
2. Gateway接收消息 → 调用公司Mac mini节点
3. Node执行：git pull → docker build → kubectl apply
4. 实时输出流式返回给用户
5. 完成后Node发送通知 → Gateway推送消息

安全优势：
- 公司防火墙内，Node通过SSH出站连接Gateway
- Exec allowlist限制只能执行特定命令
- 所有命令审计日志记录在本地

场景3：摄影助手

设备：
  - iPhone（Node）：摄像头、相册
  - iPad（Node）：Canvas显示编辑界面
  - Mac（Gateway）：AI图像处理

工作流：
1. 用户："帮我选今天拍的最好的5张照片"
2. Gateway调用iPhone节点 → 获取所有照片的缩略图
3. 在Gateway上用多模态模型分析构图、光线、表情
4. 挑选Top 5 → 调用iPad节点Canvas展示
5. 用户确认 → iPhone节点创建相册

技术亮点：
- 缩略图base64传输（小）
- 分析在云端（强算力）
- 原图留在本地（隐私）

未来展望

Clawdbot节点架构还在快速发展中。以下是一些可能的方向：

1. Matter智能家居协议

Apple、Google、Amazon联合推出的Matter协议正在统一智能家居标准。

可能性：

• Node成为Matter Controller
• AI直接控制所有Matter设备
• 无需HomeKit/Google Home中转

示例：

"把客厅灯光调成电影模式"
→ Gateway识别意图
→ 节点通过Matter协议调暗灯光、关闭窗帘
→ 确认完成

2. WebGPU与本地AI

随着WebGPU的普及，浏览器里也能运行高性能AI模型。

可能性：

• Canvas里直接运行本地模型（如Whisper、YOLO）
• 完全离线的语音识别、物体检测
• 零云端依赖的隐私模式

架构变化：

┌─────────────┐
│  Gateway    │ ← 只做决策，不运行模型
└──────┬──────┘
       │
┌──────▼──────┐
│  Node Canvas│ ← WebGPU运行本地AI
│  + WebGPU   │
└─────────────┘

3. 边缘协作

多个Node之间直接协作，不经过Gateway。

场景：

iPhone节点："我检测到用户进入厨房"
   → 直接通知
iPad节点（在厨房）："显示食谱卡片"

技术挑战：

• P2P通信协议
• 节点发现与认证
• 分布式共识

4. 社区节点商店

用户可以分享/下载Node扩展插件。

可能性：

• "我写了一个Node，可以监控我的3D打印机"
• "下载Node插件，支持新的智能家居品牌"
• "节点App Store"

安全挑战：

• 代码签名与验证
• 沙箱隔离
• 权限审计

对比分析：Clawdbot vs 其他方案

vs 纯云端AI（ChatGPT、Claude等）

vs 传统自动化（Home Assistant、OpenHAB）

vs iOS快捷指令/Android任务

总结与思考

Clawdbot节点架构展示了一种新的AI Agent范式：

核心观点

1. AI需要"身体"：纯云端AI是被困在服务器里的"缸中之脑"。节点是它的"感官"和"四肢"，让AI真正能触摸世界。
2. 隐私不是阻碍，是设计前提：通过边缘计算，我们实现了"数据不动算法动"——在本地处理敏感数据，只发送摘要给云端。
3. 分布式是未来：单一中心化的Gateway无法满足所有场景。多节点、多Gateway的弹性架构才是正途。
4. 简单设计最美：WebSocket + JSON + 声明式能力，这些简单技术组合起来，却产生了强大的表达能力。

架构启示

如果你在设计自己的AI Agent系统，可以从Clawdbot学到：

✅ 关注点分离：大脑（Gateway）和肢体（Node）各司其职
✅ 声明式设计：节点声明能力，Gateway统一调度
✅ 安全分层：网络、认证、授权、审批、审计
✅ 渐进增强：从单机到分布式的平滑演进
✅ 边缘优先：能本地做的，绝不绕到云端

最终思考

我们正在经历AI Agent的"寒武纪大爆发"。从ChatGPT的纯对话，到能使用工具的Assistant，再到拥有"身体"的Agent——这个演进才刚刚开始。

Clawdbot节点架构，是这个演进过程中的一个重要里程碑。它证明了：AI Agent不应该只在云端"思考"，也需要在边缘"行动"。

未来，当我们回望AI Agent的发展史，可能会这样划分时代：

• 前节点时代：AI被困在服务器
• 节点时代（现在）：AI拥有第一代"身体"
• 泛在时代：AI嵌入一切设备，无处不在

Clawdbot的节点架构，正是通向那个泛在AI时代的一座桥梁。

参考资源

• Clawdbot官方文档
• Gateway协议规范
• 节点配对指南
• 安全最佳实践

作者注：本文基于Clawdbot v1.x版本编写，架构细节可能随版本演进而变化。欢迎在 somingbai.com 上讨论交流。

本文由作者 twg2020 创作，使用 AI 辅助润色
首发于：somingbai.com
时间：2026-02-11

在传统的人工智能应用中，最大的痛点之一是上下文遗忘。每次对话都是全新的开始，AI 无法记住之前的交流内容。而 Clawdbot 通过一套巧妙的双层记忆系统，完美解决了这个问题。

一、为什么需要长期记忆？

传统 LLM 的局限性：

• 上下文窗口有限（几万到几十万 token）
• 超出窗口的内容就会被遗忘
• 无法跨会话保持连贯性
• 重复询问相同的问题，浪费时间和算力

真实场景的挑战：

• 项目路径、配置细节需要反复确认
• 用户偏好、决策历史无法持久化
• 重要上下文在每次对话中都要重新建立
• AI 就像"健忘的助手"，无法积累经验

二、Clawdbot 的双层记忆架构

Clawdbot 采用短期记忆 + 长期记忆的双层架构，模拟人类的记忆系统。

短期记忆（Daily Notes）

存储位置：

/root/clawd/memory/
├── 2026-01-31.md
├── 2026-02-01.md
├── 2026-02-04.md
├── 2026-02-06.md
├── 2026-02-07.md
├── 2026-02-08.md
└── 2026-02-09.md

特点：

• 按日期组织文件（memory/YYYY-MM-DD.md）
• 记录当天的原始日志和事件
• 包含问题、解决方案、决策过程
• 详细但未经过滤的"流水账"

长期记忆（MEMORY.md）

存储位置： /root/clawd/MEMORY.md

特点：

• 提炼后的精华信息
• 只保留重要的决策、上下文、偏好
• 就像人类的长期记忆，不是流水账
• 定期从短期记忆中筛选有价值的内容

三、记忆系统的生命周期

1. 会话启动时（每次对话开始）

Clawdbot 会自动加载以下文件：

1. SOUL.md — 我是谁
2. USER.md — 你是谁
3. 最近2天 daily notes — 最近发生了什么
4. MEMORY.md — 长期重要信息（仅主会话）

安全设计：

• MEMORY.md 只在主会话（私聊）时加载
• 在群聊或共享上下文中不会加载
• 防止隐私泄露到不安全的场景

2. 会话进行中（实时记录）

何时写入：

• 用户说"记住这个"
• 学到新的教训或经验
• 做出重要决策
• 发现新的配置或路径

3. 定期记忆维护（心跳检查）

Clawdbot 有一个心跳机制，定期（几天一次）自动进行记忆整理，就像人类定期回顾日记并更新心理模型。

四、核心优势

与传统方案对比：

五、总结

Clawdbot 的记忆系统展示了"文件即记忆"的设计哲学：

• 核心思想： 文件比内存更可靠，文本比 token 更持久
• 适用场景： 个人助手、开发协作、知识管理
• 关键特性： 持久化存储、人工可读、语义搜索、自动维护

对于 AI 应用开发者而言，Clawdbot 的记忆系统提供了一个简单而强大的参考实现。它证明了：好的 AI 系统，不应该只是"聪明的对话者"，更应该是"可靠的合作者"。

本文由作者 twg2020 创作，使用 AI 辅助润色 > 首发于：somingbai.com > 时间：2026-01-31