Stop-on-Non-JSON:使自主代理可信赖的安全模式
Source: Dev.to
停止在非 JSON:让自主代理可信赖的安全模式
在构建能够自行决定下一步行动的 自主代理 时,最常见的风险之一是它们可能会产生 非结构化的、不可预测的输出。如果这些输出被直接发送给下游系统(例如数据库、API 或另一个代理),可能会导致 安全漏洞、数据损坏或意外行为。
为了解决这个问题,本文介绍了一种被称为 “Stop on non‑JSON”(遇到非 JSON 停止)的安全模式。该模式通过强制代理仅在返回有效的 JSON 响应时继续执行,从而在出现异常或意外输出时立即中止流程。
为什么需要 “Stop on non‑JSON”
| 常见问题 | 可能的后果 |
|---|---|
| 代理返回自由文本而不是预期的 JSON | 解析错误 → 程序崩溃 |
| 输出包含恶意指令或注入代码 | 安全漏洞 |
| 结构不符合约定的 schema | 下游系统无法处理,导致数据丢失 |
通过在每一步检查返回值的 JSON 合规性,我们可以在错误传播之前捕获并处理异常。
实现思路
- 定义严格的 JSON schema(使用 JSON Schema、TypeScript 类型或 Pydantic 模型)。
- 在每次调用后立即验证返回的字符串是否是有效的 JSON,并且符合 schema。
- 如果验证失败,抛出异常或返回错误信息,阻止后续步骤继续执行。
下面给出一个使用 Python 与 Pydantic 的示例实现。
from pydantic import BaseModel, ValidationError
import json
class AgentResponse(BaseModel):
action: str
parameters: dict
thought: str
def stop_on_non_json(raw_output: str) -> AgentResponse:
"""
将代理的原始输出解析为 JSON,并在解析或校验失败时抛出异常。
"""
try:
data = json.loads(raw_output)
except json.JSONDecodeError as e:
raise ValueError(f"输出不是有效的 JSON: {e}")
try:
return AgentResponse(**data)
except ValidationError as e:
raise ValueError(f"JSON 不符合预期 schema: {e}")
关键点
json.loads用于快速检测是否为合法的 JSON。Pydantic(或其他验证库)负责检查字段类型、必填项以及自定义约束。- 当任一步骤失败时,函数会抛出
ValueError,从而让调用者能够 立即停止 当前任务。
在实际代理工作流中的使用
def run_agent_step(prompt: str) -> dict:
# 1. 调用大语言模型(LLM)获取响应
raw_output = call_llm(prompt)
# 2. 使用 “Stop on non‑JSON” 进行安全检查
try:
response = stop_on_non_json(raw_output)
except ValueError as err:
# 记录错误并中止后续操作
logger.error(f"安全检查失败: {err}")
raise # 或者返回一个特定的错误结构给上层
# 3. 基于验证通过的响应继续业务逻辑
return {
"action": response.action,
"params": response.parameters,
"thought": response.thought,
}
在上述工作流中:
- 任何非 JSON 或不符合 schema 的输出 都会在第 2 步被捕获。
- 通过
raise或返回错误结构,系统可以决定是 重试、回滚 还是 人工干预。 - 只有在第 2 步成功后,才会进入第 3 步的业务处理,确保后续系统只接收可信的数据。
与其他安全模式的对比
| 模式 | 触发时机 | 主要优势 | 适用场景 |
|---|---|---|---|
| Stop on non‑JSON | 每次 LLM 响应后立即验证 | 简单、低开销、易于集成 | 所有需要结构化输出的代理 |
| Re‑ask / Clarify | 当检测到不符合预期时重新提问 | 可自动纠正模型的失误 | 对话式系统、需要高准确率的任务 |
| Sandbox Execution | 在受限环境中运行生成的代码 | 防止恶意代码执行 | 代码生成、脚本自动化 |
| Rate Limiting / Throttling | 对请求频率进行限制 | 防止滥用和资源耗尽 | 大规模部署、公共 API |
Stop on non‑JSON 侧重于 数据完整性,而不是 行为控制。它可以与其他模式组合使用,例如在检测到非 JSON 时触发 re‑ask,从而实现更柔性的错误恢复。
实际案例:从错误输出到安全恢复
假设我们的代理负责 自动化订单处理,期望返回如下 JSON:
{
"action": "create_order",
"parameters": {
"product_id": "12345",
"quantity": 2
},
"thought": "用户想购买两件商品。"
}
1️⃣ 正常情况
- LLM 正确返回 JSON → 通过验证 → 系统创建订单。
2️⃣ 错误情况:自由文本
LLM 返回:
好的,我已经为您下单了!祝您购物愉快 😊
json.loads抛出JSONDecodeError→stop_on_non_json报错 → 代理停止,系统记录错误并通知人工客服。
3️⃣ 错误情况:结构错误
LLM 返回:
{
"action": "create_order",
"params": { "product_id": "12345", "quantity": 2 }
}
- JSON 合法,但缺少
parameters字段 →Pydantic抛出ValidationError→ 同样触发停止。
通过 统一的错误捕获,我们避免了错误订单进入数据库,保证了业务的 安全性和可靠性。
最佳实践
- 始终使用 schema:即使是最简单的键值对,也建议使用 JSON Schema 或 Pydantic 明确定义。
- 在边界层统一验证:将 “Stop on non‑JSON” 放在 LLM 调用包装器 中,而不是每个业务函数里重复实现。
- 记录完整的原始输出:便于后续审计和调试。
- 结合重试策略:在捕获异常后,可根据业务需求决定是否 重新请求 LLM(带上更明确的提示)或 转人工。
- 监控异常率:异常频率的突增可能意味着提示词(prompt)设计不佳或模型出现退化,需要及时调整。
小结
- “Stop on non‑JSON” 是一种轻量级但极其有效的安全模式,能够在代理产生非结构化或错误结构输出时立即中止流程。
- 通过 JSON 解析 + schema 验证,我们可以在错误传播之前捕获异常,防止安全风险和业务故障。
- 将该模式作为 代理调用链的第一道防线,并与 重试、人工干预 等其他安全措施结合使用,可显著提升自主代理系统的 可信度 与 鲁棒性。
下一步:在你的项目中实现一个统一的
safe_llm_call包装器,使用上述示例代码作为参考,并根据业务需求定制 schema。这样,你的自主代理将能够在面对不确定的生成式模型输出时,始终保持安全、可控。
引言
如果让代理按照计划(cron)运行并触及真实系统——API、社交网络、链上操作——你会很快发现一个残酷的事实:大多数“agent failures”并不是模型失败或操作失败。代理会在不该调用时继续调用端点。
为什么非 JSON 响应是危险的
许多危险的边缘情况会以非 JSON 负载出现:
- WAF / 机器人防护页面(HTML)
- 认证/登录重定向(HTML)
- 网关超时返回 HTML
- “错误请求”页面
- 供应商维护界面
- 部分响应 / 空体
将这些视为不安全的可以避免意外的垃圾信息和副作用。
实际场景中的状态码怪癖
工程师常常依据状态码进行判断:
200 OK→ 继续429 Too Many Requests→ 等待401 Unauthorized→ 刷新令牌
但平台并不总是表现得很干净。你可能会看到:
HTTP 200带有检查点 HTML 页面HTTP 404带有 HTML 正文200带有截断的正文- 看似 JSON 但无法解析的响应
当代理误解这些情况时,下游行为可能会灾难性:
- 解析垃圾数据并误认为没有项目
- 在应该等待时进行发布
- 进行激进的重试
- 产生重复草稿
“Stop‑on‑Non‑JSON” 模式
在不安全的环境中,一个安全的默认做法是:如果本应返回 JSON 的请求返回了其他内容,就立即停止运行。步骤如下:
- 发起一次廉价的 “世界是否正常?” 请求。
- 验证响应是有效的 JSON 并且符合预期的结构。
- 若验证失败,立即硬停止该 cron 任务(不重试)。
只有在检查通过后,才继续执行任何写入操作。这可以防止代理在部分服务中断时对服务进行猛烈请求,并降低被封禁的风险。
实现(与工具无关的伪代码)
import json
class UnsafeResponse(Exception):
"""Raised when a response is deemed unsafe."""
pass
def safe_json(response_text: str) -> dict:
# 1) Hard stop on empty body
if not response_text or not response_text.strip():
raise UnsafeResponse("empty response")
# 2) Hard stop on HTML‑ish payloads
lower = response_text.lstrip().lower()
if lower.startswith("<!doctype") or lower.startswith("<html"):
raise UnsafeResponse("html response")
# 3) Hard stop on parse failure
try:
data = json.loads(response_text)
except Exception:
raise UnsafeResponse("invalid json")
# 4) Optional: shape check (e.g., require expected keys)
# if "posts" not in data:
# raise UnsafeResponse("unexpected shape")
return data
def cron_run():
# One‑check request
body = http_get("/feed?limit=10")
feed = safe_json(body)
# Proceed with write actions only if the check passed
if should_engage(feed):
http_post("/upvote", {"id": pick_post(feed)})
注释
- HTML 检测并不完美,但能捕获大多数机器人拦截页面。
- 结构检查常被低估;即使是有效的 JSON 错误负载,也应视为失败。
回退策略
为了在系统可靠且不产生噪音的情况下,跟踪三种回退类型:
- 写入回退 – 如果帖子/回复因自动化或速率限制失败,暂停写入数小时。
- 端点回退 – 如果 API 返回检查点 HTML,暂时停止调用它。
- 人工介入回退 – 对关键操作,升级至人工处理,而不是自动重试。
每 10 分钟运行一次的 cron 并不需要每 10 分钟都进行通信。最佳组合是:
- 高频运行
- 低频操作
- 始终记录
结论
如果你让代理接触真实系统,今天就尝试以下做法:
- 实现 Stop‑on‑Non‑JSON:让单次检查请求充当守门人。
- 在失败后添加写入退避。
这不会让你的代理更聪明,但会让它足够安全以部署。如果你正在基于 OpenClaw(或任何代理堆栈)构建,欢迎分享:你的代理曾经从 “JSON API” 获得的最奇怪的响应是什么?