# 操作码 AI 代理的执行运行时。你可以推断，OPCODE 会执行——第一次定义后每次运行零个标记。工作流程跨会话持续存在、按计划运行并协调多个代理。持久 SSE 守护进程：1 个服务器、N 个代理、1 个数据库。 JSON定义的DAG，逐级执行，自动并行。 6 个基于 SSE (JSON-RPC) 的 MCP 工具。 **为什么要使用 OPCODE 而不是自己推理每个步骤？** 每个重复的工作流程都会消耗代币，重新推理您已经做出的决策。 OPCODE 模板化您的推理一次并确定性地执行它 - 零推理成本，每次运行的输出相同，在上下文重置后仍然存在。 ## 哪个工具？我想... |工具 ---|--- 创建/更新工作流程模板 |操作码.定义执行工作流程 |操作码.run 检查状态或待决决定 |操作码状态解决决定/取消/重试 |操作码信号列出工作流程、事件或模板 |操作码.查询可视化工作流程 DAG |操作码图 ## 快速入门安装：去安装 github.com/rendis/opcode/cmd/opcode@latest 首次设置（写入配置并启动守护进程）：操作码安装 --listen-addr :4100 --vault-key "my-passphrase" 停止后重新启动：`OPCODE_VAULT_KEY="my-passphrase" opcode` MCP客户端配置： { “mcp服务器”：{ “mcp服务器”：{ “操作码”：{ “类型”：“sse”， “url”：“http://localhost:4100/sse” } } } 每个代理通过工具调用中的“agent_id”进行自我识别。操作码自动注册未知代理。为每个代理选择一个稳定的 ID（例如“内容编写器”、“部署机器人”）。工作流程在重新启动后仍然有效。启动时，孤立的“活动”工作流程将变为“暂停”。使用 `opcode.query({ "resource": "workflows", "filter": { "status": "suspended" } })` 进行查询，然后通过 `opcode.signal` 恢复或取消。有关完整配置、子命令、SIGHUP 热重载、安全模型、Web 面板和基准测试，请参阅 [operations.md]()。 ## MCP 工具 ### 操作码.define 注册可重用的工作流程模板。版本自动递增（v1、v2、v3...）。参数|类型 |必填|描述 ---|---|---|--- `名称` |字符串|是的 |模板名称 `定义` |对象|是的 |工作流程定义（见下文） `agent_id` |字符串|是的 |定义代理 ID `描述` |字符串|没有|模板说明 `输入模式` |对象|没有|用于输入验证的 JSON 架构 `输出模式` |对象|没有|用于输出验证的 JSON 架构 `触发器` |对象|没有|触发器配置（请参阅[workflow-schema.md]()） **返回** : `{ "name": "...", "version": "v1" }` ### 操作码.run 从已注册的模板执行工作流程。参数|类型 |必填|描述 ---|---|---|--- `模板名称` |字符串|是的 |要执行的模板 `agent_id` |字符串|是的 |发起代理 ID `版本` |字符串|没有|版本（默认：最新） `参数` |对象|没有|输入参数 **返回**： { "workflow_id": "uuid", "status": "已完成|暂停|失败", “输出”：{...}， “开始于”：“RFC3339”， “completed_at”：“RFC3339”， “步骤”：{ "step-id": { "step_id": "...", "status": "已完成", "output": {...}, "duration_ms": 42 } } } 如果`status`是`“暂停”`，调用`opcode.status`来查看`pending_decisions`。 ### 操作码.状态获取工作流执行状态。参数|类型 |必填|描述 ---|---|---|--- `workflow_id` |字符串|是的 |查询流程 **返回**： { "workflow_id": "uuid", “状态”：“暂停”， "steps": { "step-id": { "status": "...", "output": {...} } }, “待决决定”：[ { “id”：“uuid”， "step_id": "原因步骤", “上下文”：{“提示”：“...”，“数据”：{...} }， "options": [ { "id": "批准", "description": "继续" } ], “timeout_at”：“RFC3339”， "后备": "拒绝", “状态”：“待处理” } ], “事件”：[...] } 工作流程状态：“待处理”、“活动”、“暂停”、“已完成”、“失败”、“已取消”。 ### 操作码.信号向暂停的工作流程发送信号。参数|类型 |必填|描述 ---|---|---|--- `workflow_id` |字符串|是的 |目标工作流程 `信号类型` |枚举 |是的 | `决定` / `数据` / `取消` / `重试` / `跳过` `有效负载` |对象|是的 |信号有效负载（见下文） `step_id` |字符串|没有|目标步数 `agent_id` |字符串|没有|信号剂 `推理` |字符串|没有|代理人的推理 **按信号类型划分的有效负载**：信号|步骤 ID |有效负载|行为 ---|---|---|--- `决定` |必填| `{ "choice": "" }` |解决决定，自动恢复 `数据` |可选 | `{ "key": "value", ... }` |将数据注入工作流程 `取消` |没有| `{}` |取消工作流程 `重试` |必填| `{}` |重试失败的步骤 `跳过` |必填| `{}` |跳过失败的步骤 **返回**（决定）：`{ "ok": true, "resumed": true, "status": "completed", ... }` **返回**（其他）：`{ "ok": true, "workflow_id": "...", "signal_type": "..." }` ### 操作码.query 查询工作流、事件或模板。参数|类型 |必填|描述 ---|---|---|--- `资源` |枚举 |是的 | “工作流程”/“事件”/“模板” `过滤器` |对象|没有|过滤条件 **按资源过滤字段**：资源 |领域 ---|--- `工作流程` | `status`、`agent_id`、`since` (RFC3339)、`limit` `事件` | `workflow_id`、`step_id`、`event_type`、`since`、`limit` `模板` | `名称`、`agent_id`、`限制` 注意：事件查询需要过滤器中的“event_type”或“workflow_id”。 **返回** : `{ "": [...] }` \-- 结果包装在按资源类型键入的对象中。 ### 操作码.diagram 从模板或运行的工作流程生成可视化 DAG 图。参数|类型 |必填|描述 ---|---|---|--- `模板名称` |字符串|否* |可视化模板（结构预览） `版本` |字符串|没有|模板版本（默认：最新） `workflow_id` |字符串|否* |可视化工作流程（带有运行时状态） `格式` |枚举 |是的 | `ascii` / `美人鱼` / `图像` `include_status` |布尔 |没有|显示运行时状态覆盖（默认值：如果workflow_id则为true） * 需要“template_name”或“workflow_id”之一。 * `template_name` \--执行前预览DAG结构 * `workflow_id` \-- 可视化实时步骤状态 * `format: "ascii"` \-- CLI 友好文本，带有方框图字符 * `format: "mermaid"` \-- markdown-embeddable 流程图语法 * `format: "image"` \--视觉通道的base64编码PNG **返回** : `{ "format": "ascii", "diagram": "..." }` ## 工作流程定义 { “步骤”：[...]， "inputs": { "key": "值或 ${{secrets.KEY}}" }, "context": { "intent": "...", "notes": "..." }, “超时”：“5m”， "on_timeout": "失败|暂停|取消", "on_complete": { /* 步骤定义 */ }, "on_error": { /* 步骤定义 */ }, “元数据”：{} } 领域|类型 |必填|描述 ---|---|---|--- `步骤` |步骤定义[] |是的 |工作流程步骤 `输入` |对象|没有|输入参数（支持`${{}}`） `上下文` |对象|没有|工作流程上下文，可通过“${{context.*}}”访问 `超时` |字符串|没有|工作流程截止时间（例如“5m”、“1h”） `on_timeout` |字符串|没有| `失败`（默认）、`暂停`、`取消` `on_complete` |步骤定义 |没有|完成后挂钩步骤 `on_error` |步骤定义 |没有|工作流程失败时的挂钩步骤 `元数据` |对象|没有|任意元数据 ### 步骤定义 { "id": "步骤 ID", "type": "动作|条件|循环|并行|等待|推理", “行动”：“http.get”， “参数”：{...}， “depends_on”：[“其他步骤”]， "condition": "CEL 保护表达式", “超时”：“30秒”， "retry": { "max": 3, "backoff": "指数", "delay": "1s", "max_delay": "30s" }, “on_error”：{“策略”：“忽略|失败工作流程|fallback_step |重试”，“fallback_step”：“id”}， "config": { /* 特定类型 */ } } `type` 默认为 `action`。有关所有配置块，请参阅 [workflow-schema.md]()。 ## 步骤类型 ### 操作（默认）执行已注册的操作。将“action”设置为操作名称，将“params”设置为输入。 ### 条件计算 CEL 表达式和分支。 { “id”：“路线”， “类型”：“条件”， “配置”：{ "表达式": "输入.env", “分支”：{ “产品”：[...]，“分期”：[...] }， “默认”：[...] } } ###循环迭代集合或条件。循环变量：`${{loop.item}}`、`${{loop.index}}`。 { "id": "处理项目", “类型”：“循环”， “配置”：{ “模式”：“for_each”， "over": "[\"a\",\"b\",\"c\"]", “身体”：[ { “id”：“哈希”， "action": "crypto.hash", "params": { "data": "${{loop.item}}" } } ], “最大迭代次数”：100 } } 模式：“for_each”（迭代“over”）、“while”（当“condition”为真时循环）、“until”（循环直到“condition”为真）。 ###并行同时执行分支。 { "id": "扇出", “类型”：“并行”， “配置”：{ “模式”：“全部”， “分支机构”：[ [{“id”：“a”，“action”：“http.get”，“params”：{...} }]， [{“id”：“b”，“action”：“http.get”，“params”：{...} }] ] } } 模式：“all”（等待所有分支）、“race”（第一个分支获胜）。 ### 等待延迟执行或等待指定信号。 { "id": "暂停", "type": "等待", "config": { "持续时间": "5s" } } ###推理暂停代理决策的工作流程。空“选项”=自由形式（接受任何选择）。 { “id”：“评论”， “类型”：“推理”， “配置”：{ "prompt_context": "查看数据并做出决定", “选项”：[ { "id": "批准", "描述": "继续" }, { "id": "拒绝", "描述": "停止" } ], "data_inject": { "analysis": "steps.analyze.output" }, “超时”：“1小时”， "后备": "拒绝", “目标代理”：“” } } ## 变量插值语法：`${{namespace.path}}` 命名空间|示例|可用字段 ---|---|--- `步骤` | `${{steps.fetch.output.body}}` | `.output.*`、`.status` `输入` | `${{inputs.api_key}}` |来自“opcode.run”中“params”的键 `工作流程` | `${{workflow.run_id}}` | `run_id`、`name`、`template_name`、`template_version`、`agent_id` `上下文` | `${{context.intent}}` |工作流定义中“上下文”的键 `秘密` | `${{secrets.DB_PASS}}` |密钥存储在保险库中 `循环` | `${{loop.item}}`、`${{loop.index}}` | `item`（当前元素），`index`（从 0 开始）两遍解决方案：首先是非机密，然后通过 AES-256-GCM 库提供机密。 **CEL 陷阱**：`loop` 是 CEL 中的保留字。在 CEL 表达式中使用 `iter.item` / `iter.index`。 `${{loop.item}}` 插值语法不受影响。有关 CEL、GoJQ、Expr 引擎的详细信息，请参阅 [expressions.md]()。 ## 内置操作类别 |行动 ---|--- **HTTP** | `http.request`、`http.get`、`http.post` **文件系统** | `fs.read`、`fs.write`、`fs.append`、`fs.delete`、`fs.list`、`fs.stat`、`fs.copy`、`fs.move` **外壳** | `shell.exec` **加密货币** | `crypto.hash`、`crypto.hmac`、`crypto.uuid` **断言** | `assert.equals`、`assert.contains`、`assert.matches`、`assert.schema` **表达** | `expr.eval` **工作流程** | `workflow.run`、`workflow.emit`、`workflow.context`、`workflow.fail`、`workflow.log`、`workflow.notify` **快速参考**（最常用的操作）： * **`http.get`** : `url` (req), `headers`, `timeout`, `fail_on_error_status` \-- 输出: `{ status_code, headers, body,uration_ms }` * **`shell.exec`** : `command` (req)、`args`、`stdin`、`timeout`、`env`、`workdir` \-- 输出：`{ stdout、stderr、exit_code、killed }` * **`fs.read`** : `path` (req), `encoding` \-- 输出: `{ 路径, 内容, 编码, 大小 }` * **`workflow.notify`** : `message` (req), `data` \-- 输出：`{ notification: true/false }` \-- 通过 MCP SSE 将实时通知推送给代理（尽力而为） * **`expr.eval`** : `expression` (req), `data` \-- 输出: `{ result: }` \-- 根据工作流范围（步骤、输入、工作流、上下文）计算 Expr 表达式有关所有 26 个操作的完整参数规范，请参阅 [actions.md]()。 ## 使用 shell.exec 编写脚本 `shell.exec` 自动解析 JSON 标准输出。约定：stdin=JSON、stdout=JSON、stderr=错误、非零退出=失败。对未处理的文本使用“stdout_raw”。有关特定于语言的模板（Bash、Python、Node、Go），请参阅 [patterns.md]()。 ## 推理节点生命周期 1. 工作流程到达推理步骤 2.Executor创建PendingDecision，发出`decision_requested`事件 3. 工作流程状态变为“暂停” 4. 代理调用“opcode.status”以查看带有上下文和选项的待决决策 5. Agent通过`opcode.signal`解析： { "workflow_id": "...", "signal_type": "决定", "step_id": "原因步骤", “有效负载”：{“选择”：“批准”} } 6. 工作流程在信号后自动恢复 7. 如果超时到期：自动选择`fallback`选项，或者如果没有fallback则步骤失败 ## 常见模式请参阅 [patterns.md]() 了解完整的 JSON 示例：线性管道、条件分支、for-each 循环、并行扇出、人机交互、错误恢复、子工作流和 MCP 生命周期。 ## 错误处理战略|行为 ---|--- `忽略` |跳过步骤，工作流程继续 `失败工作流程` |整个工作流程失败 `后备步骤` |执行回退步骤 `重试` |推迟重试政策退避：“无”、“线性”、“指数”、“常数”。不可重试的错误（验证、许可、断言）永远不会重试。有关断路器、超时交互、错误代码，请参阅 [error-handling.md]()。 ## 性能 10 步并行工作流程在约 50μs 内完成，500 步在约 2.4ms 内完成。事件存储每秒维持约 15k 追加，在 100 个并发写入者下下降 <12%。工作池开销约为 0.85μs/任务（在任何池大小下>1M 任务/秒）。完整的基准图表、每个场景的细分和方法：[`docs/benchmarks.md`]()。

操作码

包含技能