# How to Create a Custom Content Transformation Workflow in LangGraph

> Learn to create a custom content transformation workflow in LangGraph. Define a Transformation record with an LLM prompt and conditionally invoke it for your needs.

- Repository: [Luis Novo/open-notebook](https://github.com/lfnovo/open-notebook)
- Tags: how-to-guide
- Published: 2026-06-28

---

**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`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/domain/transformation.py). This Pydantic model stores the LLM prompt template and execution metadata.

```python

# 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`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/transformation.py). This file defines a single-node state machine that handles model provisioning, prompt rendering, and response cleaning.

```python

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

```python
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`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py). This graph manages extraction, saving, embedding, and conditionally triggers transformations based on the `apply_transformations` list.

```python

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

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

```json
{
  "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

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

```python

# 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`](https://github.com/lfnovo/open-notebook/blob/main/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`](https://github.com/lfnovo/open-notebook/blob/main/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`](https://github.com/lfnovo/open-notebook/blob/main/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`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source.py)), which conditionally routes to the transformation graph ([`open_notebook/graphs/transformation.py`](https://github.com/lfnovo/open-notebook/blob/main/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`](https://github.com/lfnovo/open-notebook/blob/main/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`](https://github.com/lfnovo/open-notebook/blob/main/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`](https://github.com/lfnovo/open-notebook/blob/main/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.