S3-Dokumenten-Pipeline

version: "1.0"

name: document-classifier
type: llm
model: connic/gemini-3.5-flash-lite
description: "Classifies extracted documents and routes them to the right workflow"

system_prompt: |
  You receive structured extraction output from the intake agent. Your job is
  to make a routing decision.

  Classification rules:
  - Assign a primary category and sub-category.
  - Set a confidence score (0-100). Below 70 means "needs human review".
  - Determine the routing destination:
    - invoice -> "accounts-payable"
    - contract -> "legal-review"
    - report -> "analytics-team"
    - letter/memo -> "general-inbox"
    - form -> "data-entry"
    - other -> "manual-review"
  - Add tags for searchability (e.g., "q4-2025", "vendor:acme", "urgent").

  If the extraction flagged truncation, mark confidence as maximum 60 and
  route to manual-review regardless of type.

temperature: 0.2
output_schema: classification-result

version: "1.0"

name: document-intake
type: llm
model: connic/gemini-3.7-flash
description: "Processes files uploaded to S3 - extracts text, metadata, and structured content"

system_prompt: |
  You process documents uploaded to an S3 bucket.

  Your job:
  1. Analyze the document from the content object. Check content.encoding
     to know if it's raw text ("utf-8") or base64-encoded binary ("base64").
  2. Extract the title, author, creation date, and a concise summary.
  3. Identify the document type: contract, invoice, report, letter, form, memo, or other.
  4. Pull out key entities: people, organizations, dates, monetary amounts.
  5. If the file is an image, describe what you see and extract any visible text.

  Always include the S3 key in your output for downstream traceability.
  For large documents, focus on the first 10 pages and note if content was truncated.

temperature: 0.2
output_schema: intake-result

retry_options:
  attempts: 3
  max_delay: 30

version: "1.0"

name: document-pipeline
type: sequential
description: "Full document processing: intake from S3, then classify and route"

agents:
  - document-intake
  - document-classifier

"""Middleware for document-intake.

Validates file size and type before running the LLM. Rejects unsupported
formats with StopProcessing to avoid wasted tokens.

The S3 inbound connector delivers a JSON payload with:
bucket, key, size, etag, event_name, event_time, _s3,
and optionally content: {text, content_type, size_bytes, encoding}
"""

import json
from typing import Any

from connic import StopProcessing

SUPPORTED_TYPES = {
    "application/pdf",
    "text/plain",
    "text/csv",
    "text/markdown",
    "text/html",
    "image/png",
    "image/jpeg",
    "image/webp",
    "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
}

MAX_SIZE_BYTES = 50 * 1024 * 1024


