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

> Explore the content ingestion pipeline flow in open notebook's source.py. Learn how the content-core library extracts, persists, and transforms content.

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

---

**The content ingestion pipeline in [`open_notebook/graphs/source.py`](https://github.com/lfnovo/open-notebook/blob/main/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`](https://github.com/lfnovo/open-notebook/blob/main/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`](https://github.com/lfnovo/open-notebook/blob/main/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`](https://github.com/lfnovo/open-notebook/blob/main/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:

```python
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:

```python
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:

```python
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`](https://github.com/lfnovo/open-notebook/blob/main/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`](https://github.com/lfnovo/open-notebook/blob/main/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`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/domain/notebook.py), generates a vector representation of the extracted text, enabling semantic search capabilities across the notebook's content repository.