fastrag.ingestion.events#

Events emitted during pipeline execution.

Two families live here: step lifecycle events (Step*), emitted by the evaluator around every step regardless of what it does, and domain events (Document*, ItemFailed), emitted by a step itself through its ExecutionContext to describe what it did to one item. Both are propagated to whatever sink a Pipeline context manager installs.

Every event carries two names for the step that produced it, and they answer different questions:

step is the instance name — by default the producing class’s own name (ChunkerRouter, SitemapCrawler, MilvusVectorStore.save). It identifies which object in the graph emitted this, and two differently configured instances of the same family are told apart by it.
event_name is the stage name — the label event_name resolves to for that class (Fetching, Parsing, Chunking, Embedding, Indexing, …). It identifies what kind of work this was, so a consumer can group every chunker in a graph under one heading without knowing any concrete class. A step whose family declares no event_name falls back to its own class name, so this is never empty for an event a real step emitted.

Both are plain strings with an empty default, so an event constructed by hand (a test double, a consumer replaying a stream) stays valid without naming either.

Classes

`DocumentChunked`(document[, chunks, step, ...])	Emitted when a document has been split into chunks.
`DocumentEmbedded`(document[, step, event_name])	Emitted when a document or chunk receives vector embeddings.
`DocumentFetched`(source[, step, event_name])	Emitted when a source item has been fetched.
`DocumentParsed`(document[, step, event_name])	Emitted when a document has been parsed.
`DocumentSaved`(document[, step, event_name])	Emitted when a document is persisted into a vector store.
`DocumentsPruned`([step, stale, deleted, ...])	Emitted once by a vector store's sync step, after its upstream is exhausted.
`ItemFailed`([step, uri, error, source_id, ...])	Emitted when a single item fails within a step.
`ItemFinished`(item_id[, elapsed, error])	Emitted when a step is done with one item, successfully or not.
`ItemRetrying`(item_id[, attempt, delay, error])	Emitted when a transient failure or timeout is about to be retried.
`ItemStarted`(item_id[, uri, step, ...])	Emitted when a step starts working on one item.
`PipelineEvent`()	Base class for all events emitted during pipeline execution.
`StepCompleted`(step[, event_name])	Emitted when a step completes processing all items.
`StepFailed`(step[, error, event_name])	Emitted when a step encounters an unrecoverable error.
`StepProgress`(step[, processed_count, event_name])	Emitted incrementally as items process through a step.
`StepStarted`(step[, event_name])	Emitted when a pipeline step begins processing.

class fastrag.ingestion.events.PipelineEvent[source]#

Bases: object

Base class for all events emitted during pipeline execution.

class fastrag.ingestion.events.StepStarted(step, event_name='')[source]#

Bases: PipelineEvent

Emitted when a pipeline step begins processing.

Parameters:

step (str)
event_name (str)

step#

Name of the step that started.

Type:: str

event_name#

Stage label of the step that started — see this module’s docstring.

Type:: str

class fastrag.ingestion.events.StepProgress(step, processed_count=0, event_name='')[source]#

Bases: PipelineEvent

Emitted incrementally as items process through a step.

Parameters:

step (str)
processed_count (int)
event_name (str)

step#

Name of the step reporting progress.

Type:: str

processed_count#

Items successfully processed so far.

Type:: int

event_name#

Stage label of the reporting step — see this module’s docstring.

Type:: str

class fastrag.ingestion.events.StepCompleted(step, event_name='')[source]#

Bases: PipelineEvent

Emitted when a step completes processing all items.

Parameters:

step (str)
event_name (str)

step#

Name of the step that completed.

Type:: str

event_name#

Stage label of the step that completed — see this module’s docstring.

Type:: str

class fastrag.ingestion.events.StepFailed(step, error=<factory>, event_name='')[source]#

Bases: PipelineEvent

Emitted when a step encounters an unrecoverable error.

Parameters:

step (str)
error (Exception)
event_name (str)

step#

Name of the step that failed.

Type:: str

error#

The exception instance that triggered failure.

Type:: Exception

event_name#

Stage label of the step that failed — see this module’s docstring.

Type:: str

class fastrag.ingestion.events.DocumentFetched(source, step='', event_name='')[source]#

Bases: PipelineEvent

Emitted when a source item has been fetched.

Parameters:

source (Source)
step (str)
event_name (str)

source#

The fetched item. source.cached is True when it was served from the cache rather than freshly fetched.

Type:: fastrag.ingestion.models.Source

step#

Name of the step that fetched it.

Type:: str

event_name#

Stage label of that step — see this module’s docstring.

Type:: str

class fastrag.ingestion.events.DocumentParsed(document, step='', event_name='')[source]#

Bases: PipelineEvent

Emitted when a document has been parsed.

Parameters:

document (Document)
step (str)
event_name (str)

document#

The parsed document. document.cached is True when it was served from the cache rather than freshly parsed.

Type:: fastrag.models.Document

step#

Name of the step that parsed it.

Type:: str

event_name#

Stage label of that step — see this module’s docstring.

Type:: str

