miniOpenClaw —— 从零构建一个 Python AI Agent 框架

项目目标

使用 Python 从零实现一个名为 miniOpenClaw 的轻量级 AI Agent 框架，以渐进式、模块化的方式逐章构建，最终形成一个具备 Gateway 接入、多渠道适配、会话管理、ReAct 推理、工具调用、技能编排、记忆系统和 MCP 协议支持的完整 Agent 平台。

总体要求

• Python 版本：3.11+（使用 asyncio.TaskGroup、StrEnum 等新特性）
• 项目根目录：/data/workspace/learnNote/miniOpenClaw
• 每个章节对应一个独立可运行的子目录（如 day1-gateway/、day2-channel/），同时在项目根目录维护一个持续演进的 miniclaw/ 核心库
• 全面使用 async/await 异步编程模型
• 代码风格遵循 Python 惯例（snake_case 函数/变量，PascalCase 类名），通过 ruff 或 flake8 检查
• 每章需包含：可运行的示例代码（example/）、单元测试（覆盖率 ≥ 80%）、README.md 说明文档

章节依赖关系

Day1(Gateway) → Day2(Channel) → Day3(Session) → Day4(Agent Runtime)
                                                       ↑
Day5(Tools) ──────────────────────────────────────────┘
Day6(Skills) ─────────────────────────────────────────┘
Day7(Memory) ─────────────────────────────────────────┘
Day8(MCP) ──→ Day5(Tools)

---

章节规划

Day 1 — Gateway 与消息协议

目标： 搭建 Agent 平台的通信基座——基于 WebSocket 的 Gateway 服务，定义统一消息协议，实现消息路由和事件总线

• 定义 GatewayMessage 协议格式（基于 Pydantic）：
  • 字段：msg_id、msg_type（枚举：text / command / event / error）、source、target、payload、timestamp
  • 支持 JSON 序列化/反序列化与校验
• 实现 GatewayServer：
  • 基于 websockets 库实现异步 WebSocket 服务端
  • 连接管理：维护活跃连接池，处理连接/断开事件
  • 消息收发：接收客户端消息 → 解析为 GatewayMessage → 分发处理
• 实现 MessageRouter：
  • 基于 msg_type 的消息路由分发
  • 支持注册自定义消息处理器（handler 模式）
• 实现 EventBus：
  • 异步发布-订阅模式，解耦组件间通信
  • 支持事件注册、发布和监听器管理
• 交付物： gateway 包、协议序列化测试、WebSocket 连接集成测试、EventBus 单元测试

Day 2 — Channel 适配器

目标： 抽象消息渠道层，使 Agent 能够对接 CLI、Web 聊天、Webhook 等不同输入/输出渠道

• 定义 Channel 抽象基类（ABC）：
  • 核心接口：receive() 接收用户输入 → GatewayMessage，send() 将 Agent 响应发送到渠道
  • 生命周期方法：start() / stop()
• 实现 CLIChannel：
  • 基于 asyncio 的终端交互式输入/输出
  • 支持命令行参数启动、优雅退出（Ctrl+C）
• 实现 WebChatChannel：
  • 基于 aiohttp 的简易 Web 聊天服务端
  • 提供 HTTP 接口接收消息、WebSocket 推送响应
• 实现 WebhookChannel：
  • 接收外部 HTTP 回调（如企业微信、飞书、Slack 等消息推送）
  • 统一转换为 GatewayMessage 格式
• Channel 与 Gateway 的集成：Channel 作为 Gateway 的消息来源，通过 EventBus 解耦
• 交付物： channel 包、各 Channel 单元测试、CLI + WebChat 集成示例

Day 3 — 会话管理（Session）

目标： 实现多用户、多轮对话的会话管理机制，为 Agent 提供对话上下文

• 定义 Session 数据模型（Pydantic BaseModel）：
  • 字段：session_id、user_id、channel、created_at、updated_at、messages（消息历史列表）、metadata
• 实现 SessionManager：
  • 会话的创建、获取、更新、过期清理
  • 基于 user_id + channel 的会话路由（同一用户在不同渠道可有不同会话）
  • 会话超时机制：支持配置 TTL，自动清理过期会话
• 实现 SessionStorage 接口及多种后端：
  • InMemoryStorage：基于 dict 的内存存储（开发/测试用）
  • FileStorage：基于 JSON 文件的持久化存储
  • 预留接口：方便后续扩展 Redis 等存储后端
• 会话与 Gateway 集成：收到消息时自动查找/创建对应会话
• 交付物： session 包、会话 CRUD 测试、会话过期清理测试、多存储后端测试

Day 4 — Agent 运行时（Agent Runtime）

目标： 实现 Agent 的核心推理循环（ReAct 模式），完成"接收问题 → 思考 → 行动 → 观察 → 回答"的完整闭环

