其实你自己就能做出 OpenClaw（教程在这）

说实话，我第一次听说 OpenClaw 的时候，以为这是什么只有博士级工程师才能搞懂的黑科技。结果呢？一旦拆开来看，其实挺简单的。

整个东西本质上就是一个网关——把 AI 连接到你的消息应用，给它一些操作电脑的工具，然后让它记住你是谁。就这样。难的地方在于，当你同时跑多个渠道时，怎么让它稳定运行。

我的做法是：从零开始，自己重新实现了核心组件。没有花哨的框架，就是一个消息 API、一个大模型，加上一些耐心。说真的，你也完全可以做到。

没人说的那个问题

浏览器里的 ChatGPT 和 Claude，说白了不太适合做真正的工作。别误会，它们确实很聪明。但是：

每次对话都从头开始。它不知道你叫什么、昨天做了什么，什么都不知道。你永远在重新介绍自己。

你得主动去找它。它不会主动来找你。想让它早上七点帮你检查日历？不行，它只在你坐在那儿打字的时候才工作。

它被困在一个文本框里。不能运行命令，不能帮你浏览网页，除了聊天什么实际的事都做不了。

它只活在一个标签页里。你的生活分散在 WhatsApp、Telegram、Discord、Slack 各处，但你的 AI？被困在自己的小窗口里。

如果有一个 AI 住在你真正在用的消息应用里，记得所有事情，能控制你的电脑，在你自己的硬件上 7×24 小时运行，那会怎样？

那就是 OpenClaw。我们来做一个。

从最简单的开始

首先，做一个能响应 Telegram 消息的 AI：

  1. # bot-v0.py - 最简单的 AI 机器人
  2. import os
  3. import anthropic
  4. from telegram import Update
  5. from telegram.ext import Application, MessageHandler, filters
  7. client = anthropic.Anthropic()
  9. async def handle_message(update: Update, context):
  10.     user_message = update.message.text
  11.     response = client.messages.create(
  12.         model="claude-sonnet-4-5-20250929",
  13.         max_tokens=1024,
  14.         messages=[{"role": "user", "content": user_message}]
  15.     )
  16.     await update.message.reply_text(response.content[0].text)
  18. app = Application.builder().token(os.getenv("TELEGRAM_BOT_TOKEN")).build()
  19. app.add_handler(MessageHandler(filters.TEXT, handle_message))
  20. app.run_polling()

运行它，发一条消息，得到回复。挺酷的。但也完全没用，因为它没有任何记忆。问它”我刚才说了什么？”它完全不知道。

让它记住你

解决方法很简单——保存对话历史。我用的是 JSONL 文件，因为它崩溃安全。每行是一条消息，程序中途挂掉最多丢一行。

  1. # bot-v1.py - 带持久会话的机器人
  2. import json
  3. import os
  4. import anthropic
  5. from telegram import Update
  6. from telegram.ext import Application, MessageHandler, filters
  8. client = anthropic.Anthropic()
  9. SESSIONS_DIR = "./sessions"
  10. os.makedirs(SESSIONS_DIR, exist_ok=True)
  12. def get_session_path(user_id):
  13.     return os.path.join(SESSIONS_DIR, f"{user_id}.jsonl")
  15. def load_session(user_id):
  16.     """从磁盘加载对话历史"""
  17.     path = get_session_path(user_id)
  18.     messages = []
  19.     if os.path.exists(path):
  20.         with open(path, "r") as f:
  21.             for line in f:
  22.                 if line.strip():
  23.                     messages.append(json.loads(line))
  24.     return messages
  26. def append_to_session(user_id, message):
  27.     """向会话文件追加一条消息"""
  28.     path = get_session_path(user_id)
  29.     with open(path, "a") as f:
  30.         f.write(json.dumps(message) + "\n")
  32. def save_session(user_id, messages):
  33.     """用完整消息列表覆写会话文件"""
  34.     path = get_session_path(user_id)
  35.     with open(path, "w") as f:
  36.         for message in messages:
  37.             f.write(json.dumps(message) + "\n")
  39. async def handle_message(update: Update, context):
  40.     user_id = str(update.effective_user.id)
  41.     user_message = update.message.text
  42.     messages = load_session(user_id)
  43.     user_msg = {"role": "user", "content": user_message}
  44.     messages.append(user_msg)
  45.     append_to_session(user_id, user_msg)
  46.     response = client.messages.create(
  47.         model="claude-sonnet-4-5-20250929",
  48.         max_tokens=4096,
  49.         messages=messages
  50.     )
  51.     assistant_msg = {"role": "assistant", "content": response.content[0].text}
  52.     append_to_session(user_id, assistant_msg)
  53.     await update.message.reply_text(response.content[0].text)
  55. app = Application.builder().token(os.getenv("TELEGRAM_BOT_TOKEN")).build()
  56. app.add_handler(MessageHandler(filters.TEXT, handle_message))
  57. app.run_polling()

现在你可以真正进行对话了：

你： 我叫 Rolly 机器人： 很高兴认识你，Rolly！

（几小时后……）

你： 我叫什么名字？ 机器人： 你叫 Rolly！

这正是 OpenClaw 存储对话的方式：~/.openclaw/agents/<agentId>/sessions/<sessionId>.jsonl。一个会话，一个文件。重启所有东西，数据还在。

赋予它个性

