À mesure que les applications de grands modèles de langage (LLM) évoluent de simples bots de questions-réponses vers des systèmes complexes de RAG (Retrieval-Augmented Generation) multi-agents, l'observabilité est devenue un goulot d'étranglement critique. Dans une configuration RAG à agent unique, une requête s'écoule de manière linéaire : requête, récupération, génération. Cependant, dans les architectures multi-agents, les requêtes se divisent en sous-agents parallèles, chacun effectuant des tâches distinctes telles que l'analyse de documents, la recherche sémantique ou la vérification des faits. Cette complexité entraîne une fragmentation des traces, où une seule requête utilisateur est dispersée entre des dizaines de spans non liées, rendant le débogage presque impossible.
La journalisation traditionnelle est insuffisante car elle manque de contexte. C'est ici qu'OpenTelemetry (OTel) brille. En mettant en œuvre des schémas d'instrumentation personnalisés, vous pouvez recoller ces traces fragmentées pour offrir une vue unifiée de votre infrastructure IA. Dans cet article, nous explorons comment structurer ces schémas à l'aide de Python.
Le défi des workflows d'IA distribués
Imaginez un système de support client avec trois agents : un Agent de tri, un Agent de récupération des connaissances et un Générateur de réponses. Lorsqu'un utilisateur soumet une requête, l'Agent de tri détermine l'intention et délègue les tâches. Si l'intention est technique, l'Agent de connaissances recherche dans une base de données vectorielle ; s'il s'agit de facturation, il interroge une base de données SQL.
Sans corrélation appropriée, les spans générées par ces agents apparaissent comme des îlots isolés dans votre tableau de bord de surveillance des performances des applications (APM). Vous ne pouvez pas voir que la latence élevée du Générateur de réponses était causée par un délai d'attente dans la recherche vectorielle de l'Agent de connaissances. Pour résoudre ce problème, nous devons propager le contexte explicitement à travers les limites des agents.
Mise en œuvre d'une instrumentation personnalisée avec OpenTelemetry
La stratégie centrale consiste à créer un wrapper personnalisé qui capture le contexte d'exécution de chaque agent. Nous utilisons opentelemetry-api pour gérer manuellement les spans, garantissant que les appels imbriqués dans la logique interne d'un agent sont regroupés sous une seule span parente. Cela est particulièrement utile lorsque vos agents utilisent plusieurs bibliothèques (par exemple, LangChain, LlamaIndex ou des clients HTTP personnalisés) qui ne s'instrumentent pas automatiquement de manière optimale.
Voici un exemple pratique d'un décorateur personnalisé qui enveloppe la méthode d'exécution d'un agent, garantissant une structure de trace cohérente :
import functools
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode
tracer = trace.get_tracer(__name__)
def agent_span(agent_name: str):
"""
Un décorateur qui enveloppe la méthode d'un agent dans une span OpenTelemetry dédiée.
"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
with tracer.start_as_current_span(f"agent.{agent_name}") as span:
try:
# Définir des attributs pour un meilleur filtrage dans les outils APM
span.set_attribute("agent.name", agent_name)
span.set_attribute("args", str(args)[:100]) # Assainir les entrées
# Exécuter la logique de l'agent
result = func(*args, **kwargs)
# Marquer le succès
span.set_status(Status(StatusCode.OK))
return result
except Exception as e:
# Marquer l'échec et enregistrer l'erreur
span.set_status(Status(StatusCode.ERROR, str(e)))
span.record_exception(e)
raise
return wrapper
return decorator
# Exemple d'utilisation
class KnowledgeRetrievalAgent:
@agent_span("knowledge_retrieval")
def search_docs(self, query: str):
# Simuler une recherche coûteuse dans une base de données vectorielle
return {"doc_id": "123", "content": "La réponse est 42."}
Dans ce schéma, le décorateur @agent_span agit comme un conteneur pour toutes les opérations internes effectuées par cet agent spécifique. Lorsque vous enchaînez plusieurs agents, les spans enfants (issues de la logique interne) sont automatiquement imbriquées dans la span de l'agent, créant ainsi une vue hiérarchique plutôt qu'une liste plate d'événements.
Corrélation des spans à travers les limites des services
Pour les systèmes multi-agents déployés en tant que microservices, la propagation du contexte est vitale. Vous devez vous assurer que le TraceId et le SpanId sont transmis via des en-têtes (par exemple, les en-têtes HTTP dans les API REST ou les en-têtes de messages dans Kafka). Les utilitaires de propagation de contexte d'OpenTelemetry gèrent cela automatiquement si l'instrumentation de votre client ou serveur HTTP est activée. Cependant, pour les courtiers de messages personnalisés ou les bus d'événements internes, vous devrez peut-être injecter manuellement le support de contexte :
from opentelemetry.propagate import inject
def send_message_to_agent(queue, message, agent_name):
# Injecter le contexte de trace dans les métadonnées du message
headers = {}
inject(headers)
queue.publish({
"agent": agent_name,
"payload": message,
"trace_headers": headers
})
Conclusion
Résoudre la fragmentation des traces dans les systèmes RAG multi-agents ne consiste pas seulement à collecter plus de données ; il s'agit de structurer ces données de manière intelligente. En adoptant des schémas d'instrumentation personnalisés avec OpenTelemetry, les développeurs peuvent obtenir une visibilité granulaire sur les interactions des agents, identifier les goulots d'étranglement dans les workflows parallèles et, in fine, construire des applications IA plus fiables. À mesure que ces systèmes deviennent plus complexes, l'observabilité passera d'une fonctionnalité souhaitable à la colonne vertébrale de l'excellence opérationnelle.