OpenClaw：本地部署 AI Agent 实践

┌─────────────────────────────────────────────────────────────────┐
│                      OpenClaw 系统架构                           │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │                    外部渠道层                            │   │
│  │  ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐        │   │
│  │  │WhatsApp │ │Telegram │ │Discord  │ │ WebChat │        │   │
│  │  └────┬────┘ └────┬────┘ └────┬────┘ └────┬────┘        │   │
│  └───────┼──────────┼──────────┼──────────┼────────────────┘   │
│          │          │          │          │                    │
│          └──────────┴────┬─────┴──────────┘                    │
│                            ↓                                    │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │                   Gateway 网关层                         │   │
│  │  ┌─────────────────────────────────────────────────┐    │   │
│  │  │            gateway_v2 (端口: 18789)              │    │   │
│  │  │  • 消息路由     • 权限验证     • 会话管理        │    │   │
│  │  └─────────────────────────────────────────────────┘    │   │
│  └─────────────────────────────┬───────────────────────────┘   │
│                                ↓                                │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │                    Agent 核心层                          │   │
│  │  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐    │   │
│  │  │ 规划器   │ │ 记忆系统 │ │ 技能系统 │ │ 执行器   │    │   │
│  │  │(Planning)│ │(Memory)  │ │(Skills)  │ │(Action)  │    │   │
│  │  └──────────┘ └──────────┘ └──────────┘ └──────────┘    │   │
│  └─────────────────────────────┬───────────────────────────┘   │
│                                ↓                                │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │                    LLM 服务层                            │   │
│  │  ┌──────────┐ ┌──────────┐ ┌──────────┐                 │   │
│  │  │ DeepSeek │ │  Qwen3   │ │ 其他模型 │                 │   │
│  │  │   V3     │ │  Coder   │ │  (API)   │                 │   │
│  │  └──────────┘ └──────────┘ └──────────┘                 │   │
│  └─────────────────────────────────────────────────────────┘   │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

# gateway_v2/gateway_v2.py 核心逻辑
 
class Gateway:
    """
    OpenClaw 网关服务
    - 端口: 18789
    - 功能: 消息路由、权限验证、会话管理
    """
    
    def __init__(self):
        self.port = 18789
        self.sessions = SessionManager()
        self.skill_registry = SkillRegistry()
        
    async def handle_message(self, message: Message):
        """处理接收到的消息"""
        # 1. 解析消息来源
        channel = message.channel  # whatsapp/telegram/discord/webchat
        
        # 2. 获取或创建会话
        session = self.sessions.get_or_create(
            user_id=message.user_id,
            channel=channel
        )
        
        # 3. 权限验证
        if not self.auth.verify(message.user_id):
            return "抱歉，您没有权限使用此服务。"
        
        # 4. 消息过滤（避免循环）
        if message.content.startswith("[rust_im]"):
            return None  # 过滤 Rust IM 系统消息
        
        # 5. 调用 Agent 处理
        response = await self.agent.process(
            message=message.content,
            session=session
        )
        
        return response

┌─────────────────────────────────────────────────────────────┐
│                   OpenClaw 三层记忆系统                      │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌─────────────────────────────────────────────────────┐   │
│  │              🧠 永久记忆 (Permanent)                  │   │
│  │  ┌─────────────────────────────────────────────┐    │   │
│  │  │ 文件: IDENTITY.md + SOUL.md                 │    │   │
│  │  │ 作用: 定义 AI 核心身份和性格                 │    │   │
│  │  │ 特点: 注入系统提示词，不随会话改变           │    │   │
│  │  └─────────────────────────────────────────────┘    │   │
│  └─────────────────────────────────────────────────────┘   │
│                            │                                │
│                            ↓                                │
│  ┌─────────────────────────────────────────────────────┐   │
│  │              📚 长期记忆 (Long-term)                  │   │
│  │  ┌─────────────────────────────────────────────┐    │   │
│  │  │ 文件: MEMORY.md + 向量数据库                 │    │   │
│  │  │ 作用: 存储重要事件、学习成果、用户偏好        │    │   │
│  │  │ 特点: 跨会话持久化，通过关键词/向量检索       │    │   │
│  │  │ 写入时机: 重大决策、重要信息、学习成果        │    │   │
│  │  └─────────────────────────────────────────────┘    │   │
│  └─────────────────────────────────────────────────────┘   │
│                            │                                │
│                            ↓                                │
│  ┌─────────────────────────────────────────────────────┐   │
│  │              💬 短期记忆 (Short-term)                 │   │
│  │  ┌─────────────────────────────────────────────┐    │   │
│  │  │ 文件: memory/YYYY-MM-DD.md + SQLite 缓存    │    │   │
│  │  │ 作用: 当前会话上下文、临时信息               │    │   │
│  │  │ 特点: 近5轮核心会话，自动管理                │    │   │
│  │  └─────────────────────────────────────────────┘    │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

