Suivre les 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:
        """
        OPÉRATION AU NIVEAU DU TOUR : Ceci représente un tour.
        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 statistiques 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

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 de routage, prétraitement de base
        print(f"  L1: Routing to OpenAI for: {user_input}")

        # Ceci est la limite de tour de conversation - encapsuler avec le contexte de thread
        with weave.thread(thread_id):
            # Appeler le niveau 2 directement - 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:
        """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 de routage, sélection du fournisseur
        print(f"  L1: Routing to Anthropic for: {user_input}")

        # Appeler le niveau 2 - cela 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, limite de débit, gestion de session
        auth_token = "anthropic_key_xyz_authenticated"

         # Ceci est la limite de tour de conversation - encapsuler avec le contexte de thread au niveau 3
        with weave.thread(thread_id):
            # Appeler le niveau 3 - imbriquer 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 aléatoirement à des fins de 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 de 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 de 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 de conversation avec différentes profondeurs de chaîne d'appels
    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}")

        # Même thread_id utilisé pour 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 : thread unique avec 5 tours de conversation
# - Tours OpenAI : contexte de thread au niveau 2 dans 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 dans 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 de tour de conversation marquées à différentes profondeurs de pile d'appels
# - Les opérations de support dans la chaîne d'appels sont suivies comme des appels imbriqués, et non comme des tours de conversation

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 ligne de commande pour la reprise de session
parser = argparse.ArgumentParser()
parser.add_argument("--session-id", help="Existing session ID to resume")
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"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 conversation
    history = load_history(thread_id)
    print(f"Active thread ID: {thread_ctx.thread_id}")

    # La logique de votre application ici.

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:
    """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 du 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 du 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 d'application : app_req_789
# └── Tour de conversation : process_order() ← Orchestration principale
#
# Thread d'infrastructure : app_req_789_infra
# ├── Tour de conversation : authenticate_user() ← Opération d'infrastructure 1
# ├── Tour de conversation : call_payment_gateway() ← Opération d'infrastructure 2
# └── Tour de conversation : update_inventory() ← Opération d'infrastructure 3
#
# Thread de logique : app_req_789_logic
# ├── Tour de conversation : validate_order() ← Opération de logique métier 1
# ├── Tour de conversation : calculate_pricing() ← Opération de logique métier 2
# └── Tour de conversation : apply_business_rules() ← Opération de logique métier 3
#
# Avantages :
# - Séparation claire des responsabilités entre les threads
# - Pas de passage des ID de threads en paramètre
# - Surveillance indépendante des couches application/infrastructure/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   # 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]

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

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

from datetime import datetime, timedelta

# Obtenir les threads lancé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")]
))