Merge data agent runtime updates

This commit is contained in:
wuyang6
2026-05-21 20:54:57 +08:00
20 changed files with 3016 additions and 37 deletions
+47 -2
View File
@@ -583,6 +583,22 @@ bash "$HOME/zk-data-agent/scripts/install-from-git.sh"
- 安装前端依赖并构建。
- 安装并启动用户级 systemd 服务。
如果要启用“平台账号 = Linux 用户”的托管工作区,需要使用 root/systemd system 服务部署:
```bash
cd "$HOME/zk-data-agent"
sudo -E env PATH="$PATH" bash scripts/deploy-ubuntu.sh main --system-service --enable-linux-accounts
```
启用后:
- 平台注册/登录账号时会同步创建同名 Linux 用户。
- 平台密码会同步设置为 Linux 用户密码。
- Linux 用户允许 SSH 登录。
- 本机托管工作区会使用 `/home/<account_id>/zk-agent/`
- `python_exec``python_package``bash` 会在对应 Linux 用户身份下执行。
- Jupyter 远程工作区保持现有逻辑,不参与本机 Linux 用户隔离。
`.env.deploy` 会保存:
```text
@@ -595,6 +611,8 @@ CLAW_BACKEND_PORT
CLAW_FRONTEND_HOST
CLAW_FRONTEND_PORT
CLAW_API_URL
CLAW_SERVICE_SCOPE
CLAW_ENABLE_LINUX_ACCOUNTS
CLAW_NPM_BIN
CLAW_NPX_BIN
CLAW_NODE_BIN
@@ -618,6 +636,12 @@ bash scripts/update-server-fast.sh
bash scripts/deploy-ubuntu.sh
```
root/system 服务更新:
```bash
sudo -E env PATH="$PATH" bash scripts/deploy-ubuntu.sh main --system-service --enable-linux-accounts
```
部署指定分支:
```bash
@@ -632,7 +656,7 @@ bash scripts/deploy-ubuntu.sh main --force
### 服务管理
部署脚本会安装两个用户级 systemd 服务:
部署脚本默认安装用户级 systemd 服务;以 root 执行或指定 `--system-service` 时安装 system 级服务。服务名默认是
```text
zk-data-agent-backend
@@ -646,6 +670,13 @@ systemctl --user status zk-data-agent-backend
systemctl --user status zk-data-agent-frontend
```
system 级服务使用:
```bash
systemctl status zk-data-agent-backend
systemctl status zk-data-agent-frontend
```
查看日志:
```bash
@@ -653,12 +684,25 @@ journalctl --user -u zk-data-agent-backend -f
journalctl --user -u zk-data-agent-frontend -f
```
system 级日志使用:
```bash
journalctl -u zk-data-agent-backend -f
journalctl -u zk-data-agent-frontend -f
```
重启服务:
```bash
systemctl --user restart zk-data-agent-backend zk-data-agent-frontend
```
system 级重启使用:
```bash
systemctl restart zk-data-agent-backend zk-data-agent-frontend
```
停止服务:
```bash
@@ -683,6 +727,7 @@ Ubuntu 部署建议:
- Python `3.10.14`
- Node.js `20``22`
- `npm`
- 启用 Linux 账号工作区时,还需要 `passwd``python3``python3-venv`,并要求服务以 root/systemd system 方式运行。
首次安装如果缺 Python 编译依赖,可以执行:
@@ -690,7 +735,7 @@ Ubuntu 部署建议:
bash scripts/deploy-ubuntu.sh --bootstrap-system
```
`--bootstrap-system` 会使用 `sudo apt-get` 安装系统依赖;应用代码、虚拟环境、前端依赖、运行数据和 systemd 用户服务仍然位于当前用户目录。
`--bootstrap-system` 会使用 `sudo apt-get` 安装系统依赖;普通模式下应用代码、虚拟环境、前端依赖、运行数据和 systemd 用户服务仍然位于当前用户目录。启用 `--enable-linux-accounts` 后,账号工作区位于 `/home/<account_id>/zk-agent/`
## 开发验证
+157 -2
View File
@@ -11,6 +11,7 @@ import csv
import hashlib
import json
import os
import pwd
import queue
import re
import shutil
@@ -67,6 +68,9 @@ from src.session_store import (
)
from src.token_budget import calculate_token_budget
LINUX_ACCOUNT_WORKSPACE_NAME = 'zk-agent'
LINUX_ACCOUNT_ENV = 'CLAW_ENABLE_LINUX_ACCOUNTS'
STATIC_DIR = Path(__file__).resolve().parents[2] / 'frontend' / 'legacy-static'
API_TOOL_CONTENT_MAX_CHARS = 20000
@@ -626,6 +630,8 @@ class AgentState:
if not account_id:
return self.session_directory.parent
safe_id = _safe_account_id(account_id)
if _linux_accounts_enabled():
return _linux_account_workspace(safe_id).resolve()
return (self.session_directory.parent / 'accounts' / safe_id).resolve()
def account_paths(self, account_id: str | None) -> dict[str, Path]:
@@ -825,7 +831,15 @@ class AgentState:
for directory in paths.values():
if directory.name != '.venv':
directory.mkdir(parents=True, exist_ok=True)
self._ensure_python_env(paths['python_env'])
runtime_user = (
_safe_account_id(account_id)
if account_id and _linux_accounts_enabled()
else None
)
if runtime_user:
_ensure_linux_user_exists(runtime_user)
_chown_path_for_linux_user(paths['base'], runtime_user, recursive=True)
self._ensure_python_env(paths['python_env'], account_id)
permissions = AgentPermissions(
allow_file_write=config.allow_write,
allow_shell_commands=config.allow_shell,
@@ -837,6 +851,7 @@ class AgentState:
session_directory=paths['sessions'],
scratchpad_root=paths['scratchpad'],
python_env_dir=paths['python_env'],
runtime_user=runtime_user,
enabled_skill_names=self.enabled_skill_names(account_id),
)
model_config = ModelConfig(
@@ -936,12 +951,37 @@ class AgentState:
def lock(self) -> Lock:
return self._lock
def _ensure_python_env(self, env_dir: Path) -> None:
def _ensure_python_env(self, env_dir: Path, account_id: str | None = None) -> None:
python_bin = env_dir / 'bin' / 'python'
pip_bin = env_dir / 'bin' / 'pip'
if python_bin.exists() and pip_bin.exists():
return
env_dir.parent.mkdir(parents=True, exist_ok=True)
runtime_user = _safe_account_id(account_id) if account_id and _linux_accounts_enabled() else None
if runtime_user:
_ensure_linux_user_exists(runtime_user)
env_dir.parent.mkdir(parents=True, exist_ok=True)
_chown_path_for_linux_user(env_dir.parent, runtime_user, recursive=True)
runtime_python = os.environ.get('CLAW_RUNTIME_PYTHON_BIN') or shutil.which('python3')
if not runtime_python:
raise RuntimeError('Linux runtime requires python3 or CLAW_RUNTIME_PYTHON_BIN')
subprocess.run(
[
'runuser',
'-u',
runtime_user,
'--',
runtime_python,
'-m',
'venv',
str(env_dir),
],
check=True,
capture_output=True,
text=True,
timeout=120,
)
return
venv.EnvBuilder(with_pip=True, symlinks=False).create(env_dir)
@@ -3785,6 +3825,107 @@ def _safe_account_id(account_id: str | None) -> str:
return normalized[:80] or 'default'
def _linux_accounts_enabled() -> bool:
return os.environ.get(LINUX_ACCOUNT_ENV, '').strip().lower() in {
'1',
'true',
'yes',
'on',
}
def _linux_account_workspace(account_id: str) -> Path:
return Path('/home') / account_id / LINUX_ACCOUNT_WORKSPACE_NAME
def _validate_linux_username(account_id: str) -> None:
if not re.fullmatch(r'[a-z_][a-z0-9_-]{1,31}', account_id):
raise HTTPException(
status_code=400,
detail='启用 Linux 账号时,账号名必须以小写字母或下划线开头,只包含小写字母、数字、下划线或中划线,长度 2-32',
)
def _ensure_linux_account(account_id: str, password: str) -> None:
_validate_linux_username(account_id)
if os.geteuid() != 0:
raise HTTPException(
status_code=500,
detail='CLAW_ENABLE_LINUX_ACCOUNTS=1 需要后端以 root 身份运行',
)
try:
pwd.getpwnam(account_id)
except KeyError:
subprocess.run(
[
'useradd',
'-m',
'-d',
f'/home/{account_id}',
'-s',
'/bin/bash',
account_id,
],
check=True,
capture_output=True,
text=True,
timeout=30,
)
chpasswd = subprocess.run(
['chpasswd'],
input=f'{account_id}:{password}\n',
check=False,
capture_output=True,
text=True,
timeout=30,
)
if chpasswd.returncode != 0:
detail = (chpasswd.stderr or chpasswd.stdout or '').strip()
raise HTTPException(status_code=500, detail=f'同步 Linux 用户密码失败:{detail}')
workspace = _linux_account_workspace(account_id)
for child in (
workspace / 'sessions',
workspace / 'python',
workspace / 'memory',
workspace / 'integrations',
):
child.mkdir(parents=True, exist_ok=True)
_chown_path_for_linux_user(workspace, account_id, recursive=True)
def _ensure_linux_user_exists(account_id: str) -> None:
try:
pwd.getpwnam(account_id)
except KeyError as exc:
raise RuntimeError(
f'Linux runtime user does not exist: {account_id}. '
'Create the account through the platform first, or disable CLAW_ENABLE_LINUX_ACCOUNTS.'
) from exc
def _chown_path_for_linux_user(path: Path, account_id: str, *, recursive: bool = False) -> None:
if os.name != 'posix':
return
try:
user_info = pwd.getpwnam(account_id)
except KeyError:
return
target = path.resolve(strict=False)
try:
os.chown(target, user_info.pw_uid, user_info.pw_gid)
except PermissionError:
return
except FileNotFoundError:
return
if not recursive or not target.is_dir():
return
for child in target.rglob('*'):
try:
os.chown(child, user_info.pw_uid, user_info.pw_gid)
except (PermissionError, FileNotFoundError):
continue
def _admin_token() -> str:
return os.environ.get('ZK_ADMIN_TOKEN') or 'admin'
@@ -3933,6 +4074,8 @@ def _admin_create_account(state: AgentState, account_id: str) -> dict[str, Any]:
safe_id = _safe_account_id(account_id)
if safe_id == 'default':
raise HTTPException(status_code=400, detail='账号名不合法')
if _linux_accounts_enabled():
_validate_linux_username(safe_id)
users = _load_users_file(state)
user_rows = users.setdefault('users', [])
if any(
@@ -3941,6 +4084,8 @@ def _admin_create_account(state: AgentState, account_id: str) -> dict[str, Any]:
for item in user_rows
):
raise HTTPException(status_code=400, detail='账号已存在')
if _linux_accounts_enabled():
_ensure_linux_account(safe_id, '123456')
user_rows.append(
{
'id': safe_id,
@@ -3984,8 +4129,18 @@ def _admin_delete_account(state: AgentState, account_id: str) -> dict[str, Any]:
encoding='utf-8',
)
base = _accounts_root(state) / safe_id
if _linux_accounts_enabled():
base = _linux_account_workspace(safe_id)
if base.exists():
shutil.rmtree(base)
if _linux_accounts_enabled():
subprocess.run(
['passwd', '-l', safe_id],
check=False,
capture_output=True,
text=True,
timeout=30,
)
state._clear_agents_for_account(safe_id)
return {'account_id': safe_id, 'deleted': True}
@@ -0,0 +1,478 @@
# 10. Agent 记忆机制调研与 ZK Data Agent 对比
本文整理主流 Agent / AI 产品的记忆实现方式,并对照 ZK Data Agent 当前实现。目标不是判断哪一种“最好”,而是说明不同记忆机制分别解决什么问题,以及为什么我们当前选择“用户记忆 + Skill 使用记忆 + 异步整理队列 + Markdown 可编辑文件”的路线。
调研时间:2026-05-19。
## 1. 结论摘要
主流记忆实现大致分为六类:
| 类型 | 代表 | 核心做法 | 适合场景 |
|------|------|----------|----------|
| 产品级长期记忆 | ChatGPT Memory | 平台自动保存用户偏好和事实,并在后续对话中注入 | 通用个人助手 |
| 会话状态记忆 | OpenAI Agents SDK Sessions、AutoGen Memory | 自动保存历史消息或把外部记忆注入上下文 | 线程连续对话 |
| 文件化项目记忆 | Claude Code `CLAUDE.md`、Claw 基座 memory files | 通过项目/用户级 Markdown 文件向 Agent 注入稳定规则 | 工程项目、团队约定 |
| 图/向量检索记忆 | LangGraph Store、Mem0、Zep/Graphiti | 抽取事实,存入向量库或知识图谱,按语义检索 | 长期、跨会话、海量事实 |
| Agent 自主管理记忆 | Letta / MemGPT | Agent 有显式 memory blocks 和 archival memory,可读写管理 | 状态型 Agent、长期角色 |
| 框架内置任务记忆 | CrewAI | 短期、长期、实体、上下文记忆组合 | 多 Agent 任务协作 |
ZK Data Agent 当前更接近:
```text
文件化项目记忆
+ 产品级用户记忆
+ Skill 作用域记忆
+ 异步记忆整理队列
```
它没有优先做向量库或知识图谱,而是选择 Markdown 文件作为最终记忆正文。这个取舍适合当前团队场景:记忆内容需要能被用户看到、编辑、删除,并且要按 Skill 作用域精准注入。
## 2. 主流实现机制
### 2.1 ChatGPT Memory:产品级个人长期记忆
ChatGPT Memory 的核心是平台级用户记忆。它会保存用户偏好、事实和历史对话中有持续价值的信息,并在后续对话中使用。用户可以查看、管理、删除保存的记忆,也可以关闭相关能力。
机制特点:
- 记忆作用域是用户账号。
- 由产品后台判断哪些内容值得保存。
- 注入方式对用户透明,用户看到的是“助手更了解我”。
- 适合通用个人助手,不适合表达复杂业务流程结构。
和我们的关系:
ZK Data Agent 的“用户记忆”借鉴了这个方向,但没有把全部记忆做成黑盒。我们把最终正文落到 `user.md`,并在 UI 里允许用户编辑。
### 2.2 OpenAI Agents SDK Sessions:会话状态记忆
OpenAI Agents SDK 的 Sessions 主要解决“同一个会话线程里自动保留历史上下文”。开发者不需要每轮手动传入完整历史,Session 会保存对话项,并在下一轮运行时自动带上。
机制特点:
- 更偏 conversation state,而不是长期个人偏好。
- 适合多轮会话连续执行。
- 常见实现是 SQLite / SQLAlchemy / 自定义 session backend。
- 记忆对象主要是消息历史,不是抽象后的长期知识。
和我们的关系:
ZK Data Agent 也有 session 持久化,但我们把它和“长期记忆”分开:
```text
session.json
保存当前会话消息、工具调用、产物和运行事件。
memory/user.md、memory/skills/*.md
保存跨会话长期偏好和 Skill 使用经验。
```
这个区分很重要:会话历史服务“恢复当前任务”,长期记忆服务“下次任务更懂用户和业务”。
### 2.3 Claude Code / OpenClaw / Claw:文件化项目记忆
Claude Code 使用 `CLAUDE.md` 作为项目或用户级记忆文件,常用于保存仓库规则、构建命令、代码风格、项目约定等。OpenClaw / Claw 类 Code Agent 基座通常也会保留这条路线:从全局或工作目录发现记忆文件,并把内容注入上下文。
在当前仓库里,对应实现主要是:
```text
src/agent_context.py
src/session_memory_compact.py
```
其中 `agent_context.py` 负责发现全局和目录级 memory files`session_memory_compact.py` 负责会话压缩场景下的 session memory 摘要。
机制特点:
- 记忆是文本文件,天然可读、可版本化。
- 非常适合工程项目规则和团队约定。
- 注入通常按目录/项目作用域进行。
- 记忆更新更多依赖人工维护,而不是完全自动。
和我们的关系:
ZK Data Agent 继承了“文件化、可编辑、可解释”的优点,但把作用域进一步细分:
```text
user.md
用户级偏好和稳定习惯。
skills/<skill-name>.md
某个 Skill 的使用经验、踩坑、格式偏好和边界修正。
```
也就是说,我们不是只有“项目记忆”,而是增加了“Skill 记忆”这一层。
### 2.4 LangGraph:线程状态 + 长期 Memory Store
LangGraph 把 memory 分成 short-term memory 和 long-term memory。短期记忆通常跟 thread 绑定,用来维持一次会话;长期记忆通过 store 按 namespace 保存,可以跨 thread 召回。它还把长期记忆进一步拆成 semantic、episodic、procedural 等类型。
机制特点:
- thread state 解决会话内上下文。
- store 解决跨会话长期信息。
- 支持按 user id / namespace 组织记忆。
- 长期记忆可以由应用逻辑或 Agent 写入、搜索、更新。
和我们的关系:
ZK Data Agent 当前没有引入通用 Store / VectorStore,而是用文件系统和 SQLite 队列实现一个轻量版本:
```text
namespace = account_id + memory kind + skill_name
storage = Markdown files + SQLite queue
retrieval = user memory always considered, skill memory按启用 Skill 精准注入
```
这比 LangGraph Store 简单,但更直接服务我们当前的 Skill 工作台。
### 2.5 Mem0:独立记忆层
Mem0 更像一个独立 memory layer。典型链路是:从对话中抽取事实,存入记忆系统;后续根据 query 检索相关记忆,再注入给模型。它强调 add / search / update / delete 这类记忆 API,也支持面向用户、Agent、session 等维度组织。
机制特点:
- 记忆层和 Agent 框架解耦。
- 常见存储后端是向量、图或混合检索。
- 强调自动抽取、去重、更新和语义召回。
- 适合大规模个性化 Agent 或跨应用记忆服务。
和我们的关系:
ZK Data Agent 目前没有把记忆做成独立检索服务。原因是我们的高频需求不是“从海量事实里语义搜索”,而是“把少量稳定经验准确注入到对应 Skill”。如果未来 Skill 记忆膨胀,可以在 Markdown 之外增加 Mem0 类似的检索层。
### 2.6 Letta / MemGPTAgent 自主管理内存
Letta 延续 MemGPT 思路,把 Agent 看成有长期状态的主体。它通常区分 core memory 和 archival memorycore memory 是短小、常驻上下文的重要信息;archival memory 是更大的外部记忆空间,Agent 可以通过工具读写。
机制特点:
- Agent 可以主动管理自己的记忆。
- core memory 常驻,archival memory 需要检索。
- 适合长期角色 Agent、个人助理、需要自我状态连续性的 Agent。
- 复杂度更高,需要更强的记忆写入约束和审计。
和我们的关系:
ZK Data Agent 没有让主 Agent 在执行链路里自由修改记忆。我们把记忆写入放到后台 worker,避免主任务因为记忆整理变慢或出错。这是一个更保守的团队平台取舍。
### 2.7 Zep / Graphiti:时间感知知识图谱记忆
Zep / Graphiti 代表的是 temporal knowledge graph 路线:从对话或事件中抽取实体和关系,形成带时间属性的知识图谱。它解决的问题不是简单偏好记忆,而是“事实如何随时间变化”“实体关系如何演进”。
机制特点:
- 记忆结构是实体、关系、事件、时间。
- 适合复杂事实网络和时间演化。
- 检索结果可以包含关系路径和上下文。
- 实现成本和运维复杂度高于 Markdown 或向量检索。
和我们的关系:
标签边界、业务规则、Skill 使用经验目前更适合文本化规则,不一定需要图谱。但如果未来要做“用户、Skill、数据集、标签、错误类型、修复策略”之间的关系分析,图谱路线会有价值。
### 2.8 CrewAI:多 Agent 任务记忆
CrewAI 的记忆体系主要服务多 Agent 协作,通常包含 short-term memory、long-term memory、entity memory 和 contextual memory。它关注的是任务过程中多个 Agent 如何共享上下文和持续改进。
机制特点:
- 和 Crew / Agent / Task 结构绑定。
- 强调任务协作过程中的上下文复用。
- 对实体、任务经验有独立组织方式。
和我们的关系:
ZK Data Agent 当前不是多 Agent 编排优先,而是单个工作台 Agent + Skill 能力包优先。Skill 记忆在某种程度上承担了“任务经验记忆”的角色。
### 2.9 AutoGenMemory 组件注入上下文
AutoGen 的 AgentChat 提供 Memory 抽象,可以把 list memory、vector memory 等组件挂到 AssistantAgent 上。运行时 Memory 会根据消息更新上下文,或把检索结果添加到模型输入。
机制特点:
- Memory 是 Agent 可插拔组件。
- 可以使用简单列表,也可以接向量检索。
- 更偏框架扩展点,而不是产品级记忆管理 UI。
和我们的关系:
ZK Data Agent 的记忆也可以理解为一个可插拔上下文组件,但我们额外做了用户 UI、Skill 作用域和后台队列。
## 3. ZK Data Agent 当前实现
实现入口:
```text
src/personal_memory.py
backend/api/server.py
frontend/app/components/assistant-ui/threadlist-sidebar.tsx
```
### 3.1 存储结构
每个账号有独立记忆目录:
```text
.port_sessions/accounts/<account_id>/memory/
user.md
skills/
<skill-name>.md
memory.db
```
其中:
- `user.md`:用户级长期记忆。
- `skills/<skill>.md`:某个 Skill 的使用记忆。
- `memory.db`:事件队列、状态和 revision 账本。
### 3.2 注入逻辑
模型调用前,后端调用:
```text
memory_manager.render_injection(account_id, enabled_skill_names)
```
注入规则:
```text
用户记忆
账号级,作为长期偏好注入。
Skill 使用记忆
只读取当前启用 Skill 对应的 skills/<skill>.md。
冲突优先级
用户本轮明确要求 > 个性化记忆。
```
这避免了一个常见问题:所有记忆都无差别注入导致上下文污染。
### 3.3 生成时机
每次交互结束后,后端调用:
```text
memory_manager.enqueue_interaction(...)
```
系统不会每轮同步整理记忆,而是先检测信号:
```text
显式记忆词:
记住、以后、下次、默认、总是、不要、应该、固定
纠错词:
不对、不是这样、格式错、之前说过、还是不行
Skill 经验:
skill、工具、流程、格式
工具经验:
模型返回的工具参数不是合法 JSON
```
命中后写入 SQLite pending 队列。显式记忆优先级更高。
### 3.4 异步整理
后台 worker 每 5 秒扫描账号事件,每次最多处理 8 条 pending event
```text
pending -> processing -> done / failed
```
整理方式:
1. 读取已有 `user.md` 和相关 `skills/<skill>.md`
2. 把一批事件交给模型做“整理式合并”。
3. 模型必须输出 JSON
```json
{
"user_memory": "完整 Markdown 或空字符串",
"skill_memories": {
"skill-name": "完整 Markdown"
}
}
```
4. 如果模型输出不可解析,则走 fallback 规则。
5. 写入 Markdown 文件,并更新 revision。
### 3.5 用户可编辑
前端左下角“记忆”入口支持:
- 查看用户记忆行数。
- 查看 Skill 记忆列表。
- 编辑用户记忆。
- 编辑某个 Skill 记忆。
- 查看记忆队列状态。
管理后台只看队列、用量等统计,不展示其他用户具体记忆内容。
## 4. 对比表
| 维度 | ChatGPT | Claude Code / OpenClaw | LangGraph / Mem0 / Zep | Letta | ZK Data Agent |
|------|---------|--------------------|-------------------------|-------|---------------|
| 主要目标 | 个人助手个性化 | 项目规则注入 | 长期检索记忆 | 状态型长期 Agent | 团队 Skill 工作台 |
| 记忆粒度 | 用户 | 用户/项目/目录 | 用户/线程/实体/namespace | Agent memory block | 用户 + Skill |
| 存储形态 | 平台内部 | Markdown 文件 | Store / 向量 / 图 | Core + archival memory | Markdown + SQLite queue |
| 生成时机 | 产品后台自动 | 多为人工维护 | 自动抽取 / API 写入 | Agent 主动管理 | 交互结束后异步整理 |
| 检索方式 | 平台决定 | 直接注入文件 | 语义搜索 / 图检索 | Core 常驻 + archival 检索 | 用户记忆 + 当前 Skill 记忆注入 |
| 可编辑性 | 用户可管理 | 文件可编辑 | 取决于产品/API | 通常需要工具/API | UI 可编辑 Markdown |
| 适合业务流程沉淀 | 中 | 中 | 高,但工程复杂 | 高,但复杂 | 高,且轻量 |
| 风险 | 黑盒、难按业务作用域隔离 | 容易依赖人工维护 | 检索和更新复杂 | 主链路复杂度高 | 暂无语义召回和图谱能力 |
## 5. 为什么当前方案适合我们
### 5.1 我们需要的是 Skill 使用经验,而不只是用户偏好
通用记忆多关注“用户是谁、用户喜欢什么”。我们的高频需求更像:
```text
product-data 生成数据时,用户偏好什么确认流程?
标签大师判断时,哪些边界经常被纠正?
online-mining-v2 查询线上日志时,哪些字段和表更稳定?
某个 Skill 写文件时,模型容易踩什么坑?
```
这些经验天然和 Skill 绑定。因此 `skills/<skill>.md` 比单一用户记忆更准确。
### 5.2 我们需要可审计、可编辑,而不是完全黑盒
团队平台里,记忆不能只存在模型或向量库内部。用户需要能看到:
- 记住了什么。
- 为什么下一次会注入。
- 哪里可以手动修改。
- 哪些记忆是用户级,哪些是 Skill 级。
Markdown 文件在这点上比纯向量库更直接。
### 5.3 主链路不能被记忆整理拖慢
数据生成、线上挖掘、标签判断本身就是长任务。记忆整理如果同步放在主链路里,会增加延迟和失败面。
当前设计是:
```text
主链路:只读取已有记忆 + 入队事件
后台:异步整理、合并、失败重试/记录
```
这和团队生产工具的稳定性要求更匹配。
### 5.4 Skill 作用域注入能降低上下文污染
如果所有记忆每次都注入,模型会被无关偏好干扰。当前只注入启用 Skill 的记忆:
```text
启用 product-data -> 注入 product-data 使用记忆
启用 label-master -> 注入 label-master 使用记忆
未启用某 Skill -> 不注入该 Skill 记忆
```
这使记忆更像“能力使用手册的增量补丁”,而不是一坨全局上下文。
## 6. 当前不足和后续方向
### 6.1 缺少语义召回
当前 Skill 记忆是按 Skill 文件整体注入,不做向量检索。如果某个 Skill 记忆变得很长,可能需要:
- 按章节拆分。
- 引入轻量 embedding 检索。
- 只注入和当前 query 相关的片段。
### 6.2 缺少结构化 schema
Markdown 易编辑,但不方便做强约束。后续可以让 Skill 记忆同时存在:
```text
human.md
structured.json
```
其中 Markdown 给人看,JSON 给程序做筛选和校验。
### 6.3 缺少记忆质量评估
目前能看到队列状态,但还没有系统评估:
- 哪些记忆被注入。
- 注入后是否减少纠错。
- 哪些记忆过期。
- 哪些 Skill 记忆导致误导。
后续可以把 memory revision 与 session outcome 关联起来。
### 6.4 缺少跨用户团队记忆
当前是账号级记忆。团队共性经验仍主要沉淀在 Skill 本体里。未来可以区分:
```text
个人 Skill 记忆
某个用户自己的偏好和使用习惯。
团队 Skill 记忆
多人使用后沉淀的稳定经验,经 review 后合入 Skill。
```
这样可以形成从“个人经验”到“团队 Skill 知识”的晋升路径。
## 7. 建议的技术路线
短期保持当前架构:
```text
Markdown 可编辑记忆
SQLite 异步队列
按 Skill 注入
UI 可查看可修改
后台可观测队列
```
中期增强:
```text
记忆片段化
记忆注入日志
过期/冲突检测
记忆质量指标
```
长期可选:
```text
向量检索:解决 Skill 记忆膨胀后的相关片段召回。
图谱记忆:解决用户、Skill、标签、数据集、错误类型之间的关系分析。
团队记忆晋升:把多用户共性 Skill 经验 review 后写回 Git Skill。
```
## 8. 资料来源
- OpenAI HelpChatGPT Memory FAQ
https://help.openai.com/en/articles/8590148-memory-faq
- OpenAI Agents SDKSessions
https://openai.github.io/openai-agents-python/sessions/
- Anthropic Claude CodeMemory
https://docs.anthropic.com/en/docs/claude-code/memory
- LangChain / LangGraphMemory concepts
https://docs.langchain.com/oss/python/concepts/memory
- Mem0 documentation
https://docs.mem0.ai/
- Letta documentation
https://docs.letta.com/
- Zep / Graphiti documentation
https://help.getzep.com/
- CrewAI Memory concepts
https://docs.crewai.com/concepts/memory
- Microsoft AutoGen AgentChat Memory
https://microsoft.github.io/autogen/dev/user-guide/agentchat-user-guide/memory.html
- ZK Data Agent 当前实现
`src/personal_memory.py``docs/technical-architecture/05-workspace-memory-observability.md`
@@ -0,0 +1,663 @@
# Workspace Runtime 设计稿
## 背景
当前项目已经有了平台账号、会话目录、Jupyter 远程工作区、Skill/Tools 和文件产物管理,但这些能力还没有被一个统一的“执行环境”概念串起来。
现在的问题不是单纯缺少登录账号,而是需要回答:
```text
谁在使用 Agent
-> 当前会话绑定到哪个工作区
-> 工具以什么身份、在什么目录、用什么权限执行
-> 产物在哪里保存、展示和下载
```
因此,账户体系升级不应该只看账号密码,而应该引入 `Workspace Runtime` 作为平台账号和工具执行之间的核心抽象。
## 核心结论
账户体系分两层:
```text
平台账号 Account
负责登录、角色、会话、Skill 配置、模型配置、记忆和集成状态。
工作区运行时 Workspace Runtime
负责执行身份、工作目录、文件读写、Python 环境、远程连接和进程管理。
```
平台账号不直接等价于 Linux 账号,也不直接等价于 Jupyter 账号。平台账号可以绑定不同类型的 runtime。
## 目标
1. 统一本机工作区、Linux 子账户工作区、Jupyter 工作区和未来 SSH 工作区。
2. 让 Tool handler 不关心执行位置,只面向统一 runtime 执行。
3. 明确权限来源,避免把远程工作区误认为平台托管沙盒。
4. 让每个 session 的输入、输出、scratchpad、Python 环境和文件下载有稳定归属。
5. 为后续多用户、资源限制、审计、团队空间和远程执行打基础。
## 非目标
1. 不在第一阶段实现完整企业 SSO。
2. 不把所有账号体系直接迁移到 Linux PAM。
3. 不强制所有远程工作区都变成平台托管沙盒。
4. 不要求 Skill 感知 runtime 的具体实现细节。
## 对象模型
### Account
平台账号是 Web 产品层的身份。
```text
Account
id
username
display_name
role
status
created_at
updated_at
```
职责:
- 登录和会话 token。
- 模型选择。
- Skill 启用状态。
- 用户记忆。
- 第三方集成状态。
- 默认 workspace runtime 策略。
### Session
Session 是一次 Agent 对话任务。
```text
Session
id
account_id
runtime_id
title
status
created_at
updated_at
```
职责:
- 保存对话历史。
- 绑定一个 runtime。
- 保存工具调用、活动步骤和最终结果。
- 关联输入文件和输出 artifact。
Session 一旦绑定远程 runtime,刷新页面后也应该恢复到同一个 runtime。
### Workspace Runtime
Workspace Runtime 是工具执行的真实环境。
```text
WorkspaceRuntime
id
account_id
session_id
type
root
permissions_source
status
created_at
updated_at
```
`type` 可以是:
```text
local_process
local_linux_user
remote_jupyter
remote_ssh
```
职责:
- 决定 bash/python/file 工具在哪里执行。
- 决定输入输出文件在哪里。
- 决定 Python 环境在哪里。
- 决定进程如何启动、停止和清理。
- 决定文件如何展示、下载和转在线文档。
### Artifact
Artifact 是输入和输出文件的统一抽象。
```text
Artifact
id
account_id
session_id
runtime_id
kind: input | output | scratchpad
uri
name
size
mime
created_at
```
`uri` 可以是:
```text
file:///home/<account_id>/zk-agent/sessions/<session_id>/output/a.jsonl
jupyter://<session_id>/root/zk_agent_workspaces/<session_id>/output/a.jsonl
ssh://<runtime_id>/home/user/zk_agent_workspaces/<session_id>/output/a.jsonl
```
文件列表只需要展示 metadata。下载或转在线文档时,再通过 runtime 拉取内容。
### Executor
Executor 是 Tool handler 和 Runtime 之间的执行适配层。
```text
Executor
run_bash(command, cwd, timeout)
run_python(code_or_file, cwd, timeout)
read_file(path)
write_file(path, content)
list_files(path)
open_file_stream(path)
cancel(run_id)
```
Tool handler 不应该自己判断是在本地、Jupyter 还是 SSH。它只调用当前 session 的 executor。
## Runtime 类型
### local_process
当前已有的默认模式。工具在服务进程所在机器上执行,目录由平台约定。
```text
.port_sessions/accounts/<account_id>/sessions/<session_id>/
```
适合:
- 本地开发。
- 单用户调试。
- 早期兼容。
问题:
- 多用户隔离主要靠代码路径约束。
- 工具进程和平台服务权限一致,风险较高。
### local_linux_user
平台托管的标准多用户工作区。
```text
平台账号: banisherwy
Linux runtime user: banisherwy
workspace root: /home/banisherwy/zk-agent/sessions/<session_id>
```
服务进程可以是 root,工具进程切换到普通 Linux 用户执行。
```text
root backend
-> runuser -u banisherwy -- <runner command>
```
职责分工:
```text
root 服务
创建 runtime 用户
初始化 workspace
设置 owner 和权限
启停进程
管理平台账号和 session
普通 Linux 用户
执行 bash/python
拥有自己的 workspace
拥有自己的 Python 虚拟环境
只能写自己的目录
```
推荐目录:
```text
/home/<account_id>/zk-agent/
sessions/
<session_id>/
input/
output/
scratchpad/
session.json
python/
.venv/
memory/
integrations/
```
推荐约定:
```text
平台用户名 = Linux 用户名
平台密码 = Linux 用户密码
Linux 用户允许 SSH 登录
```
这样用户体验更直接:
- 在平台创建账号时,同步创建同名 Linux 用户。
- 用户可以使用同一套账号密码登录 Web 平台和 SSH。
- Agent 工具执行时也使用同一个 Linux 用户身份。
- 文件 owner、进程 owner、SSH 登录用户和平台用户名一致,便于排查和审计。
但两者在架构语义上仍然保留分层:
```text
平台账号体系
登录、角色、session、Skill、模型配置。
Linux 用户体系
执行隔离、文件权限、进程权限、资源限制。
```
也就是说,账号名和密码保持一致,但平台仍然保留自己的登录态、session、角色和配置管理。Linux 账号负责机器级登录和执行权限。
需要注意:
- 用户名必须同时满足平台账号规范和 Linux 用户名规范。
- 修改平台密码时必须同步修改 Linux 密码。
- 禁用平台账号时,也应该禁用 Linux 登录或锁定 Linux 用户。
- 删除平台账号时,需要明确是否保留 `/home/<account_id>/zk-agent/` 数据。
- root 服务创建用户和改密码时必须走受控 helper,不能把用户输入拼成 shell 命令。
### remote_jupyter
用户授权的远程工作区。
语义是:
```text
用户把自己已有权限的 Jupyter 环境接入平台。
平台代替用户在这个环境里执行。
```
这不是平台托管沙盒。权限边界来自用户提供的 Jupyter 凭证。
```text
Account: banisherwy
Session: xxx
Runtime: remote_jupyter
Root: /root/zk_agent_workspaces/<session_id>
Permissions source: Jupyter password/token 对应的远程用户权限
```
平台需要保证:
- Jupyter 凭证只绑定当前 account/session。
- 刷新后 runtime 状态可恢复。
- 文件列表 metadata-only。
- 下载时通过 Jupyter API 流式拉取。
- 转在线文档时按需拉取,不默认同步大文件。
- 用户明确知道 Agent 在远程环境里的权限等同于该 Jupyter 用户。
平台不能保证:
- 远程机器上的文件权限隔离。
- 远程 Jupyter 用户不是 root。
- 远程挂载目录的访问范围。
短期建议:`remote_jupyter` 先保持当前逻辑,不作为账户体系升级的主战场。
当前已经具备:
- session 级 Jupyter 绑定。
- 刷新后恢复远程工作区状态。
- 输出文件 metadata-only 展示。
- 下载时通过 Jupyter API 流式读取。
- 转在线文档时按需拉取。
因此下一步账户体系升级优先处理本机托管 runtime 和平台账号,不主动重构 Jupyter 执行链路。后续只需要让 Jupyter 工作区在概念上挂到 `WorkspaceRuntime` 模型下。
### remote_ssh
未来可扩展的用户授权远程工作区。
语义和 remote_jupyter 类似:
```text
用户提供 SSH 连接能力。
平台代替用户在远程机器上执行。
权限边界来自 SSH 凭证对应的远程用户。
```
remote_ssh 更适合:
- 远程机器没有 Jupyter。
- 需要更完整 shell 能力。
- 需要使用远程开发机的挂载盘、GPU、模型目录。
但它也更复杂:
- SSH 凭证管理。
- 长连接和心跳。
- relay / OTP / 扫码登录。
- 文件传输和断线恢复。
- 进程树管理。
因此优先级应低于 `local_linux_user` 和已有 `remote_jupyter`
## 权限边界
需要在 UI 和文档中明确区分两类工作区:
```text
平台托管工作区
平台负责权限隔离。
典型类型: local_linux_user。
用户授权工作区
用户提供凭证。
平台不创建权限边界,只复用用户已有权限。
典型类型: remote_jupyter, remote_ssh。
```
UI 可以显示:
```text
当前工作区:Jupyter 远程工作区
权限来源:用户提供的 Jupyter 凭证
Agent 权限:等同于该远程环境当前登录用户
```
或者:
```text
当前工作区:平台托管工作区
执行身份:banisherwy
Agent 权限:普通 Linux 用户权限
```
## Tool 调用关系
目标关系:
```text
Agent Loop
-> Tool handler
-> RuntimeResolver(session_id)
-> Executor
-> local process / linux user / jupyter / ssh
```
工具不应该散落处理路径和远程协议。
例如:
```text
python_exec
-> executor.run_python(...)
write_file
-> executor.write_file(...)
download_artifact
-> executor.open_file_stream(...)
```
这样后续新增 runtime 时,尽量只新增 executor,不重写每个工具。
## 文件策略
### 输入文件
输入文件应该同步到当前 runtime 的 `input/`
```text
local_linux_user
上传文件 -> /home/<account_id>/zk-agent/sessions/<session_id>/input/
remote_jupyter
上传文件 -> 通过 Jupyter API 写入 /root/zk_agent_workspaces/<session_id>/input/
```
### 输出文件
输出文件默认放到当前 runtime 的 `output/`
```text
output/
records.jsonl
report.md
samples.csv
```
对远程 runtime,平台只保存 metadata。
```text
name
size
mtime
uri
runtime_id
```
点击下载时再流式读取。点击转在线文档时再按需拉取,并设置大小限制。
## Python 环境策略
每个 runtime 应有自己的 Python 环境。
```text
local_linux_user
/home/<account_id>/zk-agent/python/.venv
remote_jupyter
/root/zk_agent_workspaces/.zk-agent-python/.venv
```
初始化时只做最小准备:
- 创建 venv。
- 配置 pip 源。
- 不预装大量包。
缺包时由 Agent 根据任务安装,安装也发生在当前 runtime 内。
## 进程管理
每个工具执行必须有 run id 和 process group。
```text
run_id
account_id
session_id
runtime_id
executor_pid 或 remote_execution_id
status
started_at
updated_at
```
停止任务时:
- local_process:杀本地进程组。
- local_linux_user:杀对应 runtime 用户下该 run 的进程组。
- remote_jupyter:中断 kernel 或关闭对应执行任务。
- remote_ssh:杀远程进程组。
不能只停止 Web 请求,否则会出现“前端以为停了,后台 Python 还在跑”的问题。
## 持久化建议
建议把当前 JSON 账号体系逐步迁到 SQLite。
第一阶段可新增这些表:
```text
accounts
id
username
password_hash
role
status
created_at
updated_at
account_sessions
token_hash
account_id
created_at
updated_at
expires_at
workspace_runtimes
id
account_id
session_id
type
root
status
config_json
created_at
updated_at
artifacts
id
account_id
session_id
runtime_id
kind
uri
name
size
mime
created_at
```
敏感信息不要直接明文落库。Jupyter 密码、SSH key、token 至少需要加密或放入受控 secret store。
## 与现有实现的关系
当前已有能力可以映射到新模型:
```text
frontend/app/lib/claw-auth.ts
Account 登录态原型。
.port_sessions/accounts/<account_id>
local_process 模式下的 account workspace。
backend/api/server.py::account_paths
Runtime path resolver 的雏形。
src/jupyter_runtime.py
remote_jupyter executor 的雏形。
RunManager / RunStateStore
run id、活动状态、停止任务的雏形。
frontend 文件面板
Artifact list/download 的雏形。
```
所以这不是推翻重来,而是把已有能力抽象成更稳定的边界。
## 演进路线
### Phase 0:明确概念,不改执行路径
- 在代码和文档中引入 Workspace Runtime 术语。
- 把现有 `.port_sessions/accounts/<account_id>` 视为 `local_process` runtime。
- UI 显示当前工作区类型。
- 对 Jupyter 工作区补充权限提示。
### Phase 1:抽象 RuntimeResolver 和 Executor
- 新增 `RuntimeResolver`,根据 account/session 找当前 runtime。
- 新增统一 `Executor` 接口。
- 先把 `python_exec``bash`、文件工具迁到 executor。
- 保持现有 local 和 Jupyter 行为不变。
### Phase 2:账号存储升级
-`users.json``auth_sessions.json` 迁到 SQLite。
- 增加 `account_id``role``status``expires_at`
- 增加 session token 清理。
- 管理后台去掉 `admin/admin` 和默认 `123456`
### Phase 3local_linux_user runtime
- root 服务创建与平台账号同名的 Linux 用户。
- 平台密码同步设置为 Linux 用户密码。
- Linux 用户允许 SSH 登录。
- 初始化 `/home/<account_id>/zk-agent/`
- 工具执行切到普通 Linux 用户。
- Python venv、session、output 全部进入用户 home。
- 停止任务时按 process group 清理。
### Phase 4:资源限制和审计
- ulimit / cgroup。
- 每账号磁盘 quota。
- 工具执行审计。
- 大文件下载限流。
- session/output 清理策略。
### Phase 5remote_ssh runtime
- 在 remote_jupyter 稳定后再考虑。
- 重点解决认证、relay、长连接、文件传输和远程进程清理。
## 关键待决问题
1. 平台账号是否允许用户自注册,还是只允许管理员创建?
2. 用户自注册时,是否允许自动创建同名 Linux 用户?
3. 删除账号时,是否删除 Linux 用户,是否保留 home 目录?
4. 本机平台托管 workspace 是否统一迁到 `/home/<account_id>/zk-agent/`
5. Jupyter 凭证如何加密保存?
6. 远程 workspace 产物保留多久?
7. 大文件下载、在线文档转换和文件预览的大小限制是多少?
8. 是否需要团队空间:一个 workspace 被多个账号共享?
## 推荐决策
短期建议:
```text
保留平台账号体系。
引入 Workspace Runtime 抽象。
继续稳定 remote_jupyter。
账号存储从 JSON 迁到 SQLite。
开始设计 local_linux_user,但不要立即替换所有执行路径。
```
中期建议:
```text
服务可以 root 运行。
平台账号创建时同步创建同名普通 Linux 用户。
平台密码和 Linux 密码保持一致。
Linux 用户允许 SSH 登录。
工具执行统一通过 runtime executor。
本机默认工作区逐步迁到 /home/<account_id>/zk-agent。
```
长期建议:
```text
平台账号负责产品身份。
Workspace Runtime 负责执行环境。
Artifact 负责跨 runtime 文件抽象。
Executor 负责工具执行适配。
```
这样账户体系、Linux 子账户、Jupyter/SSH 远程工作区、文件下载、Python 环境和工具执行可以合到一个统一设计里,而不是继续各自生长。
+1
View File
@@ -14,6 +14,7 @@
7. [online-mining-v2 Skill 实现](07-online-mining-v2.md)
8. [label-master Skill 实现](08-label-master.md)
9. [外部系统 Skill:ELK、SQL、模型迭代](09-external-skills.md)
10. [Agent 记忆机制调研与对比](10-memory-research.md)
## 一句话定位
+5 -1
View File
@@ -11,6 +11,7 @@ import {
accountSessionOutputRoot,
accountSessionRoot,
accountSessionScratchpadRoot,
chownAccountPath,
getCurrentAccount,
} from "@/lib/claw-auth";
@@ -334,14 +335,16 @@ async function getLastUserText(
}
async function ensureSessionDirectories(accountId: string, sessionId: string) {
const sessionRoot = accountSessionRoot(accountId, sessionId);
await Promise.all([
mkdir(accountSessionRoot(accountId, sessionId), { recursive: true }),
mkdir(sessionRoot, { recursive: true }),
mkdir(accountSessionInputRoot(accountId, sessionId), { recursive: true }),
mkdir(accountSessionOutputRoot(accountId, sessionId), { recursive: true }),
mkdir(accountSessionScratchpadRoot(accountId, sessionId), {
recursive: true,
}),
]);
await chownAccountPath(accountId, sessionRoot, true);
}
function renderSessionRuntimeContext(accountId: string, sessionId: string) {
@@ -394,6 +397,7 @@ async function saveFilePart(
await mkdir(uploadDir, { recursive: true });
const filePath = path.join(uploadDir, `${Date.now()}-${filename}`);
await writeFile(filePath, bytes);
await chownAccountPath(accountId, filePath);
return [
`[文件已上传] ${filename}`,
+5
View File
@@ -51,6 +51,11 @@ const technicalDocs = {
file: "09-external-skills.md",
image: "/doc-assets/technical/09-external-skills.png",
},
"10-memory-research": {
title: "Agent 记忆机制调研与对比",
file: "10-memory-research.md",
image: null,
},
} as const;
type TechnicalDocSlug = keyof typeof technicalDocs;
+117 -15
View File
@@ -1,17 +1,19 @@
import { spawn } from "node:child_process";
import { randomBytes, scryptSync, timingSafeEqual } from "node:crypto";
import { mkdir, readFile, writeFile } from "node:fs/promises";
import path from "node:path";
import { cookies } from "next/headers";
const AUTH_ROOT = path.join(
const AUTH_STATE_ROOT = path.join(
/* turbopackIgnore: true */ process.cwd(),
"../../.port_sessions/accounts",
);
const USERS_PATH = path.join(AUTH_ROOT, "users.json");
const SESSIONS_PATH = path.join(AUTH_ROOT, "auth_sessions.json");
const USERS_PATH = path.join(AUTH_STATE_ROOT, "users.json");
const SESSIONS_PATH = path.join(AUTH_STATE_ROOT, "auth_sessions.json");
const COOKIE_NAME = "claw_account_session";
const SESSION_MAX_AGE_SECONDS = 60 * 60 * 24 * 30;
const SESSION_REFRESH_INTERVAL_MS = 60 * 60 * 12 * 1000;
const LINUX_ACCOUNT_WORKSPACE_NAME = "zk-agent";
type UserRecord = {
id: string;
@@ -51,6 +53,7 @@ export async function registerAccount(username: string, password: string) {
createdAt: new Date().toISOString(),
};
usersFile.users.push(account);
await ensureLinuxAccount(account.id, password);
await writeUsers(usersFile);
await createAccountDirectories(account.id);
return createSession(account);
@@ -65,6 +68,7 @@ export async function loginAccount(username: string, password: string) {
if (!account || !verifyPassword(password, account.passwordHash)) {
throw new Error("账号或密码错误");
}
await ensureLinuxAccount(account.id, password);
await createAccountDirectories(account.id);
return createSession(account);
}
@@ -101,15 +105,18 @@ export async function getCurrentAccount(): Promise<AccountSession | null> {
}
export function accountUploadRoot(accountId: string) {
return path.join(AUTH_ROOT, accountId, "sessions");
return path.join(accountBaseRoot(accountId), "sessions");
}
export function accountOutputRoot(accountId: string) {
return path.join(AUTH_ROOT, accountId, "sessions");
return path.join(accountBaseRoot(accountId), "sessions");
}
export function accountBaseRoot(accountId: string) {
return path.join(AUTH_ROOT, accountId);
if (linuxAccountsEnabled()) {
return path.join("/home", accountId, LINUX_ACCOUNT_WORKSPACE_NAME);
}
return path.join(AUTH_STATE_ROOT, accountId);
}
export function accountSessionInputRoot(accountId: string, sessionId: string) {
@@ -161,15 +168,18 @@ async function setSessionCookie(token: string) {
}
async function createAccountDirectories(accountId: string) {
await Promise.all(
["sessions"].map((name) =>
mkdir(path.join(AUTH_ROOT, accountId, name), { recursive: true }),
),
);
const base = accountBaseRoot(accountId);
await Promise.all([
mkdir(path.join(base, "sessions"), { recursive: true }),
mkdir(path.join(base, "python"), { recursive: true }),
mkdir(path.join(base, "memory"), { recursive: true }),
mkdir(path.join(base, "integrations"), { recursive: true }),
]);
await chownAccountPath(accountId, base, true);
}
async function readUsers(): Promise<UsersFile> {
await mkdir(AUTH_ROOT, { recursive: true });
await mkdir(AUTH_STATE_ROOT, { recursive: true });
try {
return JSON.parse(await readFile(USERS_PATH, "utf8")) as UsersFile;
} catch {
@@ -178,12 +188,12 @@ async function readUsers(): Promise<UsersFile> {
}
async function writeUsers(payload: UsersFile) {
await mkdir(AUTH_ROOT, { recursive: true });
await mkdir(AUTH_STATE_ROOT, { recursive: true });
await writeFile(USERS_PATH, JSON.stringify(payload, null, 2), "utf8");
}
async function readSessions(): Promise<SessionsFile> {
await mkdir(AUTH_ROOT, { recursive: true });
await mkdir(AUTH_STATE_ROOT, { recursive: true });
try {
return JSON.parse(await readFile(SESSIONS_PATH, "utf8")) as SessionsFile;
} catch {
@@ -192,12 +202,20 @@ async function readSessions(): Promise<SessionsFile> {
}
async function writeSessions(payload: SessionsFile) {
await mkdir(AUTH_ROOT, { recursive: true });
await mkdir(AUTH_STATE_ROOT, { recursive: true });
await writeFile(SESSIONS_PATH, JSON.stringify(payload, null, 2), "utf8");
}
function normalizeUsername(username: string) {
const cleanUsername = username.trim().toLowerCase();
if (linuxAccountsEnabled()) {
if (!/^[a-z_][a-z0-9_-]{1,31}$/.test(cleanUsername)) {
throw new Error(
"启用 Linux 账号时,账号只能包含小写字母、数字、下划线或中划线,且必须以小写字母或下划线开头,长度 2-32",
);
}
return cleanUsername;
}
if (!/^[a-z0-9._-]{2,40}$/.test(cleanUsername)) {
throw new Error(
"账号只能包含小写字母、数字、点、下划线或中划线,长度 2-40",
@@ -230,3 +248,87 @@ function shouldRefreshSession(value?: string) {
if (Number.isNaN(timestamp)) return true;
return Date.now() - timestamp > SESSION_REFRESH_INTERVAL_MS;
}
function linuxAccountsEnabled() {
return ["1", "true", "yes", "on"].includes(
(process.env.CLAW_ENABLE_LINUX_ACCOUNTS ?? "").trim().toLowerCase(),
);
}
async function ensureLinuxAccount(accountId: string, password: string) {
if (!linuxAccountsEnabled()) return;
if (process.platform !== "linux") {
throw new Error("CLAW_ENABLE_LINUX_ACCOUNTS=1 只支持 Linux 部署环境");
}
if (typeof process.getuid === "function" && process.getuid() !== 0) {
throw new Error(
"CLAW_ENABLE_LINUX_ACCOUNTS=1 需要 Web 服务以 root 身份运行",
);
}
await runCommand("id", ["-u", accountId], { allowFailure: true }).then(
async (result) => {
if (result.code === 0) return;
await runCommand("useradd", [
"-m",
"-d",
`/home/${accountId}`,
"-s",
"/bin/bash",
accountId,
]);
},
);
await runCommand("chpasswd", [], {
input: `${accountId}:${password}\n`,
});
}
export async function chownAccountPath(
accountId: string,
targetPath: string,
recursive = false,
) {
if (!linuxAccountsEnabled()) return;
await runCommand("chown", [
...(recursive ? ["-R"] : []),
accountId,
targetPath,
]);
}
function runCommand(
command: string,
args: string[],
options: { input?: string; allowFailure?: boolean } = {},
) {
return new Promise<{ code: number; stdout: string; stderr: string }>(
(resolve, reject) => {
const child = spawn(command, args, {
stdio: ["pipe", "pipe", "pipe"],
});
let stdout = "";
let stderr = "";
child.stdout.on("data", (chunk) => {
stdout += chunk.toString();
});
child.stderr.on("data", (chunk) => {
stderr += chunk.toString();
});
child.on("error", reject);
child.on("close", (code) => {
const exitCode = code ?? 1;
if (exitCode !== 0 && !options.allowFailure) {
reject(
new Error(
`${command} ${args.join(" ")} 失败:${stderr || stdout || exitCode}`,
),
);
return;
}
resolve({ code: exitCode, stdout, stderr });
});
if (options.input) child.stdin.write(options.input);
child.stdin.end();
},
);
}
@@ -0,0 +1,478 @@
# 10. Agent 记忆机制调研与 ZK Data Agent 对比
本文整理主流 Agent / AI 产品的记忆实现方式,并对照 ZK Data Agent 当前实现。目标不是判断哪一种“最好”,而是说明不同记忆机制分别解决什么问题,以及为什么我们当前选择“用户记忆 + Skill 使用记忆 + 异步整理队列 + Markdown 可编辑文件”的路线。
调研时间:2026-05-19。
## 1. 结论摘要
主流记忆实现大致分为六类:
| 类型 | 代表 | 核心做法 | 适合场景 |
|------|------|----------|----------|
| 产品级长期记忆 | ChatGPT Memory | 平台自动保存用户偏好和事实,并在后续对话中注入 | 通用个人助手 |
| 会话状态记忆 | OpenAI Agents SDK Sessions、AutoGen Memory | 自动保存历史消息或把外部记忆注入上下文 | 线程连续对话 |
| 文件化项目记忆 | Claude Code `CLAUDE.md`、Claw 基座 memory files | 通过项目/用户级 Markdown 文件向 Agent 注入稳定规则 | 工程项目、团队约定 |
| 图/向量检索记忆 | LangGraph Store、Mem0、Zep/Graphiti | 抽取事实,存入向量库或知识图谱,按语义检索 | 长期、跨会话、海量事实 |
| Agent 自主管理记忆 | Letta / MemGPT | Agent 有显式 memory blocks 和 archival memory,可读写管理 | 状态型 Agent、长期角色 |
| 框架内置任务记忆 | CrewAI | 短期、长期、实体、上下文记忆组合 | 多 Agent 任务协作 |
ZK Data Agent 当前更接近:
```text
文件化项目记忆
+ 产品级用户记忆
+ Skill 作用域记忆
+ 异步记忆整理队列
```
它没有优先做向量库或知识图谱,而是选择 Markdown 文件作为最终记忆正文。这个取舍适合当前团队场景:记忆内容需要能被用户看到、编辑、删除,并且要按 Skill 作用域精准注入。
## 2. 主流实现机制
### 2.1 ChatGPT Memory:产品级个人长期记忆
ChatGPT Memory 的核心是平台级用户记忆。它会保存用户偏好、事实和历史对话中有持续价值的信息,并在后续对话中使用。用户可以查看、管理、删除保存的记忆,也可以关闭相关能力。
机制特点:
- 记忆作用域是用户账号。
- 由产品后台判断哪些内容值得保存。
- 注入方式对用户透明,用户看到的是“助手更了解我”。
- 适合通用个人助手,不适合表达复杂业务流程结构。
和我们的关系:
ZK Data Agent 的“用户记忆”借鉴了这个方向,但没有把全部记忆做成黑盒。我们把最终正文落到 `user.md`,并在 UI 里允许用户编辑。
### 2.2 OpenAI Agents SDK Sessions:会话状态记忆
OpenAI Agents SDK 的 Sessions 主要解决“同一个会话线程里自动保留历史上下文”。开发者不需要每轮手动传入完整历史,Session 会保存对话项,并在下一轮运行时自动带上。
机制特点:
- 更偏 conversation state,而不是长期个人偏好。
- 适合多轮会话连续执行。
- 常见实现是 SQLite / SQLAlchemy / 自定义 session backend。
- 记忆对象主要是消息历史,不是抽象后的长期知识。
和我们的关系:
ZK Data Agent 也有 session 持久化,但我们把它和“长期记忆”分开:
```text
session.json
保存当前会话消息、工具调用、产物和运行事件。
memory/user.md、memory/skills/*.md
保存跨会话长期偏好和 Skill 使用经验。
```
这个区分很重要:会话历史服务“恢复当前任务”,长期记忆服务“下次任务更懂用户和业务”。
### 2.3 Claude Code / OpenClaw / Claw:文件化项目记忆
Claude Code 使用 `CLAUDE.md` 作为项目或用户级记忆文件,常用于保存仓库规则、构建命令、代码风格、项目约定等。OpenClaw / Claw 类 Code Agent 基座通常也会保留这条路线:从全局或工作目录发现记忆文件,并把内容注入上下文。
在当前仓库里,对应实现主要是:
```text
src/agent_context.py
src/session_memory_compact.py
```
其中 `agent_context.py` 负责发现全局和目录级 memory files`session_memory_compact.py` 负责会话压缩场景下的 session memory 摘要。
机制特点:
- 记忆是文本文件,天然可读、可版本化。
- 非常适合工程项目规则和团队约定。
- 注入通常按目录/项目作用域进行。
- 记忆更新更多依赖人工维护,而不是完全自动。
和我们的关系:
ZK Data Agent 继承了“文件化、可编辑、可解释”的优点,但把作用域进一步细分:
```text
user.md
用户级偏好和稳定习惯。
skills/<skill-name>.md
某个 Skill 的使用经验、踩坑、格式偏好和边界修正。
```
也就是说,我们不是只有“项目记忆”,而是增加了“Skill 记忆”这一层。
### 2.4 LangGraph:线程状态 + 长期 Memory Store
LangGraph 把 memory 分成 short-term memory 和 long-term memory。短期记忆通常跟 thread 绑定,用来维持一次会话;长期记忆通过 store 按 namespace 保存,可以跨 thread 召回。它还把长期记忆进一步拆成 semantic、episodic、procedural 等类型。
机制特点:
- thread state 解决会话内上下文。
- store 解决跨会话长期信息。
- 支持按 user id / namespace 组织记忆。
- 长期记忆可以由应用逻辑或 Agent 写入、搜索、更新。
和我们的关系:
ZK Data Agent 当前没有引入通用 Store / VectorStore,而是用文件系统和 SQLite 队列实现一个轻量版本:
```text
namespace = account_id + memory kind + skill_name
storage = Markdown files + SQLite queue
retrieval = user memory always considered, skill memory按启用 Skill 精准注入
```
这比 LangGraph Store 简单,但更直接服务我们当前的 Skill 工作台。
### 2.5 Mem0:独立记忆层
Mem0 更像一个独立 memory layer。典型链路是:从对话中抽取事实,存入记忆系统;后续根据 query 检索相关记忆,再注入给模型。它强调 add / search / update / delete 这类记忆 API,也支持面向用户、Agent、session 等维度组织。
机制特点:
- 记忆层和 Agent 框架解耦。
- 常见存储后端是向量、图或混合检索。
- 强调自动抽取、去重、更新和语义召回。
- 适合大规模个性化 Agent 或跨应用记忆服务。
和我们的关系:
ZK Data Agent 目前没有把记忆做成独立检索服务。原因是我们的高频需求不是“从海量事实里语义搜索”,而是“把少量稳定经验准确注入到对应 Skill”。如果未来 Skill 记忆膨胀,可以在 Markdown 之外增加 Mem0 类似的检索层。
### 2.6 Letta / MemGPTAgent 自主管理内存
Letta 延续 MemGPT 思路,把 Agent 看成有长期状态的主体。它通常区分 core memory 和 archival memorycore memory 是短小、常驻上下文的重要信息;archival memory 是更大的外部记忆空间,Agent 可以通过工具读写。
机制特点:
- Agent 可以主动管理自己的记忆。
- core memory 常驻,archival memory 需要检索。
- 适合长期角色 Agent、个人助理、需要自我状态连续性的 Agent。
- 复杂度更高,需要更强的记忆写入约束和审计。
和我们的关系:
ZK Data Agent 没有让主 Agent 在执行链路里自由修改记忆。我们把记忆写入放到后台 worker,避免主任务因为记忆整理变慢或出错。这是一个更保守的团队平台取舍。
### 2.7 Zep / Graphiti:时间感知知识图谱记忆
Zep / Graphiti 代表的是 temporal knowledge graph 路线:从对话或事件中抽取实体和关系,形成带时间属性的知识图谱。它解决的问题不是简单偏好记忆,而是“事实如何随时间变化”“实体关系如何演进”。
机制特点:
- 记忆结构是实体、关系、事件、时间。
- 适合复杂事实网络和时间演化。
- 检索结果可以包含关系路径和上下文。
- 实现成本和运维复杂度高于 Markdown 或向量检索。
和我们的关系:
标签边界、业务规则、Skill 使用经验目前更适合文本化规则,不一定需要图谱。但如果未来要做“用户、Skill、数据集、标签、错误类型、修复策略”之间的关系分析,图谱路线会有价值。
### 2.8 CrewAI:多 Agent 任务记忆
CrewAI 的记忆体系主要服务多 Agent 协作,通常包含 short-term memory、long-term memory、entity memory 和 contextual memory。它关注的是任务过程中多个 Agent 如何共享上下文和持续改进。
机制特点:
- 和 Crew / Agent / Task 结构绑定。
- 强调任务协作过程中的上下文复用。
- 对实体、任务经验有独立组织方式。
和我们的关系:
ZK Data Agent 当前不是多 Agent 编排优先,而是单个工作台 Agent + Skill 能力包优先。Skill 记忆在某种程度上承担了“任务经验记忆”的角色。
### 2.9 AutoGenMemory 组件注入上下文
AutoGen 的 AgentChat 提供 Memory 抽象,可以把 list memory、vector memory 等组件挂到 AssistantAgent 上。运行时 Memory 会根据消息更新上下文,或把检索结果添加到模型输入。
机制特点:
- Memory 是 Agent 可插拔组件。
- 可以使用简单列表,也可以接向量检索。
- 更偏框架扩展点,而不是产品级记忆管理 UI。
和我们的关系:
ZK Data Agent 的记忆也可以理解为一个可插拔上下文组件,但我们额外做了用户 UI、Skill 作用域和后台队列。
## 3. ZK Data Agent 当前实现
实现入口:
```text
src/personal_memory.py
backend/api/server.py
frontend/app/components/assistant-ui/threadlist-sidebar.tsx
```
### 3.1 存储结构
每个账号有独立记忆目录:
```text
.port_sessions/accounts/<account_id>/memory/
user.md
skills/
<skill-name>.md
memory.db
```
其中:
- `user.md`:用户级长期记忆。
- `skills/<skill>.md`:某个 Skill 的使用记忆。
- `memory.db`:事件队列、状态和 revision 账本。
### 3.2 注入逻辑
模型调用前,后端调用:
```text
memory_manager.render_injection(account_id, enabled_skill_names)
```
注入规则:
```text
用户记忆
账号级,作为长期偏好注入。
Skill 使用记忆
只读取当前启用 Skill 对应的 skills/<skill>.md。
冲突优先级
用户本轮明确要求 > 个性化记忆。
```
这避免了一个常见问题:所有记忆都无差别注入导致上下文污染。
### 3.3 生成时机
每次交互结束后,后端调用:
```text
memory_manager.enqueue_interaction(...)
```
系统不会每轮同步整理记忆,而是先检测信号:
```text
显式记忆词:
记住、以后、下次、默认、总是、不要、应该、固定
纠错词:
不对、不是这样、格式错、之前说过、还是不行
Skill 经验:
skill、工具、流程、格式
工具经验:
模型返回的工具参数不是合法 JSON
```
命中后写入 SQLite pending 队列。显式记忆优先级更高。
### 3.4 异步整理
后台 worker 每 5 秒扫描账号事件,每次最多处理 8 条 pending event
```text
pending -> processing -> done / failed
```
整理方式:
1. 读取已有 `user.md` 和相关 `skills/<skill>.md`
2. 把一批事件交给模型做“整理式合并”。
3. 模型必须输出 JSON
```json
{
"user_memory": "完整 Markdown 或空字符串",
"skill_memories": {
"skill-name": "完整 Markdown"
}
}
```
4. 如果模型输出不可解析,则走 fallback 规则。
5. 写入 Markdown 文件,并更新 revision。
### 3.5 用户可编辑
前端左下角“记忆”入口支持:
- 查看用户记忆行数。
- 查看 Skill 记忆列表。
- 编辑用户记忆。
- 编辑某个 Skill 记忆。
- 查看记忆队列状态。
管理后台只看队列、用量等统计,不展示其他用户具体记忆内容。
## 4. 对比表
| 维度 | ChatGPT | Claude Code / OpenClaw | LangGraph / Mem0 / Zep | Letta | ZK Data Agent |
|------|---------|--------------------|-------------------------|-------|---------------|
| 主要目标 | 个人助手个性化 | 项目规则注入 | 长期检索记忆 | 状态型长期 Agent | 团队 Skill 工作台 |
| 记忆粒度 | 用户 | 用户/项目/目录 | 用户/线程/实体/namespace | Agent memory block | 用户 + Skill |
| 存储形态 | 平台内部 | Markdown 文件 | Store / 向量 / 图 | Core + archival memory | Markdown + SQLite queue |
| 生成时机 | 产品后台自动 | 多为人工维护 | 自动抽取 / API 写入 | Agent 主动管理 | 交互结束后异步整理 |
| 检索方式 | 平台决定 | 直接注入文件 | 语义搜索 / 图检索 | Core 常驻 + archival 检索 | 用户记忆 + 当前 Skill 记忆注入 |
| 可编辑性 | 用户可管理 | 文件可编辑 | 取决于产品/API | 通常需要工具/API | UI 可编辑 Markdown |
| 适合业务流程沉淀 | 中 | 中 | 高,但工程复杂 | 高,但复杂 | 高,且轻量 |
| 风险 | 黑盒、难按业务作用域隔离 | 容易依赖人工维护 | 检索和更新复杂 | 主链路复杂度高 | 暂无语义召回和图谱能力 |
## 5. 为什么当前方案适合我们
### 5.1 我们需要的是 Skill 使用经验,而不只是用户偏好
通用记忆多关注“用户是谁、用户喜欢什么”。我们的高频需求更像:
```text
product-data 生成数据时,用户偏好什么确认流程?
标签大师判断时,哪些边界经常被纠正?
online-mining-v2 查询线上日志时,哪些字段和表更稳定?
某个 Skill 写文件时,模型容易踩什么坑?
```
这些经验天然和 Skill 绑定。因此 `skills/<skill>.md` 比单一用户记忆更准确。
### 5.2 我们需要可审计、可编辑,而不是完全黑盒
团队平台里,记忆不能只存在模型或向量库内部。用户需要能看到:
- 记住了什么。
- 为什么下一次会注入。
- 哪里可以手动修改。
- 哪些记忆是用户级,哪些是 Skill 级。
Markdown 文件在这点上比纯向量库更直接。
### 5.3 主链路不能被记忆整理拖慢
数据生成、线上挖掘、标签判断本身就是长任务。记忆整理如果同步放在主链路里,会增加延迟和失败面。
当前设计是:
```text
主链路:只读取已有记忆 + 入队事件
后台:异步整理、合并、失败重试/记录
```
这和团队生产工具的稳定性要求更匹配。
### 5.4 Skill 作用域注入能降低上下文污染
如果所有记忆每次都注入,模型会被无关偏好干扰。当前只注入启用 Skill 的记忆:
```text
启用 product-data -> 注入 product-data 使用记忆
启用 label-master -> 注入 label-master 使用记忆
未启用某 Skill -> 不注入该 Skill 记忆
```
这使记忆更像“能力使用手册的增量补丁”,而不是一坨全局上下文。
## 6. 当前不足和后续方向
### 6.1 缺少语义召回
当前 Skill 记忆是按 Skill 文件整体注入,不做向量检索。如果某个 Skill 记忆变得很长,可能需要:
- 按章节拆分。
- 引入轻量 embedding 检索。
- 只注入和当前 query 相关的片段。
### 6.2 缺少结构化 schema
Markdown 易编辑,但不方便做强约束。后续可以让 Skill 记忆同时存在:
```text
human.md
structured.json
```
其中 Markdown 给人看,JSON 给程序做筛选和校验。
### 6.3 缺少记忆质量评估
目前能看到队列状态,但还没有系统评估:
- 哪些记忆被注入。
- 注入后是否减少纠错。
- 哪些记忆过期。
- 哪些 Skill 记忆导致误导。
后续可以把 memory revision 与 session outcome 关联起来。
### 6.4 缺少跨用户团队记忆
当前是账号级记忆。团队共性经验仍主要沉淀在 Skill 本体里。未来可以区分:
```text
个人 Skill 记忆
某个用户自己的偏好和使用习惯。
团队 Skill 记忆
多人使用后沉淀的稳定经验,经 review 后合入 Skill。
```
这样可以形成从“个人经验”到“团队 Skill 知识”的晋升路径。
## 7. 建议的技术路线
短期保持当前架构:
```text
Markdown 可编辑记忆
SQLite 异步队列
按 Skill 注入
UI 可查看可修改
后台可观测队列
```
中期增强:
```text
记忆片段化
记忆注入日志
过期/冲突检测
记忆质量指标
```
长期可选:
```text
向量检索:解决 Skill 记忆膨胀后的相关片段召回。
图谱记忆:解决用户、Skill、标签、数据集、错误类型之间的关系分析。
团队记忆晋升:把多用户共性 Skill 经验 review 后写回 Git Skill。
```
## 8. 资料来源
- OpenAI HelpChatGPT Memory FAQ
https://help.openai.com/en/articles/8590148-memory-faq
- OpenAI Agents SDKSessions
https://openai.github.io/openai-agents-python/sessions/
- Anthropic Claude CodeMemory
https://docs.anthropic.com/en/docs/claude-code/memory
- LangChain / LangGraphMemory concepts
https://docs.langchain.com/oss/python/concepts/memory
- Mem0 documentation
https://docs.mem0.ai/
- Letta documentation
https://docs.letta.com/
- Zep / Graphiti documentation
https://help.getzep.com/
- CrewAI Memory concepts
https://docs.crewai.com/concepts/memory
- Microsoft AutoGen AgentChat Memory
https://microsoft.github.io/autogen/dev/user-guide/agentchat-user-guide/memory.html
- ZK Data Agent 当前实现
`src/personal_memory.py``docs/technical-architecture/05-workspace-memory-observability.md`
@@ -14,6 +14,7 @@
7. [online-mining-v2 Skill 实现](07-online-mining-v2.md)
8. [label-master Skill 实现](08-label-master.md)
9. [外部系统 Skill:ELK、SQL、模型迭代](09-external-skills.md)
10. [Agent 记忆机制调研与对比](10-memory-research.md)
## 一句话定位
+105 -11
View File
@@ -3,7 +3,7 @@ set -euo pipefail
# Ubuntu 一键部署/更新脚本。
# - 首次执行会交互式生成 .env.deploy,本机保存,不提交 git。
# - 后续执行会拉取代码、更新依赖、构建前端、安装/重启用户级 systemd 服务。
# - 后续执行会拉取代码、更新依赖、构建前端、安装/重启 systemd 服务。
ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
ENV_FILE="${ROOT_DIR}/.env.deploy"
@@ -12,14 +12,17 @@ DEPLOY_INSTANCE="${CLAW_DEPLOY_INSTANCE:-zk-data-agent}"
BACKEND_SERVICE="${CLAW_BACKEND_SERVICE:-${DEPLOY_INSTANCE}-backend}"
FRONTEND_SERVICE="${CLAW_FRONTEND_SERVICE:-${DEPLOY_INSTANCE}-frontend}"
USER_SYSTEMD_DIR="${HOME}/.config/systemd/user"
SYSTEM_SYSTEMD_DIR="/etc/systemd/system"
BRANCH=""
FORCE=0
SKIP_GIT=0
BOOTSTRAP_SYSTEM=0
SERVICE_SCOPE="${CLAW_SERVICE_SCOPE:-}"
ENABLE_LINUX_ACCOUNTS="${CLAW_ENABLE_LINUX_ACCOUNTS:-}"
usage() {
cat <<EOF
Usage: bash scripts/deploy-ubuntu.sh [branch] [--force] [--skip-git] [--bootstrap-system]
Usage: bash scripts/deploy-ubuntu.sh [branch] [--force] [--skip-git] [--bootstrap-system] [--system-service] [--user-service] [--enable-linux-accounts]
Options:
branch 要部署的 git 分支;不传则使用当前分支
@@ -27,6 +30,12 @@ Options:
--skip-git 跳过 git 拉取,只部署当前工作区
--bootstrap-system
强制使用 sudo 安装 Ubuntu 系统依赖;首次安装会默认执行
--system-service
安装 root/system 级 systemd 服务;以 root 执行时默认启用
--user-service
安装当前用户级 systemd 服务
--enable-linux-accounts
启用平台账号同名 Linux 用户工作区;要求 system 服务和 root 权限
EOF
}
@@ -44,6 +53,18 @@ while [[ $# -gt 0 ]]; do
BOOTSTRAP_SYSTEM=1
shift
;;
--system-service)
SERVICE_SCOPE="system"
shift
;;
--user-service)
SERVICE_SCOPE="user"
shift
;;
--enable-linux-accounts)
ENABLE_LINUX_ACCOUNTS="1"
shift
;;
-h|--help)
usage
exit 0
@@ -60,6 +81,9 @@ while [[ $# -gt 0 ]]; do
esac
done
REQUESTED_SERVICE_SCOPE="${SERVICE_SCOPE}"
REQUESTED_ENABLE_LINUX_ACCOUNTS="${ENABLE_LINUX_ACCOUNTS}"
log() {
printf "\n\033[1;34m==>\033[0m %s\n" "$1"
}
@@ -77,6 +101,16 @@ configure_instance_names() {
DEPLOY_INSTANCE="${CLAW_DEPLOY_INSTANCE:-${DEPLOY_INSTANCE:-zk-data-agent}}"
BACKEND_SERVICE="${CLAW_BACKEND_SERVICE:-${DEPLOY_INSTANCE}-backend}"
FRONTEND_SERVICE="${CLAW_FRONTEND_SERVICE:-${DEPLOY_INSTANCE}-frontend}"
if [[ -z "${SERVICE_SCOPE}" ]]; then
if [[ "${EUID}" == "0" ]]; then
SERVICE_SCOPE="system"
else
SERVICE_SCOPE="user"
fi
fi
if [[ -z "${ENABLE_LINUX_ACCOUNTS}" ]]; then
ENABLE_LINUX_ACCOUNTS="${CLAW_ENABLE_LINUX_ACCOUNTS:-0}"
fi
}
require_command() {
@@ -120,6 +154,9 @@ bootstrap_system_packages() {
libssl-dev zlib1g-dev libbz2-dev libreadline-dev libsqlite3-dev \
libffi-dev liblzma-dev libncursesw5-dev xz-utils tk-dev \
systemd
if [[ "${ENABLE_LINUX_ACCOUNTS}" == "1" ]]; then
sudo apt-get install -y passwd python3 python3-venv
fi
}
configure_nginx_upload_limit() {
@@ -180,15 +217,42 @@ export CLAW_BACKEND_PORT=$(printf "%q" "${backend_port}")
export CLAW_FRONTEND_HOST=$(printf "%q" "${frontend_host}")
export CLAW_FRONTEND_PORT=$(printf "%q" "${frontend_port}")
export CLAW_API_URL=$(printf "%q" "${claw_api_url}")
export CLAW_SERVICE_SCOPE=$(printf "%q" "${SERVICE_SCOPE}")
export CLAW_ENABLE_LINUX_ACCOUNTS=$(printf "%q" "${ENABLE_LINUX_ACCOUNTS:-0}")
EOF
chmod 600 "${ENV_FILE}"
}
upsert_env_export() {
local key="$1"
local value="$2"
local tmp_file
tmp_file="$(mktemp)"
grep -v -E "^export ${key}=" "${ENV_FILE}" >"${tmp_file}" || true
printf 'export %s=%s\n' "${key}" "$(printf "%q" "${value}")" >>"${tmp_file}"
install -m 0600 "${tmp_file}" "${ENV_FILE}"
rm -f "${tmp_file}"
}
ensure_env_file() {
if [[ -f "${ENV_FILE}" ]]; then
# shellcheck disable=SC1090
source "${ENV_FILE}"
if [[ -n "${REQUESTED_SERVICE_SCOPE}" ]]; then
SERVICE_SCOPE="${REQUESTED_SERVICE_SCOPE}"
export CLAW_SERVICE_SCOPE="${REQUESTED_SERVICE_SCOPE}"
fi
if [[ -n "${REQUESTED_ENABLE_LINUX_ACCOUNTS}" ]]; then
ENABLE_LINUX_ACCOUNTS="${REQUESTED_ENABLE_LINUX_ACCOUNTS}"
export CLAW_ENABLE_LINUX_ACCOUNTS="${REQUESTED_ENABLE_LINUX_ACCOUNTS}"
fi
configure_instance_names
if [[ -n "${REQUESTED_SERVICE_SCOPE}" ]]; then
upsert_env_export CLAW_SERVICE_SCOPE "${SERVICE_SCOPE}"
fi
if [[ -n "${REQUESTED_ENABLE_LINUX_ACCOUNTS}" ]]; then
upsert_env_export CLAW_ENABLE_LINUX_ACCOUNTS "${ENABLE_LINUX_ACCOUNTS}"
fi
if [[ -z "${OPENAI_API_KEY:-}" ]]; then
fail "${ENV_FILE} 缺少 OPENAI_API_KEY,请编辑该文件补齐。"
fi
@@ -197,6 +261,8 @@ ensure_env_file() {
echo "OPENAI_BASE_URL=${OPENAI_BASE_URL:-}"
echo "OPENAI_MODEL=${OPENAI_MODEL:-}"
echo "OPENAI_TIMEOUT_SECONDS=${OPENAI_TIMEOUT_SECONDS:-3600}"
echo "CLAW_SERVICE_SCOPE=${SERVICE_SCOPE}"
echo "CLAW_ENABLE_LINUX_ACCOUNTS=${CLAW_ENABLE_LINUX_ACCOUNTS:-0}"
echo "OPENAI_API_KEY=已配置"
return
fi
@@ -214,6 +280,9 @@ ensure_env_file() {
prompt_value base_url "请输入 OPENAI_BASE_URL" "http://model.mify.ai.srv/v1"
prompt_value model "请输入 OPENAI_MODEL" "tongyi/deepseek-v4-pro"
configure_instance_names
if [[ "${ENABLE_LINUX_ACCOUNTS}" == "1" && "${SERVICE_SCOPE}" != "system" ]]; then
fail "--enable-linux-accounts 需要 --system-service,并建议使用 sudo/root 执行部署。"
fi
prompt_value backend_host "请输入后端监听地址" "127.0.0.1"
prompt_value backend_port "请输入后端端口" "8765"
prompt_value frontend_host "请输入前端监听地址" "0.0.0.0"
@@ -304,14 +373,21 @@ EOF
install_systemd_service() {
local service_name="$1"
local template_path="$2"
local target_path="${USER_SYSTEMD_DIR}/${service_name}.service"
local target_dir="${USER_SYSTEMD_DIR}"
local wanted_by="default.target"
if [[ "${SERVICE_SCOPE}" == "system" ]]; then
target_dir="${SYSTEM_SYSTEMD_DIR}"
wanted_by="multi-user.target"
fi
local target_path="${target_dir}/${service_name}.service"
local tmp_path
tmp_path="$(mktemp)"
mkdir -p "${USER_SYSTEMD_DIR}"
mkdir -p "${target_dir}"
sed \
-e "s#__PROJECT_ROOT__#${ROOT_DIR}#g" \
-e "s#__BACKEND_SERVICE__#${BACKEND_SERVICE}#g" \
-e "s#__FRONTEND_SERVICE__#${FRONTEND_SERVICE}#g" \
-e "s#WantedBy=default.target#WantedBy=${wanted_by}#g" \
"${template_path}" >"${tmp_path}"
install -m 0644 "${tmp_path}" "${target_path}"
rm -f "${tmp_path}"
@@ -319,17 +395,30 @@ install_systemd_service() {
install_services() {
require_command systemctl "当前系统不支持 systemd,无法安装服务。"
log "安装/更新用户级 systemd 服务"
if [[ "${ENABLE_LINUX_ACCOUNTS}" == "1" && "${SERVICE_SCOPE}" != "system" ]]; then
fail "CLAW_ENABLE_LINUX_ACCOUNTS=1 需要 system 服务。请用 sudo bash scripts/deploy-ubuntu.sh --system-service --enable-linux-accounts。"
fi
log "安装/更新 ${SERVICE_SCOPE} systemd 服务"
install_systemd_service "${BACKEND_SERVICE}" "${ROOT_DIR}/deploy/systemd/zk-data-agent-backend.service.template"
install_systemd_service "${FRONTEND_SERVICE}" "${ROOT_DIR}/deploy/systemd/zk-data-agent-frontend.service.template"
systemctl --user daemon-reload
systemctl --user enable "${BACKEND_SERVICE}" "${FRONTEND_SERVICE}"
if [[ "${SERVICE_SCOPE}" == "system" ]]; then
systemctl daemon-reload
systemctl enable "${BACKEND_SERVICE}" "${FRONTEND_SERVICE}"
else
systemctl --user daemon-reload
systemctl --user enable "${BACKEND_SERVICE}" "${FRONTEND_SERVICE}"
fi
}
restart_services() {
log "重启服务"
systemctl --user restart "${BACKEND_SERVICE}"
systemctl --user restart "${FRONTEND_SERVICE}"
if [[ "${SERVICE_SCOPE}" == "system" ]]; then
systemctl restart "${BACKEND_SERVICE}"
systemctl restart "${FRONTEND_SERVICE}"
else
systemctl --user restart "${BACKEND_SERVICE}"
systemctl --user restart "${FRONTEND_SERVICE}"
fi
}
health_check() {
@@ -348,8 +437,13 @@ health_check() {
sleep 0.5
done
warn "健康检查失败,输出最近日志。"
journalctl --user -u "${BACKEND_SERVICE}" -n 80 --no-pager || true
journalctl --user -u "${FRONTEND_SERVICE}" -n 80 --no-pager || true
if [[ "${SERVICE_SCOPE}" == "system" ]]; then
journalctl -u "${BACKEND_SERVICE}" -n 80 --no-pager || true
journalctl -u "${FRONTEND_SERVICE}" -n 80 --no-pager || true
else
journalctl --user -u "${BACKEND_SERVICE}" -n 80 --no-pager || true
journalctl --user -u "${FRONTEND_SERVICE}" -n 80 --no-pager || true
fi
exit 1
}
+19 -3
View File
@@ -11,6 +11,7 @@ ENV_FILE="${ROOT_DIR}/.env.deploy"
DEPLOY_INSTANCE="${CLAW_DEPLOY_INSTANCE:-zk-data-agent}"
BACKEND_SERVICE="${CLAW_BACKEND_SERVICE:-${DEPLOY_INSTANCE}-backend}"
FRONTEND_SERVICE="${CLAW_FRONTEND_SERVICE:-${DEPLOY_INSTANCE}-frontend}"
SERVICE_SCOPE="${CLAW_SERVICE_SCOPE:-}"
BRANCH="${1:-}"
log() {
@@ -26,6 +27,13 @@ configure_instance_names() {
DEPLOY_INSTANCE="${CLAW_DEPLOY_INSTANCE:-${DEPLOY_INSTANCE:-zk-data-agent}}"
BACKEND_SERVICE="${CLAW_BACKEND_SERVICE:-${DEPLOY_INSTANCE}-backend}"
FRONTEND_SERVICE="${CLAW_FRONTEND_SERVICE:-${DEPLOY_INSTANCE}-frontend}"
if [[ -z "${SERVICE_SCOPE}" ]]; then
if [[ "${EUID}" == "0" ]]; then
SERVICE_SCOPE="system"
else
SERVICE_SCOPE="user"
fi
fi
}
resolve_npm_bin() {
@@ -99,12 +107,20 @@ npm_bin="$(resolve_npm_bin)"
export PATH="$(dirname "${npm_bin}"):${PATH}"
"${npm_bin}" run build
log "重启用户服务"
systemctl --user restart "${BACKEND_SERVICE}" "${FRONTEND_SERVICE}"
log "重启 ${SERVICE_SCOPE} 服务"
if [[ "${SERVICE_SCOPE}" == "system" ]]; then
systemctl restart "${BACKEND_SERVICE}" "${FRONTEND_SERVICE}"
else
systemctl --user restart "${BACKEND_SERVICE}" "${FRONTEND_SERVICE}"
fi
log "健康检查"
sleep 1
systemctl --user --no-pager --lines=0 status "${BACKEND_SERVICE}" "${FRONTEND_SERVICE}"
if [[ "${SERVICE_SCOPE}" == "system" ]]; then
systemctl --no-pager --lines=0 status "${BACKEND_SERVICE}" "${FRONTEND_SERVICE}"
else
systemctl --user --no-pager --lines=0 status "${BACKEND_SERVICE}" "${FRONTEND_SERVICE}"
fi
echo
echo "快速更新完成。"
+35
View File
@@ -0,0 +1,35 @@
# model-labeling
对已有数据调用线上模型接口批量打标。
典型输入:
- `output/records.jsonl`
- `output/training.jsonl`
- `output/eval_planning.csv`
- `output/records.csv`
典型流程:
1. 确认输入文件路径。
2. 确认线上模型 `generate` URL。
3.`batch_label_model.py``dry_run=true` 识别格式。
4. 批量请求模型,输出 `output/model_predictions.jsonl`
执行示例:
```json
{
"script_path": "skills/model-labeling/scripts/batch_label_model.py",
"stdin": {
"input_path": "output/records.jsonl",
"model_url": "http://example/generate",
"output_path": "output/model_predictions.jsonl",
"parameters": {
"max_new_tokens": 64
}
},
"timeout_seconds": 600,
"max_output_chars": 20000
}
```
+159
View File
@@ -0,0 +1,159 @@
---
name: model-labeling
description: 对已有数据集调用线上模型 generate 接口批量打标,支持 canonical records、训练 jsonl、评测 CSV 和同事流转表格的统一归一化。
when_to_use: 当用户已有一份数据,或刚通过 product-data / online-mining-v2 生成数据后,希望指定线上模型 URL 批量请求模型、得到预测标签、对比真实标签或产出标注结果时使用。
aliases: batch-labeling, model-annotation, online-model-labeling, 模型打标
allowed_tools: read_file, write_file, grep_search, glob_search, ask_user_question, python_exec
---
# Model Labeling
使用这个 skill 处理“已有数据 -> 统一样本格式 -> 调线上模型接口批量打标 -> 输出预测结果”的流程。
线上模型 URL 经常变化,**不要猜 URL**。如果用户没有明确提供 `http://.../generate` 或等价接口地址,必须先询问用户。
## 能力组织
```text
skills/model-labeling/
SKILL.md
knowledge/
input_formats.md
scripts/
batch_label_model.py
```
`batch_label_model.py` 是 portable script,只依赖 Python 标准库。它会:
- 识别 canonical records JSON/JSONL。
- 识别 product-data 导出的训练 JSONL。
- 识别 eval/planning CSV,优先使用 `newPrompt`
- 识别同事流转 CSV,使用 `query``prev_session``context``function`
- 对未知格式返回结构化错误,要求用户提供字段映射,不要擅自转换。
- 调用用户提供的模型接口,默认请求体为:
```json
{
"inputs": "<prompt>",
"parameters": {
"max_new_tokens": 64
}
}
```
## 交互规则
开始执行前必须确认:
1. **输入数据路径**:用户给出的文件路径,或上一轮产物路径。
2. **模型接口 URL**:必须是用户明确提供的 URL;没有就问。
3. **输入格式是否可识别**canonical records、训练 jsonl、eval CSV、同事流转表格可以直接处理。
4. **未知格式的字段映射**:如果脚本提示 unknown format,需要问用户:
- 哪列是 query
- 哪列是 prompt
- 哪列是真实标签?
- 是否有历史上下文、context
5. **输出路径**:默认写到当前 session 的 `output/model_predictions.jsonl`
不要在没有 URL 的情况下开始打标。不要把临时结果写到项目根目录。所有输出优先放当前 session 的 `output/`
## 推荐流程
### 1. 找到输入文件
如果用户说“刚才生成的数据”“这份数据”,先用会话文件面板或 `glob_search` / `read_file` 找到实际路径。常见路径:
```text
output/records.jsonl
output/training.jsonl
output/eval_planning.csv
output/records.csv
```
### 2. 确认模型 URL
如果用户没有提供 URL,直接问:
```text
请提供这次要调用的线上模型 generate 接口 URL,例如 http://.../generate。
```
如果 `ask_user_question` 可用,优先使用;不可用就普通回复提问并停止。
### 3. 先 dry run 识别格式
先执行一次 `dry_run=true`,只识别格式和样例,不请求模型:
```json
{
"script_path": "skills/model-labeling/scripts/batch_label_model.py",
"stdin": {
"input_path": "output/records.jsonl",
"model_url": "http://example/generate",
"dry_run": true,
"max_records": 3
},
"timeout_seconds": 60,
"max_output_chars": 20000
}
```
如果返回 `ok=false``needs_mapping=true`,必须把错误和已识别字段展示给用户,让用户说明映射。
### 4. 批量请求模型
确认格式后执行:
```json
{
"script_path": "skills/model-labeling/scripts/batch_label_model.py",
"stdin": {
"input_path": "output/records.jsonl",
"model_url": "http://example/generate",
"output_path": "output/model_predictions.jsonl",
"parameters": {
"max_new_tokens": 64
},
"timeout_seconds": 60
},
"timeout_seconds": 600,
"max_output_chars": 20000
}
```
`output_path` 使用相对 `output/...`,平台会路由到当前 session output 目录。
### 5. 展示结果
执行完成后,简短展示:
- 输入格式。
- 处理条数、成功数、失败数。
- 输出路径。
- 抽 3 条预测样例。
如果存在真实标签,说明输出里包含 `gold_label`,后续可以继续做准确率或错误分析。
## 输出格式
默认输出 JSONL,一行一条紧凑 JSON:
```json
{"index":0,"request_id":"...","query":"...","gold_label":"complex=false\nAgent(tag=\"地图导航\")","prediction":"...","ok":true,"latency_ms":123}
```
如果请求失败:
```json
{"index":0,"query":"...","gold_label":"...","prediction":"","ok":false,"error":"HTTP 500 ...","latency_ms":123}
```
默认不把完整 prompt 写入结果,避免文件过大。如确实需要排查,可传 `include_prompt=true`
## 注意事项
- URL、鉴权 header、特殊请求体字段都以用户提供为准。
- 默认接口字段是 `inputs``parameters`;如果用户说明接口不同,需要在脚本输入里传 `request_template`
- 大批量请求前先小样本 dry run。
- 打标脚本只负责请求模型和记录预测结果,不负责修改原始数据。
- 后续准确率统计、错误聚类、补数计划可以再交给其他 skill。
@@ -0,0 +1,81 @@
# Model Labeling 输入格式
本 skill 的脚本会把不同来源的数据统一成 sample:
```json
{
"index": 0,
"request_id": "",
"query": "",
"prompt": "",
"gold_label": "",
"source_format": ""
}
```
## canonical records
识别条件:
- JSONL 每行是对象,或 JSON 数组 / `{ "records": [...] }`
- 对象包含 `turn.query`
字段来源:
- `query``turn.query`
- `request_id``source.request_id`
- `gold_label``complex=true/false` + `label.target`
- `prompt`:按 product-data 的 planning prompt 规则生成
## training jsonl
识别条件:
- 每行是对象。
- 包含 `instruction``system`
字段来源:
- `prompt``system + instruction` 包成 chat template;如果已有 `prompt` 则直接使用。
- `gold_label``output`
- `query`:尽力从 `[当前query]` 后的 `用户:` 抽取。
## eval/planning CSV
识别条件:
- CSV 表头包含 `newPrompt``query`
字段来源:
- `prompt`:优先 `newPrompt`
- `query``query`
- `gold_label``code标签`,如果有 `complex` 列则组合成两行
## 同事流转 CSV
识别条件:
- CSV 表头包含 `query``function`
字段来源:
- `query``query`
- `request_id``request_id`
- `gold_label``function`
- `prompt`:根据 `query``prev_session``context` 生成 planning prompt
## 未知格式
如果不能识别,脚本返回:
```json
{
"ok": false,
"needs_mapping": true,
"columns": ["..."],
"error": "..."
}
```
此时必须询问用户字段映射,不要猜。
+600
View File
@@ -0,0 +1,600 @@
#!/usr/bin/env python3
from __future__ import annotations
"""批量调用线上模型 generate 接口给数据打标。
输入通过 stdin 或 --input 传 JSON 对象,输出稳定 JSON 对象到 stdout。
脚本只依赖 Python 标准库,便于在本地、Linux runtime 用户或远程工作区迁移执行。
"""
import argparse
import csv
import json
import os
import re
import sys
import time
import urllib.error
import urllib.request
from io import StringIO
from pathlib import Path
from typing import Any
DEFAULT_SYSTEM_PROMPT = "你是小爱同学,中文智能语音助手。"
DEFAULT_PARAMETERS = {"max_new_tokens": 64}
class LabelingError(ValueError):
"""输入参数或数据格式错误。"""
def main() -> int:
try:
payload = load_payload()
result = run(payload)
emit({"ok": True, **result})
return 0
except LabelingError as exc:
emit({"ok": False, "error": str(exc), **getattr(exc, "extra", {})})
return 1
except Exception as exc: # noqa: BLE001 - CLI 需要稳定 JSON 错误
emit({"ok": False, "error": str(exc)})
return 1
def load_payload() -> dict[str, Any]:
parser = argparse.ArgumentParser()
parser.add_argument("--input", "-i", help="JSON 参数文件;不传则从 stdin 读取")
args = parser.parse_args()
text = Path(args.input).read_text(encoding="utf-8") if args.input else sys.stdin.read()
if not text.strip():
raise LabelingError("input JSON is required")
payload = json.loads(text)
if not isinstance(payload, dict):
raise LabelingError("input JSON must be an object")
return payload
def emit(payload: dict[str, Any]) -> None:
print(json.dumps(payload, ensure_ascii=False, separators=(",", ":")))
def run(payload: dict[str, Any]) -> dict[str, Any]:
input_path = require_string(payload, "input_path")
model_url = require_string(payload, "model_url")
dry_run = bool(payload.get("dry_run", False))
include_prompt = bool(payload.get("include_prompt", False))
max_records = optional_int(payload.get("max_records"))
start_index = int(payload.get("start_index") or 0)
timeout_seconds = float(payload.get("timeout_seconds") or 60)
output_path = str(payload.get("output_path") or "output/model_predictions.jsonl")
parameters = payload.get("parameters")
if parameters is None:
parameters = dict(DEFAULT_PARAMETERS)
if not isinstance(parameters, dict):
raise LabelingError("parameters must be an object")
headers = payload.get("headers")
if headers is None:
headers = {"Content-Type": "application/json"}
if not isinstance(headers, dict):
raise LabelingError("headers must be an object")
request_template = payload.get("request_template")
if request_template is not None and not isinstance(request_template, dict):
raise LabelingError("request_template must be an object")
source = read_input_file(input_path)
samples = normalize_samples(
source,
field_mapping=payload.get("field_mapping"),
system_prompt=str(payload.get("system_prompt") or DEFAULT_SYSTEM_PROMPT),
session_num=int(payload.get("session_num") or 5),
session_time_minutes=int(payload.get("session_time_minutes") or 5),
)
if start_index:
samples = samples[start_index:]
if max_records is not None:
samples = samples[:max_records]
summary = {
"input_path": input_path,
"source_format": source["format"],
"total_samples": len(samples),
"sample_preview": preview_samples(samples),
}
if dry_run:
return {**summary, "dry_run": True}
if not model_url.startswith(("http://", "https://")):
raise LabelingError("model_url must start with http:// or https://")
results: list[dict[str, Any]] = []
ok_count = 0
output = resolve_runtime_path(output_path)
output.parent.mkdir(parents=True, exist_ok=True)
with output.open("w", encoding="utf-8") as file:
for sample in samples:
item = request_one(
sample,
model_url=model_url,
parameters=parameters,
headers={str(k): str(v) for k, v in headers.items()},
timeout_seconds=timeout_seconds,
request_template=request_template,
include_prompt=include_prompt,
)
if item.get("ok"):
ok_count += 1
results.append(item)
file.write(json.dumps(item, ensure_ascii=False, separators=(",", ":")) + "\n")
return {
**summary,
"dry_run": False,
"output_path": str(output),
"success_count": ok_count,
"failure_count": len(results) - ok_count,
"result_preview": results[:3],
}
def request_one(
sample: dict[str, Any],
*,
model_url: str,
parameters: dict[str, Any],
headers: dict[str, str],
timeout_seconds: float,
request_template: dict[str, Any] | None,
include_prompt: bool,
) -> dict[str, Any]:
prompt = str(sample["prompt"])
body = build_request_body(prompt, parameters, request_template)
started = time.time()
base = {
"index": sample["index"],
"request_id": sample.get("request_id", ""),
"query": sample.get("query", ""),
"gold_label": sample.get("gold_label", ""),
"source_format": sample.get("source_format", ""),
}
if include_prompt:
base["prompt"] = prompt
try:
request = urllib.request.Request(
model_url,
data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
headers=headers,
method="POST",
)
with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
text = response.read().decode("utf-8", errors="replace")
status = getattr(response, "status", 200)
latency_ms = int((time.time() - started) * 1000)
decoded = try_json(text)
return {
**base,
"prediction": extract_prediction(decoded, text),
"ok": 200 <= int(status) < 300,
"status": int(status),
"latency_ms": latency_ms,
"response": decoded if decoded is not None else text,
}
except urllib.error.HTTPError as exc:
text = exc.read().decode("utf-8", errors="replace")
return {
**base,
"prediction": extract_prediction(try_json(text), text),
"ok": False,
"status": exc.code,
"latency_ms": int((time.time() - started) * 1000),
"error": f"HTTP {exc.code}: {text[:500]}",
}
except Exception as exc: # noqa: BLE001 - 单条失败不中断整体批次
return {
**base,
"prediction": "",
"ok": False,
"latency_ms": int((time.time() - started) * 1000),
"error": str(exc),
}
def build_request_body(
prompt: str,
parameters: dict[str, Any],
request_template: dict[str, Any] | None,
) -> dict[str, Any]:
if request_template is None:
return {"inputs": prompt, "parameters": parameters}
return replace_placeholders(request_template, {"prompt": prompt, "parameters": parameters})
def replace_placeholders(value: Any, variables: dict[str, Any]) -> Any:
if isinstance(value, str):
if value == "{{prompt}}":
return variables["prompt"]
if value == "{{parameters}}":
return variables["parameters"]
return value.replace("{{prompt}}", str(variables["prompt"]))
if isinstance(value, list):
return [replace_placeholders(item, variables) for item in value]
if isinstance(value, dict):
return {key: replace_placeholders(item, variables) for key, item in value.items()}
return value
def read_input_file(path: str) -> dict[str, Any]:
file_path = resolve_runtime_path(path)
if not file_path.exists():
raise LabelingError(f"input_path not found: {path}")
suffix = file_path.suffix.lower()
text = file_path.read_text(encoding="utf-8", errors="replace")
if suffix in {".csv", ".tsv"}:
delimiter = "\t" if suffix == ".tsv" else ","
rows = list(csv.DictReader(StringIO(text), delimiter=delimiter))
return {"format": "table", "path": str(file_path), "rows": rows, "columns": list(rows[0].keys()) if rows else []}
if suffix == ".jsonl":
rows = [json.loads(line) for line in text.splitlines() if line.strip()]
return {"format": "jsonl", "path": str(file_path), "rows": rows}
decoded = json.loads(text)
if isinstance(decoded, dict) and isinstance(decoded.get("records"), list):
decoded = decoded["records"]
if isinstance(decoded, list):
return {"format": "json", "path": str(file_path), "rows": decoded}
raise_with_mapping("JSON input must be an array or {records:[...]}", columns=list(decoded.keys()) if isinstance(decoded, dict) else [])
def normalize_samples(
source: dict[str, Any],
*,
field_mapping: Any,
system_prompt: str,
session_num: int,
session_time_minutes: int,
) -> list[dict[str, Any]]:
rows = source.get("rows")
if not isinstance(rows, list) or not rows:
raise LabelingError("input file contains no rows")
if isinstance(field_mapping, dict):
return samples_from_mapping(rows, field_mapping, source["format"], system_prompt)
first = rows[0]
if not isinstance(first, dict):
raise_with_mapping("rows must be objects")
if is_canonical_record(first):
return [
sample_from_record(index, row, system_prompt, session_num, session_time_minutes)
for index, row in enumerate(rows)
if isinstance(row, dict)
]
if is_training_row(first):
return [sample_from_training(index, row, system_prompt) for index, row in enumerate(rows) if isinstance(row, dict)]
if source["format"] == "table":
columns = list(first.keys())
if "newPrompt" in columns:
return [sample_from_eval_row(index, row) for index, row in enumerate(rows) if isinstance(row, dict)]
if "query" in columns and "function" in columns:
return [sample_from_flow_row(index, row, system_prompt) for index, row in enumerate(rows) if isinstance(row, dict)]
raise_with_mapping("unrecognized table format", columns=columns)
raise_with_mapping("unrecognized JSON/JSONL format", columns=list(first.keys()))
def sample_from_record(
index: int,
record: dict[str, Any],
system_prompt: str,
session_num: int,
session_time_minutes: int,
) -> dict[str, Any]:
source = record.get("source") if isinstance(record.get("source"), dict) else {}
turn = record.get("turn") if isinstance(record.get("turn"), dict) else {}
return {
"index": index,
"request_id": str(source.get("request_id") or ""),
"query": str(turn.get("query") or ""),
"prompt": build_planning_prompt(record, system_prompt, session_num, session_time_minutes),
"gold_label": combined_label(record),
"source_format": "canonical_record_v1",
}
def sample_from_training(index: int, row: dict[str, Any], system_prompt: str) -> dict[str, Any]:
prompt = str(row.get("prompt") or "")
if not prompt:
system = str(row.get("system") or system_prompt)
instruction = str(row.get("instruction") or row.get("input") or "")
prompt = wrap_chat_prompt(system, instruction)
return {
"index": index,
"request_id": str(row.get("request_id") or ""),
"query": extract_query_from_prompt(prompt),
"prompt": prompt,
"gold_label": str(row.get("output") or row.get("target") or ""),
"source_format": "training_jsonl",
}
def sample_from_eval_row(index: int, row: dict[str, Any]) -> dict[str, Any]:
gold = str(row.get("code标签") or row.get("function") or row.get("target") or "")
complex_value = row.get("complex")
if complex_value not in (None, "") and not gold.startswith("complex="):
gold = f"complex={normalize_bool_literal(complex_value)}\n{gold}".rstrip()
prompt = str(row.get("newPrompt") or row.get("prompt") or "")
query = str(row.get("query") or "")
if not prompt:
prompt = build_planning_prompt(minimal_record(query, {}, [], gold), DEFAULT_SYSTEM_PROMPT, 5, 5)
return {
"index": index,
"request_id": str(row.get("request_id") or ""),
"query": query,
"prompt": prompt,
"gold_label": gold,
"source_format": "eval_csv",
}
def sample_from_flow_row(index: int, row: dict[str, Any], system_prompt: str) -> dict[str, Any]:
record = record_from_flow_row(row)
return {
"index": index,
"request_id": str(row.get("request_id") or ""),
"query": str(row.get("query") or ""),
"prompt": build_planning_prompt(record, system_prompt, 5, 5),
"gold_label": str(row.get("function") or ""),
"source_format": "flow_csv",
}
def samples_from_mapping(
rows: list[Any],
mapping: dict[str, Any],
source_format: str,
system_prompt: str,
) -> list[dict[str, Any]]:
query_field = str(mapping.get("query") or "")
prompt_field = str(mapping.get("prompt") or "")
label_field = str(mapping.get("label") or mapping.get("target") or "")
request_id_field = str(mapping.get("request_id") or "")
if not query_field and not prompt_field:
raise LabelingError("field_mapping must provide query or prompt")
samples: list[dict[str, Any]] = []
for index, row in enumerate(rows):
if not isinstance(row, dict):
continue
prompt = str(row.get(prompt_field) or "")
query = str(row.get(query_field) or "")
if not prompt:
record = minimal_record(query, {}, [], str(row.get(label_field) or ""))
prompt = build_planning_prompt(record, system_prompt, 5, 5)
samples.append(
{
"index": index,
"request_id": str(row.get(request_id_field) or ""),
"query": query or extract_query_from_prompt(prompt),
"prompt": prompt,
"gold_label": str(row.get(label_field) or ""),
"source_format": f"{source_format}_mapped",
}
)
return samples
def is_canonical_record(row: dict[str, Any]) -> bool:
return isinstance(row.get("turn"), dict) and bool(row["turn"].get("query"))
def is_training_row(row: dict[str, Any]) -> bool:
return any(key in row for key in ("instruction", "system", "output", "prompt"))
def build_planning_prompt(
record: dict[str, Any],
system_prompt: str,
session_num: int,
session_time_minutes: int,
) -> str:
instruction = build_training_instruction(record, session_num, session_time_minutes)
return wrap_chat_prompt(system_prompt, instruction)
def wrap_chat_prompt(system_prompt: str, instruction: str) -> str:
return (
f"<|im_start|>system\n{system_prompt}<|im_end|>\n"
f"<|im_start|>user\n{instruction}<|im_end|>\n"
"<|im_start|>assistant\n"
)
def build_training_instruction(record: dict[str, Any], session_num: int, session_time_minutes: int) -> str:
turn = record.get("turn") if isinstance(record.get("turn"), dict) else {}
query = str(turn.get("query") or "")
context = record.get("context") if isinstance(record.get("context"), dict) else {}
prev_session = record.get("prev_session") if isinstance(record.get("prev_session"), list) else []
current_ts = optional_int(turn.get("timestamp"))
history = render_history(prev_session, current_ts, session_num, session_time_minutes)
return (
"请参考用户的[当前query]、[对话历史]、[知识注入]、[系统状态]识别出[当前query]的[function]结果,[function]是python的code形式。\n"
"[知识注入]\n"
f"{json.dumps({'location': str(context.get('location') or ''), 'rag': str(context.get('rag') or '')}, ensure_ascii=False, indent=0)}\n"
"[系统状态]\n"
"{}\n"
"[对话历史]\n"
f"{history}"
"[当前query]\n"
f"用户: {query}\n"
"[function]\n"
)
def render_history(prev_session: list[Any], current_ts: int | None, session_num: int, session_time_minutes: int) -> str:
usable: list[dict[str, Any]] = []
for item in prev_session[-session_num:]:
if not isinstance(item, dict):
continue
ts = optional_int(item.get("timestamp"))
if current_ts is not None and ts is not None:
if abs(current_ts - ts) > session_time_minutes * 60_000:
continue
usable.append(item)
if not usable:
return ""
lines: list[str] = []
for item in usable:
query = str(item.get("query") or "").strip()
tts = str(item.get("tts") or "").strip()
if query:
lines.append(f"用户: {query}")
if tts:
lines.append(f"小爱: {tts}")
return "\n".join(lines) + ("\n" if lines else "")
def record_from_flow_row(row: dict[str, Any]) -> dict[str, Any]:
prev_session = parse_json_cell(row.get("prev_session"), default=[])
context = parse_json_cell(row.get("context"), default={})
return minimal_record(
str(row.get("query") or ""),
context if isinstance(context, dict) else {},
prev_session if isinstance(prev_session, list) else [],
str(row.get("function") or ""),
request_id=str(row.get("request_id") or ""),
timestamp=optional_int(row.get("timestamp")),
)
def minimal_record(
query: str,
context: dict[str, Any],
prev_session: list[Any],
target: str,
*,
request_id: str = "",
timestamp: int | None = None,
) -> dict[str, Any]:
return {
"source": {"request_id": request_id, "timestamp": timestamp},
"turn": {"query": query, "timestamp": timestamp},
"prev_session": prev_session,
"context": context,
"label": {"target": target},
"dimensions": {},
}
def combined_label(record: dict[str, Any]) -> str:
target = ""
label = record.get("label") if isinstance(record.get("label"), dict) else {}
if isinstance(label, dict):
target = str(label.get("target") or "")
dimensions = record.get("dimensions") if isinstance(record.get("dimensions"), dict) else {}
complex_value = dimensions.get("complex") if isinstance(dimensions, dict) else None
if isinstance(complex_value, bool):
return f"complex={'true' if complex_value else 'false'}\n{target}".rstrip()
return target
def extract_query_from_prompt(prompt: str) -> str:
match = re.search(r"\[当前query\]\s*\n用户[:]\s*(.+)", prompt)
return match.group(1).strip() if match else ""
def normalize_bool_literal(value: Any) -> str:
text = str(value).strip().lower()
return "true" if text in {"true", "1", "yes", "y", "", "复杂"} else "false"
def parse_json_cell(value: Any, default: Any) -> Any:
if value in (None, ""):
return default
if isinstance(value, (dict, list)):
return value
try:
return json.loads(str(value))
except json.JSONDecodeError:
return default
def resolve_runtime_path(path: str) -> Path:
raw = Path(path).expanduser()
if raw.is_absolute():
return raw
scratchpad = Path(str(Path.cwd()))
if os.environ.get("PYTHON_EXEC_SCRATCHPAD"):
scratchpad = Path(os.environ["PYTHON_EXEC_SCRATCHPAD"]).expanduser()
parts = raw.parts
if parts and parts[0] in {"output", "outputs"}:
return scratchpad.parent / "output" / Path(*parts[1:])
if parts and parts[0] in {"input", "inputs"}:
return scratchpad.parent / "input" / Path(*parts[1:])
if parts and parts[0] in {"scratchpad", "scratch"}:
return scratchpad / Path(*parts[1:])
return raw
def optional_int(value: Any) -> int | None:
if value in (None, ""):
return None
try:
return int(float(str(value)))
except (TypeError, ValueError):
return None
def extract_prediction(decoded: Any, text: str) -> str:
if isinstance(decoded, dict):
for key in ("generated_text", "text", "output", "response", "result"):
value = decoded.get(key)
if isinstance(value, str):
return value.strip()
outputs = decoded.get("outputs")
if isinstance(outputs, list) and outputs:
first = outputs[0]
if isinstance(first, str):
return first.strip()
if isinstance(first, dict):
return extract_prediction(first, json.dumps(first, ensure_ascii=False))
choices = decoded.get("choices")
if isinstance(choices, list) and choices:
first = choices[0]
if isinstance(first, dict):
message = first.get("message")
if isinstance(message, dict) and isinstance(message.get("content"), str):
return message["content"].strip()
if isinstance(first.get("text"), str):
return first["text"].strip()
return text.strip()
def try_json(text: str) -> Any:
try:
return json.loads(text)
except json.JSONDecodeError:
return None
def preview_samples(samples: list[dict[str, Any]]) -> list[dict[str, Any]]:
return [
{
"index": item.get("index"),
"query": item.get("query"),
"gold_label": item.get("gold_label"),
"prompt_preview": str(item.get("prompt") or "")[:200],
}
for item in samples[:3]
]
def require_string(payload: dict[str, Any], key: str) -> str:
value = payload.get(key)
if not isinstance(value, str) or not value.strip():
raise LabelingError(f"{key} is required")
return value.strip()
def raise_with_mapping(message: str, columns: list[str] | None = None) -> None:
exc = LabelingError(message)
exc.extra = {"needs_mapping": True, "columns": columns or []} # type: ignore[attr-defined]
raise exc
if __name__ == "__main__":
raise SystemExit(main())
+2
View File
@@ -40,6 +40,7 @@ class ToolExecutionContext:
permissions: AgentPermissions
scratchpad_directory: Path | None = None
python_env_dir: Path | None = None
runtime_user: str | None = None
extra_env: dict[str, str] = field(default_factory=dict)
cancel_event: Any | None = None
process_registry: Any | None = None
@@ -162,6 +163,7 @@ def build_tool_context(
if config.python_env_dir
else None
),
runtime_user=config.runtime_user,
extra_env=dict(extra_env or {}),
cancel_event=cancel_event,
process_registry=process_registry,
+56 -3
View File
@@ -6,6 +6,7 @@ import hashlib
import io
import json
import os
import pwd
import re
import selectors
import shutil
@@ -1644,6 +1645,8 @@ def _resolve_path(
else context.root / expanded
)
resolved = candidate.resolve(strict=not allow_missing)
if _is_session_workspace_path(resolved, context):
return resolved
try:
resolved.relative_to(context.root)
except ValueError as exc:
@@ -1674,6 +1677,17 @@ def _session_logical_path(path: Path, context: ToolExecutionContext) -> Path | N
return None
def _is_session_workspace_path(path: Path, context: ToolExecutionContext) -> bool:
if context.scratchpad_directory is None:
return False
session_root = context.scratchpad_directory.parent.resolve(strict=False)
try:
path.resolve(strict=False).relative_to(session_root)
return True
except ValueError:
return False
def _display_path(path: Path, context: ToolExecutionContext) -> str:
resolved = path.resolve()
try:
@@ -1687,10 +1701,25 @@ def _execution_cwd(context: ToolExecutionContext) -> Path:
return Path(context.jupyter_runtime.binding.workspace_cwd)
if context.scratchpad_directory is not None:
context.scratchpad_directory.mkdir(parents=True, exist_ok=True)
_chown_to_runtime_user(context.scratchpad_directory, context)
return context.scratchpad_directory
return context.root
def _chown_to_runtime_user(path: Path, context: ToolExecutionContext) -> None:
runtime_user = (context.runtime_user or '').strip()
if not runtime_user or os.name != 'posix':
return
try:
user_info = pwd.getpwnam(runtime_user)
except KeyError:
return
try:
os.chown(path, user_info.pw_uid, user_info.pw_gid)
except (FileNotFoundError, PermissionError, OSError):
return
def _is_platform_app_root(root: Path) -> bool:
return all((root / marker).exists() for marker in _PLATFORM_ROOT_MARKERS)
@@ -1886,8 +1915,10 @@ def _write_file(arguments: dict[str, Any], context: ToolExecutionContext) -> str
elif target.exists():
raise ToolExecutionError(f'Path exists and is not a file: {target}')
target.parent.mkdir(parents=True, exist_ok=True)
_chown_to_runtime_user(target.parent, context)
final_text = (previous_text or '') + content if append else content
target.write_text(final_text, encoding='utf-8')
_chown_to_runtime_user(target, context)
rel = _display_path(target, context)
new_sha256 = hashlib.sha256(final_text.encode('utf-8')).hexdigest()
action = 'append_file' if append else 'write_file'
@@ -2007,6 +2038,7 @@ def _edit_file(arguments: dict[str, Any], context: ToolExecutionContext) -> str:
before_sha256 = hashlib.sha256(current.encode('utf-8')).hexdigest()
updated = current.replace(old_text, new_text) if replace_all else current.replace(old_text, new_text, 1)
target.write_text(updated, encoding='utf-8')
_chown_to_runtime_user(target, context)
rel = _display_path(target, context)
replaced = occurrences if replace_all else 1
after_sha256 = hashlib.sha256(updated.encode('utf-8')).hexdigest()
@@ -2091,6 +2123,7 @@ def _notebook_edit(arguments: dict[str, Any], context: ToolExecutionContext) ->
cell.setdefault('execution_count', None)
updated = json.dumps(notebook, ensure_ascii=True, indent=1) + '\n'
target.write_text(updated, encoding='utf-8')
_chown_to_runtime_user(target, context)
after_sha256 = hashlib.sha256(updated.encode('utf-8')).hexdigest()
rel = _display_path(target, context)
return (
@@ -2360,7 +2393,7 @@ def _run_python_exec(arguments: dict[str, Any], context: ToolExecutionContext) -
stderr=subprocess.PIPE,
text=True,
env=_build_subprocess_env(context),
**_subprocess_isolation_kwargs(),
**_subprocess_execution_kwargs(context),
)
_register_child_process(context, process)
if stdin and process.stdin is not None:
@@ -2482,6 +2515,24 @@ def _subprocess_isolation_kwargs() -> dict[str, Any]:
return {}
def _subprocess_execution_kwargs(context: ToolExecutionContext) -> dict[str, Any]:
runtime_user = (context.runtime_user or '').strip()
if not runtime_user or os.name != 'posix':
return _subprocess_isolation_kwargs()
try:
user_info = pwd.getpwnam(runtime_user)
except KeyError as exc:
raise ToolExecutionError(f'Linux runtime user not found: {runtime_user}') from exc
def demote() -> None:
os.setsid()
os.initgroups(runtime_user, user_info.pw_gid)
os.setgid(user_info.pw_gid)
os.setuid(user_info.pw_uid)
return {'preexec_fn': demote}
def _terminate_process(process: subprocess.Popen[Any]) -> None:
if process.poll() is not None:
return
@@ -2616,6 +2667,7 @@ def _run_python_package(arguments: dict[str, Any], context: ToolExecutionContext
text=True,
timeout=timeout_seconds,
env=_build_subprocess_env(context),
**_subprocess_execution_kwargs(context),
)
except subprocess.TimeoutExpired as exc:
stdout = exc.stdout if isinstance(exc.stdout, str) else ''
@@ -2729,6 +2781,7 @@ def _run_bash(arguments: dict[str, Any], context: ToolExecutionContext) -> str:
text=True,
timeout=context.command_timeout_seconds,
env=_build_subprocess_env(context),
**_subprocess_execution_kwargs(context),
)
stdout = completed.stdout or ''
stderr = completed.stderr or ''
@@ -4063,13 +4116,13 @@ def _stream_bash(
command,
shell=True,
executable='/bin/bash',
cwd=context.root,
cwd=_execution_cwd(context),
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1,
env=_build_subprocess_env(context),
**_subprocess_isolation_kwargs(),
**_subprocess_execution_kwargs(context),
)
_register_child_process(context, process)
except (ToolPermissionError, ToolExecutionError, OSError, subprocess.SubprocessError) as exc:
+1
View File
@@ -171,6 +171,7 @@ class AgentRuntimeConfig:
session_directory: Path = field(default_factory=lambda: (Path('.port_sessions') / 'agent').resolve())
scratchpad_root: Path = field(default_factory=lambda: (Path('.port_sessions') / 'scratchpad').resolve())
python_env_dir: Path | None = None
runtime_user: str | None = None
enabled_skill_names: tuple[str, ...] | None = None
+6
View File
@@ -192,6 +192,7 @@ def serialize_runtime_config(runtime_config: AgentRuntimeConfig) -> JSONDict:
if runtime_config.python_env_dir is not None
else None
),
'runtime_user': runtime_config.runtime_user,
'enabled_skill_names': (
list(runtime_config.enabled_skill_names)
if runtime_config.enabled_skill_names is not None
@@ -256,6 +257,11 @@ def deserialize_runtime_config(payload: JSONDict) -> AgentRuntimeConfig:
if payload.get('python_env_dir') is not None
else None
),
runtime_user=(
str(payload['runtime_user'])
if payload.get('runtime_user') is not None
else None
),
enabled_skill_names=enabled_skill_names,
)