Document ingestion pipeline

I keep an ingest manifest with doc_id, content hash and status. Each scheduled run lists the source, hashes every file and compares it with the manifest. Unchanged files are skipped. For changed files I delete all chunks with that doc_id in the vector store, then upsert the new chunks using deterministic ids made from doc_id and chunk index, so a rerun after a crash just overwrites the same ids. Files missing from the source trigger a delete by doc_id. After the write succeeds I update the manifest. I also run a periodic reconciliation that compares the manifest with the ids actually in the store, which catches orphans from partial failures. That gives incremental cost, idempotent reruns and no stale answers.

An external scheduler such as cron, a CI job or an orchestrator fires on a fixed interval. It calls POST /api/v1/run/<flow-id> with an API key and tweaks for the source path, target collection and a reindex flag. Inside Langflow, a custom change-detector hashes source files and compares them with the ingest manifest. It passes new or changed files forward and emits deletes for removed ones. File loaders read the text, and a custom metadata component attaches doc_id, path, hash, model name and access tags. The text splitter creates chunks with deterministic ids. The embeddings component embeds them in batches, and the vector store component deletes old chunks per doc_id and then upserts the new ones. Finally the manifest is updated, a run report goes to the team, and traces are kept in LangSmith or Langfuse.

Modified time tells you a file was touched, not that its content changed. A copy, a checkout or a sync tool can reset timestamps and trigger expensive re-embedding of identical text. Clocks can also be wrong or get preserved during a copy, so a real change could be missed. A content hash changes only when the bytes change, which makes the decision deterministic and repeatable. The manifest stores the last successfully ingested hash per document, so comparing the two gives three clear sets: new, changed and unchanged. Removed files show up as entries in the manifest that no longer exist in the source. I still record modified time as metadata, because it is useful for display and for freshness filters.

I treat access tags as a required field, not a nice-to-have. The custom metadata component looks up tags from the source system or a mapping file. If a document has no resolvable tag, it fails closed: the document is marked failed or skipped and is never upserted untagged. The splitter must copy document metadata onto every chunk, and just before the upsert I add a validation step that checks every chunk for doc_id, content hash, embedding model and tenant or access tags. I also add a test fixture that asserts tags survive splitting. Retrieval flows should always apply a tag filter, so a missing tag hides content instead of exposing it. The run report counts these rejections so someone fixes the mapping.

Upsert by deterministic id overwrites chunks 0 to 6. Chunks 7, 8 and 9 from the old version are never touched, so they stay in the index as stale content and can still be retrieved. That is why the vector store step first deletes all chunks matching the changed doc_id and then upserts the new set. An alternative with a smaller gap is to upsert the new chunks first and then delete any chunk for that doc_id whose id is not in the new set, or whose content hash differs from the new hash. Both approaches depend on doc_id being stored as filterable metadata on every chunk, which is another reason that metadata must never be lost.

Langflow flows are run through the API, so I keep scheduling outside the flow. A cron job, CI pipeline or orchestrator calls POST /api/v1/run/<flow-id> with an API key. I use tweaks to override component settings per run, such as the source path, the target collection name, or a flag that forces a full reindex. That lets one flow on the canvas serve several knowledge bases without duplicating it. A webhook works when the source system can push change events instead of polling. I add a lock so two runs never overlap on the same collection, and I send the run report counts to the team so a silent failure is noticed quickly rather than discovered by users.

I attach metadata in a custom component before the text splitter so every chunk inherits it. The core fields are doc_id, source path or URL, title, document type, last modified time or version, content hash and the embedding model name. The ids and hash drive sync and deletion. Source and title let the RAG flow cite where an answer came from. Type, date and department let retrieval narrow the search, for example only current HR policies. The most important fields are access or tenant tags, because retrieval filters on them to stop users seeing documents they should not see. I treat those as required and reject a document rather than index it without them, since a missing tag is a data leak, not a cosmetic issue.

With delete-then-upsert, that document has no chunks until the next successful run, so retrieval silently misses it. The key protection is that the manifest is updated only after the upsert succeeds. On the next run, the hash still differs from the manifest, so the document is reprocessed and the state heals itself. To shrink the gap, I can reorder the steps: upsert the new chunks, then delete old chunks for that doc_id that are not in the new id set. Readers briefly see the new content plus a few leftovers instead of nothing. For strict needs, I build a fresh collection and switch retrieval to it only after validation.

