Ir para o conteúdo

Observabilidade

src.observabilidade

Módulo de observabilidade para classificação de intents.

Este módulo fornece ferramentas para registrar e analisar eventos de classificação do Pede AI. Permite:

  • Registrar eventos: Cada classificação é logada em um CSV com detalhes como confiança, caminho usado (lookup, RAG, LLM) e o exemplo mais similar.
  • Analisar dados: Consultas DuckDB prontas para extrair insights dos logs, como casos de baixa confiança e distribuição de caminhos.

Componentes principais:

  • ObservabilidadeLogger: Logger thread-safe para registrar eventos de classificação em CSV.
Example 
from src.observabilidade import ObservabilidadeLogger
from src.observabilidade.consultas import distribuicao_caminhos

# Registrar evento
logger = ObservabilidadeLogger('logs/eventos.csv')
logger.registrar(
    thread_id='sessao_123',
    mensagem='Quero um X-Burguer',
    mensagem_norm='querer x-burguer',
    intent='pedido_lanche',
    confidence=0.95,
    caminho='rag_forte',
    top1_texto='quero um x-burguer',
    top1_intencao='pedido_lanche',
)

# Analisar distribuição de caminhos
dist = distribuicao_caminhos('logs/eventos.csv')
for item in dist:
    print(f'{item["caminho"]}: {item["total"]} eventos')
Note

Os logs são armazenados em CSV para facilitar análise posterior com DuckDB, pandas ou ferramentas de visualização.

See Also
  • ObservabilidadeLogger: Classe principal para registrar eventos.
  • src.observabilidade.consultas: Funções de análise com DuckDB.

ClarificacaoLogger(csv_path: Path | str)

Bases: BaseCsvLogger

Logger thread-safe para registrar eventos de clarificacao.

Source code in src/observabilidade/base_logger.py 
37
38
39
40
41
42
43
44
45
46
def __init__(self, csv_path: Path | str) -> None:
    """Inicializa o logger criando o arquivo CSV se necessario.

    Args:
        csv_path: Caminho para o arquivo CSV.
    """
    self._csv_path = Path(csv_path).resolve()
    self._csv_path.parent.mkdir(parents=True, exist_ok=True)
    self._lock = threading.Lock()
    self._inicializar_csv()

ExtracaoLogger(csv_path: Path | str)

Bases: BaseCsvLogger

Logger thread-safe para registrar eventos de extracao de itens.

Source code in src/observabilidade/base_logger.py 
37
38
39
40
41
42
43
44
45
46
def __init__(self, csv_path: Path | str) -> None:
    """Inicializa o logger criando o arquivo CSV se necessario.

    Args:
        csv_path: Caminho para o arquivo CSV.
    """
    self._csv_path = Path(csv_path).resolve()
    self._csv_path.parent.mkdir(parents=True, exist_ok=True)
    self._lock = threading.Lock()
    self._inicializar_csv()

FunilLogger(csv_path: Path | str)

Bases: BaseCsvLogger

Logger thread-safe para registrar transicoes no funil.

Source code in src/observabilidade/base_logger.py 
37
38
39
40
41
42
43
44
45
46
def __init__(self, csv_path: Path | str) -> None:
    """Inicializa o logger criando o arquivo CSV se necessario.

    Args:
        csv_path: Caminho para o arquivo CSV.
    """
    self._csv_path = Path(csv_path).resolve()
    self._csv_path.parent.mkdir(parents=True, exist_ok=True)
    self._lock = threading.Lock()
    self._inicializar_csv()

HandlerLogger(csv_path: Path | str)

Bases: BaseCsvLogger

Logger thread-safe para registrar execucoes de handlers.

Source code in src/observabilidade/base_logger.py 
37
38
39
40
41
42
43
44
45
46
def __init__(self, csv_path: Path | str) -> None:
    """Inicializa o logger criando o arquivo CSV se necessario.

    Args:
        csv_path: Caminho para o arquivo CSV.
    """
    self._csv_path = Path(csv_path).resolve()
    self._csv_path.parent.mkdir(parents=True, exist_ok=True)
    self._lock = threading.Lock()
    self._inicializar_csv()

ObservabilidadeLogger(csv_path: Path | str)

Bases: BaseCsvLogger

Logger thread-safe para registrar eventos de classificacao.

