Understanding the LangGraph Workflow Architecture for Content Ingestion

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 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 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. 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 defines the execution flow:

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

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:

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:

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	Defines the LangGraph workflow (content_process, save_source, transform_content) and compiles source_graph
open_notebook/graphs/transformation.py	Implements the transformation node that calls the LLM and stores resulting insights
commands/source_commands.py	Exposes process_source and run_transformation commands for API and UI consumption
open_notebook/domain/source.py	Domain model representing the Source record with vectorize() and add_insight helpers
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 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.

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