Pipeline IA événementiel : Cloud Storage → Pub/Sub/Eventarc → Cloud Run → BigQuery

Le service comptable de Transports Valmont, transporteur routier de 400 personnes (entreprise fictive), reçoit chaque jour entre 300 et 600 documents : factures de carburant, relances d'impayés, réclamations de clients, avoirs de sous-traitants. Une assistante les ouvre un par un pour les classer et repérer ceux qui ne peuvent pas attendre. Une relance avant mise en demeure est restée trois semaines dans une boîte partagée l'an dernier : 1 200 € de pénalités.

La demande métier tient en une phrase : « dès qu'un document arrive, on veut savoir ce que c'est, combien, et si c'est urgent ». La demande technique est plus exigeante : chaque document doit produire exactement une ligne d'analyse, même quand le réseau coupe, que le quota Gemini sature ou que Pub/Sub livre deux fois le même message. Ce projet construit ce pipeline, le déploie sur Google Cloud, le teste avec dix documents fictifs dont un vide et un message empoisonné, puis le supprime en vérifiant qu'il ne reste rien.

Ce que vous livrez

• Un worker FastAPI sur Cloud Run, privé, appelé par un abonnement push Pub/Sub authentifié.
• Une analyse Gemini 2.5 Flash (Vertex AI, europe-west1) en sortie structurée validée par Pydantic.
• Une table BigQuery partitionnée où un MERGE sur l'ID de message garantit une ligne par document.
• Des réessais à trois niveaux (dans la requête, par Pub/Sub, puis file de lettres mortes).
• Deux comptes de service aux droits limités à une ressource chacun quand c'est possible.
• Des journaux JSON et une métrique basée sur les journaux, ventilée par statut.
• make cloud-up, make e2e, make cloud-down : création, test réel et suppression vérifiée.

Architecture

flowchart LR
  U["Dépôt d'un document<br/>gs://…/entrant/"] --> GCS[("Cloud Storage")]
  GCS -- "notification OBJECT_FINALIZE" --> T["Sujet Pub/Sub<br/>lab-pipeline-evt-docs"]
  T --> S["Abonnement push<br/>10 tentatives, 10 à 60 s"]
  S -- "POST + jeton OIDC" --> W["Cloud Run<br/>worker"]
  W -- "lecture par génération" --> GCS
  W -- "sortie structurée" --> G["Vertex AI<br/>gemini-2.5-flash"]
  W -- "MERGE sur message_id" --> BQ[("BigQuery<br/>analyses")]
  S -- "après 10 échecs" --> DLQ["Sujet de lettres mortes"]
  W -. "journaux JSON" .-> L["Cloud Logging<br/>+ métrique"]

Stack, coût, durée

Coût mesuré du test complet (10 documents) : 4 664 tokens Gemini, soit moins d'un centime au tarif public de Gemini 2.5 Flash. Extrapolé à 1 000 documents, l'ordre de grandeur est de 0,40 $ de Gemini et une dizaine de centimes de BigQuery (détail en leçon 7). Durée : 6 h de lecture et de pratique ; make cloud-up prend environ 1 min 40 s, make e2e environ 6 min, dont 5 min 45 s d'attente des 10 tentatives avant la file de lettres mortes.

Prérequis

• Python et HTTP courants ; avoir déjà déployé un service sur Cloud Run aide.
• Un projet Google Cloud où vous êtes propriétaire (création de comptes de service et de liaisons IAM), gcloud et bq installés, gcloud auth application-default login fait.
• Sans projet Google Cloud : make test (16 tests, sans réseau) et la lecture des leçons restent possibles.

Programme

Le dépôt complet est dans files/ ; commencez par son README.