Boucle d'appel des agents IA personnalisés post-démo et stratégies de contrôle des coûts de tokens
makedream2026년 7월 28일
0
Computing/Software
Comments (0)
Log in to leave a comment
No posts yet
makedreammakedreamLog in to leave a comment
No posts yet
Importer un framework et créer un agent IA avec des données internes se déroule généralement sans heurts jusqu'à la phase de prototype. Le véritable problème commence au moment où ce code est mis en production. Un agent qui fonctionnait parfaitement à l'écran se retrouve piégé par une saisie inattendue, répétant ses appels à l'infini, et générant parfois des milliers d'euros de frais d'API pendant le week-end.
La majeure parte des pannes constatées sur le terrain ne provient pas des limites du modèle LLM lui-même, mais plutôt de pannes de gestion d'état et de l'absence de contrôle périphérique. Bien que vous ayez conçu un outil personnalisé à la demande de la direction, il est temps de mettre fin à cette situation où vous ne pouvez plus accomplir votre travail principal parce que vous passez vos journées à déboguer des hallucinations d'agents et à traiter des factures de tokens vertigineuses.
Une fois sorti de l'environnement de démo, l'agent se heurte à trois obstacles majeurs : l'opération en boîte noire, les boucles non déterministes et les risques de fuite de données.
Si les entrées et sorties de chaque étape — processus de raisonnement (Chain-of-Thought), appels d'outils externes (Tool Call) et recherches dans la base de données vectorielle — ne sont pas conservées dans les logs, il devient impossible d'identifier la cause d'un problème. Localiser l'endroit où le prompt a déraillé peut nécessiter une journée entière de recherche.
Le problème devient plus grave lorsqu'il s'agit d'une boucle infinie. Un agent doté d'une architecture ReAct (Reasoning + Acting) renvoie continuellement la même requête si le résultat de l'exécution d'un outil est ambigu.
┌─────────────────────────────────────────────────────────────────────────┐ │ ReAct Architecture Loop │ │ │ │ ┌────────────┐ User Query ┌────────────┐ Tool Call Request │ │ │ User │ ──────────────> │ Main LLM │ ────────────────────┐ │ │ └────────────┘ └────────────┘ │ │ │ ▲ ▼ │ │ │ Observation ┌──────┐ │ │ │ (Ambiguous/Failed) │ Tool │ │ │ └─────────────────────── │ A │ │ │ └──────┘ │ │ * Problem: When Observation fails, LLM retries Tool A endlessly. │ └─────────────────────────────────────────────────────────────────────────┘
Dès lors qu'il n'arrive pas à sortir des conditions de fin et sollicite sans cesse l'API avec les mêmes arguments, le budget mensuel de tokens s'épuise en quelques minutes. Les pannes en chaîne, où l'agent principal et les sous-agents s'appellent mutuellement et réessayent en boucle, représentent plus de 30 % de la cause de toutes les pannes système.
Un autre problème survient lorsque l'historique de conversation s'accumule continuellement et remplit la fenêtre de contexte, ce qui dégrade les performances mêmes du modèle. De plus, les prompts codés en dur sous forme de chaînes de caractères dans le code Python vous obligent à recompiler et redéployer l'ensemble du système, même pour modifier une simple tournure de phrase.
Il convient de séparer la solution de débogage local de l'outil d'observabilité en production afin de définir précisément les points de collecte. En phase de développement, on s'appuie sur Arize Phoenix basé sur OpenTelemetry pour vérifier la qualité des plongements (embeddings) RAG et les appels d'outils. En environnement de production, on intègre Langfuse avec ClickHouse en backend pour surveiller en temps réel les logs de traçabilité et la consommation de tokens.
┌─────────────────────────────────────────────────────────────────────────┐ │ Semantic Caching & Routing Flow │ │ │ │ Client Query │ │ │ │ │ ▼ │ │ ┌───────────┐ Similarity >= 0.92? ┌─────────────────────────────┐ │ │ │ Redis Vector │ ─────────────────────────> │ Return Cached Response │ │ │ │ Cache │ (Cache Hit) │ (Latency -88%, Cost -86%) │ │ │ └───────────┘ └─────────────────────────────┘ │ │ │ │ │ │ (Cache Miss) │ │ ▼ │ │ ┌───────────┐ Execution & Save Cache ┌─────────────────────────────┐ │ │ │ External │ ────────────────────────> │ Store Result as Vector in │ │ │ │ LLM API │ │ Backend Database │ │ │ └───────────┘ └─────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────────────┘
Les frais d'API générés par la répétition des mêmes requêtes sont bloqués par une couche de mise en cache sémantique (Semantic Caching).
D'après les données d'analyse publiées par AWS sur 60 000 requêtes, la mise en place d'un cache sémantique basé sur la similitude cosinus permet de réduire les coûts d'inférence LLM jusqu'à 86 % et d'améliorer le temps de réponse de 88 %. Le framework RouteLLM de LMSYS permet également de réduire les coûts de 85 % en répartissant les requêtes entre modèles à coût élevé et modèles à bas coût en fonction de la complexité des questions.
La mise en cache sémantique est configurée selon les étapes suivantes :
Pour bloquer fondamentalement les boucles infinies, il faut intégrer directement un modèle de coupe-circuit (Circuit Breaker) vérifiant l'état de l'agent dans le flux de contrôle du pipeline. S'en remettre aux options de réessai par défaut fournies par les frameworks risque de provoquer des exceptions qui paralyseront l'ensemble du service. La méthode la plus sûre consiste à hasher le nom de l'outil et les valeurs des arguments avec SHA-256 pour les ajouter à une liste, puis à forcer la bascule vers un autre nœud si le même hash est accumulé 3 fois consécutives.
`python
import hashlib
from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
steps: int
tool_hashes: List[str]
def hash_tool_call(tool_name: str, tool_args: str) -> str:
raw_str = f"{tool_name}:{tool_args}"
return hashlib.sha256(raw_str.encode('utf-8')).hexdigest()
def agent_circuit_breaker_router(state: AgentState) -> str:
# 1. Si le nombre total d'exécutions dépasse 5, redirection vers le nœud de secours (fallback)
if state["steps"] > 5:
return "fallback_graceful_node"
# 2. Bloquer immédiatement si le même Tool avec le même Argument est appelé 3 fois de suite
hashes = state.get("tool_hashes", [])
if len(hashes) >= 3 and hashes[-1] == hashes[-2] == hashes[-3]:
return "fallback_graceful_node"
# 3. Vérifier le mot-clé de fin normale
last_message = state["messages"][-1]
if "FINAL_ANSWER" in last_message.content:
return "end"
return "continue_tools"
`
Indépendamment des sorties non déterministes du LLM, ce routeur évalue les conditions directement dans l'environnement d'exécution Python, ce qui empêche physiquement l'entrée dans une boucle causée par une hallucination.
Abandonner les prompts dans le code empêche de détecter les erreurs de régression (Regression Failure), où des fonctionnalités qui marchaient bien deviennent inutilisables après avoir modifié le modèle. Les prompts doivent être extraits du code Python et gérés dans des fichiers YAML indépendants.
`yaml
name: "agent_reasoning"
version: "1.2.0"
model: "gpt-4o"
temperature: 0.1
messages:
Les prompts ainsi séparés sont automatiquement testés dans le pipeline CI/CD en combinant Pytest avec DeepEval, un framework d'évaluation open source. Les métriques G-Eval servent de référence pour valider la qualité des réponses du modèle avant tout déploiement.
`python
import pytest
from deepeval import assert_test
from deepeval.metrics import GEval, TaskCompletionMetric
from deepeval.test_case import LLMTestCase, SingleTurnParams
correctness_metric = GEval(
name="Précision et respect du schéma",
criteria="La réponse du LLM répond-elle précisément à la question et respecte-t-elle parfaitement le format JSON demandé ?",
evaluation_params=[SingleTurnParams.ACTUAL_OUTPUT, SingleTurnParams.EXPECTED_OUTPUT],
threshold=0.7
)
@pytest.mark.parametrize(
"user_input, expected_output",
[
("2024년 1분기 매출 데이터를 요약해줘.", "1분기 총 매출은 50억 원입니다."),
("퇴직금 계산 규정을 알려줘.", "퇴직금은 근속연수 1년에 대해 30일분 이상의 평균임금입니다.")
]
)
def test_agent_regression(user_input, expected_output):
actual_output = run_in_house_agent(user_input)
test_case = LLMTestCase(
input=user_input,
actual_output=actual_output,
expected_output=expected_output
)
# Interrompre le build si le score est inférieur au seuil défini
assert_test(test_case, [correctness_metric, TaskCompletionMetric(threshold=0.8)])
`
Le système d'évaluation s'articule autour de trois étapes :
deepeval test run s'exécute automatiquement.`yaml
name: AI Agent Evaluation Gate
on:
pull_request:
branches: [ main ]
jobs:
eval-gate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install Dependencies
run: |
pip install poetry
poetry install
- name: Run DeepEval Suite
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: |
poetry run deepeval test run tests/test_evals.py
`
Pour empêcher les fuites de données internes, un garde-fou doit dépersonnaliser les données d'identification personnelle (PII) avant que les requêtes API ne sortent du réseau. L'intégration du moteur Microsoft Presidio au niveau de la passerelle permet de masquer automatiquement les numéros d'enregistrement d'état civil, les e-mails, les numéros de téléphone et les identifiants d'employés.
`python
from presidio_analyzer import AnalyzerEngine, PatternRecognizer
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig
analyzer = AnalyzerEngine()
employee_id_recognizer = PatternRecognizer(
supported_entity="EMPLOYEE_ID",
regex="EMP-[0-9]{6}",
score=0.95
)
analyzer.registry.add_recognizer(employee_id_recognizer)
anonymizer = AnonymizerEngine()
def sanitize_user_prompt(raw_prompt: str) -> str:
results = analyzer.analyze(
text=raw_prompt,
entities=["PERSON", "PHONE_NUMBER", "EMAIL_ADDRESS", "EMPLOYEE_ID"],
language="en"
)
anonymized_result = anonymizer.anonymize(
text=raw_prompt,
analyzer_results=results,
operators={
"DEFAULT": OperatorConfig("replace", {"new_value": "<REDACTED>"}),
"EMPLOYEE_ID": OperatorConfig("mask", {"chars_to_mask": 6, "masking_char": "*", "from_end": True})
}
)
return anonymized_result.text
`
Lors de l'intégration d'une recherche RAG, il convient également de veiller à ne récupérer que les documents correspondant au niveau d'autorisation de l'utilisateur. Les bases de données vectorielles telles que Qdrant ou Pinecone gèrent cette séparation des droits via le filtrage de métadonnées (Metadata Filtering) lors de la recherche par similitude.
`python
from qdrant_client import QdrantClient
from qdrant_client.http import models
client = QdrantClient(host="localhost", port=6333)
def search_documents_with_rbac(query_vector: list, user_department: str, user_clearance_level: int):
search_result = client.search(
collection_name="enterprise_knowledge_base",
query_vector=query_vector,
query_filter=models.Filter(
must=[
models.FieldCondition(
key="department",
match=models.MatchValue(value=user_department)
),
models.FieldCondition(
key="security_level",
range=models.Range(lte=user_clearance_level)
)
]
),
limit=5
)
return search_result
`
La règle fondamentale pour se protéger contre l'injection de prompts consiste à séparer complètement le canal du prompt système de celui des saisies utilisateur. Les opérations critiques modifiant la base de données ou appelant des API externes doivent nécessiter une validation humaine (Human-in-the-Loop) ou voir leurs exécutions restreintes à un bac à sable (sandbox) isolé.
L'exploitation continue d'un système en interne en remplacement de modules SaaS nécessite une routine de contrôle périodique.
| Fréquence | Élément de contrôle opérationnel | Action détaillée |
|---|---|---|
| Quotidien | Taux d'erreur et consommation de tokens | Vérification du taux d'erreur HTTP 5xx et de la consommation de tokens par département sur le tableau de bord Langfuse |
| Quotidien | Historique de blocage des coupe-circuits | Collecte et correction des noms d'outils et des motifs d'arguments des sessions interrompues par la détection de boucle |
| Hebdomadaire | Extraction des requêtes en échec de recherche RAG | Sélection des requêtes avec un score de similitude inférieur à 0,6 pour les ajouter au jeu de données de test |
| Hebdomadaire | Vérification des faux positifs du masquage PII | Échantillonnage des logs Presidio pour vérifier l'absence d'omissions de masquage |
| Mensuel | Mise à jour des critères d'évaluation CI/CD | Modification des cas de test automatisés pour refléter les évolutions métier |
Si vous hésitez entre un hébergement propre (basé sur vLLM) et un abonnement à une API externe, faites le calcul en vous basant sur votre trafic quotidien.
L'abonnement mensuel pour une instance AWS EC2 g5.2xlarge équipée d'un GPU NVIDIA A10G s'élève à environ 880 $. À l'inverse, le tarif de l'API GPT-4o est de 2,50 $ par million de tokens en entrée et de 10,00 $ par million de tokens en sortie. Pour un trafic quotidien inférieur à 50 millions de tokens, la formule d'abonnement API reste plus avantageuse compte tenu de la charge de maintenance DevOps et des coûts fixes de GPU. Il devient pertinent de basculer vers un auto-hébergement basé sur vLLM lorsque le trafic dépasse 100 millions de tokens par jour ou qu'un isolement strict sur réseau interne est obligatoire.
Pour pallier les pannes du modèle principal, mettez en place une architecture de routage de secours à 4 niveaux.
┌─────────────────────────────────────────────────────────────────────────┐ │ 4-Tier Graceful Degradation Architecture │ │ │ │ [Tier 1] Primary High-Performance Model (e.g., GPT-4o) │ │ │ │ │ ▼ (API Failure / Timeout / Circuit Breaker) │ │ [Tier 2] Lightweight Routing Model (e.g., GPT-4o-mini / On-Prem vLLM) │ │ │ │ │ ▼ (Continuous Outage) │ │ [Tier 3] Deterministic Regex & SQL Rule Engine │ │ │ │ │ ▼ (Unrecoverable Error) │ │ [Tier 4] Static Error Message & Async Admin Ticket Generation │ └─────────────────────────────────────────────────────────────────────────┘
L'optimisation d'un système d'agent personnalisé s'effectue par tranches de 90 jours.
| Période | Objectif d'application | Actions concrètes |
|---|---|---|
| Jours 1 à 30 | Visibilité et protection des données | Installation du moteur de masquage Presidio et intégration du SDK de journalisation Langfuse |
| Jours 31 à 60 | Blocage des boucles et réduction des coûts | Application de la mise en cache sémantique Redis et connexion du coupe-circuit en Python |
| Jours 61 à 90 | Automatisation de la validation et contrôle des accès | Gestion des prompts via Git, intégration CI/CD de DeepEval et application des métadonnées RBAC sur Qdrant |
Le premier mois est consacré à la mise en place d'une solution d'observabilité garantissant la journalisation de chaque requête et la prévention des fuites de PII. Le deuxième mois intègre la mise en cache des prompts récurrents et un routeur d'interruption de boucle pour éliminer les dépenses imprévues. Enfin, le dernier mois finalise la gestion de version des prompts et le système de tests automatisés, garantissant une exploitation stable sans crainte de dysfonctionnement de l'agent à chaque déploiement.