Open Notebook Content Processing Pipeline: How content-core Extracts and Processes Data

Open Notebook ingests raw files and web pages through a LangGraph workflow that delegates extraction to the content-core library, handling documents, web scraping, and audio transcription before embedding results for semantic search.

The lfnovo/open-notebook repository implements a robust content processing pipeline that transforms unstructured data into queryable knowledge. At the heart of this system lies a sophisticated LangGraph workflow defined in open_notebook/graphs/source.py, which orchestrates extraction, normalization, and persistence using the specialized content-core library.

Three-Stage Content Processing Pipeline

The workflow proceeds through three logical stages, each handling distinct aspects of data ingestion and preparation.

Stage 1: Configuration and Speech-to-Text Setup

The pipeline initializes by preparing a content_state dictionary with extraction parameters. The system queries ModelManager for default Speech-to-Text (STT) models and injects ContentSettings to specify processing engines for documents and URLs.

In open_notebook/graphs/source.py (lines 62-71), the workflow retrieves the default STT model configuration:

from open_notebook.ai.models import ModelManager, Model

model_manager = ModelManager()
defaults = await model_manager.get_defaults()
stt = await Model.get(defaults.default_speech_to_text_model)

These settings populate the content_state with audio_provider and audio_model keys, enabling downstream transcription capabilities.

Stage 2: Extraction via content_core.extract_content

The content_process node (lines 34-78 in source.py) calls extract_content(content_state), which performs the heavy lifting:

• Document handling: PDFs, Word documents, and markdown files parse into plain text
• Web scraping: HTML pages fetch and strip boilerplate content
• Audio/Video transcription: YouTube videos and audio files process through the configured STT model or fallback to the internal Whisper model
• Output formatting: Results coerce to Markdown via output_format = "markdown"

The function returns a ProcessSourceState dataclass from content_core.common containing title, content, url, file_path, and error diagnostics.

Stage 3: Error Handling, Persistence, and Embedding

The save_source node validates extraction success by checking for sentinel values where title == "Error" and content starts with "Failed to extract content:" (lines 80-92). When detected, the pipeline raises a ValueError to mark the job as failed and eligible for retry.

Upon successful extraction, the system:

• Persists the extracted text to the Source record (source.full_text = content_state.content)
• Stores metadata including URL and file path
• Optionally calls source.vectorize() when embed=True to vector-index content for semantic search (lines 31-38)

LangGraph Workflow Architecture

The workflow graph chains nodes declaratively as: START → content_process → save_source → (optional) transform_content → END.

This structure enables asynchronous execution with built-in retry logic. The StateGraph assembly (lines 84-99 in source.py) ensures deterministic processing paths that can be monitored and extended through custom transformations.

Implementation Examples

Running the Pipeline Manually

Execute the graph asynchronously from scripts or REPLs:

from open_notebook.graphs.source import source_graph
from open_notebook.domain.notebook import Source

state = {
    "content_state": {
        "url": "https://example.com/article.html",
        # optional: "file_path": "/tmp/upload.pdf"

    },
    "apply_transformations": [],
    "source_id": "src_12345",
    "notebook_ids": ["nb_1"],
    "source": await Source.get("src_12345"),
    "embed": True,
}

result = await source_graph.ainvoke(state)
print("Processed source:", result["source"])

Customizing Speech-to-Text for Video Content

Override STT providers for specific YouTube extractions:

custom_state = {
    "content_state": {
        "url": "https://www.youtube.com/watch?v=abcde",
        "audio_provider": stt.provider,
        "audio_model": stt.name,
    },
    "source_id": "src_12345",
    "notebook_ids": ["nb_1"],
    "embed": True,
}

await source_graph.ainvoke(custom_state)

Adding Post-Processing Transformations

Queue transformations that execute via the separate transform_graph:

from open_notebook.domain.transformation import Transformation

summary = Transformation(
    name="summarise", 
    title="Summary", 
    prompt="Summarise the text."
)

state["apply_transformations"] = [summary]
await source_graph.ainvoke(state)

Key Source Files and Components

• open_notebook/graphs/source.py: Main LangGraph workflow orchestrating extraction, saving, embedding, and optional transformations
• content_core/__init__.py: External library providing extract_content and ProcessSourceState dataclasses
• open_notebook/domain/content_settings.py: Centralizes defaults for content-processing engines and STT preferences
• open_notebook/ai/models.py: Supplies default model configuration including Speech-to-Text settings
• open_notebook/graphs/transformation.py: Defines the separate graph executing user-specified transformations on extracted text

Summary

• The pipeline uses LangGraph to orchestrate a deterministic, asynchronous workflow for content ingestion
• content-core handles heterogeneous inputs including documents, web pages, and media with automatic transcription
• The system supports configurable STT models, markdown output formatting, and semantic search vectorization
• Extensible architecture allows custom transformations via a separate processing graph

Frequently Asked Questions

What file types does the content processing pipeline support?

The pipeline supports PDFs, Word documents, markdown files, HTML web pages, and YouTube videos. Audio and video content processes using either the configured Speech-to-Text model or the internal Whisper fallback, with output standardized to Markdown format.

How does the pipeline handle extraction errors?

The system checks for sentinel values where title == "Error" and content starts with "Failed to extract content:". When detected in open_notebook/graphs/source.py (lines 80-92), it raises a ValueError to mark the job as failed and eligible for retry, preventing corrupted data from persisting to the database.

Can I customize the Speech-to-Text provider for video transcription?

Yes. Query the ModelManager for the default STT model and inject the audio_provider and audio_model fields into the content_state dictionary before invoking source_graph.ainvoke(). This overrides the default engine for specific extractions while maintaining global defaults for other operations.

What is the purpose of the transformation graph in the pipeline?

The transformation graph, defined in open_notebook/graphs/transformation.py, executes user-defined transformations such as summarization or entity extraction after initial content extraction completes. The workflow conditionally routes to this graph when apply_transformations contains valid Transformation objects, enabling post-processing without blocking the core extraction path.

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