在你的智能体中使用 Galvanize
使用免费 API 密钥,在 LangChain、LlamaIndex 或 OpenAI Agents SDK 中用 Galvanize 检查用户输入和工具结果。
本页面向在自己的代码中调用可选 zn Cloud API 的开发者。 只想保护你的编程智能体?打开你自己的终端(不是智能体的聊天窗口),运行
npx -y zn-gate init,然后运行npx -y zn-gate check。分步说明:安装。
Galvanize 2 是 zn Cloud API 背后的模型。你的智能体把一段文本(用户消息、检索到的文档、工具结果)发给它,在文本继续传递之前得到 allow 或 block。下面每个集成都只是围绕一次 POST https://api.usezn.com/analyze 调用的几行代码。同一个端点也可通过 /v30/analyze 访问。
你需要一个密钥。Free 账户就够了:在仪表板中创建账户,复制一个密钥(以 zn_ 开头),并把它放在环境变量中,而不是写进代码。
API 会原样接收你发送的文本。zn-gate 会在任何内容发往云端之前,先在你的机器上移除它检测到的密钥;这些代码片段不会这样做,检索到的文档和工具结果都会原样发送。不要发送密钥、密码或个人数据,或先把它们脱敏,例如使用 zn-gate Python 包中的 redact_secrets(密钥脱敏)。云端保留哪些内容:隐私政策。
想要在本机运行、免费且离线的检查?请改用本地集成:LangChain、CrewAI 与 LlamaIndex。
export ZN_API_KEY=zn_live_your_key
三者共用的一个 helper
import os, requests
def analyze(text: str) -> dict:
r = requests.post("https://api.usezn.com/analyze", json={"input": text}, timeout=12,
headers={"Authorization": "Bearer " + os.environ["ZN_API_KEY"]})
r.raise_for_status() # 失败即关闭:出错会拦住文本,而不是放行
return r.json()
analyze("...")["verdict"] 为 "block" 或 "allow"。完整响应见 POST /analyze。请使用 10 到 12 秒的超时:超过 2,000 个字符的输入会按窗口评分,耗时可能更长。
LangChain
已在 Python 3.11 上用 langchain-core 1.6.12 测试。在提示词之前(或工具之后)放一个 RunnableLambda,在模型看到之前拦住被阻止的文本。提示词必须只有一个输入变量,因为 guard 返回的是字符串:
from langchain_core.runnables import RunnableLambda
def guard(text: str) -> str:
if analyze(text)["verdict"] == "block":
raise ValueError("Blocked by Galvanize")
return text
chain = RunnableLambda(guard) | prompt | llm # 对工具的输出使用同一个 guard
安装:pip install langchain-core requests。
LlamaIndex
已用 llama-index-core 0.14.25 测试。节点后处理器会丢弃被 Galvanize 阻止的检索片段,因此被标记的片段不会到达模型:
from llama_index.core.postprocessor.types import BaseNodePostprocessor
class GalvanizeFilter(BaseNodePostprocessor):
def _postprocess_nodes(self, nodes, query_bundle=None):
return [n for n in nodes if analyze(n.get_content())["verdict"] == "allow"]
engine = index.as_query_engine(node_postprocessors=[GalvanizeFilter()])
安装:pip install llama-index-core requests。每个片段无论保留还是丢弃,都算一次 API 调用(超过 2,000 个字符则更多),所以请用 similarity_top_k 只检索少量片段。
OpenAI Agents SDK
已用 openai-agents 0.24.0 测试。设置了 run_in_parallel=False 的输入 guardrail 会在第一个智能体启动之前运行,当输入被阻止时抛出 InputGuardrailTripwireTriggered。如果不设置 run_in_parallel=False,SDK 会让 guardrail 与智能体同时运行,模型可能在拦截生效之前就已经回答或调用工具:
import asyncio
import requests
from agents import Agent, GuardrailFunctionOutput, input_guardrail
def last_user_text(items):
for m in reversed(items):
if isinstance(m, dict) and m.get("role") == "user":
c = m.get("content", "")
return c if isinstance(c, str) else " ".join(p.get("text", "") for p in c if isinstance(p, dict))
return ""
@input_guardrail(run_in_parallel=False)
async def galvanize(ctx, agent, user_input):
text = user_input if isinstance(user_input, str) else last_user_text(user_input)
if not text.strip():
return GuardrailFunctionOutput(output_info="empty", tripwire_triggered=False)
try:
verdict = (await asyncio.to_thread(analyze, text))["verdict"]
except requests.RequestException:
return GuardrailFunctionOutput(output_info="zn API error", tripwire_triggered=True)
return GuardrailFunctionOutput(output_info=verdict, tripwire_triggered=verdict == "block")
agent = Agent(name="Support", instructions="Help the user.", input_guardrails=[galvanize])
安装:pip install openai-agents requests。在 Runner.run(agent, user_message) 外捕获 InputGuardrailTripwireTriggered,并返回固定的回复。如果无法连接 zn API 或调用被拒绝,guardrail 会抛出同样的 tripwire(失败即关闭);exception.guardrail_result.output.output_info 可以区分这两种情况。当输入是列表时(例如来自 to_input_list()),只检查最新的一条用户消息,因此每一轮的费用相同。对于工具结果,请在工具返回之前在工具内部调用 analyze。
限制与价格
数字与套餐与配额和定价页面一致。Team 套餐不在此列出:请参见套餐与配额。
| 套餐 | 价格 | 每月调用次数 |
|---|---|---|
| Free | $0 | 10,000 |
| Pro | $19 / 月,或 $190 / 年;超额每 1,000 次调用 $0.10 | 250,000 |
| Scale | $99 / 月,或 $990 / 年 | 2,000,000 |
- 一次调用最多覆盖
input的 2,000 个字符。每多 1,000 个字符(不足也算)多算一次调用。每个请求最多 20,000 个字符。 - 每个请求都会计数,包括重试和被阻止的请求。对检查两次的文本,请缓存判定结果。
- 错误:
400(缺少input,或超过 20,000 个字符)、401(密钥错误或已撤销)、402和TRIAL_ENDED(Team 订阅未生效)、429和MONTHLY_LIMIT_REACHED(配额用完)或INSUFFICIENT_QUOTA(此输入所需调用次数超过剩余次数)。参见错误。 - 遇到这些错误时 helper 会抛出异常,因此智能体会停下,而不是把未检查的文本传下去。对于低影响的读取操作,你可以捕获错误并回退到本地 zn-gate 规则。
- Free 使用与 Pro 和 Scale 相同的模型:每次调用都会运行规则和 Galvanize 2。
- 模型不可用时,API 仍可能返回
200,但只使用规则:此时decided_by以ml-error开头(例如ml-error-rules-fallback)。如果你需要模型的判断,请把它当作错误处理。