API 参考文档¶
OpenClaw-Py 对外暴露两套 API:WebSocket RPC 和 OpenAI 兼容 HTTP。
1. Gateway WebSocket API¶
连接¶
使用 JSON 帧协议 (v3)。每一帧包含以下结构:
请求帧 (Client → Server)¶
响应帧 (Server → Client)¶
错误响应:
{
"type": "res",
"id": "same-request-id",
"ok": false,
"error": {
"code": "NOT_FOUND",
"message": "Session not found"
}
}
事件帧 (Server → Client,推送)¶
认证¶
连接后第一个请求必须是 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_id 或 channelId | { 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"
}
示例:
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 检查¶
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 图标¶
支持的 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 可用的记忆工具:
memory_search¶
搜索记忆中的相关内容。
memory_add¶
添加新的记忆条目。
memory_update_preference¶
更新用户偏好。
memory_forget¶
删除记忆条目。
memory_stats¶
获取记忆系统统计信息。
创建记忆工具¶
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 命令¶
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
}
}
}