OpenClaw智能体七文件架构：面向工业级落地的模块化设计

1. 项目概述：OpenClaw 不是玩具，是智能体开发的“乐高工厂”

如果你刚在 GitHub 上点开 OpenClaw 的仓库，看到满屏的.py、.yaml和Dockerfile，第一反应可能是——这哪是智能体框架，分明是一份加密电报。别急，我去年用它从零搭起一个能自动处理客户邮件+生成周报+同步钉钉日程的轻量级办公助手，前后踩了二十多个坑，才真正摸清它不是“另一个大模型调用封装”，而是一套面向真实工作流的智能体组装系统。OpenClaw 的核心价值，从来不在炫技式的单点能力，而在它把智能体拆解成可插拔、可验证、可复用的七个标准模块——就像汽车厂不卖整车，而是把发动机、变速箱、底盘、ECU 控制单元分开交付，你按需组合，还能自己换零件。这七个文件，就是它的“标准件清单”。它们分别是：config.yaml（全局调度中枢）、agent.py（智能体行为骨架）、tool_registry.py（工具货架目录）、memory.py（短期记忆缓存）、orchestrator.py（多步任务编排器）、llm_client.py（大模型通信协议栈）、docker-compose.yml（生产环境部署契约）。新手常犯的致命错误，是把agent.py当成主程序去改，结果改完发现记忆不持久、工具调用失败、多轮对话乱序——问题根本不在 agent，而在memory.py的缓存策略没对齐orchestrator.py的状态机设计，或者llm_client.py的重试逻辑和tool_registry.py的超时阈值打架。这篇文章不讲“怎么跑通 demo”，只带你一寸一寸拆开这七块核心拼图，看清每颗螺丝的咬合方向、每根线缆的信号流向、每个接口的容错边界。适合三类人：刚学完 LangChain 想进阶的开发者、需要快速落地垂直场景智能体的产品经理、以及被“智能体=大模型+提示词”这种简化说法耽误已久的业务方。你不需要会写大模型训练代码，但得明白为什么config.yaml里一个max_retries: 3的配置，会决定整个服务在高并发下的熔断表现。

2. 核心设计逻辑：为什么是这七个文件？而不是五个或九个？

2.1 拆解本质：OpenClaw 解决的是“智能体工业化落地”的工程瓶颈

很多人误以为 OpenClaw 是为了“让写智能体更简单”，其实恰恰相反——它是为了让智能体更难出错、更容易审计、更便于交接。我们来算一笔账：一个典型企业级智能体要满足四个硬性指标——响应延迟 ≤ 1.2 秒（用户无感等待）、工具调用成功率 ≥ 99.3%（金融/客服场景底线）、上下文记忆准确率 ≥ 98.7%（避免张冠李戴）、故障平均恢复时间 ≤ 47 秒（SRE 要求）。LangChain 这类胶水框架，在单机 demo 阶段很顺滑，但一旦接入真实 API（比如飞书审批流、用友 NC 接口、内部风控引擎），就会暴露三个结构性缺陷：第一，状态散落——记忆存在 Redis，工具调用日志打在本地文件，LLM 请求链路埋点在中间件，故障时根本无法串联还原；第二，耦合过重——改一个工具的参数，要动 agent 逻辑、memory 缓存策略、甚至 llm_client 的重试机制；第三，契约模糊——没人定义“工具调用失败”到底指网络超时、API 返回 401 还是 JSON 解析异常，导致下游编排器无法做差异化重试。OpenClaw 的七文件架构，就是针对这三点的手术刀式解法。它把智能体生命周期切分为七个正交关注点：配置（谁来管）、行为（做什么）、工具（用什么）、记忆（记什么）、编排（怎么做）、通信（怎么问）、部署（在哪跑）。每个文件只解决一个问题，且通过明确定义的输入/输出契约交互。比如tool_registry.py只负责两件事：一是注册工具函数（带类型注解和描述），二是返回一个标准化的ToolCallResult对象（含success: bool,data: dict,error: str,latency_ms: int四个字段）。orchestrator.py拿到这个对象后，不用关心工具内部怎么实现，只根据success字段决定走成功分支还是失败分支，根据latency_ms动态调整后续步骤的超时阈值。这种设计，让新人接手时，可以先只看config.yaml理清数据流向，再聚焦agent.py理解业务逻辑，最后深入llm_client.py优化性能——而不是面对一个 2000 行的main.py文件发呆。

