Tracer des threads

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")

import weave

class ConversationAgent:
    @weave.op
    def process_user_message(self, message: str) -> str:
        """
        TURN-LEVEL OPERATION: This represents one conversation turn.
        Only this function will be counted in thread statistics.
        """
        # Stocker le message de l’utilisateur
        # Générer la réponse de l’IA au moyen d’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:
        """NESTED CALL: Implementation details, not counted in thread stats."""
        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 une base de données vectorielle, requête sur une base de connaissances, etc.
        return "retrieved_context"

    @weave.op
    def _classify_intent(self, message: str) -> str:
        # Logique de classification de l’intention
        return "general_inquiry"

    @weave.op
    def _call_llm(self, message: str, context: str) -> str:
        # Appel d’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 de thread établi automatiquement
agent = ConversationAgent()

# Établir le contexte de thread : chaque appel à process_user_message devient un tour de conversation
with weave.thread() as thread_ctx:  # Génère automatiquement thread_id
    print(f"Thread ID: {thread_ctx.thread_id}")

    # Chaque appel à process_user_message crée 1 tour de conversation + 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 : un thread de 3 tours de conversation, soit environ 15 à 20 appels au total (appels imbriqués compris)

# Variante : 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 de cette session
    agent.process_user_message("Can you summarize what we discussed?") # Tour 2 de cette session

import weave
import random
import asyncio

class OpenAIProvider:
    """OpenAI branch: 2 levels deep call chain to turn boundary"""

    @weave.op
    def route_to_openai(self, user_input: str, thread_id: str) -> str:
        """Level 1: Route and prepare OpenAI request"""
        # Validation des entrées, logique d’acheminement, prétraitement de base
        print(f"  L1: Routing to OpenAI for: {user_input}")

        # Limite du tour de conversation : encapsuler avec le contexte de thread
        with weave.thread(thread_id):
            # Appeler directement le niveau 2, ce qui 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:
        """Level 2: TURN BOUNDARY - Execute OpenAI API call"""
        print(f"    L2: Executing OpenAI API call")
        response = f"OpenAI GPT-4 response: {user_input}"
        return response


class AnthropicProvider:
    """Anthropic branch: 3 levels deep call chain to turn boundary"""

    @weave.op
    def route_to_anthropic(self, user_input: str, thread_id: str) -> str:
        """Level 1: Route and prepare Anthropic request"""
        # Validation des entrées, logique d’acheminement, sélection du fournisseur
        print(f"  L1: Routing to Anthropic for: {user_input}")

        # Appeler le niveau 2, ce qui crée la profondeur de la chaîne d’appels
        return self.authenticate_anthropic(user_input, thread_id)

    @weave.op
    def authenticate_anthropic(self, user_input: str, thread_id: str) -> str:
        """Level 2: Handle Anthropic authentication and setup"""
        print(f"    L2: Authenticating with Anthropic")

        # Authentification, limitation de débit, gestion de session
        auth_token = "anthropic_key_xyz_authenticated"

         # Limite du tour de conversation : encapsuler avec le contexte de thread au niveau 3
        with weave.thread(thread_id):
            # Appeler le niveau 3, ce qui imbrique 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:
        """Level 3: TURN BOUNDARY - Execute Anthropic API call"""
        print(f"      L3: Executing Anthropic API call with auth")
        response = f"Anthropic Claude response (auth: {auth_token[:15]}...): {user_input}"
        return response


class MultiProviderAgent:
    """Main agent that routes between providers with different call chain depths"""

    def __init__(self):
        self.openai_provider = OpenAIProvider()
        self.anthropic_provider = AnthropicProvider()

    def handle_conversation_turn(self, user_input: str, thread_id: str) -> str:
        """
        Route to different providers with imbalanced call chain depths.
        Thread context is applied at different nesting levels in each chain.
        """
        # Choisir un fournisseur au hasard pour la démonstration
        use_openai = random.choice([True, False])

        if use_openai:
            print(f"Choosing OpenAI (2-level call chain)")
            # OpenAI : niveau 1 → niveau 2 (limite du tour de conversation)
            response = self.openai_provider.route_to_openai(user_input, thread_id)
            return f"[OpenAI Branch] {response}"
        else:
            print(f"Choosing Anthropic (3-level call chain)")
            # Anthropic : niveau 1 → niveau 2 → niveau 3 (limite du tour de conversation)
            response = self.anthropic_provider.route_to_anthropic(user_input, thread_id)
            return f"[Anthropic Branch] {response}"


