Content Ingestion Flow in Open Notebook's LangGraph Source Workflow

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. 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), which prepares extraction settings and delegates to the content_core library.

Settings preparation (lines 35‑51): 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).

Speech-to-Text configuration (lines 62‑73): If configured, the default STT model provider and name are added to the state.

Extraction execution (line 78): 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): 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).

Phase 2: Source Persistence via save_source

The save_source node (lines 110‑140) 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).
2. Populate fields: Assigns the extracted URL/file path to asset (line 119), raw Markdown to full_text (line 120), and updates the title only if it remains a placeholder (lines 122‑124).
3. Database write: Calls await source.save() (line 126).
4. Optional embedding: If state["embed"] is True, invokes source.vectorize() to generate semantic search vectors (lines 131‑138).
5. Return: Places the saved Source object back onto the graph state as source (line 140).

Phase 3: Optional Transformations

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

Branch decision (lines 143‑160): 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): 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): The resulting insight attaches to the source through source.add_insight.

The conditional edge is declared using add_conditional_edges (lines 94‑96).

Phase 4: Finalization

The workflow terminates at the END node (lines 192‑199) after all transformations complete or immediately after persistence if no transformations are requested.

Graph State and Data Flow

The SourceState TypedDict (lines 19‑27) carries all mutable data between steps:

# 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:

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:

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:

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: Defines the LangGraph state machine, nodes, and edges (source.py)
• open_notebook/domain/content_settings.py: Contains ContentSettings defaults for extraction engines (content_settings.py)
• open_notebook/domain/notebook.py: Implements Source model and vectorize() method (notebook.py)
• open_notebook/graphs/transformation.py: Separate LangGraph for content transformations (transformation.py)
• content_core/__init__.py: Provides extract_content for OCR, PDF parsing, and STT (content_core)

Summary

• The content ingestion flow follows four distinct phases: extraction, persistence, optional transformation, and finalization.
• content_process (lines 34‑78) handles URL fetching, OCR, and speech-to-text conversion through the content_core library.
• save_source (lines 110‑140) persists data to the database and conditionally calls source.vectorize() for semantic search.
• trigger_transformations (lines 143‑160) conditionally routes to the transformation graph based on user-selected operations.
• The SourceState TypedDict (lines 19‑27) 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. 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, 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) 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, 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) 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}