2.2 为什么不是五个？——“记忆”与“编排”必须分离

有人会问：memory.py和orchestrator.py能不能合并？毕竟都是管“状态”的。实测下来绝对不行。去年我们给某银行做信贷初筛助手时，就吃过这个亏。最初把对话历史缓存和多步决策逻辑全塞进一个state_manager.py，结果出现一个诡异现象：当用户连续问“上个月放款总额？”、“其中小微企业占比？”、“TOP3 区域是哪些？”时，第三问的答案总是错的。排查三天才发现，state_manager.py在处理第二问时，把“小微企业占比”的计算结果缓存进了内存，但第三问的 prompt 模板里引用了“上个月放款总额”这个变量，而缓存层没做变量依赖追踪，导致第三问拿到的仍是第一问的原始数据。OpenClaw 强制分离的深意就在这里：memory.py是纯数据容器，只提供get(key),set(key, value, ttl),clear()三个原子操作，所有数据结构（如对话树、知识图谱快照）都由上层决定；orchestrator.py则是状态机驱动器，它维护一个有限状态机（FSM），每个状态对应一个明确的业务意图（如WAITING_FOR_AMOUNT,VALIDATING_REGION,GENERATING_REPORT），状态迁移由tool_call_result.success和 LLM 的intent_classification输出共同触发。二者通过memory_key字符串解耦——orchestrator告诉memory：“把当前用户 ID 的current_intent设为GENERATING_REPORT”，memory不关心这个 intent 是什么含义，只忠实执行。这种分离让调试变得极其清晰：如果状态跳错了，看orchestrator的 FSM 定义；如果数据读取异常，直接查memory的get日志。我们后来加了个小技巧：在memory.py的set方法里，强制要求传入source: str参数（如"orchestrator","llm_client"），日志里就能一眼看出是谁污染了缓存。

2.3 为什么不是九个？——“LLM 通信”与“工具注册”已足够抽象

也有团队想把llm_client.py拆成openai_adapter.py、qwen_adapter.py、local_vllm_client.py三个文件。OpenClaw 的作者在设计文档里明确反对这种过早抽象。理由很实在：95% 的企业项目，上线时只会用一种 LLM 后端（要么是公司统一采购的 Qwen，要么是私有化部署的 DeepSeek），切换模型是季度级决策，不是每次迭代都要做的事。强行拆分会带来两个成本：第一，orchestrator.py要多维护一套 adapter 工厂模式，增加 200+ 行胶水代码；第二，config.yaml里llm_provider: qwen这种配置，会变成llm_adapter: qwen_v1、llm_endpoint: http://qwen-api:8000/v1/chat/completions、llm_api_key: ${QWEN_KEY}三行，配置复杂度指数上升。OpenClaw 的方案是：llm_client.py内部用策略模式，但对外只暴露一个call_llm(prompt: str, tools: List[ToolSpec]) -> LLMResponse接口。具体实现藏在llm_client._get_provider_client()私有方法里，根据config.yaml的llm.provider字段动态加载。这样既保留了扩展性（真要切模型，改一行配置+加一个 provider 类），又避免了日常开发的认知负担。同理，tool_registry.py也没拆成http_tool.py、db_tool.py、file_tool.py，因为所有工具最终都归一为Callable[[dict], ToolCallResult]，差异只在初始化参数（URL、DB connection string、file path），这些全由config.yaml的toolssection 管理。我们实测过，一个 5 人团队用这套架构，三个月内迭代了 17 个新工具（从查 CRM 到调用内部风控 API），没人动过tool_registry.py的核心逻辑，只在 config 里加 YAML 片段。

