AI-upflux-docprocessor/docs/modulos.md

Documentação dos Módulos

1. app.py — Aplicação Principal

Ponto de entrada da API REST. Define os endpoints FastAPI, autentica as requisições e orquestra o processamento das guias.

Autenticação

Toda requisição ao endpoint /process deve incluir o header:

X-API-Key: <chave configurada no Secrets Manager>

Se a chave estiver ausente ou incorreta, a API retorna HTTP 403 Forbidden.

Função process_guia(guia: dict) → dict

Processa uma única guia de autorização do início ao fim.

Etapas:

1. Itera sobre todos os anexos da guia (guia.anexos).
2. Para cada anexo, extrai o texto via OCR (em paralelo, usando asyncio.gather).
3. Concatena todos os textos extraídos em uma única string (file_content).
4. Avalia cada serviço médico (guia.servicos) em paralelo, chamando o agente de IA.
5. Adiciona os tempos de processamento ao objeto da guia.

Campos adicionados à guia:

Campo	Tipo	Descrição
anexo.textoExtraido	string	Texto extraído via OCR do arquivo
anexo.pageCount	int	Número de páginas extraídas
anexo.tempoExtracaoSegundos	float	Tempo de extração do anexo em segundos
anexo.error	string	Mensagem de erro (apenas se falhou)
guia.avaliacaoAgente	list	Lista de avaliações dos serviços
guia.tempoProcessamento.extracaoSegundos	float	Tempo total de extração de todos os anexos
guia.tempoProcessamento.agentSegundos	float	Tempo total do agente para todos os serviços
guia.tempoProcessamento.totalSegundos	float	Tempo total de processamento da guia

Função _extract_anexo(anexo_idx, anexo)

Função interna de process_guia. Processa um único anexo:

• Lê o campo urlAnexo ou URLAnexo para obter a URI S3.
• Se a URI for inválida ou não começar com s3://, define textoExtraido = "" e retorna None.
• Chama extract_text_from_s3_document para baixar e executar OCR.
• Em caso de erro, armazena a mensagem em anexo.error e retorna texto vazio.
• Retorna a string formatada "--- <nomeArquivo> ---\n<texto>".

Modelo de Requisição ProcessRequest

class ProcessRequest(BaseModel):
    operadora: dict   # Metadados da operadora (passado diretamente na resposta)
    guias: list[dict] # Lista de guias a serem avaliadas

Endpoints

Método	Rota	Autenticação	Descrição
POST	/process	X-API-Key	Processa todas as guias da requisição
GET	/health	Não	Verifica se o serviço está ativo
GET	/rules	Não	Lista os códigos de procedimento com regras ativas

---

2. services/document_extractor.py — Extração de Documentos

Responsável por baixar arquivos do S3 e extrair texto via AWS Textract.

Clientes AWS

Dois clientes boto3 são criados no início do módulo:

• _s3_input: cliente S3 autenticado com as credenciais do Secrets Manager (para acesso ao bucket de entrada de documentos).
• _textract: cliente Textract usando as credenciais de ambiente padrão.

Função parse_s3_uri(s3_uri: str) → tuple[str, str]

Converte uma URI S3 no formato s3://bucket/chave em uma tupla (bucket, chave).

Erros:

• Lança ValueError se o scheme não for s3://.
• Lança ValueError se o bucket ou a chave estiverem ausentes.

Exemplos:

parse_s3_uri("s3://meu-bucket/pasta/arquivo.pdf")
# → ("meu-bucket", "pasta/arquivo.pdf")

Função extract_text_from_s3_document(bucket: str, key: str) → tuple[str, int]

Função assíncrona principal do módulo. Baixa o arquivo e extrai o texto conforme o tipo:

Extensão	Comportamento
.png, .jpg, .jpeg	Envia os bytes diretamente ao Textract. Retorna (texto, 1).
.pdf	Separa o PDF em páginas individuais. Cada página é enviada ao Textract separadamente (em paralelo). Retorna (texto_completo, num_paginas).
Outros	Retorna ("", 0) sem processar.

