OpenClaw 智能体实战：从零搭建你的第一个 AI 员工

3.1.2 完整代码

import requests
import json

# 通义千问 API 配置（新手需先在阿里云申请 API-KEY）
API_KEY = "你的 API-KEY"
API_URL = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation"

class TaskParser:
    def __init__(self, api_key):
        self.api_key = api_key
        self.headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        # 提示词模板（核心：规范输出格式）
        self.prompt_template = """
你是 OpenClaw 智能体的任务解析专家，必须严格按照以下 JSON 格式输出结果，不允许添加任何额外文字：
{
  "任务类型": "字符串",
  "目标": "字符串",
  "步骤": ["步骤 1", "步骤 2"],
  "所需技能": ["技能名称"],
  "参数": {"参数名": "值"}
}
用户指令：{user_instruction}
"""

    def parse(self, user_instruction):
        prompt = self.prompt_template.format(user_instruction=user_instruction)
        payload = {
            "model": "qwen-turbo",
            "input": {"messages": [{"role": "user", "content": prompt}]},
            "parameters": {"result_format": "json", "temperature": 0.1}
        }
        try:
            response = requests.post(API_URL, headers=self.headers, json=payload)
            response.raise_for_status()
            result = response.json()["output"]["choices"][0]["message"]["content"]
            return json.loads(result)
        except Exception as e:
            return {"error": f"解析失败：{str(e)}"}

if __name__ == "__main__":
    parser = TaskParser(api_key=API_KEY)
    task = parser.parse("整理桌面的杂乱文件")
    print(json.dumps(task, ensure_ascii=False, indent=2))

3.2 核心组件 2：技能调用器

3.2.1 技能的本质

技能是可复用的 Python 函数，需遵循统一接口。

3.2.2 完整代码

import os
import shutil

class SkillPool:
    def __init__(self):
        self.skills = {
            "文件遍历": self.traverse_files,
            "文件夹创建": self.create_folders,
            "文件移动": self.move_files,
            "统计输出": self.print_statistics
        }

    def get_skill(self, skill_name):
        return self.skills.get(skill_name, None)

    def traverse_files(self, target_path):
        try:
            target_path = os.path.expanduser(target_path)
            file_list = []
            for filename in os.listdir(target_path):
                file_path = os.path.join(target_path, filename)
                if os.path.isfile(file_path):
                    file_list.append(file_path)
            return {"status": "success", "data": file_list}
        except Exception as e:
            return {"status": "failed", "error": str(e)}

    def create_folders(self, target_path, folder_names):
        try:
            target_path = os.path.expanduser(target_path)
            created_folders = []
            for folder in folder_names:
                folder_path = os.path.join(target_path, folder)
                if not os.path.exists(folder_path):
                    os.makedirs(folder_path)
                created_folders.append(folder_path)
            return {"status": "success", "data": created_folders}
        except Exception as e:
            return {"status": "failed", "error": str(e)}

    def move_files(self, file_list, target_path, category_rules):
        try:
            target_path = os.path.expanduser(target_path)
            success_count = 0
            failed_files = []
            for file_path in file_list:
                file_ext = os.path.splitext(file_path)[1].lower()
                target_folder = None
                for category, exts in category_rules.items():
                    if file_ext in exts:
                        target_folder = os.path.join(target_path, category)
                        break
                if not target_folder:
                    continue
                try:
                    shutil.move(file_path, target_folder)
                    success_count += 1
                except Exception as e:
                    failed_files.append({"file": file_path, "error": str(e)})
            return {"status": "success", "data": {"成功移动数量": success_count, "失败文件": failed_files}}
        except Exception as e:
            return {"status": "failed", "error": str(e)}

    def print_statistics(self, result_data):
        print("\n" + "=" * 50)
        print("文件整理完成！统计信息如下：")
        print(f"成功移动文件数量：{result_data['成功移动数量']}")
        print(f"失败文件数量：{len(result_data['失败文件'])}")
        print("=" * 50 + "\n")
        return {"status": "success", "data": "统计信息输出完成"}

class SkillInvoker:
    def __init__(self, skill_pool):
        self.skill_pool = skill_pool

    def invoke(self, skill_name, **kwargs):
        skill = self.skill_pool.get_skill(skill_name)
        if not skill:
            return {"status": "failed", "error": f"未找到技能：{skill_name}"}
        return skill(**kwargs)

if __name__ == "__main__":
    skill_pool = SkillPool()
    invoker = SkillInvoker(skill_pool)
    result = invoker.invoke("文件遍历", target_path="~/Desktop")
    print(result)

3.3 核心组件 3：记忆模块

使用 JSON 文件存储用户偏好和任务历史，无需复杂数据库。

3.3.1 完整代码

import json
import time