3. 七大核心文件逐行精解：不只是“是什么”，更是“为什么这么写”

3.1config.yaml：不是配置文件，是智能体的“宪法性文档”

OpenClaw 的config.yaml看似平平无奇，实则是整个系统的“宪法”。它不定义功能，而定义契约边界。我们来看一个生产环境的真实片段：

# config.yaml global: timeout_ms: 8000 max_retries: 3 fallback_strategy: "return_error" llm: provider: "qwen" endpoint: "https://qwen-api.internal/v1/chat/completions" api_key: "${QWEN_API_KEY}" # 环境变量注入，非明文 model: "qwen2-72b-instruct" temperature: 0.3 max_tokens: 2048 tools: - name: "fetch_customer_info" type: "http" url: "https://crm.internal/api/v2/customers/{customer_id}" method: "GET" timeout_ms: 3000 retry_on: ["5xx", "timeout"] - name: "send_approval_request" type: "http" url: "https://feishu.internal/open-apis/approval/v1/instances" method: "POST" timeout_ms: 5000 retry_on: ["429", "timeout"] memory: backend: "redis" redis_url: "redis://redis.internal:6379/1" default_ttl_sec: 3600 orchestrator: state_machine: initial_state: "WAITING_FOR_CUSTOMER_ID" states: WAITING_FOR_CUSTOMER_ID: on_enter: ["prompt_for_customer_id"] transitions: - event: "customer_id_provided" target: "FETCHING_CUSTOMER_INFO" FETCHING_CUSTOMER_INFO: on_enter: ["call_tool:fetch_customer_info"] transitions: - event: "tool_success" target: "ASKING_APPROVAL" - event: "tool_failure" target: "HANDLE_CUSTOMER_NOT_FOUND"

关键点解析：

• global.timeout_ms是系统级熔断阀：不是某个工具的超时，而是整个智能体单次请求的总耗时上限。orchestrator.py在启动时会创建一个asyncio.wait_for的顶层包裹，一旦超过 8 秒，直接抛出TimeoutError，跳过所有后续步骤，走fallback_strategy。这比在每个工具里设超时更可靠，因为能防住 LLM 本身卡死（比如 prompt 太长导致 qwen 服务端 hang 住）。

• tools[].retry_on的设计智慧：它不写retry_on: [429, 500, 503]，而用字符串枚举["429", "5xx", "timeout"]。为什么？因为 HTTP 状态码的语义是分层的：429（限流）必须重试，但要加退避（exponential backoff）；5xx（服务端错误）可以重试，但可能无效；timeout必须重试，且下次要缩短超时阈值。llm_client.py里的重试逻辑会根据这个字符串，动态选择重试策略。我们曾在线上遇到飞书审批接口偶发 502，配置retry_on: ["5xx"]后，失败率从 0.7% 降到 0.03%。

• orchestrator.state_machine是业务逻辑的可视化表达：这里没有一行 Python 代码，却完整定义了用户旅程。on_enter指定进入状态时触发的动作（prompt_for_customer_id是一个预定义的 prompt 模板名），transitions定义状态迁移条件。orchestrator.py的核心循环就是：读当前状态 → 执行on_enter动作 → 等待事件（LLM 输出 or 工具返回）→ 查transitions表 → 跳转新状态。这种 DSL（领域特定语言）写法，让产品经理都能看懂流程图，也方便用graphviz自动生成状态机图谱用于审计。

提示：config.yaml里所有${VAR_NAME}都必须在运行时存在，否则服务启动失败。我们用了一个小技巧：在 Docker 启动脚本里加set -o nounset，确保缺失环境变量时容器立刻退出，而不是静默降级——这对生产环境至关重要。

3.2agent.py：行为骨架，而非业务大脑

agent.py是最常被误解的文件。新手总想在这里写业务逻辑，比如“如果用户问还款，就调用还款工具”。这是灾难的开始。agent.py的唯一职责，是接收输入、触发 orchestrator、返回输出。它像一个快递分拣员：用户消息进来，它不拆包，只看面单（intent），然后把包裹（message object）扔进对应的传送带（orchestrator 的事件队列）。以下是精简后的核心逻辑：

