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

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). 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.

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, 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

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) submits an embed_source command:

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). 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) 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

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

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

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

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 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.

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