深度解析 OpenClaw 技术实现原理：从入门到实践

# ❌ 不好的设计：所有逻辑耦合在一起
def handle_message(user_input):
    if "订餐" in user_input:
        # 直接调用外卖 API
        order_food()
    elif "查询" in user_input:
        # 直接查数据库
        query_database()
    # ... 更多 if-else

# ✅ 好的设计：职责分离
class Gateway:
    """只负责接收和路由消息"""
    def route_message(self, message):
        # 根据类型路由到不同的处理器
        pass

class Assistant:
    """只负责理解意图和生成回复"""
    def process(self, message):
        # 调用 LLM 理解用户意图
        pass

class Skill:
    """只负责执行具体任务"""
    def execute(self, action):
        # 执行订餐、查询等具体操作
        pass

# weather-skill/SKILL.md
name: Weather Query
description: 查询指定城市的天气信息
triggers:
  - "天气"
  - "气温"
  - "下雨"
parameters:
  - name: city
    type: string
    description: 城市名称
    required: true
implementation: python weather.py

// Gateway 接收 WhatsApp 消息
interface Message {
  id: string;
  from: string;        // 用户 ID
  to: string;          // 机器人 ID
  text: string;        // 消息内容
  timestamp: number;
}

// Gateway 处理逻辑
async function handleMessage(msg: Message) {
  // 1. 验证用户
  const user = await authenticateUser(msg.from);
  
  // 2. 查找或创建会话
  const session = await getOrCreateSession(user.id);
  
  // 3. 加载历史上下文
  const context = await memory.loadContext(session.id);
  
  // 4. 路由到 Assistant
  const response = await assistant.process({
    message: msg.text,
    context: context,
    userId: user.id
  });
  
  // 5. 返回回复
  return response;
}

# Assistant 处理逻辑
class Assistant:
    def process(self, request):
        # 1. 构建 Prompt
        prompt = self.build_prompt(
            user_message=request.message,
            context=request.context,
            available_skills=self.get_skill_descriptions()
        )
        
        # 2. 调用 LLM
        llm_response = self.llm.generate(prompt)
        
        # 3. 解析意图
        intent = self.parse_intent(llm_response)
        
        # 4. 执行技能或直接回复
        if intent.needs_skill:
            result = self.execute_skill(intent.skill_name, intent.params)
            return self.generate_response(result)
        else:
            return llm_response

# Weather Skill 实现
# weather_skill/weather.py

import requests

def get_weather(city: str) -> dict:
    """查询天气信息"""
    # 调用天气 API
    api_url = f"https://api.weather.com/v1/{city}"
    response = requests.get(api_url)
    
    data = response.json()
    return {
        "city": city,
        "temperature": data["temp"],
        "condition": data["condition"],
        "humidity": data["humidity"]
    }

# 入口函数
def main(params):
    city = params.get("city", "北京")
    weather_data = get_weather(city)
    
    # 格式化输出
    return f"{city}今天{weather_data['condition']}，" \
           f"气温{weather_data['temperature']}℃，" \
           f"湿度{weather_data['humidity']}%"

# Memory 系统实现
class Memory:
    def __init__(self):
        self.vector_db = MilvusClient()
        self.embedding_model = SentenceTransformer()
    
    def save_conversation(self, session_id, message, response):
        """保存对话到记忆"""
        # 1. 生成向量
        embedding = self.embedding_model.encode(message)
        
        # 2. 存储到向量数据库
        self.vector_db.insert({
            "session_id": session_id,
            "message": message,
            "response": response,
            "embedding": embedding,
            "timestamp": time.time()
        })
    
    def retrieve_context(self, session_id, query, top_k=5):
        """检索相关上下文"""
        # 1. 生成查询向量
        query_embedding = self.embedding_model.encode(query)
        
        # 2. 向量搜索
        results = self.vector_db.search(
            collection_name="conversations",
            query_vector=query_embedding,
            filter={"session_id": session_id},
            limit=top_k
        )
        
        return results

