# Understanding the Sources Service Architecture for Content Ingestion in Open Notebook

> Explore the sources_service architecture for Open Notebook content ingestion. Discover its three-layer pipeline, API handling, command orchestration, and LangGraph workflow execution for efficient content processing.

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

---

**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`](https://github.com/lfnovo/open-notebook/blob/main/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`](https://github.com/lfnovo/open-notebook/blob/main/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`](https://github.com/lfnovo/open-notebook/blob/main/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

```python
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

```python
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

```python
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

```python
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`](https://github.com/lfnovo/open-notebook/blob/main/api/sources_service.py)** – Wraps low-level API calls; returns domain objects (`Source`, `SourceProcessingResult`).
- **[`commands/source_commands.py`](https://github.com/lfnovo/open-notebook/blob/main/commands/source_commands.py)** – Implements the `process_source` SurrealDB command with retry logic and status tracking.
- **[`open_notebook/graphs/source.py`](https://github.com/lfnovo/open-notebook/blob/main/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`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/transformation.py)** – Graph used by `transform_content` to call LLMs and generate insights.
- **[`open_notebook/domain/notebook.py`](https://github.com/lfnovo/open-notebook/blob/main/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`](https://github.com/lfnovo/open-notebook/blob/main/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`](https://github.com/lfnovo/open-notebook/blob/main/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`](https://github.com/lfnovo/open-notebook/blob/main/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`](https://github.com/lfnovo/open-notebook/blob/main/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`](https://github.com/lfnovo/open-notebook/blob/main/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.