Document Intake — Triage + Semantic Search

Goal: every uploaded document gets simultaneously triaged for routing (classify → priority → entity extraction → assigned team) AND indexed for retrieval (chunked → embedded → searchable). One ingest source, two parallel pipelines.

Why this matters: an enterprise document-intake system needs both. Triage answers “what is this and where does it go?” (a SQL-shaped question — counts, dashboards, routing rules). Retrieval answers “show me similar documents to this one” (a semantic question — RAG, Q&A, related-cases recall).

Primitives used: Storage + Pipelines (two pipelines, same internal-S3 source) + Tables + Vector.

Shape:


   document upload  ──►  Storage bucket
                              │
                              ▼  internal-S3 source matches both rules
                    ┌─────────┴─────────┐
                    │                   │
                    ▼                   ▼
            document_triage    text_embedding_index
            Scriptum pipeline   Scriptum pipeline
                    │                   │
                    ▼                   ▼
            Tables: intake      Vector: intake_vec
            (one row per doc:   (N chunks per doc,
             classify, route,    semantic recall)
             entities, urgency)
                    │                   │
                    ▼                   ▼
            Routing layer       "Similar cases"
            (assigns to team)   (RAG / agent grounding)

Compared to Reviews Dashboard: same fan-out shape, different templates. Reviews focuses on per-record analytics; this recipe focuses on per-document routing + retrieval.

Prerequisites

dodil CLI + dodil auth login

A bucket — kb-intake:


dodil data bucket create kb-intake -d "Document intake pipeline"

No engine setup step — the Tables engine is auto-on and vector capacity is provisioned on demand when you run dodil data vector collection add below.

1. Browse the two templates

document_triage (warehouse-compatible, emits structured rows) and text_embedding_index (vector-compatible). They accept overlapping modalities:


# Tables-side
dodil data table templates --search triage -o json | jq '.templates[] | {id, name, acceptedExtensions}'
 
# Vector-side
dodil data vector templates --search text -o json | jq '.templates[] | {id, name, acceptedExtensions}'

document_triage (templates/core/) accepts txt, md, html, htm, pdf, docx, eml, msg; text_embedding_index accepts txt, md, html, htm, pdf, docx — so .eml / .msg reach triage but not the vector index. Inspect each template’s contract to understand what it produces:


dodil data template get document_triage -o json | jq '.contract'
dodil data template get text_embedding_index -o json | jq '.contract'

2. Create the two destinations

A. Pipeline-bound table for triage


dodil data table pipeline create intake -b kb-intake \
  --description "Document intake — triage classification + entities" \
  --template document_triage

Not dodil data table create. That verb builds a plain Delta table from a column spec (--columns-json, --merge-key, --partition-column) and has no template or pipeline flags. Template-bound tables are dodil data table pipeline create (-t/--template, optional --folder-prefix) — the CLI face of POST /:bucket/pipelines/table.

document_triage declares table_name document_triage, warehouse_mode merge, merge_keys ["input_ref"], and outputs triage, entities, s3, warehouse. The flattened per-document fields it typically lands are source_key, document_type, topic, language, priority, entities, routing_team, urgency_score — but the exact column set is the template’s ScriptContract, so read it with dodil data template get document_triage rather than trusting this list.

B. Pipeline-mode vector collection for retrieval


dodil data vector collection add intake_vec -b kb-intake \
  --description "Document intake — semantic chunks for retrieval" \
  --template text_embedding_index

Capture identifiers. Both destinations are pipelines; the ?facet= split is what separates them:


export TABLE_PIPELINE_ID=$(dodil data table pipeline list -b kb-intake -o json \
  | jq -r '.pipelines[] | select(.name == "intake") | .pipelineId')
export VECTOR_PIPELINE_ID=$(dodil data vector collection list -b kb-intake -o json \
  | jq -r '.pipelines[] | select(.name == "intake_vec") | .pipelineId')
 
# Two rules now exist on this bucket
dodil data ingest list -b kb-intake -o json \
  | jq '.rules[] | {name, pipelineId, includePatterns, includeMimeTypes, enabled}'

3. Upload an intake document

Stage a sample document representing an inbound customer support PDF:


curl -sSL https://example.com/customer-letter.pdf -o customer-letter.pdf 2>/dev/null || \
cat > customer-letter.txt <<'EOF'
Subject: Service interruption — extended outage on May 23
 
Dear Support Team,
 
Our company experienced a complete service outage from May 23 09:00 to May 23 14:30 UTC,
affecting our production billing systems. This is the third such incident in two months.
 
