How to Create a Custom Content Transformation Workflow in LangGraph

You create a custom content transformation workflow in LangGraph by defining a Transformation record with an LLM prompt template, then letting the source graph conditionally invoke the transformation graph for each transformation listed in a source's apply_transformations field.

The open-notebook repository leverages LangGraph to orchestrate asynchronous, typed state machines that process notebook sources. By implementing a Transformation domain model and wiring it into the source ingestion pipeline, you can automatically transform content using custom LLM prompts. This guide walks through the exact implementation found in the lfnovo/open-notebook codebase.

Understanding the Transformation Architecture

Open-notebook separates transformation logic into three distinct layers: the domain model that stores transformation configurations, the dedicated transformation graph that executes LLM calls, and the source graph that orchestrates the entire ingestion workflow.

The Transformation Domain Model

At the core of the system is the Transformation class defined in open_notebook/domain/transformation.py. This Pydantic model stores the LLM prompt template and execution metadata.

# open_notebook/domain/transformation.py

class Transformation(ObjectModel):
    table_name: ClassVar[str] = "transformation"
    name: str                     # short identifier

    title: str                    # human-readable label

    description: str              # UI description

    prompt: str                   # LLM prompt template

    apply_default: bool           # automatically run on new sources?

The prompt field accepts a template string that will be rendered by ai-prompter at execution time. The apply_default boolean determines whether new sources automatically include this transformation in their processing queue.

The Transformation Graph

The actual LLM execution happens in open_notebook/graphs/transformation.py. This file defines a single-node state machine that handles model provisioning, prompt rendering, and response cleaning.

# open_notebook/graphs/transformation.py

class TransformationState(TypedDict):
    input_text: str
    source: Source
    transformation: Transformation
    output: str

async def run_transformation(state: dict, config: RunnableConfig) -> dict:
    # 1. Resolve content (source.full_text or supplied input_text)

    # 2. Build a system prompt from transformation.prompt + optional defaults

    # 3. Provision the appropriate model via provision_langchain_model

    # 4. Invoke the chain, clean the response, store the insight on the source

    # 5. Return {"output": cleaned_content}

The graph is compiled as a simple linear flow:

graph = StateGraph(TransformationState).add_node("agent", run_transformation) \
                                       .add_edge(START, "agent") \
                                       .add_edge("agent", END) \
                                       .compile()

The Source Graph Orchestrator

The main workflow lives in open_notebook/graphs/source.py. This graph manages extraction, saving, embedding, and conditionally triggers transformations based on the apply_transformations list.

# open_notebook/graphs/source.py

class SourceState(TypedDict):
    content_state: ProcessSourceState
    apply_transformations: List[Transformation]
    source_id: str
    notebook_ids: List[str]
    source: Source
    transformation: Annotated[list, operator.add]
    embed: bool

async def transform_content(state: TransformationState) -> Optional[dict]:
    result = await transform_graph.ainvoke(
        dict(input_text=state["source"].full_text,
             transformation=state["transformation"])
    )
    await state["source"].add_insight(state["transformation"].title,
                                      result["output"])

The workflow uses conditional edges to route to the transformation node only when apply_transformations is non-empty:

workflow = StateGraph(SourceState) \
    .add_node("content_process", content_process) \
    .add_node("save_source", save_source) \
    .add_node("transform_content", transform_content) \
    .add_edge(START, "content_process") \
    .add_edge("content_process", "save_source") \
    .add_conditional_edges("save_source", trigger_transformations,
                            ["transform_content"]) \
    .add_edge("transform_content", END) \
    .compile()

Creating a Custom Transformation Step-by-Step

To implement a custom workflow, you define the transformation record, configure its prompt template, and ensure it appears in the source's execution list.

1. Define the Transformation Record

Create a Transformation instance via API or directly in the database. The following JSON payload creates a summarization transformation:

{
  "name": "summarize",
  "title": "Summarize Content",
  "description": "Generate a concise summary of the source text.",
  "prompt": "You are a summarizer. Provide a 3-sentence summary of the following text:\n\n{{input_text}}",
  "apply_default": true
}