async def before(content: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    """Reject unsupported file types and oversized files before LLM runs."""
    parts = content.get("parts", [])
    for part in parts:
        if isinstance(part, dict) and "text" in part:
            try:
                payload = json.loads(part["text"])
            except (json.JSONDecodeError, TypeError):
                continue

            if not isinstance(payload, dict):
                continue

            context["s3_bucket"] = payload.get("bucket", "")
            context["s3_key"] = payload.get("key", "")
            context["pipeline_stage"] = "intake"

            file_size = payload.get("size", 0)
            if file_size > MAX_SIZE_BYTES:
                raise StopProcessing(
                    f"File too large: {file_size} bytes "
                    f"(max {MAX_SIZE_BYTES // (1024*1024)} MB). "
                    f"Key: {payload.get('key', 'unknown')}"
                )

            file_content = payload.get("content", {})
            if isinstance(file_content, dict):
                ct = file_content.get("content_type", "")
                if ct and ct not in SUPPORTED_TYPES:
                    raise StopProcessing(
                        f"Unsupported file type: {ct}. "
                        f"Supported: {', '.join(sorted(SUPPORTED_TYPES))}"
                    )

    return content


async def after(response: str, context: dict[str, Any]) -> str:
    """Track processing metrics."""
    context["pipeline_stage"] = "intake_complete"
    return response

{
  "type": "object",
  "description": "Document classification and routing decision",
  "properties": {
    "category": {
      "type": "string",
      "enum": ["contract", "invoice", "report", "letter", "form", "memo", "other"]
    },
    "sub_category": {
      "type": "string",
      "description": "More specific classification (e.g. 'nda', 'sow', 'quarterly-report')"
    },
    "confidence": {
      "type": "number",
      "minimum": 0,
      "maximum": 100,
      "description": "Classification confidence; below 70 triggers human review"
    },
    "route_to": {
      "type": "string",
      "enum": ["accounts-payable", "legal-review", "analytics-team", "general-inbox", "data-entry", "manual-review"],
      "description": "Destination team or workflow"
    },
    "tags": {
      "type": "array",
      "items": {"type": "string"},
      "description": "Searchable tags for the document"
    },
    "needs_human_review": {
      "type": "boolean",
      "description": "True if confidence is low or content was truncated"
    }
  },
  "required": ["category", "confidence", "route_to"]
}

{
  "type": "object",
  "description": "Document intake extraction result",
  "properties": {
    "s3_key": {
      "type": "string",
      "description": "Original S3 object key for traceability"
    },
    "title": {
      "type": "string",
      "description": "Document title"
    },
    "author": {
      "type": "string",
      "description": "Document author if identifiable"
    },
    "creation_date": {
      "type": "string",
      "description": "Document creation or issue date (YYYY-MM-DD)"
    },
    "document_type": {
      "type": "string",
      "enum": ["contract", "invoice", "report", "letter", "form", "memo", "other"]
    },
    "summary": {
      "type": "string",
      "description": "Concise summary of the document content"
    },
    "entities": {
      "type": "object",
      "properties": {
        "people": {"type": "array", "items": {"type": "string"}},
        "organizations": {"type": "array", "items": {"type": "string"}},
        "dates": {"type": "array", "items": {"type": "string"}},
        "amounts": {"type": "array", "items": {"type": "string"}}
      }
    },
    "truncated": {
      "type": "boolean",
      "description": "True if the document was too long and content was cut"
    },
    "page_count": {
      "type": "integer",
      "description": "Number of pages processed"
    }
  },
  "required": ["s3_key", "document_type", "summary"]
}

version: "1.0"

# Both agents are tool-less, so no mocks are needed. The payload mimics the S3
# inbound connector shape (bucket/key/size/content) that the middleware reads -
# the document text is supplied inline via content.text, since the middleware
# extracts content from the JSON payload rather than from attached files.
defaults:
  runs: 1
  timeout_s: 90

tests:
  - name: extracts_invoice_metadata
    payload: '{"bucket": "documents-inbox", "key": "inbox/INV-2025-0042.txt", "size": 612, "content": {"content_type": "text/plain", "encoding": "utf-8", "text": "INVOICE\n\nAcme Supplies GmbH\nInvoice Number: INV-2025-0042\nIssue Date: 2025-11-14\nBill To: Northwind Trading Ltd\n\nIndustrial floor mats  10  45.00 EUR  450.00 EUR\nSafety signage pack     4  22.50 EUR   90.00 EUR\n\nSubtotal   540.00 EUR\nVAT (19%)  102.60 EUR\nTotal Due  642.60 EUR\n\nPayment terms: net 30."}}'
    expected_result: output.title != "" or output.document_type != ""

version: "1.0"

# End-to-end through the sequential pipeline: intake extracts the document, then
# the classifier routes it. No custom tools, so no mocks. The invoice text is
# supplied inline via content.text the way the middleware reads it.
defaults:
  runs: 1
  timeout_s: 150

tests:
  - name: invoice_is_classified_and_routed
    payload: '{"bucket": "documents-inbox", "key": "inbox/INV-2025-0042.txt", "size": 612, "content": {"content_type": "text/plain", "encoding": "utf-8", "text": "INVOICE\n\nAcme Supplies GmbH\nInvoice Number: INV-2025-0042\nIssue Date: 2025-11-14\nBill To: Northwind Trading Ltd\n\nIndustrial floor mats  10  45.00 EUR  450.00 EUR\nSafety signage pack     4  22.50 EUR   90.00 EUR\n\nSubtotal   540.00 EUR\nVAT (19%)  102.60 EUR\nTotal Due  642.60 EUR\n\nPayment terms: net 30."}}'
    expected_result: output.category != "" and output.route_to != ""