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

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 receives multipart/form-data or JSON requests
2. Graph Execution: 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 chunks text and generates vectors

Step 1: API Reception and File Handling

Incoming requests hit the FastAPI endpoint in api/routers/sources.py, which delegates to api/sources_service.py. The service handles file uploads by storing binaries in UPLOADS_FOLDER before constructing a SourceCreate model.

# 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 using LangGraph's StateGraph. The workflow compiles three primary nodes with conditional edges:

# 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

# 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:

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():

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

# 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 implements the vectorization logic with exponential backoff retries:

@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 to detect content type and split text appropriately, while 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:

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:

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:

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

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

Programmatic Vectorization with Python

Manually trigger embedding for an existing source:

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 handles multipart uploads and JSON payloads, storing files in UPLOADS_FOLDER
• Orchestration: 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 save to SurrealDB with asset metadata and full text
• Vectorization: Asynchronous embed_source commands handle chunking via open_notebook/utils/chunking.py and embedding via 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, 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 uses detect_content_type and chunk_text from open_notebook/utils/chunking.py to split documents into appropriate segments before embedding. It then calls generate_embeddings from 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.

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