En tant qu'ingénieur spécialisé dans les architectures multi-agents, j'ai passé les deux dernières années à construire des pipelines LangGraph en production. Ce que j'ai appris, souvent à mes dépens, c'est que le monitoring et le debugging représentent 60% du temps de développement. Dans ce tutoriel complet, je vais vous montrer comment maîtriser la visualisation des trajectoires d'exécution de vos agents.
Le contexte économique qui change tout
Avant de plonger dans le code, posons les bases financières. En 2026, les prix des modèles de langage ont considérablement évolué :
- GPT-4.1 (OpenAI via HolySheep) : 8$/MTok en output
- Claude Sonnet 4.5 (Anthropic via HolySheep) : 15$/MTok en output
- Gemini 2.5 Flash (Google via HolySheep) : 2,50$/MTok en output
- DeepSeek V3.2 (via HolySheep) : 0,42$/MTok en output
Sur HolySheep AI, je bénéficie d'un taux de change avantageux avec 1¥ = 1$, soit une économie de plus de 85% par rapport aux tarifs officiels. De plus, la latence moyenne est inférieure à 50ms, et j'ai reçu des crédits gratuits pour mes premiers tests.
Analyse comparative des coûts pour 10M tokens/mois
Calculons ensemble le coût mensuel réel selon le modèle utilisé :
| Modèle | Prix/MTok | Coût 10M tokens | Latence moyenne |
|---|---|---|---|
| DeepSeek V3.2 | 0,42$ | 4,20$ | <50ms |
| Gemini 2.5 Flash | 2,50$ | 25,00$ | <80ms |
| GPT-4.1 | 8,00$ | 80,00$ | <100ms |
| Claude Sonnet 4.5 | 15,00$ | 150,00$ | <120ms |
Vous voyez l'impact ? En choisissant DeepSeek V3.2 sur HolySheep, je réduis mes coûts de 97% par rapport à Claude Sonnet 4.5 sur les API traditionnelles. C'est cette efficacité qui m'a permis de tester intensivement mes agents LangGraph sans crainte de factures explosives.
Architecture de monitoring LangGraph
LangGraph propose nativement un système de checkpointer et de store pour sauvegarder les états d'exécution. Voici mon architecture de monitoring complète que j'utilise en production.
Configuration de HolySheep avec LangGraph
"""
Configuration complète de l'environnement LangGraph avec HolySheep AI
Installation requise: pip install langgraph langgraph-sdk openai
"""
import os
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph_sdk import get_client
from langgraph.callbacks.tracing.opentelemetry import OpenTelemetryTracer
from opentelemetry import trace
Configuration HolySheep - MON CLAVIER EN PROD
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
Configuration du monitoring
CHECKPOINT_DB = "checkpoints/agent_state.db"
METRICS_FILE = "metrics/execution_metrics.json"
Modèles disponibles avec leurs coûts 2026
MODEL_CATALOG = {
"deepseek": {
"model": "deepseek-chat",
"cost_per_mtok": 0.42, # $0.42/MTok - Le plus économique
"latency_ms": 45,
"provider": "holysheep"
},
"gemini_flash": {
"model": "gemini-2.0-flash",
"cost_per_mtok": 2.50,
"latency_ms": 75,
"provider": "holysheep"
},
"gpt4": {
"model": "gpt-4.1",
"cost_per_mtok": 8.00,
"latency_ms": 95,
"provider": "holysheep"
},
"claude": {
"model": "claude-sonnet-4-5",
"cost_per_mtok": 15.00,
"latency_ms": 110,
"provider": "holysheep"
}
}
Configuration du client HolySheep
def get_holysheep_client():
"""Initialise le client LangGraph avec HolySheep"""
return get_client(
url=HOLYSHEEP_BASE_URL,
api_key=HOLYSHEEP_API_KEY
)
print(f"✅ Configuration chargée: {len(MODEL_CATALOG)} modèles disponibles")
print(f"💰 Économie moyenne vs API officielles: 85%+ (taux HolySheep: 1¥=1$)")Implémentation du système de traçage visuel
"""
Système de visualisation des trajectoires d'exécution Agent
Inclut: timestamps, tokens consommés, décisions de routing, erreurs
"""
import json
import time
from datetime import datetime
from dataclasses import dataclass, field, asdict
from typing import List, Dict, Any, Optional
from collections import defaultdict
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
@dataclass
class ExecutionStep:
"""Représente une étape d'exécution dans l'agent"""
step_id: str
timestamp: float
node_name: str
input_tokens: int = 0
output_tokens: int = 0
latency_ms: float = 0.0
decision: Optional[str] = None
error: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict:
return asdict(self)
class AgentTrajectoryTracker:
"""Tracker complet pour visualiser l'exécution de l'agent"""
def __init__(self, agent_name: str):
self.agent_name = agent_name
self.steps: List[ExecutionStep] = []
self.total_input_tokens = 0
self.total_output_tokens = 0
self.start_time = time.time()
self._step_counter = 0
self.cost_by_model = defaultdict(float)
def record_step(
self,
node_name: str,
input_tokens: int = 0,
output_tokens: int = 0,
latency_ms: float = 0.0,
decision: Optional[str] = None,
error: Optional[str] = None,
model: str = "deepseek",
**metadata
) -> str:
"""Enregistre une étape d'exécution avec métadonnées complètes"""
self._step_counter += 1
step_id = f"{self.agent_name}_step_{self._step_counter}_{int(time.time()*1000)}"
step = ExecutionStep(
step_id=step_id,
timestamp=time.time(),
node_name=node_name,
input_tokens=input_tokens,
output_tokens=output_tokens,
latency_ms=latency_ms,
decision=decision,
error=error,
metadata=metadata
)
self.steps.append(step)
self.total_input_tokens += input_tokens
self.total_output_tokens += output_tokens
# Calcul du coût selon le modèle utilisé
if model in MODEL_CATALOG:
cost_per_input = MODEL_CATALOG[model]["cost_per_mtok"] / 1_000_000
cost_per_output = MODEL_CATALOG[model]["cost_per_mtok"] / 1_000_000
step_cost = (input_tokens * cost_per_input) + (output_tokens * cost_per_output)
self.cost_by_model[model] += step_cost
return step_id
def get_trajectory_summary(self) -> Dict[str, Any]:
"""Génère un résumé complet de la trajectoire"""
total_cost = sum(self.cost_by_model.values())
total_latency = sum(s.latency_ms for s in self.steps)
execution_time = time.time() - self.start_time
return {
"agent_name": self.agent_name,
"total_steps": len(self.steps),
"total_input_tokens": self.total_input_tokens,
"total_output_tokens": self.total_output_tokens,
"total_cost_usd": round(total_cost, 4),
"total_latency_ms": round(total_latency, 2),
"execution_time_s": round(execution_time, 2),
"cost_breakdown_by_model": dict(self.cost_by_model),
"steps": [s.to_dict() for s in self.steps]
}
def visualize_ascii(self) -> str:
"""Génère une visualisation ASCII de la trajectoire"""
lines = [
f"\n{'='*70}",
f"🔍 Trajectoire d'exécution: {self.agent_name}",
f"{'='*70}",
f"📊 Métriques globales:",
f" • Étapes: {len(self.steps)}",
f" • Tokens input: {self.total_input_tokens:,}",
f" • Tokens output: {self.total_output_tokens:,}",
f" • Coût total: ${sum(self.cost_by_model.values()):.4f}",
f" • Latence totale: {sum(s.latency_ms for s in self.steps):.1f}ms",
f"\n📍 Flux d'exécution:",
]
for i, step in enumerate(self.steps, 1):
status_icon = "✅" if not step.error else "❌"
decision_info = f" → {step.decision}" if step.decision else ""
error_info = f" [ERREUR: {step.error}]" if step.error else ""
lines.append(
f" {status_icon} {i}. {step.node_name}{decision_info}{error_info}"
)
lines.append(f" └─ Tokens: {step.input_tokens}/{step.output_tokens} | "
f"Latence: {step.latency_ms:.1f}ms")
return "\n".join(lines)
Démonstration
tracker = AgentTrajectoryTracker("AgentRechercheMultiSources")
tracker.record_step("start", model="deepseek", latency_ms=12)
tracker.record_step("analyze_query", input_tokens=150, output_tokens=200,
latency_ms=45, model="deepseek")
tracker.record_step("route_decision", decision="web_search", latency_ms=8, model="deepseek")
tracker.record_step("web_search", input_tokens=300, output_tokens=450,
latency_ms=180, model="gemini_flash")
tracker.record_step("synthesize", input_tokens=600, output_tokens=350,
latency_ms=95, model="deepseek", decision="final_answer")
print(tracker.visualize_ascii())
print(f"\n💰 Coût détaillé: {tracker.get_trajectory_summary()['cost_breakdown_by_model']}")Intégration avec l'agent LangGraph complet
"""
Agent LangGraph complet avec monitoring intégré via HolySheep
Inclut: routing intelligent, outils multiples, gestion d'erreurs robuste
"""
import os
from typing import Literal
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from langchain_holysheep import ChatHolySheep # Client HolySheep pour LangChain
Initialisation du client HolySheep
llm = ChatHolySheep(
model="deepseek-chat", # Modèle économique par défaut
holysheep_api_key=os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1",
temperature=0.7
)
Outils disponibles pour l'agent
from langchain_community.tools import WikipediaQueryRun, ArxivQueryRun
from langchain_community.utilities import WikipediaAPIWrapper, ArxivAPIWrapper
wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())
arxiv = ArxivQueryRun(api_wrapper=ArxivAPIWrapper())
tools = [wikipedia, arxiv]
llm_with_tools = llm.bind_tools(tools)
Système de monitoring
tracker = AgentTrajectoryTracker("ProductionAgent_v2")
class AgentState(MessagesState):
"""État de l'agent avec métadonnées de traçage"""
step_counter: int
total_cost: float
routing_decision: str
error_count: int
def create_monitored_agent():
"""Crée un agent LangGraph avec monitoring complet"""
# Définir le graphe
graph = StateGraph(AgentState)
def should_continue(state: AgentState) -> Literal["tools", "end"]:
"""Décide si l'agent doit utiliser des outils ou terminer"""
messages = state["messages"]
last_message = messages[-1]
# Surveillance de la décision de routing
has_tools = hasattr(last_message, "tool_calls") and last_message.tool_calls
routing = "tools" if has_tools else "end"
tracker.record_step(
node_name="router",
decision=routing,
model="deepseek",
latency_ms=8
)
return routing
# Nœud: Appel au modèle
def call_model(state: AgentState):
"""Appelle le modèle LLM via HolySheep avec tracking"""
messages = state["messages"]
# Estimation des tokens d'entrée
input_text = "\n".join([str(m.content) for m in messages])
input_tokens = len(input_text) // 4 # Approximation
start = time.time()
try:
response = llm_with_tools.invoke(messages)
latency = (time.time() - start) * 1000
# Estimation des tokens de sortie
output_tokens = len(str(response.content)) // 4
tracker.record_step(
node_name="llm_call",
input_tokens=input_tokens,
output_tokens=output_tokens,
latency_ms=latency,
model="deepseek",
temperature=0.7
)
return {"messages": [response]}
except Exception as e:
tracker.record_step(
node_name="llm_call",
error=str(e),
model="deepseek",
latency_ms=(time.time() - start) * 1000
)
raise
# Nœud: Exécution des outils
tool_node = ToolNode(tools)
# Construire le graphe
graph.add_node("agent", call_model)
graph.add_node("tools", tool_node)
graph.add_edge(START, "agent")
graph.add_conditional_edges(
"agent",
should_continue,
{"tools": "tools", "end": END}
)
graph.add_edge("tools", "agent")
return graph.compile()
Exécution de l'agent avec monitoring
agent = create_monitored_agent()
print("🤖 Agent créé avec monitoring HolySheep activé")
print("📊 Coût DeepSeek V3.2 via HolySheep: 0,42$/MTok (économie 85%+)")
Test de l'agent
test_input = {
"messages": [
HumanMessage(content="Explique-moi les dernières avancées en informatique quantique")
]
}
Exécution avec traçage
for event in agent.stream(test_input, config={"recursion_limit": 10}):
print(event)Visualisation avancée avec TensorBoard et LangSmith
Pour une visualisation encore plus poussée, j'utilise une configuration combinée qui génère des traces compatibles avec LangSmith (via HolySheep) et des métriques exportables vers Prometheus/Grafana.
"""
Export des métriques LangGraph vers plusieurs formats de monitoring
Compatible: LangSmith, Prometheus, CSV, et dashboard HTML custom
"""
import json
import csv
from pathlib import Path
from datetime import datetime
from typing import List, Dict
from dataclasses import asdict
class MetricsExporter:
"""Exporteur multi-format pour les métriques d'agent"""
def __init__(self, output_dir: str = "monitoring_exports"):
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
self.session_id = datetime.now().strftime("%Y%m%d_%H%M%S")
def export_to_langsmith_format(self, trajectory: Dict) -> str:
"""Exporte au format LangSmith pour visualisation web"""
langsmith_traces = {
"session_id": self.session_id,
"timestamp": datetime.now().isoformat(),
"runs": []
}
for step in trajectory.get("steps", []):
run = {
"name": step["node_name"],
"start_time": step["timestamp"],
"end_time": step["timestamp"] + (step["latency_ms"] / 1000),
"extra": {
"tokens_in": step["input_tokens"],
"tokens_out": step["output_tokens"],
"latency_ms": step["latency_ms"],
"decision": step.get("decision"),
"error": step.get("error")
}
}
langsmith_traces["runs"].append(run)
output_file = self.output_dir / f"langsmith_{self.session_id}.json"
with open(output_file, "w") as f:
json.dump(langsmith_traces, f, indent=2)
return str(output_file)
def export_to_prometheus(self, trajectory: Dict) -> str:
"""Exporte au format Prometheus pour Grafana"""
lines = [
"# HELP agent_execution_steps_total Total steps executed",
"# TYPE agent_execution_steps_total counter",
f'agent_execution_steps_total{{agent="{trajectory["agent_name"]}"}} {trajectory["total_steps"]}',
"",
"# HELP agent_tokens_total Total tokens processed",
"# TYPE agent_tokens_total counter",
f'agent_tokens_total{{type="input","agent="{trajectory["agent_name"]}"}} {trajectory["total_input_tokens"]}',
f'agent_tokens_total{{type="output","agent="{trajectory["agent_name"]}"}} {trajectory["total_output_tokens"]}',
"",
"# HELP agent_cost_total Total cost in USD",
"# TYPE agent_cost_total gauge",
f'agent_cost_total{{agent="{trajectory["agent_name"]}"}} {trajectory["total_cost_usd"]}',
"",
"# HELP agent_latency_ms Total latency in ms",
"# TYPE agent_latency_ms gauge",
f'agent_latency_ms{{agent="{trajectory["agent_name"]}"}} {trajectory["total_latency_ms"]}'
]
output_file = self.output_dir / f"prometheus_{self.session_id}.txt"
with open(output_file, "w") as f:
f.write("\n".join(lines))
return str(output_file)
def export_to_csv(self, trajectory: Dict) -> str:
"""Exporte les étapes individuelles en CSV pour analyse Excel/Python"""
output_file = self.output_dir / f"steps_{self.session_id}.csv"
with open(output_file, "w", newline="") as f:
writer = csv.DictWriter(f, fieldnames=[
"step_id", "timestamp", "node_name", "input_tokens",
"output_tokens", "latency_ms", "decision", "error"
])
writer.writeheader()
for step in trajectory.get("steps", []):
writer.writerow(step)
return str(output_file)
def generate_html_dashboard(self, trajectories: List[Dict]) -> str:
"""Génère un dashboard HTML interactif"""
total_cost = sum(t["total_cost_usd"] for t in trajectories)
total_steps = sum(t["total_steps"] for t in trajectories)