Source code in src/observabilidade/base_logger.py 
37
38
39
40
41
42
43
44
45
46
def __init__(self, csv_path: Path | str) -> None:
    """Inicializa o logger criando o arquivo CSV se necessario.

    Args:
        csv_path: Caminho para o arquivo CSV.
    """
    self._csv_path = Path(csv_path).resolve()
    self._csv_path.parent.mkdir(parents=True, exist_ok=True)
    self._lock = threading.Lock()
    self._inicializar_csv()

registrar(thread_id: str, mensagem: str, mensagem_norm: str, intent: str, confidence: float, caminho: str, top1_texto: str, top1_intencao: str, lookup: str = '', rag_top1: str = '', rag_sim: str = '', rag_intent: str = '', llm_raw: str = '', llm_intent: str = '') -> None

Registra um evento de classificacao no CSV.

Parameters:

Name Type Description Default
lookup str

Intencao encontrada por lookup (ou vazio).

 ''
rag_top1 str

Texto do top1 do RAG (ou vazio).

 ''
rag_sim str

Similaridade do top1 do RAG (ou vazio).

 ''
rag_intent str

Intencao do top1 do RAG (ou vazio).

 ''
llm_raw str

Resposta crua do LLM (ou vazio).

 ''
llm_intent str

Intencao extraida do LLM (ou vazio).

 ''

Raises:

Type Description
ValueError

Se caminho nao estiver em CAMINHOS_VALIDOS.
Source code in src/observabilidade/logger.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def registrar(
    self,
    thread_id: str,
    mensagem: str,
    mensagem_norm: str,
    intent: str,
    confidence: float,
    caminho: str,
    top1_texto: str,
    top1_intencao: str,
    lookup: str = '',
    rag_top1: str = '',
    rag_sim: str = '',
    rag_intent: str = '',
    llm_raw: str = '',
    llm_intent: str = '',
) -> None:
    """Registra um evento de classificacao no CSV.

    Args:
        lookup: Intencao encontrada por lookup (ou vazio).
        rag_top1: Texto do top1 do RAG (ou vazio).
        rag_sim: Similaridade do top1 do RAG (ou vazio).
        rag_intent: Intencao do top1 do RAG (ou vazio).
        llm_raw: Resposta crua do LLM (ou vazio).
        llm_intent: Intencao extraida do LLM (ou vazio).

    Raises:
        ValueError: Se ``caminho`` nao estiver em ``CAMINHOS_VALIDOS``.
    """
    super().registrar(
        thread_id=thread_id,
        mensagem=mensagem,
        mensagem_norm=mensagem_norm,
        intent=intent,
        confidence=confidence,
        caminho=caminho,
        top1_texto=top1_texto,
        top1_intencao=top1_intencao,
        lookup=lookup,
        rag_top1=rag_top1,
        rag_sim=rag_sim,
        rag_intent=rag_intent,
        llm_raw=llm_raw,
        llm_intent=llm_intent,
    )

get_clarificacao_logger() -> ClarificacaoLogger | None

Retorna o logger de clarificacao configurado.

Source code in src/observabilidade/registry.py 
93
94
95
def get_clarificacao_logger() -> ClarificacaoLogger | None:
    """Retorna o logger de clarificacao configurado."""
    return _registry.get('clarificacao')  # type: ignore[return-value]

get_extracao_logger() -> ExtracaoLogger | None

Retorna o logger de extracao configurado.

Source code in src/observabilidade/registry.py 
103
104
105
def get_extracao_logger() -> ExtracaoLogger | None:
    """Retorna o logger de extracao configurado."""
    return _registry.get('extracao')  # type: ignore[return-value]

get_funil_logger() -> FunilLogger | None

Retorna o logger de funil configurado.

Source code in src/observabilidade/registry.py 
123
124
125
def get_funil_logger() -> FunilLogger | None:
    """Retorna o logger de funil configurado."""
    return _registry.get('funil')  # type: ignore[return-value]

get_handler_logger() -> HandlerLogger | None

Retorna o logger de handler configurado.

Source code in src/observabilidade/registry.py 
113
114
115
def get_handler_logger() -> HandlerLogger | None:
    """Retorna o logger de handler configurado."""
    return _registry.get('handler')  # type: ignore[return-value]

get_obs_logger() -> ObservabilidadeLogger

Retorna o logger de observabilidade configurado.

Source code in src/observabilidade/registry.py 
83
84
85
def get_obs_logger() -> ObservabilidadeLogger:
    """Retorna o logger de observabilidade configurado."""
    return _registry.get_required('observabilidade')  # type: ignore[return-value]