# workspace/AGENTS.md 定义的内存管理规则
 
class MemorySystem:
    """
    OpenClaw 三层记忆系统实现
    """
    
    def __init__(self, workspace_path: str):
        # 永久记忆：身份定义
        self.identity = self.load_file(f"{workspace_path}/IDENTITY.md")
        self.soul = self.load_file(f"{workspace_path}/SOUL.md")
        
        # 长期记忆：重要信息存储
        self.long_term_file = f"{workspace_path}/MEMORY.md"
        self.vector_db = VectorDatabase(f"{workspace_path}/memory.db")
        
        # 短期记忆：会话缓存
        self.today_memory = f"{workspace_path}/memory/{date.today()}.md"
        self.session_cache = SQLiteCache()
    
    def remember(self, info: str, importance: str = "short"):
        """
        记忆写入规则：
        - "记住这个" → 更新 memory/YYYY-MM-DD.md
        - 学到教训 → 更新 AGENTS.md, TOOLS.md
        - 犯错 → 记录下来，让未来的你不重蹈覆辙
        """
        if importance == "permanent":
            # 不允许程序修改永久记忆
            pass
        elif importance == "long":
            # 重要信息存入长期记忆
            self.append_to_file(self.long_term_file, info)
            self.vector_db.store(info)
        else:
            # 临时信息存入短期记忆
            self.append_to_file(self.today_memory, info)
            self.session_cache.add(info)
    
    def recall(self, query: str) -> str:
        """
        记忆检索优先级：
        1. 短期记忆（SQLite 缓存）→ 当前会话上下文
        2. 长期记忆（MEMORY.md + 向量检索）→ 重要历史信息
        3. 永久记忆（IDENTITY.md）→ 核心身份定义
        """
        # 优先从短期记忆获取
        if self.session_cache.has(query):
            return self.session_cache.get(query)
        
        # 向量检索长期记忆
        results = self.vector_db.search(query, top_k=5)
        if results:
            return results
        
        # 返回永久记忆（身份定义）
        return self.identity

┌─────────────────────────────────────────────────────────────┐
│                    OpenClaw 技能系统                         │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  workspace/skills/                                          │
│  ├── playwright-mcp/          # 浏览器自动化技能            │
│  │   └── SKILL.md             # 技能定义文件               │
│  ├── doubao-chat/             # 豆包对话技能               │
│  │   └── SKILL.md                                         │
│  ├── screenshot/              # 截图技能                   │
│  │   └── SKILL.md                                         │
│  └── ...                                                   │
│                                                             │
│  技能定义格式 (SKILL.md):                                    │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ # 技能名称                                           │   │
│  │ ## 功能描述                                          │   │
│  │ ## 使用方法                                          │   │
│  │ ## 参数说明                                          │   │
│  │ ## 示例                                              │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

class SkillRegistry:
    """
    技能注册中心
    """
    
    def __init__(self):
        self.skills = {}
        self.load_skills()
    
    def load_skills(self):
        """从 workspace/skills/ 加载所有技能"""
        skills_dir = "workspace/skills"
        for skill_path in glob.glob(f"{skills_dir}/*/SKILL.md"):
            skill = self.parse_skill(skill_path)
            self.skills[skill.name] = skill
    
    def select_skill(self, user_message: str) -> Optional[Skill]:
        """根据用户消息选择合适的技能"""
        for skill in self.skills.values():
            if skill.matches(user_message):
                return skill
        return None
    
    def execute(self, skill_name: str, params: dict) -> str:
        """执行技能"""
        skill = self.skills.get(skill_name)
        if skill:
            return skill.execute(**params)
        return "技能不存在"

# 多渠道消息格式适配
class ChannelAdapter:
    def adapt_message(self, raw_message: dict, channel: str) -> Message:
        """适配不同渠道的消息格式"""
        
        if channel == "webchat":
            return Message(
                user_id=raw_message["userId"],
                content=raw_message["content"],
                channel="webchat"
            )
        
        elif channel == "rust_im":
            # Rust IM 服务消息
            return Message(
                user_id=raw_message["sender_id"],
                content=raw_message["text"],
                channel="rust_im",
                group_id=raw_message.get("group_id")
            )
        
        elif channel == "discord":
            return Message(
                user_id=str(raw_message["author"]["id"]),
                content=raw_message["content"],
                channel="discord",
                guild_id=raw_message.get("guild_id")
            )
        
        # ... 其他渠道

