Las implementaciones empresariales de LangChain en 2026 necesitarán registros de auditoría y aplicación de políticas en cada llamada de herramienta. Este tutorial envuelve la herramienta Scavio LangChain con una capa de gobernanza: límites de tasa por agente, listas de consultas permitidas y registros de auditoría estructurados que alimentan un SIEM.
Requisitos previos
- Python 3.10+
- LangChain 0.2+
- Una clave API de Scavio
- Un destino de registro (Loki, Datadog, CloudWatch)
Guia paso a paso
Paso 1: Envuelva la herramienta Scavio
Subclase la herramienta LangChain con ganchos de políticas.
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()Paso 2: Agregar registro de auditoría
Cada llamada registra agent_id, consulta y recuento de resultados.
Python
def audit(agent_id, query, result):
print(json.dumps({'ts': time.time(), 'agent': agent_id, 'query': query, 'results': len(result.get('organic_results', []))}))Paso 3: Hacer cumplir la lista de permitidos
Bloquear consultas que contengan patrones bloqueados.
Python
BLOCKED = ['ssn', 'credit card']
def allowed(query):
return not any(b in query.lower() for b in BLOCKED)Paso 4: Aplicar límites de tarifas
Depósito de tokens por agente.
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 TruePaso 5: Registrar la herramienta gobernada
Los agentes utilizan el envoltorio gobernado, no el original.
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')Ejemplo en 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'))Ejemplo en 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;
}Salida esperada
JSON
Structured audit log per call, policy blocks for sensitive queries, per-agent rate limit enforcement. Ready for SOC 2 evidence collection.