indexes — Source-controlled vector store sync (delete, update, upsert)

Source-controlled vector store sync (delete, update, upsert)

Module langchain_classic/indexes/ — Source-controlled vector store sync (delete, update, upsert). Part of the Retrieval & Data group.

🎯 Why This Module Exists

This module is documented from the indexes_deep.md Phase 1B research artifact. See sections below for design decisions, algorithms, error philosophy, performance, and evolution clues.

Source files

Sub-directories

Retrieval & Data

Architecture

flowchart LR
  subgraph Core["langchain_classic/indexes/"]
    BASE["Base abstraction"]
    IMPLS["Concrete implementations"]
    BASE --> IMPLS
  end
  subgraph Up["Upstream callers"]
    U1["chains/"]
    U2["agents/"]
  end
  subgraph Down["Downstream"]
    D1["langchain_core"]
    D2["External providers"]
  end
  Up --> Core
  Core --> Down

Data Flow

sequenceDiagram
  participant Caller as Caller
  participant Indexes as indexes
  participant Core as langchain_core
  participant Ext as External
  Caller->>Indexes: invoke(input)
  Note over Indexes: Validate & route
  Indexes->>Core: delegate primitive
  Core->>Ext: provider call
  Ext-->>Core: response
  Core-->>Indexes: typed output
  Indexes-->>Caller: result

Existence rationale

indexes/ is the home of incremental ingestion into a vectorstore. The naïve approach — wipe and re-embed everything on every refresh — is slow and expensive. The index() algorithm uses a record manager (a SQL table of (key, hash, source_id, updated_at)) to compute the diff between what we have and what should be there, and emits only the deltas as adds and deletes. This is the difference between a 5-second nightly refresh and a 5-hour one.

Design decisions

Decision	Choice	Alternative	Rationale
Record manager backed by SQL	`SQLRecordManager` (SQLAlchemy)	In-memory; vectorstore-side metadata	Persistent across processes; doesn't depend on vectorstore having metadata filtering
Hash-based change detection	`hash(page_content + metadata)` keyed by `(source_id, key)`	Timestamp-only	Detects content edits even when source mtime is stale
Source-aware cleanup	`cleanup="full"` deletes everything not present in this run; `"incremental"` deletes only stale rows from the same `source_id`; `None` never deletes	One mode	"full" is for one-shot rebuilds; "incremental" is for periodic per-source refreshes
Batch upsert	Documents processed in `batch_size` chunks	One-by-one	Vectorstore embeddings amortise across batches
Deterministic IDs	`key` is hash-derived → upsert is idempotent	Random UUIDs	Re-running the same indexing job is a no-op rather than a duplicate flood

Algorithm deep-dive — `index()`

Pseudocode.

existing = record_manager.list_keys(source_ids=[source])  # set of (key, hash)
new_keys = set()
for batch in batches(docs, batch_size):
    for doc in batch:
        key = hash(doc.page_content)
        if (source, key) not in existing or existing[key].hash != current_hash(doc):
            vectorstore.add_documents([doc], ids=[key])  # upsert
            record_manager.update([(key, source, current_hash(doc), now)])
        new_keys.add(key)

if cleanup == "full":
    stale = record_manager.list_keys() - new_keys
elif cleanup == "incremental":
    stale = record_manager.list_keys(source_ids=[source]) - new_keys
else:
    stale = set()
vectorstore.delete(stale)
record_manager.delete_keys(stale)
return {"num_added": ..., "num_updated": ..., "num_deleted": ..., "num_skipped": ...}

Edge cases.

Two docs with identical content: same key → second is a no-op upsert; not a duplicate.
Source that produces fewer docs than last run: incremental cleanup deletes the missing ones; None leaves them (orphans).
Vectorstore that doesn't support deterministic IDs: index() raises with a clear error.

Performance characteristics

Best case (no changes): O(N_existing) record-manager read + zero vectorstore work.
Worst case (full rewrite): O(N_docs) embed + upsert + delete.
Bottleneck: embedding throughput (the vectorstore is usually fast; the embedding API is the queue).

Evolution clues

SQLRecordManager started as the only backend; an interface (RecordManager ABC) and an aindex async variant followed. The cleanup modes accreted over time as users hit different ingestion patterns (one-shot rebuild vs source-by-source refresh).

Step-by-Step Walkthrough

Follow a typical operation through the indexes/ module:

Caller invokes the public surface — usually .invoke(input) on a Runnable, or a legacy __call__(inputs_dict) on a Chain subclass.
Pydantic validation — every entry point is a BaseModel; constructor args, input dicts, and tool args_schema are validated before any expensive work runs. Fail-fast on type errors.
CallbackManager.configure — a fresh manager is built per call, fans on_*_start events to all subscribers (LangSmith tracer, stdout handler, custom logging), and stamps a UUID4 run_id for tree reconstruction.
indexes-specific logic — the module's core method runs: prompt rendering, retrieval, tool execution, memory load, document split, etc. See the deep-dive sections above.
Provider call — typically a single HTTP request to OpenAI/Anthropic/etc., or a local SDK call. Retries are handled by tenacity with exponential backoff and a configurable max-attempts knob.
Streaming fan-out — if the provider streams tokens, each chunk fires on_llm_new_token through the callback manager so downstream observers (UI, log, tracer) see partial output.
OutputParser pass — if a parser is attached, the raw provider output is parsed into a structured Python object; OutputParserException may trigger a retry chain (RetryWithErrorOutputParser).
Memory write-back — for chains/agents that have a BaseMemory attached, save_context(inputs, outputs) persists the turn before returning to the caller.
on_*_end fan-out — the callback manager closes the run span; tracers serialise the Run tree and ship it to LangSmith asynchronously.
Result returned — caller receives a typed object (AIMessage, dict, BaseModel, list[Document], ...) matching the module's declared output schema.

What Could Go Wrong

Failure Mode	Trigger	Detection	Recovery
Provider 5xx / rate-limit	OpenAI/Anthropic/etc. transient error	HTTP status from SDK	tenacity retries with exponential backoff up to `max_retries`; then surface to caller
Output parser failure	LLM returned malformed JSON / wrong schema	`OutputParserException` in parser	`RetryWithErrorOutputParser` re-prompts with the parser error included; or surface
Pydantic validation	Caller passes wrong-shaped dict	`ValidationError` at `__init__` / `invoke`	Surface immediately — no fallback is correct here
Callback handler exception	User-supplied tracer raises	Try/except around each fan-out	Logged and swallowed by default; `raise_error=True` to opt in to propagation
Provider auth missing	`OPENAI_API_KEY` unset	SDK `AuthenticationError`	Surface with a helpful message; no retry (would just fail again)

Performance Profile

Hot Paths

Provider HTTP round-trip dominates wall-clock time (100ms–30s depending on model and prompt length).
Pydantic validation is repeated on every invoke; cached schema parsing keeps it under 1ms for typical inputs.
Callback fan-out is synchronous by default — a slow handler will slow every call. Async tracers (LangSmith) push work to a background thread to avoid blocking.

Resource Usage

Memory: token-history buffers (in memory/) and document caches (in retrieval pipelines) are the largest steady-state allocations.
I/O: bounded by provider rate limits; vector store I/O for retrieval-flavored modules.
Concurrency: async support via a* methods; agents and parallel chains use asyncio.gather for concurrent tool calls.

Directory Map

Source tree for indexes

indexes/
├── __init__.py
├── _api.py
├── _sql_record_manager.py
├── graph.py
├── prompts/
└── vectorstore.py