第26章 Omnigent 自定义策略详解：CEL 与 Python 实现
自定义策略
当内置策略无法满足你的需求时，你有两种选择：
* `cel_policy`：直接在配置中编写 CEL 表达式。无需部署代码，建议从这里入手。 * Python 策略：注册在服务器上的完整 Python 函数。仅当逻辑涉及状态、外部调用，或 CEL 非图灵完备表达式无法描述的场景时才使用。
使用 `cel_policy`
`cel_policy` 是一个内置策略，它对每个事件求值一个 CEL 表达式。CEL 是一种安全、沙箱化的表达式语言——无循环、无 I/O、无副作用——因此它是大多数自定义规则的理想默认选择。 表达式接收一个包含 `type` 和 `data` 字段的事件对象（即所有策略看到的同一个 `PolicyEvent`）。返回 `true` 即可拒绝该操作。
示例
拒绝特定工具：
```yaml policies: no_dangerous_tool: type: function handler: omnigent.policies.builtins.safety.cel_policy factory_params: expression: "event.type == 'tool_call' && event.data.name == 'dangerous_tool'" reason: "dangerous_tool is not allowed in this session." ```
限制在特定目录外的写入操作：
```yaml policies: confine_writes: type: function handler: omnigent.policies.builtins.safety.cel_policy factory_params: expression: > event.type == 'tool_call' && event.data.name in ['sys_os_write', 'sys_os_edit'] && !event.data.arguments.path.startsWith('/home/user/project/') reason: "Writes are only allowed inside /home/user/project/." ```
使用 `state_updates` 追踪调用：
返回一个 `state_updates` 列表，以在会话状态中累加计数器或标志。引擎在策略触发后应用这些变更，因此后续事件可通过 `event.session_state` 读取更新后的值。
```yaml policies: track_tool_calls: type: function handler: omnigent.policies.builtins.safety.cel_policy factory_params: expression: > event.type == 'tool_call' ? {"result": "ALLOW", "state_updates": [ {"key": "tool_count", "action": "increment", "value": 1}, {"key": "last_tool", "action": "set", "value": event.data.name} ]} : {"result": "ALLOW"} ```
状态在整个会话期间持久化。配套策略可以读取 `event.session_state.tool_count` 来根据累计总数进行门控。
只要规则能表示为事件字段上的布尔值，就使用 `cel_policy`。你无需编写或部署代码，即可享受相同的策略生命周期（ALLOW / ASK / DENY）。
Python 策略
当 `cel_policy` 不够用时，使用 Python 策略——例如，当规则需要跨调用累积状态、数据库查找，或 CEL 无法表达的控制流时。
1. 编写策略函数
策略是一个 Python 函数，接收事件并返回 `ALLOW`、`ASK`、`DENY` 或 `None`（无意见）：
```python from omnigent.policies.schema import PolicyEvent, PolicyResponse def my_policy(event: PolicyEvent) -> PolicyResponse | None: if event["type"] != "tool_call": return None if event["data"]["name"] == "dangerous_tool": return {"result": "DENY", "reason": "Blocked."} return {"result": "ALLOW"} ```
如果策略需要参数，请使用工厂模式。该函数接收配置并返回求值器：
```python def block_domains(blocked_domains: list[str]) -> callable: blocked = frozenset(d.lower() for d in blocked_domains) def evaluate(event: PolicyEvent) -> PolicyResponse | None: if event["type"] != "tool_call": return None url = event["data"]["arguments"].get("url", "") for domain in blocked: if domain in url.lower(): return {"result": "DENY", "reason": f"Domain {domain} blocked."} return {"result": "ALLOW"} return evaluate ```
2. 在服务器上注册
为了让你的策略被 Omnigent 发现并在 UI 中可见，请执行以下两步：
从模块中导出 `POLICY_REGISTRY`：
```python # myorg/policies.py POLICY_REGISTRY = [ { "handler": "myorg.policies.block_domains", "kind": "factory", "name": "Block Domains", "description": "Block web access to specific domains.", "params_schema": { "type": "object", "properties": { "blocked_domains": { "type": "array", "items": {"type": "string"}, "description": "Domains to block" } }, "required": ["blocked_domains"] } } ] ```
将模块添加到服务器配置中：
```yaml # config.yaml policy_modules: - myorg.policies ```
然后使用该配置启动服务器：
```bash omnigent server -c config.yaml ```
`-c` 是 `--config` 的简写。
注册后，你的自定义策略将与内置策略并列出现。当你在聊天中要求 Omnigent 添加策略时，它可以选择这些策略；它们也会显示在 UI 的设置面板中。
3. 使用它
一旦注册，你的自定义策略的工作方式与任何内置策略相同。有关应用它的所有方式（聊天、Omnigent YAML 或服务器配置），参见“添加策略”。
参考：PolicyEvent
每个策略接收一个 `PolicyEvent` 字典。使用 `type` 字段过滤你关注的事件：
| 事件类型 | 触发时机 | 主要数据字段 | | :--- | :--- | :--- | | request | 用户发送消息 | `user_content`（输入文本），`attachments`（包含 `{filename, content_type, text}` 的列表） | | tool_call | Omnigent 即将调用工具 | `name`（工具名称），`arguments`（字典） | | tool_result | 工具已返回结果 | `result`（工具输出）；`event["request_data"]` 携带原始 `tool_call` 负载 | | response | Omnigent 即将发送助手消息 | 助手消息文本 |
对于你不处理的事件类型，返回 `None`。
对于 `request` 事件，文本/代码附件（例如 CSV）在 `text` 中携带解码后的内容。存档、Office 文档和数据库（`.zip`, `.docx`, `.xlsx`, `.pptx`, `.db`, `.sqlite`, `.sqlite3`）仅列出其 `filename` 和 `content_type`，但 `text` 为空。这样策略可以根据名称和类型进行操作，而无需扫描其内容。
每个事件还包含：
* `event["target"]`：在 `tool_call` / `tool_result` 时为工具名称；其他情况下为 `None`。 * `event["session_state"]`：由先前策略累积的每会话键值对只读字典。通过返回值中的 `state_updates` 进行变更，而非直接修改此字典。 * `event["context"]`：会话元数据（见下文）。
EventContext
`event["context"]` 是一个字典，包含以下字段：
| 字段 | 类型 | 描述 | | :--- | :--- | :--- | | `actor.run_as` | str | 认证用户邮箱，例如 "alice@example.com"。未知时为空字符串。 | | `actor.client_id` | str | OAuth 客户端 ID。未知时为空字符串。 | | `usage.input_tokens` | float | 本会话累计消耗的输入 token 数。 | | `usage.output_tokens` | float | 本会话累计消耗的输出 token 数。 | | `usage.total_cost_usd` | float | 累计 LLM 成本（美元）。在首次 LLM 调用前或定价不可用时为 0.0。 | | `model` | str \| None | 当前模型，例如 "databricks-claude-sonnet-4-6"。未知时为 None。 | | `harness` | str \| None | 当前 harness，例如 "codex-native"。在 Web / API 路径上为 None。 | | `labels` | dict | 对话护栏标签的只读快照。 |
参考：PolicyResponse
策略可调用对象返回 `None`（弃权）或包含以下字段的字典：
| 字段 | 类型 | 必需 | 描述 | | :--- | :--- | :--- | :--- | | `result` | str | 是 | "ALLOW", "DENY", 或 "ASK"（不区分大小写）。 | | `reason` | str | 否 | 人类可读的解释。在 ASK 时显示给用户；在 DENY 时记录日志。ALLOW 时可选。 | | `data` | any | 否 | 替代负载。如果出现在 ALLOW 中，执行点会用此值替换原始事件内容——适用于返回去除 PII 的工具参数版本。 | | `state_updates` | list | 否 | 会话状态变更的有序列表。在 ALLOW 和 DENY 时应用；在 ASK 时扣留，待批准。参见 StateUpdateEntry。 | | `set_labels` | dict[str, str] | 否 | 标签键值写入。通过策略声明的 `set_labels` 允许列表进行过滤。 |
返回 `None` 被视为弃权（等同于 `{"result": "ALLOW"}`）。
```python
# 最小
{"result": "ALLOW"}
# 完整
{
"result": "DENY",
"reason": "Blocked.",
"data":
参考：StateUpdateEntry
`state_updates` 中的每个条目是一个字典：
| 字段 | 类型 | 必需 | 描述 | | :--- | :--- | :--- | :--- | | `key` | str | 是 | 要变更的状态键，例如 "call_count"。 | | `action` | str | 是 | "set", "increment", "delete", "append" 之一。 | | `value` | any | 是（"delete" 除外） | 操作数。set、increment 和 append 需要；delete 忽略。 |
引擎将更新应用于 `event["session_state"]`，以便后续策略求值能看到新值。
```python # 覆盖 {"key": "status", "action": "set", "value": "escalated"} # 数值增量（若不存在则初始化为 0） {"key": "call_count", "action": "increment", "value": 1} # 追加到列表（若不存在则初始化为 []） {"key": "seen_tools", "action": "append", "value": "sys_os_shell"} # 完全移除该键 {"key": "temp_flag", "action": "delete"} ```
