From ee596a33e02b4a7c56d1671457b023b6755c9073 Mon Sep 17 00:00:00 2001 From: wuyang6 Date: Wed, 20 May 2026 12:29:11 +0800 Subject: [PATCH 1/3] Add Linux account workspace runtime --- README.md | 49 +- backend/api/server.py | 159 ++++- .../11-workspace-runtime.md | 663 ++++++++++++++++++ frontend/app/app/api/chat/route.ts | 6 +- frontend/app/lib/claw-auth.ts | 132 +++- scripts/deploy-ubuntu.sh | 116 ++- scripts/update-server-fast.sh | 22 +- src/agent_tool_core.py | 2 + src/agent_tools.py | 59 +- src/agent_types.py | 1 + src/session_store.py | 6 + 11 files changed, 1178 insertions(+), 37 deletions(-) create mode 100644 docs/technical-architecture/11-workspace-runtime.md diff --git a/README.md b/README.md index 79b5eac..c9d72d0 100644 --- a/README.md +++ b/README.md @@ -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//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//zk-agent/`。 ## 开发验证 diff --git a/backend/api/server.py b/backend/api/server.py index 206e082..6fdd383 100644 --- a/backend/api/server.py +++ b/backend/api/server.py @@ -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} diff --git a/docs/technical-architecture/11-workspace-runtime.md b/docs/technical-architecture/11-workspace-runtime.md new file mode 100644 index 0000000..9e599a2 --- /dev/null +++ b/docs/technical-architecture/11-workspace-runtime.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//zk-agent/sessions//output/a.jsonl +jupyter:///root/zk_agent_workspaces//output/a.jsonl +ssh:///home/user/zk_agent_workspaces//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//sessions// +``` + +适合： + +- 本地开发。 +- 单用户调试。 +- 早期兼容。 + +问题： + +- 多用户隔离主要靠代码路径约束。 +- 工具进程和平台服务权限一致，风险较高。 + +### local_linux_user + +平台托管的标准多用户工作区。 + +```text +平台账号: banisherwy +Linux runtime user: banisherwy +workspace root: /home/banisherwy/zk-agent/sessions/ +``` + +服务进程可以是 root，工具进程切换到普通 Linux 用户执行。 + +```text +root backend + -> runuser -u banisherwy -- +``` + +职责分工： + +```text +root 服务 + 创建 runtime 用户 + 初始化 workspace + 设置 owner 和权限 + 启停进程 + 管理平台账号和 session + +普通 Linux 用户 + 执行 bash/python + 拥有自己的 workspace + 拥有自己的 Python 虚拟环境 + 只能写自己的目录 +``` + +推荐目录： + +```text +/home//zk-agent/ + sessions/ + / + 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//zk-agent/` 数据。 +- root 服务创建用户和改密码时必须走受控 helper，不能把用户输入拼成 shell 命令。 + +### remote_jupyter + +用户授权的远程工作区。 + +语义是： + +```text +用户把自己已有权限的 Jupyter 环境接入平台。 +平台代替用户在这个环境里执行。 +``` + +这不是平台托管沙盒。权限边界来自用户提供的 Jupyter 凭证。 + +```text +Account: banisherwy +Session: xxx +Runtime: remote_jupyter +Root: /root/zk_agent_workspaces/ +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//zk-agent/sessions//input/ + +remote_jupyter + 上传文件 -> 通过 Jupyter API 写入 /root/zk_agent_workspaces//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//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/ + 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/` 视为 `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 3：local_linux_user runtime + +- root 服务创建与平台账号同名的 Linux 用户。 +- 平台密码同步设置为 Linux 用户密码。 +- Linux 用户允许 SSH 登录。 +- 初始化 `/home//zk-agent/`。 +- 工具执行切到普通 Linux 用户。 +- Python venv、session、output 全部进入用户 home。 +- 停止任务时按 process group 清理。 + +### Phase 4：资源限制和审计 + +- ulimit / cgroup。 +- 每账号磁盘 quota。 +- 工具执行审计。 +- 大文件下载限流。 +- session/output 清理策略。 + +### Phase 5：remote_ssh runtime + +- 在 remote_jupyter 稳定后再考虑。 +- 重点解决认证、relay、长连接、文件传输和远程进程清理。 + +## 关键待决问题 + +1. 平台账号是否允许用户自注册，还是只允许管理员创建？ +2. 用户自注册时，是否允许自动创建同名 Linux 用户？ +3. 删除账号时，是否删除 Linux 用户，是否保留 home 目录？ +4. 本机平台托管 workspace 是否统一迁到 `/home//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//zk-agent。 +``` + +长期建议： + +```text +平台账号负责产品身份。 +Workspace Runtime 负责执行环境。 +Artifact 负责跨 runtime 文件抽象。 +Executor 负责工具执行适配。 +``` + +这样账户体系、Linux 子账户、Jupyter/SSH 远程工作区、文件下载、Python 环境和工具执行可以合到一个统一设计里，而不是继续各自生长。 diff --git a/frontend/app/app/api/chat/route.ts b/frontend/app/app/api/chat/route.ts index 927743b..fca7a2e 100644 --- a/frontend/app/app/api/chat/route.ts +++ b/frontend/app/app/api/chat/route.ts @@ -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}`, diff --git a/frontend/app/lib/claw-auth.ts b/frontend/app/lib/claw-auth.ts index e5346d7..493c69e 100644 --- a/frontend/app/lib/claw-auth.ts +++ b/frontend/app/lib/claw-auth.ts @@ -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 { } 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 { - 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 { } 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 { - 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 { } 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(); + }, + ); +} diff --git a/scripts/deploy-ubuntu.sh b/scripts/deploy-ubuntu.sh index 284ba4f..52532c1 100755 --- a/scripts/deploy-ubuntu.sh +++ b/scripts/deploy-ubuntu.sh @@ -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 <\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 } diff --git a/scripts/update-server-fast.sh b/scripts/update-server-fast.sh index 0e4c29d..8ac9ae8 100755 --- a/scripts/update-server-fast.sh +++ b/scripts/update-server-fast.sh @@ -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 "快速更新完成。" diff --git a/src/agent_tool_core.py b/src/agent_tool_core.py index 0ec8e0f..183e10d 100644 --- a/src/agent_tool_core.py +++ b/src/agent_tool_core.py @@ -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, diff --git a/src/agent_tools.py b/src/agent_tools.py index 550548e..8327039 100644 --- a/src/agent_tools.py +++ b/src/agent_tools.py @@ -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: diff --git a/src/agent_types.py b/src/agent_types.py index 2678956..895d173 100644 --- a/src/agent_types.py +++ b/src/agent_types.py @@ -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 diff --git a/src/session_store.py b/src/session_store.py index abf67e3..25ec176 100644 --- a/src/session_store.py +++ b/src/session_store.py @@ -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, ) From 101a4a57c3c64e8c71f10beb9579e81cda7983e3 Mon Sep 17 00:00:00 2001 From: wuyang6 Date: Wed, 20 May 2026 15:53:06 +0800 Subject: [PATCH 2/3] Add memory research technical doc --- .../10-memory-research.md | 478 ++++++++++++++++++ docs/technical-architecture/README.md | 1 + frontend/app/app/doc/[slug]/page.tsx | 5 + .../technical-docs/10-memory-research.md | 478 ++++++++++++++++++ .../doc-assets/technical-docs/README.md | 1 + 5 files changed, 963 insertions(+) create mode 100644 docs/technical-architecture/10-memory-research.md create mode 100644 frontend/app/public/doc-assets/technical-docs/10-memory-research.md diff --git a/docs/technical-architecture/10-memory-research.md b/docs/technical-architecture/10-memory-research.md new file mode 100644 index 0000000..72859bd --- /dev/null +++ b/docs/technical-architecture/10-memory-research.md @@ -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/.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 / MemGPT：Agent 自主管理内存 + +Letta 延续 MemGPT 思路，把 Agent 看成有长期状态的主体。它通常区分 core memory 和 archival memory：core 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 AutoGen：Memory 组件注入上下文 + +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//memory/ + user.md + skills/ + .md + memory.db +``` + +其中： + +- `user.md`：用户级长期记忆。 +- `skills/.md`：某个 Skill 的使用记忆。 +- `memory.db`：事件队列、状态和 revision 账本。 + +### 3.2 注入逻辑 + +模型调用前，后端调用： + +```text +memory_manager.render_injection(account_id, enabled_skill_names) +``` + +注入规则： + +```text +用户记忆 + 账号级，作为长期偏好注入。 + +Skill 使用记忆 + 只读取当前启用 Skill 对应的 skills/.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/.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/.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 Help：ChatGPT Memory FAQ + https://help.openai.com/en/articles/8590148-memory-faq +- OpenAI Agents SDK：Sessions + https://openai.github.io/openai-agents-python/sessions/ +- Anthropic Claude Code：Memory + https://docs.anthropic.com/en/docs/claude-code/memory +- LangChain / LangGraph：Memory 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` diff --git a/docs/technical-architecture/README.md b/docs/technical-architecture/README.md index a7c3d4d..8e713bd 100644 --- a/docs/technical-architecture/README.md +++ b/docs/technical-architecture/README.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) ## 一句话定位 diff --git a/frontend/app/app/doc/[slug]/page.tsx b/frontend/app/app/doc/[slug]/page.tsx index 8bcd8b3..b660502 100644 --- a/frontend/app/app/doc/[slug]/page.tsx +++ b/frontend/app/app/doc/[slug]/page.tsx @@ -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; diff --git a/frontend/app/public/doc-assets/technical-docs/10-memory-research.md b/frontend/app/public/doc-assets/technical-docs/10-memory-research.md new file mode 100644 index 0000000..72859bd --- /dev/null +++ b/frontend/app/public/doc-assets/technical-docs/10-memory-research.md @@ -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/.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 / MemGPT：Agent 自主管理内存 + +Letta 延续 MemGPT 思路，把 Agent 看成有长期状态的主体。它通常区分 core memory 和 archival memory：core 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 AutoGen：Memory 组件注入上下文 + +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//memory/ + user.md + skills/ + .md + memory.db +``` + +其中： + +- `user.md`：用户级长期记忆。 +- `skills/.md`：某个 Skill 的使用记忆。 +- `memory.db`：事件队列、状态和 revision 账本。 + +### 3.2 注入逻辑 + +模型调用前，后端调用： + +```text +memory_manager.render_injection(account_id, enabled_skill_names) +``` + +注入规则： + +```text +用户记忆 + 账号级，作为长期偏好注入。 + +Skill 使用记忆 + 只读取当前启用 Skill 对应的 skills/.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/.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/.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 Help：ChatGPT Memory FAQ + https://help.openai.com/en/articles/8590148-memory-faq +- OpenAI Agents SDK：Sessions + https://openai.github.io/openai-agents-python/sessions/ +- Anthropic Claude Code：Memory + https://docs.anthropic.com/en/docs/claude-code/memory +- LangChain / LangGraph：Memory 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` diff --git a/frontend/app/public/doc-assets/technical-docs/README.md b/frontend/app/public/doc-assets/technical-docs/README.md index a7c3d4d..8e713bd 100644 --- a/frontend/app/public/doc-assets/technical-docs/README.md +++ b/frontend/app/public/doc-assets/technical-docs/README.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) ## 一句话定位 From a3a4cdca14278dc989f4c0c79b2d98ea41ee2c4b Mon Sep 17 00:00:00 2001 From: wuyang6 Date: Thu, 21 May 2026 19:10:06 +0800 Subject: [PATCH 3/3] Add model labeling skill --- skills/model-labeling/README.md | 35 + skills/model-labeling/SKILL.md | 159 +++++ .../model-labeling/knowledge/input_formats.md | 81 +++ .../scripts/batch_label_model.py | 600 ++++++++++++++++++ 4 files changed, 875 insertions(+) create mode 100644 skills/model-labeling/README.md create mode 100644 skills/model-labeling/SKILL.md create mode 100644 skills/model-labeling/knowledge/input_formats.md create mode 100755 skills/model-labeling/scripts/batch_label_model.py diff --git a/skills/model-labeling/README.md b/skills/model-labeling/README.md new file mode 100644 index 0000000..8a2337b --- /dev/null +++ b/skills/model-labeling/README.md @@ -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 +} +``` diff --git a/skills/model-labeling/SKILL.md b/skills/model-labeling/SKILL.md new file mode 100644 index 0000000..5ec5d67 --- /dev/null +++ b/skills/model-labeling/SKILL.md @@ -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": "", + "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。 diff --git a/skills/model-labeling/knowledge/input_formats.md b/skills/model-labeling/knowledge/input_formats.md new file mode 100644 index 0000000..6e702bd --- /dev/null +++ b/skills/model-labeling/knowledge/input_formats.md @@ -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": "..." +} +``` + +此时必须询问用户字段映射，不要猜。 diff --git a/skills/model-labeling/scripts/batch_label_model.py b/skills/model-labeling/scripts/batch_label_model.py new file mode 100755 index 0000000..f177fab --- /dev/null +++ b/skills/model-labeling/scripts/batch_label_model.py @@ -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())