Agentsop Idempotent Ingestion

Idempotent Ingestion · Re-Ingest-Correctness SOP

Third-person operating model for a coder agent that owns ingestion correctness across repeated runs. The audience is the LLM agent writing or reviewing the pipeline code — not the end user.

One sentence: The interesting run is the second one. A correct pipeline hashes each document, looks the hash up in a docstore, and inserts / updates / skips accordingly — so re-running over unchanged docs changes nothing.

This skill is an ENHANCE overlay over [[llamaindex]]. The base skill names IngestionPipeline with docstore + UPSERTS_AND_DELETE as a hardening must-have (OP-08, anti-pattern A10) but stops at "use it". This overlay surfaces the correctness contract — what idempotency actually means, how the hash decides the action, and how it generalises beyond LlamaIndex.

1. 何时激活 (Activation Rules)

Activate this skill whenever any of the following holds:

1. The codebase contains an ingestion call (VectorStoreIndex.from_documents, pipeline.run(documents=...), vectorstore.add_documents, a custom embed
   • upsert loop) and that call will execute more than once over a corpus that can change between runs.
2. The user mentions any of: re-ingest, re-index, scheduled / nightly index refresh, incremental document updates, docstore, doc hash, upsert, RecordManager, "keep the index in sync with the source".
3. A bug report says "I see the same answer chunk twice", "retrieval returns duplicates", "deleted a file but it still shows up in answers", "the index keeps growing every night even though nothing changed".
4. PR review: new ingestion code that builds the index with no docstore / no hash-keyed dedup and the corpus is live (LlamaIndex base skill A10).
5. A production RAG is about to ship and the ingestion side has only ever been tested as a cold first run.

Do not activate when:

• The index is built once from a static corpus and never refreshed — there is no second run to make idempotent.
• The corpus is tiny and rebuilding from scratch each run is measurably cheaper than maintaining a docstore (rare; verify, don't assume).
• The task is purely query-time (retrieval, rerank, synthesis) with no ingestion path in scope.

2. 核心心智模型 (Core Mental Model)

Three principles. Violating any one means a second run can duplicate or corrupt the index — regardless of how good the retrieval stack downstream is.

Principle 1 — Ingestion must be idempotent; the hash is the decision

Re-running the pipeline over an unchanged document is a no-op. Run it once, run it a thousand times — same index. The mechanism: each document is reduced to a stable content hash (and a stable doc id); that hash is looked up in a docstore (a key→hash record store separate from the vector store); the lookup result decides the action.

            doc_id seen?           hash changed?
                │                       │
   ┌────────────┴───────────┐    ┌──────┴──────┐
   no                       yes  no            yes
   │                         │   │              │
INSERT (embed + upsert)      │  SKIP        UPDATE
                             └──────────────►(delete old chunks,
                                              re-embed, upsert)

The hash is the whole game. Without it, the pipeline cannot tell "I have seen this exact content" from "this is new" — so it re-embeds and re-adds everything, and the vector store accumulates duplicates. (LlamaIndex base skill failure #3 / A10; developers.llamaindex.ai/.../loading/ingestion_pipeline/.)

Principle 2 — The docstore is a separate ledger, not the vector store

The vector store holds embeddings keyed by node id. The docstore holds, per document, {doc_id → content_hash} (and the node ids derived from it). They are two stores with two jobs:

The dedup decision happens against the docstore, before any embedding call — so an unchanged corpus costs zero embedding tokens on re-run. If the docstore is ephemeral (in-memory, lost on restart), every cold start looks like a first run and re-embeds the world. The docstore must be persisted to the same durability tier as the vector store (SimpleDocumentStore.persist(), Redis, MongoDB, Postgres). A docstore that doesn't survive the process is not a docstore.

Principle 3 — Deletes don't propagate for free

Insert and update are driven by documents that are present. A document that disappeared from the source emits no event — so a naive upsert pipeline never learns it should remove the orphaned chunks. Stale content keeps getting retrieved long after the source file is gone. Propagating deletes requires the pipeline to compare the full set of doc_ids seen this run against the set in the docstore, and purge the difference. In LlamaIndex this is the _AND_DELETE half of DocstoreStrategy.UPSERTS_AND_DELETE; in LangChain it is cleanup="full" / cleanup="incremental". Choosing the upsert-only strategy silently accepts stale ghosts — sometimes correct (append-only corpus), usually not.

3. SOP 工作流 (Agentic Protocol)

Five stages. Each gates the next. Stop and remediate at the first failure. The gate that matters most is Stage 4 — prove the second run is a no-op.

Stage 0 — Frame the re-run

Before code, answer:

1. Run cadence: how does the pipeline get re-triggered? (cron / webhook / CI / manual). If the honest answer is "never, it's one-shot" → this skill is over-engineering; stop.
2. Change shape: do source docs get added only, or also edited and deleted? This decides upsert-only vs upsert-and-delete (Principle 3).
3. Doc identity: what is the stable doc_id? It must survive across runs — a file path, a CMS id, a URL. Not a random uuid generated at load time (that makes every run look new). LlamaIndex derives a hash from content + doc_id; if doc_id is unstable, idempotency breaks even with a docstore.
4. Durability tier: where does the docstore live so it survives restarts?

Stage 1 — Attach a persisted docstore

from llama_index.core.ingestion import IngestionPipeline, DocstoreStrategy
from llama_index.core.storage.docstore import SimpleDocumentStore
from llama_index.core.node_parser import SentenceSplitter
from llama_index.embeddings.openai import OpenAIEmbedding

pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=1024, chunk_overlap=20),
        OpenAIEmbedding(model="text-embedding-3-small"),
    ],
    docstore=SimpleDocumentStore(),          # the dedup ledger
    vector_store=vector_store,               # the embedding store
    docstore_strategy=DocstoreStrategy.UPSERTS_AND_DELETE,
)