set_clarificacao_logger(logger: ClarificacaoLogger | None) -> None

Configura o logger de clarificacao.

Source code in src/observabilidade/registry.py 
 98
 99
100
def set_clarificacao_logger(logger: ClarificacaoLogger | None) -> None:
    """Configura o logger de clarificacao."""
    _registry.set('clarificacao', logger)

set_extracao_logger(logger: ExtracaoLogger | None) -> None

Configura o logger de extracao.

Source code in src/observabilidade/registry.py 
108
109
110
def set_extracao_logger(logger: ExtracaoLogger | None) -> None:
    """Configura o logger de extracao."""
    _registry.set('extracao', logger)

set_funil_logger(logger: FunilLogger | None) -> None

Configura o logger de funil.

Source code in src/observabilidade/registry.py 
128
129
130
def set_funil_logger(logger: FunilLogger | None) -> None:
    """Configura o logger de funil."""
    _registry.set('funil', logger)

set_handler_logger(logger: HandlerLogger | None) -> None

Configura o logger de handler.

Source code in src/observabilidade/registry.py 
118
119
120
def set_handler_logger(logger: HandlerLogger | None) -> None:
    """Configura o logger de handler."""
    _registry.set('handler', logger)

set_obs_logger(logger: ObservabilidadeLogger) -> None

Configura o logger de observabilidade.

Source code in src/observabilidade/registry.py 
88
89
90
def set_obs_logger(logger: ObservabilidadeLogger) -> None:
    """Configura o logger de observabilidade."""
    _registry.set('observabilidade', logger)

src.observabilidade.base_logger

Classe base para loggers CSV thread-safe.

Elimina boilerplate duplicado em todos os loggers de observabilidade. Cada logger concreto precisa implementar apenas headers e _to_row.

Example 
from src.observabilidade.base_logger import BaseCsvLogger


class MeuLogger(BaseCsvLogger):
    @property
    def headers(self) -> list[str]:
        return ['timestamp', 'mensagem']

    def _to_row(self, **kwargs) -> list:
        return [kwargs.get('timestamp', ''), kwargs.get('mensagem', '')]

BaseCsvLogger(csv_path: Path | str)

Bases: ABC

Classe base abstrata para loggers CSV thread-safe.

Subclasses devem implementar: - headers: propriedade com lista de colunas do CSV. - _to_row(**kwargs): metodo que converte kwargs em lista de valores.

Inicializa o logger criando o arquivo CSV se necessario.

Parameters:

Name Type Description Default
csv_path Path | str

Caminho para o arquivo CSV.

 required
Source code in src/observabilidade/base_logger.py 
37
38
39
40
41
42
43
44
45
46
def __init__(self, csv_path: Path | str) -> None:
    """Inicializa o logger criando o arquivo CSV se necessario.

    Args:
        csv_path: Caminho para o arquivo CSV.
    """
    self._csv_path = Path(csv_path).resolve()
    self._csv_path.parent.mkdir(parents=True, exist_ok=True)
    self._lock = threading.Lock()
    self._inicializar_csv()

headers: list[str] abstractmethod property

Retorna lista de colunas do CSV.

csv_path: Path property

Retorna caminho absoluto do arquivo CSV.

registrar(**kwargs: str | float | int | None) -> None

Registra uma linha no CSV de forma thread-safe.

Parameters:

Name Type Description Default
**kwargs str | float | int | None

Dados a registrar (str, float, int ou None). Serão convertidos via _to_row().

 {}
Source code in src/observabilidade/base_logger.py 
74
75
76
77
78
79
80
81
82
83
84
def registrar(self, **kwargs: str | float | int | None) -> None:
    """Registra uma linha no CSV de forma thread-safe.

    Args:
        **kwargs: Dados a registrar (str, float, int ou None).
            Serão convertidos via _to_row().
    """
    row = self._to_row(**kwargs)
    with self._lock, open(self._csv_path, 'a', encoding='utf-8', newline='') as f:
        writer = csv.writer(f)
        writer.writerow(row)

src.observabilidade.registry

Registry central para loggers de observabilidade.

Usa dict interno em vez de variaveis de modulo — zero global statements. Mantem aliases retroativos para compatibilidade com codigo existente.

Example 
from src.observabilidade.registry import set_logger, get_logger

set_logger('observabilidade', ObservabilidadeLogger('logs/obs.csv'))
logger = get_logger('observabilidade')