mini-openclaw/
├── gateway/
│   ├── __init__.py
│   └── server.py          # WebSocket 服务器
├── assistant/
│   ├── __init__.py
│   └── agent.py           # LLM 调用逻辑
├── skills/
│   ├── __init__.py
│   ├── base.py            # 技能基类
│   └── weather/
│       ├── SKILL.md       # 技能描述
│       └── main.py        # 技能实现
├── memory/
│   ├── __init__.py
│   └── store.py           # 简单的内存存储
├── config.yaml            # 配置文件
└── main.py                # 入口文件

# config.yaml
gateway:
  host: "localhost"
  port: 8765
  
assistant:
  model: "gpt-3.5-turbo"
  api_key: "${OPENAI_API_KEY}"
  max_tokens: 1000
  
memory:
  type: "simple"  # 使用简单的内存存储
  max_history: 10
  
skills:
  directory: "./skills"
  auto_load: true

# gateway/server.py
import asyncio
import websockets
import json
from typing import Dict

class Gateway:
    def __init__(self, host: str, port: int):
        self.host = host
        self.port = port
        self.sessions: Dict[str, dict] = {}
    
    async def handle_connection(self, websocket, path):
        """处理 WebSocket 连接"""
        session_id = None
        
        try:
            async for message in websocket):
                data = json.loads(message)
                
                # 处理不同类型的消息
                if data["type"] == "connect":
                    # 创建新会话
                    session_id = self.create_session(data["user_id"])
                    await websocket.send(json.dumps({
                        "type": "connected",
                        "session_id": session_id
                    }))
                
                elif data["type"] == "message":
                    # 处理用户消息
                    response = await self.process_message(
                        session_id, 
                        data["content"]
                    )
                    await websocket.send(json.dumps({
                        "type": "response",
                        "content": response
                    }))
        
        except Exception as e:
            print(f"Error: {e}")
    
    def create_session(self, user_id: str) -> str:
        """创建新会话"""
        import uuid
        session_id = str(uuid.uuid4())
        self.sessions[session_id] = {
            "user_id": user_id,
            "created_at": time.time(),
            "messages": []
        }
        return session_id
    
    async def process_message(self, session_id: str, message: str) -> str:
        """处理消息（调用 Assistant）"""
        from assistant.agent import Assistant
        
        assistant = Assistant()
        response = await assistant.process(session_id, message)
        
        # 保存到会话历史
        self.sessions[session_id]["messages"].append({
            "role": "user",
            "content": message
        })
        self.sessions[session_id]["messages"].append({
            "role": "assistant",
            "content": response
        })
        
        return response
    
    def run(self):
        """启动服务器"""
        start_server = websockets.serve(
            self.handle_connection,
            self.host,
            self.port
        )
        
        asyncio.get_event_loop().run_until_complete(start_server)
        asyncio.get_event_loop().run_forever()

if __name__ == "__main__":
    gateway = Gateway("localhost", 8765)
    print("🦞 Mini OpenClaw Gateway running on ws://localhost:8765")
    gateway.run()

# assistant/agent.py
import openai
import yaml
from typing import List, Dict

