Les chatbots des secteurs réglementés (banque, assurance, santé) ont besoin de RAG avec masquage des PII, contrôle strict des sources et journalisation d'audit. Ce tutoriel construit cette architecture en utilisant Scavio pour l'ancrage sur le web public et un nettoyeur local de PII.
Prérequis
- Python 3.10+
- Une clé API Scavio
- Presidio ou un détecteur de PII équivalent
- Postgres avec pgvector pour les journaux de citations
Parcours
Étape 1: Nettoyer les entrées pour les PII
Ne jamais envoyer de PII aux API en aval.
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine
analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()
def scrub(text):
results = analyzer.analyze(text=text, language='en')
return anonymizer.anonymize(text=text, analyzer_results=results).textÉtape 2: Ancrer la réponse avec Scavio
Récupérer du contenu public faisant autorité pour compléter les documents internes.
import os, requests
API_KEY = os.environ['SCAVIO_API_KEY']
def ground(question):
r = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY},
json={'query': question, 'platform': 'google'})
return r.json().get('organic_results', [])[:5]Étape 3: Journaliser chaque citation
Les régulateurs exigent une provenance pour chaque déclaration du chatbot.
import psycopg2, json
def log_citation(session_id, question, sources):
conn = psycopg2.connect(os.environ['DATABASE_URL'])
with conn.cursor() as c:
c.execute('INSERT INTO citations(session_id, question, sources_json) VALUES (%s, %s, %s)',
(session_id, question, json.dumps(sources)))
conn.commit()Étape 4: Composer la réponse avec un prompt protégé
Demander au LLM de citer chaque affirmation et de refuser en cas de doute.
SYSTEM = '''You are a regulated-industry assistant.
Rules:
1. Cite every factual claim with [source N].
2. If unsure, say "I cannot answer with confidence."
3. Never repeat customer PII back in the answer.'''Étape 5: Imposer un parcours de révision humaine
Les réponses à faible confiance sont dirigées vers un humain. Chaque réponse a un score de risque.
def risk_score(answer, sources):
if not sources: return 1.0
if 'cannot answer' in answer.lower(): return 0.2
return 0.5Exemple Python
import os, requests
API_KEY = os.environ['SCAVIO_API_KEY']
def rag_answer(question):
clean = scrub(question)
sources = ground(clean)
log_citation('sess-1', clean, sources)
return {'question': clean, 'sources': sources}
print(rag_answer('What are the KYC rules for my account?'))Exemple JavaScript
const API_KEY = process.env.SCAVIO_API_KEY;
export async function ragAnswer(question) {
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: question, platform: 'google' })
});
const sources = ((await r.json()).organic_results || []).slice(0, 5);
return { question, sources };
}Sortie attendue
PII-scrubbed questions, typed public-web citations, durable audit log of every answer. Compliance team gets a per-session export on demand.