现在我们的机器人太泛泛了，无聊。用一个 SOUL 文件给它一些性格（对，OpenClaw 真的就叫这个名字）：

  1. # bot-v2.py - 带个性的机器人
  2. SOUL = """
  3. # 你是谁
  4. **名字：** Jarvis
  5. **角色：** 个人 AI 助手
  7. ## 个性
  8. - 真正有帮助，而不是表演式的帮助
  9. - 省掉"好问题！"——直接帮忙就好
  10. - 有自己的观点，你可以不同意
  11. - 需要简洁时简洁，需要详细时详细
  13. ## 边界
  14. - 私密的事情保持私密
  15. - 有疑问时，在对外行动前先询问
  16. - 你不是用户的代言人——代替他们发送消息要谨慎
  18. ## 记忆
  19. 记住对话中的重要细节。
  20. 如果某件事值得记，就写下来。
  21. """
  23. async def handle_message(update: Update, context):
  24.     user_id = str(update.effective_user.id)
  25.     messages = load_session(user_id)
  26.     user_msg = {"role": "user", "content": update.message.text}
  27.     messages.append(user_msg)
  28.     append_to_session(user_id, user_msg)
  29.     response = client.messages.create(
  30.         model="claude-sonnet-4-5-20250929",
  31.         max_tokens=4096,
  32.         system=SOUL,  # <-- 个性在这里注入
  33.         messages=messages
  34.     )
  35.     assistant_msg = {"role": "assistant", "content": response.content[0].text}
  36.     append_to_session(user_id, assistant_msg)
  37.     await update.message.reply_text(response.content[0].text)

SOUL 在每次调用时作为系统提示注入。现在你不是在和”通用 AI 助手 #47”对话，而是在和 Jarvis 对话。

在 OpenClaw 中，这个文件放在 ~/.openclaw/workspace/SOUL.md。想写什么都行，可以幽默，可以严肃，可以给它一个背景故事。越具体，它的行为就越一致。

真正做点事（工具）

一个只会聊天的机器人相当有限。如果它能运行命令、读取文件、搜索网页呢？

  1. import subprocess
  3. TOOLS = [
  4.     {
  5.         "name": "run_command",
  6.         "description": "在用户的电脑上运行一个 shell 命令",
  7.         "input_schema": {
  8.             "type": "object",
  9.             "properties": {
  10.                 "command": {"type": "string", "description": "要运行的命令"}
  11.             },
  12.             "required": ["command"]
  13.         }
  14.     },
  15.     {
  16.         "name": "read_file",
  17.         "description": "从文件系统读取一个文件",
  18.         "input_schema": {
  19.             "type": "object",
  20.             "properties": {
  21.                 "path": {"type": "string", "description": "文件路径"}
  22.             },
  23.             "required": ["path"]
  24.         }
  25.     },
  26.     {
  27.         "name": "write_file",
  28.         "description": "向文件写入内容",
  29.         "input_schema": {
  30.             "type": "object",
  31.             "properties": {
  32.                 "path": {"type": "string", "description": "文件路径"},
  33.                 "content": {"type": "string", "description": "要写入的内容"}
  34.             },
  35.             "required": ["path", "content"]
  36.         }
  37.     },
  38.     {
  39.         "name": "web_search",
  40.         "description": "在网络上搜索信息",
  41.         "input_schema": {
  42.             "type": "object",
  43.             "properties": {
  44.                 "query": {"type": "string", "description": "搜索关键词"}
  45.             },
  46.             "required": ["query"]
  47.         }
  48.     }
  49. ]
  51. def execute_tool(name, input):
  52.     if name == "run_command":
  53.         result = subprocess.run(
  54.             input["command"], shell=True,
  55.             capture_output=True, text=True, timeout=30
  56.         )
  57.         return result.stdout + result.stderr
  58.     elif name == "read_file":
  59.         with open(input["path"], "r") as f:
  60.             return f.read()
  61.     elif name == "write_file":
  62.         with open(input["path"], "w") as f:
  63.             f.write(input["content"])
  64.         return f"已写入 {input['path']}"
  65.     elif name == "web_search":
  66.         # 简化版——实际使用时请接入真正的搜索 API
  67.         return f"搜索结果：{input['query']}"
  68.     return f"未知工具：{name}"

现在我们需要 Agent 循环。当 AI 想使用工具时，我们运行它并把结果返回：

  1. def serialize_content(content):
  2.     """将 API 响应内容块转换为可 JSON 序列化的字典"""
  3.     serialized = []
  4.     for block in content:
  5.         if hasattr(block, "text"):
  6.             serialized.append({"type": "text", "text": block.text})
  7.         elif block.type == "tool_use":
  8.             serialized.append({
  9.                 "type": "tool_use",
  10.                 "id": block.id,
  11.                 "name": block.name,
  12.                 "input": block.input
  13.             })
  14.     return serialized
  16. def run_agent_turn(messages, system_prompt):
  17.     """运行一个完整的 Agent 轮次（可能涉及多次工具调用）"""
  18.     while True:
  19.         response = client.messages.create(
  20.             model="claude-sonnet-4-5-20250929",
  21.             max_tokens=4096,
  22.             system=system_prompt,
  23.             tools=TOOLS,
  24.             messages=messages
  25.         )
  26.         content = serialize_content(response.content)
  27.         # 如果 AI 完成了（没有工具调用），返回文本
  28.         if response.stop_reason == "end_turn":
  29.             text = ""
  30.             for block in response.content:
  31.                 if hasattr(block, "text"):
  32.                     text += block.text
  33.             messages.append({"role": "assistant", "content": content})
  34.             return text, messages
  35.         # 处理工具调用
  36.         if response.stop_reason == "tool_use":
  37.             messages.append({"role": "assistant", "content": content})
  38.             tool_results = []
  39.             for block in response.content:
  40.                 if block.type == "tool_use":
  41.                     print(f"  工具: {block.name}({json.dumps(block.input)})")
  42.                     result = execute_tool(block.name, block.input)
  43.                     tool_results.append({
  44.                         "type": "tool_result",
  45.                         "tool_use_id": block.id,
  46.                         "content": str(result)
  47.                     })
  48.             messages.append({"role": "user", "content": tool_results})