LoggerRegistry()

Registry central para loggers — sem globals.

Usa dict interno para armazenar loggers por nome.

Source code in src/observabilidade/registry.py 
33
34
def __init__(self) -> None:
    self._loggers: dict[str, object] = {}

get(nome: str) -> object | None

Retorna logger por nome, ou None se nao configurado.

Source code in src/observabilidade/registry.py 
36
37
38
def get(self, nome: str) -> object | None:
    """Retorna logger por nome, ou None se nao configurado."""
    return self._loggers.get(nome)

set(nome: str, logger: object) -> None

Configura logger por nome.

Source code in src/observabilidade/registry.py 
40
41
42
def set(self, nome: str, logger: object) -> None:
    """Configura logger por nome."""
    self._loggers[nome] = logger

get_required(nome: str) -> object

Retorna logger ou levanta RuntimeError se nao configurado.

Source code in src/observabilidade/registry.py 
44
45
46
47
48
49
def get_required(self, nome: str) -> object:
    """Retorna logger ou levanta RuntimeError se nao configurado."""
    logger = self._loggers.get(nome)
    if logger is None:
        raise RuntimeError(f'Logger "{nome}" nao configurado. Use set_logger().')
    return logger

reset() -> None

Limpa todos os loggers (util para testes).

Source code in src/observabilidade/registry.py 
51
52
53
def reset(self) -> None:
    """Limpa todos os loggers (util para testes)."""
    self._loggers.clear()

get_logger(nome: str) -> object | None

Retorna logger por nome, ou None se nao configurado.

Source code in src/observabilidade/registry.py 
60
61
62
def get_logger(nome: str) -> object | None:
    """Retorna logger por nome, ou None se nao configurado."""
    return _registry.get(nome)

set_logger(nome: str, logger: object) -> None

Configura logger por nome.

Source code in src/observabilidade/registry.py 
65
66
67
def set_logger(nome: str, logger: object) -> None:
    """Configura logger por nome."""
    _registry.set(nome, logger)

get_required_logger(nome: str) -> object

Retorna logger ou levanta RuntimeError se nao configurado.

Source code in src/observabilidade/registry.py 
70
71
72
def get_required_logger(nome: str) -> object:
    """Retorna logger ou levanta RuntimeError se nao configurado."""
    return _registry.get_required(nome)

reset_loggers() -> None

Limpa todos os loggers (util para testes).

Source code in src/observabilidade/registry.py 
75
76
77
def reset_loggers() -> None:
    """Limpa todos os loggers (util para testes)."""
    _registry.reset()

get_obs_logger() -> ObservabilidadeLogger

Retorna o logger de observabilidade configurado.

Source code in src/observabilidade/registry.py 
83
84
85
def get_obs_logger() -> ObservabilidadeLogger:
    """Retorna o logger de observabilidade configurado."""
    return _registry.get_required('observabilidade')  # type: ignore[return-value]

set_obs_logger(logger: ObservabilidadeLogger) -> None

Configura o logger de observabilidade.

Source code in src/observabilidade/registry.py 
88
89
90
def set_obs_logger(logger: ObservabilidadeLogger) -> None:
    """Configura o logger de observabilidade."""
    _registry.set('observabilidade', logger)

get_clarificacao_logger() -> ClarificacaoLogger | None

Retorna o logger de clarificacao configurado.

Source code in src/observabilidade/registry.py 
93
94
95
def get_clarificacao_logger() -> ClarificacaoLogger | None:
    """Retorna o logger de clarificacao configurado."""
    return _registry.get('clarificacao')  # type: ignore[return-value]

set_clarificacao_logger(logger: ClarificacaoLogger | None) -> None

Configura o logger de clarificacao.

Source code in src/observabilidade/registry.py 
 98
 99
100
def set_clarificacao_logger(logger: ClarificacaoLogger | None) -> None:
    """Configura o logger de clarificacao."""
    _registry.set('clarificacao', logger)

get_extracao_logger() -> ExtracaoLogger | None

Retorna o logger de extracao configurado.

Source code in src/observabilidade/registry.py 
103
104
105
def get_extracao_logger() -> ExtracaoLogger | None:
    """Retorna o logger de extracao configurado."""
    return _registry.get('extracao')  # type: ignore[return-value]

set_extracao_logger(logger: ExtracaoLogger | None) -> None

Configura o logger de extracao.

