Skip to content

Stream

Stream turns a continuous feed of documents into grounded zettelkasten notes. Where you normally hand the coordinator a source or two by hand, stream watches a source — a drop directory, an SEC EDGAR poll, a custom API — and, for every new document, drives a grounded-extraction graph that writes citation-backed claims into the zettelkasten.

It is a separate, optional package (pip install "angelo[stream]"), only wired into a project that inits --with-zettelkasten. It sits on top of the coordinator (it drives extraction graphs) and writes into the zettelkasten — so it doesn't add a new store, it just keeps the existing one fed. Earnings filings are the first worked recipe, but stream is general-purpose: any document source, any schema.

Documents don't have to be finished to be ingested. Stream can also micro-batch a live stream in real time — a source that arrives over time (a live earnings-call transcript, a captioning feed, a log tail) is accumulated and extracted incrementally as it is released, so notes appear while the stream is still rolling out rather than only after it ends. See Live streaming below.

The pipeline

Every document travels the same path, whether it arrived by a one-shot call or a long-running daemon:

  1. Feed — a source adapter (FeedAdapter) turns raw inputs into typed Documents. The built-in drop_dir feed watches a folder; custom feeds (an EDGAR poller, a webhook, a queue) are registered as entry points.
  2. Route — the router maps each document to one or more projects (e.g. one project per ticker, so a company's filings accumulate and stay comparable), and the schema selector maps it to one or more schemas (the extraction lenses to apply).
  3. Drive — for each routed document, stream builds a coordinator task graph and runs it with a portable driver — the agent backend that actually does the extraction.
  4. Write — grounded claims land in the routed project's synthesis spine in the zettelkasten, verbatim quotes and all. From there they behave like any other note: they cluster, feed spines, and populate matrices.

Because step 3 is just a coordinator run, everything you know about rigor, waves, and grounded extraction applies unchanged — stream is orchestration glue, not a second extraction engine.

Portable drivers

The driver decouples stream from any one agent runtime. Three ship in the box:

Driver Backend Set
claude (default) Anthropic API ANTHROPIC_API_KEY
gpt OpenAI / OpenAI-compatible (vLLM, local) OPENAI_API_KEY (+ base_url)
cursor Cursor's agent runtime —

This is why stream can run headless, in CI, or on a server with no editor attached — it doesn't depend on Cursor being open. Provider SDKs are lazy, opt-in extras: pip install "angelo[stream]" pulls both Claude and GPT, or install just anthropic / openai for one. Importing stream stays cheap — the SDKs, the coordinator, and the zettelkasten store are all imported lazily.

Two ways to run it

Imperative — call the facade directly, good for scripts and notebooks:

import stream

report = stream.apply(
    ["./inbox/AAPL-10Q.pdf"],
    project="aapl",
    schemas=["earnings"],
    driver="claude",
)

Declarative — describe the whole pipeline once in stream.yaml, then run a single pass or a long-lived daemon:

import stream

stream.process_once("stream.yaml")               # drain the feeds once
stream.StreamDaemon.from_file().serve_forever()   # watch forever

Or from the CLI entry point, angelo-stream, for one-shot or daemon runs.

Extract once, attach to many projects

Pass projects=[...] (instead of, or alongside, the singular project=) and the expensive LLM extraction runs once — the claim is written a single time into the source graph and attached to each listed project's spine:

report = stream.apply(
    ["./inbox/AAPL-10Q.pdf"],
    projects=["tech-fund", "macro-fund"],   # one extraction, attached to both
    schemas=["earnings"],
)

A single project behaves exactly as before. In stream.yaml this happens automatically: a document the router sends to several projects is grouped by its exact project set and extracted once.

Anatomy of a stream.yaml

A worked earnings config shows the moving parts (see examples/earnings/stream.yaml):

defaults:
  driver: claude          # claude | gpt | cursor
  concurrency: 2          # parallel agents per wave
  rigor: medium           # low | medium | high
  template: extraction    # built-in grounded-extraction template
  project: earnings       # fallback project when routing yields none

archive: local            # keep a durable copy of each original: none | local | dvc

feeds:
  - name: inbox
    type: drop_dir
    options:
      path: ./inbox
      glob: "*.pdf"
      delete_after: true  # remove the file once its extraction run succeeds

schemas:                  # document metadata -> schema name(s), UNION across rules
  default: []             # [] = semantic-only ingestion (no spine)
  rules:
    - match: {doc_type: pdf}
      schemas: [earnings]

router:                   # document -> project name(s)
  default: [earnings]
  axes:
    - type: metadata
      key: ticker
      template: "{value}"  # ticker AAPL -> project "AAPL"

The pieces:

  • defaults — the driver, model, concurrency, and rigor every run inherits, plus the fallback project when routing yields nothing.
  • archive — how the original document is preserved before the feed deletes it: none, local (a gitignored content-addressed store under .angelo/stream/), or dvc (push to the configured DVC/S3 remote).
  • feeds — the sources. drop_dir is built in; register custom feed factories under the angelo.stream.feeds entry-point group.
  • schemas — which extraction lenses apply, matched on document metadata. An empty default means semantic-only ingestion (notes without a spine).
  • router — which project(s) a document scopes into. One project per axis value (per ticker, per topic) keeps related sources comparable over time.

Tabular data bypasses extraction

Not every document should be mined for claims. A datasets: block routes CSV/JSON/Parquet documents around grounded extraction: the numbers are the data, so stream writes a dataset note pointing at the file (copied into the source folder, or referenced live) and adds it to the routed project. Those datasets can then be linked to filings' claims with measures / supports / derives-from.

Live streaming: one box per call

The feeds above deliver finished documents (a PDF, a filing). Some sources instead arrive over time — a live earnings-call transcript, a captioning stream, a log tail. Stream handles these by micro-batching in real time: a segmenting live feed accumulates the stream and extracts each new slice as it is released, so claims land in one box per logical source (one box per call, not one box per transcript clipping) while notes still appear as the source rolls out — you don't wait for the stream to finish.

The base class SegmentingLiveFeed owns the mechanics; a concrete feed only implements read_deltas(), returning whatever text newly arrived:

from stream import SegmentingLiveFeed, StreamDelta

class MyLiveFeed(SegmentingLiveFeed):
    def read_deltas(self):
        for chunk in self._client.drain():          # your source
            yield StreamDelta(
                stream_key="acme-fy25q1-call",       # the ONE box this stream fills
                text=chunk.text,
                final=chunk.is_last,
                metadata={"ticker": "ACME"},         # routed like any document
            )

On each poll the base appends new text to the stream's single accumulating file and, once flush_threshold new characters have arrived (or on a final delta), emits one Document that:

  • shares the stream's box name, and is marked refresh=True so the box's content pointer is refreshed rather than duplicated;
  • carries a slices window covering only the new characters (plus a small overlap backstep) so extraction mines just the delta, while quote grounding verifies against the whole accumulated transcript.

This relies on a small growable-source primitive in the store: create_source(update=True) refreshes an existing box's content hash and busts its grounding cache, and the extraction template threads the per-delta slices. The upshot is real-time-ish extraction (notes lag by ~one poll interval, since each delta runs an extractor → scribe → auditor wave) that still yields a single, coherent box per call.

Because a live feed buffers across polls it must be a persistent instance. The daemon keeps feeds alive across passes, so SegmentingLiveFeed works under StreamDaemon (and angelo-stream); a bare one-shot process_once rebuilds feeds each call and is meant for stateless feeds like drop_dir.

Try it without a live source: ReplayFeed

The built-in replay feed replays a saved transcript in timed chunks — a worked example of the base and the way to develop or test a live pipeline with no live source:

feeds:
  - name: call
    type: replay
    options:
      path: ./transcripts/acme-fy25q1.txt
      stream_key: acme-fy25q1-call
      chunk_size: 800
      chunk_interval: 5        # release a chunk every 5s (0 = all at once)
      flush_threshold: 1500    # extract once this many new chars accumulate
      overlap: 200             # re-read the last 200 chars each delta

A vendor-specific live feed (e.g. an LSEG earnings-call subscription) is not part of angelo core: it lives in a separate package that pip installs angelo, subclasses SegmentingLiveFeed, and registers under the angelo.stream.feeds entry point — exactly like the EDGAR example feed.

Where it fits

Stream is deliberately outside the core architecture picture on the home page: the coordinator, memory, and zettelkasten stand on their own, and stream is the thing you add when you want the zettelkasten continuously fed rather than filled by hand. If you're doing one-off literature reviews you never need it; if you're tracking a live corpus (quarterly filings, a paper feed, an incident stream) it's the front door.

  • Package overview: stream/README.md
  • Behavioral contract for agents: the bundled stream rule, installed with --with-zettelkasten.

Semantically related entries from the memory graph.