Galvanize no seu agente
Verifique a entrada do usuário e os resultados de ferramentas com o Galvanize a partir do LangChain, LlamaIndex ou OpenAI Agents SDK, com uma chave de API gratuita.
Esta página é para desenvolvedores que chamam a API opcional do zn Cloud a partir do próprio código. Só quer proteger o seu agente de código? Abra o seu próprio terminal (não o chat do agente), execute
npx -y zn-gate inite depoisnpx -y zn-gate check. Passo a passo: Instalar.
O Galvanize 2 é o modelo por trás da API do zn Cloud. Seu agente envia um texto (uma mensagem do usuário, um documento recuperado, um resultado de ferramenta) e recebe allow ou block antes de o texto seguir adiante. Cada integração abaixo são poucas linhas em volta de uma chamada a POST https://api.usezn.com/analyze. O mesmo endpoint também está disponível como /v30/analyze.
Você precisa de uma chave. Uma conta Free basta: crie-a no painel, copie uma chave (ela começa com zn_) e guarde-a em uma variável de ambiente, não no seu código.
A API recebe o texto exatamente como você o envia. O zn-gate remove na sua máquina os segredos que detecta antes que qualquer coisa vá para a nuvem; estes snippets não fazem isso, e enviam como estão os documentos recuperados e os resultados de ferramentas. Não envie chaves, senhas nem dados pessoais, ou oculte-os antes, por exemplo com redact_secrets do pacote Python do zn-gate (Redação de segredos). O que a nuvem guarda: Política de Privacidade.
Quer verificações que rodam na sua máquina, grátis e offline? Use as integrações locais: LangChain, CrewAI e LlamaIndex.
export ZN_API_KEY=zn_live_your_key
Um helper compartilhado pelas três
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() # falha fechada: um erro interrompe o texto, não o deixa passar
return r.json()
analyze("...")["verdict"] é "block" ou "allow". A resposta completa está em POST /analyze. Use um timeout de 10 a 12 segundos: entradas com mais de 2.000 caracteres são avaliadas em janelas e podem demorar mais.
LangChain
Testado com langchain-core 1.6.12 no Python 3.11. Coloque um RunnableLambda antes do prompt, ou depois de uma ferramenta, para barrar um texto bloqueado antes que o modelo o veja. O prompt deve ter uma única variável de entrada, porque guard retorna uma string:
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 # use o mesmo guard na saída de uma ferramenta
Instalação: pip install langchain-core requests.
LlamaIndex
Testado com llama-index-core 0.14.25. Um pós-processador de nós descarta os trechos recuperados que o Galvanize bloqueia, de modo que um trecho sinalizado nunca chega ao modelo:
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()])
Instalação: pip install llama-index-core requests. Cada trecho, mantido ou descartado, é uma chamada à API (mais se passar de 2.000 caracteres); por isso recupere poucos com similarity_top_k.
OpenAI Agents SDK
Testado com openai-agents 0.24.0. Um guardrail de entrada com run_in_parallel=False roda antes de o primeiro agente começar e lança InputGuardrailTripwireTriggered quando a entrada é bloqueada. Sem run_in_parallel=False, o SDK roda o guardrail ao mesmo tempo que o agente, e o modelo pode responder ou chamar ferramentas antes de o bloqueio chegar:
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])
Instalação: pip install openai-agents requests. Capture InputGuardrailTripwireTriggered em volta de Runner.run(agent, user_message) e responda com uma mensagem fixa. Se a API da zn não puder ser alcançada ou rejeitar a chamada, o guardrail lança o mesmo tripwire (falha fechada); exception.guardrail_result.output.output_info distingue os dois casos. Quando a entrada é uma lista (por exemplo de to_input_list()), só a última mensagem do usuário é verificada, então cada turno custa o mesmo. Para resultados de ferramentas, chame analyze dentro da ferramenta antes de ela retornar.
Limites e preços
Os mesmos números de Planos e cotas e da página de preços. Os planos Team não estão listados aqui: veja Planos e cotas.
| Plano | Preço | Chamadas / mês |
|---|---|---|
| Free | $0 | 10.000 |
| Pro | $19 / mês, ou $190 / ano; excedente a $0,10 por 1.000 chamadas | 250.000 |
| Scale | $99 / mês, ou $990 / ano | 2.000.000 |
- Uma chamada cobre até 2.000 caracteres de
input. Cada 1.000 caracteres a mais, ou fração, é uma chamada extra. O máximo é 20.000 caracteres por requisição. - Toda requisição conta, inclusive novas tentativas e as bloqueadas. Guarde em cache o veredito de um texto que você verifica duas vezes.
- Erros:
400(seminput, ou mais de 20.000 caracteres),401(chave incorreta ou revogada),402comTRIAL_ENDED(assinatura Team inativa),429comMONTHLY_LIMIT_REACHED(cota esgotada) ouINSUFFICIENT_QUOTA(esta entrada precisa de mais chamadas do que restam). Veja Erros. - Seu helper lança uma exceção nesses erros, então o agente para em vez de repassar texto não verificado. Em leituras de baixo impacto, você pode capturar o erro e recorrer às regras locais do zn-gate.
- O Free usa os mesmos modelos que o Pro e o Scale: toda chamada executa as regras e o Galvanize 2.
- Quando o modelo está indisponível, a API ainda pode responder
200só com as regras:decided_bypassa a começar comml-error(por exemploml-error-rules-fallback). Se você precisa da opinião do modelo, trate isso como um erro.