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

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

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

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

第一篇的阶段目标是：

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

把当前的 EchoProvider 升级为可接真实模型的统一抽象层，支持 OpenAI-Compatible API

做什么

• 定义统一 LLMProvider 接口（输入消息，输出文本）
• 增加 OpenAICompatibleProvider
• 从 .env 读取 OPENAI_API_KEY、OPENAI_BASE_URL、LLM_MODEL
• 接入超时、重试、错误映射

不做什么

• 不做多模态（图片/音频）请求
• 不做函数调用（Tool Call 在第 3、4 篇）
• 不做多模型路由与自动降级
• 不使用langchian, 等中后期再升级到langchain

代码下载路径:openclaw_py

4.1 定义统一协议与错误类型

class LLMProvider(ABC): """Provider abstraction for text chat completion.""" @abstractmethod async def complete(self, messages: list[ChatMessage]) -> str: raise NotImplementedError

错误分层（建议保留）：

 
    
    
      
 
       
class ProviderConfigError(ProviderError): ... 
 
       
class ProviderAuthError(ProviderError): … class ProviderTimeoutError(ProviderError): … class ProviderRequestError(ProviderError): … class ProviderServerError(ProviderError): … class ProviderResponseError(ProviderError): …
 
      

 
意义：后面 API 层可以按错误类型映射成更清晰的 HTTP 响应，而不是统一 500。
 

4.2 OpenAI-Compatible 适配器

 
    
    
      
 
       
class OpenAICompatibleProvider(LLMProvider): def __init__(self, api_key: str, model_name: str, base_url: str = "https://api.openai.com/v1", timeout_seconds: float = 20.0, max_retries: int = 2): self.api_key = api_key.strip() self.model_name = model_name self.base_url = self._normalize_base_url(base_url) self.timeout_seconds = timeout_seconds self.max_retries = max(0, max_retries) @staticmethod def _normalize_base_url(base_url: str) -> str: base = base_url.rstrip("/") if not base.endswith("/v1"): base = f"{base}/v1" return base
 
      
 

_normalize_base_url() 自动补 /v1，避免你在 .env 写成 https://api.openai.com 时路径拼错。

4.3 请求、重试与响应解析

 
    
    
      
 
       
payload = { "model": self.model_name, "messages": messages, 
 
       
}
 
       
attempts = self.max_retries + 1 for attempt in range(attempts):
 
       
try: async with httpx.AsyncClient(timeout=self.timeout_seconds) as client: response = await client.post(url, headers=headers, json=payload) except httpx.TimeoutException as exc: if attempt < self.max_retries: continue raise ProviderTimeoutError("LLM request timed out") from exc
 
      
 
状态码映射：
 
    
    
      
 
       
if response.status_code == 401: raise ProviderAuthError("Provider auth failed (401)") 
 
       
if 400 <= response.status_code < 500:
 
       
raise ProviderRequestError(...) 
 
       
if response.status_code >= 500:
 
       
...
 
      
 
响应解析：
 
    
    
      
 
       
data = response.json() 
 
       
content = data["choices"][0]["message"]["content"] if not isinstance(content, str) or not content.strip():
 
       
raise ProviderResponseError("Empty content in provider response") 
 
       
return content.strip()
 
      

 
4.4 Provider 工厂：把选择逻辑从Agent拆出去
 
    
    
      
 
       
def build_provider_from_settings() -> LLMProvider: provider = settings.llm_provider.lower().strip() if provider == "openai_compatible": return OpenAICompatibleProvider( api_key=settings.openai_api_key, model_name=settings.llm_model, base_url=settings.openai_base_url, timeout_seconds=settings.llm_timeout_seconds, max_retries=settings.llm_max_retries, ) return EchoProvider()
 
      
 
这一步看起来小，但决定了后面扩展成本：
 

新增 ClaudeProvider 时，只需要"加一个类 + 工厂一个分支"，不用动 Agent 主流程。

4.5 Agent 侧只保留一行依赖

 
    
    
      
 
       
class Agent: def __init__(self, provider: LLMProvider | None = None) -> None: self.provider = provider or build_provider_from_settings()
 
      
 
这就是"依赖倒置"在这里最朴素、最有价值的落地。
 

1. 先抽象 provider 协议：complete(messages) -> text
2. 增加 OpenAI-Compatible 实现
3. 在 Agent 中注入 provider 实例
4. 增加基础单测（成功、超时、401 错误）
5. 提供一个可直接测试的 /v1/chat

• openclaw_py/app/core/llm_provider.py
• openclaw_py/app/config.py
• openclaw_py/app/core/agent.py
• openclaw_py/.env
• openclaw_py/tests/

• 配置正确时，/v1/chat 能返回真实模型结果
• Provider 层接口可替换，不影响 Agent 外部调用方式
• openclaw_py/tests/test_provider.py 提供单元测试

验证 Provider 的三件事：

1. 成功路径能否正常解析 choices[0].message.content
2. 401 是否能准确抛 ProviderAuthError
3. 超时是否按 max_retries 真的重试

当前测试用了 httpx.MockTransport，这个做法：

• 不依赖真实外网
• 不消耗 token
• 可稳定复现错误场景

例如超时重试测试里，max_retries=2，最终断言 calls["count"] == 3，能精准验证"首调 + 2 次重试"是否生效。

配置了env中的模型api后, 就可以启动项目, 然后点击"http://127.0.0.1:7788/docs"中的chat接口进行普通的对话测试

第 3 篇进入 Agent Loop，补齐：

• 多轮消息历史
• Tool call 循环
• 结束条件与最大轮数保护

如果你已经看到这里，说明你对"手搓 OpenClaw"是真的感兴趣。

这套系列会持续把代码和踩坑都开源出来，不走玄学，尽量做到每篇都能复现。

如果这篇对你有帮助，欢迎随手支持一下：

• 点个赞，让我知道这条路线是有价值的
• 点个关注，后续 3~12 篇更新不会错过
• 点个收藏，后面实操时可以随时回来看代码片段
• 有余力的话，来个打赏，我会把更多时间投入到高质量连载里

你的每一次反馈，都会直接影响这个系列更新的速度和深度。

我们下一篇见。