class Assistant:
    def __init__(self):
        # 加载配置
        with open("config.yaml") as f:
            config = yaml.safe_load(f)
        
        self.model = config["assistant"]["model"]
        openai.api_key = config["assistant"]["api_key"]
        
        # 加载技能
        self.skills = self.load_skills()
    
    def load_skills(self) -> List[dict]:
        """加载所有技能"""
        skills = []
        # 这里简化处理，实际应该扫描 skills 目录
        skills.append({
            "name": "weather",
            "description": "查询天气信息",
            "trigger_words": ["天气", "气温", "下雨"]
        })
        return skills
    
    async def process(self, session_id: str, message: str) -> str:
        """处理用户消息"""
        # 1. 构建系统提示
        system_prompt = self.build_system_prompt()
        
        # 2. 获取历史上下文
        from memory.store import MemoryStore
        memory = MemoryStore()
        context = memory.get_context(session_id)
        
        # 3. 调用 LLM
        messages = [
            {"role": "system", "content": system_prompt},
            *context,
            {"role": "user", "content": message}
        ]
        
        response = openai.ChatCompletion.create(
            model=self.model,
            messages=messages,
            functions=self.get_function_definitions(),
            temperature=0.7
        )
        
        # 4. 检查是否需要调用技能
        if response.choices[0].message.get("function_call"):
            function_name = response.choices[0].message["function_call"]["name"]
            arguments = json.loads(
                response.choices[0].message["function_call"]["arguments"]
            )
            
            # 执行技能
            skill_result = await self.execute_skill(function_name, arguments)
            
            # 让 LLM 基于结果生成自然语言回复
            final_response = openai.ChatCompletion.create(
                model=self.model,
                messages=[
                    *messages,
                    response.choices[0].message,
                    {"role": "function", "name": function_name, "content": skill_result}
                ]
            )
            
            return final_response.choices[0].message.content
        
        return response.choices[0].message.content
    
    def build_system_prompt(self) -> str:
        """构建系统提示"""
        skill_descriptions = "\n".join([
            f"- {skill['name']}: {skill['description']}"
            for skill in self.skills
        ])
        
        return f"""你是一个有用的 AI 助手。

可用技能：
{skill_descriptions}

当需要使用技能时，请调用相应的函数。"""
    
    def get_function_definitions(self) -> List[dict]:
        """获取函数定义（用于 Function Calling）"""
        return [
            {
                "name": "get_weather",
                "description": "查询指定城市的天气信息",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {
                            "type": "string",
                            "description": "城市名称，如：北京、上海"
                        }
                    },
                    "required": ["city"]
                }
            }
        ]
    
    async def execute_skill(self, skill_name: str, params: dict) -> str:
        """执行技能"""
        if skill_name == "get_weather":
            from skills.weather.main import get_weather
            return get_weather(params["city"])
        
        return "未知技能"

# skills/weather/main.py
import requests
from typing import Dict

def get_weather(city: str) -> str:
    """
    查询天气信息
    
    Args:
        city: 城市名称
    
    Returns:
        天气信息描述
    """
    # 这里使用免费的天气 API（示例）
    # 实际使用时请替换为真实的 API
    
    # 模拟 API 调用
    weather_data = {
        "北京": {"temp": "15-25", "condition": "晴", "humidity": "45%"},
        "上海": {"temp": "18-26", "condition": "多云", "humidity": "60%"},
        "深圳": {"temp": "22-28", "condition": "晴", "humidity": "70%"}
    }
    
    if city in weather_data:
        data = weather_data[city]
        return f"{city}今天{data['condition']}，气温{data['temp']}℃，湿度{data['humidity']}"
    else:
        return f"抱歉，暂时没有{city}的天气信息"

if __name__ == "__main__":
    # 测试
    print(get_weather("北京"))

# memory/store.py
from typing import List, Dict
import time

class MemoryStore:
    def __init__(self):
        # 使用简单的内存存储（生产环境应使用数据库）
        self.conversations: Dict[str, List[dict]] = {}
        self.max_history = 10
    
    def save_message(self, session_id: str, role: str, content: str):
        """保存消息到历史"""
        if session_id not in self.conversations:
            self.conversations[session_id] = []
        
        self.conversations[session_id].append({
            "role": role,
            "content": content,
            "timestamp": time.time()
        })
        
        # 限制历史长度
        if len(self.conversations[session_id]) > self.max_history:
            self.conversations[session_id] = \
                self.conversations[session_id][-self.max_history:]
    
    def get_context(self, session_id: str) -> List[dict]:
        """获取上下文"""
        return self.conversations.get(session_id, [])
    
    def clear_session(self, session_id: str):
        """清除会话历史"""
        if session_id in self.conversations:
            del self.conversations[session_id]

# main.py
import asyncio
from gateway.server import Gateway

def main():
    """启动 Mini OpenClaw"""
    print("🦞 正在启动 Mini OpenClaw...")
    print("📖 这是最小化的 OpenClaw 实现")
    print("💡 用于学习 AI Agent 架构")
    print("-" * 50)
    
    gateway = Gateway("localhost", 8765)
    gateway.run()

if __name__ == "__main__":
    main()

pip install websockets openai pyyaml requests

export OPENAI_API_KEY="your-api-key-here"

python main.py

