Galvanize en tu agente

Esta página es para desarrolladores que llaman a la API opcional de zn Cloud desde su propio código. ¿Solo quieres proteger tu agente de código? Abre tu propia terminal (no el chat del agente), ejecuta npx -y zn-gate init y después npx -y zn-gate check. Paso a paso: Instalar.

Galvanize 2 es el modelo detrás de la API de zn Cloud. Tu agente le envía un texto (un mensaje del usuario, un documento recuperado, un resultado de herramienta) y recibe allow o block antes de que el texto avance. Cada integración de abajo son unas pocas líneas alrededor de una llamada a POST https://api.usezn.com/analyze. El mismo endpoint está disponible también como /v30/analyze.

Necesitas una clave. Basta una cuenta Free: créala en el panel, copia una clave (empieza por zn_) y guárdala en una variable de entorno, no en tu código.

La API recibe el texto tal como lo envías. zn-gate quita en tu máquina los secretos que detecta antes de que nada vaya a la nube; estos snippets no lo hacen, y envían tal cual los documentos recuperados y los resultados de herramientas. No envíes claves, contraseñas ni datos personales, o redáctalos antes, por ejemplo con redact_secrets del paquete de Python de zn-gate (Redacción de secretos). Qué guarda la nube: Política de privacidad.

¿Quieres comprobaciones que se ejecuten en tu máquina, gratis y sin conexión? Usa mejor las integraciones locales: LangChain, CrewAI y LlamaIndex.

export ZN_API_KEY=zn_live_your_key

Un helper compartido por las tres

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()  # falla cerrado: un error detiene el texto, no lo deja pasar
    return r.json()

analyze("...")["verdict"] es "block" o "allow". La respuesta completa está en POST /analyze. Usa un timeout de 10 a 12 segundos: las entradas de más de 2.000 caracteres se puntúan por ventanas y pueden tardar más.

LangChain

Probado con langchain-core 1.6.12 en Python 3.11. Pon un RunnableLambda delante del prompt, o después de una herramienta, para detener un texto bloqueado antes de que lo vea el modelo. El prompt debe tener una sola variable de entrada, porque guard devuelve un 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  # usa el mismo guard con la salida de una herramienta

Instalar: pip install langchain-core requests.

LlamaIndex

Probado con llama-index-core 0.14.25. Un postprocesador de nodos descarta los fragmentos recuperados que Galvanize bloquea, así un fragmento marcado nunca llega al 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()])

Instalar: pip install llama-index-core requests. Cada fragmento, conservado o descartado, es una llamada a la API (más si pasa de 2.000 caracteres), así que recupera pocos con similarity_top_k.

OpenAI Agents SDK

Probado con openai-agents 0.24.0. Un guardrail de entrada con run_in_parallel=False se ejecuta antes de que arranque el primer agente y lanza InputGuardrailTripwireTriggered cuando la entrada está bloqueada. Sin run_in_parallel=False, el SDK ejecuta el guardrail a la vez que el agente, y el modelo puede responder o llamar a herramientas antes de que llegue el bloqueo:

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])

Instalar: pip install openai-agents requests. Captura InputGuardrailTripwireTriggered alrededor de Runner.run(agent, user_message) y responde con un mensaje fijo. Si no se puede alcanzar la API de zn o rechaza la llamada, el guardrail lanza el mismo tripwire (falla cerrado); exception.guardrail_result.output.output_info distingue los dos casos. Cuando la entrada es una lista (por ejemplo de to_input_list()), solo se comprueba el último mensaje del usuario, así que cada turno cuesta lo mismo. Para resultados de herramientas, llama a analyze dentro de la herramienta antes de que devuelva.

Límites y precios

Mismas cifras que Planes y cuotas y la página de precios. Los planes Team no aparecen aquí: consulta Planes y cuotas.

• Una llamada cubre hasta 2.000 caracteres de input. Cada 1.000 caracteres más, o fracción, es una llamada extra. El máximo es 20.000 caracteres por petición.
• Cada petición cuenta, también los reintentos y las bloqueadas. Guarda en caché el veredicto de un texto que compruebas dos veces.
• Errores: 400 (sin input, o más de 20.000 caracteres), 401 (clave incorrecta o revocada), 402 con TRIAL_ENDED (suscripción Team no activa), 429 con MONTHLY_LIMIT_REACHED (cuota agotada) o INSUFFICIENT_QUOTA (esta entrada necesita más llamadas de las que quedan). Consulta Errores.
• Tu helper lanza una excepción con esos errores, así que el agente se detiene en lugar de pasar texto sin comprobar. Para lecturas de bajo impacto puedes capturar el error y recurrir a las reglas locales de zn-gate.
• Free usa los mismos modelos que Pro y Scale: cada llamada ejecuta las reglas y Galvanize 2.
• Cuando el modelo no está disponible, la API puede responder igualmente 200 solo con las reglas: decided_by empieza entonces por ml-error (por ejemplo ml-error-rules-fallback). Si necesitas la opinión del modelo, trátalo como un error.