Nota: Operações bloqueantes do boto3 são executadas em threads separadas via asyncio.to_thread para não bloquear o event loop.

Funções Auxiliares

_split_pdf_pages(pdf_bytes: bytes) → list[bytes]

Usa PyPDF2 para ler o PDF e separar cada página em um arquivo PDF individual (em bytes). Necessário porque o Textract aceita no máximo uma página por chamada via bytes.

_textract_detect_bytes(file_bytes: bytes) → str

Chama textract.detect_document_text com os bytes do arquivo e retorna o texto extraído.

_extract_text_from_textract_response(response: dict) → str

Filtra os blocos da resposta do Textract, mantendo apenas blocos do tipo LINE, e os concatena em uma string com quebras de linha.

---

3. services/authorization.py — Avaliação de Serviços

Avalia se um serviço médico deve ser aprovado ou reprovado, consultando o agente de IA.

Função evaluate_servico(servico, guia, file_content) → dict

Parâmetros:

Parâmetro	Tipo	Descrição
servico	dict	Objeto do serviço médico com codigoServico e outros campos
guia	dict	Guia completa (atendimento, guia, histórico)
file_content	str	Texto OCR concatenado de todos os anexos

Fluxo:

1. Extrai o codigoServico e mantém apenas os dígitos numéricos.
2. Verifica se o código está presente no dicionário RULES (carregado do rules.yaml).
   • Se não estiver: retorna resultado = "SKIPPED" imediatamente.
3. Monta um payload JSON com atendimento, guia, servico e historico.
4. Chama run_agent(query, code, file_content) de forma assíncrona.
5. Tenta parsear a resposta JSON do agente para extrair o campo status.
6. Suporta respostas encapsuladas em blocos de código markdown (```json ... ```).

Resposta:

{
  "codigoServico": "40808130",
  "resultado": "Aprovado",
  "agentOutput": "{ \"status\": \"Aprovado\", ... }",
  "input_tokens": 1523,
  "output_tokens": 87,
  "tempoAgentSegundos": 4.32
}

Resultado possível:

Valor	Condição
"Aprovado"	Agente retornou status: Aprovado
"Reprovado"	Agente retornou status: Reprovado ou erro no parse
"SKIPPED"	Código de serviço não possui regras configuradas

---

4. services/result_store.py — Persistência de Resultados

Salva os resultados do processamento no S3 de saída.

Função save_results(results: list[dict]) → None

Persiste cada guia processada como um arquivo JSON separado no S3.

Localização no S3:

s3://<OUTPUT_BUCKET>/<API_VERSION>/<codigoGuiaLocal>_<YYYYMMDD_HHMMSS>.json

Exemplo:

s3://upflux-doc-analyzer/v1/GUIA-2025-0001_20251117_103000.json

Comportamento:

• Todas as guias são salvas em paralelo com asyncio.gather.
• Erros são silenciados (except Exception: pass) para não impedir a resposta da API.
• Usa o cliente S3 padrão de ambiente (sem credenciais explícitas).

---

5. utils/langgraph_agent.py — Agente de IA (LangGraph)

Núcleo de decisão do sistema. Define o fluxo do agente ReAct usando LangGraph e executa a avaliação via AWS Bedrock (Claude).

Carregamento das Regras

Ao importar o módulo, o arquivo rules.yaml é lido e dois dicionários são populados:

• RULES: mapeamento de código de procedimento → critérios de auto-aprovação (texto livre).
• MIN_DOC: mapeamento de código de procedimento → documentação mínima exigida (texto livre).

AgentState (TypedDict)

Estado compartilhado entre os nós do grafo LangGraph:

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

O campo messages acumula todas as mensagens trocadas (System, Human, AI, Tool) ao longo da execução.

Função create_agent(file_content: str) → CompiledGraph

Cria e compila o grafo LangGraph para uma execução específica.

Ferramenta disponível:

• check(expression: str) → str: Retorna o conteúdo OCR dos documentos. O agente pode chamar essa ferramenta quando o JSON da guia não for suficiente para a decisão.

Nós do grafo:

