Pipeline d'intelligence contractuelle

Un service Python qui accepte un PDF de contrat, l'uploade dans un vault de projet Alloovium, attend la fin du traitement, puis interroge le document avec des prompts ciblés pour extraire des données structurées d'obligation, de paiement et de jalon. Les données extraites sont ensuite poussées vers un système externe comme Procore ou Jira.

À la fin de ce guide, vous aurez une fonction async réutilisable qui prend un chemin de contrat et retourne un dataclass Python typé avec tous les champs extraits — prêt à être inséré dans votre base de données ou transmis à une API en aval.

• Une clé API Alloovium avec les scopes scopes. Voir le Démarrage rapide pour la création de clé.
• Python 3.11+ et deps.
• Un contrat au format PDF. Alloovium accepte aussi DOCX.
• Optionnel : un compte Procore ou Jira pour l'étape 5.

bash
pip install httpx

Chaque document dans Alloovium vit à l'intérieur d'un projet. Si vous avez un projet existant, sautez l'étape de création et utilisez directement son ID. Les projets sont permanents — vous en créeriez typiquement un par chantier et le réutiliseriez pour tous les documents de ce chantier.

python
import httpx
import asyncio

BASE_URL = "https://api.alloovium.com/api/v1"
API_KEY = "ak_live_YOUR_KEY_HERE"

HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}


async def create_project(client: httpx.AsyncClient, name: str) -> str:
    """Create a project and return its ID."""
    resp = await client.post(
        f"{BASE_URL}/projects",
        json={"name": name, "project_type": "commercial"},
        headers=HEADERS,
    )
    resp.raise_for_status()
    return resp.json()["id"]

python
import os
from pathlib import Path


async def upload_contract(
    client: httpx.AsyncClient,
    project_id: str,
    contract_path: str,
) -> str:
    """Upload a contract PDF and return the document ID."""
    path = Path(contract_path)
    filename = path.name
    file_size = path.stat().st_size

    # Step 1a — prepare: request a signed upload URL
    resp = await client.post(
        f"{BASE_URL}/projects/{project_id}/documents/bulk-prepare",
        json={
            "documents": [
                {
                    "filename": filename,
                    "content_type": "application/pdf",
                    "size": file_size,
                }
            ]
        },
        headers=HEADERS,
    )
    resp.raise_for_status()
    prepared = resp.json()["documents"][0]
    doc_id = prepared["document_id"]
    upload_url = prepared["upload_url"]
    upload_fields = prepared.get("upload_fields", {})

    # Step 1b — stream the file directly to S3
    with open(contract_path, "rb") as f:
        if upload_fields:
            # Multipart POST to S3
            s3_resp = await client.post(
                upload_url,
                data=upload_fields,
                files={"file": (filename, f, "application/pdf")},
            )
        else:
            # Simple PUT to presigned URL
            s3_resp = await client.put(
                upload_url,
                content=f.read(),
                headers={"Content-Type": "application/pdf"},
            )
    s3_resp.raise_for_status()

    # Step 1c — confirm: tell Alloovium the upload succeeded
    confirm_resp = await client.post(
        f"{BASE_URL}/projects/{project_id}/documents/bulk-confirm",
        json={"document_ids": [doc_id]},
        headers=HEADERS,
    )
    confirm_resp.raise_for_status()

    return doc_id

Après confirmation, Alloovium traite le document de manière asynchrone — OCR, extraction de texte, chunking et embedding. Les temps de traitement typiques sont de 30–120 secondes pour un PDF de contrat standard. Interrogez l'endpoint de statut du document jusqu'à ce que le statut soit status.

python
import asyncio


async def wait_for_processing(
    client: httpx.AsyncClient,
    project_id: str,
    doc_id: str,
    timeout_seconds: int = 300,
) -> None:
    """Poll until the document status is 'processed'. Raises on timeout."""
    deadline = asyncio.get_event_loop().time() + timeout_seconds
    delay = 5.0

    while asyncio.get_event_loop().time() < deadline:
        resp = await client.get(
            f"{BASE_URL}/projects/{project_id}/documents/{doc_id}",
            headers=HEADERS,
        )
        resp.raise_for_status()
        status = resp.json().get("status")

        if status == "processed":
            return
        if status in ("failed", "error"):
            raise RuntimeError(f"Document processing failed: {resp.json()}")

        await asyncio.sleep(delay)
        delay = min(delay * 1.5, 30.0)  # cap at 30s

    raise TimeoutError(f"Document {doc_id} did not process within {timeout_seconds}s")