更新消息处理器：

  1. async def handle_message(update: Update, context):
  2.     user_id = str(update.effective_user.id)
  3.     messages = load_session(user_id)
  4.     messages.append({"role": "user", "content": update.message.text})
  5.     response_text, messages = run_agent_turn(messages, SOUL)
  6.     save_session(user_id, messages)
  7.     await update.message.reply_text(response_text)

现在你可以发消息：

你： 创建一个叫 hello.py 的文件，打印 hello world，然后运行它

机器人： （使用 write_file 创建 hello.py） （使用 run_command 执行它） 完成！我创建了 hello.py 并运行了它。输出：”hello world”

AI 自己判断了用哪些工具、按什么顺序用。全程只靠一条文字消息。

别让它把你的电脑搞崩

我们在通过 Telegram 运行 shell 命令，这很吓人。如果有人黑了你的账号，让它执行 rm -rf / 怎么办？

我们需要权限控制：

  1. import re
  3. SAFE_COMMANDS = {"ls", "cat", "head", "tail", "wc", "date", "whoami", "echo"}
  4. DANGEROUS_PATTERNS = [r"\brm\b", r"\bsudo\b", r"\bchmod\b", r"\bcurl.*\|.*sh"]
  6. # 持久化白名单
  7. APPROVALS_FILE = "./exec-approvals.json"
  9. def load_approvals():
  10.     if os.path.exists(APPROVALS_FILE):
  11.         with open(APPROVALS_FILE) as f:
  12.             return json.load(f)
  13.     return {"allowed": [], "denied": []}
  15. def save_approval(command, approved):
  16.     approvals = load_approvals()
  17.     key = "allowed" if approved else "denied"
  18.     if command not in approvals[key]:
  19.         approvals[key].append(command)
  20.     with open(APPROVALS_FILE, "w") as f:
  21.         json.dump(approvals, f, indent=2)
  23. def check_command_safety(command):
  24.     """返回 'safe'、'approved' 或 'needs_approval'"""
  25.     base_cmd = command.strip().split()[0] if command.strip() else ""
  26.     if base_cmd in SAFE_COMMANDS:
  27.         return "safe"
  28.     approvals = load_approvals()
  29.     if command in approvals["allowed"]:
  30.         return "approved"
  31.     for pattern in DANGEROUS_PATTERNS:
  32.         if re.search(pattern, command):
  33.             return "needs_approval"
  34.     return "needs_approval"

更新 run_command 的处理逻辑：

  1. if name == "run_command":
  2.     cmd = input["command"]
  3.     safety = check_command_safety(cmd)
  4.     if safety == "needs_approval":
  5.         # 实际的机器人中，你会通过 Telegram 提示用户并等待回复
  6.         # 为简化起见，这里直接记录日志并拒绝
  7.         print(f"  ⚠  已拦截: {cmd}（需要授权）")
  8.         return "权限被拒绝。该命令需要授权。"
  9.     result = subprocess.run(
  10.         cmd, shell=True, capture_output=True,
  11.         text=True, timeout=30
  12.     )
  13.     return result.stdout + result.stderr

安全命令立即执行。可疑命令被拦截。授权记录保存到 exec-approvals.json，同一条命令不会被问两次。

网关模式

这里开始变得有意思了。现在我们有了一个 Telegram 机器人，但 Discord 呢？WhatsApp 呢？Slack 呢？

你可以写多个独立的机器人。但那样的话对话和记忆就是分开的——Telegram 上的 AI 不知道你在 Discord 上聊了什么。

解决方法：一个统一的中央网关，处理所有渠道。注意，我们的 run_agent_turn 函数对 Telegram 一无所知，它只接收消息、返回文本。所以我们可以随意添加任何界面。

来证明一下，在 Telegram 旁边加一个 HTTP API：

  1. from flask import Flask, request, jsonify
  2. import threading
  4. flask_app = Flask(__name__)
  6. @flask_app.route("/chat", methods=["POST"])
  7. def chat():
  8.     data = request.json
  9.     user_id = data["user_id"]
  10.     messages = load_session(user_id)
  11.     messages.append({"role": "user", "content": data["message"]})
  12.     response_text, messages = run_agent_turn(messages, SOUL)
  13.     save_session(user_id, messages)
  14.     return jsonify({"response": response_text})
  16. # 在后台线程中运行 HTTP API
  17. threading.Thread(target=lambda: flask_app.run(port=5000), daemon=True).start()
  19. # Telegram 机器人照常运行
  20. app = Application.builder().token(os.getenv("TELEGRAM_BOT_TOKEN")).build()
  21. app.add_handler(MessageHandler(filters.TEXT, handle_message))
  22. app.run_polling()

测试一下：

  1. # 通过 Telegram
  2. 你: 我叫 Rolly
  3. 机器人: 很高兴认识你，Rolly！
  5. # 通过 HTTP——使用你的 Telegram 用户 ID，这样就能命中同一个会话
  6. curl -X POST http://127.0.0.1:5000/chat \
  7.   -H "Content-Type: application/json" \
  8.   -d '{"user_id": "YOUR_TELEGRAM_USER_ID", "message": "我叫什么名字？"}'
  9. {"response": "你叫 Rolly！"}

同一个 Agent，同一个会话，同一份记忆，两个界面。这就是网关。

OpenClaw 对 Telegram、Discord、WhatsApp、Slack、Signal、iMessage 等都是这样做的，全部通过一个配置文件搞定。

对话太长怎么办

聊了几周之后，会话文件变得很大，最终会触碰模型的 token 上限。怎么办？