Per document, the manifest stores doc_id, source path, the last ingested content hash, the embedding model name, the chunk count or chunk ids, the status (ingested, skipped or failed), an error reason and a last run timestamp. It must be durable and outside the flow's memory, because each run is stateless. A small relational table or a key-value store works well. A JSON file in object storage is fine for small corpora, but it needs care with concurrent writers. The manifest is the source of truth for what the pipeline believes is indexed, so updates should happen per document after a successful upsert, not once at the start of the run.

Several design choices together make reruns safe. Deterministic chunk ids built from doc_id and chunk index mean an upsert replaces rather than duplicates. The delete for a doc_id is safe to repeat. The change detector compares against the manifest, so a document finished in an earlier attempt shows as unchanged and is skipped. The manifest is written only after success, so partial work is redone rather than lost. The remaining risk is two runs at the same time interleaving deletes and upserts on one document. I prevent that with a lock, such as a row in the manifest database or the orchestrator's single-concurrency setting, and a second trigger either exits or waits.

I start from how the content will be retrieved and answered. Chunks should be large enough to hold a complete idea, such as a section or a procedure step, and small enough that one retrieved chunk is mostly relevant. They must also fit within the embedding model's input limit. Overlap keeps sentences that cross a boundary findable from either side, but it increases storage and embedding calls. I would not guess: I build a small evaluation set of real questions with known answer passages, try two or three size and overlap settings, and measure retrieval hit rate. The chosen values become part of the configuration, and changing them means a full reindex.

The embeddings component sends chunks in batches, and transient errors such as rate limits or timeouts get retries with backoff. If a batch still fails, its documents go onto a failed-docs list with the error, and their manifest entries are not updated to the new hash. That means the next run sees them as changed and tries again. I avoid upserting part of a document: if any of its chunks failed, the whole document is treated as failed, so the index never holds a half-updated version. The run report shows the failed count, and an alert fires if the same document fails repeatedly, which usually points to bad input rather than a flaky provider.

The Langflow API key lives in the scheduler's secret store, such as CI secrets or the orchestrator's vault, and never in the repo or the flow export. I use a dedicated key for this job so it can be rotated or revoked on its own. The Langflow server should not be open to the public internet. I put it behind a private network or gateway with TLS. Tweaks are powerful because they can change the source path or target collection, so I restrict the allowed values in the custom components, for example with an allow-list of source roots and collections. Credentials for the vector store and embedding provider are kept as Langflow secrets or environment variables, not in plain component fields.

With a pure content hash, it won't notice, and that is a security bug, because retrieval would keep filtering on the old tags. I fix this by computing a second fingerprint over the metadata that matters, especially access and tenant tags, and storing it in the manifest. The change detector treats a change in either the content hash or the metadata fingerprint as a change. For a metadata-only change I could update metadata in place if the vector store supports it. The simpler and safer path is to rerun the delete and upsert for that doc_id. If permissions change often, an event-driven trigger from the permission system is better than waiting for the next scheduled run.

Every chunk carries a tenant id and access tags in its metadata, set by the metadata component and checked before the upsert. Retrieval flows must always apply a filter on tenant and permitted tags. That filter comes from the authenticated user's context, never from free text in the prompt. For stronger isolation, or for tenants with contractual requirements, I use separate collections per tenant and pass the target collection name through tweaks. Separate collections cost more to operate but make cross-tenant leaks much harder and make deleting one tenant's data simple. Either way, I add a test that queries as tenant A and asserts that no tenant B chunks come back.

First, I build in the canvas and run it in the Playground against a small fixture folder. The folder covers a new file, an edited file, a deleted file, an empty file, an unsupported format and a document without access tags. I check the vector store directly: the expected chunk counts, no orphan ids and metadata present on every chunk. Then I run it twice to prove idempotency, because the second run should report zero changes. I call the API with tweaks pointing at a staging collection so production is never touched. The custom components are Python classes, so I unit test the change-detector and metadata logic outside Langflow. Traces in LangSmith or Langfuse help me inspect each step.

