Data Ingestion & Sync — механизм, который систематически переносит данные из внешних источников во внутренний knowledge/data layer и поддерживает локальное представление актуальным: делает backfill, incremental sync, deduplication, upsert, deletion handling, retries и checkpoints.
№54 Connectors умеет обращаться к конкретной внешней системе; №55 orchestrates recurring ingestion, backfill, checkpoints, dedupe, update/delete propagation. №56 Document Parsing/OCR превращает конкретный fetched blob/document в извлечённые структурированные элементы; №55 доставляет и переобрабатывает такие blobs. №08 RAG ищет знания во время запроса; №55 подготавливает corpus заранее. №42 Events & Triggers может инициировать sync; №55 владеет sync state. №57 Queues & Workers выполняет work units; №55 определяет сами ingestion jobs и checkpoints. №61 Provenance позже углубит lineage; №55 обязан уже сохранять source identity/version.
Prerequisites: №08 RAG, №10 State, №42 Events, №46 Observability, №50 Contracts, №54 Connectors. Forward references: №56 Parsing/OCR, №57 Queues/Workers, №58 Scheduler, №59 Broker, №60 Artifact Store, №61 Provenance, №62 Caching, №63 Retry, №66 Vector DB & Embeddings.
REQUEST-TIME: обычно NO — это background pipeline. CONTROL PLANE: YES — source registry, schedules, cursors, backfill state, pause/resume, schema versions. DATA PLANE: YES — source objects, blobs, normalized records, tombstones. OFFLINE: YES — reindex/backfill/reprocessing, validation, quality checks.
Success: local corpus converges toward source state within defined freshness SLA. Retryable: transient source/worker errors and rate limits. Permanent: unsupported schema/object, revoked auth, deterministic parse failure after policy-defined attempts. Idempotency: same source version can be replayed safely. Persist: source_id, external_id, version/hash, cursor/checkpoint, sync run, last_seen/updated, tombstone state, parse/index versions. Trace: source→fetch→transform→upsert/delete→index.
№55 не владеет provider API adapter, content parsing algorithms, embedding model, vector DB internals, queue engine, scheduler platform или full data governance. Она владеет THE STATEFUL PROCESS THAT MOVES AND RECONCILES SOURCE DATA OVER TIME.
Agent вызывает connector прямо в момент запроса. Хорошо для небольших sources, current transactional state, редких запросов.
Плюсы: freshest source, no duplicate storage.
Минусы: latency, rate limits, provider availability, weak full-text/vector search.
Система заранее копирует/нормализует нужное представление и строит индексы.
Плюсы: fast search, embeddings, joins, offline processing.
Минусы: freshness problem, storage, delete handling, sync complexity.
Пройти весь доступный dataset и создать initial local representation. Нужен при первом подключении или полной rebuild.
Периодически запросить только новые/изменённые records. Типовой MVP.
Provider сообщает об изменении. Быстро, но всё равно нужен recovery/reconciliation path.
Сравнить local state с authoritative source, чтобы поймать потерянные события и drift.
Source не изменился, но поменялся parser/chunker/embedding/model/schema — нужно переобработать.
Ингестировать конкретный object/collection по запросу workflow, а не весь source.
{
"source_id": "SRC-...",
"tenant_id": "tenant_A",
"connector_id": "drive:brand_A",
"resource_scope": "folder:123",
"sync_mode": "INCREMENTAL",
"status": "ACTIVE",
"schedule": "*/15m",
"cursor": "opaque:...",
"freshness_sla_s": 1800,
"parser_profile": "documents.v2",
"index_profile": "knowledge.v3",
"delete_policy": "TOMBSTONE",
"last_success_at": "...",
"last_error": null
}Source registry отвечает:
Last durable point: cursor C42 / updated_at T / page token.
Fetch changes after C42. Upsert/delete all records. Persist derived artifacts/index state.
Advance to C43 only when processing policy says window is complete.
Provider change token/page cursor should be treated as opaque state, not interpreted by model.
updated_at > last_seen works only if ordering/time semantics are reliable; usually add overlap window.
Use timestamp plus deterministic tie-breaker ID to avoid missing records with equal timestamps.
source_key =
tenant_id
+ connector_id
+ object_type
+ external_id
source_version =
provider_version
OR updated_at
OR content_hash
upsert(record):
existing = load(source_key)
if existing.version == source_version:
return NOOP
write canonical source record
write provenance
enqueue derived processing if needed
mark version currentПовтор может возникнуть из-за:
Sync должен проектироваться так, будто один object обязательно придёт повторно.
| Version axis | Пример | Что вызывает |
|---|---|---|
| SOURCE VERSION | Document updated_at/content hash changed. | Refetch/reparse/reindex. |
| PARSER VERSION | document parser v2 → v3. | Reparse same source blob. |
| CHUNKER VERSION | Chunking strategy changed. | Rechunk/reembed. |
| EMBEDDING VERSION | New embedding model. | Re-embed current chunks. |
| INDEX SCHEMA | New metadata/filter fields. | Reindex/migrate derived records. |
updated_at для всей pipeline lineage. Храните версии source и derived processing раздельно.Решается stable event/object IDs + idempotent upsert.
Можно использовать content hash для storage optimization, но не смешивать source identity.
Это уже knowledge resolution/entity layer, не базовая sync responsibility.
Удалить local record и derived chunks/index entries. Подходит, если governance требует полного удаления.
Сохранить minimal metadata/provenance, но исключить content из retrieval.
Provider объект архивирован/скрыт, но не удалён. Canonical status отражает это отдельно.
Если provider не даёт delete events, периодическая inventory reconciliation выявляет исчезнувшие objects.
Discovers/fetches external file metadata and blob/reference.
Tracks source identity/version, downloads/stores blob, schedules processing, maintains sync state.
Extracts text/tables/images/OCR/layout into structured document representation.
Change event → dedupe → fetch current object → idempotent upsert. Хорошо для freshness.
Periodic change feed or inventory scan catches missed/expired webhook events.
Time from source change to durable local availability.
Time since last successful run/checkpoint.
Known items/events/jobs waiting to process.
Например, 95% updates searchable within N minutes.
| Failure | Recommended behavior |
|---|---|
| One object parse fails | Mark object PROCESSING_FAILED, preserve source/blob ref, continue source run if safe. |
| Provider transient outage | Bounded retry/backoff, keep checkpoint unchanged for affected window. |
| Permission revoked | Pause source / AUTH_REQUIRED; do not hammer provider. |
| Rate limited | Respect Retry-After, reduce concurrency, resume from checkpoint. |
| Schema drift | Quarantine affected records, alert, keep raw payload/ref, do not silently corrupt canonical schema. |
| Index write fails after source record stored | Mark derived index stale/pending; retry indexing without refetch if possible. |
В распределённой pipeline трудно гарантировать literally one delivery across source/network/worker/store.
Разрешить replay, но сделать upsert/dedupe безопасными и deterministic.
Advance progress only after effects required by your consistency contract are durable.
| Model | Meaning | Use case |
|---|---|---|
| EVENTUAL | Local representation converges after bounded delay. | Most knowledge/RAG corpora. |
| READ-THROUGH VERIFY | Use local corpus for discovery, then live-read critical object before action. | Policies/current status/transactional data. |
| SNAPSHOT | Use consistent source snapshot/version for a batch/research job. | Audits, reproducible reports. |
| STRONG LIVE | Do not rely on mirror for authoritative current state. | Balances, permissions, mutable transactional state. |
Ингестированные документы складываются в один index без tenant/ACL metadata.
tenant, source scope, owner/groups/classification сохраняются вместе с record/chunks.
RAG/query layer применяет authorization filters before returning evidence to model.
{
"record_ref": "doc://tenant_A/123",
"source_id": "SRC-drive-brandA",
"external_id": "drive-file-abc",
"source_version": "v42",
"content_hash": "sha256:...",
"fetched_at": "...",
"connector_version": "2.3.1",
"parser_version": "documents.v2",
"index_version": "knowledge.v3",
"status": "CURRENT"
}Минимальная lineage позволяет понять:
№61 позже расширит lineage как отдельную тему.
source_id, run_id, mode, checkpoint start/end.
items/pages/events discovered.
new / updated / unchanged / deleted.
fetch/parse/index/auth/rate-limit counts.
source-change-to-searchable latency.
Queued items and oldest age.
deletions propagated to all derived layers.
missing/orphaned/version-mismatch counts.
All expected objects mirrored once logically.
Changed source version updates local/derived records.
Deleted object disappears from retrieval/derived indexes.
Same page/event/object repeated without duplication.
Worker dies before/after write; restart neither loses nor corrupts object.
429/backoff preserves progress and does not storm provider.
Lost webhook is later repaired by poll/reconcile.
Permission changes propagate to searchable representation.
Source change → local searchable availability.
Successful runs/items by source and stage.
Oldest pending item/event and queue depth.
Reconciliation mismatches per source snapshot.
Time until removed source is absent from retrieval.
Items refetched/reprocessed without source/pipeline change.
Objects stuck in deterministic processing error.
API calls/bytes/compute per changed object and per source.
ingestion/ ├── sources.py ├── discover.py ├── sync.py ├── checkpoints.py ├── upsert.py ├── deletes.py ├── reconcile.py └── tests/ tables: sources sync_runs source_objects processing_state sources: source_id connector_id tenant_id scope cursor mode last_success_at source_objects: source_key external_id source_version content_hash status last_seen_at deleted_at parser_version index_version
Queues, brokers, distributed workers and sophisticated CDC can be added later if scale actually requires them.
| Вопрос | Ответ |
|---|---|
| Стоит ли реализовывать? | Да, если система хранит локальный knowledge corpus/index, который должен обновляться. Если всегда используется live connector read — может быть не нужен. |
| Separate Component? | YES. Stateful long-running sync responsibility отдельно от connector adapter и request-time RAG. |
| Минимум 80% ценности? | Source registry, backfill, cursor, source version/hash, idempotent upsert, delete handling, reconciliation, freshness metrics. |
| Когда overkill? | Строить Kafka/CDC/data-lake platform для 300 документов, которые можно раз в 15 минут проверить простым worker. |
| Trigger? | Нужен локальный searchable/indexed corpus, offline processing, embeddings или устойчивость к source latency. |
| Как измерить uplift? | Freshness lag, retrieval coverage, drift, sync failures, source API pressure, stale-answer rate, cost per changed object. |
| Можно ли rule/tool/code вместо LLM-agent? | Да, полностью. Ingestion/sync — deterministic pipeline; LLM может использоваться позже только внутри content extraction/classification tasks. |
Provider access и long-running reconciliation — разные responsibilities.
At-least-once + idempotent upsert лучше fragile exactly-once assumptions.
Progress advances only after required effects are committed.
external_id alone insufficient; know whether content actually changed.
Parser/chunker/embedding/index may require reprocess without source change.
Removed source must disappear from all derived retrieval surfaces.
Webhooks/cursors are efficient, reconciliation is the safety net.
Deduplication must not erase source identity/authority/access context.
«Sync healthy» means meeting explicit lag/completeness objectives.
EXTERNAL SOURCE
↓
№54 CONNECTOR
list / fetch / webhook normalize
↓
SOURCE REGISTRY
scope + tenant + mode + cursor + SLA
↓
DISCOVER CHANGES
backfill
incremental
webhook
reconciliation
↓
STABLE SOURCE KEY
+ source version/hash
↓
FETCH SOURCE OBJECT / BLOB
↓
STORE SOURCE REPRESENTATION + PROVENANCE
↓
[ №56 PARSING / EXTRACTION if needed ]
↓
NORMALIZE
↓
IDEMPOTENT UPSERT
├─ NEW
├─ UPDATE
├─ NOOP
└─ DELETE / TOMBSTONE
↓
DERIVED PROCESSING
chunks
embeddings
search index
graph
↓
VERIFY DURABLE RESULT
↓
ADVANCE CHECKPOINT
↓
FRESHNESS / LAG / DRIFT METRICS
PERIODICALLY:
AUTHORITATIVE SOURCE SNAPSHOT
↕
LOCAL CORPUS
↓
RECONCILE DIFFERENCES
CORE PRINCIPLE:
INGESTION GETS DATA IN.
SYNC KEEPS IT TRUE ENOUGH
FOR THE SYSTEM'S FRESHNESS CONTRACT.
CONNECTOR KNOWS HOW TO TALK TO THE SOURCE.
SYNC KNOWS HOW TO REMEMBER PROGRESS,
REPLAY SAFELY,
PROPAGATE CHANGES
AND REPAIR DRIFT.
A. Related ECC ideas. Context-as-cache, scoped memory, lifecycle hooks, selective capabilities, feature flags, deterministic enforcement, provider-neutral adapters and eval-gated learning are applied only where relevant to №55 Data Ingestion & Sync.
B–E. Existing boundary and placement. The existing conceptual boundary, class PRODUCTION, default CONDITIONAL and owner Knowledge / Research Engine + Production Fabric remain authoritative. Runtime/control/data/offline placement is unchanged; durable state stays outside model context.
F–H. Hooks and contracts. Use bounded PRE_MODEL/POST_MODEL, PRE_TOOL/POST_TOOL, CHECKPOINT and TASK_COMPLETED events as applicable. Illustrative fields and canonical contracts are defined in NEW_CONTRACTS_SPEC.md; no universal schema is implied.
I–J. Security and evaluation. Host-side schema, permission, secret, budget, idempotency and audit checks take precedence over LLM output. Optional mechanisms require a feature flag and WITH/WITHOUT ablation; measure quality, acceptance, correction, latency, cost, escalations and severe errors.
K–L. Task profiles and cross-references. A TaskProfile selects the relevant skill, tool/context slice, memory scope and enforcement profile independently from FAST/STANDARD/DEEP. See cross-reference map, hook spec and ablation plan. Provider adapters remain outside the core.