龙虾-OpenClaw一文详细了解-手搓OpenClaw-1

龙虾-OpenClaw一文详细了解-手搓OpenClaw-1

这一系列我会用 Python 一步步手搓一个“可运行、可扩展、可解释”的 OpenClaw 简化版。
第一篇先不追求功能多，而是先搭好最重要的骨架：服务入口、会话并发模型、最小 Agent Loop。

0. 为什么要手搓 OpenClaw

OpenClaw 很强，但完整工程体量也很大。对于大多数开发者来说，直接阅读全量代码会有三个痛点：

• 模块多：Gateway、Agent、Tools、Sessions、Channels 互相耦合
• 路径长：一条消息从输入到回复，跨越多个子系统
• 调试难：没有自己的“最小版本”，很难定位问题

所以这个系列采用一个更实用的学习路径：
先做最小闭环，再逐步补齐能力。

1. 目标

用 Python 从 0 到 1 复现 OpenClaw 的核心能力：

• Agent Loop（工具调用 + 多轮推理）
• Session 与并发隔离
• 记忆系统（短期 + 长期）
• Skills 系统（分层加载）
• Web/Telegram 等渠道接入

第一篇的阶段目标是：

• 跑起 FastAPI 服务
• 打通一个最小/v1/chat对话接口
• 具备会话隔离与并发控制（每会话锁 + 全局信号量）

2. 目标架构

3. 技术选型

• 语言：Python 3.11+
• 服务框架：FastAPI（Web/Dashboard/API 一体）
• 模型接入：先用简化 Provider 抽象，后续切真实 OpenAI Compatible
• 并发模型：asyncio+ 每会话Lock+ 全局Semaphore
• 配置与数据模型：Pydantic v2+pydantic-settings
• 日志：标准logging（后续可升级 JSON 结构化日志）
• 任务调度：APScheduler（后续阶段引入）
• 数据库：第一篇先不强依赖，阶段二引入PostgreSQL + pgvector

4. 第一篇完成了什么

本篇配套工程目录：openclaw_py下载地址:https://gitee.com/wisdomfriend/openclaw_py

已经有这些关键模块：

• app/main.py：FastAPI 启动入口
• app/api/routes_chat.py：最小聊天接口
• app/session/manager.py：会话复用 + 并发控制
• app/core/agent.py：最小 Agent Loop（先支持 echo 与工具占位）
• app/core/tools.py：工具注册与调用骨架
• run.py：一键启动uvicorn
• .env与requirements.txt：本地运行基础配置

5. 如何运行（可复制）

cdopenclaw_py python-mvenv .venv .venv\Scripts\activate pipinstall-rrequirements.txt python run.py

服务启动后：

• 健康检查：http://127.0.0.1:7788/health
• OpenAPI 文档：http://127.0.0.1:7788/docs

测试聊天接口（PowerShell）：

curl-XPOST"http://127.0.0.1:7788/v1/chat"^-H"Content-Type: application/json"^-d"{\"session_id\":\"demo\",\"message\":\"hello\"}"

6. 设计细节：为什么 SessionManager 要先做

很多人一开始只关心 LLM 调用，但真实系统更容易先死在并发上。
如果同一个session_id同时进两条消息，不加锁就会出现：

• 历史消息交错写入
• 上下文错乱，回复“串台”
• 工具调用顺序被打乱

所以第一篇就把这一层打好：

• 同一会话：串行（asyncio.Lock）
• 全局会话：限流（asyncio.Semaphore）

这一步是后面做“记忆系统”和“多渠道接入”的地基。

from__future__importannotationsimportasynciofromcontextlibimportasynccontextmanagerfromtypingimportAsyncIteratorfromapp.configimportsettingsfromapp.core.agentimportAgentclassSessionManager:def__init__(self,max_concurrent:int=4)->None:self._sessions:dict[str,Agent]={}self._locks:dict[str,asyncio.Lock]={}self._semaphore=asyncio.Semaphore(max_concurrent)defget_or_create(self,session_id:str)->Agent:ifsession_idnotinself._sessions:self._sessions[session_id]=Agent()returnself._sessions[session_id]defget_lock(self,session_id:str)->asyncio.Lock:ifsession_idnotinself._locks:self._locks[session_id]=asyncio.Lock()returnself._locks[session_id]@asynccontextmanagerasyncdefacquire(self,session_id:str)->AsyncIterator[None]:lock=self.get_lock(session_id)asyncwithlock:asyncwithself._semaphore:yielddef__len__(self)->int:returnlen(self._sessions)_session_manager=SessionManager(max_concurrent=settings.max_concurrent_agents)defget_session_manager()->SessionManager:return_session_manager

7. 这篇没做什么

为了保证节奏清晰，第一篇有意不做：

• 不接真实大模型 API（先保留 Provider 接口）
• 不做复杂工具协议（先保留 Tool Runtime 骨架）
• 不做数据库与向量检索（留给阶段二）
• 不做 Telegram/Discord 真接入（留给阶段三）

8. 常见坑与排查

• 端口占用：把.env里的PORT改掉
• 全局环境依赖冲突：优先用.venv，不要直接污染系统 Python

9. 下一篇预告

下一篇我会接入真实的OpenAI-Compatible Provider，把当前的EchoProvider替换掉，内容包括：

• .env增加OPENAI_API_KEY与OPENAI_BASE_URL
• Provider 抽象如何兼容不同模型厂商
• 超时、重试、错误映射的最小实现

到那时，这个项目就会从“骨架”进化成“真能聊天的 Agent”。