One bad file should not block thousands of good updates. If the run failed on the first odd PDF, the whole knowledge base would go stale until someone intervened. Recording the file as skipped with a reason keeps the run going and gives a clear list to act on. The skip is written to the manifest, so it is visible and is not retried every night unless the file's hash changes. The risk is silent gaps, so the skipped count appears in the run report, and a sudden jump in skips, for example after a source system starts exporting a new format, should raise an alert.

The doc_id should be stable for the life of a document. If the source system offers a stable id, such as a CMS page id or a drive file id, I use it. If I only have paths, the doc_id is derived from a normalised path, which means a rename or move looks like a delete plus a new document. That is still correct: the old chunks are deleted and new ones are created. The cost is re-embedding unchanged text. If renames are common, I can detect them by matching the content hash of a vanished file with that of a new file and reuse the embeddings. I would add that only if the cost justifies it.

I estimate from our own measurements, not assumptions. The work scales with changed content, not total corpus size. Embedding calls are roughly the number of changed chunks divided by the batch size, and the chunk count is the changed text length divided by the chunk size minus overlap. I measure average latency per batch and the provider's rate limit, and those give the wall-clock time. Cost is the number of tokens embedded times the provider's current published rate. Vector store writes scale with upserted plus deleted chunks. For a full reindex I apply the same formula to the whole corpus, which is why reindexes are planned rather than accidental.

The incremental design already helps, because a normal run touches only changed files. The bottlenecks become listing and hashing the source, embedding throughput and vector store write rate. For listing, I use the source system's change feed or modified-since queries to avoid hashing everything, and then hash only the candidates. For throughput, the scheduler can shard the work by calling the flow several times with tweaks for different source paths. Each shard has its own lock, and all of them write to the same manifest. Batches are sized to provider limits. A full reindex goes into a new collection in the background, so the live index stays usable.

There are three layers. Traces in LangSmith or Langfuse show each component's inputs, outputs, timings and errors, so I can see exactly where a document failed. The run report summarises added, updated, deleted, skipped and failed counts and is sent to the team's channel. Alerts fire on conditions such as the run not happening at all, failed counts above a threshold, the same document failing repeatedly, or a deleted count that is suspiciously large. That last one often means the source mount was empty, and in that case the pipeline should stop before deleting rather than wipe the index.

If the source path is empty because a mount failed or credentials expired, the change detector would see every manifest entry as deleted. I add a safety check: if the delete set is above a threshold, either a percentage of the corpus or an absolute count, the run stops before any deletion and alerts. A source listing error is treated as a failure, not as an empty folder. Large deletes then require an explicit tweak, such as an allow-mass-delete flag, on a manual run. I also rely on the manifest and versioned collections so a bad run can be reversed.

Even with careful logic, drift happens through manual edits, partial failures or bugs. I run a periodic reconciliation job, often weekly, that lists distinct doc_ids and their content hashes from the vector store metadata and compares them with the manifest. Doc_ids found in the store but not in the manifest are orphans to delete. Manifest entries with no chunks are re-queued. Chunks whose hash or embedding model differs from the manifest are stale and get reingested. This depends on doc_id, content hash and model name being stored on every chunk. The job reports its findings before fixing anything, so a bug in it cannot wipe the index.

Retrieval flows query the same vector store collection and pass metadata filters alongside the similarity search. The tenant and access tags restrict results to what the user may see. Document type or source path narrows the search to, say, policies only. Version or modified time can prefer the latest content. Title and source path are returned with each chunk so the answer can cite its source. The embedding model name lets a retrieval flow confirm it is embedding the query with the same model. If ingestion drops a field, the retrieval filter either leaks data or returns nothing, which is why metadata is validated before the upsert.

I trigger a full reindex when the meaning of existing vectors or chunks changes. That covers a new embedding model, new chunk size or overlap, a change to the metadata schema that retrieval depends on, a splitter or loader fix that changes extracted text, and recovery after confirmed drift or corruption. The scheduler passes the full-reindex flag as a tweak, and the change detector then treats every file as changed. I prefer writing into a new target collection, also set by tweak, validating it and then pointing retrieval at it. That avoids a period where the live index mixes old and new chunks.