// openclaw.iflow.json
{
  "name": "OpenClaw",
  "version": "2.0.0",
  "gateway": {
    "port": 18789,
    "host": "0.0.0.0"
  },
  "llm": {
    "provider": "deepseek",
    "model": "deepseek-chat",
    "api_key": "${DEEPSEEK_API_KEY}"
  },
  "memory": {
    "vector_db": "./data/memory.db",
    "max_context_tokens": 8000
  },
  "skills": {
    "enabled": ["playwright-mcp", "doubao-chat", "screenshot"]
  }
}

# 启动主服务
cd F:\vsspace\showvoice\openclaw-iflow
python gateway_v2/run.py
 
# 或使用守护进程
python xiaolongxia_daemon.py

# .env 文件
DEEPSEEK_API_KEY=your_deepseek_key
QWEN_API_KEY=your_qwen_key
OPENAI_API_KEY=your_openai_key  # 可选

┌─────────────────────────────────────────────────────────────┐
│                   OpenClaw 消息处理流程                      │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   用户消息 ──→ ┌─────────────────────────────────────┐      │
│                │  1. Gateway 接收消息                  │      │
│                │     • 解析来源渠道                    │      │
│                │     • 权限验证                        │      │
│                └───────────────┬─────────────────────┘      │
│                                ↓                            │
│                ┌─────────────────────────────────────┐      │
│                │  2. 记忆系统加载上下文               │      │
│                │     • 加载短期记忆（当前会话）        │      │
│                │     • 检索长期记忆（相关历史）        │      │
│                │     • 注入永久记忆（身份定义）        │      │
│                └───────────────┬─────────────────────┘      │
│                                ↓                            │
│                ┌─────────────────────────────────────┐      │
│                │  3. 技能匹配与选择                   │      │
│                │     • 分析用户意图                    │      │
│                │     • 匹配合适技能                    │      │
│                └───────────────┬─────────────────────┘      │
│                                ↓                            │
│                ┌─────────────────────────────────────┐      │
│                │  4. LLM 推理生成响应                 │      │
│                │     • 构造提示词                      │      │
│                │     • 调用大模型 API                  │      │
│                └───────────────┬─────────────────────┘      │
│                                ↓                            │
│                ┌─────────────────────────────────────┐      │
│                │  5. 执行动作（如需要）               │      │
│                │     • 调用技能执行                    │      │
│                │     • 获取执行结果                    │      │
│                └───────────────┬─────────────────────┘      │
│                                ↓                            │
│                ┌─────────────────────────────────────┐      │
│                │  6. 更新记忆并返回响应               │      │
│                │     • 存储重要信息到记忆             │      │
│                │     • 返回响应给用户                  │      │
│                └─────────────────────────────────────┘      │
│                                                             │
└─────────────────────────────────────────────────────────────┘

# 心跳轮询处理
class HeartbeatHandler:
    """
    心跳机制：定时检查并主动行动
    """
    
    def on_heartbeat(self):
        """收到心跳轮询时的处理逻辑"""
        
        # 检查事项（轮换执行，每天 2-4 次）
        checks = [
            self.check_emails,      # 紧急邮件？
            self.check_calendar,    # 近期事件？
            self.check_mentions,    # 社交通知？
            self.check_weather,     # 天气提醒？
        ]
        
        # 随机选择一项检查
        check = random.choice(checks)
        result = check()
        
        if result.needs_attention:
            # 主动联系用户
            return result.message
        else:
            # 保持安静
            return "HEARTBEAT_OK"
    
    def should_be_proactive(self) -> bool:
        """判断是否应该主动联系"""
        # 深夜保持安静
        if self.is_night_time():
            return False
        
        # 主人很忙时不打扰
        if self.user_is_busy():
            return False
        
        # 有重要信息需要通知
        return self.has_important_info()

class GroupChatHandler:
    """
    群聊处理规则
    """
    
    def should_respond(self, message: GroupMessage) -> bool:
        """
        判断是否应该回复群聊消息
        
        响应时机：
        - 直接被提及或被问问题
        - 可以提供真正的价值
        - 纠正重要的错误信息
        - 被要求总结时
        """
        
        # 被提及
        if self.is_mentioned(message):
            return True
        
        # 能提供价值
        if self.can_contribute(message):
            return True
        
        # 需要纠正错误
        if self.needs_correction(message):
            return True
        
        return False
    
    def should_stay_quiet(self, message: GroupMessage) -> bool:
        """
        保持沉默的时机：
        - 只是人类之间的闲聊
        - 已经有人回答了问题
        - 回复只是"是的"或"不错"
        - 对话进行得很好
        """
        
        if self.is_small_talk(message):
            return True
        
        if self.already_answered(message):
            return True
        
        return False