把旧消息压缩成摘要，保留最近的：

  1. def estimate_tokens(messages):
  2.     """粗略估算 token 数：约每 4 个字符 1 个 token"""
  3.     return sum(len(json.dumps(m)) for m in messages) // 4
  5. def compact_session(user_id, messages):
  6.     """当上下文太长时压缩旧消息"""
  7.     if estimate_tokens(messages) < 100_000:  # 约 128k 窗口的 80%
  8.         return messages  # 不需要压缩
  9.     split = len(messages) // 2
  10.     old, recent = messages[:split], messages[split:]
  11.     print("  正在压缩会话历史...")
  12.     summary = client.messages.create(
  13.         model="claude-sonnet-4-5-20250929",
  14.         max_tokens=2000,
  15.         messages=[{
  16.             "role": "user",
  17.             "content": (
  18.                 "简洁地总结这段对话。保留：\n"
  19.                 "- 关于用户的关键事实（姓名、偏好）\n"
  20.                 "- 重要决定\n"
  21.                 "- 未完成的任务或待办事项\n\n"
  22.                 f"{json.dumps(old, indent=2)}"
  23.             )
  24.         }]
  25.     )
  26.     compacted = [{
  27.         "role": "user",
  28.         "content": f"[之前的对话摘要]\n{summary.content[0].text}"
  29.     }] + recent
  30.     save_session(user_id, compacted)
  31.     return compacted

在处理器中加上这一行：

  1. async def handle_message(update: Update, context):
  2.     user_id = str(update.effective_user.id)
  3.     messages = load_session(user_id)
  4.     messages = compact_session(user_id, messages)  # <-- 加这一行
  5.     messages.append({"role": "user", "content": update.message.text})
  6.     response_text, messages = run_agent_turn(messages, SOUL)
  7.     save_session(user_id, messages)
  8.     await update.message.reply_text(response_text)

机器人仍然记得重要信息，会话文件也保持在可管理的大小。

永不丢失的记忆

会话给你对话记忆。但如果你重置会话或开启新会话呢？一切都消失了。

我们需要长期记忆——永久保存的文件。

添加这些工具：

  1. {
  2.     "name": "save_memory",
  3.     "description": "将重要信息保存到长期记忆中。用于用户偏好、关键事实，以及任何值得跨会话记住的内容。",
  4.     "input_schema": {
  5.         "type": "object",
  6.         "properties": {
  7.             "key": {
  8.                 "type": "string",
  9.                 "description": "简短标签，例如 'user-preferences'、'project-notes'"
  10.             },
  11.             "content": {
  12.                 "type": "string",
  13.                 "description": "要记住的信息"
  14.             }
  15.         },
  16.         "required": ["key", "content"]
  17.     }
  18. },
  19. {
  20.     "name": "memory_search",
  21.     "description": "在长期记忆中搜索相关信息。在对话开始时使用，以回忆之前会话的上下文。",
  22.     "input_schema": {
  23.         "type": "object",
  24.         "properties": {
  25.             "query": {
  26.                 "type": "string",
  27.                 "description": "要搜索的内容"
  28.             }
  29.         },
  30.         "required": ["query"]
  31.     }
  32. }

实现它们：

  1. MEMORY_DIR = "./memory"
  3. # 在 execute_tool 中添加这些分支：
  4. elif name == "save_memory":
  5.     os.makedirs(MEMORY_DIR, exist_ok=True)
  6.     filepath = os.path.join(MEMORY_DIR, f"{input['key']}.md")
  7.     with open(filepath, "w") as f:
  8.         f.write(input["content"])
  9.     return f"已保存到记忆：{input['key']}"
  10. elif name == "memory_search":
  11.     query = input["query"].lower()
  12.     results = []
  13.     if os.path.exists(MEMORY_DIR):
  14.         for fname in os.listdir(MEMORY_DIR):
  15.             if fname.endswith(".md"):
  16.                 with open(os.path.join(MEMORY_DIR, fname), "r") as f:
  17.                     content = f.read()
  18.                 if any(word in content.lower() for word in query.split()):
  19.                     results.append(f"--- {fname} ---\n{content}")
  20.     return "\n\n".join(results) if results else "未找到匹配的记忆。"

更新你的 SOUL：

  1. SOUL = """
  2. # 你是谁
  3. ...现有个性...
  5. ## 记忆
  6. 你有一个长期记忆系统。
  7. - 使用 save_memory 存储重要信息（用户偏好、关键事实、项目详情）。
  8. - 在对话开始时使用 memory_search 回忆之前会话的上下文。
  9. 记忆文件以 Markdown 格式存储在 ./memory/ 目录中。
  10. """

试一试：

你： 记住我最喜欢的餐厅是 Elvies，我喜欢在周末去。

机器人： （使用 save_memory 写入 Rolly-profile.md） 好的——已保存你的餐厅偏好。

（重置会话或重启机器人）

你： 今晚去哪里吃饭？

机器人： （使用 memory_search 搜索”餐厅 晚餐 最爱”） 去 Elvies 怎么样？我知道那是你最喜欢的。这个周末去吗？

记忆之所以持久，是因为它存在文件里，而不是会话里。重启所有东西——记忆还在。

防止竞争条件

这里有一个隐藏的 bug：如果两条消息同时到达怎么办？

你在 Telegram 发”查一下我的日历”，同时通过 HTTP 发”今天天气怎么样”。两个请求都试图加载同一个会话，都试图写入它，最终数据就损坏了。

解决方法：每个会话一把锁。

  1. # 加到你的机器人里
  2. from collections import defaultdict
  4. session_locks = defaultdict(threading.Lock)

包装你的处理器：

  1. async def handle_message(update: Update, context):
  2.     user_id = str(update.effective_user.id)
  3.     with session_locks[user_id]:
  4.         messages = load_session(user_id)
  5.         messages = compact_session(user_id, messages)
  6.         messages.append({"role": "user", "content": update.message.text})
  7.         response_text, messages = run_agent_turn(messages, SOUL)
  8.         save_session(user_id, messages)
  9.     await update.message.reply_text(response_text)