We require:
1. A formal incident root-cause analysis (RCA) within 5 business days
2. Service credits per our SLA (4.5h downtime exceeds 99.9% monthly threshold)
3. A direct contact for escalation: Jane Smith ([email protected], +1-555-0142)
 
Please escalate to your account executive. Our contract: ACME-CONTRACT-2024-INT-091.
 
Sincerely,
Operations Team, Example Corp
EOF
mv customer-letter.txt customer-letter.pdf 2>/dev/null || true
dodil data object create ./customer-letter.pdf -b kb-intake -k inbox/2026-05-27/customer-letter.pdf

Two pipelines fire in parallel.

4. Watch both pipelines complete


# Tables pipeline — produces ONE row per document
dodil data ingest jobs -b kb-intake -p "$TABLE_PIPELINE_ID" -o json \
  | jq '.jobs[] | {pipeline: "triage", object: .object.key, status, rowsWritten}'
 
# Vector pipeline — produces MANY chunks per document
dodil data ingest jobs -b kb-intake -p "$VECTOR_PIPELINE_ID" -o json \
  | jq '.jobs[] | {pipeline: "embed", object: .object.key, status, chunksCreated, embeddingsWritten}'

Different “shapes” per pillar:

Pipeline	Output per document	Latency typical
`document_triage`	1 row (classification + entities + routing decision)	~1–5 s
`text_embedding_index`	N chunks (token-based chunking) → N embeddings	~3–15 s

The triage pipeline finishes first usually; your routing layer can act on triage results before retrieval is fully indexed.

5. Query the triage table — routing logic

After the Tables pipeline drains, the document’s triage row is queryable. It is an ordinary Postgres-wire table — use whatever client you already have:

psql


psql "postgresql://$DODIL_USER:$DODIL_TOKEN@pg.uk-lon-1.dodil.io:5432/kb-intake?sslmode=require" \
  -c "SELECT * FROM intake WHERE source_key = 'inbox/2026-05-27/customer-letter.pdf'"

Python (psycopg)


import psycopg
 
with psycopg.connect(
    host="pg.uk-lon-1.dodil.io", port=5432, dbname="kb-intake",
    user=DODIL_USER, password=DODIL_TOKEN,
    sslmode="require",    # the pg wire terminates TLS; never `prefer`
) as conn, conn.cursor() as cur:
    cur.execute(
        "SELECT * FROM intake WHERE source_key = %s",
        ("inbox/2026-05-27/customer-letter.pdf",),
    )
    print(cur.fetchone())

The Postgres wire is the only door with prepared statements, so %s binding works here and nowhere else.

dodil data


dodil data sql -b kb-intake \
  "SELECT * FROM intake WHERE source_key = 'inbox/2026-05-27/customer-letter.pdf'"

Sample row (template-specific fields):


{
  "source_key": "inbox/2026-05-27/customer-letter.pdf",
  "document_type": "complaint",
  "topic": "service_outage",
  "priority": "high",
  "urgency_score": 0.87,
  "entities": ["ACME-CONTRACT-2024-INT-091", "Jane Smith", "[email protected]", "Example Corp"],
  "routing_team": "tier-2-support",
  "language": "en",
  "extracted_at": 1716840000000000
}

Operational dashboard queries


-- What's incoming today?
SELECT document_type, priority, COUNT(*) AS n
FROM intake
WHERE extracted_at >= DATE_TRUNC('day', NOW())
GROUP BY document_type, priority
ORDER BY priority DESC, n DESC;
 
-- Top topics for tier-2 routing queue
SELECT topic, COUNT(*) AS n
FROM intake
WHERE routing_team = 'tier-2-support'
  AND urgency_score > 0.5
GROUP BY topic
ORDER BY n DESC
LIMIT 10;
 
-- Outliers needing immediate review
SELECT source_key, document_type, urgency_score
FROM intake
WHERE urgency_score > 0.8
  AND extracted_at >= NOW() - INTERVAL '1 hour'
ORDER BY urgency_score DESC;

These power your team-routing layer — e.g. a worker process polls the table every minute and dispatches each new high-urgency document to a queue / Slack / Jira based on routing_team.

6. Query the vector collection — “similar documents”

When a triage decision is made (or an analyst is reading a specific document), the next question is usually “have we seen this before?” Vector handles it:

The route takes text (JSON) or an uploaded file (multipart). It does not take a stored object key — there is no s3Key field. For a document already in the bucket, query by its subject line / body text, or re-upload the bytes as a multipart file query:

dodil data


dodil data search "extended service outage, RCA request, SLA credits" \
  -b kb-intake --table intake_vec --top-k 10

curl (text)


curl -sS -X POST "https://api.data.dodil.io/kb-intake/search/vector" \
  -H "Authorization: Bearer $DODIL_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "text": "extended service outage, RCA request, SLA credits",
    "collectionNames": ["intake_vec"],
    "topK": 10
  }' | jq '.results[] | {
       score,
       object: .object.key,
       chunkIndex,
       content: (.content | .[0:200])
     }'

curl (file)


curl -sS -X POST "https://api.data.dodil.io/kb-intake/search/vector" \
  -H "Authorization: Bearer $DODIL_TOKEN" \
  -F "file=@./customer-letter.pdf;type=application/pdf" \
  -F "collection_name=intake_vec" \
  -F "top_k=10"

This searches the entire intake_vec collection for chunks semantically similar to the query — including chunks of the document itself, if it is already indexed. Filter that out client-side on .object.key. content is on every hit; there is no includeContent opt-in.

Returns previous customer letters with similar themes (outages, RCA requests, escalation patterns) — useful context for the agent / analyst handling the new document.

7. The killer pattern — routing decision + similar-case recall in one workflow

A realistic agent workflow combining both pillars:


import os, psycopg, requests
 
K3 = "https://api.data.dodil.io"
HEADERS = {
    "Authorization": f"Bearer {os.environ['DODIL_TOKEN']}",
    "Content-Type": "application/json",
}
 
PG = dict(
    host="pg.uk-lon-1.dodil.io", port=5432, dbname="kb-intake",
    user=DODIL_USER, password=DODIL_TOKEN, sslmode="require",
)
 
def handle_intake(object_key: str, query_text: str):
    """
    1. Get triage decision from Tables (over the Postgres wire — table SQL is
       the tables-gateway's, not the control plane's).
    2. Find semantically similar past documents from Vector.
    3. Hand decision + context to the routing layer.
    """
 
    # 1. Pull the triage row. No freshness argument exists — reads are
    #    read-your-writes by default, decided by a frontier check.
    with psycopg.connect(**PG) as conn, conn.cursor(
        row_factory=psycopg.rows.dict_row
    ) as cur:
        cur.execute("SELECT * FROM intake WHERE source_key = %s", (object_key,))
        row = cur.fetchone()
 
    if row is None:
        # Triage pipeline hasn't drained yet — retry shortly
        return {"status": "pending"}
 
    decision = {
        "type": row["document_type"],
        "topic": row["topic"],
        "priority": row["priority"],
        "team": row["routing_team"],
        "entities": row["entities"],
    }
 
    # 2. Find similar past documents. The route has no metadata pre-filter, so
    #    over-fetch and drop the self-hit client-side.
    similar = requests.post(
        f"{K3}/kb-intake/search/vector",
        headers=HEADERS,
        json={
            "text": query_text,
            "collectionNames": ["intake_vec"],
            "topK": 20,
        },
    ).json()
 
    context = [
        {"key": h["object"]["key"], "score": h["score"], "preview": h["content"][:200]}
        for h in similar["results"]
        if h["object"]["key"] != object_key
    ][:5]
 
    return {
        "status": "ready",
        "decision": decision,
        "similar_cases": context,
    }
 
# Use it
result = handle_intake(
    "inbox/2026-05-27/customer-letter.pdf",
    "extended service outage, RCA request, SLA credits",
)
print(result)

This is the pattern that makes K3 different from a stack with separate document AI + vector DB products: both signals land from one upload, queryable side-by-side.

8. Operational patterns

Scope similar-case recall to “same team’s queue”

When the routing layer dispatches a document to a team, you want similar-case retrieval scoped to that team’s history, not the whole org. There is no pre-filter on the search route, so this is a SQL narrow + a client-side intersection:


# Step 1: SQL — keys of past tier-2 support docs
dodil data sql -b kb-intake -o json \
  "SELECT source_key FROM intake WHERE routing_team = 'tier-2-support'" \
  | jq -r '.rows[].source_key' | sort > /tmp/tier2.keys
 
# Step 2: over-fetch semantically, then intersect on the object key
curl -sS -X POST "https://api.data.dodil.io/kb-intake/search/vector" \
  -H "Authorization: Bearer $DODIL_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "text": "recurring service outage SLA breach",
    "collectionNames": ["intake_vec"],
    "topK": 200
  }' | jq -r '.results[] | "\(.score)\t\(.object.key)"' \
     | grep -Ff /tmp/tier2.keys | head -5