Source code in src/observabilidade/registry.py 
108
109
110
def set_extracao_logger(logger: ExtracaoLogger | None) -> None:
    """Configura o logger de extracao."""
    _registry.set('extracao', logger)

get_handler_logger() -> HandlerLogger | None

Retorna o logger de handler configurado.

Source code in src/observabilidade/registry.py 
113
114
115
def get_handler_logger() -> HandlerLogger | None:
    """Retorna o logger de handler configurado."""
    return _registry.get('handler')  # type: ignore[return-value]

set_handler_logger(logger: HandlerLogger | None) -> None

Configura o logger de handler.

Source code in src/observabilidade/registry.py 
118
119
120
def set_handler_logger(logger: HandlerLogger | None) -> None:
    """Configura o logger de handler."""
    _registry.set('handler', logger)

get_funil_logger() -> FunilLogger | None

Retorna o logger de funil configurado.

Source code in src/observabilidade/registry.py 
123
124
125
def get_funil_logger() -> FunilLogger | None:
    """Retorna o logger de funil configurado."""
    return _registry.get('funil')  # type: ignore[return-value]

set_funil_logger(logger: FunilLogger | None) -> None

Configura o logger de funil.

Source code in src/observabilidade/registry.py 
128
129
130
def set_funil_logger(logger: FunilLogger | None) -> None:
    """Configura o logger de funil."""
    _registry.set('funil', logger)

src.observabilidade.logger

Logger de observabilidade para classificacao de intents.

Herda BaseCsvLogger — boilerplate CSV e thread safety sao herdados.

CAMINHOS_VALIDOS = frozenset({'lookup', 'rag_forte', 'llm_rag', 'llm_fixo', 'desconhecido'}) module-attribute

Caminhos validos para o parametro caminho.

HEADERS = ['timestamp', 'thread_id', 'mensagem', 'mensagem_norm', 'intent', 'confidence', 'caminho', 'top1_texto', 'top1_intencao', 'lookup', 'rag_top1', 'rag_sim', 'rag_intent', 'llm_raw', 'llm_intent'] module-attribute

Cabecalhos do CSV de eventos de classificacao.

ObservabilidadeLogger(csv_path: Path | str)

Bases: BaseCsvLogger

Logger thread-safe para registrar eventos de classificacao.

Source code in src/observabilidade/base_logger.py 
37
38
39
40
41
42
43
44
45
46
def __init__(self, csv_path: Path | str) -> None:
    """Inicializa o logger criando o arquivo CSV se necessario.

    Args:
        csv_path: Caminho para o arquivo CSV.
    """
    self._csv_path = Path(csv_path).resolve()
    self._csv_path.parent.mkdir(parents=True, exist_ok=True)
    self._lock = threading.Lock()
    self._inicializar_csv()

registrar(thread_id: str, mensagem: str, mensagem_norm: str, intent: str, confidence: float, caminho: str, top1_texto: str, top1_intencao: str, lookup: str = '', rag_top1: str = '', rag_sim: str = '', rag_intent: str = '', llm_raw: str = '', llm_intent: str = '') -> None

Registra um evento de classificacao no CSV.

Parameters:

Name Type Description Default
lookup str

Intencao encontrada por lookup (ou vazio).

 ''
rag_top1 str

Texto do top1 do RAG (ou vazio).

 ''
rag_sim str

Similaridade do top1 do RAG (ou vazio).

 ''
rag_intent str

Intencao do top1 do RAG (ou vazio).

 ''
llm_raw str

Resposta crua do LLM (ou vazio).

 ''
llm_intent str

Intencao extraida do LLM (ou vazio).

 ''

Raises:

Type Description
ValueError

Se caminho nao estiver em CAMINHOS_VALIDOS.
Source code in src/observabilidade/logger.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def registrar(
    self,
    thread_id: str,
    mensagem: str,
    mensagem_norm: str,
    intent: str,
    confidence: float,
    caminho: str,
    top1_texto: str,
    top1_intencao: str,
    lookup: str = '',
    rag_top1: str = '',
    rag_sim: str = '',
    rag_intent: str = '',
    llm_raw: str = '',
    llm_intent: str = '',
) -> None:
    """Registra um evento de classificacao no CSV.

    Args:
        lookup: Intencao encontrada por lookup (ou vazio).
        rag_top1: Texto do top1 do RAG (ou vazio).
        rag_sim: Similaridade do top1 do RAG (ou vazio).
        rag_intent: Intencao do top1 do RAG (ou vazio).
        llm_raw: Resposta crua do LLM (ou vazio).
        llm_intent: Intencao extraida do LLM (ou vazio).

    Raises:
        ValueError: Se ``caminho`` nao estiver em ``CAMINHOS_VALIDOS``.
    """
    super().registrar(
        thread_id=thread_id,
        mensagem=mensagem,
        mensagem_norm=mensagem_norm,
        intent=intent,
        confidence=confidence,
        caminho=caminho,
        top1_texto=top1_texto,
        top1_intencao=top1_intencao,
        lookup=lookup,
        rag_top1=rag_top1,
        rag_sim=rag_sim,
        rag_intent=rag_intent,
        llm_raw=llm_raw,
        llm_intent=llm_intent,
    )