# agent.py class OpenClawAgent: def __init__(self, config: Config): self.config = config self.orchestrator = Orchestrator(config.orchestrator) self.memory = MemoryBackend(config.memory) async def handle_message(self, user_id: str, message: str) -> AgentResponse: # 1. 从 memory 读取用户当前状态（state key: f"{user_id}:state"） current_state = await self.memory.get(f"{user_id}:state", default="INITIAL") # 2. 将用户消息 + 当前状态，喂给 orchestrator # orchestrator 返回的是一个包含下一步动作的指令包 orchestration_plan = await self.orchestrator.plan( user_id=user_id, current_state=current_state, user_input=message ) # 3. 执行 orchestrator 规划的动作（可能是调工具、发 prompt、跳转状态） result = await self._execute_plan(orchestration_plan) # 4. 更新 memory 中的状态和对话历史 await self.memory.set(f"{user_id}:state", orchestration_plan.next_state) await self.memory.append(f"{user_id}:history", { "role": "user", "content": message, "timestamp": time.time() }) return AgentResponse( content=result.content, status=result.status, next_state=orchestration_plan.next_state )

关键洞察：

• agent.py绝不解析用户意图。意图识别（intent classification）是orchestrator.py的工作。agent.py只负责把原始消息原样传过去。这样做的好处是：当你要把意图识别从规则引擎换成微调的小模型时，只需改orchestrator.py，agent.py一行不动。

• agent.py不持有任何业务数据。用户 ID、对话历史、当前状态，全部通过self.memory接口读写。这意味着你可以轻松把memory后端从 Redis 换成 PostgreSQL（比如要支持 SQL 查询历史），只要MemoryBackend接口不变，agent.py完全无感。

• agent.py的handle_message方法是唯一的入口。所有外部调用（HTTP API、WebSocket、钉钉机器人回调）都必须经过它。我们在线上加了 Prometheus metrics：openclaw_agent_handle_duration_seconds{status="success"}，监控每个请求的耗时分布，快速定位是 LLM 卡顿还是工具慢。

3.3tool_registry.py：工具不是函数，是带 SLA 的服务契约

tool_registry.py是 OpenClaw 的“工具应用商店”。它不关心工具怎么实现，只关心工具承诺了什么。核心是ToolSpec数据类和register_tool装饰器：

# tool_registry.py from dataclasses import dataclass from typing import Callable, Dict, Any @dataclass class ToolSpec: name: str description: str # 给 LLM 看的自然语言描述 parameters: Dict[str, str] # {param_name: param_description} required: list[str] timeout_ms: int retry_on: list[str] class ToolRegistry: _tools: Dict[str, Callable] = {} _specs: Dict[str, ToolSpec] = {} @classmethod def register_tool(cls, spec: ToolSpec): def decorator(func: Callable) -> Callable: cls._tools[spec.name] = func cls._specs[spec.name] = spec return func return decorator @classmethod def get_tool(cls, name: str) -> Callable: return cls._tools.get(name) @classmethod def get_spec(cls, name: str) -> ToolSpec: return cls._specs.get(name)

使用示例（tools/fetch_customer_info.py）：

# tools/fetch_customer_info.py from tool_registry import ToolRegistry, ToolSpec @ToolRegistry.register_tool( ToolSpec( name="fetch_customer_info", description="根据客户ID查询客户基本信息、信用等级、历史订单数", parameters={ "customer_id": "客户唯一标识，如 CUST-2024-001" }, required=["customer_id"], timeout_ms=3000, retry_on=["5xx", "timeout"] ) ) async def fetch_customer_info(customer_id: str) -> dict: # 真实实现：调用 CRM API async with aiohttp.ClientSession() as session: async with session.get( f"https://crm.internal/api/v2/customers/{customer_id}", timeout=aiohttp.ClientTimeout(total=3.0) ) as resp: if resp.status == 200: return await resp.json() else: raise Exception(f"CRM API error: {resp.status}")

