Galvanize dans votre agent

Cette page s'adresse aux développeurs qui appellent l'API optionnelle zn Cloud depuis leur propre code. Vous voulez seulement protéger votre agent de code ? Ouvrez votre propre terminal (pas le chat de l'agent), lancez npx -y zn-gate init, puis npx -y zn-gate check. Pas à pas : Installer.

Galvanize 2 est le modèle derrière l'API zn Cloud. Votre agent lui envoie un texte (un message d'utilisateur, un document récupéré, un résultat d'outil) et reçoit allow ou block avant que le texte n'aille plus loin. Chaque intégration ci-dessous tient en quelques lignes autour d'un appel à POST https://api.usezn.com/analyze. Le même endpoint est aussi disponible sous /v30/analyze.

Il vous faut une clé. Un compte Free suffit : créez-le dans le tableau de bord, copiez une clé (elle commence par zn_) et gardez-la dans une variable d'environnement, pas dans votre code.

L'API reçoit le texte tel que vous l'envoyez. zn-gate retire sur votre machine les secrets qu'il détecte avant que quoi que ce soit parte vers le cloud ; ces snippets ne le font pas, et envoient tels quels les documents récupérés et les résultats d'outils. N'envoyez ni clés, ni mots de passe, ni données personnelles, ou masquez-les avant, par exemple avec redact_secrets du paquet Python de zn-gate (Caviardage des secrets). Ce que le cloud conserve : Politique de confidentialité.

Vous voulez des vérifications qui s'exécutent sur votre machine, gratuites et hors ligne ? Utilisez plutôt les intégrations locales : LangChain, CrewAI et LlamaIndex.

export ZN_API_KEY=zn_live_your_key

Un helper partagé par les trois

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()  # échec fermé : une erreur arrête le texte, elle ne le laisse pas passer
    return r.json()

analyze("...")["verdict"] vaut "block" ou "allow". La réponse complète est décrite dans POST /analyze. Utilisez un timeout de 10 à 12 secondes : les entrées de plus de 2 000 caractères sont évaluées par fenêtres et peuvent prendre plus de temps.

LangChain

Testé avec langchain-core 1.6.12 sur Python 3.11. Placez un RunnableLambda devant le prompt, ou après un outil, pour arrêter un texte bloqué avant que le modèle ne le voie. Le prompt doit avoir une seule variable d'entrée, car guard renvoie une chaîne :

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  # utilisez le même guard sur la sortie d'un outil

Installation : pip install langchain-core requests.

LlamaIndex

Testé avec llama-index-core 0.14.25. Un postprocesseur de nœuds écarte les passages récupérés que Galvanize bloque, de sorte qu'un passage signalé n'atteint jamais le modèle :

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

Installation : pip install llama-index-core requests. Chaque passage, conservé ou écarté, correspond à un appel d'API (davantage au-delà de 2 000 caractères) : récupérez-en peu avec similarity_top_k.

OpenAI Agents SDK

Testé avec openai-agents 0.24.0. Un guardrail d'entrée avec run_in_parallel=False s'exécute avant le démarrage du premier agent et lève InputGuardrailTripwireTriggered quand l'entrée est bloquée. Sans run_in_parallel=False, le SDK exécute le guardrail en même temps que l'agent, et le modèle peut répondre ou appeler des outils avant que le blocage n'arrive :

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

Installation : pip install openai-agents requests. Interceptez InputGuardrailTripwireTriggered autour de Runner.run(agent, user_message) et répondez par un message fixe. Si l'API zn est injoignable ou rejette l'appel, le guardrail lève le même tripwire (échec fermé) ; exception.guardrail_result.output.output_info distingue les deux cas. Quand l'entrée est une liste (par exemple issue de to_input_list()), seul le dernier message de l'utilisateur est vérifié, donc chaque tour coûte autant. Pour les résultats d'outils, appelez analyze dans l'outil avant qu'il ne renvoie.

Limites et prix

Mêmes chiffres que Plans et quotas et que la page des tarifs. Les plans Team ne figurent pas ici : voir Plans et quotas.

• Un appel couvre jusqu'à 2 000 caractères d'input. Chaque tranche de 1 000 caractères supplémentaires, même entamée, compte pour un appel de plus. Le maximum est de 20 000 caractères par requête.
• Chaque requête compte, y compris les nouvelles tentatives et les requêtes bloquées. Mettez en cache le verdict d'un texte que vous vérifiez deux fois.
• Erreurs : 400 (pas d'input, ou plus de 20 000 caractères), 401 (clé incorrecte ou révoquée), 402 avec TRIAL_ENDED (abonnement Team inactif), 429 avec MONTHLY_LIMIT_REACHED (quota épuisé) ou INSUFFICIENT_QUOTA (cette entrée demande plus d'appels qu'il n'en reste). Voir Erreurs.
• Votre helper lève une exception sur ces erreurs : l'agent s'arrête au lieu de transmettre un texte non vérifié. Pour des lectures à faible impact, vous pouvez intercepter l'erreur et vous rabattre sur les règles locales de zn-gate.
• Free utilise les mêmes modèles que Pro et Scale : chaque appel exécute les règles et Galvanize 2.
• Quand le modèle est indisponible, l'API peut quand même répondre 200 avec les seules règles : decided_by commence alors par ml-error (par exemple ml-error-rules-fallback). Si vous avez besoin de l'avis du modèle, traitez-le comme une erreur.