Documentation Index
Fetch the complete documentation index at: https://wb-21fd5541-john-wbdocs-2044-rename-serverless-products.mintlify.app/llms.txt
Use this file to discover all available pages before exploring further.
Avec les Threads de W&B Weave, vous pouvez suivre et analyser les conversations à plusieurs tours dans vos applications LLM. Les threads regroupent des Appels liés sous un thread_id commun, afin que vous puissiez visualiser des sessions complètes et suivre des métriques au niveau de la conversation d’un tour à l’autre. Vous pouvez créer des threads par programmation et les visualiser dans la Weave UI.
Pour commencer avec les Threads, procédez comme suit :
- Familiarisez-vous avec les bases des Threads.
- Essayez les exemples de code, qui montrent des modèles d’utilisation courants et des cas d’utilisation concrets.
Les threads sont utiles lorsque vous souhaitez organiser et analyser :
- Des conversations à plusieurs tours
- Des flux de travail basés sur des sessions
- Toute séquence d’opérations liées.
Les threads vous permettent de regrouper les Appels par contexte, ce qui facilite la compréhension de la façon dont votre système répond au fil de plusieurs étapes. Par exemple, vous pouvez suivre la session d’un utilisateur, la chaîne de décisions d’un agent ou une requête complexe qui s’étend sur les couches d’infrastructure et de logique métier.
En structurant votre application avec des threads et des tours, vous obtenez des métriques plus claires et une meilleure visibilité dans le Weave UI. Au lieu de voir chaque op de bas niveau, vous pouvez vous concentrer sur les étapes de haut niveau qui comptent.
Un Thread est un regroupement logique d’Appels associés qui partagent un même contexte conversationnel. Un Thread :
- Dispose d’un
thread_id unique
- Contient un ou plusieurs tours
- Maintient le contexte d’un Appel à l’autre
- Représente des sessions utilisateur complètes ou des flux d’interaction
Un tour est une opération de haut niveau au sein d’un Thread, affichée dans l’UI sous forme de lignes distinctes dans une vue de thread. Chaque tour :
- représente une étape logique dans une conversation ou un flux de travail
- est l’enfant direct d’un contexte de thread et peut inclure des appels imbriqués de niveau inférieur (non affichés dans les statistiques au niveau du thread).
Un Appel désigne toute exécution d’une fonction décorée avec @weave.op dans votre application.
- Les Appels de tour sont des opérations de premier niveau qui lancent de nouveaux tours
- Les Appels imbriqués sont des opérations de niveau inférieur au sein d’un tour
Une Trace capture la pile complète des Appels pour une seule opération. Les threads regroupent les traces qui font partie d’une même conversation logique ou session. En d’autres termes, un thread est composé de plusieurs tours de conversation, chacun représentant une partie de la conversation. Pour en savoir plus sur les Traces, voir l’Aperçu du Tracing.
Dans la barre latérale de Weave, sélectionnez Threads pour accéder à la vue en liste des threads.
- Répertorie les threads récents de votre projet
- Les colonnes incluent le nombre de tours, l’heure de début et la date de la dernière mise à jour
- Cliquez sur une ligne pour ouvrir son volet de détails
Volet de détails des threads
- Cliquez sur n’importe quelle ligne pour ouvrir le volet de détails correspondant.
- Affiche tous les tours au sein d’un thread.
- Les tours sont listés dans l’ordre de leur démarrage (selon leur heure de début, et non leur durée ou leur heure de fin).
- Inclut des métadonnées au niveau de l’appel (latence, entrées, sorties).
- Peut aussi afficher le contenu des messages ou des données structurées s’ils ont été enregistrés.
- Pour voir l’exécution complète d’un tour, vous pouvez l’ouvrir depuis le volet de détails du thread. Cela vous permet d’explorer toutes les opérations imbriquées survenues pendant ce tour précis.
- Si un tour inclut des messages extraits d’appels LLM, ils apparaissent dans le volet de chat de droite. Ces messages proviennent généralement d’appels effectués par des intégrations prises en charge (par exemple,
openai.ChatCompletion.create) et doivent répondre à des critères spécifiques pour s’afficher. Pour plus d’informations, voir Comportement de la vue de chat.
Comportement de la vue Chat
Qu’est-ce qui est considéré comme un message ?
openai.ChatCompletion.createanthropic.Anthropic.completion
Que se passe-t-il si aucun message n’est présent ?
Interactions entre les tours de conversation et le chat
- Cliquer sur un tour de conversation fait défiler le volet de chat jusqu’à l’emplacement du message correspondant (comportement d’épinglage).
- Faire défiler le volet de chat met en surbrillance le tour de conversation correspondant dans la liste de gauche.
Accéder à la vue de trace et en revenir
- Pour suivre une session ou une conversation, utilisez le gestionnaire de contexte
weave.thread().
- Décorez les opérations logiques avec
@weave.op pour les suivre en tant que tours ou Appels imbriqués.
- Si vous passez un
thread_id, Weave l’utilise pour regrouper toutes les opérations de ce bloc dans le même thread. Si vous omettez le thread_id, Weave en génère automatiquement un unique.
La valeur de retour de weave.thread() est un objet ThreadContext doté d’une propriété thread_id, que vous pouvez consigner, réutiliser ou transmettre à d’autres systèmes.
Les contextes weave.thread() imbriqués démarrent toujours un nouveau thread, sauf si vous réutilisez le même thread_id. La fin d’un contexte enfant n’interrompt ni n’écrase le contexte parent. Cela permet d’obtenir des structures de threads ramifiées ou une orchestration en couches des threads, selon la logique de votre application.
Création de thread de base
weave.thread() pour regrouper une ou plusieurs opérations sous un thread_id commun. C’est le moyen le plus simple de commencer à utiliser les threads dans votre application.
import weave
@weave.op
def say_hello(name: str) -> str:
return f"Hello, {name}!"
# Démarrer un nouveau contexte de thread
with weave.thread() as thread_ctx:
print(f"Thread ID: {thread_ctx.thread_id}")
say_hello("Bill Nye the Science Guy")
Implémentation manuelle d’une boucle d’agent
@weave.op et du contexte weave.thread(). Chaque appel à process_user_message crée un nouveau tour dans le thread. Vous pouvez utiliser ce modèle lorsque vous créez votre propre boucle d’agent et que vous souhaitez contrôler entièrement la gestion du contexte et de l’imbrication.
Utilisez l’ID de thread généré automatiquement pour les interactions de courte durée, ou transmettez un ID de session personnalisé (comme user_session_123) pour conserver le contexte du thread entre les sessions.
import weave
class ConversationAgent:
@weave.op
def process_user_message(self, message: str) -> str:
"""
OPÉRATION AU NIVEAU DU TOUR : Représente un tour de conversation.
Seule cette fonction sera comptabilisée dans les statistiques du thread.
"""
# Stocker le message de l'utilisateur
# Générer la réponse IA via des appels imbriqués
response = self._generate_response(message)
# Stocker la réponse de l'assistant
return response
@weave.op
def _generate_response(self, message: str) -> str:
"""APPEL IMBRIQUÉ : Détails d'implémentation, non comptabilisés dans les stats du thread."""
context = self._retrieve_context(message) # Autre appel imbriqué
intent = self._classify_intent(message) # Autre appel imbriqué
response = self._call_llm(message, context) # Appel LLM (imbriqué)
return self._format_response(response) # Dernier appel imbriqué
@weave.op
def _retrieve_context(self, message: str) -> str:
# Recherche dans la base vectorielle, requête dans la base de connaissances, etc.
return "retrieved_context"
@weave.op
def _classify_intent(self, message: str) -> str:
# Logique de classification des intentions
return "general_inquiry"
@weave.op
def _call_llm(self, message: str, context: str) -> str:
# Appel API OpenAI/Anthropic/etc
return "llm_response"
@weave.op
def _format_response(self, response: str) -> str:
# Logique de mise en forme de la réponse
return f"Formatted: {response}"
# Utilisation : contexte du thread établi automatiquement
agent = ConversationAgent()
# Établir le contexte du thread - chaque appel à process_user_message devient un tour
with weave.thread() as thread_ctx: # Génère automatiquement un thread_id
print(f"Thread ID: {thread_ctx.thread_id}")
# Chaque appel à process_user_message crée 1 tour + plusieurs appels imbriqués
agent.process_user_message("Hello, help with setup") # Tour 1
agent.process_user_message("What languages do you recommend?") # Tour 2
agent.process_user_message("Explain Python vs JavaScript") # Tour 3
# Résultat : thread avec 3 tours, ~15-20 appels au total (imbriqués inclus)
# Alternative : utiliser un thread_id explicite pour le suivi de session
session_id = "user_session_123"
with weave.thread(session_id) as thread_ctx:
print(f"Session Thread ID: {thread_ctx.thread_id}") # "user_session_123"
agent.process_user_message("Continue our previous conversation") # Tour 1 dans cette session
agent.process_user_message("Can you summarize what we discussed?") # Tour 2 dans cette session
Agent manuel avec profondeur d’appel asymétrique
thread_id, mais la frontière du tour apparaît à différents niveaux de la pile selon la logique du fournisseur. C’est utile lorsque les appels doivent être tracés différemment selon les backends, tout en étant regroupés dans le même thread.
import weave
import random
import asyncio
class OpenAIProvider:
"""Branche OpenAI : chaîne d'appels de 2 niveaux de profondeur jusqu'à la frontière de tour"""
@weave.op
def route_to_openai(self, user_input: str, thread_id: str) -> str:
"""Niveau 1 : Acheminer et préparer la requête OpenAI"""
# Validation des entrées, logique de routage, prétraitement de base
print(f" N1: Routage vers OpenAI pour: {user_input}")
# Ceci est la frontière de tour — entourer avec le contexte thread
with weave.thread(thread_id):
# Appeler directement le niveau 2 — cela crée la profondeur de la chaîne d'appels
return self.execute_openai_call(user_input)
@weave.op
def execute_openai_call(self, user_input: str) -> str:
"""Niveau 2 : FRONTIÈRE DE TOUR — Exécuter l'appel API OpenAI"""
print(f" N2: Exécution de l'appel API OpenAI")
response = f"Réponse de OpenAI GPT-4 : {user_input}"
return response
class AnthropicProvider:
"""Branche Anthropic : chaîne d'appels de 3 niveaux de profondeur jusqu'à la frontière de tour"""
@weave.op
def route_to_anthropic(self, user_input: str, thread_id: str) -> str:
"""Niveau 1 : Acheminer et préparer la requête Anthropic"""
# Validation des entrées, logique de routage, sélection du fournisseur
print(f" N1: Routage vers Anthropic pour: {user_input}")
# Call Level 2 - this creates call chain depth
return self.authenticate_anthropic(user_input, thread_id)
@weave.op
def authenticate_anthropic(self, user_input: str, thread_id: str) -> str:
"""Niveau 2 : Gérer l'authentification et la configuration Anthropic"""
print(f" N2: Authentification avec Anthropic")
# Authentification, limitation de débit, gestion de session
auth_token = "anthropic_key_xyz_authenticated"
# Ceci est la frontière de tour — entourer avec le contexte thread au niveau 3
with weave.thread(thread_id):
# Appeler le niveau 3 — imbriquant davantage la chaîne d'appels
return self.execute_anthropic_call(user_input, auth_token)
@weave.op
def execute_anthropic_call(self, user_input: str, auth_token: str) -> str:
"""Niveau 3 : FRONTIÈRE DE TOUR — Exécuter l'appel API Anthropic"""
print(f" N3: Exécution de l'appel API Anthropic avec auth")
response = f"Réponse de Anthropic Claude (auth : {auth_token[:15]}...) : {user_input}"
return response
class MultiProviderAgent:
"""Agent principal qui route entre les fournisseurs avec différentes profondeurs de chaîne d'appels"""
def __init__(self):
self.openai_provider = OpenAIProvider()
self.anthropic_provider = AnthropicProvider()
def handle_conversation_turn(self, user_input: str, thread_id: str) -> str:
"""
Router vers différents fournisseurs avec des profondeurs de chaîne d'appels inégales.
Le contexte thread est appliqué à différents niveaux d'imbrication dans chaque chaîne.
"""
# Choisir aléatoirement un fournisseur pour la démonstration
use_openai = random.choice([True, False])
if use_openai:
print(f"Choix d'OpenAI (chaîne d'appels à 2 niveaux)")
# OpenAI : Niveau 1 → Niveau 2 (frontière de tour)
response = self.openai_provider.route_to_openai(user_input, thread_id)
return f"[Branche OpenAI] {response}"
else:
print(f"Choix d'Anthropic (chaîne d'appels à 3 niveaux)")
# Anthropic : Niveau 1 → Niveau 2 → Niveau 3 (frontière de tour)
response = self.anthropic_provider.route_to_anthropic(user_input, thread_id)
return f"[Branche Anthropic] {response}"
async def main():
agent = MultiProviderAgent()
conversation_id = "nested_depth_conversation_999"
# Multi-turn conversation with different call chain depths
conversation_turns = [
"Qu'est-ce que l'apprentissage profond ?",
"Expliquez la rétropropagation dans les réseaux de neurones",
"Comment fonctionnent les mécanismes d'attention ?",
"Quelle est l'architecture transformer ?",
"Comparez les CNN aux RNN"
]
print(f"Démarrage de la conversation : {conversation_id}")
for i, user_input in enumerate(conversation_turns, 1):
print(f"\n--- Tour {i} ---")
print(f"Utilisateur : {user_input}")
# Même thread_id utilisé à travers différentes profondeurs de chaîne d'appels
response = agent.handle_conversation_turn(user_input, conversation_id)
print(f"Agent : {response}")
if __name__ == "__main__":
asyncio.run(main())
# Résultat attendu : Un seul thread avec 5 tours
# - Tours OpenAI : contexte thread au niveau 2 dans la chaîne d'appels
# Pile d'appels : route_to_openai() → execute_openai_call() ← contexte thread ici
# - Tours Anthropic : contexte thread au niveau 3 dans la chaîne d'appels
# Pile d'appels : route_to_anthropic() → authenticate_anthropic() → execute_anthropic_call() ← contexte thread ici
# - Tous les tours partagent thread_id : « nested_depth_conversation_999 »
# - Frontières de tour marquées à différentes profondeurs de pile d'appels
# - Opérations de support dans la chaîne d'appels suivies comme des appels imbriqués, pas des tours
Reprendre une session précédente
thread_id à None, car cela désactiverait complètement le regroupement des threads. À la place, fournissez toujours un ID de thread valide. Si vous devez créer un nouveau thread, générez un identifiant unique à l’aide d’une fonction comme generate_id().
Lorsqu’aucun thread_id n’est spécifié, l’implémentation interne de Weave génère automatiquement un UUID v7 aléatoire. Vous pouvez reproduire ce comportement dans votre propre fonction generate_id() ou utiliser n’importe quelle chaîne unique de votre choix.
import weave
import uuidv7
import argparse
def generate_id():
"""Génère un identifiant de thread unique en utilisant UUID v7."""
return str(uuidv7.uuidv7())
@weave.op
def load_history(session_id):
"""Charge l'historique de conversation pour la session donnée."""
# Votre implémentation ici
return []
# Analyser les arguments de ligne de commande pour la reprise de session
parser = argparse.ArgumentParser()
parser.add_argument("--session-id", help="ID de session existant à reprendre")
args = parser.parse_args()
# Déterminer l'ID de thread : reprendre une session existante ou en créer une nouvelle
if args.session_id:
thread_id = args.session_id
print(f"Reprise de la session : {thread_id}")
else:
thread_id = generate_id()
print(f"Démarrage d'une nouvelle session : {thread_id}")
# Établir le contexte de thread pour le suivi des appels
with weave.thread(thread_id) as thread_ctx:
# Charger ou initialiser l'historique de conversation
history = load_history(thread_id)
print(f"ID de thread actif : {thread_ctx.thread_id}")
# Votre logique applicative ici...
ThreadContext partagé. Utilisez ce modèle si vous souhaitez analyser ou surveiller différentes parties du système indépendamment, tout en les rattachant à une session commune.
import weave
from contextlib import contextmanager
from typing import Dict
# Contexte de thread global pour la coordination des threads imbriqués
class ThreadContext:
def __init__(self):
self.app_thread_id = None
self.infra_thread_id = None
self.logic_thread_id = None
def setup_for_request(self, request_id: str):
self.app_thread_id = f"app_{request_id}"
self.infra_thread_id = f"{self.app_thread_id}_infra"
self.logic_thread_id = f"{self.app_thread_id}_logic"
# Instance globale
thread_ctx = ThreadContext()
class InfrastructureLayer:
"""Gère toutes les opérations d'infrastructure dans un thread dédié"""
@weave.op
def authenticate_user(self, user_id: str) -> Dict:
# Logique d'authentification...
return {"user_id": user_id, "authenticated": True}
@weave.op
def call_payment_gateway(self, amount: float) -> Dict:
# Traitement du paiement...
return {"status": "approved", "amount": amount}
@weave.op
def update_inventory(self, product_id: str, quantity: int) -> Dict:
# Gestion des stocks...
return {"product_id": product_id, "updated": True}
def execute_operations(self, user_id: str, order_data: Dict) -> Dict:
"""Exécute toutes les opérations d'infrastructure dans le contexte de thread dédié"""
with weave.thread(thread_ctx.infra_thread_id):
auth_result = self.authenticate_user(user_id)
payment_result = self.call_payment_gateway(order_data["amount"])
inventory_result = self.update_inventory(order_data["product_id"], order_data["quantity"])
return {
"auth": auth_result,
"payment": payment_result,
"inventory": inventory_result
}
class BusinessLogicLayer:
"""Gère la logique métier dans un thread dédié"""
@weave.op
def validate_order(self, order_data: Dict) -> Dict:
# Logique de validation...
return {"valid": True}
@weave.op
def calculate_pricing(self, order_data: Dict) -> Dict:
# Calculs de tarification...
return {"total": order_data["amount"], "tax": order_data["amount"] * 0.08}
@weave.op
def apply_business_rules(self, order_data: Dict) -> Dict:
# Règles métier...
return {"rules_applied": ["standard_processing"], "priority": "normal"}
def execute_logic(self, order_data: Dict) -> Dict:
"""Exécute toute la logique métier dans le contexte de thread dédié"""
with weave.thread(thread_ctx.logic_thread_id):
validation = self.validate_order(order_data)
pricing = self.calculate_pricing(order_data)
rules = self.apply_business_rules(order_data)
return {"validation": validation, "pricing": pricing, "rules": rules}
class OrderProcessingApp:
"""Orchestrateur principal de l'application"""
def __init__(self):
self.infra = InfrastructureLayer()
self.business = BusinessLogicLayer()
@weave.op
def process_order(self, user_id: str, order_data: Dict) -> Dict:
"""Traitement principal des commandes — devient un tour dans le thread de l'application"""
# Exécute les opérations imbriquées dans leurs threads dédiés
infra_results = self.infra.execute_operations(user_id, order_data)
logic_results = self.business.execute_logic(order_data)
# Orchestration finale
return {
"order_id": f"order_12345",
"status": "completed",
"infra_results": infra_results,
"logic_results": logic_results
}
# Utilisation avec coordination du contexte de thread global
def handle_order_request(request_id: str, user_id: str, order_data: Dict):
# Configuration du contexte de thread pour cette requête
thread_ctx.setup_for_request(request_id)
# Exécution dans le contexte de thread de l'application
with weave.thread(thread_ctx.app_thread_id):
app = OrderProcessingApp()
result = app.process_order(user_id, order_data)
return result
# Exemple d'utilisation
order_result = handle_order_request(
request_id="req_789",
user_id="user_001",
order_data={"product_id": "laptop", "quantity": 1, "amount": 1299.99}
)
# Structure de threads attendue :
#
# Thread applicatif : app_req_789
# └── Tour : process_order() ← Orchestration principale
#
# Thread d'infrastructure : app_req_789_infra
# ├── Tour : authenticate_user() ← Opération d'infrastructure 1
# ├── Tour : call_payment_gateway() ← Opération d'infrastructure 2
# └── Tour : update_inventory() ← Opération d'infrastructure 3
#
# Thread de logique : app_req_789_logic
# ├── Tour : validate_order() ← Opération de logique métier 1
# ├── Tour : calculate_pricing() ← Opération de logique métier 2
# └── Tour : apply_business_rules() ← Opération de logique métier 3
#
# Avantages :
# - Séparation claire des responsabilités entre les threads
# - Pas de passage des IDs de thread en paramètre
# - Surveillance indépendante des couches application/infrastructure/logique
# - Coordination globale via le contexte de thread
POST /threads/query
class ThreadsQueryReq:
project_id: str
limit: Optional[int] = None
offset: Optional[int] = None
sort_by: Optional[list[SortBy]] = None # Champs pris en charge : thread_id, turn_count, start_time, last_updated
sortable_datetime_after: Optional[datetime] = None # Filtrer les threads par optimisation de granularité
sortable_datetime_before: Optional[datetime] = None # Filtrer les threads par optimisation de granularité
class ThreadSchema:
thread_id: str # Identifiant unique du thread
turn_count: int # Nombre d'appels de tour dans ce thread
start_time: datetime # Heure de début la plus ancienne des appels de tour dans ce thread
last_updated: datetime # Heure de fin la plus récente des appels de tour dans ce thread
class ThreadsQueryRes:
threads: List[ThreadSchema]
Requête sur les threads récemment actifs
my-project par l’ID réel de votre projet.
# Obtenir les threads les plus récemment actifs
response = client.threads_query(ThreadsQueryReq(
project_id="my-project",
sort_by=[SortBy(field="last_updated", direction="desc")],
limit=50
))
for thread in response.threads:
print(f"Thread {thread.thread_id}: {thread.turn_count} turns, last active {thread.last_updated}")
Interroger les threads par niveau d’activité
# Obtenir les threads avec le plus d'activité (le plus de tours)
response = client.threads_query(ThreadsQueryReq(
project_id="my-project",
sort_by=[SortBy(field="turn_count", direction="desc")],
limit=20
))
Interroger uniquement les threads récents
days dans timedelta.
from datetime import datetime, timedelta
# Obtenir les threads démarrés au cours des dernières 24 heures
yesterday = datetime.now() - timedelta(days=1)
response = client.threads_query(ThreadsQueryReq(
project_id="my-project",
sortable_datetime_after=yesterday,
sort_by=[SortBy(field="start_time", direction="desc")]
))