For production, swap SimpleDocumentStore for a server-backed one (RedisDocumentStore, MongoDocumentStore, PostgresDocumentStore) so the ledger is shared across workers and survives restarts. A SimpleDocumentStore that is never .persist()-ed is the #1 cause of "it re-embeds everything every night".

Stage 2 — Pin stable doc ids and let the pipeline hash

docs = SimpleDirectoryReader("./data", filename_as_id=True).load_data()
# filename_as_id=True → doc.id_ is the file path → stable across runs.
# LlamaIndex then computes the content hash internally per node.
pipeline.run(documents=docs)

Hard rules:

• doc_id comes from a stable external identity (path / CMS id / URL), never a fresh uuid per load. Set filename_as_id=True or assign doc.id_ explicitly.
• Do not mutate document text in a non-deterministic transformation before the hash is taken (e.g. injecting datetime.now() into metadata) — that makes the hash change every run and defeats the skip path. Keep volatile metadata out of the hashed payload.
• The embedding model is part of the index identity: changing it requires a full re-embed (LlamaIndex base skill A2), not an incremental upsert.

Stage 3 — Choose the upsert strategy deliberately

Default for any system that mirrors a mutable source: UPSERTS_AND_DELETE. DUPLICATES_ONLY is not a re-ingest strategy — it only de-dupes within a single .run(); a second .run() with the same docs will not skip them.

Stage 4 — Prove the second run is a no-op (the gate)

Before merging, the PR must include — and CI must run — a test that runs the pipeline twice and asserts the second run did nothing observable:

def test_reingest_is_idempotent(pipeline, docs, vector_store):
    n1 = len(pipeline.run(documents=docs))      # cold run
    count_after_first = vector_store.count()

    n2 = len(pipeline.run(documents=docs))       # identical re-run
    count_after_second = vector_store.count()

    assert n2 == 0,                       "re-run re-processed unchanged docs"
    assert count_after_second == count_after_first, "re-run duplicated vectors"

pipeline.run() returns the list of nodes it actually processed; on a clean idempotent re-run over unchanged input that list is empty. If it isn't, the docstore is missing, ephemeral, or the doc ids / hashes are unstable — go back to Stage 1-2. Then add the edit case (change one doc → exactly that doc's chunks replaced) and, for UPSERTS_AND_DELETE, the delete case (remove one doc → its chunks gone, others untouched). See OP-06 / OP-07.

Stage 5 — Schedule, persist, observe

• Persist the docstore after every run (storage_context.persist() / server-backed store auto-persists) — otherwise the next cold start re-embeds everything.
• Run on a schedule, not by hand (LlamaIndex base skill OP-08): cron / Airflow / serverless cron. The whole point of idempotency is that a missed-or-doubled trigger is harmless.
• Log per run: {docs_in, inserted, updated, skipped, deleted, embedding_calls}. On a steady-state corpus, inserted+updated+deleted should trend to ~0 and skipped ≈ docs_in. A run that re-embeds everything every night is the loud signal that idempotency is broken.