A cleaner option if you know the split up front: give each team its own collection and pin collectionNames at query time — collection scoping is a request field, unlike metadata filtering.

Replay one pipeline after template update

If document_triage ships a new version (or you re-tune its prompt) but text_embedding_index is unchanged, replay only the Tables side:


TABLE_RULE_ID=$(dodil data ingest list -b kb-intake -p "$TABLE_PIPELINE_ID" -o json | jq -r '.rules[0].ruleId')
 
SOURCE_ID=$(dodil data source list -b kb-intake -o json \
  | jq -r '.sources[] | select(.name == "internal") | .sourceId')
 
# Re-dispatch every object through ONLY the Tables pipeline
dodil data ingest trigger -b kb-intake -s "$SOURCE_ID" --rule "$TABLE_RULE_ID"

The Vector collection is untouched.

Pause incoming docs without losing history


TABLE_RULE_ID=...
VECTOR_RULE_ID=...
 
dodil data ingest update "$TABLE_RULE_ID" -b kb-intake --enabled=false
dodil data ingest update "$VECTOR_RULE_ID" -b kb-intake --enabled=false
# New uploads accumulate in Storage; existing data in Tables + Vector is unchanged.
# Re-enable when ready, then `trigger-discovery --full-sync` to catch up.

Common gotchas

Symptom	Cause	Fix
Triage row exists but `intake_vec` has no chunks for the document	Vector pipeline is slower; check job status	`dodil data ingest jobs -b kb-intake -p $VECTOR_PIPELINE_ID -o json` — wait for the matching object’s job to be `COMPLETED`
`routing_team` is null on a row	`document_triage` couldn’t classify (too short / non-English / empty)	Inspect `errorDetails` on the job; consider pre-filtering at the upload layer for known-bad inputs
Narrowing a semantic query to a large key set is awkward	The search route has no metadata pre-filter — k3-api sends the template a hardcoded empty filter	Give each team/category its own collection and pin `collectionNames`, which is a request field. Otherwise over-fetch and intersect client-side
Re-uploading the same document accumulates rows in `intake`	The merge key isn’t stable across runs	`document_triage` declares `warehouse_mode: "merge"` with `merge_keys: ["input_ref"]`, so a re-run of the same object should merge, not append. If it appends, check that `input_ref` is identical between runs
Tables sees the document but routing decision feels wrong	`document_triage`’s classification confidence isn’t surfaced as a field	Custom-tune the triage by switching to a different template, or hook a downstream Scriptum step that rewrites the decision

Cleanup


TABLE_RULE_ID=$(dodil data ingest list -b kb-intake -p "$TABLE_PIPELINE_ID" -o json | jq -r '.rules[0].ruleId')
VECTOR_RULE_ID=$(dodil data ingest list -b kb-intake -p "$VECTOR_PIPELINE_ID" -o json | jq -r '.rules[0].ruleId')
 
dodil data ingest update "$TABLE_RULE_ID" -b kb-intake --enabled=false
dodil data ingest update "$VECTOR_RULE_ID" -b kb-intake --enabled=false
 
dodil data ingest delete "$TABLE_RULE_ID" -b kb-intake
dodil data ingest delete "$VECTOR_RULE_ID" -b kb-intake
# Both destinations are pipelines — `pipeline delete` takes either id
dodil data pipeline delete "$TABLE_PIPELINE_ID" -b kb-intake
dodil data pipeline delete "$VECTOR_PIPELINE_ID" -b kb-intake
 
# The table rows survive the pipeline; drop them separately
dodil data table delete intake -b kb-intake
 
dodil data bucket delete kb-intake

Document Intake — Triage + Semantic Search

Prerequisites

1. Browse the two templates

2. Create the two destinations

A. Pipeline-bound table for triage

B. Pipeline-mode vector collection for retrieval

3. Upload an intake document

4. Watch both pipelines complete

5. Query the triage table — routing logic

psql

Python (psycopg)

dodil data

Operational dashboard queries

6. Query the vector collection — “similar documents”

dodil data

curl (text)

curl (file)

7. The killer pattern — routing decision + similar-case recall in one workflow

8. Operational patterns

Scope similar-case recall to “same team’s queue”

Replay one pipeline after template update

Pause incoming docs without losing history

Common gotchas

Cleanup

See also