跳转至

API 参考文档

OpenClaw-Py 对外暴露两套 API:WebSocket RPCOpenAI 兼容 HTTP


1. Gateway WebSocket API

连接

ws://localhost:18777/ws

使用 JSON 帧协议 (v3)。每一帧包含以下结构:

请求帧 (Client → Server)

{
  "type": "req",
  "id": "unique-request-id",
  "method": "method.name",
  "params": { ... }
}

响应帧 (Server → Client)

{
  "type": "res",
  "id": "same-request-id",
  "ok": true,
  "payload": { ... }
}

错误响应:

{
  "type": "res",
  "id": "same-request-id",
  "ok": false,
  "error": {
    "code": "NOT_FOUND",
    "message": "Session not found"
  }
}

事件帧 (Server → Client,推送)

{
  "type": "event",
  "event": "chat.delta",
  "seq": 42,
  "payload": { ... }
}

认证

连接后第一个请求必须是 connect,携带认证参数:

{
  "type": "req",
  "id": "1",
  "method": "connect",
  "params": {
    "minProtocol": 1,
    "maxProtocol": 3,
    "clientName": "my-app",
    "auth": { "token": "your-auth-token" }
  }
}

RPC 方法列表

连接与健康

方法 参数 返回 说明
connect minProtocol, maxProtocol, clientName, auth.token { protocol, server } 认证并建立连接
health (无) { status, uptime, connections } 健康检查

聊天

方法 参数 返回 说明
chat.send message, agentId?, sessionId?, model?, provider?, systemPrompt?, attachments?, abortPrevious?, temperature? { completed } 发送消息 (触发 agent 回复)。自动进行参数校验、内容净化、时间上下文注入
chat.abort sessionId, agentId? { aborted } 中止当前生成
chat.history sessionId, agentId? { messages } 获取会话消息历史
chat.resend sessionId?, agentId?, provider?, model? { completed } 重新发送最后一条用户消息 (regenerate)
chat.edit message, sessionId?, agentId?, provider?, model? { completed } 编辑最后一条用户消息并重新运行

聊天过程中服务器推送以下事件:

事件 payload 说明
chat.delta { sessionId, delta, role } 增量文本
chat.tool_call { sessionId, tool, args } 工具调用
chat.tool_result { sessionId, tool, result } 工具结果
chat.done { sessionId, usage } 回复完成
chat.error { sessionId, error } 错误

会话

方法 参数 返回 说明
sessions.list (无) { sessions: [{ agentId, file, path, size }] } 列出会话
sessions.preview path, limit? { messages } 预览会话消息
sessions.delete path { deleted } 删除会话
sessions.reset path { reset } 重置会话(清空内容保留文件)

Agent 管理

方法 参数 返回 说明
agents.list (无) [{ id, name, model }] 列出 Agent
agents.bindings (无) [{ pattern, agentId }] 查看路由绑定

通道

方法 参数 返回 说明
channels.list (无) [{ id, type, status }] 列出通道
channels.status channel_idchannelId { channel_id, name, running, status, metrics? } 通道状态

模型

方法 参数 返回 说明
models.list (无) [{ id, provider, ... }] 列出可用模型
models.probe model, provider? { available, latency } 探测模型可用性

配置

方法 参数 返回 说明
config.get key? { value } 获取配置
config.set key, value { ok } 设置配置
config.patch patch (JSON object) { ok } 批量更新配置

浏览器

方法 参数 返回 说明
browser.status profile? { started, profile, tabCount, ... } 查看浏览器状态
browser.start profile? { started, sessionId } 启动 Playwright 浏览器
browser.stop profile? { started: false } 停止浏览器
browser.tabs profile? { tabs: [...] } 列出所有 tab
browser.open url, profile? { opened, tabId, url, title } 新 tab 打开 URL
browser.navigate url, profile? { navigated, url, title } 在当前 tab 导航
browser.click ref, profile? { clicked, ref } 点击元素 (CSS 选择器)
browser.type ref, text, profile? { typed, ref } 向元素填入文本
browser.screenshot profile?, fullPage? { screenshotB64, mimeType, sizeBytes } 真实 Playwright 截图
browser.snapshot profile? { htmlLength, htmlPreview } DOM 快照
browser.evaluate fn, profile? { result } 在页面执行 JS
browser.profiles (无) { profiles } 列出浏览器 Profile
browser.createProfile name { created, profile } 创建 Profile
browser.deleteProfile name { deleted } 删除 Profile
browser.focus tabId, profile? { focused } 切换活跃 tab
browser.close tabId, profile? { closed } 关闭 tab