# test_client.py
import asyncio
import websockets
import json

async def test():
    uri = "ws://localhost:8765"
    async with websockets.connect(uri) as websocket:
        # 连接
        await websocket.send(json.dumps({
            "type": "connect",
            "user_id": "test_user"
        }))
        
        response = await websocket.recv()
        print("连接响应:", response)
        
        # 发送消息
        await websocket.send(json.dumps({
            "type": "message",
            "content": "今天北京天气怎么样？"
        }))
        
        response = await websocket.recv()
        print("AI 回复:", json.loads(response)["content"])

asyncio.run(test())

mkdir skills/news

# skills/news/SKILL.md
name: News Query
description: 查询最新新闻
triggers:
  - "新闻"
  - "今日头条"
parameters:
  - name: category
    type: string
    description: 新闻类别（科技/体育/娱乐）
    required: false

# skills/news/main.py
def get_news(category: str = "general") -> str:
    """获取新闻"""
    # 调用新闻 API
    # 返回格式化的新闻内容
    pass

# channels/telegram.py
from telegram import Update
from telegram.ext import Updater, MessageHandler, Filters

class TelegramChannel:
    def __init__(self, token: str, gateway: Gateway):
        self.updater = Updater(token=token)
        self.gateway = gateway
    
    def handle_message(self, update: Update, context):
        """处理 Telegram 消息"""
        user_id = str(update.effective_user.id)
        message = update.message.text
        
        # 转发到 Gateway
        response = await self.gateway.process_message(user_id, message)
        
        # 返回回复
        update.message.reply_text(response)
    
    def start(self):
        """启动 Telegram Bot"""
        dispatcher = self.updater.dispatcher
        dispatcher.add_handler(MessageHandler(Filters.text, self.handle_message))
        self.updater.start_polling()

# memory/vector_store.py
from pymilvus import connections, Collection
from sentence_transformers import SentenceTransformer

class VectorMemory:
    def __init__(self):
        # 连接 Milvus
        connections.connect("default", host="localhost", port="19530")
        
        # 加载 Embedding 模型
        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')
        
        # 获取集合
        self.collection = Collection("conversations")
    
    def save(self, session_id: str, message: str, response: str):
        """保存对话到向量数据库"""
        # 生成向量
        embedding = self.encoder.encode(message).tolist()
        
        # 插入数据
        self.collection.insert([
            [session_id],
            [message],
            [response],
            [embedding]
        ])
    
    def search(self, query: str, top_k: int = 5) -> list:
        """语义搜索相关对话"""
        # 生成查询向量
        query_vector = self.encoder.encode(query).tolist()
        
        # 向量搜索
        results = self.collection.search(
            data=[query_vector],
            anns_field="embedding",
            param={"metric_type": "L2", "params": {"nprobe": 10}},
            limit=top_k
        )
        
        return results

❌ 问题：
- Token 数量：1000 个技能 × 平均 200 tokens = 200,000 tokens
- 成本：每次请求都要花费大量 tokens
- 速度：LLM 处理时间变长（5-10秒）
- 准确性：LLM 难以在大量选项中做出正确选择

class SmartSkillLoader:
    """智能技能加载器"""
    
    def __init__(self, all_skills: list):
        self.all_skills = all_skills
        self.skill_embeddings = self.precompute_embeddings()
    
    def get_relevant_skills(self, user_message: str, top_k: int = 5):
        """获取相关技能（两阶段）"""
        
        # 阶段 1: 快速筛选（本地，不调用 LLM）
        candidate_skills = self.quick_filter(user_message, top_k=10)
        
        # 阶段 2: LLM 精确选择
        final_skills = self.llm_refine(user_message, candidate_skills, top_k=5)
        
        return final_skills
    
    def quick_filter(self, message: str, top_k: int):
        """快速筛选：关键词 + 向量搜索"""
        
        # 1. 关键词匹配
        keyword_matches = []
        for skill in self.all_skills:
            if any(keyword in message for keyword in skill.trigger_words):
                keyword_matches.append(skill)
        
        # 2. 向量相似度搜索
        query_embedding = self.encode(message)
        vector_matches = self.vector_search(
            query_embedding, 
            self.skill_embeddings, 
            top_k=top_k
        )
        
        # 3. 合并去重
        all_candidates = list(set(keyword_matches + vector_matches))
        
        return all_candidates[:top_k]
    
    def llm_refine(self, message: str, candidates: list, top_k: int):
        """LLM 精确选择（只处理少量候选）"""
        
        # 构建提示（只有 10 个候选，而不是 1000 个）
        prompt = f"""用户消息：{message}

候选技能：
{self.format_skills(candidates)}

请选择最合适的 1-{top_k} 个技能。"""
        
        response = self.llm.generate(prompt)
        selected_skills = self.parse_llm_response(response)
        
        return selected_skills