关键设计点：

• description和parameters是给 LLM 的“说明书”：orchestrator.py在调用 LLM 前，会把所有已注册工具的description和parameters拼成一段 system prompt，让 LLM 知道“我能用哪些工具，每个工具要什么参数”。这比硬编码 prompt 更灵活——加一个新工具，LLM 自动学会用它。

• timeout_ms和retry_on是工具的 SLA 声明：llm_client.py在调用工具时，会严格遵守这个超时，并根据retry_on列表决定是否重试。我们曾把某个内部风控工具的timeout_ms从 5000 改成 2000，结果整体 P95 延迟下降了 35%，因为避免了长时间等待。

• 工具函数签名必须是async def：OpenClaw 默认异步执行，强制要求所有工具可 await。这杜绝了同步阻塞操作拖垮整个事件循环。我们有个血泪教训：早期一个工具用了requests.get（同步），导致整个服务在高并发下连接池耗尽，改成aiohttp后稳定如山。

3.4memory.py：记忆不是“记住”，而是“精准召回”

memory.py的核心矛盾是：既要低延迟（毫秒级读写），又要强一致性（不能读到脏数据），还要支持复杂查询（比如“查用户 A 最近三次的还款请求”）。OpenClaw 的解法是分层存储 + 显式键约定：

# memory.py import redis.asyncio as redis import json import time from typing import Optional, Dict, Any, List class MemoryBackend: def __init__(self, config: MemoryConfig): self.redis = redis.from_url(config.redis_url) self.default_ttl = config.default_ttl_sec async def get(self, key: str, default=None) -> Optional[Any]: """通用 get，支持任意类型序列化""" data = await self.redis.get(key) if data is None: return default try: return json.loads(data) except json.JSONDecodeError: return data # 原始字节，如二进制文件 async def set(self, key: str, value: Any, ttl: Optional[int] = None): """通用 set，自动序列化""" if isinstance(value, (dict, list, str, int, float, bool)): data = json.dumps(value, ensure_ascii=False) else: data = str(value) # 兜底 await self.redis.setex(key, ttl or self.default_ttl, data) async def append(self, key: str, item: Dict): """向列表末尾追加，用于对话历史""" # 使用 Redis List，LPUSH + LTRIM 保证长度 await self.redis.lpush(key, json.dumps(item, ensure_ascii=False)) await self.redis.ltrim(key, 0, 99) # 只保留最近 100 条 async def search_history(self, user_id: str, limit: int = 10) -> List[Dict]: """按用户 ID 搜索历史（约定 key 格式）""" key = f"{user_id}:history" items = await self.redis.lrange(key, 0, limit-1) return [json.loads(i) for i in items if i]

键命名规范（约定大于配置）：

• f"{user_id}:state"—— 当前状态机状态（字符串）
• f"{user_id}:history"—— 对话历史列表（Redis List）
• f"{user_id}:session:{session_id}:context"—— 单次会话上下文（JSON）
• f"tool_cache:{tool_name}:{hash(params)}"—— 工具调用结果缓存（带 TTL）

这种设计带来的实操优势：

• 调试极简：线上出问题？直接redis-cli连上去GET user_123:state，一秒知道用户卡在哪。
• 冷启动友好：新用户第一次访问，get("user_123:state")返回None，agent.py自动触发INITIAL状态，无需额外判断。
• 缓存穿透防护：search_history方法里，如果lrange返回空列表，agent.py会触发默认欢迎语，而不是报错。

注意：memory.py里所有await操作都加了try/except包裹，捕获ConnectionError和TimeoutError，并返回default值。我们宁可返回过期数据，也不让一次 Redis 故障拖垮整个智能体。

3.5orchestrator.py：状态机才是真正的“智能”所在

如果说agent.py是快递员，orchestrator.py就是物流调度中心。它的核心是StateMachine类，基于config.yaml的orchestrator.state_machine配置动态生成：