定时任务

方法 参数 返回 说明
cron.list (无) [{ id, cron, ... }] 列出定时任务
cron.add cron, message, agentId? { id } 添加任务
cron.remove id { ok } 删除任务
cron.history jobId?, limit? { records, count } 执行历史

工具

方法 参数 返回 说明
tools.list (无) [{ name, description }] 列出可用工具
tools.exec_approve requestId, approved { ok } 审批命令执行请求

计划管理

方法 参数 返回 说明
plan.list (无) { plans, count } 列出计划
plan.get planId { id, status, steps, ... } 获取计划详情
plan.resume planId { resumed, plan? } 恢复暂停的计划
plan.delete planId { deleted } 删除计划

备份

方法 参数 返回 说明
backup.export (无) { path, files } 导出备份
backup.status (无) { backups: [{ path, size }], count } 备份列表与状态

设备配对

方法 参数 返回 说明
device.initiate deviceName { code } 发起配对
device.complete code, response { paired } 完成配对

日志

方法 参数 返回 说明
logs.tail limit?, json?, localTime? { lines, count, path } 读取最近日志行
system.logs lines?, level? { logs, count } 按级别过滤日志

系统

方法 参数 返回 说明
system.info (无) { platform, python, uptime } 系统信息
system.event text, mode? { ok, eventType, mode } 发送系统事件
system.heartbeat.last (无) { enabled, lastHeartbeatAt } 查询最近心跳
system.presence (无) { entries } 查询组件在线状态

扩展

方法 参数 返回 说明
usage.get days? { totalTokens, estimatedCost, ... } 使用量与费用统计
doctor.run (无) { checks, summary } 运行诊断检查
skills.list (无) { skills, count } 列出可用 Skills
tts.speak text, voice?, rate?, pitch? { audio, mimeType, sizeBytes } 文本转语音
tts.voices (无) { voices, count } TTS 语音列表
update.check (无) { currentVersion, updateAvailable } 检查更新
web.status (无) { connected, uptime } 网页状态

注意: 以下方法当前返回 NOT_IMPLEMENTED 错误: wizard.start, wizard.step, push.send, voicewake.status

TTS(文本转语音)

tts.speak 使用 edge-tts 实现,支持多种语言和语音。

请求参数:

参数 类型 必填 说明
text string 要转换的文本
voice string 语音ID,默认 en-US-AvaNeural
rate string 语速,如 +50%-20%
pitch string 音调,如 +50Hz-20Hz

常用语音:

语音ID 语言 性别
en-US-AvaNeural 英语(美国)
en-US-AndrewNeural 英语(美国)
en-GB-SoniaNeural 英语(英国)
zh-CN-XiaoxiaoNeural 中文
zh-CN-YunxiNeural 中文
ja-JP-NanamiNeural 日语

响应:

{
  "audio": "base64-encoded-mp3-audio",
  "mimeType": "audio/mpeg",
  "sizeBytes": 12345,
  "voice": "en-US-AvaNeural"
}

示例:

pyclaw gateway call tts.speak --params '{"text": "Hello, world!", "voice": "en-US-AvaNeural"}'

2. OpenAI 兼容 HTTP API

POST /v1/chat/completions

完全兼容 OpenAI Chat Completions API,支持流式和非流式。

请求

curl -X POST http://localhost:18778/v1/chat/completions \
  -H "Authorization: Bearer $PYCLAW_AUTH_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "gpt-4o",
    "messages": [
      {"role": "system", "content": "You are a helpful assistant."},
      {"role": "user", "content": "Hello!"}
    ],
    "stream": true
  }'

非流式响应

{
  "id": "chatcmpl-...",
  "object": "chat.completion",
  "created": 1709000000,
  "model": "gpt-4o",
  "choices": [{
    "index": 0,
    "message": {
      "role": "assistant",
      "content": "Hello! How can I help?"
    },
    "finish_reason": "stop"
  }],
  "usage": {
    "prompt_tokens": 20,
    "completion_tokens": 10,
    "total_tokens": 30
  }
}

流式响应 (SSE)

data: {"id":"chatcmpl-...","object":"chat.completion.chunk","choices":[{"delta":{"content":"Hello"},"index":0}]}

data: {"id":"chatcmpl-...","object":"chat.completion.chunk","choices":[{"delta":{"content":"!"},"index":0}]}

