Appliquez des politiques d'exécution sur les outils LangChain en enveloppant chaque outil dans une couche de politique qui vérifie les limites de débit, les filtres de contenu, les plafonds budgétaires et les autorisations d'accès avant l'exécution. Les agents LangChain prennent des décisions autonomes sur les outils, ce qui signifie que des outils non contraints peuvent épuiser les budgets API, divulguer des requêtes sensibles ou appeler des outils de manière inappropriée. Un wrapper de politique intercepte chaque appel d'outil, applique des règles, enregistre les violations et bloque les actions non autorisées sans modifier le code sous-jacent de l'outil.
Prérequis
- Python 3.8+ installé
- bibliothèque langchain installée
- bibliothèque requests installée
- Une clé API Scavio depuis scavio.dev
Parcours
Étape 1: Définir les règles de politique
Créez une configuration de politique qui spécifie les limites de débit, les requêtes bloquées, les plafonds budgétaires et les autorisations requises pour chaque outil.
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 = []Étape 2: Créer le wrapper de politique
Construisez un décorateur qui intercepte les appels d'outils et applique toutes les vérifications de politique avant d'autoriser l'exécution.
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'))Étape 3: Appliquer les politiques aux outils LangChain
Enveloppez les fonctions d'outil LangChain avec la couche de politique afin que chaque appel soit vérifié avant l'exécution.
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'))Étape 4: Enregistrer les violations pour examen
Enregistrez toutes les violations de politique avec contexte pour examen de sécurité et réglage de la politique.
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))Étape 5: Tester l'application
Exécutez des scénarios de test pour vérifier que les limites de débit, les filtres de contenu et les plafonds budgétaires fonctionnent correctement.
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()Exemple 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'))Exemple 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);Sortie attendue
JSON
A policy enforcement layer for LangChain tools that blocks unauthorized queries, enforces rate and budget limits, and logs all violations for review.