# orchestrator.py from typing import Dict, List, Optional, Any from dataclasses import dataclass @dataclass class StateTransition: event: str target: str condition: Optional[str] = None # 如 "result.success == True" @dataclass class StateDefinition: on_enter: List[str] # ["prompt:welcome", "call_tool:check_user_status"] transitions: List[StateTransition] class Orchestrator: def __init__(self, config: OrchestratorConfig): self.config = config self.state_machine = self._build_state_machine(config.state_machine) def _build_state_machine(self, config_sm: dict) -> Dict[str, StateDefinition]: """将 YAML 配置转为内存中的状态机映射""" sm = {} for state_name, state_def in config_sm["states"].items(): sm[state_name] = StateDefinition( on_enter=state_def.get("on_enter", []), transitions=[ StateTransition(**t) for t in state_def.get("transitions", []) ] ) return sm async def plan(self, user_id: str, current_state: str, user_input: str) -> Plan: """根据当前状态和用户输入，规划下一步""" # 1. 获取当前状态定义 state_def = self.state_machine.get(current_state) if not state_def: return Plan(next_state="ERROR", actions=[]) # 2. 执行 on_enter 动作（可能是发 prompt 或调工具） actions = [] for action in state_def.on_enter: if action.startswith("prompt:"): # 加载 prompt 模板，注入上下文 template = load_prompt(action.split(":")[1]) prompt = template.format(user_input=user_input, history=await self._get_history(user_id)) actions.append(PromptAction(prompt=prompt)) elif action.startswith("call_tool:"): tool_name = action.split(":")[1] tool_spec = ToolRegistry.get_spec(tool_name) # 构造工具调用参数（从 user_input 或 memory 中提取） params = await self._extract_params(tool_spec, user_id, user_input) actions.append(ToolCallAction(tool_name=tool_name, params=params)) # 3. 返回规划（不执行，只规划） return Plan(next_state=current_state, actions=actions)

关键机制：

• on_enter是声明式编程：你不用写if current_state == "WAITING_FOR_CUSTOMER_ID": send_welcome_prompt()，而是配置"on_enter": ["prompt:welcome"]。orchestrator自动加载prompts/welcome.j2模板，渲染后作为下一步 prompt。模板里可以写{% if user_has_profile %}欢迎回来{% else %}请先登录{% endif %}，变量从memory读取。

• 状态迁移是事件驱动：orchestrator.plan()只返回“计划”，不执行。执行由agent.py的_execute_plan()完成。执行结果（如工具返回{"success": true, "data": {...}}）会作为event（如"tool_success"）再次喂给orchestrator.plan()，触发状态迁移。这种“计划-执行-反馈”循环，让状态机完全可控。

• Plan对象是契约载体：它包含next_state（下一个状态）和actions（要执行的动作列表）。agent.py拿到Plan后，顺序执行actions，每个动作的返回值（如 LLM 的 response、工具的 result）都会被收集，作为下一轮plan的输入。我们加了个 debug 模式：在Plan里记录debug_info: {"reason": "user said 'I want to apply', so enter APPLYING state"}，方便回溯决策逻辑。

3.6llm_client.py：通信协议栈，不是“调 API”那么简单

llm_client.py是 OpenClaw 的“网络协议栈”。它不只发请求，还负责重试、熔断、降级、可观测性。核心是LLMClient类：