class SkillRouter:
    """技能分类路由器"""
    
    def __init__(self):
        # 定义技能分类
        self.categories = {
            "weather": ["天气", "气温", "下雨", "晴"],
            "file": ["文件", "读取", "保存", "删除"],
            "network": ["网络", "请求", "下载", "上传"],
            "calendar": ["日历", "日程", "会议", "提醒"],
            # ... 更多分类
        }
        
        # 每个分类下的技能
        self.skill_categories = {
            "weather": [weather_skill1, weather_skill2, ...],
            "file": [file_skill1, file_skill2, ...],
            # ...
        }
    
    def route(self, user_message: str) -> list:
        """先判断类别，再加载对应技能"""
        
        # 1. 判断类别（轻量级分类器）
        category = self.classify_category(user_message)
        
        # 2. 只加载该类别的技能（而不是全部 1000 个）
        relevant_skills = self.skill_categories.get(category, [])
        
        return relevant_skills
    
    def classify_category(self, message: str) -> str:
        """分类用户消息"""
        # 简单的关键词匹配
        for category, keywords in self.categories.items():
            if any(keyword in message for keyword in keywords):
                return category
        
        return "general"  # 默认类别

class DynamicSkillLoader:
    """动态技能加载器"""
    
    def __init__(self):
        # 只在内存中保存技能摘要
        self.skill_summaries = self.load_summaries()
        self.full_skills = {}  # 缓存已加载的完整技能
    
    def load_summaries(self):
        """加载所有技能的摘要（轻量级）"""
        return {
            "weather": {
                "summary": "查询天气",
                "keywords": ["天气", "气温"],
                "embedding": [0.1, 0.2, ...],  # 预计算的向量
            },
            "news": {
                "summary": "查询新闻",
                "keywords": ["新闻", "头条"],
                "embedding": [0.3, 0.4, ...],
            },
            # ... 1000 个技能的摘要
        }
    
    def get_skills(self, user_message: str, top_k: int = 5):
        """动态加载相关技能"""
        
        # 1. 用摘要做快速筛选
        query_embedding = self.encode(user_message)
        relevant_names = self.vector_search(
            query_embedding, 
            self.skill_summaries,
            top_k=top_k
        )
        
        # 2. 动态加载完整技能信息
        full_skills = []
        for skill_name in relevant_names:
            if skill_name not in self.full_skills:
                # 从磁盘加载完整信息（延迟加载）
                full_desc = self.load_from_disk(skill_name)
                self.full_skills[skill_name] = full_desc
            
            full_skills.append(self.full_skills[skill_name])
        
        return full_skills

# skills/weather/SKILL.md

# 摘要部分（用于快速筛选）
summary: "查询天气信息"
keywords: ["天气", "气温", "下雨"]
embedding: [0.123, 0.456, ...]  # 预计算的向量

# 详细信息部分（只在需要时加载）
detailed_description: |
  这个技能可以查询全球任何城市的天气信息。
  支持当前天气、未来7天预报、历史天气查询。
  
  参数：
  - city: 城市名称（必需）
  - days: 预报天数（可选，1-7）
  
  示例用法：
  - "今天北京天气怎么样？"
  - "上海未来3天的天气"

