Content Ingestion Pipeline Flow in source.py Using the content-core Library

The content ingestion pipeline in open_notebook/graphs/source.py is a LangGraph workflow that extracts text and metadata using the content-core library, persists the data as a Source record with optional vector embeddings, and routes the flow through optional content transformations before completion.

The lfnovo/open-notebook repository implements this robust document processing system to handle diverse input formats—from PDFs to YouTube URLs—preparing them for semantic search and AI-driven transformations. This workflow orchestrates the content-core library within a declarative graph structure that can be invoked asynchronously from any service layer.

Pipeline Architecture Overview

The ingestion workflow consists of four discrete stages managed by a StateGraph compiled into source_graph. Each stage handles specific responsibilities: content extraction, persistence, conditional routing, and transformation application.

Stage 1: Content Extraction with content-core

The content_process node (lines 34-88 in open_notebook/graphs/source.py) initializes the pipeline by building a ContentSettings object and injecting default processing engines. It calls content_core.extract_content to pull text, metadata, and transcripts from the input source.

If the content-core library returns soft-failure signals—such as a title equal to "Error" or empty content—the pipeline converts these into exceptions. This ensures the job is marked as failed and can be retried rather than persisting corrupted data.

Stage 2: Source Persistence and Vectorization

The save_source node (lines 110-135) retrieves the Source record by source_id and persists the extracted URL or file path, full text, and title. When the state contains "embed": True, this node triggers source.vectorize() to generate a vector embedding for semantic search capabilities.

This stage materializes the processed content into the database model defined in open_notebook/domain/notebook.py, making it available for notebook associations and retrieval.

Stage 3: Conditional Transformation Routing

The trigger_transformations node (lines 143-159) acts as a conditional router. If the user supplied any Transformation objects in the apply_transformations state field, the function creates Send edges for each transformation, passing the saved Source and specific Transformation configuration to the next node.

If no transformations are requested, the graph bypasses the transformation stage and proceeds directly to END.

Stage 4: Transformation Application

The transform_content node (lines 162-181) executes the generic transformation graph defined in open_notebook/graphs/transformation.py. It processes the source text according to the specified transformation rules and stores the result as an insight attached to the Source record.

This stage enables automated summarization, keyword extraction, or custom AI transformations on the ingested content.

Graph Wiring and State Management

The LangGraph construction (lines 85-100) wires these nodes into a cohesive workflow:

from langgraph.graph import StateGraph, START, END
from open_notebook.graphs.source import content_process, save_source, transform_content, trigger_transformations

workflow = StateGraph(SourceState)
workflow.add_node("content_process", content_process)
workflow.add_node("save_source", save_source)
workflow.add_node("transform_content", transform_content)

workflow.add_edge(START, "content_process")
workflow.add_edge("content_process", "save_source")
workflow.add_conditional_edges(
    "save_source", trigger_transformations, ["transform_content"]
)
workflow.add_edge("transform_content", END)

source_graph = workflow.compile()

The SourceState dictionary manages the workflow context, containing fields such as content_state (extraction parameters), source_id, apply_transformations, embed (boolean flag), and accumulators for the source object and transformation results.

Executing the Pipeline

You can invoke the compiled graph from a FastAPI endpoint or standalone script. The following example demonstrates integration with a web service:

from open_notebook.graphs.source import source_graph
from langgraph.graph import RunnableConfig

async def ingest_source(
    source_id: str, 
    notebook_ids: list[str], 
    apply_transformations: list, 
    embed: bool
):
    state = {
        "content_state": {
            "url": "https://example.com/article",
            "document_engine": "auto",
            "url_engine": "auto",
            "output_format": "markdown",
        },
        "apply_transformations": apply_transformations,
        "source_id": source_id,
        "notebook_ids": notebook_ids,
        "source": None,
        "transformation": [],
        "embed": embed,
    }
    
    result = await source_graph.ainvoke(state, config=RunnableConfig())
    return result["source"]

For local testing without a web framework:

import asyncio
from open_notebook.graphs.source import source_graph
from open_notebook.domain.transformation import Transformation

async def demo():
    state = {
        "content_state": {"url": "https://example.com/article"},
        "apply_transformations": [Transformation(name="summarize", title="Summary")],
        "source_id": "src_123",
        "notebook_ids": ["nb_1"],
        "source": None,
        "transformation": [],
        "embed": True,
    }
    out = await source_graph.ainvoke(state)
    print(f"Title: {out['source'].title}")
    print(f"Preview: {out['source'].full_text[:100]}")

asyncio.run(demo())

Summary

• Four-stage pipeline: The workflow extracts content via content-core, persists to the database, conditionally routes through transformations, and applies AI processing.
• Error handling: Soft failures from content-core are converted to exceptions to prevent corrupted data persistence.
• Vector embedding: Optional semantic search capability triggered by the embed state flag during the save_source stage.
• Declarative graph: LangGraph structure defined in open_notebook/graphs/source.py enables asynchronous execution and conditional branching based on transformation requirements.
• Extensible: The Transformation system allows custom processing of ingested content without modifying the core extraction logic.

Frequently Asked Questions

What is the content-core library's role in the ingestion pipeline?

The content-core library provides the extract_content function that handles format-specific parsing for PDFs, web pages, YouTube videos, and other document types. In source.py, the content_process node wraps this library to convert raw URLs or files into structured text and metadata, normalizing diverse inputs into a consistent format for the Open Notebook system.

How does the pipeline handle extraction failures?

The content_process function detects soft-failure signals from content-core, such as titles equal to "Error" or empty content strings. Rather than persisting these invalid states, the code raises exceptions that mark the LangGraph job as failed. This allows the ingestion task to be retried through the standard error handling mechanisms without polluting the database with incomplete records.

What triggers the transformation stage in the graph?

The trigger_transformations node (lines 143-159) checks the apply_transformations list in the current state. If the list contains Transformation objects, the function returns Send edges targeting the transform_content node for each transformation. If the list is empty, the conditional routing bypasses transformations and proceeds directly to END, completing the workflow after persistence.

How are vector embeddings generated during ingestion?

When the state["embed"] field evaluates to True, the save_source node calls source.vectorize() on the persisted Source object. This method, implemented in open_notebook/domain/notebook.py, generates a vector representation of the extracted text, enabling semantic search capabilities across the notebook's content repository.

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