Observabilité de niveau production pour les agents IA : une approche basée sur le code minimal et la configuration d’abord
deviennent plus complexes, l’exploitation forestière et la surveillance traditionnelles ne suffisent pas. Ce dont les équipes ont réellement besoin, c’est observabilité: la possibilité de suivre les décisions des agents, d’évaluer automatiquement la qualité des réponses et de détecter les dérives au fil du temps, sans écrire ni maintenir de grandes quantités de code d’évaluation et de télémétrie personnalisé.
Par conséquent, les équipes doivent adopter la bonne plate-forme d’observabilité tout en se concentrant sur la tâche principale consistant à construire et à améliorer l’orchestration des agents. Et intégrez leur application à la plate-forme d’observabilité avec une surcharge minimale de leur code fonctionnel. Dans cet article, je vais montrer comment vous pouvez configurer une plate-forme d’observabilité d’IA open source pour effectuer les opérations suivantes en utilisant une approche de code minimal :
- LLM en tant que juge: configurez des évaluateurs prédéfinis pour noter les réponses en termes d’exactitude, de pertinence, d’hallucination et plus encore. Affichez les scores sur plusieurs exécutions avec des journaux et des analyses détaillés.
- Tests à grande échelle: Configurez des ensembles de données pour stocker des cas de tests de régression afin de mesurer la précision par rapport aux réponses attendues de la vérité terrain. Détectez de manière proactive les LLM et la dérive des agents.
- Données de fusion: suivez les métriques (latence, utilisation des jetons, dérive du modèle), les événements (appels API, appels LLM, utilisation des outils), les journaux (interaction utilisateur, exécution de l’outil, prise de décision de l’agent) avec des traces détaillées, le tout sans télémétrie détaillée ni code d’instrumentation.
Nous utiliserons Langfuse pour l’observabilité. Il est open source et indépendant du framework et peut fonctionner avec les frameworks d’orchestration et les fournisseurs LLM populaires.
Application multi-agents
Pour cette démonstration, j’ai joint le code LangGraph d’une application Customer Service. L’application accepte les tickets de l’utilisateur, les classe en technique, facturation ou les deux à l’aide d’un agent de triage, puis les achemine vers l’agent du support technique, l’agent du support de facturation ou vers les deux. Ensuite, un agent finaliseur synthétise la réponse des deux agents dans un format cohérent et plus lisible. L’organigramme est le suivant :
Le code est joint ici
# --------------------------------------------------
# 0. Load .env
# --------------------------------------------------
from dotenv import load_dotenv
load_dotenv(override=True)
# --------------------------------------------------
# 1. Imports
# --------------------------------------------------
import os
from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import AzureChatOpenAI
from langfuse import Langfuse
from langfuse.langchain import CallbackHandler
# --------------------------------------------------
# 2. Langfuse Client (WORKING CONFIG)
# --------------------------------------------------
langfuse = Langfuse(
host="https://cloud.langfuse.com",
public_key=os.environ["LANGFUSE_PUBLIC_KEY"] ,
secret_key=os.environ["LANGFUSE_SECRET_KEY"]
)
langfuse_callback = CallbackHandler()
os.environ["LANGGRAPH_TRACING"] = "false"
# --------------------------------------------------
# 3. Azure OpenAI Setup
# --------------------------------------------------
llm = AzureChatOpenAI(
azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
api_version=os.environ.get("AZURE_OPENAI_API_VERSION", "2025-01-01-preview"),
temperature=0.2,
callbacks=[langfuse_callback], # 🔑 enables token usage
)
# --------------------------------------------------
# 4. Shared State
# --------------------------------------------------
class AgentState(TypedDict, total=False):
ticket: str
category: str
technical_response: str
billing_response: str
final_response: str
# --------------------------------------------------
# 5. Agent Definitions
# --------------------------------------------------
def triage_agent(state: dict) -> dict:
with langfuse.start_as_current_observation(
as_type="span",
name="triage_agent",
input={"ticket": state["ticket"]},
) as span:
span.update_trace(name="Customer Service Query - LangGraph Demo")
response = llm.invoke([
{
"role": "system",
"content": (
"Classify the query as one of: "
"Technical, Billing, Both. "
"Respond with only the label."
),
},
{"role": "user", "content": state["ticket"]},
])
raw = response.content.strip().lower()
if "both" in raw:
category = "Both"
elif "technical" in raw:
category = "Technical"
elif "billing" in raw:
category = "Billing"
else:
category = "Technical" # ✅ safe fallback
span.update(output={"raw": raw, "category": category})
return {"category": category}
def technical_support_agent(state: dict) -> dict:
with langfuse.start_as_current_observation(
as_type="span",
name="technical_support_agent",
input={
"ticket": state["ticket"],
"category": state.get("category"),
},
) as span:
response = llm.invoke([
{
"role": "system",
"content": (
"You are a technical support specialist. "
"Provide a clear, step-by-step solution."
),
},
{"role": "user", "content": state["ticket"]},
])
answer = response.content
span.update(output={"technical_response": answer})
return {"technical_response": answer}
def billing_support_agent(state: dict) -> dict:
with langfuse.start_as_current_observation(
as_type="span",
name="billing_support_agent",
input={
"ticket": state["ticket"],
"category": state.get("category"),
},
) as span:
response = llm.invoke([
{
"role": "system",
"content": (
"You are a billing support specialist. "
"Answer clearly about payments, invoices, or accounts."
),
},
{"role": "user", "content": state["ticket"]},
])
answer = response.content
span.update(output={"billing_response": answer})
return {"billing_response": answer}
def finalizer_agent(state: dict) -> dict:
with langfuse.start_as_current_observation(
as_type="span",
name="finalizer_agent",
input={
"ticket": state["ticket"],
"technical": state.get("technical_response"),
"billing": state.get("billing_response"),
},
) as span:
parts = [
f"Technical:\n{state['technical_response']}"
for k in ["technical_response"]
if state.get(k)
] + [
f"Billing:\n{state['billing_response']}"
for k in ["billing_response"]
if state.get(k)
]
if not parts:
final = "Error: No agent responses available."
else:
response = llm.invoke([
{
"role": "system",
"content": (
"Combine the following agent responses into ONE clear, professional, "
"customer-facing answer. Do not mention agents or internal labels. "
f"Answer the user's query: '{state['ticket']}'."
),
},
{"role": "user", "content": "\n\n".join(parts)},
])
final = response.content
span.update(output={"final_response": final})
return {"final_response": final}
# --------------------------------------------------
# 6. LangGraph Construction
# --------------------------------------------------
builder = StateGraph(AgentState)
builder.add_node("triage", triage_agent)
builder.add_node("technical", technical_support_agent)
builder.add_node("billing", billing_support_agent)
builder.add_node("finalizer", finalizer_agent)
builder.set_entry_point("triage")
# Conditional routing
builder.add_conditional_edges(
"triage",
lambda state: state["category"],
{
"Technical": "technical",
"Billing": "billing",
"Both": "technical",
"__default__": "technical", # ✅ never dead-end
},
)
# Sequential resolution
builder.add_conditional_edges(
"technical",
lambda state: state["category"],
{
"Both": "billing", # Proceed to billing if Both
"__default__": "finalizer",
},
)
builder.add_edge("billing", "finalizer")
builder.add_edge("finalizer", END)
graph = builder.compile()
# --------------------------------------------------
# 9. Main
# --------------------------------------------------
if __name__ == "__main__":
print("===============================================")
print(" Conditional Multi-Agent Support System (Ready)")
print("===============================================")
print("Enter 'exit' or 'quit' to stop the program.\n")
while True:
# Get user input for the ticket
ticket = input("Enter your support query (ticket): ")
# Check for exit command
if ticket.lower() in ["exit", "quit"]:
print("\nExiting the support system. Goodbye!")
break
if not ticket.strip():
print("Please enter a non-empty query.")
continue
try:
# --- Run the graph with the user's ticket ---
result = graph.invoke(
{"ticket": ticket},
config={"callbacks": [langfuse_callback]},
)
# --- Print Results ---
category = result.get('category', 'N/A')
print(f"\n✅ Triage Classification: **{category}**")
# Check which agents were executed based on the presence of a response
executed_agents = []
if result.get("technical_response"):
executed_agents.append("Technical")
if result.get("billing_response"):
executed_agents.append("Billing")
print(f"🛠️ Agents Executed: {', '.join(executed_agents) if executed_agents else 'None (Triage Failed)'}")
print("\n================ FINAL RESPONSE ================\n")
print(result["final_response"])
print("\n" + "="*60 + "\n")
except Exception as e:
# This is important for debugging: print the exception type and message
print(f"\nAn error occurred during processing ({type(e).__name__}): {e}")
print("\nPlease try another query.")
print("\n" + "="*60 + "\n")
Configuration de l’observabilité
Pour configurer Langfuse, accédez à https://cloud.langfuse.com/et créez un compte avec un niveau de facturation (niveau passe-temps avec des limites généreuses disponibles), puis créez un projet. Dans les paramètres du projet, vous pouvez générer les clés publiques et secrètes qui doivent être fournies au début du code. Vous devez également ajouter la connexion LLM, qui sera utilisée pour l’évaluation LLM-as-a-Judge.
Configuration LLM en tant que juge
Il s’agit du cœur de la configuration d’évaluation des performances des agents. Ici, vous pouvez configurer divers évaluateurs prédéfinis à partir de la bibliothèque d’évaluateurs qui noteront les réponses sur divers critères tels que la concision, l’exactitude, l’hallucination, la critique de réponse, etc. Ceux-ci devraient suffire pour la plupart des cas d’utilisation, sinon des évaluateurs personnalisés peuvent également être configurés. Voici une vue de la bibliothèque Evaluator :
Sélectionnez l’évaluateur, par exemple Pertinence, que vous souhaitez utiliser. Vous pouvez choisir de l’exécuter pour des traces nouvelles ou existantes ou pour des exécutions d’ensembles de données. De plus, examinez l’invite d’évaluation pour vous assurer qu’elle répond à votre objectif d’évaluation. Plus important encore, la requête, la génération et les autres variables doivent être correctement mappées à la source (généralement aux entrées et sorties de la trace de l’application). Pour notre cas, il s’agira respectivement des données du ticket saisies par l’utilisateur et de la réponse générée par l’agent finaliseur. De plus, pour les exécutions de jeux de données, vous pouvez comparer les réponses générées aux réponses Ground Truth stockées comme résultats attendus (expliqués dans les sections suivantes).
Voici la configuration pour le ‘Précision GT‘évaluation que j’ai configurée pour les nouvelles exécutions d’ensembles de données, ainsi que le mappage des variables. L’aperçu de l’invite d’évaluation est également représenté. La plupart des évaluateurs notent dans une fourchette de 0 à 1 :
Pour la démo du service client, j’ai configuré 3 évaluateurs – Pertinence, concision qui fonctionnent pour toutes les nouvelles traces, et Précision GTqui se déploie uniquement pour les exécutions de Dataset.
Configuration des ensembles de données
Créez un ensemble de données à utiliser comme référentiel de scénarios de test. Ici, vous pouvez stocker des cas de test avec la requête d’entrée et la réponse idéale attendue. Pour créer l’ensemble de données, vous avez 3 choix : créer un enregistrement à la fois, télécharger un CSV de requêtes et de réponses attendues ou, très facilement, ajouter des entrées et des sorties. directement depuis les traces applicatives dont les réponses sont jugées de bonne qualité par les experts humains.
Voici l’ensemble de données que j’ai créé pour la démo. Il s’agit d’un mélange de requêtes techniques, de facturation ou « les deux », et j’ai créé tous les enregistrements à partir des traces d’application :
C’est ça! La configuration est terminée et nous sommes prêts à exécuter l’observabilité.
Résultats d’observabilité
La page d’accueil de Langfuse est un tableau de bord contenant plusieurs graphiques utiles. Il montre le nombre de traces d’exécution, les scores et les moyennes en un coup d’œil, les traces par temps, l’utilisation et le coût du modèle, etc.
Données de fusion
Les données d’observabilité les plus utiles sont disponibles dans l’option « Traçage », qui affiche des vues résumées et détaillées de toutes les exécutions. Voici une vue du tableau de bord décrivant l’heure, le nom, l’entrée, la sortie et les mesures cruciales de latence et d’utilisation des jetons. Notez que pour chaque exécution d’agent de notre application, 2 traces d’évaluation sont générées pour le Concision et Pertinence évaluateurs que nous avons mis en place.
Regardons les détails de l’une des exécutions de l’application Customer Service. Sur le panneau de gauche, le flux des agents est représenté à la fois sous forme d’arborescence et d’organigramme. Il montre les nœuds LangGraph (agents) et les appels LLM ainsi que l’utilisation du jeton. Si nos agents avaient eu des appels d’outils ou des étapes humaines, ils auraient également été représentés ici. Notez que les scores d’évaluation pour Concision et Pertinence sont également représentés en haut, qui valent respectivement 0,40 et 1 pour cette exécution. En cliquant dessus, vous verrez le motif du score et un lien pour accéder à la trace de l’évaluateur.
Sur la droite, pour chaque appel d’agent, LLM et outil, nous pouvons voir l’entrée et la sortie générée. Par exemple, nous voyons ici que la requête a été classée dans la catégorie « Les deux ». Par conséquent, dans le graphique de gauche, elle montre que les agents d’assistance technique et de facturation ont été appelés, ce qui confirme que notre flux fonctionne comme prévu.
En haut du panneau de droite, il y a le ‘Ajouter aux ensembles de données bouton. À n’importe quelle étape de l’arborescence, ce bouton, lorsque vous cliquez dessus, ouvrira un panneau comme celui illustré ci-dessous, dans lequel vous pourrez ajouter l’entrée et la sortie de cette étape directement à un ensemble de données de test créé dans la section précédente. Il s’agit d’une fonctionnalité utile permettant aux experts humains d’ajouter des requêtes utilisateur fréquentes et de bonnes réponses à l’ensemble de données pendant les opérations normales de l’agent, créant ainsi un référentiel de tests de régression avec un minimum d’effort. À l’avenir, lorsqu’il y aura une mise à niveau ou une version majeure de l’application, l’ensemble de données de régression pourra être exécuté et les résultats générés pourront être évalués par rapport aux résultats attendus (vérité terrain) enregistrés ici à l’aide de l’option ‘Précision GT‘ évaluateur que nous avons créé lors de la configuration LLM-as-a-juge. Cela permet de détecter précocement la dérive LLM (ou dérive d’agent) et de prendre des mesures correctives.
Voici une des traces d’évaluation (Concision) pour cette trace d’application. L’évaluateur justifie la note de 0,4 qu’il a attribuée à cette réponse.
Partitions
L’option Scores dans Langfuse affiche une liste de toutes les évaluations des différents évaluateurs actifs ainsi que leurs scores. Le tableau de bord Analytics est plus pertinent, où deux scores peuvent être sélectionnés et des mesures telles que la moyenne et l’écart type ainsi que les lignes de tendance peuvent être visualisées.
Tests de régression
Avec les ensembles de données, nous sommes prêts à exécuter des tests de régression en utilisant le référentiel de cas de test des requêtes et des résultats attendus. Nous avons stocké 4 requêtes dans notre ensemble de données de régression, avec un mélange de requêtes techniques, de facturation et « les deux ».
Pour cela, nous pouvons exécuter le code ci-joint qui obtient l’ensemble de données pertinent et exécute l’expérience. Tous les tests sont enregistrés avec les scores moyens. Nous pouvons visualiser le résultat d’un test sélectionné avec Concision, précision GT et pertinence scores pour chaque cas de test dans un seul tableau de bord. Et si nécessaire, la trace détaillée est accessible pour voir le raisonnement du score.
Vous pouvez consulter le code ici.
from langfuse import get_client
from langfuse.openai import OpenAI
from langchain_openai import AzureChatOpenAI
from langfuse import Langfuse
import os
# Initialize client
from dotenv import load_dotenv
load_dotenv(override=True)
langfuse = Langfuse(
host="https://cloud.langfuse.com",
public_key=os.environ["LANGFUSE_PUBLIC_KEY"] ,
secret_key=os.environ["LANGFUSE_SECRET_KEY"]
)
llm = AzureChatOpenAI(
azure_deployment=os.environ.get("AZURE_OPENAI_DEPLOYMENT_NAME"),
api_version=os.environ.get("AZURE_OPENAI_API_VERSION", "2025-01-01-preview"),
temperature=0.2,
)
# Define your task function
def my_task(*, item, **kwargs):
question = item.input['ticket']
response = llm.invoke([{"role": "user", "content": question}])
raw = response.content.strip().lower()
return raw
# Get dataset from Langfuse
dataset = langfuse.get_dataset("Regression")
# Run experiment directly on the dataset
result = dataset.run_experiment(
name="Production Model Test",
description="Monthly evaluation of our production model",
task=my_task # see above for the task definition
)
# Use format method to display results
print(result.format())
Points clés à retenir
- L’observabilité de l’IA n’a pas besoin d’être lourde en code.
La plupart des fonctionnalités d’évaluation, de traçage et de test de régression pour les agents LLM peuvent être activées via la configuration plutôt que par le code personnalisé, ce qui réduit considérablement les efforts de développement et de maintenance. - Des workflows d’évaluation riches peuvent être définis de manière déclarative.
Des fonctionnalités telles que la notation LLM-as-a-Judge (pertinence, concision, hallucination, précision de la vérité terrain), le mappage des variables et les invites d’évaluation sont configurés directement dans la plateforme d’observabilité, sans écrire de logique d’évaluation sur mesure. - Les ensembles de données et les tests de régression sont des fonctionnalités axées sur la configuration.
Les référentiels de cas de test, les exécutions d’ensembles de données et les comparaisons avec la vérité terrain peuvent être configurés et réutilisés via l’interface utilisateur ou une configuration simple, permettant aux équipes d’exécuter des tests de régression sur plusieurs versions d’agent avec un minimum de code supplémentaire. - L’observabilité complète de MELT est prête à l’emploi.
Les métriques (latence, utilisation des jetons, coût), les événements (appels LLM et outils), les journaux et les traces sont automatiquement capturés et corrélés, évitant ainsi le besoin d’une instrumentation manuelle dans les flux de travail des agents. - Instrumentation minimale, visibilité maximale.
Grâce à l’intégration légère du SDK, les équipes bénéficient d’une visibilité approfondie sur les chemins d’exécution multi-agents, les résultats d’évaluation et les tendances de performances, permettant ainsi aux développeurs de se concentrer sur la logique des agents plutôt que sur la plomberie d’observabilité.
Conclusion
À mesure que les agents LLM deviennent plus complexes, l’observabilité n’est plus facultative. Sans cela, les systèmes multi-agents se transforment rapidement en boîtes noires difficiles à évaluer, déboguer et améliorer.
Une plateforme d’observabilité de l’IA décharge ce fardeau des développeurs et du code d’application. Utiliser un approche basée sur le code minimal et la configuration d’abordles équipes peuvent activer l’évaluation LLM en tant que juge, les tests de régression et l’observabilité MELT complète sans créer ni maintenir de pipelines personnalisés. Cela réduit non seulement les efforts d’ingénierie, mais accélère également le passage du prototype à la production.
En adoptant une plateforme open source indépendante du framework comme Langfuse, les équipes bénéficient d’un source unique de vérité pour les performances des agents, ce qui rend les systèmes d’IA plus faciles à faire confiance, à évoluer et à exploiter à grande échelle.
Vous voulez en savoir plus ? L’application agentique de service client présentée ici suit un modèle d’architecture gestionnaire-travailleur, qui ne fait pas travailler dans CrewAI. Découvrez comment observabilité m’a aidé à résoudre ce problème bien connu avec le processus hiérarchique gestionnaire-employé de CrewAI, en traçant les réponses des agents à chaque étape et en les affinant pour que l’orchestration fonctionne comme elle le devrait. Analyse complète ici : Pourquoi l’architecture Manager-Worker de CrewAI échoue – et comment y remédier
Connectez-vous avec moi et partagez vos commentaires sur www.linkedin.com/in/partha-sarkar-lets-talk-AI
Toutes les images et données utilisées dans cet article sont générées de manière synthétique. Chiffres et code créés par moi
