FinanceAI

Multi-Agent Financial Analysis Platform — Sprint 7: Testing, Quality & Robustness

FinanceAI is a modular, AI-powered financial analysis platform built on FastAPI, LangGraph, RAG, and LLMs. This repository contains the foundational layer (Sprint 1) — all AI and RAG components will be added in subsequent sprints.

Architecture Overview

financeai/
│
├── app/                    # FastAPI application layer
│   ├── main.py             # ASGI entry-point, route definitions
│   ├── config.py           # Settings (pydantic-settings)
│   └── dependencies.py     # Dependency injection providers
│
├── domain/                 # Pure domain models — no framework deps
│   ├── schemas.py          # Pydantic v2 response/request schemas
│   └── exceptions.py       # Custom exception hierarchy
│
├── services/               # Application services (Sprint 2+)
├── agents/                 # LangGraph agents       (Sprint 3+)
├── graph/                  # LangGraph graph defs   (Sprint 3+)
├── rag/                    # RAG pipeline           (Sprint 4+)
│
├── infrastructure/
│   └── logger.py           # Structured logger factory
│
├── tests/
│   └── test_health.py      # Endpoint smoke tests
│
└── logs/                   # Auto-created at runtime

Clean Architecture Layers

Layer	Responsibility
`app/`	HTTP concerns, routing, DI wiring
`domain/`	Business contracts (schemas, exceptions)
`services/`	Use-case orchestration
`agents/`	AI agent definitions
`infrastructure/`	Cross-cutting concerns (logging, DB)

Tech Stack (Sprint 1)

Library	Purpose
FastAPI	ASGI web framework
Uvicorn	ASGI server
Pydantic v2	Data validation & serialisation
pydantic-settings	Environment-based configuration
python-dotenv	`.env` file loading
pytest + httpx	Testing

Installation

Prerequisites

Python 3.12+
pip or uv

Steps

# 1. Clone the repository
git clone https://github.com/your-org/financeai.git
cd financeai/financeai

# 2. Create and activate a virtual environment
python -m venv .venv
# Windows
.venv\Scripts\activate
# macOS / Linux
source .venv/bin/activate

# 3. Install dependencies
pip install -r requirements.txt

# 4. Copy and configure environment variables
cp .env.example .env
# Edit .env as needed

Running Locally

uvicorn app.main:app --reload

The API will be available at:

URL	Description
`http://localhost:8000/`	Root welcome message
`http://localhost:8000/health`	Health-check endpoint
`http://localhost:8000/docs`	Swagger UI
`http://localhost:8000/redoc`	ReDoc documentation

Running Tests

pytest

Expected output:

tests/test_health.py::TestRootEndpoint::test_root_endpoint_returns_200    PASSED
tests/test_health.py::TestRootEndpoint::test_root_endpoint_body_structure  PASSED
tests/test_health.py::TestRootEndpoint::test_root_endpoint_message_value   PASSED
tests/test_health.py::TestHealthEndpoint::test_health_endpoint_returns_200              PASSED
tests/test_health.py::TestHealthEndpoint::test_health_endpoint_body_has_required_fields PASSED
tests/test_health.py::TestHealthEndpoint::test_health_endpoint_status_value             PASSED
tests/test_health.py::TestHealthEndpoint::test_health_endpoint_service_value            PASSED
tests/test_health.py::TestHealthEndpoint::test_health_endpoint_version_format           PASSED
tests/test_health.py::TestHealthEndpoint::test_health_endpoint_environment_value        PASSED

Environment Variables

Variable	Default	Description
`ENVIRONMENT`	`development`	Deployment environment
`DEBUG`	`true`	Enable debug mode
`LOG_LEVEL`	`INFO`	Logging verbosity

Sprint Roadmap

Sprint	Focus
1 — Foundation ✅	Project bootstrap, config, logging, health endpoints
2 — RAG Foundation ✅	PDF ingestion, chunking, embeddings, ChromaDB, semantic search
3 — AI Agents	LangGraph agent definitions, orchestration graph
4 — Observability	Tracing, metrics, structured logging at scale
5 — Governance	Rate limiting, audit logs, access control

Code Quality

# Format with Black
black .

# Lint with Ruff
ruff check .

# Type-check with mypy
mypy .

Sprint 2 — RAG Foundation

O que é RAG?

Retrieval Augmented Generation (RAG) permite que modelos de linguagem respondam com base em documentos privados — como relatórios financeiros — que não fazem parte do seu treinamento original.

Em vez de depender apenas do conhecimento do LLM, o RAG recupera trechos relevantes do documento e os fornece como contexto para a geração de resposta.

Fluxo

PDF Financeiro
    ↓
PyPDFLoader      — carrega páginas
    ↓
Chunker          — divide em trechos de 1 000 chars (overlap 200)
    ↓