Setting apply_default: true automatically adds this transformation to every new source's apply_transformations list.

2. Configure the Prompt Template

The prompt field uses standard templating with {{input_text}} as the primary variable. The ai-prompter library renders this template at runtime, injecting the source's full_text or any explicitly provided input_text.

3. Set Execution Flags

When apply_default is false, you must manually append the transformation to specific sources. This allows for on-demand processing rather than automatic execution.

Implementing the Workflow in Code

You can programmatically create transformations and trigger them using the domain models and graph interfaces.

Creating a Transformation Programmatically

from open_notebook.domain.transformation import Transformation
from open_notebook.domain.source import Source
from open_notebook.graphs.source import source_graph

# Create a Transformation record

my_transform = await Transformation.create(
    name="extract_keywords",
    title="Extract Keywords",
    description="Return a comma-separated list of key terms from the text.",
    prompt="Extract the most important keywords from the following content:\n\n{{input_text}}",
    apply_default=False,          # run only when explicitly requested

)

# Attach it to a specific source

source = await Source.get("source:123")
source.apply_transformations.append(my_transform)
await source.save()

Triggering the Transformation Graph

Manually invoke the source graph with your transformation configured in the state:

# Trigger the workflow manually

await source_graph.ainvoke(
    {
        "content_state": {...},          # already extracted content state

        "apply_transformations": [my_transform],
        "source_id": source.id,
        "notebook_ids": ["notebook:1"],
        "source": source,
        "embed": True,
    }
)

All calls are asynchronous, and the graph handles error classification via classify_error in open_notebook/utils/error_classifier.py.

How the Graph Executes Transformations

When a source is ingested, the workflow follows this precise execution path:

1. Content Processing: The content_process node extracts raw text from URLs or files.
2. Persistence: The save_source node persists the source to the database.
3. Conditional Routing: The trigger_transformations function checks if apply_transformations is non-empty. If it contains transformations, the graph emits a Send for each transformation, routing to transform_content.
4. Transformation Execution: For each transformation, the system:
   • Resolves the content via source.full_text
   • Builds the system prompt using the transformation's template
   • Provisions the appropriate model via provision_langchain_model from open_notebook/ai/provision.py
   • Invokes the LLM chain and cleans the response
   • Stores the result using source.add_insight
5. Completion: The workflow ends after all transformations complete, with results available in source.insights.

Summary

• Define transformations using the Transformation domain model in open_notebook/domain/transformation.py, specifying a name, title, and LLM prompt template.
• Configure automatic execution by setting apply_default: true, or manually append transformations to specific sources for on-demand processing.
• Execute via LangGraph by invoking the source graph (open_notebook/graphs/source.py), which conditionally routes to the transformation graph (open_notebook/graphs/transformation.py) based on the apply_transformations list.
• Store results using source.add_insight, which persists transformation outputs as notebook insights accessible via the API.
• Handle errors asynchronously through the built-in error classification system that manages exceptions during transformation execution.

Frequently Asked Questions

What is the difference between the transformation graph and source graph?

The transformation graph (open_notebook/graphs/transformation.py) is a single-node state machine responsible for executing one specific LLM transformation, rendering prompts, and cleaning outputs. The source graph (open_notebook/graphs/source.py) is the orchestration layer that handles content extraction, saving, embedding, and conditionally invokes the transformation graph for each transformation attached to a source.

How do I make a transformation run automatically on every new source?

Set the apply_default field to true when creating the Transformation record. This flag tells the source graph to automatically include this transformation in the apply_transformations list for every new source ingested into the system.

Can I use different LLM models for different transformations?

Yes. The run_transformation function in open_notebook/graphs/transformation.py calls provision_langchain_model to select and configure the appropriate model at execution time. You can extend the provisioning logic to select different models based on transformation metadata or configuration.

How are transformation results stored and accessed?

Transformation results are stored as insights on the source object. The run_transformation function calls source.add_insight(state["transformation"].title, result["output"]), which persists the cleaned LLM output. The UI and API then read from source.insights to display the transformed content alongside the original source.

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