Avec le document traité, créez une conversation scopée au projet et posez des questions d'extraction ciblées. L'API chat utilise des server-sent events (SSE) pour le streaming. Pour les scripts d'extraction batch, il est souvent plus simple d'accumuler la réponse complète plutôt que de traiter le stream en temps réel — l'exemple ci-dessous fait cela.

python
async def create_conversation(
    client: httpx.AsyncClient,
    project_id: str,
) -> str:
    """Create a conversation scoped to the project and return its ID."""
    resp = await client.post(
        f"{BASE_URL}/conversations",
        json={"project_id": project_id, "title": "Contract extraction"},
        headers=HEADERS,
    )
    resp.raise_for_status()
    return resp.json()["id"]

python
async def ask_and_collect(
    client: httpx.AsyncClient,
    conversation_id: str,
    question: str,
) -> str:
    """Send a message and collect the full streamed answer as a string."""
    full_answer: list[str] = []

    async with client.stream(
        "POST",
        f"{BASE_URL}/conversations/{conversation_id}/messages",
        json={"content": question},
        headers={**HEADERS, "Accept": "text/event-stream"},
        timeout=120.0,
    ) as stream:
        async for line in stream.aiter_lines():
            if not line.startswith("data:"):
                continue
            payload = line[5:].strip()
            if payload in ("", "[DONE]"):
                continue

            import json
            try:
                event = json.loads(payload)
            except json.JSONDecodeError:
                continue

            event_type = event.get("type")
            if event_type == "token":
                full_answer.append(event.get("token", ""))
            elif event_type == "complete":
                # complete event contains the full answer too
                return event.get("answer", "".join(full_answer))

    return "".join(full_answer)

python
OBLIGATION_PROMPT = """
List all contractor obligations from this contract.
For each obligation return a JSON object with these fields:
- clause_ref: the clause number (e.g. "12.3")
- description: a one-sentence plain-language description
- party: who is obligated ("contractor", "principal", or "both")
- deadline: ISO date if a deadline is stated, otherwise null

Return a JSON array. No markdown, no commentary — raw JSON only.
"""

PAYMENT_TERMS_PROMPT = """
Extract all payment terms from this contract.
Return a JSON object with:
- payment_schedule: array of {milestone, amount_aud, due_date}
- retention_percent: number (e.g. 5 for 5%)
- retention_release_condition: string
- late_payment_interest_percent: number or null
- payment_claim_period_days: number

Raw JSON only.
"""

MILESTONE_PROMPT = """
List all project milestones and completion dates from this contract.
Return a JSON array where each item has:
- name: milestone name
- date: ISO date or null if not stated
- liquidated_damages_per_day_aud: number or null

Raw JSON only.
"""

Les LLMs retournent de manière fiable du JSON valide quand le prompt est suffisamment spécifique. Utilisez un helper d'extraction simple qui retire les balises markdown accidentelles, puis analysez avec func.

python
import json
import re
from dataclasses import dataclass, field
from typing import Optional


def extract_json(raw: str) -> object:
    """Strip markdown fences and parse JSON from an LLM response."""
    # Strip leading/trailing whitespace and any markdown code fence markers
    cleaned = raw.strip()
    if cleaned.startswith("json"):
        cleaned = cleaned[4:]
    # Remove lines that are only backtick fence markers
    lines = [ln for ln in cleaned.splitlines() if ln.strip() not in ("json", "")]
    cleaned = "\n".join(lines).strip()
    return json.loads(cleaned)


@dataclass
class PaymentTerms:
    payment_schedule: list[dict]
    retention_percent: float
    retention_release_condition: str
    late_payment_interest_percent: Optional[float]
    payment_claim_period_days: int


@dataclass
class ContractIntelligence:
    project_id: str
    document_id: str
    obligations: list[dict] = field(default_factory=list)
    payment_terms: Optional[PaymentTerms] = None
    milestones: list[dict] = field(default_factory=list)


async def extract_contract_intelligence(
    client: httpx.AsyncClient,
    project_id: str,
    doc_id: str,
) -> ContractIntelligence:
    """Run all extraction prompts and return a populated ContractIntelligence."""
    conv_id = await create_conversation(client, project_id)

    raw_obligations = await ask_and_collect(client, conv_id, OBLIGATION_PROMPT)
    raw_payment = await ask_and_collect(client, conv_id, PAYMENT_TERMS_PROMPT)
    raw_milestones = await ask_and_collect(client, conv_id, MILESTONE_PROMPT)

    obligations = extract_json(raw_obligations)
    payment_data = extract_json(raw_payment)
    milestones = extract_json(raw_milestones)

    payment_terms = PaymentTerms(
        payment_schedule=payment_data.get("payment_schedule", []),
        retention_percent=float(payment_data.get("retention_percent", 0)),
        retention_release_condition=payment_data.get("retention_release_condition", ""),
        late_payment_interest_percent=payment_data.get("late_payment_interest_percent"),
        payment_claim_period_days=int(payment_data.get("payment_claim_period_days", 0)),
    )

    return ContractIntelligence(
        project_id=project_id,
        document_id=doc_id,
        obligations=obligations if isinstance(obligations, list) else [],
        payment_terms=payment_terms,
        milestones=milestones if isinstance(milestones, list) else [],
    )