HTTP 端点也一样：

  1. @flask_app.route("/chat", methods=["POST"])
  2. def chat():
  3.     data = request.json
  4.     user_id = data["user_id"]
  5.     with session_locks[user_id]:
  6.         messages = load_session(user_id)
  7.         messages = compact_session(user_id, messages)
  8.         messages.append({"role": "user", "content": data["message"]})
  9.         response_text, messages = run_agent_turn(messages, SOUL)
  10.         save_session(user_id, messages)
  11.     return jsonify({"response": response_text})

完成。同一个用户的消息排队处理，不同用户并行运行，不再有数据损坏。

让它自己醒来

现在你的 AI 只在你跟它说话时才工作。但如果你想让它每天早上检查你的邮件呢？或者提醒你开会呢？

你需要定时任务——心跳机制。

  1. import schedule
  2. import time
  4. def setup_heartbeats():
  5.     """配置周期性 Agent 任务"""
  6.     def morning_briefing():
  7.         print("\n⏰ 心跳：早间简报")
  8.         # 使用独立的会话 key，避免 cron 污染主聊天
  9.         session_key = "cron:morning-briefing"
  10.         with session_locks[session_key]:
  11.             messages = load_session(session_key)
  12.             messages.append({
  13.                 "role": "user",
  14.                 "content": "早上好！查一下今天的日期，给我一句励志名言。"
  15.             })
  16.             response_text, messages = run_agent_turn(messages, SOUL)
  17.             save_session(session_key, messages)
  18.         print(f"  {response_text}\n")
  19.         # 生产环境中，你还会把这条消息发到 Telegram/Discord
  21.     schedule.every().day.at("07:30").do(morning_briefing)
  23.     # 在后台线程中运行调度器
  24.     def scheduler_loop():
  25.         while True:
  26.             schedule.run_pending()
  27.             time.sleep(60)
  29.     threading.Thread(target=scheduler_loop, daemon=True).start()
  31. # 在 run_polling() 之前调用
  32. setup_heartbeats()

关键点：每个心跳任务使用自己的会话 key（cron:morning-briefing），让定时任务和你的主聊天历史保持分离。

测试时，改成每分钟运行一次：

  1. schedule.every(1).minutes.do(morning_briefing)

你会在终端看到它触发。测完改回每天一次。

多个 Agent

一个 Agent 很有用，但随着任务增多，一种个性无法胜任所有事情。研究助手需要不同的指令，和通用助手不一样。

解决方案：多个 Agent 配置加路由。

  1. AGENTS = {
  2.     "main": {
  3.         "name": "Jarvis",
  4.         "soul": SOUL,  # 我们现有的 SOUL
  5.         "session_prefix": "agent:main",
  6.     },
  7.     "researcher": {
  8.         "name": "Scout",
  9.         "soul": """你是 Scout，一位研究专家。
  10. 你的工作：找到信息并注明来源。每一个论断都需要证据。
  11. 使用工具收集数据。要全面但简洁。
  12. 用 save_memory 保存重要发现，供其他 Agent 参考。""",
  13.         "session_prefix": "agent:researcher",
  14.     },
  15. }
  17. def resolve_agent(message_text):
  18.     """根据前缀命令将消息路由到正确的 Agent"""
  19.     if message_text.startswith("/research "):
  20.         return "researcher", message_text[len("/research "):]
  21.     return "main", message_text

更新你的处理器：

  1. async def handle_message(update: Update, context):
  2.     user_id = str(update.effective_user.id)
  3.     agent_id, message_text = resolve_agent(update.message.text)
  4.     agent = AGENTS[agent_id]
  5.     session_key = f"{agent['session_prefix']}:{user_id}"
  6.     with session_locks[session_key]:
  7.         messages = load_session(session_key)
  8.         messages = compact_session(session_key, messages)
  9.         messages.append({"role": "user", "content": message_text})
  10.         response_text, messages = run_agent_turn(messages, agent["soul"])
  11.         save_session(session_key, messages)
  12.     await update.message.reply_text(f"[{agent['name']}] {response_text}")

试一试：

你： 今天天气怎么样？ [Jarvis] 天气不错！具体数据可以查一下天气服务。

你： /research Python 异步编程的最佳实践是什么？ [Scout] 这是我找到的内容…… （使用 web_search、save_memory 收集并存储发现） 核心实践包括：1）使用 asyncio.gather 处理并发任务……

你： Scout 发现了哪些关于 Python 异步的内容？ [Jarvis] （使用 memory_search） Scout 的研究发现，核心异步最佳实践包括……

每个 Agent 有自己的对话历史，但共享记忆。Scout 保存研究成果，Jarvis 之后可以搜索到。它们通过共享文件协同工作。

完整版代码

