Vérificateur de conformité sous-traitants

Un service de vérification qui prend un ensemble de documents de conformité de sous-traitants — certificats d'assurance, licences professionnelles, inductions sécurité — les uploade dans Alloovium, puis exécute une série de questions Oui/Non structurées sur chaque document pour vérifier les niveaux de couverture, les dates d'expiration et les endorsements requis. Le résultat est une matrice de conformité par sous-traitant qui peut être stockée dans votre ERP ou envoyée à un chef de projet comme rapport.

• Une clé API avec les scopes scopes.
• Python 3.11+ avec dep installé.
• Documents de conformité en fichiers PDF (certificats d'assurance, licences professionnelles, tickets de sécurité).
• Un projet dans Alloovium à utiliser comme vault de documents de conformité, ou la capacité d'en créer un par sous-traitant.

Utilisez le même flux d'upload bulk prepare/confirm que dans le guide d'intelligence contractuelle. Pour un vérificateur de conformité, il est judicieux de créer un projet par sous-traitant afin que les frontières de documents soient claires lors des requêtes.

python
import httpx
import asyncio
from pathlib import Path

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 upload_compliance_documents(
    client: httpx.AsyncClient,
    project_id: str,
    file_paths: list[str],
) -> list[str]:
    """Upload multiple compliance docs and return their document IDs."""
    doc_entries = []
    for path_str in file_paths:
        p = Path(path_str)
        doc_entries.append({
            "filename": p.name,
            "content_type": "application/pdf",
            "size": p.stat().st_size,
        })

    resp = await client.post(
        f"{BASE_URL}/projects/{project_id}/documents/bulk-prepare",
        json={"documents": doc_entries},
        headers=HEADERS,
    )
    resp.raise_for_status()
    prepared_docs = resp.json()["documents"]

    doc_ids: list[str] = []
    for prepared, path_str in zip(prepared_docs, file_paths):
        doc_id = prepared["document_id"]
        upload_url = prepared["upload_url"]
        upload_fields = prepared.get("upload_fields", {})

        with open(path_str, "rb") as f:
            if upload_fields:
                s3_resp = await client.post(
                    upload_url,
                    data=upload_fields,
                    files={"file": (Path(path_str).name, f, "application/pdf")},
                )
            else:
                s3_resp = await client.put(
                    upload_url,
                    content=f.read(),
                    headers={"Content-Type": "application/pdf"},
                )
        s3_resp.raise_for_status()
        doc_ids.append(doc_id)

    confirm_resp = await client.post(
        f"{BASE_URL}/projects/{project_id}/documents/bulk-confirm",
        json={"document_ids": doc_ids},
        headers=HEADERS,
    )
    confirm_resp.raise_for_status()
    return doc_ids

Une fois les documents traités, posez des questions Oui/Non ciblées. Structurer le prompt pour retourner un objet JSON avec les champs answer, evidence et confidence vous donne à la fois un résultat lisible par machine et une piste d'audit lisible par l'homme.

python
import json
import re


async def ask_and_collect(
    client: httpx.AsyncClient,
    conversation_id: str,
    question: str,
) -> str:
    """Stream a chat response and return the full answer text."""
    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
            try:
                event = json.loads(payload)
            except json.JSONDecodeError:
                continue
            if event.get("type") == "token":
                full_answer.append(event.get("token", ""))
            elif event.get("type") == "complete":
                return event.get("answer", "".join(full_answer))
    return "".join(full_answer)


def extract_json(raw: str) -> object:
    cleaned = re.sub(r"^```(?:json)?\s*", "", raw.strip(), flags=re.MULTILINE)
    cleaned = re.sub(r"```\s*$", "", cleaned, flags=re.MULTILINE)
    return json.loads(cleaned.strip())

python
def build_insurance_check_prompt(
    coverage_type: str,
    required_amount_aud: float,
) -> str:
    return f"""
Review this insurance certificate.

Check: Does this certificate provide {coverage_type} coverage of at least
AUD {required_amount_aud:,.0f}?

Return a JSON object with exactly these fields:
- answer: "yes", "no", or "unclear"
- coverage_amount_aud: the actual coverage amount as a number, or null if not found
- expiry_date: ISO date string of the policy expiry, or null
- insurer: name of the insurer, or null
- policy_number: the policy number, or null
- evidence: one sentence quoting the relevant clause from the document
- confidence: "high", "medium", or "low"

Raw JSON only. No markdown.
"""


PUBLIC_LIABILITY_CHECK = build_insurance_check_prompt(
    "public liability", 20_000_000
)
WORKERS_COMP_CHECK = build_insurance_check_prompt(
    "workers compensation", 0  # statutory — any valid policy passes
)
PROFESSIONAL_INDEMNITY_CHECK = build_insurance_check_prompt(
    "professional indemnity", 5_000_000
)

python
from dataclasses import dataclass
from typing import Optional


@dataclass
class InsuranceCheckResult:
    check_name: str
    answer: str  # "yes", "no", or "unclear"
    coverage_amount_aud: Optional[float]
    expiry_date: Optional[str]
    insurer: Optional[str]
    policy_number: Optional[str]
    evidence: str
    confidence: str
    passed: bool


def parse_insurance_check(
    check_name: str,
    raw_answer: str,
    required_amount_aud: float,
) -> InsuranceCheckResult:
    data = extract_json(raw_answer)
    assert isinstance(data, dict)

    answer = data.get("answer", "unclear")
    actual_amount = data.get("coverage_amount_aud")

    # A "yes" from the LLM AND the stated amount meeting the threshold
    passed = (
        answer == "yes"
        and (
            required_amount_aud == 0  # statutory — any valid policy
            or (actual_amount is not None and float(actual_amount) >= required_amount_aud)
        )
    )

    return InsuranceCheckResult(
        check_name=check_name,
        answer=answer,
        coverage_amount_aud=float(actual_amount) if actual_amount is not None else None,
        expiry_date=data.get("expiry_date"),
        insurer=data.get("insurer"),
        policy_number=data.get("policy_number"),
        evidence=data.get("evidence", ""),
        confidence=data.get("confidence", "low"),
        passed=passed,
    )

Exécutez les vérifications sur tous les documents de conformité uploadés et assemblez une matrice de conformité par sous-traitant. La matrice regroupe les résultats par type de vérification afin qu'un chef de projet puisse rapidement voir quelles exigences sont satisfaites.

python
from dataclasses import dataclass, field


INSURANCE_CHECKS = [
    ("Public Liability $20M", PUBLIC_LIABILITY_CHECK, 20_000_000),
    ("Workers Compensation", WORKERS_COMP_CHECK, 0),
    ("Professional Indemnity $5M", PROFESSIONAL_INDEMNITY_CHECK, 5_000_000),
]

LICENCE_CHECK_PROMPT = """
Is this a valid Australian contractor or trade licence?
Return JSON with:
- answer: "yes" or "no"
- licence_number: string or null
- licence_type: string (e.g. "Electrical Contractor") or null
- expiry_date: ISO date or null
- issuing_authority: string or null
- evidence: one sentence from the document
Raw JSON only.
"""


@dataclass
class ComplianceMatrix:
    subcontractor_name: str
    project_id: str
    insurance_checks: list[InsuranceCheckResult] = field(default_factory=list)
    licence_check: Optional[dict] = None

    @property
    def fully_compliant(self) -> bool:
        return all(c.passed for c in self.insurance_checks)


async def build_compliance_matrix(
    client: httpx.AsyncClient,
    subcontractor_name: str,
    project_id: str,
) -> ComplianceMatrix:
    """Run all compliance checks and return the matrix."""
    matrix = ComplianceMatrix(
        subcontractor_name=subcontractor_name,
        project_id=project_id,
    )

    # One conversation per check keeps context clean
    for check_name, prompt, required_amount in INSURANCE_CHECKS:
        conv_resp = await client.post(
            f"{BASE_URL}/conversations",
            json={"project_id": project_id, "title": f"Check: {check_name}"},
            headers=HEADERS,
        )
        conv_resp.raise_for_status()
        conv_id = conv_resp.json()["id"]

        raw = await ask_and_collect(client, conv_id, prompt)
        result = parse_insurance_check(check_name, raw, required_amount)
        matrix.insurance_checks.append(result)

    # Licence check
    conv_resp = await client.post(
        f"{BASE_URL}/conversations",
        json={"project_id": project_id, "title": "Licence check"},
        headers=HEADERS,
    )
    conv_resp.raise_for_status()
    conv_id = conv_resp.json()["id"]
    raw_licence = await ask_and_collect(client, conv_id, LICENCE_CHECK_PROMPT)
    matrix.licence_check = extract_json(raw_licence)

    return matrix

En production, vous voulez que la vérification de conformité s'exécute automatiquement quand un sous-traitant uploade un nouveau document. Utilisez un webhook ou une tâche en arrière-plan déclenchée par les changements de statut de document. L'exemple ci-dessous montre un endpoint FastAPI qui reçoit un événement de document traité et exécute la vérification de conformité.

python
# webhook_receiver.py — example FastAPI handler
from fastapi import FastAPI, BackgroundTasks
import httpx

app = FastAPI()


@app.post("/webhooks/document-processed")
async def on_document_processed(
    payload: dict,
    background_tasks: BackgroundTasks,
) -> dict:
    """Handle a document-processed event from Alloovium."""
    project_id = payload.get("project_id")
    doc_id = payload.get("document_id")

    if not project_id or not doc_id:
        return {"status": "ignored"}

    background_tasks.add_task(
        run_compliance_check_for_project, project_id, doc_id
    )
    return {"status": "queued"}


async def run_compliance_check_for_project(
    project_id: str,
    doc_id: str,
) -> None:
    """Background task: re-run compliance matrix after a new document arrives."""
    async with httpx.AsyncClient(timeout=180.0) as client:
        # Determine subcontractor name from project metadata
        proj_resp = await client.get(
            f"{BASE_URL}/projects/{project_id}",
            headers=HEADERS,
        )
        proj_resp.raise_for_status()
        sub_name = proj_resp.json().get("name", "Unknown")

        matrix = await build_compliance_matrix(client, sub_name, project_id)

        if not matrix.fully_compliant:
            failing = [c.check_name for c in matrix.insurance_checks if not c.passed]
            print(f"COMPLIANCE GAP: {sub_name} failing {failing}")
            # TODO: send notification to project manager

• Pipeline d'intelligence contractuelle — combinez avec l'extraction contractuelle pour vérifier les exigences d'assurance stipulées dans le contrat principal.
• Référence API Chat — schéma des événements SSE et options de portée des conversations.
• Référence API Vault — endpoints de recherche pour trouver des documents par tags de métadonnées.