• 实现 LLMProvider 抽象与 OpenAIProvider：
  • 定义统一的 LLM 调用接口：chat(messages, tools?) → response
  • 基于 openai 库实现 OpenAI / 兼容 API 的调用
  • 支持配置：模型名称、temperature、max_tokens 等参数
  • 流式响应支持（streaming）
• 实现 AgentRuntime（ReAct 循环）：
  • 核心循环：用户输入 → LLM 推理 → 判断是否需要调用工具 → 调用工具获取结果 → 将结果反馈 LLM → 最终回答
  • 最大迭代次数限制（防止无限循环）
  • 错误处理：工具调用失败时的重试与降级策略
• 实现 AgentConfig：
  • 系统提示词（system prompt）配置
  • LLM 参数配置
  • 运行时行为配置（最大迭代次数、超时时间等）
• Agent 与 Session 集成：每次对话基于 Session 的消息历史构建 LLM 上下文
• 交付物： agent 包、ReAct 循环单元测试（使用 mock LLM）、与真实 LLM 的集成示例

Day 5 — 工具系统（Tool System）

目标： 构建可扩展的工具注册与调用框架，使 Agent 具备调用外部能力（搜索、计算、文件操作等）的能力

• 定义 Tool 基类与 @tool 装饰器：
  • Tool 描述信息：name、description、parameters（JSON Schema 格式，用于 LLM function calling）
  • 工具执行接口：async execute(**kwargs) → ToolResult
  • @tool 装饰器：将普通 async 函数快速注册为工具，自动从类型标注生成 JSON Schema
• 实现 ToolRegistry：
  • 工具注册、查找、列举
  • 自动生成符合 OpenAI function calling 格式的工具描述列表
  • 工具名冲突检测
• 实现内置工具集（builtins/）：
  • datetime_tool：获取当前日期时间
  • calculator_tool：安全的数学表达式计算（基于 ast.literal_eval 或受限 eval）
  • http_request_tool：发起 HTTP 请求（基于 httpx）
• 工具与 AgentRuntime 集成：LLM 返回 tool_call → ToolRegistry 查找 → 执行 → 结果回传
• 交付物： tools 包、工具注册/调用测试、装饰器测试、内置工具测试、端到端工具调用示例

Day 6 — 技能系统（Skill System）

目标： 在工具之上构建更高层的"技能"抽象，支持通过 Markdown 文件定义可复用的 Agent 行为模式

• 定义 Skill 数据模型：
  • 技能元信息：name、description、author、version、tags
  • 技能内容：system_prompt（注入到系统提示词的行为指令）、required_tools（该技能依赖的工具列表）
  • 技能触发条件：trigger（关键词匹配 / 手动激活）
• 实现 SkillLoader：
  • 从 Markdown 文件加载技能定义（Front Matter 解析元信息，正文作为 system_prompt）
  • 从指定目录批量扫描加载技能
  • 技能校验：检查 required_tools 是否已注册
• 实现 SkillManager：
  • 技能的注册、激活、停用管理
  • 根据用户输入自动匹配适用技能（基于 tags / trigger）
  • 将激活技能的 system_prompt 注入到 Agent 上下文
• 技能与 AgentRuntime 集成：运行时根据激活的技能动态调整系统提示词和可用工具集
• 交付物： skills 包、Skill 加载解析测试、技能匹配测试、技能+Agent 端到端示例

Day 7 — 记忆与上下文管理（Memory & Context）

目标： 实现分层记忆系统和智能上下文组装，使 Agent 具备短期对话记忆和长期知识记忆能力

• 实现 SystemPromptBuilder（上下文组装器）：
  • 分层组装系统提示词：基础人设层 → 技能层（从已加载的 Skill 中注入相关提示词） → 工具层（自动生成工具描述） → 动态上下文层（会话历史 + 时间信息）
• 实现记忆系统：
  • ShortTermMemory：当前会话消息历史（已在 Session 中）
  • LongTermMemory：基于关键信息提取的持久化记忆（Markdown 文件存储）
  • 记忆的写入（Agent 主动调用 memory_save 工具）和读取（上下文组装时自动检索）
• 上下文窗口管理：
  • Token 计数估算（基于 tiktoken 或简单字符数估算）
  • 当上下文超过限制时的压缩策略：截断旧消息 / 摘要压缩
• 交付物： memory 包、上下文组装测试、记忆读写测试、Token 管理测试

Day 8 — MCP 协议支持（可选进阶）

目标： 实现 MCP（Model Context Protocol）客户端和服务端，使 miniOpenClaw 能够作为 MCP Host 连接外部 MCP Server，也能将自身能力暴露为 MCP Server

• 实现 MCP 基础协议：
  • JSON-RPC 2.0 消息格式
  • initialize / tools/list / tools/call 核心方法
  • stdio 传输（本地进程间通信，优先实现）
  • Streamable HTTP 传输（远程服务连接，可选）
