# Content Ingestion Flow in Open Notebook's LangGraph Source Workflow

> Explore the content ingestion flow within Open Notebooks LangGraph source workflow. Learn how web pages, PDFs, and videos are processed through extraction, persistence, transformation, and finalization.

- Repository: [Luis Novo/open-notebook](https://github.com/lfnovo/open-notebook)
- Tags: how-to-guide
- Published: 2026-06-19

---

**Open Notebook processes web pages, PDFs, and videos through a four-phase LangGraph state machine that extracts content, persists it to the database, optionally transforms it, and finalizes the ingestion.**

The complete **content ingestion flow** orchestrates raw material processing in the `lfnovo/open-notebook` repository through a compiled LangGraph defined in [`open_notebook/graphs/source.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py). This workflow coordinates extraction engines, database persistence, and AI-driven transformations using a strongly typed state object that passes mutable data between nodes.

## The Four Phases of Content Ingestion

The graph implements a linear progression with one conditional branch, handling everything from URL parsing to vector embedding.

### Phase 1: Content Extraction via `content_process`

The workflow begins at the `content_process` node (lines [34‑78](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L34-L78)), which prepares extraction settings and delegates to the `content_core` library.

**Settings preparation** (lines [35‑51](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L35-L51)): The function initializes a `ContentSettings` instance with default engines. It then injects `url_engine` and `document_engine` into the `content_state` dictionary (lines [54‑60](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L54-L60)).

**Speech-to-Text configuration** (lines [62‑73](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L62-L73)): If configured, the default STT model provider and name are added to the state.

**Extraction execution** (line [78](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L78)): The `extract_content(content_state)` function performs the heavy lifting—handling OCR, PDF parsing, video transcript extraction, and HTML-to-Markdown conversion.

**Error handling** (lines [80‑105](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L80-L105)): The node validates results, raising `ValueError` for soft failures (titles equal to "Error" or empty content) to trigger retry logic. The processed `ProcessSourceState` is returned to the graph state under the key `content_state` (line [107](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L107)).

### Phase 2: Source Persistence via `save_source`

The `save_source` node (lines [110‑140](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L110-L140)) persists extracted content to the database and optionally creates vector embeddings.

1. **Load existing record**: Uses `Source.get(state["source_id"])` to fetch the database entity (line [114](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L114)).
2. **Populate fields**: Assigns the extracted URL/file path to `asset` (line [119](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L119)), raw Markdown to `full_text` (line [120](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L120)), and updates the title only if it remains a placeholder (lines [122‑124](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L122-L124)).
3. **Database write**: Calls `await source.save()` (line [126](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L126)).
4. **Optional embedding**: If `state["embed"]` is `True`, invokes `source.vectorize()` to generate semantic search vectors (lines [131‑138](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L131-L138)).
5. **Return**: Places the saved `Source` object back onto the graph state as `source` (line [140](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L140)).

### Phase 3: Optional Transformations

After persistence, the graph conditionally branches based on user-selected transformations.

**Branch decision** (lines [143‑160](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L143-L160)): The `trigger_transformations` function inspects `state["apply_transformations"]`. If the list is empty, the flow proceeds directly to `END`; otherwise, it emits a `Send` for each transformation to the `transform_content` node.

**Transformation execution** (lines [170‑173](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L170-L173)): The `transform_content` node receives a single `Transformation` object and the current `Source`, forwarding the text to the dedicated `transform_graph` via `transform_graph.ainvoke`.

**Result storage** (line [174](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L174)): The resulting insight attaches to the source through `source.add_insight`.

The conditional edge is declared using `add_conditional_edges` (lines [94‑96](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L94-L96)).

### Phase 4: Finalization

The workflow terminates at the `END` node (lines [192‑199](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L192-L199)) after all transformations complete or immediately after persistence if no transformations are requested.

## Graph State and Data Flow

The `SourceState` TypedDict (lines [19‑27](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L19-L27)) carries all mutable data between steps:

```python

# Conceptual structure of SourceState

{
    "content_state": ProcessSourceState,    # Extraction results

    "apply_transformations": List[Transformation],  # User-selected transforms

    "source_id": str,                      # Database identifier

    "notebook_ids": List[str],             # Associated notebooks

    "source": Optional[Source],            # Populated after save_source

    "transformation": List,              # Internal accumulation buffer

    "embed": bool,                         # Vectorization flag

}

```

The graph construction wires these nodes together:

```python
workflow = StateGraph(SourceState)               # L85

workflow.add_node("content_process", content_process)   # L88

workflow.add_node("save_source", save_source)           # L89

workflow.add_node("transform_content", transform_content) # L90

workflow.add_edge(START, "content_process")          # L92

workflow.add_edge("content_process", "save_source")   # L93

workflow.add_conditional_edges(                     # L94-96

    "save_source", trigger_transformations,
    ["transform_content"]
)
workflow.add_edge("transform_content", END)          # L97

source_graph = workflow.compile()                    # L100

```

## Executing the Ingestion Graph

Invoke the compiled graph asynchronously from API layers or scripts:

```python
from open_notebook.graphs.source import source_graph
from langgraph.types import RunnableConfig

async def ingest_source(
    source_id: str,
    content_state: dict,
    embed: bool = True,
    transformations: list = [],
) -> dict:
    init_state = {
        "content_state": content_state,
        "apply_transformations": transformations,
        "source_id": source_id,
        "notebook_ids": [],
        "source": None,
        "transformation": [],
        "embed": embed,
    }
    
    result = await source_graph.ainvoke(
        init_state, 
        config=RunnableConfig()
    )
    return result  # Contains persisted Source under key "source"

```

Trigger with transformations:

```python
from open_notebook.domain.transformation import Transformation

summarise = await Transformation.get_by_name("summarise")
result = await ingest_source(
    source_id="src_123",
    content_state={"url": "https://example.com/article"},
    embed=True,
    transformations=[summarise],
)

```

## Key Source Files

- **[`open_notebook/graphs/source.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py)**: Defines the LangGraph state machine, nodes, and edges ([source.py](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py))
- **[`open_notebook/domain/content_settings.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/domain/content_settings.py)**: Contains `ContentSettings` defaults for extraction engines ([content_settings.py](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/domain/content_settings.py))
- **[`open_notebook/domain/notebook.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/domain/notebook.py)**: Implements `Source` model and `vectorize()` method ([notebook.py](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/domain/notebook.py))
- **[`open_notebook/graphs/transformation.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/transformation.py)**: Separate LangGraph for content transformations ([transformation.py](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/transformation.py))
- **[`content_core/__init__.py`](https://github.com/lfnovo/open-notebook/blob/main/content_core/__init__.py)**: Provides `extract_content` for OCR, PDF parsing, and STT ([content_core](https://github.com/lfnovo/open-notebook/blob/main/content_core/__init__.py))

## Summary

- The **content ingestion flow** follows four distinct phases: extraction, persistence, optional transformation, and finalization.
- **`content_process`** (lines [34‑78](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L34-L78)) handles URL fetching, OCR, and speech-to-text conversion through the `content_core` library.
- **`save_source`** (lines [110‑140](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L110-L140)) persists data to the database and conditionally calls `source.vectorize()` for semantic search.
- **`trigger_transformations`** (lines [143‑160](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L143-L160)) conditionally routes to the transformation graph based on user-selected operations.
- The **`SourceState`** TypedDict (lines [19‑27](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L19-L27)) maintains all mutable state across the LangGraph workflow.

## Frequently Asked Questions

### How does the source.py graph handle extraction errors?

The `content_process` node validates extraction results at lines [80‑105](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L80-L105). If the title equals "Error" or the content is empty, it raises a `ValueError` that propagates upstream, causing the graph invocation to fail and allowing the calling API to implement retry logic or surface the error to users.

### What determines whether a source gets vectorized for semantic search?

The `embed` boolean flag in the `SourceState` controls vectorization. When `state["embed"]` is `True`, the `save_source` node calls `source.vectorize()` at lines [131‑138](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L131-L138), creating embeddings for the extracted text that enable semantic search across notebook contents.

### Can I add custom transformations to the ingestion workflow?

Yes. The `trigger_transformations` function (lines [143‑160](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L143-L160)) accepts any list of `Transformation` objects in the `apply_transformations` state key. Each transformation spawns a separate `transform_content` node execution that delegates to the `transform_graph` defined in [`open_notebook/graphs/transformation.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/transformation.py), allowing you to implement custom summaries, translations, or metadata extraction.

### How does the graph maintain state between asynchronous operations?

The workflow uses the `SourceState` TypedDict (lines [19‑27](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py#L19-L27)) as a shared data structure. LangGraph passes this state dictionary between nodes immutably, with each node returning updates that merge into the global state. This design ensures that `content_process`, `save_source`, and `transform_content` all access the same `source_id`, `content_state`, and `notebook_ids` without side effects.