Les déploiements LangChain en entreprise en 2026 nécessitent des journaux d'audit et l'application de politiques sur chaque appel d'outil. Ce tutoriel enveloppe l'outil Scavio LangChain avec une couche de gouvernance : des limites de débit par agent, des listes blanches de requêtes et des journaux d'audit structurés alimentant un SIEM.
Prérequis
- Python 3.10+
- LangChain 0.2+
- Une clé API Scavio
- Une destination de journal (Loki, Datadog, CloudWatch)
Parcours
Étape 1: Envelopper l'outil Scavio
Sous-classer l'outil LangChain avec des hooks de politique.
Python
from langchain.tools import Tool
import os, requests, json, time
def scavio_raw(query):
r = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': os.environ['SCAVIO_API_KEY']},
json={'query': query})
return r.json()Étape 2: Ajouter la journalisation d'audit
Chaque appel enregistre agent_id, la requête et le nombre de résultats.
Python
def audit(agent_id, query, result):
print(json.dumps({'ts': time.time(), 'agent': agent_id, 'query': query, 'results': len(result.get('organic_results', []))}))Étape 3: Appliquer la liste blanche
Bloquer les requêtes contenant des motifs interdits.
Python
BLOCKED = ['ssn', 'credit card']
def allowed(query):
return not any(b in query.lower() for b in BLOCKED)Étape 4: Appliquer des limites de débit
Seau à jetons par agent.
Python
from collections import defaultdict
from time import time
BUCKET = defaultdict(list)
def under_limit(agent_id, limit=100, window=3600):
now = time()
BUCKET[agent_id] = [t for t in BUCKET[agent_id] if t > now - window]
if len(BUCKET[agent_id]) >= limit: return False
BUCKET[agent_id].append(now); return TrueÉtape 5: Enregistrer l'outil gouverné
Les agents utilisent le wrapper gouverné, pas celui brut.
Python
def governed(agent_id, query):
if not allowed(query): return {'error': 'policy_block'}
if not under_limit(agent_id): return {'error': 'rate_limit'}
result = scavio_raw(query)
audit(agent_id, query, result)
return result
tool = Tool.from_function(lambda q: governed('agent-1', q), name='search', description='Governed web search')Exemple Python
Python
import os, requests, json, time
API_KEY = os.environ['SCAVIO_API_KEY']
def governed_search(agent_id, query):
if 'ssn' in query.lower(): return {'error': 'policy'}
r = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY},
json={'query': query})
print(json.dumps({'ts': time.time(), 'agent': agent_id, 'query': query}))
return r.json()
print(governed_search('agent-42', 'best serp api'))Exemple JavaScript
JavaScript
const API_KEY = process.env.SCAVIO_API_KEY;
const BLOCKED = ['ssn', 'credit card'];
export async function governedSearch(agentId, query) {
if (BLOCKED.some(b => query.toLowerCase().includes(b))) return { error: 'policy' };
const r = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST',
headers: { 'x-api-key': API_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ query })
});
const d = await r.json();
console.log(JSON.stringify({ ts: Date.now(), agent: agentId, query }));
return d;
}Sortie attendue
JSON
Structured audit log per call, policy blocks for sensitive queries, per-agent rate limit enforcement. Ready for SOC 2 evidence collection.