GoogleEmbeddings — vetoriza cada trecho
    ↓
ChromaDB         — persiste em data/vector_db/
    ↓
Retriever        — busca semântica por similaridade de cosseno

Pré-requisitos

Definir a chave da API do Google Gemini no .env:

GEMINI_API_KEY=sua_chave_aqui

Sem esta chave o sistema lança ConfigurationException com mensagem amigável.

Instalação

pip install -r requirements.txt

Ingerir documentos

Copie seus PDFs para:

data/raw_reports/

Execute a ingestão:

curl -X POST http://localhost:8000/documents/ingest

Resposta esperada:

{
  "files_processed": 3,
  "documents_loaded": 210,
  "chunks_created": 430
}

Pesquisa semântica

curl "http://localhost:8000/documents/search?query=receita+liquida+petrobras&k=4"

Resposta esperada:

{
  "query": "receita liquida petrobras",
  "results": [
    "A receita líquida da Petrobras no trimestre...",
    "Os resultados operacionais demonstram..."
  ]
}

Variáveis de ambiente (Sprint 2)

Variável	Padrão	Descrição
`GEMINI_API_KEY`	``	Google Gemini API key (obrigatória para embeddings)
`CHROMA_PERSIST_DIR`	`data/vector_db`	Diretório de persistência do ChromaDB
`RAW_REPORTS_DIR`	`data/raw_reports`	Diretório com PDFs para ingestão
`CHROMA_COLLECTION_NAME`	`financeai_documents`	Nome da coleção ChromaDB

Sprint 3 — Specialized Agents

Filosofia

Um agente = uma responsabilidade.

Cada agente do FinanceAI possui escopo restrito e não conhece a existência dos demais. O acoplamento ocorre exclusivamente através da camada de serviço, respeitando os princípios SOLID (SRP, OCP, DIP).

Fluxo

ResearcherAgent   — busca dados financeiros (RAG ou mock)
    ↓
AnalystAgent      — interpreta dados com lógica de regras
    ↓
ReportAgent       — compõe relatório executivo estruturado
    ↓
ExecutiveReport   — JSON retornado pela API

Agentes

Agente	Arquivo	Input	Output
ResearcherAgent	`agents/researcher_agent.py`	`str` (empresa)	`FinancialData`
AnalystAgent	`agents/analyst_agent.py`	`FinancialData`	`FinancialAnalysis`
ReportAgent	`agents/report_agent.py`	`(str, FinancialAnalysis)`	`ExecutiveReport`

Benefícios da arquitetura

Baixo acoplamento — agentes não se importam uns com os outros
Alta coesão — cada agente faz uma coisa bem feita
Facilidade de testes — cada agente é testável isoladamente com mocks
Pronto para LangGraph — interfaces já compatíveis com Sprint 4
Substituição independente — trocar um agente não afeta os demais

Como usar

curl -X POST http://localhost:8000/analysis/mock \
  -H "Content-Type: application/json" \
  -d '{"company": "Petrobras"}'

Resposta:

{
  "company": "Petrobras",
  "summary": "A Petrobras apresenta liquidez alta e lucratividade alta...",
  "analysis": {
    "liquidity": "Alta",
    "profitability": "Alta",
    "debt_level": "Saudável",
    "strengths": ["Receita sólida de R$ 502.3 bilhões"],
    "risks": ["Exposição a variações cambiais e macroeconômicas"]
  },
  "recommendations": [
    "Considerar aumento de dividendos dado o alto nível de lucratividade.",
    "Estrutura de capital saudável — manter disciplina financeira."
  ]
}

Executar testes dos agentes

pytest tests/test_researcher_agent.py tests/test_analyst_agent.py \
       tests/test_report_agent.py tests/test_analysis_service.py -v

Sprint 4 — LangGraph Orchestration

Filosofia

Orquestração resiliente, modular e baseada em grafos.

Em vez do acoplamento de chamadas de serviço sequenciais manuais, a Sprint 4 introduz a orquestração multiagente utilizando LangGraph. Isso nos dá visibilidade de estado, resiliência contra falhas isoladas de nós e extensibilidade nativa para fluxos futuros mais complexos (como caminhos condicionais e loops de feedback).

Fluxo do Grafo

START → [research] → [analysis] → [report] → END

Shared State (FinancialGraphState): Um dicionário estruturado que flui entre todos os nós e acumula dados incrementalmente.
Nodes: Closures do grafo que capturam as instâncias injetadas dos agentes (ResearcherAgent, AnalystAgent, ReportAgent), executam a lógica correspondente e registram a saída sob a chave apropriada.
Resiliência e Coleta de Erros: Se um nó falhar (ex: falha de infraestrutura ou dados indisponíveis), a falha é capturada pelo tratamento de exceções do nó, descrita em um formato amigável e adicionada à lista errors no estado. O grafo continua avançando até END (degradação graciosa). Os nós subsequentes detectam a falta de dados e pulam a execução de forma controlada.

