Pular para o conteúdo
Triadeflow Docs

Convenções de Código e Padrões

Estas são as convenções aplicadas em todos os repositórios da Triadeflow. Claude Code deve seguir todas estas regras automaticamente, sem precisar ser lembrado.

Estrutura geral

Linguagens

  • Python 3.11+ para todo backend, agentes, scripts
  • TypeScript para frontends (admin panel, ops panel, dashboards de cliente)
  • YAML para configurações de tenant
  • Markdown para documentação

Naming

  • Arquivos Python: snake_case.py
  • Classes: PascalCase
  • Funções e variáveis: snake_case
  • Constantes: UPPER_SNAKE_CASE
  • Pastas: kebab-case (ex: agent-template, triadeflow-core)
  • Branches Git: feat/nome-descritivo, fix/..., refactor/..., docs/...

Idioma

  • Código, comentários, docstrings: inglês
  • Documentação (.md), prompts de agente, mensagens ao cliente: português brasileiro
  • Logs: inglês para operacionais, português para mensagens visíveis ao cliente final
  • Commit messages: inglês, padrão Conventional Commits

Tipagem

Tipagem é obrigatória. Não há “Python dinâmico” nesta base.

# CORRETO
def process_message(tenant_id: str, message: IncomingMessage) -> AgentResponse:
    ...


# ERRADO - sem tipos
def process_message(tenant_id, message):
    ...

Pydantic v2 para schemas

Todo dado que cruza fronteira (HTTP, fila, config, LLM input/output) é Pydantic.

from pydantic import BaseModel, Field


class TenantConfig(BaseModel):
    tenant_id: str = Field(..., min_length=3, max_length=64)
    agent_type: Literal["sdr", "vendas", "suporte", "agendamento"]
    modelo: ModelConfig
    messaging: MessagingConfig
    rag: RAGConfig

Type aliases para clareza

# Em types.py do core
TenantId = str
ConversationId = str
AgentType = Literal["sdr", "vendas", "suporte", "agendamento", "recuperacao", "cobranca", "pos-venda"]

Estrutura de pastas (padrão de cada agente)

agent-{tipo}/
├── pyproject.toml
├── Dockerfile
├── docker-compose.yml
├── Makefile
├── .env.example
├── README.md
├── CLAUDE.md                    # aponta para triadeflow-docs
├── src/
│   ├── main.py                  # FastAPI entrypoint
│   ├── agent.py                 # LangGraph workflow
│   ├── state.py                 # Pydantic state
│   ├── settings.py              # Settings via env
│   ├── prompts/
│   │   ├── system.md            # template Jinja2
│   │   └── fragments/           # tom, regras, lgpd, handoff
│   ├── tools/
│   │   ├── _registry.py         # registro de tools
│   │   ├── consultar_kb.py
│   │   ├── handoff_humano.py
│   │   └── ...
│   ├── flows/                   # LangGraph nodes
│   │   ├── classify.py
│   │   ├── qualify.py
│   │   └── handoff.py
│   ├── handlers/
│   │   └── webhook.py
│   └── utils/
├── tenants/
│   ├── _defaults.yaml           # padrão para todos os tenants
│   └── {tenant_id}/
│       ├── config.yaml
│       └── kb/                  # base de conhecimento (vetorizada)
├── eval/
│   ├── datasets/
│   │   └── golden.json
│   └── runners/
├── scripts/
│   ├── create_tenant.py
│   ├── ingest_kb.py
│   └── deploy.sh
├── tests/
└── deploy/
    └── hetzner.yml

Convenções por arquivo

main.py

Entry point FastAPI. Apenas:

  • Cria app
  • Configura middlewares (Langfuse, CORS, error handler)
  • Registra rotas (/webhook, /health, /admin)
  • Nunca tem lógica de negócio

agent.py

Define o LangGraph workflow. Apenas:

  • Constrói o grafo
  • Adiciona nodes (cada um em arquivo separado em flows/)
  • Define edges e conditional_edges
  • Compila o grafo

state.py

AgentState Pydantic. Imutável-ish (use .copy(update={...})).

class AgentState(BaseModel):
    tenant_id: str
    conversation_id: str
    messages: list[Message]
    current_step: str
    extracted_data: dict[str, Any]
    metadata: dict[str, Any]

settings.py

Settings via Pydantic Settings + env vars. Nunca leia os.environ direto.

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    anthropic_api_key: SecretStr
    redis_url: str
    qdrant_url: str
    langfuse_public_key: SecretStr
    langfuse_secret_key: SecretStr


    class Config:
        env_file = ".env"

Tools

Cada tool em arquivo próprio, decorada com @tool (LangChain) e @trace_tool (Langfuse):

tools/agendar_reuniao.py
from langchain_core.tools import tool
from triadeflow_core.observability import trace_tool


