Observabilidade (ActionSpan)

O ActionSpan é um wrapper fail-safe sobre o sistema de tracing da plataforma. Use para registrar eventos, medir duração de operações e criar hierarquias de spans para observabilidade em ferramentas como Langfuse.

Fail-safe por design

O ActionSpan nunca lança exceções. Se o tracer não estiver disponível (ctx.tracer is None) ou qualquer operação de tracing falhar, todos os métodos são no-op silenciosos. A action continua executando normalmente.

Inicialização

from actions_sdk.core.types.action_context import ActionContext
from actions_sdk.features.tracing import ActionSpan

ctx: ActionContext

span = ActionSpan(ctx)

Funciona com ActionContext e TaskContext. Extrai ctx.tracer automaticamente.

---

Métodos

event(event_name, data=None)

Registra um evento pontual (sem duração). Use para marcar momentos importantes na execução.

span = ActionSpan(ctx)

# Evento sem dados
span.event("action.inicio")

# Evento com dados
span.event("action.validacao.ok", {"cpf": "mascarado", "resultado": "valido"})
span.event("action.api.chamada", {"url": "https://api.exemplo.com", "status": 200})

Parâmetros:

Parâmetro	Tipo	Descrição
event_name	str	Nome do evento (use notação com ponto: "dominio.acao.estado")
data	dict | None	Dados adicionais do evento

---

span(name, input_data=None, span_kind=None)

Cria um sub-span para medir a duração de uma operação. Retorna um novo ActionSpan que deve ser finalizado com .end().

span = ActionSpan(ctx)

# Sub-span simples
sub = span.span("buscar_dados", {"fonte": "api_interna"})
# ... operação ...
sub.end({"registros": 42})

# Sub-span com span_kind semântico
sub_retrieval = span.span("buscar_kb", input_data={"query": "politica"}, span_kind="RETRIEVER")
sub_retrieval.end({"chunks": 3})

Parâmetros:

Parâmetro	Tipo	Descrição
name	str	Nome do sub-span
input_data	dict | None	Dados de entrada da operação
span_kind	str | None	Tipo semântico OpenInference (ver tabela abaixo)

Retorna: ActionSpan — novo span (no-op se tracer indisponível).

Span kinds disponíveis:

Valor	Quando usar
"RETRIEVER"	Operações de busca/retrieval em knowledge bases
"TOOL"	Chamadas a ferramentas externas
"CHAIN"	Pipelines de processamento sequencial
"GUARDRAIL"	Validações e filtros de segurança
"EMBEDDING"	Geração de embeddings
"RERANKER"	Re-ranking de resultados de busca
"AGENT"	Sub-agentes
"EVALUATOR"	Avaliações de resposta

---

end(output_data=None)

Finaliza o span atual, registrando os dados de saída.

sub = span.span("processar_pedido", {"pedido_id": "001"})
# ... processamento ...
sub.end({"status": "concluido", "itens": 3})

Parâmetros:

Parâmetro	Tipo	Descrição
output_data	dict | None	Dados de saída da operação

---

Métodos de conveniência por span_kind

Atalhos para criar sub-spans com span_kind pré-definido:

# Equivalentes a span.span(name, input_data, span_kind="...")
span.retriever_span("busca_kb", {"query": "garantia"})
span.tool_span("chamar_api_externa", {"url": "..."})
span.chain_span("pipeline_classificacao", {"etapas": 3})
span.guardrail_span("validar_cpf", {"cpf_mascarado": "***"})
span.embedding_span("gerar_embedding", {"texto": "..."})
span.reranker_span("rerank_resultados", {"chunks": 10})
span.agent_span("sub_agente_especializado", {})
span.evaluator_span("avaliar_resposta", {})

---

is_available()

Verifica se o tracer está disponível.

if span.is_available():
    # tracing ativo
    pass

---

Spans aninhados

O ActionSpan suporta hierarquias de spans em qualquer profundidade:

span = ActionSpan(ctx)

# Nível 1 — operação principal
span_api = span.span("integracao_api_bancaria", {"operacao": "consulta_saldo"})
span.event("operacao_iniciada", {"nivel": 1})

# Nível 2 — sub-operação
span_auth = span_api.span("autenticar_api", {"metodo": "oauth2"})
span_auth.event("auth_request", {"endpoint": "/auth/token"})

# Nível 3 — operação atômica
span_val = span_auth.span("validar_credenciais", {"tipo": "client_credentials"})
span_val.event("validacao_concluida", {"status": "valido"})
span_val.end({"resultado": "credenciais_validas"})

# Fechar nível 2
span_auth.end({"token_obtido": True})

# Continuar no nível 1
span_consulta = span_api.span("consultar_saldo", {"autenticado": True})
span_consulta.end({"saldo": 1500.00})
span_api.end({"operacao": "concluida"})

---

Boas práticas

• Sempre registre início e fim: span.event("action.inicio", {...}) no começo, evento de sucesso ou erro no fim
• Use span_kind para operações semânticas (retrieval, ferramentas, chains)
• Não inclua dados sensíveis nos payloads de tracing (CPFs, senhas, tokens)
• Chame .end() em todos os sub-spans criados com .span() — spans não finalizados podem gerar dados incompletos

---

Exemplo completo — spans aninhados

Baseado na action testar_action_span_aninhado do agente de homologação:

import traceback

from actions_sdk.core.types.action_context import ActionContext
from actions_sdk.core.types.responses import ResponseStatus, ResponseToAgent
from actions_sdk.features.conversational.state import StateManager
from actions_sdk.features.tracing import ActionSpan

ctx: ActionContext

CHAVE_PROGRESSO = "action.tracing.progresso"


def consultar_api_com_tracing():
    span = ActionSpan(ctx)
    state = StateManager(ctx)

    try:
        # Span principal
        span_principal = span.span(
            "consulta_api_externa",
            {"operacao": "buscar_dados_cliente"},
        )
        span.event("operacao_iniciada", {"nivel": 1})

        # Sub-span: autenticação
        span_auth = span_principal.span("autenticar", {"metodo": "api_key"})
        span_auth.event("auth_ok", {"expira_em": 3600})
        span_auth.end({"autenticado": True})

        # Sub-span: chamada principal (span_kind semântico)
        span_dados = span_principal.tool_span(
            "buscar_dados",
            {"endpoint": "/clientes/dados"},
        )
        span_dados.event("resposta_recebida", {"status": 200, "registros": 5})
        span_dados.end({"registros_retornados": 5})

        span_principal.end({"operacao": "concluida", "total_registros": 5})

        state.update(CHAVE_PROGRESSO, "SUCESSO ✓")

        return ResponseToAgent(
            status=ResponseStatus.SUCCESS,
            instruction="Consulta com tracing concluída. 5 registros retornados.",
            state_updates=state.get_full_state(),
        )

    except Exception as erro:
        state.update(CHAVE_PROGRESSO, "ERRO ✗")
        span.event("action.erro", {"erro": str(erro)})
        return ResponseToAgent(
            status=ResponseStatus.ERROR,
            instruction=f"Falha na consulta: {traceback.format_exc()}",
            state_updates=state.get_full_state(),
        )