好了，把所有东西拼在一起。这是一个包含所有功能的完整可运行脚本：

  1. #!/usr/bin/env python3
  2. # mini-openclaw.py - 一个极简的 OpenClaw 克隆
  3. # 运行方式：uv run --with anthropic --with schedule python mini-openclaw.py
  5. import anthropic
  6. import subprocess
  7. import json
  8. import os
  9. import re
  10. import threading
  11. import time
  12. import schedule
  13. from collections import defaultdict
  14. from datetime import datetime
  16. client = anthropic.Anthropic()
  18. # ─── 配置 ───
  19. WORKSPACE = os.path.expanduser("~/.mini-openclaw")
  20. SESSIONS_DIR = os.path.join(WORKSPACE, "sessions")
  21. MEMORY_DIR = os.path.join(WORKSPACE, "memory")
  22. APPROVALS_FILE = os.path.join(WORKSPACE, "exec-approvals.json")
  24. # ─── Agents ───
  25. AGENTS = {
  26.     "main": {
  27.         "name": "Jarvis",
  28.         "model": "claude-sonnet-4-5-20250929",
  29.         "soul": (
  30.             "你是 Jarvis，一位个人 AI 助手。\n"
  31.             "真正有帮助。省掉客套话。有自己的观点。\n"
  32.             "你有工具——主动使用它们。\n\n"
  33.             "## 记忆\n"
  34.             f"你的工作区是 {WORKSPACE}。\n"
  35.             "使用 save_memory 跨会话存储重要信息。\n"
  36.             "在对话开始时使用 memory_search 回忆上下文。"
  37.         ),
  38.         "session_prefix": "agent:main",
  39.     },
  40.     "researcher": {
  41.         "name": "Scout",
  42.         "model": "claude-sonnet-4-5-20250929",
  43.         "soul": (
  44.             "你是 Scout，一位研究专家。\n"
  45.             "你的工作：找到信息并注明来源。每个论断都需要证据。\n"
  46.             "使用工具收集数据。全面但简洁。\n"
  47.             "用 save_memory 保存重要发现，供其他 Agent 参考。"
  48.         ),
  49.         "session_prefix": "agent:researcher",
  50.     },
  51. }
  53. # ─── 工具 ───
  54. TOOLS = [
  55.     {
  56.         "name": "run_command",
  57.         "description": "运行一个 shell 命令",
  58.         "input_schema": {
  59.             "type": "object",
  60.             "properties": {
  61.                 "command": {"type": "string", "description": "要运行的命令"}
  62.             },
  63.             "required": ["command"]
  64.         }
  65.     },
  66.     {
  67.         "name": "read_file",
  68.         "description": "从文件系统读取文件",
  69.         "input_schema": {
  70.             "type": "object",
  71.             "properties": {
  72.                 "path": {"type": "string", "description": "文件路径"}
  73.             },
  74.             "required": ["path"]
  75.         }
  76.     },
  77.     {
  78.         "name": "write_file",
  79.         "description": "向文件写入内容（如需要会自动创建目录）",
  80.         "input_schema": {
  81.             "type": "object",
  82.             "properties": {
  83.                 "path": {"type": "string", "description": "文件路径"},
  84.                 "content": {"type": "string", "description": "要写入的内容"}
  85.             },
  86.             "required": ["path", "content"]
  87.         }
  88.     },
  89.     {
  90.         "name": "save_memory",
  91.         "description": "将重要信息保存到长期记忆",
  92.         "input_schema": {
  93.             "type": "object",
  94.             "properties": {
  95.                 "key": {"type": "string", "description": "简短标签（例如 'user-preferences'）"},
  96.                 "content": {"type": "string", "description": "要记住的信息"}
  97.             },
  98.             "required": ["key", "content"]
  99.         }
  100.     },
  101.     {
  102.         "name": "memory_search",
  103.         "description": "在长期记忆中搜索相关信息",
  104.         "input_schema": {
  105.             "type": "object",
  106.             "properties": {
  107.                 "query": {"type": "string", "description": "要搜索的内容"}
  108.             },
  109.             "required": ["query"]
  110.         }
  111.     },
  112. ]
  114. # ─── 权限控制 ───
  115. SAFE_COMMANDS = {"ls", "cat", "head", "tail", "wc", "date", "whoami",
  116.                  "echo", "pwd", "which", "git", "python", "node", "npm"}
  118. def load_approvals():
  119.     if os.path.exists(APPROVALS_FILE):
  120.         with open(APPROVALS_FILE) as f:
  121.             return json.load(f)
  122.     return {"allowed": [], "denied": []}
  124. def save_approval(command, approved):
  125.     approvals = load_approvals()
  126.     key = "allowed" if approved else "denied"
  127.     if command not in approvals[key]:
  128.         approvals[key].append(command)
  129.     with open(APPROVALS_FILE, "w") as f:
  130.         json.dump(approvals, f, indent=2)
  132. def check_command_safety(command):
  133.     base_cmd = command.strip().split()[0] if command.strip() else ""
  134.     if base_cmd in SAFE_COMMANDS:
  135.         return "safe"
  136.     approvals = load_approvals()
  137.     if command in approvals["allowed"]:
  138.         return "approved"
  139.     return "needs_approval"
  141. # ─── 工具执行 ───
  142. def execute_tool(name, tool_input):
  143.     if name == "run_command":
  144.         cmd = tool_input["command"]
  145.         safety = check_command_safety(cmd)
  146.         if safety == "needs_approval":
  147.             print(f"\n  ⚠️  命令: {cmd}")
  148.             confirm = input("  是否允许？(y/n): ").strip().lower()
  149.             if confirm != "y":
  150.                 save_approval(cmd, False)
  151.                 return "用户拒绝了权限。"
  152.             save_approval(cmd, True)
  153.         try:
  154.             result = subprocess.run(
  155.                 cmd, shell=True, capture_output=True, text=True, timeout=30
  156.             )
  157.             output = result.stdout + result.stderr
  158.             return output if output else "（无输出）"
  159.         except subprocess.TimeoutExpired:
  160.             return "命令在 30 秒后超时"
  161.         except Exception as e:
  162.             return f"错误: {e}"
  163.     elif name == "read_file":
  164.         try:
  165.             with open(tool_input["path"], "r") as f:
  166.                 return f.read()[:10000]
  167.         except Exception as e:
  168.             return f"错误: {e}"
  169.     elif name == "write_file":
  170.         try:
  171.             os.makedirs(os.path.dirname(tool_input["path"]) or ".", exist_ok=True)
  172.             with open(tool_input["path"], "w") as f:
  173.                 f.write(tool_input["content"])
  174.             return f"已写入 {tool_input['path']}"
  175.         except Exception as e:
  176.             return f"错误: {e}"
  177.     elif name == "save_memory":
  178.         os.makedirs(MEMORY_DIR, exist_ok=True)
  179.         filepath = os.path.join(MEMORY_DIR, f"{tool_input['key']}.md")
  180.         with open(filepath, "w") as f:
  181.             f.write(tool_input["content"])
  182.         return f"已保存到记忆：{tool_input['key']}"
  183.     elif name == "memory_search":
  184.         query = tool_input["query"].lower()
  185.         results = []
  186.         if os.path.exists(MEMORY_DIR):
  187.             for fname in os.listdir(MEMORY_DIR):
  188.                 if fname.endswith(".md"):
  189.                     with open(os.path.join(MEMORY_DIR, fname), "r") as f:
  190.                         content = f.read()
  191.                     if any(w in content.lower() for w in query.split()):
  192.                         results.append(f"--- {fname} ---\n{content}")
  193.         return "\n\n".join(results) if results else "未找到匹配的记忆。"
  194.     return f"未知工具：{name}"
  196. # ─── 会话管理 ───
  197. def get_session_path(session_key):
  198.     os.makedirs(SESSIONS_DIR, exist_ok=True)
  199.     safe_key = session_key.replace(":", "_").replace("/", "_")
  200.     return os.path.join(SESSIONS_DIR, f"{safe_key}.jsonl")
  202. def load_session(session_key):
  203.     path = get_session_path(session_key)
  204.     messages = []
  205.     if os.path.exists(path):
  206.         with open(path, "r") as f:
  207.             for line in f:
  208.                 if line.strip():
  209.                     try:
  210.                         messages.append(json.loads(line))
  211.                     except json.JSONDecodeError:
  212.                         continue
  213.     return messages
  215. def append_message(session_key, message):
  216.     with open(get_session_path(session_key), "a") as f:
  217.         f.write(json.dumps(message) + "\n")
  219. def save_session(session_key, messages):
  220.     with open(get_session_path(session_key), "w") as f:
  221.         for msg in messages:
  222.             f.write(json.dumps(msg) + "\n")
  224. # ─── 上下文压缩 ───
  225. def estimate_tokens(messages):
  226.     return sum(len(json.dumps(m)) for m in messages) // 4
  228. def compact_session(session_key, messages):
  229.     if estimate_tokens(messages) < 100_000:
  230.         return messages
  231.     split = len(messages) // 2
  232.     old, recent = messages[:split], messages[split:]
  233.     print("\n  正在压缩会话历史...")
  234.     summary = client.messages.create(
  235.         model="claude-sonnet-4-5-20250929",
  236.         max_tokens=2000,
  237.         messages=[{
  238.             "role": "user",
  239.             "content": (
  240.                 "简洁地总结这段对话。保留关键事实、"
  241.                 "决定和未完成任务：\n\n"
  242.                 f"{json.dumps(old, indent=2)}"
  243.             )
  244.         }]
  245.     )
  246.     compacted = [{
  247.         "role": "user",
  248.         "content": f"[对话摘要]\n{summary.content[0].text}"
  249.     }] + recent
  250.     save_session(session_key, compacted)
  251.     return compacted
  253. # ─── 会话锁 ───
  254. session_locks = defaultdict(threading.Lock)
  256. # ─── Agent 循环 ───
  257. def serialize_content(content):
  258.     serialized = []
  259.     for block in content:
  260.         if hasattr(block, "text"):
  261.             serialized.append({"type": "text", "text": block.text})
  262.         elif block.type == "tool_use":
  263.             serialized.append({
  264.                 "type": "tool_use", "id": block.id,
  265.                 "name": block.name, "input": block.input
  266.             })
  267.     return serialized
  269. def run_agent_turn(session_key, user_text, agent_config):
  270.     """运行一个完整的 Agent 轮次：加载会话、循环调用 LLM、保存。"""
  271.     with session_locks[session_key]:
  272.         messages = load_session(session_key)
  273.         messages = compact_session(session_key, messages)
  274.         user_msg = {"role": "user", "content": user_text}
  275.         messages.append(user_msg)
  276.         append_message(session_key, user_msg)
  277.         for _ in range(20):  # 最多工具调用轮次
  278.             response = client.messages.create(
  279.                 model=agent_config["model"],
  280.                 max_tokens=4096,
  281.                 system=agent_config["soul"],
  282.                 tools=TOOLS,
  283.                 messages=messages
  284.             )
  285.             content = serialize_content(response.content)
  286.             assistant_msg = {"role": "assistant", "content": content}
  287.             messages.append(assistant_msg)
  288.             append_message(session_key, assistant_msg)
  289.             if response.stop_reason == "end_turn":
  290.                 return "".join(
  291.                     b.text for b in response.content if hasattr(b, "text")
  292.                 )
  293.             if response.stop_reason == "tool_use":
  294.                 tool_results = []
  295.                 for block in response.content:
  296.                     if block.type == "tool_use":
  297.                         print(f"  🔧 {block.name}: {json.dumps(block.input)[:100]}")
  298.                         result = execute_tool(block.name, block.input)
  299.                         display = str(result)[:150]
  300.                         print(f"     → {display}")
  301.                         tool_results.append({
  302.                             "type": "tool_result",
  303.                             "tool_use_id": block.id,
  304.                             "content": str(result)
  305.                         })
  306.                 results_msg = {"role": "user", "content": tool_results}
  307.                 messages.append(results_msg)
  308.                 append_message(session_key, results_msg)
  309.         return "（已达到最大轮次）"
  311. # ─── 多 Agent 路由 ───
  312. def resolve_agent(message_text):
  313.     """根据前缀命令将消息路由到正确的 Agent"""
  314.     if message_text.startswith("/research "):
  315.         return "researcher", message_text[len("/research "):]
  316.     return "main", message_text
  318. # ─── 定时任务 / 心跳 ───
  319. def setup_heartbeats():
  320.     def morning_check():
  321.         print("\n⏰ 心跳：早间检查")
  322.         result = run_agent_turn(
  323.             "cron:morning-check",
  324.             "早上好！查一下今天的日期，给我一句励志名言。",
  325.             AGENTS["main"]
  326.         )
  327.         print(f"  {result}\n")
  328.     schedule.every().day.at("07:30").do(morning_check)
  329.     def scheduler_loop():
  330.         while True:
  331.             schedule.run_pending()
  332.             time.sleep(60)
  333.     threading.Thread(target=scheduler_loop, daemon=True).start()
  335. # ─── 交互式命令行 ───
  336. def main():
  337.     for d in [WORKSPACE, SESSIONS_DIR, MEMORY_DIR]:
  338.         os.makedirs(d, exist_ok=True)
  339.     setup_heartbeats()
  340.     session_key = "agent:main:repl"
  341.     print("Mini OpenClaw")
  342.     print(f"  Agents: {', '.join(a['name'] for a in AGENTS.values())}")
  343.     print(f"  工作区: {WORKSPACE}")
  344.     print("  命令: /new（重置）、/research <查询>、/quit\n")
  345.     while True:
  346.         try:
  347.             user_input = input("你: ").strip()
  348.         except (EOFError, KeyboardInterrupt):
  349.             print("\n再见！")
  350.             break
  351.         if not user_input:
  352.             continue
  353.         if user_input.lower() in ["/quit", "/exit", "/q"]:
  354.             print("再见！")
  355.             break
  356.         if user_input.lower() == "/new":
  357.             session_key = f"agent:main:repl:{datetime.now().strftime('%Y%m%d%H%M%S')}"
  358.             print("  会话已重置。\n")
  359.             continue
  360.         agent_id, message_text = resolve_agent(user_input)
  361.         agent_config = AGENTS[agent_id]
  362.         sk = (
  363.             f"{agent_config['session_prefix']}:repl"
  364.             if agent_id != "main" else session_key
  365.         )
  366.         response = run_agent_turn(sk, message_text, agent_config)
  367.         print(f"\n [{agent_config['name']}] {response}\n")
  369. if __name__ == "__main__":
  370.     main()

