# How the Source Ingestion Pipeline Extracts Content and Generates Vector Embeddings in Open Notebook

> Discover how Open Notebook's LangGraph workflow extracts content, stores data, and generates vector embeddings with background jobs and retries. Explore the lfnovo/open-notebook repository.

- Repository: [Luis Novo/open-notebook](https://github.com/lfnovo/open-notebook)
- Tags: deep-dive
- Published: 2026-06-28

---

**Open Notebook processes raw inputs through a three-stage LangGraph workflow that extracts content using content-core, persists data to the database, and asynchronously generates vector embeddings via a background job queue with automatic retry logic.**

The source ingestion pipeline in the `lfnovo/open-notebook` repository transforms URLs, uploaded files, and plain text into searchable vector embeddings. This architecture decouples content extraction from embedding generation to maintain API responsiveness while ensuring reliable processing through asynchronous job queues.

## The Three-Stage Pipeline Architecture

The pipeline is implemented as a LangGraph workflow defined in [[`open_notebook/graphs/source.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py)](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py). The `source_graph` wires together three distinct nodes that handle extraction, persistence, and optional embedding generation.

### Stage 1: Content Extraction with content-core

The `content_process` node handles the initial ingestion of raw input. It accepts a `ProcessSourceState` object containing a URL, file path, or plain text, then delegates the heavy lifting to **content-core**.

```python
processed_state = await extract_content(content_state)

```

This call converts documents to markdown format and returns structured metadata. The node supports multiple input types and normalizes them into a consistent internal representation before passing control to the next stage.

### Stage 2: Persisting Source Metadata

The `save_source` node receives the processed state and writes the extracted data back to the `Source` record. According to the source code in [`open_notebook/graphs/source.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py), this node performs three critical operations:

- Stores the asset reference (`url` or `file_path`) in the `Source.asset` field
- Saves the full extracted text to `source.full_text`
- Updates the source title if present in the metadata

```python
source.asset = Asset(url=content_state.url, file_path=content_state.file_path)
source.full_text = content_state.content
await source.save()

```

### Stage 3: Asynchronous Vector Embedding Generation

When the ingestion request includes `embed=True`, the `save_source` node triggers the embedding phase. Rather than processing inline, it calls `source.vectorize()` which submits an asynchronous job to the command queue.

The `Source.vectorize` method in [[`open_notebook/domain/notebook.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/domain/notebook.py)](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/domain/notebook.py) submits an **embed_source** command:

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

```

This approach returns immediately to the API caller while the heavy computational work happens in the background.

## Inside the Embedding Command

The actual embedding work is performed by `embed_source_command` in [[`commands/embedding_commands.py`](https://github.com/lfnovo/open-notebook/blob/main/commands/embedding_commands.py)](https://github.com/lfnovo/open-notebook/blob/main/commands/embedding_commands.py). This command handler implements a robust six-step workflow with automatic retry logic (up to five attempts for transient failures).

The process follows these steps:

1. **Load the source** using `Source.get` to retrieve the full text and metadata
2. **Delete existing embeddings** for idempotency, ensuring no duplicate vectors
3. **Detect content type** via `detect_content_type` to inform chunking strategy
4. **Chunk the text** using `chunk_text` with content-type-aware splitting
5. **Generate embeddings** in batches via `generate_embeddings` to optimize API calls
6. **Bulk-insert** the resulting vectors as `source_embedding` rows

The command is decorated with a retry policy that handles network timeouts and transient failures automatically.

## Triggering the Pipeline via API

The API entry point in [[`api/routers/sources.py`](https://github.com/lfnovo/open-notebook/blob/main/api/routers/sources.py)](https://github.com/lfnovo/open-notebook/blob/main/api/routers/sources.py) exposes the pipeline through the `POST /sources` endpoint. The handler constructs a `content_state` dictionary and supports two execution modes:

**Synchronous mode** executes `process_source` immediately using `execute_command_sync`, blocking until extraction completes.

**Asynchronous mode** creates the `Source` record first, then enqueues the `process_source` command via `CommandService.submit_command_job`, returning a `command_id` for tracking.

Both paths execute the same LangGraph graph, guaranteeing identical extraction and embedding behavior regardless of invocation method.

### Creating a Source with Embedding

```python
import requests

payload = {
    "type": "link",
    "url": "https://example.com/article",
    "title": "Example article",
    "embed": True,
    "notebooks": ["notebook-123"]
}
resp = requests.post(
    "http://localhost:5055/sources",
    json=payload,
    headers={"Content-Type": "application/json"}
)
print(resp.json())

# Returns source ID and command ID for tracking

```

### Tracking Embedding Progress

```python
command_id = resp.json()["command_id"]
status_resp = requests.get(f"http://localhost:5055/commands/{command_id}")
print(status_resp.json())

# Status transitions: "running" → "completed"

```

### Direct Command Invocation

```python
from commands.embedding_commands import embed_source_command, EmbedSourceInput

result = await embed_source_command(
    EmbedSourceInput(source_id="source-abc")
)
print(result.success, result.chunks_created)

```

## Summary

- **Content extraction** happens in the `content_process` node using `content-core` to convert inputs to markdown
- **Persistence** occurs in `save_source`, which stores full text and metadata in the `Source` model
- **Vector embedding** is handled asynchronously via `Source.vectorize()` submitting to the `embed_source_command` queue
- The embedding command performs content-type-aware chunking, batch embedding generation, and bulk insertion with automatic retry logic
- The API supports both sync and async invocation modes through [`api/routers/sources.py`](https://github.com/lfnovo/open-notebook/blob/main/api/routers/sources.py)

## Frequently Asked Questions

### How does the pipeline handle different content types like PDFs versus web pages?

The `content_process` node delegates to `content-core` for extraction, which handles multiple formats. During embedding, the `detect_content_type` function in [`commands/embedding_commands.py`](https://github.com/lfnovo/open-notebook/blob/main/commands/embedding_commands.py) determines the optimal chunking strategy, ensuring code files split differently than prose documents.

### What happens if the embedding service fails temporarily?

The `embed_source_command` includes a retry decorator that automatically retries transient failures up to five times. This handles network timeouts and temporary service outages without manual intervention, ensuring reliable processing of large document batches.

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

Yes. When creating a source via the API, omit the `embed` parameter or set it to `False`. The pipeline will extract and save the content but skip the `source.vectorize()` call. You can trigger embedding later by calling the vectorize method directly or re-submitting the embed command.

### How does the pipeline maintain data consistency between the source and its embeddings?

The embedding command first deletes all existing `source_embedding` rows for the given source ID before generating new chunks. This idempotent approach ensures that re-processing a source always results in a clean state, preventing orphaned vectors or duplicate embeddings when content updates.