Estrutura do Estado (`FinancialGraphState`)

class FinancialGraphState(TypedDict):
    company: str                   # Entrada inicial
    financial_data: NotRequired[dict] # Preenchido por research
    analysis: NotRequired[dict]       # Preenchido por analysis
    report: NotRequired[dict]         # Preenchido por report
    errors: NotRequired[list[str]]    # Acumulador de erros não fatais

Como usar a API (Retorna o estado completo)

curl -X POST http://localhost:8000/analysis/mock \
  -H "Content-Type: application/json" \
  -d '{"company": "Petrobras"}'

Resposta esperada:

{
  "company": "Petrobras",
  "financial_data": {
    "company": "Petrobras",
    "revenue": "R$ 502,3 bilhões",
    "profit": "R$ 124,6 bilhões",
    "debt": "R$ 83,1 bilhões",
    "equity": "R$ 321,4 bilhões"
  },
  "analysis": {
    "liquidity": "Moderada",
    "profitability": "Alta",
    "debt_level": "Saudável",
    "strengths": ["Receita sólida de R$ 502.3 bilhões", ...],
    "risks": [...]
  },
  "report": {
    "company": "Petrobras",
    "summary": "A Petrobras apresenta liquidez moderada...",
    "analysis": { ... },
    "recommendations": [ ... ]
  },
  "errors": []
}

Executar testes da Sprint 4

# Rodar testes específicos do grafo
pytest tests/test_graph_pipeline.py -v

# Rodar todos os testes do projeto
pytest tests/ -v

Sprint 5 — Analysis API Layer

O que foi feito

Modularização das Rotas: Lógica de endpoints totalmente extraída de main.py e organizada em pacotes dedicados sob app/routers/:
- health.py (rotas / e /health)
- documents.py (rotas /documents/ingest e /documents/search)
- analyses.py (rotas /analyses e rota legada /analysis/mock)
Novo Endpoint /analyses: Endpoint robusto e profissional para análise multiagente estruturada.
UUID por Análise: Geração dinâmica de identificador de transação (analysis_id via uuid4) para rastreamento.
Métricas de Performance: Medição precisa do tempo total de execução do pipeline no endpoint (processing_time_seconds).
Tratamento de Erros Profissional: Blindagem contra vazamento de stack traces de infraestrutura ao usuário. Erros inesperados retornam HTTP 500 genérico com mensagem limpa, e o stack trace real é gravado apenas no logger local.
Validação de Entrada Reforçada: Parâmetro company do request passa a exigir entre 2 e 120 caracteres.
Logs Transacionais Ricos: Cada requisição de análise gera registros detalhados contendo analysis_id, company, status, processing_time_seconds e errors_count.
Compatibilidade Retroativa: Rota legada /analysis/mock mantida temporariamente e sinalizada como deprecada.

Como usar a API

# Executar análise
curl -X POST http://localhost:8000/analyses \
  -H "Content-Type: application/json" \
  -d '{"company": "Petrobras"}'

Resposta esperada:

{
  "analysis_id": "9b1deb4d-3b7d-4bad-9bdd-2b0d7b3dcb6d",
  "company": "Petrobras",
  "status": "completed",
  "financial_data": {
    "company": "Petrobras",
    "revenue": "R$ 502,3 bilhões",
    "profit": "R$ 124,6 bilhões",
    "debt": "R$ 83,1 bilhões",
    "equity": "R$ 321,4 bilhões"
  },
  "analysis": {
    "liquidity": "Moderada",
    "profitability": "Alta",
    "debt_level": "Saudável",
    "strengths": ["Receita sólida de R$ 502.3 bilhões"],
    "risks": [...]
  },
  "report": {
    "company": "Petrobras",
    "summary": "A Petrobras apresenta liquidez moderada...",
    "analysis": { ... },
    "recommendations": [ ... ]
  },
  "errors": [],
  "processing_time_seconds": 0.045
}

Executar testes da Sprint 5

# Rodar testes específicos dos routers
pytest tests/test_analyses_router.py tests/test_health_router.py tests/test_documents_router.py -v

# Rodar a suíte completa de testes
pytest tests/ -v

Sprint 6 — Governance, Observability & Audit Trail

O que foi implementado