A schedule is simple, predictable and easy to rate-limit. The full change detection also catches anything missed, such as events that were never sent. The cost is freshness: an edit waits until the next run, which matters for permission changes. Webhooks, which Langflow supports, give near-real-time updates per document, but you must handle bursts, duplicate or out-of-order events, and missed events. A common design combines them. Webhooks handle urgent single-document updates, especially access changes, and a scheduled run reconciles everything. Both paths share the same components and deterministic ids, so they converge on the same state.

First, I decide policy with the data owner: exclude these sources, redact the content, or index it with tight access tags. Exclusion is handled in the change detector through allow-lists of paths or document types. Redaction belongs in a custom component placed before splitting, so raw PII is never embedded or stored as chunk text. Remember that the embedding provider receives the text, so data residency and the provider's retention terms matter. Traces in LangSmith or Langfuse can also capture document text, so I check what is logged and mask it or reduce the detail. Deletion requests must flow through the same delete-by-doc_id path.

I use built-in loaders, the text splitter, embeddings and vector store components wherever they fit, because they are maintained and easy to read on the canvas. I write custom components, which are Python classes with defined inputs and outputs, for logic specific to our pipeline. Here that means the change detector with manifest comparison and the metadata component with tag lookup and validation. Custom code should be small, unit tested and free of hard-coded secrets. I keep it in version control, because the flow and its components are part of the pipeline's contract, not just canvas settings.

Large files are streamed or split by the loader where possible, and they produce many chunks, so I cap the chunks per batch to stay within the provider's request and token limits. Batches run with bounded concurrency, and on rate-limit responses they back off with jitter instead of retrying immediately. A single huge file should not starve the rest, so I can process it in its own batches and give it a per-document time budget. If it exceeds that budget, it goes to the failed-docs list. Very large files often signal exports or logs that do not belong in the knowledge base, so size limits with a skip reason are reasonable.

Rollback is easiest when ingestion never overwrites the only good copy. For large or risky changes, I write into a new target collection set through tweaks, validate it with test queries and counts, and then switch retrieval to it while keeping the old collection for a while. For routine incremental runs, the source files themselves are the backup. I revert the bad source change or the bad component version, then rerun for the affected doc_ids, or do a full reindex into a fresh collection. Keeping flow exports and custom components in version control lets me redeploy the previous pipeline exactly.

Langflow can act as an MCP server and expose flows as tools, so it is technically easy. I would be careful, though. Ingestion writes to and deletes from the knowledge base, so an agent calling it with a wrong source path or the reindex flag could do real damage. If I expose it, I expose a narrow version: reingest a single named document, restricted to allowed sources and collections, with no mass delete, plus authentication and logging. Read-only tools such as checking a document's ingest status are safer to share. The scheduled job stays the main path, called directly through the API.

The manifest is what the pipeline believes. The chunk metadata is what is actually in the index. Storing the hash on each chunk lets a reconciliation job detect stale chunks whose hash no longer matches the current document. It also lets the upsert-then-delete approach remove leftovers by hash. Storing the embedding model name lets you find vectors from an old model during a migration, and lets retrieval check that the query is embedded with a matching model. Without these fields, verifying the index means re-reading and re-embedding everything, which defeats the point of incremental ingestion.

Message history in Langflow gives per-session chat memory, which suits conversational flows where a session id ties turns together. This flow is a batch job triggered by a scheduler, not a conversation. Each run is driven by its tweaks and by durable state in the manifest and the vector store. Using chat memory as state would be wrong, because it is not designed as a reliable ledger of what was ingested. Keeping the flow stateless between runs, with all persistent state in the manifest, makes it restartable, easy to shard and easy to reason about.

1. APOST /api/v1/run/<flow-id> with an API key and optional tweaksCorrect answer
2. BGET /api/v1/flows/<flow-id>/start with no authentication
3. CA chat message sent to the Playground
4. DA direct write into the vector store that Langflow then detects

Why: Flows run through the API at POST /api/v1/run/<flow-id>, authenticated with an API key, and tweaks can override component settings.

1. AStoring chat memory per session
2. BEncrypting metadata before it is stored
3. CPermanently editing the saved flow graph
4. DOverriding component settings for a single run, such as source path or target collectionCorrect answer

Why: Tweaks override component settings for that run only, so one flow can serve different sources, collections or a reindex flag.

1. ASeven chunks are stored and no stale data remains
2. BThe upsert fails because of an id conflict
3. CChunks 7 to 9 from the old version remain as stale chunksCorrect answer
4. DAll ten chunks are duplicated