src.observabilidade.funil_logger

Logger para progressao no funil de pedidos.

FunilLogger(csv_path: Path | str)

Bases: BaseCsvLogger

Logger thread-safe para registrar transicoes no funil.

Source code in src/observabilidade/base_logger.py 
37
38
39
40
41
42
43
44
45
46
def __init__(self, csv_path: Path | str) -> None:
    """Inicializa o logger criando o arquivo CSV se necessario.

    Args:
        csv_path: Caminho para o arquivo CSV.
    """
    self._csv_path = Path(csv_path).resolve()
    self._csv_path.parent.mkdir(parents=True, exist_ok=True)
    self._lock = threading.Lock()
    self._inicializar_csv()

src.observabilidade.handler_logger

Logger generico para execucao de handlers.

JSON_TRUNCATE_LIMIT = 200 module-attribute

Limite maximo de caracteres para campos JSON serializados.

HandlerLogger(csv_path: Path | str)

Bases: BaseCsvLogger

Logger thread-safe para registrar execucoes de handlers.

Source code in src/observabilidade/base_logger.py 
37
38
39
40
41
42
43
44
45
46
def __init__(self, csv_path: Path | str) -> None:
    """Inicializa o logger criando o arquivo CSV se necessario.

    Args:
        csv_path: Caminho para o arquivo CSV.
    """
    self._csv_path = Path(csv_path).resolve()
    self._csv_path.parent.mkdir(parents=True, exist_ok=True)
    self._lock = threading.Lock()
    self._inicializar_csv()

src.observabilidade.extracao_logger

Logger para eventos de extracao de itens do cardapio.

ExtracaoLogger(csv_path: Path | str)

Bases: BaseCsvLogger

Logger thread-safe para registrar eventos de extracao de itens.

Source code in src/observabilidade/base_logger.py 
37
38
39
40
41
42
43
44
45
46
def __init__(self, csv_path: Path | str) -> None:
    """Inicializa o logger criando o arquivo CSV se necessario.

    Args:
        csv_path: Caminho para o arquivo CSV.
    """
    self._csv_path = Path(csv_path).resolve()
    self._csv_path.parent.mkdir(parents=True, exist_ok=True)
    self._lock = threading.Lock()
    self._inicializar_csv()

src.observabilidade.clarificacao_logger

Logger para eventos de clarificacao de variantes.

ClarificacaoLogger(csv_path: Path | str)

Bases: BaseCsvLogger

Logger thread-safe para registrar eventos de clarificacao.

Source code in src/observabilidade/base_logger.py 
37
38
39
40
41
42
43
44
45
46
def __init__(self, csv_path: Path | str) -> None:
    """Inicializa o logger criando o arquivo CSV se necessario.

    Args:
        csv_path: Caminho para o arquivo CSV.
    """
    self._csv_path = Path(csv_path).resolve()
    self._csv_path.parent.mkdir(parents=True, exist_ok=True)
    self._lock = threading.Lock()
    self._inicializar_csv()

src.observabilidade.consultas

Consultas DuckDB para analise de eventos de classificacao.

Paths de arquivo sao interpolados com f-string (controlados internamente). Valores de usuario (thread_id, limit) usam parametros seguros ?.

Example 
from src.observabilidade.consultas import baixa_confianca, distribuicao_caminhos

casos = baixa_confianca('logs/classificacoes.csv', limit=10)
dist = distribuicao_caminhos('logs/classificacoes.csv')

baixa_confianca(csv_path: str, limit: int = 20) -> list[dict]

Retorna casos de baixa confianca que foram roteados para o LLM.

