stream.config

stream.config

Loader for stream.yaml -- the declarative pipeline configuration.

The YAML covers the common case end to end (driver/model, feeds, schema selection, project routing, archive policy). Anything it cannot express is a Python entry point (custom feeds/routers/templates/classifiers); see :mod:stream.contrib.

Example::

defaults:
  driver: claude          # claude | gpt | cursor
  model: ""               # optional model override
  concurrency: 2
  rigor: medium
  template: extraction
  project: research

archive: local            # none | local | dvc

feeds:
  - name: inbox
    type: drop_dir
    options: {path: ./inbox, glob: "*.pdf", delete_after: true}

schemas:
  default: []
  rules:
    - match: {doc_type: "10-Q"}
      schemas: [earnings]

router:
  default: [research]
  axes:
    - type: metadata
      key: ticker

SpineSelection dataclass

Per-document spine routing, mirroring schema selection + project routing.

rules union match-bound spine ids ({match: {...}, spines: [...]}) and axes derive spine ids from metadata (same axis shape as :class:RouterConfig); both fall back to default when nothing matches. This is layered ON TOP of the run-global StreamConfig.spines baseline, which is merged into every document's resolved set in process_once. When no spine_routing: block is present every document resolves to the empty set here and only the global baseline applies (byte-identical to before).

Source code in stream/config.py

@dataclass
class SpineSelection:
    """Per-document spine routing, mirroring schema selection + project routing.

    ``rules`` union match-bound spine ids (``{match: {...}, spines: [...]}``) and
    ``axes`` derive spine ids from metadata (same axis shape as
    :class:`RouterConfig`); both fall back to ``default`` when nothing matches.
    This is layered ON TOP of the run-global ``StreamConfig.spines`` baseline,
    which is merged into every document's resolved set in ``process_once``. When
    no ``spine_routing:`` block is present every document resolves to the empty
    set here and only the global baseline applies (byte-identical to before).
    """

    default: list[str] = field(default_factory=list)
    rules: list[dict[str, Any]] = field(default_factory=list)
    axes: list[dict[str, Any]] = field(default_factory=list)

DatasetSink dataclass

Route tabular feed documents straight into dataset notes (no LLM).

A document whose doc_type is listed in doc_types (matched by the file extension the drop-dir feed assigns, e.g. csv/json/parquet) BYPASSES grounded extraction: the sink creates a source folder, writes a dataset note pointing at the numbers, and adds the source to the routed project(s). Numbers never go through the extractor — the file IS the data.

mode controls where the bytes live (mirroring the dataset backends):

• ingest (default) — COPY the file into the source folder so the dataset is self-contained and durable even when the feed deletes the original. With top-level archive: dvc the in-folder copy is DVC-tracked (backend dvc); otherwise it is a committed sidecar (backend sidecar).
• reference — leave the file at its current path and point the dataset at it via the file fetch provider (backend api). Use for a live export a job keeps refreshing; pin a revision later with dataset(snapshot).

Empty doc_types disables the sink entirely (byte-identical to before): every document flows to the extraction path as usual.

Source code in stream/config.py

@dataclass
class DatasetSink:
    """Route tabular feed documents straight into ``dataset`` notes (no LLM).

    A document whose ``doc_type`` is listed in ``doc_types`` (matched by the file
    extension the drop-dir feed assigns, e.g. ``csv``/``json``/``parquet``)
    BYPASSES grounded extraction: the sink creates a source folder, writes a
    ``dataset`` note pointing at the numbers, and adds the source to the routed
    project(s). Numbers never go through the extractor — the file IS the data.

    ``mode`` controls where the bytes live (mirroring the dataset backends):

    * ``ingest`` (default) — COPY the file into the source folder so the dataset
      is self-contained and durable even when the feed deletes the original.
      With top-level ``archive: dvc`` the in-folder copy is DVC-tracked (backend
      ``dvc``); otherwise it is a committed sidecar (backend ``sidecar``).
    * ``reference`` — leave the file at its current path and point the dataset at
      it via the ``file`` fetch provider (backend ``api``). Use for a live export
      a job keeps refreshing; pin a revision later with ``dataset(snapshot)``.

    Empty ``doc_types`` disables the sink entirely (byte-identical to before):
    every document flows to the extraction path as usual.
    """

    doc_types: list[str] = field(default_factory=list)
    mode: str = "ingest"