# Source Content Ingestion Pipeline in Open Notebook: How Files and URLs Are Extracted and Vectorized

> Explore the Open Notebook ingestion pipeline. Learn how files and URLs are extracted, vectorized, and stored in SurrealDB with LangGraph for efficient content processing and search.

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

---

**Open Notebook processes uploaded files and URLs through an asynchronous LangGraph workflow that extracts content using `content_core`, persists records to SurrealDB, and optionally generates searchable embeddings via a background job system with automatic chunking and provider-agnostic AI integration.**

Open Notebook is an open-source knowledge management system that transforms documents and web content into vectorized knowledge. The **source content ingestion pipeline** handles MIME type detection, OCR, content extraction, and embedding generation through a modular architecture built on FastAPI, LangGraph, and SurrealDB.

## How the Pipeline Works: A 7-Step Overview

The ingestion flow moves through distinct architectural layers, from HTTP upload to vector storage:

1. **API Reception**: [`api/routers/sources.py`](https://github.com/lfnovo/open-notebook/blob/main/api/routers/sources.py) receives multipart/form-data or JSON requests
2. **Graph Execution**: [`open_notebook/graphs/source.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py) orchestrates the workflow
3. **Content Extraction**: `content_core` performs MIME type detection and text extraction
4. **Error Handling**: Failed extractions raise `ValueError` for retry logic
5. **Persistence**: `Source` records are saved to SurrealDB via the `save_source` node
6. **Job Queueing**: `Source.vectorize()` submits `embed_source` commands asynchronously
7. **Embedding Generation**: [`commands/embedding_commands.py`](https://github.com/lfnovo/open-notebook/blob/main/commands/embedding_commands.py) chunks text and generates vectors

## Step 1: API Reception and File Handling

Incoming requests hit the FastAPI endpoint in [`api/routers/sources.py`](https://github.com/lfnovo/open-notebook/blob/main/api/routers/sources.py), which delegates to [`api/sources_service.py`](https://github.com/lfnovo/open-notebook/blob/main/api/sources_service.py). The service handles file uploads by storing binaries in `UPLOADS_FOLDER` before constructing a `SourceCreate` model.

```python

# POST /api/sources (api/routers/sources.py)

source_data, upload_file = parse_source_form_data(...)
if upload_file:
    file_path = await save_uploaded_file(upload_file)
    source_data.file_path = file_path
result = SourcesService().create_source(**source_data.dict())

```

The endpoint automatically converts form strings to booleans (e.g., `embed`, `async_processing`) and validates JSON fields like `notebooks` and `transformations`.

## Step 2: The LangGraph Source Workflow

The core orchestration happens in [`open_notebook/graphs/source.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py) using LangGraph's `StateGraph`. The workflow compiles three primary nodes with conditional edges:

```python

# open_notebook/graphs/source.py

workflow = StateGraph(SourceState)
workflow.add_node("content_process", content_process)   # extraction

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

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

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 graph executes asynchronously via `await source_graph.ainvoke(state)`, ensuring non-blocking processing of large files.

## Step 3: Content Extraction with content_core

The `content_process` node delegates to `content_core.extract_content`, which detects MIME types and applies appropriate extractors:

- **PDFs**: OCR and text extraction
- **YouTube URLs**: Subtitle fetching
- **Web pages**: HTML body parsing

```python

# content_process node implementation

content_state = state["content_state"]
content_state["url_engine"] = content_settings.default_content_processing_engine_url or "auto"
content_state["document_engine"] = content_settings.default_content_processing_engine_doc or "auto"
content_state["output_format"] = "markdown"

processed_state = await extract_content(content_state)  # Returns ProcessSourceState

```

The result populates a `ProcessSourceState` object containing `title`, `content`, `url`, and `file_path`.

## Step 4: Error Handling and Validation

If extraction fails, `content_core` returns a title of `"Error"` with content prefixed by `"Failed to extract content:"`. The pipeline converts this soft failure into a hard exception to enable retry logic:

```python
if processed_state.title == "Error" and processed_state.content.startswith("Failed to extract content:"):
    raise ValueError("Could not extract content from this source …")

```

This ensures the job is marked failed in the UI rather than silently storing error strings as content.

## Step 5: Persisting Sources to SurrealDB

The `save_source` node populates the `Source` domain model and persists it via `source.save()`:

```python
source.asset = Asset(url=processed_state.url,
                    file_path=processed_state.file_path)
source.full_text = processed_state.content
if processed_state.title and (not source.title or source.title == "Processing..."):
    source.title = processed_state.title
await source.save()

```

The `Source` class in [`open_notebook/domain/notebook.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/domain/notebook.py) handles SurrealDB serialization and timestamp hooks.

## Step 6: Asynchronous Vectorization

When `embed=True` is passed during creation, the `Source.vectorize()` method queues a background job:

```python

# open_notebook/domain/notebook.py

command_id = submit_command(
    "open_notebook", "embed_source", {"source_id": str(self.id)}
)
return str(command_id)

```

This fire-and-forget pattern prevents HTTP worker thread blocking by delegating to the `surreal_commands` queue.

## Step 7: Embedding Generation and Chunking

The `embed_source` command in [`commands/embedding_commands.py`](https://github.com/lfnovo/open-notebook/blob/main/commands/embedding_commands.py) implements the vectorization logic with exponential backoff retries:

```python
@command("embed_source", app="open_notebook", retry={…})
async def embed_source_command(input: EmbedSourceInput) -> EmbedSourceOutput:
    source = await Source.get(input.source_id)
    # 1. Detect content type → chunk_text()

    # 2. Generate embeddings → generate_embeddings()

    # 3. Bulk UPSERT into source_embedding table

```

The pipeline uses [`open_notebook/utils/chunking.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/utils/chunking.py) to detect content type and split text appropriately, while [`open_notebook/utils/embedding.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/utils/embedding.py) provides provider-agnostic embedding generation via the Esperanto library.

## Practical Usage Examples

### Uploading a File via cURL

Submit a PDF for ingestion with automatic embedding:

```bash
curl -X POST http://localhost:5055/sources \
  -F type=file \
  -F notebook_id=nb123 \
  -F file=@/path/to/document.pdf \
  -F embed=true

```

The response includes the source ID and a `command_id` for polling embedding status.

### Ingesting a URL via JSON

Create a source from a web article:

```bash
curl -X POST http://localhost:5055/sources \
  -H "Content-Type: application/json" \
  -d '{
        "type": "link",
        "url": "https://example.com/article.html",
        "embed": true,
        "title": "Interesting article"
      }'

```

### Checking Embedding Status

Poll the command endpoint to track asynchronous processing:

```bash
curl http://localhost:5055/commands/abcd-1234

# Response: {"status":"completed","result":{…}}

```

### Programmatic Vectorization with Python

Manually trigger embedding for an existing source:

```python
from open_notebook.domain.notebook import Source

source = await Source.get("source-id-xyz")
job_id = await source.vectorize()  # Returns command ID

print(f"Embedding job submitted, ID={job_id}")

```

## Summary

- **Entry Point**: [`api/routers/sources.py`](https://github.com/lfnovo/open-notebook/blob/main/api/routers/sources.py) handles multipart uploads and JSON payloads, storing files in `UPLOADS_FOLDER`
- **Orchestration**: [`open_notebook/graphs/source.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py) runs a LangGraph workflow with extraction, persistence, and optional transformation nodes
- **Extraction**: `content_core` performs MIME type detection, OCR, and content extraction, returning structured `ProcessSourceState`
- **Error Handling**: Failures raise `ValueError` when titles equal `"Error"` with specific prefixes, enabling retry logic
- **Persistence**: `Source` models in [`open_notebook/domain/notebook.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/domain/notebook.py) save to SurrealDB with asset metadata and full text
- **Vectorization**: Asynchronous `embed_source` commands handle chunking via [`open_notebook/utils/chunking.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/utils/chunking.py) and embedding via [`open_notebook/utils/embedding.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/utils/embedding.py)

## Frequently Asked Questions

### How does Open Notebook handle PDF extraction and OCR?

The pipeline delegates to `content_core.extract_content`, which automatically detects MIME types and applies OCR when processing PDF documents. This occurs in the `content_process` node of [`open_notebook/graphs/source.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py), where the `document_engine` parameter defaults to `"auto"` to select the appropriate extraction strategy.

### What happens if content extraction fails?

When `content_core` cannot extract text, it returns a `ProcessSourceState` with `title="Error"` and content prefixed by `"Failed to extract content:"`. The `content_process` node detects this pattern and raises a `ValueError`, causing the LangGraph workflow to fail visibly. This allows the UI to surface the error and enable retry mechanisms rather than storing corrupted data.

### Can I ingest content without automatically generating embeddings?

Yes. The `embed` parameter is optional in both the REST API and Python SDK. When `embed=false` (or omitted), the pipeline stops after persisting the `Source` record. You can later trigger vectorization manually by calling `source.vectorize()` on the `Source` instance, which submits an `embed_source` command to the background queue.

### How does the embedding pipeline handle large documents?

The `embed_source` command in [`commands/embedding_commands.py`](https://github.com/lfnovo/open-notebook/blob/main/commands/embedding_commands.py) uses `detect_content_type` and `chunk_text` from [`open_notebook/utils/chunking.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/utils/chunking.py) to split documents into appropriate segments before embedding. It then calls `generate_embeddings` from [`open_notebook/utils/embedding.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/utils/embedding.py) to process chunks through the configured AI provider, performing bulk upserts into the `source_embedding` table with exponential jitter retries for transient failures.