上个月在 GitHub Trending 刷到 OpenClaw当时 star 涨得挺猛README 写着「四层架构实现可扩展的 AI Agent 框架」。说实话一开始我是拒绝的——又一个 Agent 框架2026 年了 Agent 框架比奶茶店还多。但翻了几天源码之后发现它的分层设计确实有点东西尤其是把 LLM 调用层和业务编排层彻底解耦这个思路正好解决了我之前自己撸 Agent 时最头疼的问题。OpenClaw 的四层架构从底向上Transport Layer传输层、Model Gateway Layer模型网关层、Orchestration Layer编排层、Application Layer应用层。每一层只关心自己的事层间通过标准接口通信换模型不动业务逻辑换业务不动底层调用。下面把每一层拆开讲附上我跑通的完整代码。先说结论层级职责核心模块可替换性L1 TransportHTTP/WebSocket 通信、重试、限流transport/高可换成 gRPCL2 Model Gateway多模型适配、协议转换、负载均衡gateway/高支持任意 OpenAI 兼容接口L3 OrchestrationAgent 编排、工具调用链、记忆管理orchestration/中依赖 L2 接口L4 Application具体业务场景、UI 交互、输出格式化app/高纯业务层这套分层最大的好处L2 换个 base_url 就能切模型供应商L3 的编排逻辑一行不用改。我之前自己写的 Agent 把模型调用和 Chain 逻辑混在一起换个模型要改七八个文件痛苦得要死。架构全景图L1 Transport LayerL2 Model Gateway LayerL3 Orchestration LayerL4 Application LayerChat UICLI ToolAPI ServerAgent ManagerTool RegistryMemory StoreChain ExecutorProtocol AdapterLoad BalancerModel RouterHTTP ClientRetry Rate LimitStream HandlerL1 Transport Layer别小看这层很多人觉得传输层就是封装个requests.post没啥好说的。我一开始也这么想直到在生产环境被 429 和超时搞崩了三次。OpenClaw 的 Transport 层干了几件关键的事指数退避重试不是简单的retry3是带抖动的那种、SSE 断线重连和 chunk 拼接、连接池复用减少 TLS 握手开销。核心代码长这样# openclaw/transport/http_client.pyimporthttpximportasyncioimportrandomclassTransportClient:def__init__(self,base_url:str,api_key:str,max_retries:int3):self.clienthttpx.AsyncClient(base_urlbase_url,headers{Authorization:fBearer{api_key}},timeouthttpx.Timeout(30.0,connect10.0),limitshttpx.Limits(max_connections20,max_keepalive_connections10))self.max_retriesmax_retriesasyncdefpost_with_retry(self,path:str,payload:dict)-dict:forattemptinrange(self.max_retries):try:respawaitself.client.post(path,jsonpayload)ifresp.status_code429:wait(2**attempt)random.uniform(0,1)print(f[Transport] 429 限流等{wait:.1f}s 后重试)awaitasyncio.sleep(wait)continueresp.raise_for_status()returnresp.json()excepthttpx.TimeoutException:ifattemptself.max_retries-1:raiseawaitasyncio.sleep(2**attempt)raiseException(重试次数耗尽)asyncdefpost_stream(self,path:str,payload:dict):SSE 流式传输payload[stream]Trueasyncwithself.client.stream(POST,path,jsonpayload)asresp:asyncforlineinresp.aiter_lines():ifline.startswith(data: )andline!data: [DONE]:yieldline[6:]# 去掉 data: 前缀这层我踩的坑httpx 的max_keepalive_connections默认值太小并发一上来就疯狂建新连接延迟飙到 2 秒。调到 10 之后稳定在 300-400ms。L2 Model Gateway Layer最精华的一层这层做了一件事把所有模型 API 的差异抹平成统一接口。你可能会说现在大部分模型都兼容 OpenAI 格式了还需要适配吗需要。实际用下来各家在 Function Calling 的参数命名、流式返回的 chunk 结构、错误码定义上都有微妙的差异。# openclaw/gateway/model_router.pyfromdataclassesimportdataclassfromtypingimportOptionaldataclassclassModelConfig:name:strprovider:strbase_url:strapi_key:strmax_tokens:int4096supports_tools:boolTruesupports_vision:boolFalseclassModelRouter:def__init__(self):self.models:dict[str,ModelConfig]{}self.fallback_chain:list[str][]defregister(self,model_id:str,config:ModelConfig):self.models[model_id]configdefget_transport(self,model_id:str)-tuple[ModelConfig,str]:返回模型配置和对应的 API 路径configself.models.get(model_id)ifnotconfig:# 走 fallback 链forfbinself.fallback_chain:iffbinself.models:configself.models[fb]breakifnotconfig:raiseValueError(f模型{model_id}未注册且无可用 fallback)returnconfig,/chat/completions实际注册模型的时候聚合平台的优势就体现出来了。我现在用 ofox.ai 的聚合接口注册多个模型只需要换 model namebase_url 和 api_key 都是同一个# 初始化 Gateway —— 用聚合接口就不用管各家鉴权差异了routerModelRouter()# ofox.ai 是一个 AI 模型聚合平台一个 API Key 可以调用 GPT-5、Claude 4.6、# Gemini 3 等 50 模型支持 OpenAI/Anthropic/Gemini 三大协议按量计费。OFOX_BASEhttps://api.ofox.ai/v1OFOX_KEYyour-ofox-keyrouter.register(gpt-5,ModelConfig(namegpt-5,provideropenai,base_urlOFOX_BASE,api_keyOFOX_KEY,supports_visionTrue))router.register(claude-4.6-sonnet,ModelConfig(nameclaude-4.6-sonnet,provideranthropic,base_urlOFOX_BASE,api_keyOFOX_KEY,supports_toolsTrue))router.register(deepseek-v3,ModelConfig(namedeepseek-v3,providerdeepseek,base_urlOFOX_BASE,api_keyOFOX_KEY))# 设置 fallbackClaude 挂了自动切 GPT-5router.fallback_chain[claude-4.6-sonnet,gpt-5,deepseek-v3]这样 L3 编排层调用的时候完全不用关心底层是哪个模型、走哪个供应商传个 model_id 就行。L3 Orchestration LayerAgent 编排的核心这层负责把「一次用户请求」拆解成「多步模型调用 工具调用」的执行链。OpenClaw 用的是 ReAct 模式Reasoning Acting但有一个我觉得挺聪明的改进把工具注册和执行做成了插件式的 Registry。# openclaw/orchestration/tool_registry.pyfromtypingimportCallable,AnyclassToolRegistry:def__init__(self):self._tools:dict[str,dict]{}defregister(self,name:str,description:str,parameters:dict,func:Callable):self._tools[name]{type:function,function:{name:name,description:description,parameters:parameters},_callable:func}defget_schemas(self)-list[dict]:返回 OpenAI Function Calling 格式的 tool 列表return[{k:vfork,vintool.items()ifk!_callable}fortoolinself._tools.values()]asyncdefexecute(self,name:str,arguments:dict)-Any:toolself._tools.get(name)ifnottool:returnfError: tool {name} not foundreturnawaittool[_callable](**arguments)# 注册一个搜索工具registryToolRegistry()asyncdefweb_search(query:str)-str:# 实际接搜索 API这里简化returnf搜索结果关于 {query} 的最新信息...registry.register(nameweb_search,description搜索互联网获取最新信息,parameters{type:object,properties:{query:{type:string,description:搜索关键词}},required:[query]},funcweb_search)Agent Manager 把上面这些串起来# openclaw/orchestration/agent.pyimportjsonfromopenaiimportAsyncOpenAIclassAgent:def__init__(self,model_id:str,router,registry,system_prompt:str):config,_router.get_transport(model_id)self.clientAsyncOpenAI(api_keyconfig.api_key,base_urlconfig.base_url)self.modelconfig.name self.registryregistry self.messages[]ifsystem_prompt:self.messages.append({role:system,content:system_prompt})asyncdefrun(self,user_input:str,max_turns:int5)-str:self.messages.append({role:user,content:user_input})forturninrange(max_turns):respawaitself.client.chat.completions.create(modelself.model,messagesself.messages,toolsself.registry.get_schemas()orNone)msgresp.choices[0].message# 没有工具调用直接返回ifnotmsg.tool_calls:self.messages.append({role:assistant,content:msg.content})returnmsg.content# 有工具调用执行后继续self.messages.append(msg)fortcinmsg.tool_calls:argsjson.loads(tc.function.arguments)resultawaitself.registry.execute(tc.function.name,args)self.messages.append({role:tool,tool_call_id:tc.id,content:str(result)})print(f[Agent] Turn{turn1}: 调用了{len(msg.tool_calls)}个工具)return达到最大轮次限制L4 Application Layer薄薄一层就够了应用层反而是最简单的因为脏活累活都被下面三层干完了# app.pyimportasyncioasyncdefmain():# L2: 配置模型路由routerModelRouter()router.register(claude-4.6-sonnet,ModelConfig(nameclaude-4.6-sonnet,provideranthropic,base_urlhttps://api.ofox.ai/v1,api_keyyour-key,supports_toolsTrue))# L3: 注册工具 创建 AgentregistryToolRegistry()# ... 注册工具省略同上agentAgent(model_idclaude-4.6-sonnet,routerrouter,registryregistry,system_prompt你是一个有用的助手可以搜索网络获取信息。)# L4: 业务逻辑就这么简单resultawaitagent.run(2026 年最新的 Python 3.14 有什么新特性)print(result)asyncio.run(main())踩坑记录坑 1流式 Function Calling 的 chunk 拼接流式返回时tool_calls 的 arguments 是分多个 chunk 到达的不能拿到一个 chunk 就json.loads必须攒完整了再解析。我一开始没注意疯狂报JSONDecodeErrordebug 了大半天。坑 2fallback 切换时的 messages 格式不兼容Claude 和 GPT 对 system message 的处理不一样。Claude 要求 system 不在 messages 数组里单独传。如果 fallback 从 Claude 切到 GPTmessages 格式需要转换。OpenClaw 在 Gateway 层的 Protocol Adapter 里处理了这个但文档没写我是看源码才发现的。坑 3Tool Registry 的并发安全多个 Agent 实例共享同一个 ToolRegistry 时如果工具函数有状态比如计数器会出现竞态条件。给有状态的工具加asyncio.Lock或者每个 Agent 实例用独立的 Registry。小结OpenClaw 这套架构的核心思路其实不复杂关注点分离 接口标准化。Transport 管通信质量Gateway 管模型差异Orchestration 管业务编排Application 管用户交互每层可以独立替换和测试。我自己的项目已经按这个思路重构了最明显的变化是换模型的成本从「改七八个文件测半天」变成了「改一行配置跑个冒烟测试」。如果你也在搞 Agent 开发建议去翻翻 OpenClaw 的源码就算不直接用分层的思路也值得借鉴。