Logs Estruturados: O logger principal foi atualizado para escrever no arquivo logs/app.log no formato estruturado timestamp | level | module | message. Os logs também continuam sendo exibidos no console e a pasta logs/ é criada automaticamente se ausente.
Audit Trail por Análise: Cada execução de análise via POST /analyses gera um arquivo JSON estruturado em logs/audits/ com o nome analysis_{analysis_id}.json.
Mapeamento de Status de Análise: Utiliza funções de monitoramento para calcular se o status é completed (sem erros) ou completed_with_errors (caso existam erros acumulados no grafo).
Camada de Monitoramento: Uma nova camada helper em infrastructure/monitoring.py rastreia quais etapas do grafo foram executadas e se o relatório final foi produzido de forma coerente.
Tolerância a Falhas de Auditoria: O salvamento do log de auditoria é isolado. Caso ocorra uma falha no sistema de arquivos ou permissão de escrita, a falha é registrada nos logs, mas a análise retorna com sucesso ao usuário final com audit_file: null (alta disponibilidade do fluxo principal).
Testes de Observabilidade: Implementados testes em test_audit_logger.py e test_monitoring.py isolados usando diretórios temporários para evitar lixo em produção.

Estrutura do Arquivo de Auditoria

{
  "analysis_id": "9b1deb4d-3b7d-4bad-9bdd-2b0d7b3dcb6d",
  "company": "Petrobras",
  "status": "completed",
  "processing_time_seconds": 1.23,
  "errors_count": 0,
  "errors": [],
  "pipeline_steps": [
    "research",
    "analysis",
    "report"
  ],
  "report_generated": true,
  "created_at": "2026-06-25T10:30:00"
}

Executar testes da Sprint 6

# Rodar testes de observabilidade e auditoria
pytest tests/test_audit_logger.py tests/test_monitoring.py -v

# Rodar todos os testes (104 passantes)
pytest tests/ -v

Sprint 7 — Testing, Quality & Robustness

O que foi implementado

** pytest-cov e Configuração no pyproject.toml**: O arquivo pyproject.toml foi configurado para executar testes sob o pytest-cov, coletando estatísticas e listando linhas ausentes para todos os pacotes chave (app, agents, services, graph, rag, infrastructure).
Fixtures Reutilizáveis em conftest.py: Criado conftest.py contendo fixtures centralizadas para a aplicação (client integrado ao TestClient) e stubs de dados (mock_financial_data, mock_financial_analysis, mock_executive_report, mock_graph_result), eliminando duplicações e mocks redundantes nos arquivos de testes individuais.
Validação de Schemas: Criado test_schemas.py contendo testes estritos de contrato para AnalysisRequest (validando limites de caracteres de 2 a 120, vazios e nulos), AnalysisAPIResponse e HealthResponse.
Robustez e Segurança contra Vazamento: Criado test_error_handling.py validando que as exceções customizadas (ConfigurationException, InfrastructureException) são corretamente estruturadas e que falhas severas e não mapeadas na API são mascaradas com HTTP 500 genérico sem vazar o stack trace interno.
Isolamento Total dos Testes: Garantia de que nenhuma chave de API externa seja requisitada, nenhum arquivo real seja gravado em caminhos de logs de produção (auditoria gravada exclusivamente usando a fixture tmp_path) e nenhum PDF físico seja necessário para execução.

Executar testes e Cobertura da Sprint 7

# Rodar a suíte completa de testes com cobertura detalhada
pytest

# Rodar testes de qualidade/schemas especificamente
pytest tests/test_schemas.py tests/test_error_handling.py -v

Name		Name	Last commit message	Last commit date
Latest commit History 1 Commit
agents		agents
app		app
data		data
domain		domain
graph		graph
infrastructure		infrastructure
rag		rag
services		services
tests		tests
.env.example		.env.example
.gitignore		.gitignore
README.md		README.md
pyproject.toml		pyproject.toml
requirements.txt		requirements.txt

Folders and files

Latest commit

History

Repository files navigation

FinanceAI

Architecture Overview

Clean Architecture Layers

Tech Stack (Sprint 1)

Installation

Prerequisites

Steps

Running Locally

Running Tests

Environment Variables

Sprint Roadmap

Code Quality

Sprint 2 — RAG Foundation

O que é RAG?

Fluxo

Pré-requisitos

Instalação

Ingerir documentos

Pesquisa semântica

Variáveis de ambiente (Sprint 2)

Sprint 3 — Specialized Agents

Filosofia

Fluxo

Agentes

Benefícios da arquitetura

Como usar

Executar testes dos agentes

Sprint 4 — LangGraph Orchestration

Filosofia

Fluxo do Grafo

Estrutura do Estado (FinancialGraphState)

Como usar a API (Retorna o estado completo)

Executar testes da Sprint 4

Sprint 5 — Analysis API Layer

O que foi feito

Como usar a API

Executar testes da Sprint 5

Sprint 6 — Governance, Observability & Audit Trail

O que foi implementado

Estrutura do Arquivo de Auditoria

Executar testes da Sprint 6

Sprint 7 — Testing, Quality & Robustness

O que foi implementado

Executar testes e Cobertura da Sprint 7

License

About

Topics

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages

Estrutura do Estado (`FinancialGraphState`)