4. 操作模型 (Operation Models)

Format: Trigger / Action / Output / Evidence. Machine-readable in intermediate/operation_candidates.json.

OP-01 AttachDocstore

• Trigger: any IngestionPipeline / ingestion path that will run more than once with no docstore wired.
• Action: pass docstore= (persisted: Redis/Mongo/Postgres in prod, SimpleDocumentStore only for tests) plus docstore_strategy=. The docstore is the dedup ledger; without it dedup is structurally impossible.
• Output: pipeline that can answer "have I seen this content?" before embedding.
• Evidence: developers.llamaindex.ai/python/framework/module_guides/loading/ingestion_pipeline/; [[llamaindex]] OP-08 / A10.

OP-02 PinStableDocId

• Trigger: docs loaded with random / per-run ids; "re-run re-adds everything" symptom.
• Action: set filename_as_id=True or assign doc.id_ from a stable external identity (path / CMS id / URL). Never a fresh uuid per load.
• Output: the same source document hashes to the same key across runs → the skip path can fire.
• Evidence: LlamaIndex Document.doc_id / SimpleDirectoryReader(filename_as_id=True) docs.

OP-03 HashBeforeEmbed

• Trigger: designing or reviewing the dedup decision point.
• Action: ensure the content hash is computed and looked up in the docstore before the embedding transformation runs, so unchanged docs cost zero embedding tokens. In LlamaIndex this is automatic given a docstore; in a manual pipeline, hash first, embed only on miss/change.
• Output: re-run cost on a steady corpus ≈ 0 embedding calls.
• Evidence: IngestionPipeline dedup section; LangChain indexing API ("avoids re-writing unchanged content, avoids re-computing embeddings").

OP-04 PickUpsertStrategy

• Trigger: choosing pipeline behaviour for the corpus's change shape.
• Action: UPSERTS (append-mostly), UPSERTS_AND_DELETE (mirror of a shrinkable source — the production default), DUPLICATES_ONLY (in-run dedup only, not cross-run). Decide from Stage 0 change shape, not by default.
• Output: documented strategy + one-line rationale.
• Evidence: DocstoreStrategy enum; LlamaIndex ingestion docs.

OP-05 PropagateDeletes

• Trigger: source docs can be deleted and stale chunks must not be retrieved.
• Action: use UPSERTS_AND_DELETE; the pipeline diffs the doc_ids seen this run against the docstore and purges orphans. (LangChain: cleanup="full" for a full snapshot run, cleanup="incremental" for streamed batches.)
• Output: deleted source docs → their vectors removed; retrieval stays in sync.
• Evidence: DocstoreStrategy.UPSERTS_AND_DELETE; LangChain index(..., cleanup=...) docs.

OP-06 ReingestNoOpTest

• Trigger: any ingestion code change merges.
• Action: run the pipeline twice over identical input; assert the second run() returns 0 processed nodes and the vector count is unchanged (Stage 4).
• Output: regression gate proving idempotency is wired through.
• Evidence: pipeline.run() returns processed nodes; direct application of Principle 1.

OP-07 IncrementalUpdateTest

• Trigger: corpus supports edits and/or deletes.
• Action: edit one doc → assert exactly that doc's old chunks are gone and new ones present, others untouched. Delete one doc (under UPSERTS_AND_DELETE) → assert its chunks gone, neighbours intact.
• Output: proof that update/delete are surgical, not full-rebuild.
• Evidence: hash-change → update path; delete path of UPSERTS_AND_DELETE.

OP-08 PersistDocstore

• Trigger: SimpleDocumentStore used, or docstore not persisted after run.
• Action: storage_context.persist(persist_dir=...) after each run, or use a server-backed docstore that auto-persists and is shared across workers.
• Output: cold start reads the ledger → does not re-embed the world.
• Evidence: LlamaIndex storage / persistence docs; Redis/Mongo/Postgres docstore integrations.

OP-09 ManualHashUpsert

• Trigger: no framework available (raw vector-store SDK, custom loader).
• Action: maintain a {doc_id → sha256(content)} table next to the vector store; on each doc compute the hash, compare, and INSERT / UPDATE (delete-then-add chunks) / SKIP; track seen ids for delete propagation.
• Output: framework-free idempotent ingestion with the same contract.
• Evidence: this is what IngestionPipeline / RecordManager automate; see R2 cross-framework table.

5. 困境决策案例 (Dilemma Cases)