Understanding the Sources Service Architecture for Content Ingestion in Open Notebook

The sources_service architecture implements a three-layer pipeline separating API handling, asynchronous command orchestration, and LangGraph-based workflow execution to process content from URLs, files, or raw text while maintaining API responsiveness.

The lfnovo/open-notebook repository implements a robust content ingestion system designed to handle diverse source materials without blocking the REST API. The sources_service architecture for content ingestion follows a layered design pattern that delegates heavy processing to background workers while providing clients with both synchronous and asynchronous interfaces.

Three-Layer Architecture Overview

The content ingestion pipeline separates concerns across three distinct layers: an API façade, a command orchestration layer, and a LangGraph workflow engine.

API Layer – SourcesService

Located in api/sources_service.py, the SourcesService provides a thin façade over the internal REST API client. It defines high-level methods including create_source, get_source, update_source, and delete_source.

When a client requests a new source (uploading a file, providing a URL, or sending raw text), the create_source method forwards the request to the backend via api_client.create_source. The response is normalized into a Source object for synchronous operations, or a SourceProcessingResult when async processing is requested. This layer ensures the REST API remains fast by immediately returning control to the client while heavy work proceeds in the background.

Command Layer – Surreal-Commands Integration

The asynchronous processing logic resides in commands/source_commands.py within the process_source_command function. This layer runs as an asynchronous job managed by surreal-commands, providing reliable retry semantics including exponential back-off, max attempts, and distinct handling of permanent versus transient errors.

The command receives a SourceProcessingInput containing the source ID, raw content_state (URL, file path, or text), target notebook IDs, transformation IDs, and an embedding flag. It loads requested Transformation objects, retrieves the Source record while storing a command ID reference for UI polling, and invokes the LangGraph workflow. Upon completion, it extracts insights via source.get_insights() and returns a SourceProcessingOutput containing timing metrics, insight counts, and embedding information.

Workflow Layer – LangGraph StateGraph

The actual content extraction and processing logic is encapsulated in open_notebook/graphs/source.py as a StateGraph (LangGraph) called source_graph. This workflow orchestrates content extraction, optional transformations, and vector embedding through three distinct nodes.

LangGraph Workflow Nodes

The source_graph implements a state-machine architecture with specific responsibilities distributed across three nodes wired as: START → content_process → save_source → conditional → transform_content → END.

content_process Node

This node calls content-core via extract_content to fetch documents or text, applying default content-processing settings and optional speech-to-text models. It validates extraction results and raises ValueError for unrecoverable failures such as unreachable URLs.

save_source Node

This node persists extracted text and metadata into the Source record, updates placeholder titles, and conditionally triggers source.vectorize() when embed=True. The embedding operation runs asynchronously after the source is saved, providing fire-and-forget semantics for vector generation used in semantic search.

transform_content Node

This optional node executes only when the caller supplies transformation IDs. It runs user-selected transformations using the transformation graph (transform_graph), with each transformation generating an insight attached to the source via source.add_insight.

Practical Implementation Examples

Creating a Source Synchronously

from api.sources_service import sources_service

# Create a source from a public URL, embed the content, and attach it to two notebooks

result = sources_service.create_source(
    notebooks=["notebook-123", "notebook-456"],
    source_type="link",
    url="https://example.com/article.pdf",
    embed=True,
)

# result is a `Source` object (sync) because async_processing=False (default)

print(f"Created source {result.id} titled '{result.title}'")

Creating a Source with Async Processing

from api.sources_service import sources_service

# Fire‑and‑forget ingestion; we get a processing handle back

proc = sources_service.create_source(
    notebooks=["notebook-123"],
    source_type="upload",
    file_path="/tmp/report.docx",
    embed=True,
    async_processing=True,
)

# `proc` is a SourceProcessingResult containing command_id, status, etc.

print(f"Async job submitted, command ID: {proc.command_id}")

Polling Async Job Status

from api.sources_service import sources_service

status = sources_service.get_source_status(proc.command_id)
print(f"Job status: {status['status']}")

Running Transformations on Existing Sources

from commands.source_commands import run_transformation_command, RunTransformationInput

input_data = RunTransformationInput(
    source_id="source-789",
    transformation_id="transformation-summarize",
)
output = await run_transformation_command(input_data)
print(f"Transformation succeeded: {output.success}, insights created: {output.processing_time}")

Key Components and File Locations

• api/sources_service.py – Wraps low-level API calls; returns domain objects (Source, SourceProcessingResult).
• commands/source_commands.py – Implements the process_source SurrealDB command with retry logic and status tracking.
• open_notebook/graphs/source.py – Defines the state-graph with nodes content_process, save_source, and transform_content.
• content_core.extract_content – Handles PDF/HTML/YouTube extraction; configurable via ContentSettings.
• open_notebook/graphs/transformation.py – Graph used by transform_content to call LLMs and generate insights.
• open_notebook/domain/notebook.py – Contains Source, Asset, and persistence methods (vectorize, add_insight).
• surreal_commands – External package providing the retry/queue infrastructure for process_source_command.

Summary

• The sources_service architecture separates API handling, command orchestration, and workflow execution to prevent HTTP request blocking during content processing.
• SourcesService in api/sources_service.py provides synchronous and asynchronous interfaces via create_source, returning either Source objects or SourceProcessingResult handles.
• The command layer in commands/source_commands.py manages background jobs using surreal-commands with built-in retry logic and exponential back-off.
• LangGraph orchestrates extraction, persistence, and transformation through a three-node state graph defined in open_notebook/graphs/source.py.
• Vector embedding occurs asynchronously via source.vectorize() immediately after the source record is saved, enabling semantic search without delaying the ingestion response.

Frequently Asked Questions

What is the sources_service architecture for content ingestion in Open Notebook?

The architecture consists of three layers: an API façade (SourcesService), an asynchronous command layer (process_source_command), and a LangGraph workflow (source_graph). This design extracts content from URLs, files, or text, applies optional LLM transformations, and generates vector embeddings while keeping the REST API responsive.

How does the content ingestion pipeline handle large files or slow URLs?

Heavy processing is offloaded to the command layer running under surreal-commands, which provides retry semantics with exponential back-off. The content_process node in open_notebook/graphs/source.py validates extraction results and raises ValueError for permanent failures, while the command layer handles transient errors automatically.

What is the difference between synchronous and asynchronous source creation?

Synchronous creation (async_processing=False) returns a fully processed Source object immediately, suitable for small text inputs. Asynchronous creation (async_processing=True) returns a SourceProcessingResult containing a command_id, allowing the client to poll status via get_source_status() while the LangGraph workflow processes large documents or URLs in the background.

Where does vector embedding occur in the ingestion workflow?

Vector embedding occurs in the save_source node of open_notebook/graphs/source.py. When embed=True, the node calls source.vectorize() after persisting the source record, implementing fire-and-forget semantics that prevent embedding delays from affecting API response times.

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