用户消息（Telegram） → Channel Plugin 解析 → Gateway 路由（session_key） → Lane Queue 排队 → Agent Runtime 执行（Pi-Embedded / CLI Runner） → Model Call → Tool Dispatch → Response

用户消息 → System Prompt 组装 → Agent Loop 开始 ├── Phase 1: Memory Recall（Hindsight 语义召回） ├── Phase 2: Skill Load（匹配 skill → 注入提示词） ├── Phase 3: Model Call（API 请求） ├── Phase 4: Tool Dispatch（解析 tool_calls → 执行 → 结果注入） ├── Phase 5: Hindsight Retain（自动记忆写入） └── Phase 6: 循环或终止

// src/channels/plugins/types.plugin.tsinterface ChannelPlugin { id: string; start: (gateway: Gateway) => Promise<void>; stop: () => Promise<void>; onMessage: (msg: ChannelMessage) => Promise<void>;}

// src/routing/session-key.ts// session_key 格式示例：// agent:main:telegram:chat:12345 ← Telegram 私聊// agent:main:slack:channel:C789 ← Slack 频道

Lane Queue (先进先出) ┌─────────┐ │ msg_001 │ → Agent Runtime 处理中 ├─────────┤ │ msg_002 │ → 等待 ├─────────┤ │ msg_003 │ → 等待 └─────────┘

// 简化的执行流程（src/agents/）async function runAgentLoop(session) { // 1. 加载会话上下文（历史消息 + 记忆） const context = await loadContext(session); // 2. 组装 System Prompt（工具列表 + 技能 + 安全策略） const systemPrompt = await buildSystemPrompt(session); // 3. 模型调用 const response = await modelCall(systemPrompt, context); // 4. 工具分发 if (response.tool_calls) { for (const toolCall of response.tool_calls) { const result = await executeTool(toolCall); context.push({ role: ”tool”, content: result }); } // 回到步骤 3，继续循环 return runAgentLoop(session); } // 5. 持久化会话 await saveSession(session); return response;}

# 简化自 hermes/agent/loop.pyasync def run_conversation(user_message, session_id):# Phase 1: 记忆召回 memories = await hindsight_recall(user_message)# Phase 2: Skill 匹配 skills = match_skills(user_message)# Phase 3: System Prompt 组装（三层：Core + Hindsight + Skill） system_prompt = assemble_prompt(memories, skills)# Phase 4: Agent Loop for turn in range(MAX_TURNS):# 4a: 模型调用 response = await model_call(system_prompt, messages)# 4b: 工具分发与执行 if response.tool_calls: results = await execute_tools(response.tool_calls) messages.extend(results)# 4c: 检查是否应该终止 if response.finish_reason == ”stop”: break# Phase 5: Hindsight 自动写入 await hindsight_retain(session_context) return messages

src/agents/tools/├── browser-tool.ts ← Playwright 浏览器自动化 + CDP 控制├── file-tools.ts ← 文件读写（read/write/search）├── terminal-tools.ts ← Shell 命令执行（foreground/background/pty）├── canvas-tools.ts ← Canvas 渲染（HTML → 图片）├── cron-tools.ts ← 定时任务调度├── session-tools.ts ← sessions_spawn / sessions_send（多 Agent 核心）├── mcp-tools.ts ← MCP 协议客户端工具├── memory-tools.ts ← 记忆写入/读取└── ...

interface ToolDefinition { name: string; description: string; parameters: JSONSchema; execute: (params: any, context: ToolContext) => Promise<ToolResult>; policy?: ToolPolicy;}

// src/agents/sandbox/tool-policy.tsinterface ToolPolicy { requireApproval?: boolean; // 执行前弹出确认对话框 sandboxOnly?: boolean; // 仅允许在 Docker 沙箱内执行 allowArgs?: string[]; // 参数白名单（如 terminal 只允许 ”ls”, ”cat”） denyArgs?: string[]; // 参数黑名单（如禁止 ”rm -rf /”） maxTimeout?: number; // 最大执行时间}

Model returns tool_calls → for each tool_call: → policy check（是否需确认？是否限沙箱？） → execute(params) → result → push to context → back to model（带工具结果）

# hermes/tools/registry.py@register_tool( name=”terminal”, description=”Execute shell commands on Linux. ” ”Foreground for short commands, ” ”background=true for long-running tasks.”, parameters={ ”command”: {”type”: ”string”, ”description”: ”The command to execute”}, ”background”: {”type”: ”boolean”, ”default”: False}, ”timeout”: {”type”: ”integer”, ”default”: 180} })async def terminal(command: str, background: bool = False, timeout: int = 180):# 终端后端选择逻辑 if background: return await run_background(command, timeout) return await run_foreground(command, timeout)

~/.openclaw/├── sessions/│ └── agent_main_telegram_12345.jsonl ← 每行一个 JSON 事件├── memory/│ ├── 2026-06-24.md ← 今日记忆（Markdown 格式）│ └── 2026-06-23.md ← 昨日记忆└── SOUL.md ← Agent 人格定义

// sessions/xxx.jsonl 的单行示例{”timestamp”:”2026-06-24T10:30:00Z”,”role”:”user”,”content”:”帮我查天气”}{”timestamp”:”2026-06-24T10:30:02Z”,”role”:”assistant”,”content”:”正在查询...”}{”timestamp”:”2026-06-24T10:30:05Z”,”role”:”tool”,”name”:”web_search”,”result”:”...”}

// 1. 启动子 Agent——本质是创建一个新的会话+Agent Runtimesessions_spawn({ agent: ”code-reviewer”, // 子 Agent 的配置名（openclaw.json 中定义） prompt: ”Review PR #42 的安全问题”, model: ”claude-sonnet-4”, // 可以指定不同模型（不同 Agent 用不同模型） context: ”项目使用 Django + PostgreSQL”,});// 2. Agent 间消息——本质是向另一个会话注入一条用户消息sessions_send({ target: ”agent:code-reviewer:session_abc123”, message: ”重点检查 SQL 注入和 XSS”,});

Kev (Orchestrator, Opus 4.5) ├── Code Reviewer (Claude Sonnet 4) ├── Security Scanner (Claude Opus 4.5) ├── SEO Optimizer (GPT-4o) ├── Content Writer (Claude Sonnet 4) │ ├── Research Sub-Agent (深度=2) │ └── Fact Checker (深度=2) ├── Data Analyst (Claude Opus) └── Publisher (Claude Sonnet)

# 主 Agent 在工具调用阶段执行delegate_task( goal=”分析这个错误日志，找出根因”, context=””” 日志文件: /var/log/hermes/error.log 环境: Ubuntu 22.04, Python 3.11 最近变更: 升级了 Hindsight 0.8.0 ”””, toolsets=[”terminal”, ”file”, ”web”],# 子 Agent 只能用这三个工具集 role=”leaf”# leaf: 不能再 delegate；orchestrator: 可以再 delegate)

Kanban Board┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐│ Lane 1 │ │ Lane 2 │ │ Lane 3 │ │ Lane 4 │ │ Lane 5 ││ 爬取数据 │ │ 清洗数据 │ │ 分析数据 │ │ 写报告 │ │ 发布 ││ Codex │ │ Codex │ │ Codex │ │ Codex │ │ Codex │└──────────┘ └──────────┘ └──────────┘ └──────────┘ └──────────┘

// 安全审计入口// src/security/audit.tsopenclaw security audit --check // 检查危险配置openclaw security fix // 自动修复已知漏洞