Haga cumplir las políticas de tiempo de ejecución en las herramientas de LangChain envolviendo cada herramienta en una capa de políticas que verifica los límites de velocidad, los filtros de contenido, los límites de presupuesto y los permisos de acceso antes de la ejecución. Los agentes de LangChain toman decisiones autónomas sobre las herramientas, lo que significa que las herramientas sin restricciones pueden agotar los presupuestos de API, filtrar consultas confidenciales o llamar a herramientas de manera inapropiada. Un contenedor de políticas intercepta cada llamada a la herramienta, aplica reglas, registra infracciones y bloquea acciones no autorizadas sin modificar el código de la herramienta subyacente.
Requisitos previos
- Python 3.8+ instalado
- biblioteca langchain instalada
- solicita biblioteca instalada
- Una clave API de Scavio de scavio.dev
Guia paso a paso
Paso 1: Definir reglas de políticas
Cree una configuración de política que especifique límites de tarifas, consultas bloqueadas, límites de presupuesto y permisos necesarios para cada herramienta.
Python
import os, time, json, requests
from collections import defaultdict
from functools import wraps
API_KEY = os.environ['SCAVIO_API_KEY']
POLICIES = {
'web_search': {
'rate_limit': 30, # calls per minute
'budget_limit': 100, # max calls per session
'blocked_patterns': ['password', 'ssn', 'credit card'],
'required_role': 'user',
},
}
usage = defaultdict(lambda: {'count': 0, 'window_start': time.time(), 'total': 0})
violation_log = []Paso 2: Crear el contenedor de políticas
Cree un decorador que intercepte llamadas a herramientas y aplique todas las comprobaciones de políticas antes de permitir la ejecución.
Python
def policy_check(tool_name: str, query: str, user_role: str = 'user') -> dict:
policy = POLICIES.get(tool_name, {})
now = time.time()
u = usage[tool_name]
# Rate limit check
if now - u['window_start'] > 60:
u['count'] = 0
u['window_start'] = now
if u['count'] >= policy.get('rate_limit', 999):
return {'allowed': False, 'reason': 'rate_limit_exceeded'}
# Budget check
if u['total'] >= policy.get('budget_limit', 9999):
return {'allowed': False, 'reason': 'budget_exceeded'}
# Content filter
query_lower = query.lower()
for pattern in policy.get('blocked_patterns', []):
if pattern in query_lower:
return {'allowed': False, 'reason': f'blocked_pattern: {pattern}'}
# Role check
required_role = policy.get('required_role', 'user')
roles_hierarchy = {'admin': 3, 'user': 2, 'guest': 1}
if roles_hierarchy.get(user_role, 0) < roles_hierarchy.get(required_role, 0):
return {'allowed': False, 'reason': 'insufficient_permissions'}
u['count'] += 1
u['total'] += 1
return {'allowed': True}
print(policy_check('web_search', 'best crm 2026'))
print(policy_check('web_search', 'find password reset link'))Paso 3: Aplicar políticas a las herramientas LangChain
La herramienta Wrap LangChain funciona con la capa de políticas para que cada llamada se verifique antes de su ejecución.
Python
def enforced_tool(tool_name: str, func):
@wraps(func)
def wrapper(query: str, **kwargs) -> str:
check = policy_check(tool_name, query)
if not check['allowed']:
violation = {
'tool': tool_name, 'query': query[:100],
'reason': check['reason'], 'time': time.time(),
}
violation_log.append(violation)
return f'[BLOCKED] {check["reason"]}'
return func(query, **kwargs)
return wrapper
def web_search(query: str) -> str:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY},
json={'platform': 'google', 'query': query}, timeout=10)
results = resp.json().get('organic_results', [])[:3]
return '\n'.join(f"{r['title']}: {r.get('snippet', '')}" for r in results)
safe_search = enforced_tool('web_search', web_search)
print(safe_search('best crm 2026'))
print(safe_search('find password reset'))Paso 4: Registrar infracciones para su revisión
Registre todas las violaciones de políticas con contexto para revisión de seguridad y ajuste de políticas.
Python
def log_violation(tool: str, query: str, reason: str):
entry = {
'tool': tool,
'query': query[:200],
'reason': reason,
'timestamp': time.time(),
}
violation_log.append(entry)
print(f'VIOLATION: {tool} - {reason} - query: {query[:50]}')
def get_violation_report() -> dict:
report = {
'total_violations': len(violation_log),
'by_reason': defaultdict(int),
'by_tool': defaultdict(int),
}
for v in violation_log:
report['by_reason'][v['reason']] += 1
report['by_tool'][v['tool']] += 1
report['by_reason'] = dict(report['by_reason'])
report['by_tool'] = dict(report['by_tool'])
return report
print(json.dumps(get_violation_report(), indent=2))Paso 5: Probar la aplicación
Ejecute escenarios de prueba para verificar que los límites de tarifas, los filtros de contenido y los límites de presupuesto funcionen correctamente.
Python
def test_policies():
# Test content filter
result = safe_search('what is my ssn number')
assert 'BLOCKED' in result, 'Content filter should block SSN query'
print('Content filter: PASS')
# Test normal query
result = safe_search('python tutorial 2026')
assert 'BLOCKED' not in result, 'Normal query should succeed'
print('Normal query: PASS')
# Test rate limiting (exhaust limit)
original_limit = POLICIES['web_search']['rate_limit']
POLICIES['web_search']['rate_limit'] = 2
usage['web_search']['count'] = 0
usage['web_search']['window_start'] = time.time()
safe_search('test query 1')
safe_search('test query 2')
result = safe_search('test query 3')
assert 'BLOCKED' in result, 'Rate limit should block third query'
print('Rate limiting: PASS')
POLICIES['web_search']['rate_limit'] = original_limit
report = get_violation_report()
print(f'Total violations logged: {report["total_violations"]}')
print('All policy tests passed')
test_policies()Ejemplo en Python
Python
import requests, os, time
from collections import defaultdict
H = {'x-api-key': os.environ['SCAVIO_API_KEY']}
usage = defaultdict(int)
def enforced_search(query, limit=30):
if usage['search'] >= limit:
return 'Rate limit exceeded'
blocked = ['password', 'ssn', 'credit card']
if any(b in query.lower() for b in blocked):
return 'Blocked by content policy'
usage['search'] += 1
data = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
json={'platform': 'google', 'query': query}).json()
return data.get('organic_results', [])[:3]
print(enforced_search('best crm 2026'))Ejemplo en JavaScript
JavaScript
const H = {'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json'};
let usage = 0;
const BLOCKED = ['password', 'ssn', 'credit card'];
async function enforcedSearch(query, limit = 30) {
if (usage >= limit) return 'Rate limit exceeded';
if (BLOCKED.some(b => query.toLowerCase().includes(b))) return 'Blocked';
usage++;
const r = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST', headers: H, body: JSON.stringify({platform: 'google', query})
});
return ((await r.json()).organic_results || []).slice(0, 3);
}
enforcedSearch('best crm 2026').then(console.log);Salida esperada
JSON
A policy enforcement layer for LangChain tools that blocks unauthorized queries, enforces rate and budget limits, and logs all violations for review.