• 实现 MCPClient：
  • 连接外部 MCP Server（如社区提供的各种 MCP 工具服务）
  • 自动发现远端工具并注册到 ToolRegistry
  • 代理工具调用：Agent 调用 → MCPClient 转发 → 远端执行 → 结果返回
• 实现 MCPServer：
  • 将 miniOpenClaw 的内置工具通过 MCP 协议暴露
  • 支持外部 MCP 客户端（如 Claude Desktop、Cursor）连接使用
• MCP 配置管理：通过配置文件定义要连接的 MCP Server 列表
• 交付物： mcp 包、MCP 协议编解码测试、MCP Client/Server 集成测试、与外部 MCP Server 对接示例

---

每章交付规范

dayN-xxx/
├── README.md           # 本章学习目标、核心概念、架构图、关键代码讲解
├── doc/
│   └── design.md       # 技术设计文档：接口定义、流程图、设计决策与取舍
├── example/
│   └── main.py         # 可独立运行的示例（或 server.py / client.py）
├── <package>/
│   ├── *.py            # 核心实现代码
│   └── test_*.py       # 单元测试（覆盖率 ≥ 80%）
├── pyproject.toml      # 本章依赖声明
└── requirements.txt    # pip 依赖（兼容性用途）

验收标准

• [ ] pytest 全部通过，覆盖率 ≥ 80%（使用 pytest --cov 验证）
• [ ] example/ 下的示例可独立运行并产生预期输出
• [ ] design.md 包含核心接口定义和流程时序图
• [ ] 与前序章节的集成点已验证通过

核心库演进规范

miniclaw/                    # 包名简写，项目名 miniOpenClaw
├── __init__.py
├── gateway/                 # Day 1: Gateway 与消息协议
│   ├── server.py            # WebSocket Server
│   ├── protocol.py          # GatewayMessage 定义
│   ├── router.py            # MessageRouter
│   └── events.py            # EventBus
├── channel/                 # Day 2: Channel 适配器
│   ├── base.py              # Channel 抽象基类
│   ├── cli.py               # CLI Channel
│   ├── webchat.py           # WebChat Channel
│   └── webhook.py           # Webhook Channel
├── session/                 # Day 3: 会话管理
│   ├── manager.py           # SessionManager
│   ├── models.py            # Session 数据模型
│   └── storage.py           # 会话持久化
├── agent/                   # Day 4: Agent 运行时
│   ├── runtime.py           # ReAct 循环
│   ├── providers.py         # LLM Provider
│   └── config.py            # Agent 配置
├── tools/                   # Day 5: 工具系统
│   ├── registry.py          # ToolRegistry
│   ├── base.py              # Tool 基类与装饰器
│   └── builtins/            # 内置工具
├── skills/                  # Day 6: 技能系统
│   ├── manager.py           # SkillManager
│   └── loader.py            # Skill 加载器
├── memory/                  # Day 7: 记忆与上下文
│   ├── context.py           # SystemPromptBuilder
│   ├── short_term.py        # 短期记忆
│   └── long_term.py         # 长期记忆
└── mcp/                     # Day 8: MCP 协议（可选）
    ├── protocol.py          # MCP 消息定义
    ├── client.py            # MCP Client
    └── server.py            # MCP Server

文档要求

• README.md： 用通俗的语言解释本章实现了什么、为什么这样设计、如何运行示例
• design.md： 包含接口设计（Python ABC / Protocol 定义）、核心流程时序图（可用 Mermaid）、与上一章的差异说明、关键设计决策的 trade-off 分析
• 所有文档使用中文编写

约束与风格

• 不使用任何现成的 AI Agent 框架（如 LangChain、AutoGen、CrewAI），可使用标准库和基础工具库
• 允许使用的第三方库：
  • websockets（WebSocket 服务）
  • aiohttp（HTTP 服务端）
  • httpx（HTTP 客户端）
  • pydantic（数据校验）
  • tiktoken（Token 计数）
  • openai（仅作为 LLM API 客户端）
  • pytest + pytest-asyncio + pytest-cov（测试与覆盖率）
  • python-dotenv（环境变量管理）
• 优先可读性，代码中对关键设计决策添加注释说明（不要冗余注释）
• 每章代码可独立运行，不强依赖后续章节
• 全面使用 async/await 异步编程模型
• 变量/函数命名遵循 Python 惯例（snake_case），类名使用 PascalCase
• 所有公开接口必须有类型标注
• 错误处理：定义自定义异常层级（MiniClawError 基类），不裸抛 Exception
• 配置管理：敏感信息（API Key 等）通过环境变量注入，不硬编码
• 日志：使用 logging 模块，关键路径记录 DEBUG/INFO 级别日志