# llm_client.py import asyncio import aiohttp import time from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type from typing import Dict, Any, List, Optional class LLMClient: def __init__(self, config: LLMConfig): self.config = config self.session = aiohttp.ClientSession( timeout=aiohttp.ClientTimeout(total=config.timeout_ms / 1000) ) # 熔断器：连续 3 次失败，熔断 60 秒 self.circuit_breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=60) @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10), retry=retry_if_exception_type((aiohttp.ClientError, asyncio.TimeoutError)) ) async def call_llm(self, prompt: str, tools: List[ToolSpec]) -> LLMResponse: if not self.circuit_breaker.is_closed: return LLMResponse(content="服务暂时不可用，请稍后再试", status="circuit_open") try: start_time = time.time() # 构造 OpenAI 兼容格式的请求体 payload = { "model": self.config.model, "messages": [{"role": "user", "content": prompt}], "tools": [self._tool_spec_to_openai(t) for t in tools], "temperature": self.config.temperature, "max_tokens": self.config.max_tokens } async with self.session.post( self.config.endpoint, headers={"Authorization": f"Bearer {self.config.api_key}"}, json=payload ) as resp: if resp.status == 200: data = await resp.json() latency = (time.time() - start_time) * 1000 return LLMResponse( content=self._parse_response(data), status="success", latency_ms=latency, raw_data=data ) else: raise LLMError(f"API returned {resp.status}") except Exception as e: self.circuit_breaker.record_failure() raise e def _tool_spec_to_openai(self, spec: ToolSpec) -> Dict: # 将 OpenClaw 的 ToolSpec 转为 OpenAI 的 tools 格式 return { "type": "function", "function": { "name": spec.name, "description": spec.description, "parameters": { "type": "object", "properties": {k: {"type": "string"} for k in spec.parameters}, "required": spec.required } } }

关键保障：

• tenacity重试 +CircuitBreaker熔断双保险：重试解决瞬时故障（网络抖动），熔断防止雪崩（上游服务持续不可用）。我们线上配置failure_threshold=3，意味着连续三次调用失败后，接下来 60 秒内所有请求直接返回降级文案，不发网络请求。

• latency_ms是性能黄金指标：每个LLMResponse都带耗时，agent.py会把它存入memory的f"{user_id}:latency_log"，用于分析 P95 延迟。我们发现 Qwen 的qwen2-72b模型在max_tokens=2048时，P95 延迟高达 12 秒，果断切成qwen2-14b，延迟降到 2.3 秒，业务方完全无感。

• _tool_spec_to_openai是协议转换器：它把 OpenClaw 的通用ToolSpec，转成目标 LLM（Qwen/OpenAI）能理解的格式。如果未来要支持 Ollama，只需改这个方法，call_llm接口完全不变。

3.7docker-compose.yml：部署契约，定义“生产环境长什么样”

docker-compose.yml是 OpenClaw 的“部署宪法”。它不只定义容器，更定义服务间契约、资源边界、安全基线：

# docker-compose.yml version: '3.8' services: openclaw-app: image: openclaw:latest environment: - CONFIG_PATH=/app/config.yaml - QWEN_API_KEY=${QWEN_API_KEY} volumes: - ./config.yaml:/app/config.yaml:ro - ./logs:/app/logs depends_on: - redis - nginx deploy: resources: limits: memory: 2G cpus: '1.0' reservations: memory: 1G restart_policy: condition: on-failure delay: 10s max_attempts: 3 redis: image: redis:7-alpine command: redis-server --save 60 1 --loglevel warning volumes: - redis-data:/data healthcheck: test: ["CMD", "redis-cli", "ping"] interval: 30s timeout: 10s retries: 3 nginx: image: nginx:alpine ports: - "8000:80" volumes: - ./nginx.conf:/etc/nginx/nginx.conf:ro volumes: redis-data:

关键实践：

• depends_on不是启动顺序，而是健康检查依赖：OpenClaw 应用启动时，会主动pingredis容器的健康检查端点（redis-cli ping），直到返回PONG才开始初始化memory.py。这比depends_on的service_healthy更可靠，因为后者只检查容器进程是否起来，不检查服务是否 ready。

• deploy.resources是 SLO 保障：limits.memory: 2G防止内存泄漏拖垮宿主机；reservations.memory: 1G确保容器总有 1G 内存可用，避免因宿主机内存紧张被 OOM kill。我们线上曾因没设reservations，导致高峰期容器被频繁 kill，restart_policy又没配delay，引发雪崩重启。

• volumes的:ro（只读）是安全基线：config.yaml和nginx.conf都挂载为只读，杜绝运行时被恶意篡改。logs目录可写，但我们在 `