Aller au contenu

Plugin Trino AI

Le plugin trino-ai-functions ajoute 23 fonctions IA scalaires natives à Trino, toutes préfixées akko_ai_*. Contrairement aux approches par connecteur, ces fonctions marchent sur n'importe quel catalogue Trino — Iceberg, PostgreSQL, Hive, Delta, S3, etc.

-- Fonctionne sur toute source
SELECT name, akko_ai_sentiment(description) FROM iceberg.banking.transactions;
SELECT akko_ai_pii(comment)                 FROM postgresql.public.reviews;

Pourquoi le préfixe akko_ai_*

Trino 466+ embarque ses propres 7 fonctions ai_* dans le catalogue llm. AKKO utilise akko_ai_* pour éviter toute collision et rendre la différenciation évidente (cache, OPA, multimodal, audit). Voir IA / Fonctions IA Trino et ADR-026.

Architecture

Requête SQL Trino
  └─→ plugin trino-ai-functions (Java, dans la JVM Trino)
       └─→ HTTPS / HTTP/2 (java.net.http)
            └─→ akko-ai-service (FastAPI + check OPA)
                 └─→ passerelle LiteLLM
                      └─→ Ollama / vLLM / OpenAI / etc.

Le plugin tourne dans la JVM du coordinateur Trino. Chaque appel SQL akko_ai_*() est une invocation de fonction qui :

  1. Construit un GET HTTP vers l'AI Service.
  2. Transmet l'identité de l'utilisateur via X-Trino-User et le nom de la fonction via X-Akko-Ai-Function: akko_ai_<nom>.
  3. S'authentifie par bearer token (AKKO_AI_SERVICE_TOKEN).
  4. Renvoie le résultat parsé, ou NULL en cas d'erreur.

Fonctions (23 au total)

Fonction Rôle Exemple
akko_ai_ask(question[, contexte]) Question/réponse akko_ai_ask('Qu''est-ce qu''un lakehouse?')
akko_ai_sentiment(texte) Sentiment akko_ai_sentiment('J''adore') → POSITIVE
akko_ai_classify(texte, catégories) Classification akko_ai_classify('serveur planté', 'bug,feature')
akko_ai_summarize(texte[, n]) Résumé akko_ai_summarize('long texte', 2)
akko_ai_translate(texte[, langue]) Traduction akko_ai_translate('Hello', 'French')
akko_ai_entities(texte) NER akko_ai_entities('Tim Cook chez Apple')
akko_ai_anomaly(valeur[, contexte]) Détection d'anomalie akko_ai_anomaly('150000', 'moyenne 45000')
akko_ai_sql(question[, schéma]) Langage naturel → SQL akko_ai_sql('Top clients par CA')
akko_ai_risk(profil) Score de risque 0-100 akko_ai_risk('solde=-5000, inactif')
akko_ai_pii(texte) Masquage PII akko_ai_pii('Jean, jean@mail.com')
akko_ai_sensitivity(texte) Classe RGPD akko_ai_sensitivity('NIR : 1 85 07...')
akko_ai_language(texte) Détection de langue akko_ai_language('Bonjour')
akko_ai_keywords(texte[, n]) Extraction de mots-clés akko_ai_keywords('détection fraude ML', 3)
akko_ai_ocr(image) Image → texte akko_ai_ocr(pdf_bytes)
akko_ai_describe_image(image[, prompt]) Légende d'image akko_ai_describe_image(img)
akko_ai_parse_document(pdf) PDF → JSON structuré akko_ai_parse_document(facture)
akko_ai_transcribe(audio) Audio → texte akko_ai_transcribe(call_bytes)
akko_ai_embed(texte) Embedding 768 dim akko_ai_embed('comment rembourser un client')
akko_ai_similarity(vec_a, vec_b) Similarité cosinus (local) akko_ai_similarity(a.embedding, b.embedding)
akko_ai_search(embedding, requête) Score de recherche sémantique (caché) akko_ai_search(doc.embedding, 'politique remboursement')