class TwoLevelSkillManager:
    """两级技能管理器"""
    
    def get_skill_prompts(self, user_message: str):
        """只加载相关技能的详细信息"""
        
        # 1. 用摘要快速筛选（只读摘要部分）
        candidate_skills = self.search_by_summary(user_message)
        
        # 2. 只加载候选技能的详细信息
        detailed_prompts = []
        for skill in candidate_skills:
            # 此时才读取完整的 detailed_description
            detailed_prompts.append(skill.load_detailed_description())
        
        return detailed_prompts

class OpenClawSkillLoader:
    """OpenClaw 的技能加载策略"""
    
    def load_skills_for_context(self, user_message: str):
        """为当前上下文加载技能"""
        
        # 1. 关键词快速匹配（毫秒级）
        keyword_matches = self.keyword_match(user_message)
        
        # 2. 向量语义搜索（毫秒级）
        semantic_matches = self.semantic_search(user_message, top_k=10)
        
        # 3. 合并结果
        candidate_skills = list(set(keyword_matches + semantic_matches))
        
        # 4. 如果候选太多，用 LLM 做最终筛选
        if len(candidate_skills) > 10:
            candidate_skills = self.llm_filter(
                user_message, 
                candidate_skills,
                max_skills=5
            )
        
        # 5. 只加载这些技能的完整信息
        return self.load_full_descriptions(candidate_skills)

# 简单方案：直接加载所有技能
# 100 个技能 × 100 tokens = 10,000 tokens
# 在可接受范围内
all_skills = load_all_skills()
response = llm.generate(user_message, all_skills)

# 推荐方案：两阶段检索
candidate_skills = quick_filter(user_message, all_skills, top_k=10)
response = llm.generate(user_message, candidate_skills)

# 推荐方案：分类 + 两阶段 + 缓存
category = classify(user_message)
category_skills = get_category_skills(category)
relevant_skills = semantic_search(user_message, category_skills, top_k=5)
response = llm.generate(user_message, relevant_skills)

def validate_input(message: str) -> bool:
    """验证用户输入"""
    # 长度限制
    if len(message) > 1000:
        return False
    
    # 敏感词过滤
    sensitive_words = ["exec", "eval", "system"]
    if any(word in message.lower() for word in sensitive_words):
        return False
    
    return True

def check_permission(user_id: str, skill_name: str) -> bool:
    """检查用户权限"""
    user_permissions = get_user_permissions(user_id)
    return skill_name in user_permissions

import subprocess

def execute_in_sandbox(code: str) -> str:
    """在沙箱中执行代码"""
    # 使用 Docker 容器隔离执行
    result = subprocess.run(
        ["docker", "run", "--rm", "python:3.9", "python", "-c", code],
        capture_output=True,
        text=True,
        timeout=10  # 超时限制
    )
    return result.stdout

import asyncio

async def batch_process(messages: list) -> list:
    """批量处理消息"""
    tasks = [process_message(msg) for msg in messages]
    return await asyncio.gather(*tasks)

from functools import lru_cache

@lru_cache(maxsize=100)
def get_skill_description(skill_name: str) -> str:
    """缓存技能描述"""
    # 从文件读取
    with open(f"skills/{skill_name}/SKILL.md") as f:
        return f.read()

from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

# 数据库连接池
engine = create_engine(
    "postgresql://localhost/openclaw",
    poolclass=QueuePool,
    pool_size=10,
    max_overflow=20
)

# Dockerfile
FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8765

CMD ["python", "main.py"]

# docker-compose.yml
version: '3.8'

services:
  openclaw:
    build: .
    ports:
      - "8765:8765"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    volumes:
      - ./data:/app/data
    restart: unless-stopped
  
  milvus:
    image: milvusdb/milvus:latest
    ports:
      - "19530:19530"
    volumes:
      - milvus_data:/var/lib/milvus

volumes:
  milvus_data:

# logging_config.py
import logging
from logging.handlers import RotatingFileHandler

def setup_logging():
    """配置日志"""
    logger = logging.getLogger("OpenClaw")
    logger.setLevel(logging.INFO)
    
    # 文件处理器
    file_handler = RotatingFileHandler(
        "logs/openclaw.log",
        maxBytes=10*1024*1024,  # 10MB
        backupCount=5
    )
    file_handler.setFormatter(logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    ))
    
    logger.addHandler(file_handler)
    return logger