Source code in src/observabilidade/consultas.py 
58
59
60
61
62
63
64
65
66
67
68
def baixa_confianca(csv_path: str, limit: int = 20) -> list[dict]:
    """Retorna casos de baixa confianca que foram roteados para o LLM."""
    path = _sanitizar_path(csv_path)
    query = f"""
        SELECT mensagem, intent, confidence, caminho
        FROM '{path}'
        WHERE caminho IN ('llm_rag', 'llm_fixo')
        ORDER BY confidence ASC
        LIMIT ?
    """
    return _executar_query(query, [limit])

distribuicao_caminhos(csv_path: str) -> list[dict]

Retorna a distribuicao de eventos por caminho de classificacao.

Source code in src/observabilidade/consultas.py 
71
72
73
74
75
76
77
78
79
80
def distribuicao_caminhos(csv_path: str) -> list[dict]:
    """Retorna a distribuicao de eventos por caminho de classificacao."""
    path = _sanitizar_path(csv_path)
    query = f"""
        SELECT caminho, COUNT(*) as total
        FROM '{path}'
        GROUP BY caminho
        ORDER BY total DESC
    """
    return _executar_query(query)

extracoes_sem_itens(csv_extracao: str, limit: int = 20) -> list[dict]

Retorna mensagens onde extrator nao encontrou itens.

Source code in src/observabilidade/consultas.py 
83
84
85
86
87
88
89
90
91
92
93
def extracoes_sem_itens(csv_extracao: str, limit: int = 20) -> list[dict]:
    """Retorna mensagens onde extrator nao encontrou itens."""
    path = _sanitizar_path(csv_extracao)
    query = f"""
        SELECT mensagem, itens_encontrados, tempo_ms
        FROM '{path}'
        WHERE itens_encontrados = 0
        ORDER BY timestamp DESC
        LIMIT ?
    """
    return _executar_query(query, [limit])

funil_com_abandono(csv_funil: str, thread_id: str | None = None) -> list[dict]

Analisa sessoes que pararam em etapas intermediarias.

Source code in src/observabilidade/consultas.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def funil_com_abandono(csv_funil: str, thread_id: str | None = None) -> list[dict]:
    """Analisa sessoes que pararam em etapas intermediarias."""
    path = _sanitizar_path(csv_funil)
    if thread_id:
        query = f"""
            SELECT thread_id, etapa_atual, intent, carrinho_size, timestamp
            FROM '{path}'
            WHERE thread_id = ?
            ORDER BY timestamp DESC
            LIMIT 50
        """
        return _executar_query(query, [thread_id])
    query = f"""
        SELECT thread_id, etapa_atual, intent, carrinho_size, timestamp
        FROM '{path}'
        ORDER BY timestamp DESC
        LIMIT 50
    """
    return _executar_query(query)

handlers_com_erro(csv_handler: str, limit: int = 20) -> list[dict]

Retorna execucoes de handlers com erro.

Source code in src/observabilidade/consultas.py 
117
118
119
120
121
122
123
124
125
126
127
def handlers_com_erro(csv_handler: str, limit: int = 20) -> list[dict]:
    """Retorna execucoes de handlers com erro."""
    path = _sanitizar_path(csv_handler)
    query = f"""
        SELECT handler, intent, input_resumo, erro, tempo_ms
        FROM '{path}'
        WHERE erro != ''
        ORDER BY timestamp DESC
        LIMIT ?
    """
    return _executar_query(query, [limit])

tempo_medio_handlers(csv_handler: str) -> list[dict]

Retorna tempo medio por handler.

Source code in src/observabilidade/consultas.py 
130
131
132
133
134
135
136
137
138
139
def tempo_medio_handlers(csv_handler: str) -> list[dict]:
    """Retorna tempo medio por handler."""
    path = _sanitizar_path(csv_handler)
    query = f"""
        SELECT handler, AVG(tempo_ms) as tempo_medio_ms, COUNT(*) as total_execucoes
        FROM '{path}'
        GROUP BY handler
        ORDER BY tempo_medio_ms DESC
    """
    return _executar_query(query)

src.observabilidade.debug_cli

CLI de debug para analisar logs do Pede AI.

ultima_sessao(thread_id: str | None = None) -> None

Mostra a linha do tempo da ultima sessao ou de uma especifica.