保存为 mini-openclaw.py 并运行：

  1. uv run --with anthropic --with schedule python mini-openclaw.py

运行效果如下：

  1. Mini OpenClaw
  2.   Agents: Jarvis, Scout
  3.   工作区: ~/.mini-openclaw
  4.   命令: /new（重置）、/research <查询>、/quit
  6. 你: 记住我最喜欢的餐厅是 Hai Cenato，我喜欢 7 点的预订
  7.   🔧 save_memory: {"key": "user-preferences", "content": "最喜欢的餐厅...
  8.      → 已保存到记忆：user-preferences
  9.  [Jarvis] 好的，已保存你的餐厅偏好——Hai Cenato，7 点预订。
  11. 你: 我项目目录里有什么？
  12.   🔧 run_command: {"command": "ls"}
  13.      → src, package.json, README.md, node_modules, ...
  14.  [Jarvis] 你的项目是标准的 Node.js 结构，有 src/、package.json，还有那些常见的文件。
  16. 你: /new
  17.   会话已重置。
  19. 你: 我喜欢在哪里吃饭？
  20.   🔧 memory_search: {"query": "restaurant favorite food"}
  21.      → --- user-preferences.md ---
  22.         最喜欢的餐厅：Hai Cenato...
  23.  [Jarvis] 你喜欢 Hai Cenato，偏好 7 点的预订。
  25. 你: /research AI Agent 的最新趋势是什么？
  26.   🔧 web_search: {"query": "AI agent trends 2025"}
  27.      → AI agent trends 2025 的搜索结果
  28.   🔧 save_memory: {"key": "research-ai-agents", ...}
  29.      → 已保存到记忆：research-ai-agents
  30.  [Scout] 这是我找到的关于当前 AI Agent 趋势的内容……

我们做了什么

从零开始，我们构建了：

  • 持久会话 — 崩溃后仍能保留的对话记忆
  • SOUL.md — 让 AI 行为保持一致的个性文件
  • 工具 + Agent 循环 — 让 AI 真正能做事情
  • 权限控制 — 防止它把你的电脑搞崩
  • 网关模式 — 一个 Agent，多个界面
  • 上下文压缩 — 处理长对话
  • 长期记忆 — 重置后仍能保留的知识
  • 命令队列 — 防止竞争条件
  • 心跳机制 — 定时任务
  • 多 Agent — 不同任务用不同的 Agent

每一个组件都解决了一个真实问题。没有魔法，只是实际可行的解决方案。

更进一步

这个原型包含了核心架构。OpenClaw 在此基础上增加了生产级功能：

浏览器自动化 — 控制真实浏览器，使用语义快照（文本而非截图，token 消耗少得多）

会话作用域 — 配置会话是按用户、按频道还是共享

插件系统 — 添加新渠道而无需修改核心代码

向量搜索 — 基于 embedding 的语义记忆匹配

子 Agent — Agent 可以为特定任务生成子 Agent

但说真的，从简单开始。先让一个渠道跑起来，按需添加工具，会话不够用了再加记忆。复杂度来自你的需求，而不是照着某个蓝图硬套。

或者直接用 OpenClaw 也行，它是开源的，处理了很多边界情况。但现在你知道它的工作原理了。

（全文完）