Une fois que vous avez l'objet cls, poussez vers votre système en aval. Ci-dessous du pseudocode pour les engagements Procore et les issues Jira — substituez vos bibliothèques client réelles ou appels REST.

python
async def push_to_procore(
    intelligence: ContractIntelligence,
    procore_project_id: int,
    procore_token: str,
) -> None:
    """Create Procore schedule items for each extracted milestone."""
    async with httpx.AsyncClient() as client:
        for milestone in intelligence.milestones:
            await client.post(
                f"https://api.procore.com/rest/v1.0/projects/{procore_project_id}/schedule/tasks",
                json={
                    "task": {
                        "name": milestone["name"],
                        "start_date": milestone.get("date"),
                        "end_date": milestone.get("date"),
                        "notes": (
                            f"LD: AUD {milestone['liquidated_damages_per_day_aud']}/day"
                            if milestone.get("liquidated_damages_per_day_aud")
                            else ""
                        ),
                    }
                },
                headers={"Authorization": f"Bearer {procore_token}"},
            )

python
async def push_obligations_to_jira(
    intelligence: ContractIntelligence,
    jira_base_url: str,
    jira_project_key: str,
    jira_token: str,
) -> None:
    """Create a Jira issue for each contractor obligation."""
    async with httpx.AsyncClient() as client:
        for obligation in intelligence.obligations:
            if obligation.get("party") not in ("contractor", "both"):
                continue

            await client.post(
                f"{jira_base_url}/rest/api/3/issue",
                json={
                    "fields": {
                        "project": {"key": jira_project_key},
                        "issuetype": {"name": "Task"},
                        "summary": f"[{obligation['clause_ref']}] {obligation['description']}",
                        "duedate": obligation.get("deadline"),
                        "labels": ["contract-obligation"],
                    }
                },
                headers={
                    "Authorization": f"Bearer {jira_token}",
                    "Content-Type": "application/json",
                },
            )

Pour les projets où les contrats sont amendés régulièrement, exécutez l'extraction sur un planning. L'exemple ci-dessous utilise un simple cron job avec Python. Pour la production, préférez une file de tâches (Celery, ARQ) ou un ordonnanceur cloud (AWS EventBridge, GCP Cloud Scheduler).

python
# cron_extract.py — run with cron: 0 8 * * 1  (every Monday at 08:00)

import asyncio
import httpx


async def run_weekly_extraction() -> None:
    """Re-extract intelligence for all active contracts."""
    async with httpx.AsyncClient(timeout=180.0) as client:
        # Fetch all projects tagged as active
        resp = await client.get(
            f"{BASE_URL}/projects?status=active&limit=100",
            headers=HEADERS,
        )
        resp.raise_for_status()
        projects = resp.json().get("data", [])

        for project in projects:
            project_id = project["id"]

            # Fetch processed documents in this project
            docs_resp = await client.get(
                f"{BASE_URL}/projects/{project_id}/documents?status=processed&limit=50",
                headers=HEADERS,
            )
            docs_resp.raise_for_status()
            docs = docs_resp.json().get("data", [])

            for doc in docs:
                if doc.get("document_type") != "contract":
                    continue
                try:
                    intelligence = await extract_contract_intelligence(
                        client, project_id, doc["id"]
                    )
                    await push_obligations_to_jira(
                        intelligence,
                        jira_base_url="https://yourorg.atlassian.net",
                        jira_project_key="CONST",
                        jira_token="JIRA_TOKEN",
                    )
                except Exception as exc:
                    # Log and continue — do not let one failure block the rest
                    print(f"Extraction failed for {doc['id']}: {exc}")


if __name__ == "__main__":
    asyncio.run(run_weekly_extraction())

• Référence API Chat — schéma complet des événements SSE, portée des conversations et format de citation.
• Référence API Vault — CRUD de projet et document, flux d'upload bulk et endpoints de recherche.
• Guide Vérificateur de conformité — ajoutez la vérification des certificats d'assurance sur ce pipeline.
• Guide Rapports automatisés — livrez des résumés hebdomadaires de statut d'obligation sur Slack.
• API Workflows — exécutez des pipelines d'extraction multi-étapes sans écrire de boucles d'interrogation.