Why: Upsert overwrites ids 0 to 6 but never touches 7 to 9, so old chunks must be deleted by doc_id.

1. AIt uses only the file size
2. BIt asks the embeddings model whether the text is new
3. CIt computes a content hash per file and compares it with the ingest manifestCorrect answer
4. DIt passes every file on every run

Why: Only files whose hash is new or differs from the manifest are passed on, and files missing from the source produce delete instructions.

1. AReject it and record the document as failed or skipped (fail closed)Correct answer
2. BUpsert it with a default public tag
3. CUpsert it and fix the tags in a later run
4. DDrop the tag field from all chunks for consistency

Why: Untagged chunks could leak through retrieval filters, so validation must block them rather than guess.

1. ALeave the Playground open overnight
2. BSwitch the vector store component to tool mode
3. CEnable chat memory with a fixed session id
4. DAn external scheduler calls POST /api/v1/run/<flow-id>, optionally with tweaksCorrect answer

Why: Langflow flows are executed through the run API or webhooks, so a scheduler outside Langflow triggers the run and can pass tweaks to override component settings.

1. AReruns overwrite the same records instead of creating duplicatesCorrect answer
2. BThey let you mix embedding models in one collection
3. CThey make embeddings more accurate
4. DThey remove the need for a text splitter

Why: Ids derived from doc_id and chunk index make writes idempotent, so retries and repeated runs replace existing chunks rather than appending copies.

1. AA random UUID generated at split time
2. BFrom doc_id and chunk index, deterministicallyCorrect answer
3. CFrom the current timestamp
4. DBy the vector store auto-increment counter

Why: Deterministic ids from doc_id and chunk index make reruns overwrite rather than duplicate chunks.

1. AIt is silently ignored with no record
2. BThe whole run fails
3. CIt is embedded as an empty vector
4. DIt is recorded as skipped and the run continuesCorrect answer

Why: Skipping with a recorded reason keeps one bad file from blocking the rest while staying visible in the report.

1. ARight after splitting
2. BAfter its chunks have been successfully upsertedCorrect answer
3. CAt the start of the run, before loading
4. DRight after the file is loaded

Why: Updating only after success means a failed document still looks changed and is retried next run.

1. AUpserts the chunks without vectors
2. BMarks the documents as ingested to avoid loops
3. CWrites the affected documents to a failed-docs list for the next runCorrect answer
4. DDeletes the source files

Why: The failed-docs list, together with the unchanged manifest hash, ensures the next run picks those documents up.

1. ALangflow flows cannot run without a scheduler
2. BSchedulers make embeddings cheaper
3. CChat inputs cannot call vector stores
4. DThe refresh is driven from outside on a fixed interval, not by a user messageCorrect answer

Why: Ingestion is a batch job on a timetable, so an external scheduler calling the API fits better than a conversational trigger.

1. AThe change is missed and chunks keep the old tagsCorrect answer
2. BThe run fails with a hash conflict
3. CTags are updated automatically on the next run
4. DThe document is deleted

Why: The content hash is unchanged, so the detector needs a metadata fingerprint too, or permission changes are silently ignored.

1. ATo make embeddings more accurate
2. BTo reduce storage
3. CBecause the splitter requires it to run
4. DBecause retrieval filters are applied per chunk, so each chunk needs tags and doc_idCorrect answer

Why: Vector search returns chunks, so filtering by tenant, access or doc_id only works if every chunk carries those fields.

1. AIn the flow's exported JSON
2. BIn the scheduler's secret store, with a dedicated key for this jobCorrect answer
3. CHard-coded in the cron script in the repo
4. DInside a tweak value visible in logs

Why: A dedicated key in a secret store can be rotated or revoked and is kept out of code and exports.

1. AOnly Prometheus
2. BLangSmith and LangfuseCorrect answer
3. CLangflow has no tracing integrations
4. DOnly the browser console

Why: LangSmith and Langfuse are the named tracing integrations for inspecting runs.

1. AKeep old vectors and add new ones to the same collection
2. BIncrease chunk overlap to compensate
3. CRe-embed only documents that changed since the last run
4. DRe-embed all documents into a new collection, then point retrieval at itCorrect answer