Plus les helpers opérationnels : akko_ai_stats(), akko_ai_health(), akko_ai_version(), akko_ai_cache_clear(), akko_ai_cb_reset().

Pattern de recherche sémantique

akko_ai_search(embedding, texte_requête) est la façon canonique de classer des lignes par pertinence sémantique. En interne il appelle akko_ai_embed(texte_requête) exactement une fois par worker (LRU, 1024 requêtes distinctes), puis calcule la similarité cosinus localement pour chaque ligne — zéro appel HTTP par ligne :

SELECT id, texte, akko_ai_search(embedding, 'comment rembourser ?') AS score
FROM iceberg.kb.documents
ORDER BY score DESC
LIMIT 10;

Équivalent mais 10 à 1000× plus lent (ré-embedding par ligne) :

-- À ÉVITER en production
SELECT id, texte, akko_ai_similarity(embedding, akko_ai_embed('comment rembourser ?')) AS score
FROM iceberg.kb.documents ORDER BY score DESC LIMIT 10;

Agnostique au stockage : la colonne embedding peut vivre dans Iceberg, Delta, Hive, PostgreSQL (vector ou ARRAY<DOUBLE>), Hudi — n'importe quel catalogue Trino qui expose un ARRAY<DOUBLE>.

Caractéristiques production

  • Résilience — circuit breaker (5 échecs → OPEN, 60 s recovery), retry exponentiel (3 tentatives), timeout par requête
  • Performance — multiplexing HTTP/2, pool de connexions, cache LRU Caffeine (10 k entrées, TTL 1 h, par utilisateur)
  • Observabilité — tracking de latence HdrHistogram (global + par fonction), MBean JMX exposé via le catalogue jmx de Trino et jmx_exporter pour Prometheus, spans Tempo côté AI Service
  • Sécurité — bearer token depuis K8s Secret, transmission de X-Trino-User, HTTPS avec option de skip de vérification de cert pour dev, allow-list OPA par fonction appliquée côté serveur
  • Robustesse — toutes les erreurs → SQL NULL via @SqlNullable (Trino n'est jamais déstabilisé)

Monitoring

Depuis SQL :

SELECT akko_ai_stats();
-- JSON avec circuit_breaker, latence p50/p95/p99, taux de hit cache, breakdown par fonction

SELECT * FROM jmx.current."dev.akko.trino.ai:type=aihttpclient";
-- 18 attributs : totaux, percentiles latence, version du plugin

Pour Prometheus, utiliser l'agent jmx_exporter standard attaché à la JVM Trino. Les métriques apparaissent comme :

trino_dev_akko_trino_ai_aihttpclient_total_requests
trino_dev_akko_trino_ai_aihttpclient_total_errors
trino_dev_akko_trino_ai_aihttpclient_latency_p95_micros
trino_dev_akko_trino_ai_aihttpclient_circuit_breaker_trips

Configuration

Tout se règle via les variables d'environnement du coordinateur Trino (dans helm/akko/values.yaml sous trino.env).

Variable Défaut Rôle
AKKO_AI_SERVICE_URL http://akko-akko-ai-service:8000 URL de base de l'AI Service
AKKO_AI_SERVICE_TOKEN (depuis Secret akko-trino-ai) Bearer token
AKKO_AI_TIMEOUT_MS 30000 Timeout par requête
AKKO_AI_CB_THRESHOLD 5 Échecs avant ouverture circuit
AKKO_AI_CB_RECOVERY_MS 60000 Temps avant HALF_OPEN
AKKO_AI_RETRY_MAX 3 Tentatives max
AKKO_AI_POOL_SIZE 16 Taille pool HTTP
AKKO_AI_CACHE_SIZE 10000 Entrées max du cache LRU
AKKO_AI_CACHE_TTL_S 3600 TTL d'entrée cache (s)
AKKO_AI_VERIFY_TLS true Skip vérif TLS (dev seulement)

Voir aussi