Skip to main content

ActionSpan

O ActionSpan é um wrapper seguro e simples para rastreamento de observabilidade em Actions usando o LangFuse. Garante que operações de tracing nunca interrompam a execução das Actions, proporcionando uma API mínima e robusta para monitoramento de performance.

O que é LangFuse?​

LangFuse é uma plataforma open-source de observabilidade para aplicações LLM que permite:

  • Rastreamento completo: Monitora chamadas LLM, APIs, retrieval e outras operações
  • Análise de performance: Acompanha latência, custos e qualidade das respostas
  • Debugging avançado: Visualiza fluxos complexos e identifica gargalos
  • Métricas em tempo real: Dashboard com insights de uso e performance

Classe ActionSpan​

Descrição e Objetivo​

A classe ActionSpan oferece uma interface segura para instrumentação de observabilidade em Actions. Seu principal objetivo é permitir rastreamento detalhado de operações sem risco de falha na execução principal da Action.

Características principais:

  • Fail-safe: Nunca quebra a Action, mesmo com erros de trace
  • API mínima: Interface simples e intuitiva

Métodos​

event(event_name: str, data: Optional[Dict[str, Any]] = None) -> None​

Registra um evento simples no trace, ideal para marcos importantes na execução.

Parâmetros obrigatórios:

  • event_name (str): Nome descritivo do evento

Parâmetros opcionais:

  • data (Dict[str, Any]): Dados adicionais do evento

Quando utilizar:

  • Marcar etapas importantes do processo
  • Registrar validações realizadas
  • Documentar coleta de dados do usuário
  • Logging de estados intermediários

Resultado esperado:

  • Evento registrado no span "tool-execution" para aquela determinada sessão do LangFuse.

Exemplos:

# Registro simples de marco
action_span.event("action_started")

# Evento com dados de validação
action_span.event("cpf_validation", {
    "cpf": "123.456.789-00",
    "valid": True,
    "validation_method": "algorithm"
})

# Coleta de dados do usuário
action_span.event("user_data_collected", {
    "field": "nome",
    "value": "João Silva",
    "source": "user_input"
})

# Estado de processamento
action_span.event("api_call_completed", {
    "endpoint": "/api/users",
    "status_code": 200,
    "response_time_ms": 150
})

span(name: str, input_data: Optional[Dict[str, Any]] = None) -> ActionSpan​

Cria um sub-span para rastrear operações específicas dentro da Action.

Parâmetros obrigatórios:

  • name (str): Nome descritivo do sub-span

Parâmetros opcionais:

  • input_data (Dict[str, Any]): Dados de entrada da operação

Quando utilizar:

  • Rastrear operações complexas (validações, APIs externas)
  • Medir performance de etapas específicas
  • Organizar hierarquicamente o trace

Resultado esperado:

  • Novo span criado dentro do span principal "tool-execution".

Exemplos:

# Validação de CPF com sub-span
def validar_cpf(cpf: str, action_span: ActionSpan):
    validation_span = action_span.span("cpf_validation", {
        "cpf": cpf,
        "validation_type": "algorithm"
    })
    
    # Lógica de validação
    is_valid = realizar_validacao_cpf(cpf)
    
    validation_span.end({
        "valid": is_valid,
        "duration_ms": 45
    })
    
    return is_valid

# Chamada para API externa
def consultar_cep(cep: str, action_span: ActionSpan):
    api_span = action_span.span("cep_api_call", {
        "cep": cep,
        "provider": "viacep"
    })
    
    try:
        response = requests.get(f"https://viacep.com.br/ws/{cep}/json/")
        data = response.json()
        
        api_span.end({
            "success": True,
            "response_data": data,
            "status_code": response.status_code
        })
        
        return data
    except Exception as e:
        api_span.end({
            "success": False,
            "error": str(e)
        })
        raise

end(output_data: Optional[Dict[str, Any]] = None) -> None​

Finaliza o span atual, registrando dados de saída e conclusão.

Parâmetros opcionais:

  • output_data (Dict[str, Any]): Dados de resultado da operação

Quando utilizar:

  • Ao final de sub-spans criados com .span()
  • Para registrar resultados finais
  • Quando quiser marcar explicitamente o fim de uma operação
  • Para incluir métricas de performance

Resultado esperado:

  • Span finalizado no LangFuse com dados de saída

Exemplos:

# Finalização simples
validation_span.end()

# Com dados de resultado
validation_span.end({
    "result": "valid",
    "confidence": 0.95,
    "duration_ms": 150,
    "checks_performed": ["format", "digit", "algorithm"]
})