Nó	Função
agent	Chama o LLM (Bedrock Claude) com o estado atual das mensagens
tools	Executa as ferramentas solicitadas pelo LLM

Fluxo condicional:

[entrada] → agent → (tem tool_calls?) → Sim → tools → agent → ...
                                       → Não → END

Função run_agent(query, code, file_content) → dict

Executa o agente para um serviço específico.

System Prompt: Instruções em inglês que descrevem:

• O papel do agente (aprovador/reprovador de procedimentos médicos).
• Os critérios de auto-aprovação do código (extraídos de RULES[code]).
• A documentação mínima exigida (extraída de MIN_DOC[code]).
• A lógica de decisão (auto-aprovação tem prioridade sobre documentação mínima).
• O formato obrigatório de resposta (JSON puro, sem markdown):

{
  "status": "Aprovado" ou "Reprovado",
  "criterio": "descrição breve do critério atendido ou motivo da negação",
  "documentos": [
    { "nome": "nome do documento", "pertence_ao_paciente": true }
  ]
}

Human Message: JSON da guia + conteúdo OCR dos documentos dentro da tag <documentos_anexados>.

Observabilidade: Cada execução é rastreada no Langfuse via CallbackHandler.

Retorno:

{
    "response": "<JSON string da resposta do agente>",
    "input_tokens": 1523,
    "output_tokens": 87
}

Os tokens são somados de todas as mensagens AI geradas durante a execução (incluindo chamadas intermediárias de ferramentas).

---

6. utils/config.py — Configurações

Carrega variáveis de ambiente com valores padrão:

AWS_REGION      = os.environ.get("AWS_REGION", "us-east-2")
BEDROCK_MODEL_ARN = os.environ["BEDROCK_MODEL_ARN"]   # obrigatória
OUTPUT_BUCKET   = os.environ.get("OUTPUT_BUCKET", "upflux-doc-analyzer")
API_VERSION     = os.environ.get("API_VERSION", "v1")
LANGFUSE_HOST   = os.environ.get("LANGFUSE_HOST", "https://cloud.langfuse.com")

BEDROCK_MODEL_ARN é obrigatória; se ausente, o serviço falha na inicialização com KeyError.

---

7. utils/secrets_manager.py — Gerenciamento de Segredos

Carrega os segredos do AWS Secrets Manager uma única vez na inicialização:

_client = boto3.client("secretsmanager", region_name="us-east-2")
_response = _client.get_secret_value(SecretId="doc-analyzer")
SECRETS: dict = json.loads(_response["SecretString"])

O dicionário SECRETS é importado pelos módulos que precisam de credenciais (app.py, document_extractor.py, langgraph_agent.py).

Chaves esperadas em SECRETS:

Chave	Usado em
API-KEY	app.py (autenticação da API)
AWS_ACCESS_KEY	document_extractor.py (S3)
AWS_SECRET_KEY	document_extractor.py (S3)
LANGFUSE-SECRET-KEY	langgraph_agent.py (Langfuse)
LANGFUSE-PUBLIC-KEY	langgraph_agent.py (Langfuse)

---

8. utils/rules.yaml — Regras de Autorização

Arquivo YAML com dois blocos principais:

rules — Critérios de Auto-Aprovação

Cada entrada é um código de procedimento TUSS mapeado para um texto descrevendo as condições que, se atendidas, permitem aprovação imediata sem auditoria médica.

Exemplos de códigos:

Código	Procedimento
40808130	Densitometria óssea (osteoporose)
31303293	DIU Hormonal (Mirena/Kyleena)
31005470	Colecistectomia (cálculo biliar/colelitíase)
40901254	Ultrassom obstétrico 1º trimestre
40901262	Ultrassom obstétrico morfológico
41501012	Cirurgia de catarata
20103190	Reabilitação do assoalho pélvico
40304906	Eco-doppler venoso

min_doc — Documentação Mínima

Para cada código, descreve quais documentos são necessários quando os critérios de auto-aprovação não são atendidos. O agente verifica a presença dessas informações nos anexos OCR.