class fastrag.ingestion.events.DocumentChunked(document, chunks=<factory>, step='', event_name='')[source]#

Bases: PipelineEvent

Emitted when a document has been split into chunks.

Parameters:

document (Document)
chunks (list[Document])
step (str)
event_name (str)

document#

The parent document that was split.

Type:: fastrag.models.Document

chunks#

The chunks produced from document — retained (rather than just a count) so a consumer can inspect each chunk’s own cached/id/metadata individually.

Type:: list[fastrag.models.Document]

step#

Name of the step that split it.

Type:: str

event_name#

Stage label of that step — see this module’s docstring.

Type:: str

class fastrag.ingestion.events.DocumentEmbedded(document, step='', event_name='')[source]#

Bases: PipelineEvent

Emitted when a document or chunk receives vector embeddings.

Parameters:

document (Document)
step (str)
event_name (str)

document#

The embedded document. document.cached is True when its embedding was served from the cache rather than freshly computed, or when skip_existing bypassed embedding for an already indexed document. In the latter case, its embedding may be None.

Type:: fastrag.models.Document

step#

Name of the step that embedded it.

Type:: str

event_name#

Stage label of that step — see this module’s docstring.

Type:: str

class fastrag.ingestion.events.DocumentSaved(document, step='', event_name='')[source]#

Bases: PipelineEvent

Emitted when a document is persisted into a vector store.

Parameters:

document (Document)
step (str)
event_name (str)

document#

The saved document.

Type:: fastrag.models.Document

step#

Name of the step that saved it.

Type:: str

event_name#

Stage label of that step — see this module’s docstring.

Type:: str

class fastrag.ingestion.events.DocumentsPruned(step='', stale=0, deleted=0, failures=0, event_name='')[source]#

Bases: PipelineEvent

Emitted once by a vector store’s sync step, after its upstream is exhausted.

Parameters:

step (str)
stale (int)
deleted (int)
failures (int)
event_name (str)

step#

Name of the sync step (MilvusVectorStore.sync).

Type:: str

stale#

How many stored documents the run did not yield.

Type:: int

deleted#

How many of them were deleted: all of them after a clean run, none after a run with failures.

Type:: int

failures#

ItemFailed events the run emitted. Any failure keeps the stale documents, since a failed item’s documents are missing from the run without being gone from the source.

Type:: int

event_name#

Stage label of that step — see this module’s docstring.

Type:: str

class fastrag.ingestion.events.ItemFailed(step='', uri='', error=<factory>, source_id='', event_name='')[source]#

Bases: PipelineEvent

Emitted when a single item fails within a step.

Parameters:

step (str)
uri (str)
error (Exception)
source_id (str)
event_name (str)

step#

Name of the step the failing item was being processed by. The first field (rather than an afterthought default), so every caller names it — a consumer (e.g. ConsoleLogger) can always attribute the failure to the stage it came from.

Type:: str

uri#

Identifier of the item that failed.

Type:: str

error#

The exception instance that triggered the failure.

Type:: Exception

source_id#

The originating Source id the failing item traces back to (its own id for a Source, or .source_id for a Document) — empty when the failing item carries neither (e.g. a whole child SourceStep failing inside Sources). Lets a consumer attribute a failure to the row a multi-source table’s step name alone cannot identify (a parser/chunker/ embedder/sink step is shared by every source; only the item it was processing says which one).

Type:: str

event_name#

Stage label of the step the item failed in — what a consumer columns the failure under, since step alone names an instance rather than a stage. See this module’s docstring.

Type:: str

class fastrag.ingestion.events.ItemStarted(item_id, uri='', step='', event_name='', source_id='')[source]#

Bases: PipelineEvent

Emitted when a step starts working on one item.

Parameters:

item_id (int)
uri (str)
step (str)
event_name (str)
source_id (str)

item_id#

Unique for this unit of work; repeated on its ItemFinished.

Type:: int

uri#

Identifier of the item.

Type:: str

step#

Name of the step working on it.

Type:: str

event_name#

Stage label of that step.

Type:: str

source_id#

The originating Source id, as on ItemFailed.

Type:: str

class fastrag.ingestion.events.ItemRetrying(item_id, attempt=1, delay=0.0, error=<factory>)[source]#

Bases: PipelineEvent

Emitted when a transient failure or timeout is about to be retried.

Parameters:

item_id (int)
attempt (int)
delay (float)
error (Exception)

item_id#

The ItemStarted.item_id of the unit of work.

Type:: int

attempt#

The attempt that just failed, 1-based.

Type:: int

delay#

Seconds before the next attempt.

Type:: float

error#

What the failed attempt raised.

Type:: Exception

class fastrag.ingestion.events.ItemFinished(item_id, elapsed=0.0, error=None)[source]#

Bases: PipelineEvent

Emitted when a step is done with one item, successfully or not.

Parameters:

item_id (int)
elapsed (float)
error (Exception | None)

item_id#

The ItemStarted.item_id of the unit of work.

Type:: int

elapsed#

Wall-clock seconds since its ItemStarted, retries included.

Type:: float

error#

What it finally raised, or None on success.

Type:: Exception | None