@tool
@trace_tool
def agendar_reuniao(
    tenant_id: str,
    lead_id: str,
    data_hora: str,
    duracao_min: int = 30,
) -> AgendamentoResultado:
    """
    Agenda uma reunião no calendário do tenant.


    Args:
        tenant_id: ID do tenant
        lead_id: ID do lead no CRM
        data_hora: ISO 8601 com timezone (ex: "2026-05-15T14:00:00-03:00")
        duracao_min: Duração em minutos


    Returns:
        AgendamentoResultado com sucesso, link e ID do evento
    """
    # implementação

Princípios de tool:

  • Uma responsabilidade
  • Idempotente quando possível
  • Tipada (input e output)
  • Logada (Langfuse via decorator)
  • Tenant-aware (sempre recebe tenant_id)
  • Erro tratado (retorna erro estruturado, nunca raise para o agente)

Async first

Toda I/O é async. Nunca requests.get() — sempre httpx.AsyncClient.

import httpx


async def fetch_lead(crm_url: str, lead_id: str) -> Lead:
    async with httpx.AsyncClient(timeout=10.0) as client:
        resp = await client.get(f"{crm_url}/leads/{lead_id}")
        resp.raise_for_status()
        return Lead.model_validate(resp.json())

Exceção: scripts CLI utilitários podem ser síncronos.

Tratamento de erro

Hierarquia de exceções

# Em triadeflow_core/exceptions.py
class TriadeflowError(Exception):
    """Base de todas exceções da plataforma"""


class ProviderError(TriadeflowError):
    """Falha em provider de mensageria"""


class CRMError(TriadeflowError):
    """Falha em integração com CRM"""


class TenantConfigError(TriadeflowError):
    """Configuração de tenant inválida ou ausente"""


class ToolExecutionError(TriadeflowError):
    """Falha na execução de uma tool"""

Retry com backoff

Para integrações externas, sempre use retry com tenacity:

from tenacity import retry, stop_after_attempt, wait_exponential


@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    retry=retry_if_exception_type(httpx.HTTPError),
)
async def call_external_api(...):
    ...

Circuit breaker para integrações instáveis

Use circuitbreaker para CRMs e gateways. Após N falhas, parar de tentar por X minutos.

Logging

structlog em todos os módulos. Nunca print().

import structlog
log = structlog.get_logger()


log.info("message_received", tenant_id=tid, conversation_id=cid)
log.warning("rag_low_score", tenant_id=tid, score=0.45, query=query[:50])
log.error("crm_failure", tenant_id=tid, error=str(e), exc_info=True)

Sempre inclua tenant_id e conversation_id quando disponíveis.

Configuração de tenant

Cada tenant tem tenants/{tenant_id}/config.yaml que segue exatamente o schema Pydantic TenantConfig. Validar na inicialização e não em runtime.

tenants/usa-salus/config.yaml
tenant_id: usa-salus
agent_type: sdr


agente:
  nome: "Assistente USA Salus"
  empresa: "USA Salus"
  contexto_negocio: |
    Empresa de saúde corporativa que oferece...


modelo:
  principal: claude-sonnet-4-20250514
  fallback: claude-haiku-4-5-20251001
  estruturadas: claude-haiku-4-5-20251001  # para tarefas simples
  temperatura: 0.7
  max_tokens: 1024


messaging:
  provider: evolution
  instance_name: usa-salus-prod
  webhook_secret: ${USA_SALUS_WEBHOOK_SECRET}


crm:
  type: kommo
  base_url: https://usa.kommo.com
  api_token: ${USA_SALUS_KOMMO_TOKEN}
  pipeline_id: 12345


rag:
  collection: usa-salus
  top_k: 5
  rerank: false


memory:
  redis_prefix: usa-salus
  short_term_size: 10
  summarize_after: 10


# ... outras seções

Variáveis de ambiente dentro de YAML são resolvidas pelo env_resolver do core na hora de carregar.

Prompt engineering

System prompts são arquivos .md com Jinja2:

{# prompts/system.md #}


# Identidade


Você é {{ agente.nome }}, assistente virtual da {{ agente.empresa }}.


# Sua empresa


{{ agente.contexto_negocio }}


# Seu objetivo


{{ agente.objetivo_principal }}


# Como você atua


{% include "fragments/tom.md" %}


# Suas ferramentas


{% include "fragments/tools.md" %}


# Regras inegociáveis


{% include "fragments/regras.md" %}
{% include "fragments/lgpd.md" %}


# Handoff humano


{% include "fragments/handoff.md" %}

Fragments reutilizáveis ficam em prompts/fragments/ e são compartilhados entre tenants do mesmo tipo de agente.

Prompt caching da Anthropic: marcar a parte fixa do system prompt como cache_control: ephemeral quando o prompt for > 1024 tokens.

Modelo selection (model router)

Use o model_router do core para escolher modelo dinamicamente baseado no tipo de tarefa:

from triadeflow_core.llm import model_router


# Conversa principal
response = await model_router.invoke(
    messages=msgs,
    task_type="conversation",  # → Sonnet
    tenant_id=tid,
)


# Sub-tarefa estruturada
classification = await model_router.invoke(
    messages=msgs,
    task_type="classification",  # → Haiku
    tenant_id=tid,
)

Tipos de tarefa: classification, extraction, simple_response, conversation, creative, analysis.

Testes

Estrutura

tests/
├── unit/                # módulos isolados
├── integration/         # com Redis, Postgres reais
├── e2e/                 # webhook completo
└── eval/                # contra dataset de qualidade

Pytest com markers

@pytest.mark.unit
def test_classify_intent():
    ...


@pytest.mark.integration
async def test_redis_memory():
    ...

Eval automatizado

Cada agente tem eval/datasets/golden.json com 50-100 casos rotulados. Rodar antes de promover qualquer mudança de prompt ou modelo.

Observabilidade

Toda chamada LLM é traceada

from triadeflow_core.observability import langfuse_handler


response = await llm.ainvoke(
    messages,
    config={
        "callbacks": [langfuse_handler],
        "metadata": {
            "tenant_id": tid,
            "agent_type": "sdr",
            "conversation_id": cid,
        }
    }
)

Tags semânticas obrigatórias

Toda chamada LLM no Langfuse precisa de:

  • tenant_id — quem é o cliente final
  • agent_type — qual tipo de agente (sdr, vendas, etc)
  • conversation_id — para correlacionar mensagens
  • flow_step — em qual passo do grafo está (classify, qualify, etc)

Annotation queue

Traces com score baixo do LLM-as-judge entram automaticamente na annotation queue. Revisar semanalmente como parte da rotina operacional.

Custos

Antes de mexer em modelo, simule

Não troque modelo de produção sem antes rodar a simulação de custo. Ver .claude/workflows/otimizar-custo.md.

Tática híbrida é padrão

Configure cada agente para usar Haiku em sub-tarefas estruturadas (classificação, extração, confirmação) e Sonnet apenas na conversa principal. Economia típica: 30-50%.

Prompt caching

System prompts longos (> 1024 tokens) devem usar prompt caching da Anthropic. Cache hit reduz custo da parte cacheada em 90%.

Segurança

Secrets

  • Nunca comite secrets. Use .env (gitignored) e ${VAR} no YAML
  • Em produção, use Hetzner Vault ou Doppler
  • API keys têm escopo mínimo necessário

LGPD

Todo agente tem fragmento prompts/fragments/lgpd.md que define:

  • Consentimento explícito antes de coletar PII
  • Direito ao esquecimento implementado
  • Logs de mensagens com PII têm retenção limitada (90 dias máximo)
  • Em saúde, tutela explícita do art. 11 LGPD

Webhooks

  • Sempre validar HMAC signature
  • Whitelist de IPs quando possível
  • Rate limiting por IP

Git e PRs

Conventional Commits

feat: adiciona tool agendar_reuniao no agent-sdr
fix: corrige timezone na conversão de Unix timestamp
refactor(core): extrai model_router para módulo separado
docs(agents/sdr): atualiza spec com BANT customizado
perf(rag): reduz top_k padrão de 8 para 5
test(eval): adiciona 20 casos golden para agent-suporte

PRs

  • Título no padrão de commit
  • Descrição com: o que muda, por quê, como testou
  • Screenshots se afetar UI
  • Checklist de impacto: arquitetura, custo, breaking change, docs

Branches

  • main — produção, protegida
  • feat/..., fix/... — desenvolvimento
  • release/v1.2.0 — preparação de release

Erros antipattern (não fazer)

from langchain import *

Sempre import explícito.

time.sleep() em código async

Use await asyncio.sleep().

Configuração hardcoded por cliente

Tudo de cliente vai em YAML, nunca em código.

Múltiplas LLM calls quando uma resolve

Otimize para latência e custo. Combine quando possível.

Catch genérico de Exception

# ERRADO
try: ... except Exception: pass


# CORRETO
try: ...
except (httpx.TimeoutException, httpx.HTTPStatusError) as e:
    log.warning("crm_timeout", error=str(e))
    raise CRMError(...)

Strings de SQL/comando construídas com f-string

Sempre use parameterização ou ORMs.

Mock de LLM em testes via patch global

Use o mock_llm fixture do core que tem dataset estruturado.

Convenção de logs operacionais

Padrão para mensagens visíveis ao cliente final (em português):

  • Saudação: tom natural, sem “Olá! Como posso ajudar?” automático
  • Confirmação: “Pronto, registrei aqui” — natural, não robotizado
  • Erro recuperável: “Tive um problema aqui, pode repetir o que você queria?”
  • Erro grave: “Vou chamar um humano pra te atender” + tool de handoff
  • Despedida: tom natural, sem clichês

Versão: 1.0 Aplica-se a: todos os repositórios da Triadeflow Última atualização: 2026-05-03