class MemoryModule:
    def __init__(self, memory_file="openclaw_memory.json"):
        self.memory_file = memory_file
        self.memory = {
            "user_preferences": {
                "default_target_path": "~/Desktop",
                "default_category_rules": {
                    "文档": [".doc", ".pdf", ".txt"],
                    "图片": [".jpg", ".png"],
                    "视频": [".mp4", ".avi"]
                }
            },
            "task_history": []
        }
        self.load_memory()

    def load_memory(self):
        try:
            with open(self.memory_file, "r", encoding="utf-8") as f:
                self.memory = json.load(f)
        except FileNotFoundError:
            self.save_memory()
        except Exception as e:
            print(f"加载记忆失败：{str(e)}")

    def save_memory(self):
        try:
            with open(self.memory_file, "w", encoding="utf-8") as f:
                json.dump(self.memory, f, ensure_ascii=False, indent=2)
        except Exception as e:
            print(f"保存记忆失败：{str(e)}")

    def update_preference(self, key, value):
        self.memory["user_preferences"][key] = value
        self.save_memory()
        return {"status": "success"}

    def add_task_history(self, task, result):
        history_item = {
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "task": task,
            "result": result
        }
        self.memory["task_history"].append(history_item)
        if len(self.memory["task_history"]) > 100:
            self.memory["task_history"].pop(0)
        self.save_memory()

if __name__ == "__main__":
    memory = MemoryModule()
    memory.update_preference("default_target_path", "~/Documents")

四、从零实操：搭建你的第一个 AI 员工

4.1 前置准备：环境与安全隔离

根据安全建议，禁止在主力工作机直接部署。建议使用虚拟机或云服务器。

1. 安装 Python 3.9+。
2. 安装依赖：pip install requests

4.2 Step1：整合核心组件

将上述三个组件整合为主程序 main.py。

from task_parser import TaskParser
from skill_invoker import SkillPool, SkillInvoker
from memory_module import MemoryModule

class OpenClawAIEmployee:
    def __init__(self, llm_api_key):
        self.task_parser = TaskParser(api_key=llm_api_key)
        self.skill_pool = SkillPool()
        self.skill_invoker = SkillInvoker(self.skill_pool)
        self.memory = MemoryModule()
        print("✅ OpenClaw AI 员工初始化完成！")

    def run(self, user_instruction):
        print(f"\n📥 收到用户指令：{user_instruction}")
        structured_task = self.task_parser.parse(user_instruction)
        if "error" in structured_task:
            print(f"❌ 任务解析失败：{structured_task['error']}")
            return

        default_path = self.memory.get_preference("default_target_path")
        if not structured_task["参数"].get("目标路径"):
            structured_task["参数"]["目标路径"] = default_path

        print("🚀 开始执行任务...")
        try:
            for step in structured_task["步骤"]:
                if "遍历" in step:
                    result = self.skill_invoker.invoke("文件遍历", target_path=structured_task["参数"]["目标路径"])
                elif "创建" in step:
                    result = self.skill_invoker.invoke("文件夹创建", target_path=structured_task["参数"]["目标路径"], folder_names=["文档", "图片"])
                elif "移动" in step:
                    result = self.skill_invoker.invoke("文件移动", file_list=[], target_path=structured_task["参数"]["目标路径"], category_rules={})
                else:
                    continue
        except Exception as e:
            print(f"❌ 任务执行失败：{str(e)}")
        finally:
            self.memory.add_task_history(structured_task, {"status": "success"})
            print("🎉 任务执行完成！")

if __name__ == "__main__":
    LLM_API_KEY = "你的通义千问 API-KEY"
    ai_employee = OpenClawAIEmployee(llm_api_key=LLM_API_KEY)
    ai_employee.run("整理我的默认文件夹里的文件")

4.3 Step2：文件组织结构

确保所有 .py 文件在同一文件夹下：

OpenClaw_AI_Employee/
├── main.py
├── task_parser.py
├── skill_invoker.py
└── memory_module.py

4.4 Step3：运行验证

执行 python main.py，观察控制台输出及文件变动。

五、企业级优化与进阶技巧

5.1 成本优化：Token 消耗控制

常见指令可缓存结构化任务，避免重复调用 API。

5.2 安全加固：权限管控

限制智能体仅能访问指定路径，禁止访问系统目录。对于删除等敏感操作，增加用户确认步骤。

六、常见问题与避坑指南

1. 导入错误：检查文件名与导入语句是否一致。
2. API 调用失败：确认 API-KEY 有效且网络通畅。
3. 权限错误：检查目标文件是否被占用，或尝试以管理员身份运行。
4. 路径错误：修改 openclaw_memory.json 中的路径为实际存在的路径。

七、总结

本文拆解了 OpenClaw 智能体的核心原理，实现了任务解析、技能调用、记忆模块三大组件，并提供了完整的文件整理 AI 员工搭建流程。开发者可在此基础上扩展技能池（如网页爬取、消息发送），落地更多自动化场景。核心始终是'让 AI 替人完成重复性工作'。