Files

2026-05-14 15:30:38 -03:00

6.2 KiB

Raw Permalink Blame History

Observabilidade com Langfuse

O que é o Langfuse

O Langfuse é uma plataforma de observabilidade para aplicações LLM. No doc-processor ele é usado para rastrear cada chamada ao agente de IA, permitindo:

Visualizar o histórico completo de mensagens de cada execução (System, Human, AI, Tool calls).
Monitorar consumo de tokens (entrada e saída) por execução e por procedimento.
Medir latência das chamadas ao modelo.
Inspecionar as respostas brutas do agente para depuração.
Auditar decisões do sistema ao longo do tempo.

Como está integrado no código

Inicialização (`utils/langgraph_agent.py`)

O cliente Langfuse é criado uma única vez na importação do módulo, usando as credenciais do Secrets Manager:

from langfuse import Langfuse
from langfuse.langchain import CallbackHandler

langfuse = Langfuse(
    secret_key=SECRETS["LANGFUSE-SECRET-KEY"],
    public_key=SECRETS["LANGFUSE-PUBLIC-KEY"],
    host=LANGFUSE_HOST,  # padrão: https://cloud.langfuse.com
)

Rastreamento de execuções (`run_agent`)

Para cada chamada ao agente, um CallbackHandler do Langfuse é instanciado e passado como callback ao LangGraph. Isso faz com que todas as mensagens trocadas durante a execução sejam enviadas automaticamente ao Langfuse:

langfuse_handler = CallbackHandler()
config = {"callbacks": [langfuse_handler]}
final_state = await agent.ainvoke(initial_state, config=config)

O CallbackHandler captura automaticamente:

O conteúdo do System Prompt e da mensagem Human enviados ao modelo.
Cada resposta do LLM (incluindo tool calls intermediárias).
As respostas das ferramentas (ToolMessage).
Tokens usados em cada step.

Flush assíncrono

Após cada execução do agente, o buffer do Langfuse é enviado ao servidor de forma assíncrona para garantir que nenhum trace seja perdido mesmo que o processo seja encerrado logo após:

await asyncio.to_thread(langfuse.flush)

Por que to_thread? O método flush() do Langfuse é bloqueante. Executá-lo em uma thread separada evita que ele bloqueie o event loop do FastAPI enquanto aguarda a resposta da rede.

Contagem de Tokens

Além do rastreamento automático pelo Langfuse, o código também acumula os tokens manualmente a partir dos metadados das mensagens AI para retorná-los na resposta da API:

input_tokens = 0
output_tokens = 0
for msg in final_state["messages"]:
    usage = getattr(msg, "usage_metadata", None)
    if usage:
        input_tokens += usage.get("input_tokens", 0)
        output_tokens += usage.get("output_tokens", 0)

Os tokens são somados de todas as mensagens AI geradas durante a execução, incluindo chamadas intermediárias quando o agente usa a ferramenta check. Esses valores são retornados em cada avaliacaoAgente:

{
  "codigoServico": "40808130",
  "resultado": "Aprovado",
  "input_tokens": 1523,
  "output_tokens": 87
}

Criando uma Organização e um Projeto no Langfuse

1. Criar conta e organização

Acesse o ip do Ec2 via http na porta 3000, após liberar no security group para o ip desejado, e clique em Sign Up.
Após o login, o Langfuse pede para criar ou entrar em uma organização. A organização agrupa todos os projetos e membros da equipe.
- Clique em Create new organization, preencha o nome e confirme.

2. Criar um projeto

Dentro da organização:

Crie um novo projeto.
Nos passos iniciais tem um botão para criar as Api keys, copie os valores de Secret e Public.

3. Apontar o sistema para o novo projeto

Atualize os segredos no AWS Secrets Manager (doc-analyzer) com os novos valores:

Basta ir no Secrets manager, escolher o segredo doc-analyzer, ir em retrieve secret value e edit. Atualize com os novos valores.

Após atualizar, talvez seja nescessário reiniciar o container/serviço para que ele carregue as novas credenciais:

# ECS — forçar novo deploy
aws ecs update-service \
  --cluster <nome-do-cluster> \
  --service <nome-do-servico> \
  --force-new-deployment

Nas próximas execuções, os traces aparecerão no novo projeto dentro da nova organização.

Configuração

Variável de Ambiente

Variável	Padrão	Descrição
`LANGFUSE_HOST`	`http://13.59.214.47:3000/`	Endereço do servidor Langfuse. Pode ser uma instância self-hosted.

Segredos (AWS Secrets Manager — `doc-analyzer`)

Chave	Descrição
`LANGFUSE-SECRET-KEY`	Chave secreta do projeto Langfuse
`LANGFUSE-PUBLIC-KEY`	Chave pública do projeto Langfuse

As chaves são obtidas no painel do Langfuse em Settings → API Keys.

Self-hosted

Para usar uma instância própria do Langfuse (ex: rodando internamente), basta configurar (variável de ambiente do ecs):

LANGFUSE_HOST=http://13.59.214.47:3000/

E atualizar as chaves correspondentes no Secrets Manager.

O que visualizar no painel

Cada chamada ao POST /process gera uma ou mais traces no Langfuse (uma por serviço avaliado). Em cada trace é possível ver:

Input: JSON da guia + conteúdo OCR dos documentos enviados ao agente.
System Prompt: Critérios de auto-aprovação e documentação mínima do procedimento avaliado.
Steps intermediários: Se o agente chamou a ferramenta check, aparece como um step separado com o conteúdo retornado.
Output: JSON final com status, criterio e lista de documentos.
Tokens e latência: Custo estimado e tempo de resposta do modelo.

Depuração

Se as traces não aparecerem no painel:

Verifique se LANGFUSE-SECRET-KEY e LANGFUSE-PUBLIC-KEY estão corretos no Secrets Manager.
Confirme que LANGFUSE_HOST aponta para o servidor correto.
Verifique os logs do container — erros de conexão com o Langfuse são silenciosos por padrão (o flush falha sem derrubar o serviço).
Confirme que o container tem acesso de rede ao host do Langfuse.

6.2 KiB Raw Permalink Blame History