async def main():
    agent = MultiProviderAgent()
    conversation_id = "nested_depth_conversation_999"

    # Conversation sur plusieurs tours avec des profondeurs de chaîne d’appels différentes
    conversation_turns = [
        "What's deep learning?",
        "Explain neural network backpropagation",
        "How do attention mechanisms work?",
        "What's the transformer architecture?",
        "Compare CNNs vs RNNs"
    ]

    print(f"Starting conversation: {conversation_id}")

    for i, user_input in enumerate(conversation_turns, 1):
        print(f"\\n--- Turn {i} ---")
        print(f"User: {user_input}")

        # Le même thread_id est utilisé à 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 de conversation
# - Tours OpenAI : contexte de thread au niveau 2 de la chaîne d’appels
#   Pile d’appels : route_to_openai() → execute_openai_call() ← contexte de thread ici
# - Tours Anthropic : contexte de thread au niveau 3 de la chaîne d’appels
#   Pile d’appels : route_to_anthropic() → authenticate_anthropic() → execute_anthropic_call() ← contexte de thread ici
# - Tous les tours partagent le thread_id : "nested_depth_conversation_999"
# - Limites des tours marquées à différentes profondeurs de la pile d’appels
# - Les opérations intermédiaires de la chaîne d’appels sont suivies comme des appels imbriqués, et non comme des tours

import weave
import uuidv7
import argparse

def generate_id():
    """Generate a unique thread ID using UUID v7."""
    return str(uuidv7.uuidv7())

@weave.op
def load_history(session_id):
    """Load conversation history for the given session."""
    # Votre implémentation ici
    return []

# Analyser les arguments de la ligne de commande pour reprendre une session
parser = argparse.ArgumentParser()
parser.add_argument("--session-id", help="Existing session ID to resume")
args = parser.parse_args()

# Déterminer l’ID du thread : reprendre la session existante ou en créer une nouvelle
if args.session_id:
    thread_id = args.session_id
    print(f"Resuming session: {thread_id}")
else:
    thread_id = generate_id()
    print(f"Starting new 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 la conversation
    history = load_history(thread_id)
    print(f"Active thread ID: {thread_ctx.thread_id}")

    # Votre logique applicative ici.

import weave
from contextlib import contextmanager
from typing import Dict

# Contexte de thread global pour coordonner les 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:
    """Handles all infrastructure operations in a dedicated thread"""

    @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:
        """Execute all infrastructure operations in dedicated thread context"""
        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:
    """Handles business logic in a dedicated thread"""

    @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:
        """Execute all business logic in dedicated thread context"""
        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:
    """Main application orchestrator"""

    def __init__(self):
        self.infra = InfrastructureLayer()
        self.business = BusinessLogicLayer()

    @weave.op
    def process_order(self, user_id: str, order_data: Dict) -> Dict:
        """Main order processing - becomes a turn in the app thread"""

        # Exécuter 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 via un contexte de thread global
def handle_order_request(request_id: str, user_id: str, order_data: Dict):
    # Configurer le contexte de thread pour cette requête
    thread_ctx.setup_for_request(request_id)

    # Exécuter 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 :
#
# App Thread: app_req_789
# └── Turn: process_order() ← Orchestration principale
#
# Infra Thread: app_req_789_infra
# ├── Turn: authenticate_user() ← Opération d’infrastructure 1
# ├── Turn: call_payment_gateway() ← Opération d’infrastructure 2
# └── Turn: update_inventory() ← Opération d’infrastructure 3
#
# Logic Thread: app_req_789_logic
# ├── Turn: validate_order() ← Opération de logique métier 1
# ├── Turn: calculate_pricing() ← Opération de logique métier 2
# └── Turn: apply_business_rules() ← Opération de logique métier 3
#
# Avantages :
# - Séparation claire des responsabilités entre les threads
# - Aucune propagation des ID de thread de fonction en fonction via les paramètres
# - Surveillance indépendante des couches app/infra/logique
# - Coordination globale via le contexte de thread

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   # Filtre les threads via l'optimisation par granules
    sortable_datetime_before: Optional[datetime] = None  # Filtre les threads via l'optimisation par granules

class ThreadSchema:
    thread_id: str           # Identifiant unique du thread
    turn_count: int          # Nombre d’appels de tour de conversation dans ce thread
    start_time: datetime     # Heure de début la plus ancienne parmi les appels de tour de conversation de ce thread
    last_updated: datetime   # Heure de fin la plus récente parmi les appels de tour de conversation de ce thread

class ThreadsQueryRes:
    threads: List[ThreadSchema]

# Obtenir les threads actifs le plus récemment
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}")

# Obtenir les threads les plus actifs (ceux qui comptent le plus de tours de conversation)
response = client.threads_query(ThreadsQueryReq(
    project_id="my-project",
    sort_by=[SortBy(field="turn_count", direction="desc")],
    limit=20
))

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")]
))