# Understanding the LangGraph Workflow Architecture for Content Ingestion

> Explore the LangGraph workflow architecture powering Open Notebooks content ingestion. Discover how it streamlines extraction, persistence, and AI transformation for indexed source records.

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

---

**Open Notebook uses LangGraph to orchestrate a three-stage async pipeline—content extraction, persistence, and optional AI transformation—that converts raw URLs or files into indexed Source records with vector embeddings and generated insights.**

The ingestion system in the [open-notebook repository](https://github.com/lfnovo/open-notebook) relies on LangGraph to manage complex state transitions when processing documents. By implementing the LangGraph workflow architecture for content ingestion, the application handles extraction failures gracefully, manages asynchronous vector embeddings, and chains LLM-powered transformations while maintaining API responsiveness.

## The Three-Stage Ingestion Pipeline

The workflow defined in [`open_notebook/graphs/source.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py) processes every source through three logical stages, each implemented as a discrete node in the StateGraph.

### Stage 1: Content Extraction with content_process

The `content_process` node initiates the pipeline by retrieving raw content using the *content-core* library via the `extract_content` function. This node configures default processing engines based on source type—document processors for files and URL handlers for web content—and dynamically injects a speech-to-text model when configured, enabling transcription of audio sources like YouTube videos.

The node implements robust error detection by checking for soft-failure indicators such as "Error" titles in the extraction response. When detected, the node raises exceptions that mark the job as **failed** and retryable rather than completing silently. Upon success, it returns a `ProcessedState` object containing `url`, `file_path`, `title`, and the extracted `content`.

### Stage 2: Persisting Sources with save_source

Following extraction, the `save_source` node loads the existing `Source` record using the provided `source_id`. It persists the extracted `Asset` and stores the full text in `source.full_text`. The node only overwrites placeholder titles—specifically when the existing title is empty or still reads "Processing…"—preserving user-edited titles during reprocessing.

When the `embed` parameter is set to `True`, the node triggers vectorization via `source.vectorize()`, creating searchable embeddings of the content. The node returns the updated `Source` instance to the graph state for downstream processing.

### Stage 3: Optional Transformations via transform_content

The final stage handles user-defined AI transformations through the separate `transformation` graph located in [`open_notebook/graphs/transformation.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/transformation.py). When transformation IDs are provided, the `trigger_transformations` node creates a `Send` event for each transformation ID, fanning out parallel processing tasks.

Each transformation executes the `run_transformation` node, which builds a prompt from the transformation's template, invokes a LangChain-backed LLM through `provision_langchain_model`, cleans "thinking" text from the response, and stores the result as an **Insight** attached to the source.

## StateGraph Topology and Control Flow

The StateGraph wiring in lines 85-99 of [`open_notebook/graphs/source.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py) defines the execution flow:

```text
START ──► content_process ──► save_source
                │
                └─► trigger_transformations? ──► transform_content ──► END

```

The graph begins at `START`, flows through `content_process` to `save_source`, and conditionally branches to `trigger_transformations` only when the `apply_transformations` flag is set. This conditional branching ensures transformations only run when explicitly requested, optimizing processing time for simple ingestion tasks.

## Asynchronous Execution and Error Handling

The workflow operates entirely asynchronously through the Surreal-Commands job queue, invoked via the `process_source` command in [`commands/source_commands.py`](https://github.com/lfnovo/open-notebook/blob/main/commands/source_commands.py). When the API receives an upload request, it immediately returns while the heavy-weight LLM calls, content extraction, and embedding jobs execute in the background.

This design keeps the UI responsive and surfaces failures as retryable jobs rather than silent completed states. The command persists a reference to the job on the source record, enabling the UI to display real-time status updates during long-running extraction or transcription tasks.

## Working with the Pipeline

### Processing a New Source

To trigger the full ingestion pipeline programmatically:

```python
from commands.source_commands import SourceProcessingInput, process_source_command

await process_source_command(
    SourceProcessingInput(
        source_id="src123",
        content_state={"url": "https://example.com/article"},
        notebook_ids=["nbk1", "nbk2"],
        transformations=["trf_summary", "trf_keypoints"],
        embed=True,
    )
)

```

### Running Manual Transformations

To apply a transformation to an already-processed source:

```python
from commands.source_commands import RunTransformationInput, run_transformation_command

await run_transformation_command(
    RunTransformationInput(
        source_id="src123",
        transformation_id="trf_summary",
    )
)

```

### Inspecting Results

After processing completes, inspect the source and its generated insights:

```python
source = await Source.get("src123")
print(source.title)                # Updated title from extraction

print(source.full_text[:200])      # First 200 chars of extracted text

insights = await source.get_insights()
for insight in insights:
    print(insight.title, insight.content[:100])

```

## Key Files in the Ingestion System

| File | Role |
|------|------|
| [`open_notebook/graphs/source.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py) | Defines the LangGraph workflow (`content_process`, `save_source`, `transform_content`) and compiles `source_graph` |
| [`open_notebook/graphs/transformation.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/transformation.py) | Implements the transformation node that calls the LLM and stores resulting insights |
| [`commands/source_commands.py`](https://github.com/lfnovo/open-notebook/blob/main/commands/source_commands.py) | Exposes `process_source` and `run_transformation` commands for API and UI consumption |
| [`open_notebook/domain/source.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/domain/source.py) | Domain model representing the `Source` record with `vectorize()` and `add_insight` helpers |
| [`api/sources_service.py`](https://github.com/lfnovo/open-notebook/blob/main/api/sources_service.py) | HTTP wrapper initiating `process_source` when users upload URLs or files |

## Summary

- **Three-stage pipeline**: Content extraction (`content_process`), persistence (`save_source`), and optional AI transformation (`transform_content`)
- **Resilient error handling**: Soft-failure detection in extraction prevents silent failures and enables automatic retries
- **Conditional execution**: Transformations only execute when `apply_transformations` is set, optimizing resource usage
- **Async architecture**: Surreal-Commands job queue keeps the API responsive during heavy LLM and embedding operations
- **Extensible design**: The graph structure in [`open_notebook/graphs/source.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py) allows easy addition of new preprocessing nodes

## Frequently Asked Questions

### What happens if content extraction fails?

The `content_process` node detects soft-failures from the content-core library by checking for error indicators like "Error" titles in the response. When detected, it raises an exception that marks the job as failed in the Surreal-Commands queue. This design makes the job automatically retryable rather than completing silently with corrupted data.

### How does the pipeline handle audio and video content?

When processing URLs like YouTube videos, the `content_process` node dynamically injects a speech-to-text model into the extraction pipeline if configured. This occurs before calling `extract_content`, enabling automatic transcription of audio content into searchable text stored in `source.full_text`.

### What is the difference between process_source and run_transformation?

`process_source_command` executes the complete LangGraph workflow including extraction, persistence, and optional transformations. `run_transformation_command` bypasses extraction and operates only on already-saved sources, applying a single transformation to generate new insights without reprocessing the original content.

### When are vector embeddings created?

Vector embeddings are generated during the `save_source` node when the `embed` parameter is set to `True` in the `SourceProcessingInput`. This triggers `source.vectorize()`, which creates searchable vector representations of the content for semantic search across notebooks.