data: [DONE]

GET /v1/models

列出所有可用模型。

响应

{
  "object": "list",
  "data": [
    {
      "id": "gpt-4o",
      "object": "model",
      "owned_by": "openai",
      "created": 1709000000
    },
    {
      "id": "claude-sonnet-4-20250514",
      "object": "model",
      "owned_by": "anthropic",
      "created": 1709000000
    }
  ]
}

POST /v1/responses

兼容 OpenAI Responses API,支持流式 SSE。


3. MCP (Model Context Protocol)

OpenClaw-Py 作为 MCP 客户端,连接外部 MCP 服务器。

配置

~/.pyclaw/pyclaw.json 中添加 MCP 服务器:

{
  "tools": {
    "mcpServers": {
      "server-name": {
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path"]
      }
    }
  }
}

传输方式

方式 配置 适用场景
stdio command + args 本地进程
HTTP url + headers 远程端点

工具发现

启动时自动调用每个 MCP 服务器的 tools/list,将发现的工具注册到 Agent 工具池。工具名称以 mcp:<server>:<tool> 格式标识。

热重载

MCP 配置支持热重载 — McpConfigWatcher 以固定间隔 (默认 5s) 轮询配置文件的 tools.mcpServers 段。 当检测到变更时,自动断开旧连接并重新连接新配置的 MCP 服务器。

CLI 检查

# 查看 MCP 服务器状态
pyclaw mcp status

# 列出所有 MCP 工具
pyclaw mcp list-tools

4. Gateway CLI 命令

pyclaw gateway 子命令组提供网关的运维与调试能力。