Source code in src/observabilidade/debug_cli.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@app.command()
def ultima_sessao(thread_id: str | None = None) -> None:
    """Mostra a linha do tempo da ultima sessao ou de uma especifica."""
    if not LOG_DIR.exists():
        console.print('[red]Diretorio logs/ nao encontrado[/red]')
        raise typer.Exit(1)

    funil_csv = LOG_DIR / 'funil.csv'
    if not funil_csv.exists():
        console.print('[yellow]Nenhum log de funil encontrado[/yellow]')
        return

    if thread_id:
        query = f"SELECT * FROM '{funil_csv}' WHERE thread_id = ? ORDER BY timestamp"  # noqa: S608 — path controlado internamente
        cols, rows = _ler_csv_duckdb(funil_csv, query, [thread_id])
    else:
        query = f"SELECT * FROM '{funil_csv}' ORDER BY timestamp DESC LIMIT 20"  # noqa: S608 — path controlado internamente
        cols, rows = _ler_csv_duckdb(funil_csv, query)

    table = Table(title='Funil de Pedidos')
    for col in cols:
        table.add_column(col)
    for row in rows:
        table.add_row(*[str(v) for v in row])
    console.print(table)

extracoes_falhas() -> None

Mostra mensagens onde nenhum item foi extraido.

Source code in src/observabilidade/debug_cli.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
@app.command()
def extracoes_falhas() -> None:
    """Mostra mensagens onde nenhum item foi extraido."""
    extracao_csv = LOG_DIR / 'extracoes.csv'
    if not extracao_csv.exists():
        console.print('[yellow]Nenhum log de extracao encontrado[/yellow]')
        return

    query = f"SELECT mensagem, tempo_ms FROM '{extracao_csv}' WHERE itens_encontrados = 0 ORDER BY timestamp DESC LIMIT 20"  # noqa: S608
    _, rows = _ler_csv_duckdb(extracao_csv, query)

    table = Table(title='Extracoes sem Resultados')
    table.add_column('Mensagem')
    table.add_column('Tempo (ms)')
    for row in rows:
        table.add_row(row[0], f'{row[1]:.2f}')
    console.print(table)

erros_handlers() -> None

Mostra erros em handlers.

Source code in src/observabilidade/debug_cli.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
@app.command()
def erros_handlers() -> None:
    """Mostra erros em handlers."""
    handler_csv = LOG_DIR / 'handlers.csv'
    if not handler_csv.exists():
        console.print('[yellow]Nenhum log de handler encontrado[/yellow]')
        return

    query = f"SELECT handler, intent, erro, tempo_ms FROM '{handler_csv}' WHERE erro != '' ORDER BY timestamp DESC LIMIT 20"  # noqa: S608
    _, rows = _ler_csv_duckdb(handler_csv, query)

    table = Table(title='Erros em Handlers')
    table.add_column('Handler')
    table.add_column('Intent')
    table.add_column('Erro')
    table.add_column('Tempo (ms)')
    for row in rows:
        table.add_row(row[0], row[1], row[2], f'{row[3]:.2f}')
    console.print(table)

classificar(mensagem: str) -> None

Testa classificacao de uma mensagem sem executar o grafo.

Nota: requer classificador configurado via main.py. Use 'python main.py' para teste interativo completo.

Source code in src/observabilidade/debug_cli.py 
 99
100
101
102
103
104
105
106
107
108
109
@app.command()
def classificar(mensagem: str) -> None:
    """Testa classificacao de uma mensagem sem executar o grafo.

    Nota: requer classificador configurado via main.py.
    Use 'python main.py' para teste interativo completo.
    """
    console.print('[yellow]Classificacao CLI desabilitada na refatoracao.[/yellow]')
    console.print(
        '[dim]Use o main.py para teste interativo com Groq + Transformers.[/dim]'
    )

extrair_teste(mensagem: str) -> None

Testa extracao de itens de uma mensagem.

Source code in src/observabilidade/debug_cli.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
@app.command()
def extrair_teste(mensagem: str) -> None:
    """Testa extracao de itens de uma mensagem."""
    itens = extrair(mensagem)
    if not itens:
        console.print(f'[red]Nenhum item extraido de "{mensagem}"[/red]')
        return

    table = Table(title=f'Extracao: "{mensagem}"')
    table.add_column('item_id')
    table.add_column('quantidade')
    table.add_column('variante')
    table.add_column('remocoes')
    for item in itens:
        table.add_row(
            item['item_id'],
            str(item['quantidade']),
            str(item.get('variante')),
            ', '.join(item.get('remocoes', [])),
        )
    console.print(table)