Observabilidade e Depuração
Logs em produção, telemetria, IDs de correlação, rastreamento de ferramentas e estratégias de inspeção em sistemas chatflow-agent.
Operar enxames multiagente em ambientes produtivos exige visibilidade abrangente. Quando um agente transfere uma conversa, invoca APIs externas ou aciona fallbacks, os operadores necessitam de logs estruturados e IDs de correlação para diagnosticar latência, investigar anomalias e auditar decisões tomadas.
O chatflow-agent disponibiliza loggers estruturados baseados na biblioteca padrão do Python em todas as camadas da arquitetura, desde webhooks de canais até requisições de rede dos mecanismos.
Hierarquia de Loggers
Todos os loggers estão organizados sob o namespace chatflow_agent. Você pode definir a verbosidade globalmente ou ajustar componentes de forma individual:
chatflow_agent
├── chatflow_agent.runner (Execução de turnos, handoffs, chamadas de ferramentas, checkpoints)
├── chatflow_agent.tools (Geração de esquemas, execução de ferramentas, validação de argumentos)
├── chatflow_agent.memory (Criação de sessões, E/S do SQLite, TTL de deduplicação)
├── chatflow_agent.engine
│ ├── chatflow_agent.engine.openai (Cargas úteis e latência de OpenAI/Groq/DeepSeek)
│ ├── chatflow_agent.engine.gemini (Requisições REST v1beta, turnos do modelo)
│ └── chatflow_agent.engine.anthropic (Cabeçalhos da Messages API, contagem de tokens)
├── chatflow_agent.telegram (Polling de atualizações, autorização de administradores, divisão de mensagens)
└── chatflow_agent.whatsapp (Validação HMAC, deduplicação, mascaramento de telefones)Configurando Logs Estruturados
Em produção, formate os logs como registros JSON de linha única para que sistemas centrais de ingestão (como Datadog, CloudWatch ou Google Cloud Logging) indexem os campos automaticamente:
import json
import logging
import sys
from chatflow_agent import Agent, Runner
class JSONLogFormatter(logging.Formatter):
"""Format log records as structured JSON."""
def format(self, record: logging.LogRecord) -> str:
log_entry = {
"timestamp": self.formatTime(record, self.datefmt),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
"module": record.module,
"lineno": record.lineno,
}
if record.exc_info:
log_entry["exception"] = self.formatException(record.exc_info)
return json.dumps(log_entry)
def setup_production_logging(log_level: int = logging.INFO) -> None:
"""Initialize structured JSON logging for chatflow-agent."""
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JSONLogFormatter())
# Configurar logger raiz
root_logger = logging.getLogger("chatflow_agent")
root_logger.setLevel(log_level)
root_logger.handlers = [handler]
root_logger.propagate = False
if __name__ == "__main__":
setup_production_logging()
agent = Agent(name="ObservabilityBot", instructions="Test logging.")
runner = Runner(starting_agent=agent)
print("Structured logging configured.")Correlação de Traces com trace_id
Quando uma ferramenta falha durante a execução, o Runner gera automaticamente um UUID trace_id e o anexa tanto ao registro interno de log quanto à mensagem enviada ao agente:
[ERROR] 2026-10-30T18:22:15Z chatflow_agent.runner: Tool execution failed [trace_id: 3c9b7e41-0f2d-4e92-bc12-f71e549d8802] - Tool: 'query_database' on agent 'BillingAgent': Connection timed out to db.internal.corp:5432Por Que a Correlação Importa
- Privacidade do Usuário: O usuário final não visualiza stack traces brutos nem endereços IP internos.
- Triagem Determinística: Quando um usuário relata "Minha consulta falhou às 18:22", os engenheiros de suporte podem pesquisar diretamente por
trace_id: 3c9b7e41...nos logs para analisar o stack trace completo e os parâmetros de entrada.
Inspecionando a Telemetria de Resposta
Toda invocação de Runner.run() ou Runner.run_async() retorna um objeto AgentResponse contendo metadados de execução:
from chatflow_agent import Agent, Runner
agent = Agent(name="Inspector", instructions="Provide analytical responses.")
runner = Runner(starting_agent=agent)
response = runner.run(
session_id="session-user-99",
user_message="Calculate quarterly forecast.",
)
# 1. Texto entregue ao usuário
print(f"Content: {response.content}")
# 2. Agente ativo final que formulou a resposta
print(f"Active Agent: {response.active_agent_name}")
# 3. Lista de nomes de ferramentas executadas durante o turno
print(f"Tools Executed: {response.tool_calls_executed}")
# 4. Detalhes do handoff se uma transferência ocorreu
if response.handoff:
print(f"Transferred to: {response.handoff.target_agent_name}")
print(f"Reason: {response.handoff.reason}")Auditoria do Histórico da Conversa
Para inspecionar os turnos de conversa gravados em uma sessão, recupere o contexto da sessão diretamente do armazenamento configurado:
from chatflow_agent import Agent, Runner
from chatflow_agent.core.memory import SessionStore
store = SessionStore()
agent = Agent(name="Support", instructions="Help customers.")
runner = Runner(starting_agent=agent, session_store=store)
# Executar turno
runner.run(session_id="sess-42", user_message="Hello, I need assistance.")
# Inspecionar sessão armazenada
session = store.get(session_id="sess-42")
if session:
print(f"Total Messages: {len(session.history)}")
for msg in session.history:
print(f"[{msg.role.value}] {msg.content}")Zero Registro de Chaves de API
Os loggers do chatflow-agent ocultam automaticamente tokens de API e cabeçalhos de autorização. Cabeçalhos de autorização em logs HTTP aparecem como Bearer sk-**** ou são omitidos das query strings, garantindo conformidade com padrões de auditoria SOC 2 e LGPD.
Configuração e Ambiente
Gerencie variáveis de ambiente, predefinições de provedores, timeouts e configuração de aplicações Twelve-Factor em implantações do chatflow-agent.
Melhores Práticas de Segurança
Diretrizes de segurança em profundidade para o chatflow-agent, isolamento de credenciais, mitigação de injeção de prompt, assinaturas de webhooks e privacidade de dados.