Why: Vectors from different models are not comparable, so the whole corpus must be re-embedded, and building a separate collection avoids serving a half-migrated index.

1. AA canvas colour theme
2. BA YAML file describing a prompt
3. CA hosted model endpoint
4. DA Python class with defined inputs and outputsCorrect answer

Why: Custom components are Python classes with inputs and outputs, used here for change detection and metadata.

1. AEach rerun would add duplicate chunks alongside the old onesCorrect answer
2. BReruns would overwrite chunks correctly
3. CRetrieval would become faster
4. DMetadata filters would stop working

Why: Random ids never match previous ones, so upserts insert new rows instead of replacing, creating duplicates.

1. ABecause the vector store requires it as the primary key
2. BTo display in the chat UI
3. CTo detect vectors from an old model during migration and avoid mixing modelsCorrect answer
4. DTo choose the chunk size automatically

Why: Vectors from different models are not comparable, so the model name lets you find and replace stale ones.

1. AKeeping text that crosses a chunk boundary retrievable from either sideCorrect answer
2. BEncrypting chunk text
3. CSpeeding up hashing
4. DReducing the number of chunks

Why: Overlap preserves context at boundaries, at the cost of extra storage and embedding calls.

1. AUse a lock so only one run per source executes at a timeCorrect answer
2. BDisable deletes
3. CLet them both run; upserts are safe
4. DDouble the batch size

Why: Concurrent runs can interleave deletes and upserts for the same doc_id, so a per-source lock prevents races.

1. AIt is re-embedded
2. BIt is deleted
3. CIt is skipped as unsupported
4. DIt is treated as unchanged and not re-embeddedCorrect answer

Why: The hash matches the manifest, so no work is done, unlike a modified-time check.

1. AStop before deleting and alert, because a mass delete exceeds the safety thresholdCorrect answer
2. BDelete every document in the manifest
3. CRun a full reindex
4. DMark all documents as skipped and continue

Why: An empty listing usually means a source failure, so a delete threshold prevents wiping the index.

1. APass a full-reindex flag as a tweak in the run requestCorrect answer
2. BDelete the Langflow database
3. CSend the flag through chat memory
4. DEdit the custom component code each time

Why: Tweaks override component settings per run, including a reindex flag the change detector honours.

1. ATotal corpus documents times batch size
2. BOne per file regardless of size
3. CNumber of changed chunks divided by batch sizeCorrect answer
4. DNumber of deleted files

Why: Only changed content is embedded, and batching groups chunks per call.

1. ABy chunk overlap
2. BBy the order of upserts
3. CBy a metadata filter on tenant and access tags derived from the user's identityCorrect answer
4. DBy the embedding model

Why: Tenant and access tags stored with each vector let retrieval filter results to what the user may see.

1. AAs an unchanged document
2. BAs a skipped file
3. CAs a failed embedding
4. DAs a delete of the old doc_id plus a new documentCorrect answer

Why: A new path gives a new doc_id, so the old chunks are deleted and the content is ingested as new.

1. AOnly failures
2. BNumber of chat sessions
3. COnly total tokens
4. DAdded, updated, deleted, skipped and failedCorrect answer

Why: The report summarises each outcome so the team can spot gaps such as skips or failures.

1. AThe API run endpoint with tweaks
2. BMessage history for per-session chat memoryCorrect answer
3. CTracing integrations
4. DCustom components

Why: Ingestion is a stateless batch job with state in the manifest, so session chat memory plays no role.

1. AEdit chunks in place in the live collection
2. BDelete the live collection first, then ingest
3. CRun incremental ingestion only for new files
4. DIngest into a new collection via tweak, validate it, then switch retrievalCorrect answer

Why: Building a separate collection avoids mixed or missing content and keeps the old one for rollback.

1. AIt can expose flows as tools to MCP clientsCorrect answer
2. BIt replaces the vector store
3. CIt schedules flows automatically
4. DIt encrypts API keys

Why: As an MCP server, Langflow makes flows callable as tools by MCP clients.

1. AEmail polling built into the vector store
2. BWebhooksCorrect answer
3. COnly manual Playground runs
4. DDatabase triggers inside the embeddings component

Why: Langflow flows can be run through the API or webhooks, so webhooks suit near-real-time single-document updates.