Merge pull request #12 from artiz/claude/library-streaming-mode-3p8gvu

feat(stream): streaming Markdown conversion, default in the CLI

Author: artiz

README.md

@@ -118,6 +118,40 @@ strict:  Foo ***both***.    |   ```rust (lang kept)  |   Name: ___
 `result.document.export_to_markdown_with(strict)` overrides the mode per call.
 Python docling has no such switch.
 
+### Streaming Markdown
+
+For embedding in real apps, `convert_streaming` returns the document's Markdown
+as an iterator of chunks instead of one big string — handy for piping a long
+document straight to stdout, an HTTP response, or a socket as it is produced:
+
+```rust
+use std::io::Write;
+use fleischwolf::{DocumentConverter, SourceDocument};
+
+let source = SourceDocument::from_file("input.pdf").unwrap();
+let mut out = std::io::stdout();
+for chunk in DocumentConverter::new().convert_streaming(source).unwrap() {
+    out.write_all(chunk.unwrap().as_bytes()).unwrap();
+}
+```
+
+The headline win is PDF. The ML pipeline already processes pages **in parallel**;
+streaming emits each page's Markdown **in document order, as soon as it is ready**
+(with a one-page look-ahead so paragraphs that wrap across a page break still
+merge), so output starts flowing before the last page is done. The conversion
+runs on a background thread and the chunk iterator applies backpressure; dropping
+it cancels the work. Concatenating every chunk is **byte-identical** to the
+buffered `export_to_markdown()`.
+
+Streaming is Markdown-only — JSON serializes docling-core's reference-based tree
+and needs every node up front. Picture placeholders and `embedded` data-URI
+images stream; the `referenced` mode writes sidecar files, so it stays on the
+buffered `export_to_markdown_with_images` path. Use
+`convert_streaming_images(source, ImageMode::Embedded)` to pick the image mode.
+
+The CLI streams Markdown by default (`--no-stream` opts back into buffering;
+`--to json` and `--images referenced` always buffer).
+
 ## Testing
 
 All commands run from the `fleischwolf/` workspace root.
@@ -171,8 +205,13 @@ cargo run -p fleischwolf-cli -- --to json crates/fleischwolf/sample.html > out.j
 cargo run -p fleischwolf-cli -- --images embedded   document.pdf
 cargo run -p fleischwolf-cli -- --images referenced document.pdf > out.md
 
-# or via the example
+# stream Markdown to stdout page by page (the CLI's default; --no-stream to buffer)
+cargo run -p fleischwolf-cli -- document.pdf
+cargo run -p fleischwolf-cli -- --no-stream document.pdf
+
+# or via the examples
 cargo run -p fleischwolf --example convert -- crates/fleischwolf/sample.md
+cargo run -p fleischwolf --example stream  -- crates/fleischwolf/sample.md
 
 # score HTML output against the latest published docling (installed from PyPI)
 scripts/conformance.sh html

crates/fleischwolf-cli/src/main.rs

@@ -3,7 +3,7 @@
 //! A stand-in for `docling.cli.main`; the full Typer-style CLI (batch mode,
 //! pipeline options) is a later phase.
 //!
-//! Usage: fleischwolf [--strict] [--to md|json] [--images MODE] [--fetch-images] <input-file>
+//! Usage: fleischwolf [--strict] [--to md|json] [--images MODE] [--fetch-images] [--no-stream] <input-file>
 //!   --to md|json       output format (default: md). `json` emits docling-core's
 //!                      native DoclingDocument JSON (export_to_dict).
 //!   --images MODE      picture handling for Markdown (mirrors docling's
@@ -15,7 +15,12 @@
 //!                      the bytes. Off by default; fetches over the network.
 //!   --strict           cleaner, more conformant Markdown instead of byte-for-byte
 //!                      docling-legacy output (Markdown only).
+//!   --no-stream        build the whole document before printing Markdown instead
+//!                      of streaming it page by page. Streaming is the default for
+//!                      Markdown (placeholder/embedded images); JSON and referenced
+//!                      images always use the buffered path.
 
+use std::io::{self, Write};
 use std::path::Path;
 use std::process::ExitCode;
 
@@ -26,13 +31,15 @@ fn main() -> ExitCode {
     let mut to = "md".to_string();
     let mut images = "placeholder".to_string();
     let mut fetch_images = false;
+    let mut no_stream = false;
     let mut bench_warm: Option<usize> = None;
     let mut path: Option<String> = None;
     let mut args = std::env::args().skip(1);
     while let Some(arg) = args.next() {
         match arg.as_str() {
             "--strict" => strict = true,
             "--fetch-images" => fetch_images = true,
+            "--no-stream" => no_stream = true,
             "--to" => to = args.next().unwrap_or_default(),
             "--images" => images = args.next().unwrap_or_default(),
             // Hidden benchmarking aid: load the PDF/image pipeline once, then time
@@ -67,7 +74,7 @@ fn main() -> ExitCode {
     };
 
     let Some(path) = path else {
-        eprintln!("usage: fleischwolf [--strict] [--to md|json] [--images MODE] [--fetch-images] <input-file>");
+        eprintln!("usage: fleischwolf [--strict] [--to md|json] [--images MODE] [--fetch-images] [--no-stream] <input-file>");
         return ExitCode::from(2);
     };
 
@@ -97,11 +104,48 @@ fn main() -> ExitCode {
         };
     }
 
-    let document = match DocumentConverter::new()
+    let converter = DocumentConverter::new()
         .strict(strict)
-        .fetch_images(fetch_images)
-        .convert(source)
-    {
+        .fetch_images(fetch_images);
+
+    // Stream Markdown by default: print each chunk as the converter produces it
+    // (page by page for PDF). JSON needs the whole tree, and the referenced image
+    // mode writes sidecar files, so both keep the buffered path. `--no-stream` opts
+    // back into buffering for the streamable cases too.
+    let is_markdown = matches!(to.as_str(), "md" | "markdown");
+    if is_markdown && image_mode != ImageMode::Referenced && !no_stream {
+        let stream = match converter.convert_streaming_images(source, image_mode) {
+            Ok(s) => s,
+            Err(e) => {
+                eprintln!("error: {e}");
+                return ExitCode::FAILURE;
+            }
+        };
+        let stdout = io::stdout();
+        let mut out = io::BufWriter::new(stdout.lock());
+        for chunk in stream {
+            match chunk {
+                Ok(s) => {
+                    if let Err(e) = out.write_all(s.as_bytes()) {
+                        eprintln!("error: writing output: {e}");
+                        return ExitCode::FAILURE;
+                    }
+                }
+                Err(e) => {
+                    let _ = out.flush();
+                    eprintln!("error: {e}");
+                    return ExitCode::FAILURE;
+                }
+            }
+        }
+        if let Err(e) = out.flush() {
+            eprintln!("error: writing output: {e}");
+            return ExitCode::FAILURE;
+        }
+        return ExitCode::SUCCESS;
+    }
+
+    let document = match converter.convert(source) {
         Ok(result) => result.document,
         Err(e) => {
             eprintln!("error: {e}");

crates/fleischwolf-core/src/lib.rs

@@ -18,4 +18,4 @@ mod markdown;
 
 pub use document::{DoclingDocument, Node, PictureImage, Table};
 pub use labels::DocItemLabel;
-pub use markdown::ImageMode;
+pub use markdown::{ImageMode, MarkdownStreamer};

crates/fleischwolf-core/src/markdown.rs

@@ -99,6 +99,128 @@ fn apply_links(body: &str, links: &[(String, String)]) -> String {
     out
 }
 
+/// Like [`apply_links`] but over a single chunk, consuming from a shared queue so
+/// the same `[anchor](href)` rewriting can be applied incrementally as Markdown is
+/// streamed out. Each queued link is matched (in document order) against `chunk`
+/// and rewritten in place; a link whose anchor is not in this chunk is carried
+/// forward in the queue for a later chunk. Anchors are recovered in document
+/// order and a chunk is always a contiguous run of whole blocks, so this
+/// reproduces [`apply_links`]' single moving cursor: the link lands in whichever
+/// chunk contains its anchor, identically to the buffered path. (A link whose
+/// anchor never appears is carried to the end and dropped — the same no-op
+/// `apply_links` performs for an unlocatable anchor.)
+fn apply_links_chunk(chunk: &str, queue: &mut Vec<(String, String)>) -> String {
+    let mut out = chunk.to_string();
+    let mut cursor = 0usize;
+    let mut carried: Vec<(String, String)> = Vec::new();
+    for (anchor_raw, href) in std::mem::take(queue) {
+        let anchor = anchor_raw
+            .replace('&', "&amp;")
+            .replace('<', "&lt;")
+            .replace('>', "&gt;");
+        if anchor.is_empty() {
+            continue;
+        }
+        if let Some(rel) = out[cursor..].find(&anchor) {
+            let at = cursor + rel;
+            let replacement = format!("[{anchor}]({href})");
+            out.replace_range(at..at + anchor.len(), &replacement);
+            cursor = at + replacement.len();
+        } else {
+            // Not in this chunk; try again when its block is flushed.
+            carried.push((anchor_raw, href));
+        }
+    }
+    *queue = carried;
+    out
+}
+
+/// Incremental Markdown serializer: feed finalized, in-document-order batches of
+/// [`Node`]s and receive Markdown chunks whose concatenation is **byte-identical**
+/// to [`to_markdown_images`] over the same nodes. This is the streaming
+/// counterpart of the buffered serializer — used to emit a document's Markdown in
+/// chunks (e.g. page by page, as the parallel PDF pipeline finishes pages) instead
+/// of building the whole string up front.
+///
+/// Only [`ImageMode::Placeholder`] and [`ImageMode::Embedded`] are streamable:
+/// [`ImageMode::Referenced`] needs a side-channel for the image bytes, which only
+/// the buffered [`to_markdown_images`] provides.
+///
+/// Each [`push`](Self::push) must contain whole blocks in reading order: a caller
+/// must not split a run of list items across two pushes (the run would render as
+/// two separate lists). Finalized PDF page batches already satisfy this.
+pub struct MarkdownStreamer {
+    strict: bool,
+    images: ImageMode,
+    compact_tables: bool,
+    /// Whether any non-empty chunk has been emitted yet (drives `\n\n` joins and
+    /// the trailing newline).
+    emitted_any: bool,
+    /// Recovered links not yet placed (strict mode), consumed in document order.
+    links: Vec<(String, String)>,
+}
+
+impl MarkdownStreamer {
+    /// Create a streamer. `compact_tables` mirrors [`DoclingDocument::compact_tables`].
+    pub fn new(strict: bool, images: ImageMode, compact_tables: bool) -> Self {
+        debug_assert!(
+            images != ImageMode::Referenced,
+            "referenced image mode is not streamable; use to_markdown_images"
+        );
+        Self {
+            strict,
+            images,
+            compact_tables,
+            emitted_any: false,
+            links: Vec::new(),
+        }
+    }
+
+    /// Render one finalized batch of nodes (plus any links recovered from the same
+    /// span, in document order) into the next Markdown chunk. Returns an empty
+    /// string when the batch produces no output (e.g. empty tables/pictures), in
+    /// which case nothing should be written.
+    pub fn push(&mut self, nodes: &[Node], links: &[(String, String)]) -> String {
+        self.links.extend(links.iter().cloned());
+        let mut ctx = Ctx {
+            strict: self.strict,
+            compact_tables: self.compact_tables,
+            images: self.images,
+            // Referenced mode is rejected at construction, so the artifact sink is
+            // never touched.
+            artifacts_dir: String::new(),
+            artifacts: Vec::new(),
+            pic_index: 0,
+        };
+        let mut blocks: Vec<String> = Vec::new();
+        render(nodes, &mut blocks, &mut ctx);
+        if blocks.is_empty() {
+            return String::new();
+        }
+        let mut body = blocks.join("\n\n");
+        if self.strict && !self.links.is_empty() {
+            body = apply_links_chunk(&body, &mut self.links);
+        }
+        let chunk = if self.emitted_any {
+            format!("\n\n{body}")
+        } else {
+            body
+        };
+        self.emitted_any = true;
+        chunk
+    }
+
+    /// Emit the trailing newline that finishes the document (empty if no content
+    /// was produced). Call exactly once, after the final [`push`](Self::push).
+    pub fn finish(self) -> String {
+        if self.emitted_any {
+            "\n".to_string()
+        } else {
+            String::new()
+        }
+    }
+}
+
 /// In `strict` mode, rewrite inline text for readability rather than byte-for-byte
 /// docling fidelity: undo the legacy `\_` underscore escaping, and tighten stray
 /// spaces around punctuation (`[ 37 , 36 ]` → `[37, 36]`, `( x )` → `(x)`). This
@@ -476,6 +598,101 @@ mod tests {
         assert_eq!(doc.export_to_markdown_with(true), "# a_b\n\nx_y\n\n- i_j\n");
     }
 
+    /// Drive a document's nodes through [`MarkdownStreamer`] in the given page
+    /// splits and assert the concatenated chunks equal the buffered serializer.
+    fn assert_stream_matches(
+        doc: &DoclingDocument,
+        strict: bool,
+        images: ImageMode,
+        splits: &[usize],
+    ) {
+        let want = to_markdown_images(doc, strict, images, "artifacts").0;
+        let mut streamer = MarkdownStreamer::new(strict, images, doc.compact_tables);
+        let mut got = String::new();
+        let mut start = 0;
+        for &end in splits {
+            // Links only matter in strict mode; feed them all with the first batch
+            // that has content (document order is preserved by the queue).
+            let links = if start == 0 {
+                doc.links.as_slice()
+            } else {
+                &[]
+            };
+            got.push_str(&streamer.push(&doc.nodes[start..end], links));
+            start = end;
+        }
+        got.push_str(&streamer.push(
+            &doc.nodes[start..],
+            if start == 0 {
+                doc.links.as_slice()
+            } else {
+                &[]
+            },
+        ));
+        got.push_str(&streamer.finish());
+        assert_eq!(
+            got, want,
+            "streamed output diverged (splits={splits:?}, strict={strict})"
+        );
+    }
+
+    #[test]
+    fn streaming_is_byte_identical_to_buffered() {
+        let mut doc = DoclingDocument::new("d");
+        doc.add_heading(1, "Title");
+        doc.add_paragraph("First paragraph.");
+        doc.push(Node::ListItem {
+            ordered: false,
+            number: 1,
+            first_in_list: true,
+            text: "a".into(),
+            level: 0,
+        });
+        doc.push(Node::ListItem {
+            ordered: false,
+            number: 2,
+            first_in_list: false,
+            text: "b".into(),
+            level: 0,
+        });
+        doc.push(Node::Code {
+            language: Some("rust".into()),
+            text: "let x = 1;".into(),
+        });
+        doc.push(Node::Table(Table {
+            rows: vec![vec!["a".into(), "b".into()], vec!["1".into(), "2".into()]],
+        }));
+        doc.push(Node::Picture {
+            caption: Some("Fig 1".into()),
+            image: None,
+        });
+        doc.add_paragraph("Last paragraph.");
+
+        // A run of list items must never straddle a split, so try splits that fall
+        // on safe block boundaries (the streaming PDF assembler guarantees this).
+        for &strict in &[false, true] {
+            for &images in &[ImageMode::Placeholder, ImageMode::Embedded] {
+                for splits in [&[][..], &[1][..], &[2][..], &[4][..], &[1, 4, 6][..]] {
+                    assert_stream_matches(&doc, strict, images, splits);
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn streaming_applies_recovered_links_in_strict_mode() {
+        let mut doc = DoclingDocument::new("d");
+        doc.add_paragraph("See LinkedIn for details.");
+        doc.add_paragraph("And GitHub too.");
+        doc.links = vec![
+            ("LinkedIn".into(), "https://lnkd/".into()),
+            ("GitHub".into(), "https://gh/".into()),
+        ];
+        // The second anchor lives in the second block, so it must be carried across
+        // the page boundary and placed when that block streams out.
+        assert_stream_matches(&doc, true, ImageMode::Placeholder, &[1]);
+    }
+
     #[test]
     fn strict_tightens_punctuation_spacing_legacy_keeps_it() {
         let mut doc = DoclingDocument::new("t");