命令 说明 关键参数
pyclaw gateway (无子命令) 启动 Gateway 服务器(等同 run --port, --bind, --auth-token
pyclaw gateway run 启动 Gateway 服务器 --port, --bind, --auth-token
pyclaw gateway status 显示服务状态 + 可选 RPC 探测 --url, --token, --password, --timeout, --no-probe, --deep, --json
pyclaw gateway probe 探测 Gateway 连通性 --url, --token, --password, --timeout, --json
pyclaw gateway call 底层 RPC 调用 METHOD (位置参数), --params, --url, --token, --password, --timeout, --json
pyclaw gateway discover 局域网发现 (mDNS/端口扫描) --timeout, --json

示例

# 查看 Gateway 状态(含探测)
pyclaw gateway status --deep --json

# 探测连通性
pyclaw gateway probe --url ws://192.168.1.100:18777

# 调用任意 RPC 方法
pyclaw gateway call health --params '{}'
pyclaw gateway call chat.send --params '{"message": "Hello"}'

# 局域网发现
pyclaw gateway discover --json

5. 本地模型管理 CLI

pyclaw models 子命令组中新增的本地模型管理命令:

命令 说明 关键参数
pyclaw models download 从 HuggingFace/ModelScope 下载模型 REPO_ID, --filename, --backend (llamacpp/mlx), --source
pyclaw models local 列出已下载的本地模型 --json
pyclaw models delete-local 删除本地模型文件 MODEL_ID
pyclaw models select 设置默认活跃模型 MODEL_ID

支持后端

后端 安装 说明
llamacpp pip install 'openclaw-py[llamacpp]' llama.cpp (GGUF 格式, CPU/GPU)
mlx pip install 'openclaw-py[mlx]' MLX (Apple Silicon 专用)
ollama 外部安装 Ollama 通过 OpenAI 兼容 API 接入

示例

# 下载 GGUF 模型
pyclaw models download Qwen/Qwen3-4B-GGUF

# 下载 MLX 模型 (Apple Silicon)
pyclaw models download mlx-community/Llama-3-8B-4bit --backend mlx

# 从 ModelScope 下载
pyclaw models download qwen/Qwen3-4B-GGUF --source modelscope

# 列出本地模型
pyclaw models local

# 设置默认模型
pyclaw models select "Qwen/Qwen3-4B-GGUF/qwen3-4b-q4_k_m.gguf"

6. Agent 工具扩展

Phase 63-65 新增的 Agent 工具:

工具 功能 注册方式
desktop_screenshot 跨平台桌面截图 (mss) + macOS 窗口截图 默认注册
send_file 向用户发送文件 (自动检测 MIME 类型) 默认注册
read_pdf PDF 文件文本提取 pyclaw[office]
read_docx Word 文档文本提取 pyclaw[office]
read_xlsx Excel 表格数据提取 pyclaw[office]
read_pptx PowerPoint 幻灯片文本提取 pyclaw[office]

7. 错误码

错误码 含义
AUTH_REQUIRED 未认证或 token 无效
NOT_FOUND 资源不存在 (会话、Agent 等)
INVALID_PARAMS 请求参数无效
RATE_LIMITED 请求频率超限
INTERNAL_ERROR 服务器内部错误
METHOD_NOT_FOUND RPC 方法不存在
ALREADY_EXISTS 资源已存在
ABORTED 操作被中止
NOT_IMPLEMENTED 方法已注册但尚未实现

8. UI 组件 API

Canvas 绘图组件 (pyclaw.ui.canvas)

提供 Canvas 的高层封装,支持矢量图形绘制。

DrawableCanvas

主要画布组件,支持各种图形绘制。

from pyclaw.ui import DrawableCanvas

canvas = DrawableCanvas(width=400, height=300)

# 绘制图形
canvas.draw_rect(10, 10, 100, 50, color="blue", fill=True)
canvas.draw_circle(200, 150, 40, color="red")
canvas.draw_line(0, 0, 400, 300, color="black", width=2)
canvas.draw_text(50, 50, "Hello", color="blue", size=20)

# 多边形和星形
canvas.draw_polygon([(0,0), (100,0), (50,86)], color="yellow")
canvas.draw_star(200, 150, 50, 25, points=5, color="gold")

# 画布操作
canvas.clear()
shape = canvas.pop_shape()  # 移除最后一个图形
canvas.remove_shape(shape)  # 移除指定图形

# 导出 PNG
png_bytes = await canvas.capture_png()

FreehandCanvas

支持手写/涂鸦的画布。

from pyclaw.ui import FreehandCanvas

canvas = FreehandCanvas(width=400, height=300, stroke_color="black", stroke_width=3)
canvas.set_stroke(color="red", width=5)
# 用户可以自由绘制

PaintStyle

图形样式配置。

属性 类型 默认值 说明
color str "#000000" 颜色
stroke_width float 1.0 线宽
fill bool True 是否填充
anti_alias bool True 抗锯齿

路径元素

用于绘制复杂路径。

from pyclaw.ui.canvas import MoveTo, LineTo, CubicTo, QuadraticTo, ClosePath

elements = [
    MoveTo(0, 50),
    CubicTo(25, 0, 75, 100, 100, 50),  # 三次贝塞尔曲线
    LineTo(150, 50),
    ClosePath(),
]
canvas.draw_path(elements, color="blue")

图标组件 (pyclaw.ui.icons)

提供 Font Awesome、Material 和 Lucide 图标支持。

Font Awesome 图标

from pyclaw.ui import fa_icon, fa_icon_button

# 创建图标
icon = fa_icon("heart", color="red", size=32)
page.add(icon)

# 创建按钮
btn = fa_icon_button("user", color="blue", on_click=handle_click, tooltip="User")

支持的 Font Awesome 图标名称: - 操作: plus, minus, xmark, check, edit, trash, copy, paste, save - 导航: arrow-left, arrow-right, arrow-up, arrow-down, home, menu - 状态: circle-check, circle-xmark, triangle-exwarning, circle-info, circle-question - 通信: comment, paper-plane, microphone - 用户: user, users, gear - 文件: file, folder, folder-open, database - AI: brain, robot, wand-magic-sparkles - 媒体: image, video, music, play, pause, stop - 其他: heart, star, magnifying-glass, filter, lock, unlock, key, clock, calendar, code, terminal, eye, eye-slash, spinner, link

Material 图标 (语义封装)

from pyclaw.ui import Icon

# 状态图标
Icon.success(color="green", size=32)  # 勾选
Icon.error(color="red")               # 错误
Icon.warning(color="orange")          # 警告
Icon.info(color="blue")               # 信息

# 操作图标
Icon.add()
Icon.edit()
Icon.delete()
Icon.copy()
Icon.save()

# 导航图标
Icon.menu()
Icon.back()
Icon.forward()
Icon.home()

# AI 图标
Icon.brain()      # 大脑
Icon.robot()      # 机器人
Icon.sparkles()   # 闪光

Lucide 图标

from pyclaw.ui import lucide_icon

icon = lucide_icon("send", color="blue", size=24)

支持的 Lucide 图标: send, sparkles, bot, message-square, terminal, code

图像画布 (pyclaw.ui.image_canvas)

支持在图像上绘制标注。

ImageCanvas

from pyclaw.ui import ImageCanvas

canvas = ImageCanvas(
    image_src="photo.jpg",
    width=600,
    height=400,
)

# 绘制标注
canvas.draw_rect(100, 100, 50, 50, color="red", stroke_width=2)
canvas.draw_arrow(150, 125, 200, 100, color="red")
canvas.draw_text(100, 160, "Note", color="red")
canvas.draw_marker(300, 200, label="Point A", color="blue")

# 更换背景图
canvas.set_image("another.png")

# 导出
png_bytes = await canvas.capture_png()

AnnotationCanvas

带工具栏的交互式标注画布。

from pyclaw.ui import AnnotationCanvas

canvas = AnnotationCanvas(
    image_src="document.png",
    width=600,
    height=400,
)

# 设置当前工具
canvas.current_tool = "pen"      # pen, rect, circle, arrow, marker, text
canvas.stroke_color = "red"
canvas.stroke_width = 3.0

# 工具栏
toolbar = canvas.build_toolbar()

画布面板 (pyclaw.ui.canvas_panel)

完整的绘图面板,带工具栏和操作按钮。

CanvasPanel

from pyclaw.ui import CanvasPanel

def on_export(png_bytes):
    with open("drawing.png", "wb") as f:
        f.write(png_bytes)

panel = CanvasPanel(
    width=600,
    height=400,
    canvas_bgcolor="white",
    on_export=on_export,
)
page.add(panel)

# 访问底层画布
panel.canvas.draw_rect(10, 10, 50, 50, color="blue")

# 程序化控制
panel.current_tool = "rect"
panel.stroke_color = "red"

SimpleCanvasPanel

简化的绘图面板,适合嵌入式使用。

from pyclaw.ui import SimpleCanvasPanel

panel = SimpleCanvasPanel(width=300, height=200)
page.add(panel)

4. Memory API

记忆管理系统提供完整的上下文管理功能,包括: - 长对话自动压缩 - 用户偏好持久化 - 对话历史保存 - 长期记忆管理

Python API

MemoryManager

统一的记忆管理接口。

from pathlib import Path
from pyclaw.memory import MemoryManager, MemoryConfig
from pyclaw.agents.types import ModelConfig

# 初始化
manager = MemoryManager(
    workspace_dir=Path("./workspace"),
    config=MemoryConfig.from_yaml(Path("config.yaml")),
    model=ModelConfig(provider="openai", model_id="gpt-4o-mini"),
)

# 使用上下文管理器
with MemoryManager(Path("./workspace")) as manager:
    # 检查上下文
    result = manager.check_context(messages)
    if result.needs_compaction:
        summary = await manager.generate_summary(result.messages_to_compact)
        messages = manager.apply_compaction(messages, summary, result)

    # 记忆操作
    manager.add_memory("用户偏好中文", category="preference")
    manager.set_preference("language", "中文")
    prefs = manager.get_all_preferences()

    # 搜索
    results = manager.search_memory("用户偏好", limit=5)

    # 持久化对话
    manager.persist_dialog(messages)

    # 统计信息
    stats = manager.get_stats()

配置选项

from pyclaw.memory.types import (
    MemoryConfig,
    CompactionConfig,
    SummaryConfig,
    DialogConfig,
    LongTermMemoryConfig,
    ToolResultConfig,
)

config = MemoryConfig(
    # 上下文压缩
    compaction=CompactionConfig(
        enabled=True,
        max_tokens=128000,
        reserve_ratio=0.3,
        min_messages_to_compact=10,
        compaction_threshold=0.85,
    ),
    # 摘要生成
    summary=SummaryConfig(
        enabled=True,
        model="gpt-4o-mini",
        max_summary_tokens=2000,
        async_mode=True,
    ),
    # 对话持久化
    dialog=DialogConfig(
        enabled=True,
        retention_days=30,
    ),
    # 长期记忆
    long_term=LongTermMemoryConfig(
        enabled=True,
        file="MEMORY.md",
    ),
    # 工具输出压缩
    tool_result=ToolResultConfig(
        enabled=True,
        max_length=2000,
        retention_days=7,
    ),
)

Agent Tools API

Agent 可用的记忆工具:

搜索记忆中的相关内容。

{
  "name": "memory_search",
  "arguments": {
    "query": "用户偏好",
    "limit": 10
  }
}

memory_add

添加新的记忆条目。

{
  "name": "memory_add",
  "arguments": {
    "content": "用户偏好使用中文交流",
    "category": "preference"
  }
}

memory_update_preference

更新用户偏好。

{
  "name": "memory_update_preference",
  "arguments": {
    "key": "language",
    "value": "中文"
  }
}

memory_forget

删除记忆条目。

{
  "name": "memory_forget",
  "arguments": {
    "id": 123
  }
}

memory_stats

获取记忆系统统计信息。

{
  "name": "memory_stats",
  "arguments": {}
}

创建记忆工具

from pyclaw.agents.tools.memory_tools import create_memory_tools
from pyclaw.memory import MemoryManager

manager = MemoryManager(Path("./workspace"))
manager.open()

# 创建所有记忆工具
tools = create_memory_tools(memory_manager=manager)

# 注册到 agent
for tool in tools:
    agent.register_tool(tool)

集成到 Agent 运行时

MemoryAwareRunner

from pyclaw.agents.memory_integration import MemoryAwareRunner

with MemoryAwareRunner(
    workspace_dir=Path("./workspace"),
    model=ModelConfig(provider="openai", model_id="gpt-4o"),
) as runner:
    # 获取工具
    tools = runner.get_tools()

    # 预处理消息(自动压缩)
    messages = await runner.preprocess_messages(messages)

    # 获取上下文用于提示词
    context = runner.get_context_for_prompt()

    # 后处理(自动保存对话)
    await runner.postprocess_turn(messages, session_key="session-1")

快速设置

from pyclaw.agents.memory_integration import setup_memory_for_agent

result = setup_memory_for_agent(
    workspace_dir=Path("./workspace"),
    model=ModelConfig(provider="openai", model_id="gpt-4o"),
)

# 返回所有组件
manager = result["manager"]
tools = result["tools"]
context = result["context"]  # 用于注入系统提示

# 使用完毕后关闭
manager.close()

文件存储布局

workspace/
├── MEMORY.md                    # 长期记忆:用户偏好、重要决策
├── memory/
│   ├── 2026-03-24.md           # 每日摘要日志
│   └── ...
├── dialog/
│   ├── 2026-03-24.jsonl        # 原始对话记录
│   └── ...
├── tool_result/
│   └── <uuid>.txt              # 长工具输出缓存
└── memory.db                   # SQLite 索引数据库

类型定义

from pyclaw.memory.types import (
    # 结果类型
    ContextCheckResult,    # 上下文检查结果
    StructuredSummary,     # 结构化摘要

    # 配置类型
    MemoryConfig,          # 完整配置
    CompactionConfig,      # 压缩配置
    SummaryConfig,         # 摘要配置
    DialogConfig,          # 对话配置
    LongTermMemoryConfig,  # 长期记忆配置
    ToolResultConfig,      # 工具输出配置

    # 枚举
    MemoryCategory,        # 记忆分类
    CompactionTrigger,     # 压缩触发类型
)

# 结构化摘要
summary = StructuredSummary(
    goals=["完成功能实现"],
    constraints=["使用 Python 3.10+"],
    progress="已完成基础功能",
    decisions=["使用 SQLite 存储"],
    next_steps=["编写测试"],
    key_context=["项目: pyclaw"],
)

# 转换为 Markdown
markdown = summary.to_markdown()

# 转换为消息
message = summary.to_message()

9. Active Memory API

Active Memory 子系统在 Agent 生成回复前自动检索相关记忆。

Python API

ActiveMemoryRecaller

主要召回器类,执行记忆检索和上下文构建。

from pathlib import Path
from pyclaw.memory.active_recall import ActiveMemoryRecaller
from pyclaw.memory.recall_types import ActiveRecallConfig, QueryMode, PromptStyle
from pyclaw.memory import MemoryManager

# 配置
config = ActiveRecallConfig(
    enabled=True,
    query_mode=QueryMode.RECENT,
    prompt_style=PromptStyle.BALANCED,
    max_summary_chars=220,
    recent_user_turns=2,
    recent_assistant_turns=1,
    cache_ttl_ms=15000,
)

# 初始化
recaller = ActiveMemoryRecaller(
    memory_manager=memory_manager,
    config=config,
    workspace_dir=Path("./workspace"),
)

# 执行召回
result = await recaller.recall(
    latest_message="What's the status of the API?",
    recent_turns=[
        {"role": "user", "content": "Previous message..."},
        {"role": "assistant", "content": "Previous response..."},
    ],
    session_key="session-123",
)

# 检查结果
if result.is_success:
    # 构建上下文消息
    context_msg = recaller.build_context_message(result)
    # 注入到消息列表
    messages.insert(0, context_msg)

ActiveRecallResult

召回操作的结果类型。

from pyclaw.memory.recall_types import ActiveRecallResult

result = ActiveRecallResult(
    status="ok",  # "ok" | "empty" | "timeout" | "unavailable"
    elapsed_ms=42,
    summary="Retrieved relevant memories...",
    search_debug={
        "backend": "hybrid",
        "configured_mode": "recent",
        "effective_mode": "recent",
        "fallback": "",
        "search_ms": 38,
        "hits": 5,
    },
)

# 便捷属性
result.is_success      # True if status == "ok" and summary exists
result.is_empty        # True if status == "empty"
result.is_timeout      # True if status == "timeout"
result.is_unavailable  # True if status == "unavailable"

QueryMode 枚举

from pyclaw.memory.recall_types import QueryMode

QueryMode.MESSAGE  # 仅使用最新用户消息
QueryMode.RECENT   # 最近对话轮次 + 最新消息
QueryMode.FULL     # 完整对话上下文

PromptStyle 枚举

from pyclaw.memory.recall_types import PromptStyle

PromptStyle.BALANCED        # 平衡上下文与记忆
PromptStyle.STRICT          # 严格使用记忆内容
PromptStyle.CONTEXTUAL      # 优先理解对话上下文
PromptStyle.RECALL_HEAVY    # 最大化记忆召回
PromptStyle.PRECISION_HEAVY # 优先精确性
PromptStyle.PREFERENCE_ONLY # 仅召回用户偏好

CLI 命令

# 启用 Active Memory(当前会话)
pyclaw active-memory on

# 禁用 Active Memory(当前会话)
pyclaw active-memory off

# 查看状态
pyclaw active-memory status

10. Wiki API

Wiki 子系统管理结构化知识库。

Python API

WikiVault

Wiki 目录管理器,处理文件 I/O 和页面管理。

from pathlib import Path
from pyclaw.memory.wiki.vault import WikiVault
from pyclaw.memory.wiki.types import WikiConfig, WikiPage

# 初始化
vault = WikiVault(
    vault_path=Path("./wiki"),
    config=WikiConfig(),
)
vault.initialize()

# 获取页面
page = vault.get_page("entities/company-acme")
if page:
    print(page.title)
    print(page.content)
    print(len(page.claims))

# 保存页面
page = WikiPage(
    path="entities/new-entity",
    title="New Entity",
    content="# New Entity\n\nDescription...",
    claims=[],
)
vault.save_page(page)

# 列出页面
pages = vault.list_pages("entities")

# 获取统计
stats = vault.get_stats()
# {"pages": 10, "claims": 25, "directories": 7, "sources": 3}

ClaimsManager

Claim 管理器,提供 CRUD 操作和矛盾检测。

from pathlib import Path
from pyclaw.memory.wiki.claims import ClaimsManager
from pyclaw.memory.wiki.types import ClaimStatus

# 初始化
manager = ClaimsManager(
    claims_dir=Path("./wiki/.openclaw-wiki/claims"),
)

# 添加 Claim
claim = manager.add_claim(
    page_path="entities/company-acme",
    text="Company uses PostgreSQL for primary database",
    evidence=[
        {
            "source_id": "doc-1",
            "source_path": "sources/architecture.md",
            "lines": "45-52",
            "weight": 1.0,
        }
    ],
    tags=["database", "infrastructure"],
    confidence=0.8,
)

# 获取 Claim
claim = manager.get_claim(claim.id)

# 更新 Claim
manager.update_claim(
    claim.id,
    confidence=0.9,
    tags=["database", "infrastructure", "verified"],
)

# 查找矛盾
contradictions = manager.find_contradictions(claim)

# 取代旧 Claim
new_claim = manager.supersede_claim(
    old_claim_id=claim.id,
    new_claim_text="Company migrated to MySQL in 2024",
    rationale="Database migration completed",
)

# 删除 Claim
manager.delete_claim(claim.id)

# 导出为 JSONL
manager.export_to_jsonl(Path("./claims.jsonl"))

Wiki 类型定义

from pyclaw.memory.wiki.types import (
    WikiConfig,
    WikiPage,
    Claim,
    ClaimStatus,
    Evidence,
)

# Claim 示例
claim = Claim(
    id="abc123",
    text="Company uses PostgreSQL",
    status=ClaimStatus.ACTIVE,
    confidence=0.8,
    evidence=[
        Evidence(
            source_id="doc-1",
            source_path="sources/architecture.md",
            lines="45-52",
            weight=1.0,
            excerpt="...",
        )
    ],
    tags=["database"],
)

# Claim 状态
ClaimStatus.ACTIVE      # 活跃
ClaimStatus.SUPERSEDED  # 已被取代
ClaimStatus.REFUTED     # 已被反驳

CLI 命令

# 初始化 Wiki
pyclaw wiki init [--path ./wiki]

11. Dreaming API

Dreaming 子系统执行记忆巩固和提升。

Python API

SignalsStore

短期记忆信号存储。

from pathlib import Path
from pyclaw.memory.dreaming.signals import SignalsStore

# 初始化
store = SignalsStore(
    store_dir=Path("./.dreams"),
)

# 添加信号
signal = store.add_signal(
    content="User prefers dark mode in the UI",
    source_ref="MEMORY.md:15",
    query="ui preferences",
    tags=["preferences", "ui"],
)

# 记录巩固(Agent 引用)
store.record_consolidation(signal.content_hash)

# 获取信号
all_signals = store.get_all_signals()
recent_signals = store.get_signals_by_age(max_age_days=30)

# 清理旧信号
removed = store.cleanup_old_signals(max_age_days=60)

# 获取统计
stats = store.get_stats()
# {"total": 100, "total_recall_queries": 250, "total_consolidations": 45}

PromotionScorer

晋升评分器,计算候选记忆的晋升分数。

from pyclaw.memory.dreaming.scoring import PromotionScorer, ScoreComponents
from pyclaw.memory.dreaming.types import DreamingConfig

# 配置
config = DreamingConfig(
    enabled=True,
    min_score=0.5,
    min_recall_count=2,
    min_unique_queries=2,
    scoring_weights={
        "frequency": 0.25,
        "relevance": 0.20,
        "diversity": 0.15,
        "recency": 0.15,
        "consolidation": 0.15,
        "conceptual": 0.10,
    },
)

# 初始化
scorer = PromotionScorer(config)

# 评分单个信号
score, components = scorer.score_signal(signal)
print(f"Score: {score}")
print(f"Components: {components.to_dict()}")

# 过滤候选
candidates = scorer.filter_candidates(
    signals=all_signals,
    min_score=0.5,
)
# 返回按分数降序排列的 PromotionCandidate 列表

DreamingPhases

三阶段记忆巩固执行器。

from pyclaw.memory.dreaming.phases import DreamingPhases
from pyclaw.memory.dreaming.types import DreamingPhase

# 初始化
phases = DreamingPhases(
    workspace_dir=Path("./workspace"),
    config=config,
)

# 执行 LIGHT 阶段
light_report = await phases.run_phase(DreamingPhase.LIGHT)

# 执行 DEEP 阶段
deep_report = await phases.run_phase(DreamingPhase.DEEP)
print(f"Promoted {deep_report.promoted_count} items")

# 执行 REM 阶段
rem_report = await phases.run_phase(DreamingPhase.REM)

Dreaming 类型定义

from pyclaw.memory.dreaming.types import (
    DreamingConfig,
    DreamingPhase,
    StorageMode,
    ShortTermSignal,
    PromotionCandidate,
    DreamingReport,
)

# 配置
config = DreamingConfig(
    enabled=True,
    cron="0 3 * * *",  # 每天凌晨 3 点
    limit=10,
    min_score=0.5,
    storage=StorageMode.INLINE,
)

# 阶段
DreamingPhase.LIGHT  # 收集信号
DreamingPhase.DEEP   # 评分提升
DreamingPhase.REM    # 反思叙事

# 存储模式
StorageMode.INLINE    # 追加到 MEMORY.md
StorageMode.SEPARATE  # 存储到独立文件
StorageMode.BOTH      # 两者都保存

配置

~/.pyclaw/pyclaw.json 中配置:

{
  "dreaming": {
    "enabled": true,
    "cron": "0 3 * * *",
    "timezone": "Asia/Shanghai",
    "limit": 10,
    "minScore": 0.5,
    "minRecallCount": 2,
    "minUniqueQueries": 2,
    "recencyHalfLifeDays": 7.0,
    "maxAgeDays": 30,
    "storage": "inline",
    "scoringWeights": {
      "frequency": 0.25,
      "relevance": 0.20,
      "diversity": 0.15,
      "recency": 0.15,
      "consolidation": 0.15,
      "conceptual": 0.10
    }
  }
}