class PromptBuilder:
    """
    OpenClaw 提示词构造器
    """
    
    def build_system_prompt(self, session: Session) -> str:
        """构造系统提示词"""
        
        # 1. 注入永久记忆（身份定义）
        identity = self.memory.get_permanent()
        
        # 2. 加载长期记忆（相关历史）
        history = self.memory.recall(session.user_id)
        
        # 3. 获取当前会话上下文
        context = self.memory.get_short_term(session.session_id)
        
        return f"""
{identity}
 
## 历史上下文
{history}
 
## 当前会话
{context}
 
## 可用技能
{self.list_skills()}
"""
    
    def build_user_prompt(self, message: str) -> str:
        """构造用户提示词"""
        return f"""
用户消息: {message}
 
请根据用户意图，选择合适的行动：
1. 直接回答（知识问题）
2. 调用技能（需要执行操作）
3. 请求更多信息（意图不明确）
"""

class AuthManager:
    """
    权限管理
    """
    
    def verify(self, user_id: str) -> bool:
        """验证用户权限"""
        
        # 检查用户是否在白名单
        if user_id in self.whitelist:
            return True
        
        # 检查用户是否被封禁
        if user_id in self.blacklist:
            return False
        
        # 默认允许（可配置）
        return self.default_allow

class SafeExecutor:
    """
    安全执行器
    """
    
    SENSITIVE_ACTIONS = [
        "send_email",
        "post_public",
        "delete_file",
        "execute_shell"
    ]
    
    def execute(self, action: str, params: dict) -> str:
        """安全执行操作"""
        
        # 敏感操作需要用户确认
        if action in self.SENSITIVE_ACTIONS:
            confirmation = self.ask_user_confirmation(action, params)
            if not confirmation:
                return "操作已取消"
        
        # 在沙箱环境中执行
        return self.sandbox.execute(action, params)

用户: 帮我写一篇关于 Java 并发的博客文章

OpenClaw:
  Thought: 需要收集并发知识点，然后组织成博客格式
  Action: skill_knowledge_search("Java 并发编程")
  Observation: 找到 synchronized、线程池、JMM 等知识点
  
  Action: generate_blog(topics=["synchronized", "线程池", "JMM"])
  Observation: 博客文章已生成
  
  Answer: 已为您生成 Java 并发编程博客文章，主要内容包括：
           1. synchronized 关键字原理
           2. 线程池核心参数与配置
           3. Java 内存模型(JMM)详解
           文章已保存到 workspace/output/java-concurrency.md

第一天对话：
用户: 我叫张三，是一名 Java 开发工程师
OpenClaw: 你好张三！很高兴认识一位 Java 工程师。

第二天对话：
用户: 我最近在学习什么？
OpenClaw: 根据记录，您是一名 Java 工程师，最近在学习 AI Agent 相关内容。
         需要我推荐一些 Agent 学习资料吗？

（跨会话记忆生效，Agent 记住了用户的身份和学习方向）

# 不同渠道消息统一适配为标准 Message 格式
class ChannelAdapter:
    def adapt_message(self, raw_message, channel):
        # WhatsApp / Telegram / Discord / WebChat
        # 统一转换为 Message(user_id, content, channel)
        ...

用户消息 → 技能匹配 → 技能执行 → 返回结果

1. 技能定义: workspace/skills/*/SKILL.md
2. 技能匹配: 根据消息内容匹配技能关键词
3. 技能执行: 在沙箱环境中安全执行
4. 结果返回: 将执行结果返回给用户

# 心跳轮询机制
def on_heartbeat():
    # 检查是否有需要主动通知的事项
    # 有 → 主动联系用户
    # 无 → 返回 HEARTBEAT_OK 保持安静
    
    if has_important_info():
        return proactive_message
    else:
        return "HEARTBEAT_OK"

1. 权限验证：检查用户是否在白名单
2. 敏感操作确认：发送邮件等操作需要用户确认
3. 沙箱执行：代码执行在隔离环境中
4. 操作日志：所有执行操作记录日志

1. 创建技能目录: workspace/skills/my-skill/
2. 编写 SKILL.md 定义文件
3. 实现技能逻辑（Python 脚本）
4. 重启服务自动加载