2026 年的企业 LangChain 部署需要对每个工具调用进行审核日志和策略执行。本教程将 Scavio LangChain 工具与治理层包装在一起：每个代理的速率限制、查询允许列表以及为 SIEM 提供数据的结构化审核日志。
前置条件
- Python 3.10+
- 浪链 0.2+
- Scavio API 密钥
- 日志目的地（Loki、Datadog、CloudWatch）
操作指南
步骤 1: 包裹 Scavio 工具
使用策略挂钩对 LangChain 工具进行子类化。
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()步骤 2: 添加审核日志记录
每次调用都会记录agent_id、查询、结果计数。
Python
def audit(agent_id, query, result):
print(json.dumps({'ts': time.time(), 'agent': agent_id, 'query': query, 'results': len(result.get('organic_results', []))}))步骤 3: 强制执行许可名单
阻止包含阻止模式的查询。
Python
BLOCKED = ['ssn', 'credit card']
def allowed(query):
return not any(b in query.lower() for b in BLOCKED)步骤 4: 应用速率限制
每个代理令牌桶。
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步骤 5: 注册受管工具
代理使用受管理的包装器，而不是原始包装器。
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')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'))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;
}预期输出
JSON
Structured audit log per call, policy blocks for sensitive queries, per-agent rate limit enforcement. Ready for SOC 2 evidence collection.