# Com métricas de performance
api_span.end({
    "success": True,
    "response_time_ms": 230,
    "payload_size_bytes": 1024,
    "cache_hit": False
})

Boas Práticas​

✅ Faça​

  • Use nomes descritivos para eventos e spans
  • Inclua dados relevantes mas não sensíveis
  • Crie sub-spans para operações importantes
  • Sempre finalize sub-spans com .end()
  • Use event() para marcos simples

❌ Evite​

  • Não incluir dados sensíveis (senhas, tokens)
  • Não criar spans desnecessariamente aninhados
  • Não fazer logging excessivo pois isso pode poluir e atrapalhar a visibilidade do que importa
  • Não usar para substituir logs tradicionais
Dica importante

O ActionSpan foi projetado para ser completamente fail-safe. Mesmo que ocorram erros no sistema de observabilidade, sua Action continuará funcionando normalmente.

Exemplo Completo: Gerador de Nome Pirata​

Para demonstrar o uso completo do ActionSpan, vamos analisar uma Action real que gera nomes piratas usando LLM. Este exemplo mostra como usar todos os métodos do ActionSpan em uma implementação prática.

Código da Action​

def gerar_nome_pirata(name: str):
    # Definir schema JSON para resposta estruturada
    pirate_response_schema = {
        "type": "json_schema",
        "json_schema": {
            "name": "pirate_name_response",
            "schema": {
                "type": "object",
                "properties": {
                    "pirate_name": {
                        "type": "string",
                        "description": "Nome de pirata gerado"
                    },
                    "pirate_title": {
                        "type": "string", 
                        "description": "Título ou alcunha do pirata"
                    },
                    "origin_explanation": {
                        "type": "string",
                        "description": "Explicação de como o nome foi derivado do nome original"
                    },
                    "pirate_backstory": {
                        "type": "string",
                        "description": "Breve história de fundo do pirata"
                    },
                    "ship_name": {
                        "type": "string",
                        "description": "Nome do navio pirata"
                    }
                },
                "required": ["pirate_name", "pirate_title", "origin_explanation", "pirate_backstory", "ship_name"],
                "additionalProperties": False
            }
        }
    }
    
    send_status_message("gerando nome pirata")
    gerar_span = action_span.span("gerar nome pirata span")
    
    # Event de início da ação
    gerar_span.event("gerar_nome_pirata_started", {
        "action": "gerar_nome_pirata",
        "input_name": name
    })
    
    # Validar entrada
    if not name or not name.strip():
        gerar_span.event("invalid_name_provided", {"name": name})
        
        return actions_sdk.ResponseToUser(
            status=actions_sdk.ResponseStatus.ERROR,
            message="❌ Nome inválido! Por favor, forneça um nome válido para gerar seu nome pirata.",
            instruction="Nome inválido fornecido pelo usuário"
        )
    
    # Preparar prompts para o LLM
    system_prompt = """Você é um criativo gerador de nomes piratas. Sua tarefa é transformar nomes comuns em nomes de piratas épicos e memoráveis.

    INSTRUÇÕES:
    - Crie um nome pirata baseado no nome fornecido
    - Inclua um título/alcunha marcante (ex: "o Terrível", "Barba Negra", "Punhal Dourado")
    - Explique como derivou o nome pirata do nome original
    - Crie uma breve história de fundo interessante (2-3 frases)
    - Invente um nome criativo para o navio do pirata
    - Seja criativo, divertido e épico!
    - Use elementos marítimos, tesouros, aventuras e lendas
    - IMPORTANTE: Responda SEMPRE em português brasileiro"""

    user_prompt = f"""Transforme o nome "{name}" em um nome de pirata épico e memorável.

    Crie:
    1. Um nome pirata único baseado em "{name}"
    2. Um título/alcunha impressionante
    3. Explicação de como derivou o nome
    4. Uma história de fundo fascinante
    5. Nome do navio pirata

    Seja criativo e divertido!
"""

    # Fazer chamada LLM com response_format estruturado
    try:
        gerar_span.event("llm_request_started", {
            "model": "completion",
            "has_response_format": True
        })
        
        llm_result = llm_helper.completion(
            system_prompt=system_prompt,
            user_prompt=user_prompt,
            response_format=pirate_response_schema,
            temperature=0.8,  # Mais criativo
            generation_name="pirate-name-generation"
        )
        
        gerar_span.event("llm_request_completed", {
            "response_received": True,
            "has_structured_data": "pirate_name" in llm_result
        })
        
    except Exception as e:
        gerar_span.event("llm_request_failed", {
            "error": str(e),
            "error_type": type(e).__name__
        })
        
        return actions_sdk.ResponseToUser(
            status=actions_sdk.ResponseStatus.ERROR,
            message="🏴‍☠️ Arrr! Algo deu errado na geração do seu nome pirata. Tente novamente, marujo!",
            instruction=f"Erro na chamada LLM: {str(e)}",
            detail=str(e)
        )
    
    # Extrair dados estruturados da resposta
    pirate_data = llm_result
    
    # Validar se recebemos todos os campos esperados
    required_fields = ["pirate_name", "pirate_title", "origin_explanation", "pirate_backstory", "ship_name"]
    missing_fields = [field for field in required_fields if field not in pirate_data]
    
    if missing_fields:
        gerar_span.event("incomplete_llm_response", {
            "missing_fields": missing_fields,
            "received_fields": list(pirate_data.keys())
        })
        
        return actions_sdk.ResponseToUser(
            status=actions_sdk.ResponseStatus.WARNING,
            message="🏴‍☠️ O nome pirata foi gerado, mas algumas informações ficaram incompletas. Aqui está o que conseguimos:",
            instruction="Resposta LLM incompleta - alguns campos faltando"
        )
    
    # Construir mensagem formatada para o usuário
    pirate_message = f"""🏴‍☠️ **AHOY, {name.upper()}!** 🏴‍☠️

⚔️ **Seu Nome Pirata:** {pirate_data['pirate_name']} {pirate_data['pirate_title']}

🗺️ **Origem do Nome:**
{pirate_data['origin_explanation']}

📜 **Sua História:**
{pirate_data['pirate_backstory']}

🚢 **Seu Navio:** {pirate_data['ship_name']}

---
*Que os ventos estejam sempre a seu favor, capitão!* ⚓"""

    # Armazenar na memória para referência futura
    memory_manager.store_user_info("nome_pirata", pirate_data['pirate_name'])
    memory_manager.store_user_info("titulo_pirata", pirate_data['pirate_title'])
    memory_manager.store_conversation_fact(f"Usuário {name} recebeu nome pirata: {pirate_data['pirate_name']} {pirate_data['pirate_title']}")
    
    # Atualizar estado com informações do pirata
    pirate_state = {
        "original_name": name,
        "pirate_name": pirate_data['pirate_name'],
        "pirate_title": pirate_data['pirate_title'],
        "ship_name": pirate_data['ship_name'],
        "generated_at": datetime.now().isoformat()
    }
    state_manager.update("pirate_identity", pirate_state)
    
    # Event de finalização
    gerar_span.event("gerar_nome_pirata_completed", {
        "success": True,
        "pirate_name": pirate_data['pirate_name'],
        "original_name": name,
        "message_length": len(pirate_message)
    })
    
    gerar_span.end()
    send_status_message("nome pirata gerado")
    
    # Retornar ResponseToUser para ir direto ao usuário
    return actions_sdk.ResponseToUser(
        status=actions_sdk.ResponseStatus.SUCCESS,
        message=pirate_message,
        instruction=f"Nome pirata gerado com sucesso para {name}",
        state_updates=pirate_state,
        memory_updates={
            "pirate_generation_completed": True,
            "last_pirate_generation": datetime.now().isoformat(),
            "generated_for": name
        }
    )

Visualização no LangFuse​

O código acima gera a seguinte estrutura de rastreamento no LangFuse:

Visualização do ActionSpan no LangFuse

Análise do Trace​

A imagem mostra como o ActionSpan organiza a observabilidade:

  1. Span Principal (tool-execution): Representa a execução completa da Action
  2. Sub-span (gerar nome pirata span): Criado com .span() para rastrear a operação específica
  3. Eventos sequenciais:
    • gerar_nome_pirata_started: Marco inicial com dados de entrada
    • llm_request_started: Início da chamada LLM
    • llm_request_completed: Finalização da chamada LLM
    • gerar_nome_pirata_completed: Conclusão da Action com resultados

Benefícios da Instrumentação​

Esta implementação demonstra como o ActionSpan oferece:

  • Visibilidade completa: Cada etapa da Action é rastreada
  • Debugging facilitado: Eventos mostram exatamente onde problemas podem ocorrer
  • Métricas de performance: Duração de cada operação é medida
  • Contexto preservado: Dados relevantes são mantidos para análise
  • Hierarquia clara: Sub-spans organizam operações complexas