# How to Add a Custom LangGraph Workflow to Open-Notebook

> Learn how to add a custom LangGraph workflow to Open-Notebook. Easily define states implement async nodes and compile your graph for powerful automation. Get started today.

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

---

**To add a custom LangGraph workflow to open-notebook, create a `TypedDict` state definition, implement async node functions that call `provision_langchain_model`, wire the graph using `StateGraph.add_node` and `add_edge`, and compile with `compile()` in the `open_notebook/graphs/` directory.**

Open-notebook is an extensible AI notebook application that leverages **LangGraph** to orchestrate asynchronous language model pipelines. The repository follows a modular architecture where each workflow resides as a self-contained graph in the `open_notebook/graphs/` directory. Adding a custom LangGraph workflow to open-notebook requires understanding the standard five-step pattern used by existing implementations like [`ask.py`](https://github.com/lfnovo/open-notebook/blob/main/ask.py) and [`transformation.py`](https://github.com/lfnovo/open-notebook/blob/main/transformation.py).

## Understanding the LangGraph Architecture in Open-Notebook

Every graph in open-notebook follows a consistent template according to the source code in `lfnovo/open-notebook`. The architecture requires five essential components:

1. **TypedDict State** – Defines mutable fields passed between nodes.
2. **Node Functions** – Async functions receiving state and `RunnableConfig`.
3. **Model Provisioning** – Calls to `open_notebook.ai.provision.provision_langchain_model`.
4. **Error Handling** – Uniform classification via `open_notebook.utils.error_classifier.classify_error`.
5. **Graph Compilation** – Building `StateGraph`, wiring edges, and calling `compile()`.

The [`transformation.py`](https://github.com/lfnovo/open-notebook/blob/main/transformation.py) file demonstrates this pattern clearly, with `TransformationState` defined at the top and the `run_transformation` node implementing lines 45-55 for model invocation.

## Step-by-Step Implementation Guide

### Define the State Schema

Start by creating a **TypedDict** class that describes all fields your workflow will mutate. In [`open_notebook/graphs/transformation.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/transformation.py), the `TransformationState` tracks inputs and outputs. Your custom state should follow this pattern, only including fields necessary for your specific use case.

### Implement Async Node Functions

Node functions must accept the state dictionary and a **`RunnableConfig`** object. According to the source code in [`transformation.py`](https://github.com/lfnovo/open-notebook/blob/main/transformation.py) (lines 45-55), nodes should:
- Build message payloads using `SystemMessage` and `HumanMessage`
- Provision models via `await provision_langchain_model()`
- Clean responses using `clean_thinking_content` and `extract_text_content`
- Handle exceptions through `classify_error`

### Wire the Graph Structure

Construct a **`StateGraph`** instance using your state type. Add nodes with `add_node()`, connect them using `add_edge()` or `add_conditional_edges()`, and establish entry points. The compilation happens at lines 71-76 in [`transformation.py`](https://github.com/lfnovo/open-notebook/blob/main/transformation.py), where `graph = agent_state.compile()` produces the executable object.

### Handle Model Provisioning and Errors

Always use `open_notebook.ai.provision.provision_langchain_model` to instantiate models. This central utility manages API keys and model routing. Wrap model calls in try-except blocks that utilize `classify_error` from `open_notebook.utils.error_classifier` to ensure consistent error propagation.

## Complete Example: Building an Echo Workflow

Here is a minimal custom workflow that echoes user input through the provisioned language model. Save this as [`open_notebook/graphs/echo.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/echo.py):

```python

# open_notebook/graphs/echo.py

from typing import TypedDict, Any

from ai_prompter import Prompter
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.runnables import RunnableConfig
from langgraph.graph import END, START, StateGraph

from open_notebook.ai.provision import provision_langchain_model
from open_notebook.utils.text_utils import clean_thinking_content, extract_text_content
from open_notebook.utils.error_classifier import classify_error

# ----------------------------------------------------------------------

# 1️⃣  State definition – only the user prompt and the model response.

# ----------------------------------------------------------------------

class EchoState(TypedDict):
    user_prompt: str
    model_response: str

# ----------------------------------------------------------------------

# 2️⃣  Node function – calls the LLM and stores the reply.

# ----------------------------------------------------------------------

async def call_echo_model(state: EchoState, config: RunnableConfig) -> dict:
    try:
        # Build the system prompt using a simple template.

        system_prompt = Prompter(template_text="You are a helpful echo bot.").render({})
        payload = [
            SystemMessage(content=system_prompt),
            HumanMessage(content=state["user_prompt"]),
        ]

        # Provision the model (model_id can be overridden per‑request).

        model = await provision_langchain_model(
            str(payload),
            config.get("configurable", {}).get("model_id"),
            "echo",
            max_tokens=500,
        )

        # Run the model and clean the reply.

        response = await model.ainvoke(payload)
        cleaned = clean_thinking_content(extract_text_content(response.content))

        return {"model_response": cleaned}
    except Exception as e:
        err_cls, user_msg = classify_error(e)
        raise err_cls(user_msg) from e

# ----------------------------------------------------------------------

# 3️⃣  Graph construction – linear flow: START → call_echo_model → END

# ----------------------------------------------------------------------

graph_state = StateGraph(EchoState)
graph_state.add_node("echo", call_echo_model)   # type: ignore[type-var]

graph_state.add_edge(START, "echo")
graph_state.add_edge("echo", END)

# 4️⃣  Expose the compiled graph for import elsewhere.

graph = graph_state.compile()

```

## Invoking Your Custom Workflow

To use the compiled graph within open-notebook or external scripts:

```python
import asyncio
from open_notebook.graphs.echo import graph

async def demo():
    result = await graph.ainvoke(
        {"user_prompt": "Hello, Open‑Notebook!"},
        config={"configurable": {"model_id": "gpt-4o"}},   # optional override

    )
    print(result["model_response"])

asyncio.run(demo())

```

## Exposing the Workflow via FastAPI (Optional)

To make your workflow accessible via HTTP endpoints, create a router in [`api/routers/echo.py`](https://github.com/lfnovo/open-notebook/blob/main/api/routers/echo.py):

```python

# api/routers/echo.py

from fastapi import APIRouter
from open_notebook.graphs.echo import graph

router = APIRouter()

@router.post("/echo")
async def run_echo(prompt: str):
    result = await graph.ainvoke(
        {"user_prompt": prompt},
        config={"configurable": {"model_id": None}},  # uses default model

    )
    return {"response": result["model_response"]}

```

Then register the router in [`api/main.py`](https://github.com/lfnovo/open-notebook/blob/main/api/main.py):

```python
from api.routers import echo
app.include_router(echo.router, prefix="/api")

```

## Reference Implementation Files

Study these existing graphs in `lfnovo/open-notebook` to understand advanced patterns:

- **[`open_notebook/graphs/ask.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/ask.py)** – Multi-step Q&A with strategy generation and vector search integration.
- **[`open_notebook/graphs/source_chat.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/source_chat.py)** – Conversational interface with context building and checkpointing.
- **[`open_notebook/graphs/transformation.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/transformation.py)** – Simple linear transformation pipeline (lines 45-55, 71-76).
- **[`open_notebook/graphs/prompt.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/graphs/prompt.py)** – Minimal test-suite pattern for basic chain validation.
- **[`open_notebook/ai/provision.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/ai/provision.py)** – Central model provisioning utility used across all workflows.

## Summary

- **Define state** using `TypedDict` to declare mutable fields passed between nodes.
- **Implement nodes** as async functions accepting `state` and `RunnableConfig`, calling `provision_langchain_model` for AI execution.
- **Handle errors** uniformly using `classify_error` from `open_notebook.utils.error_classifier`.
- **Compile graphs** with `StateGraph.add_node()`, `add_edge()`, and `compile()` to create executable workflows.
- **Place files** in `open_notebook/graphs/` and optionally expose via FastAPI routers in `api/routers/`.

## Frequently Asked Questions

### What is the minimum structure required for a LangGraph workflow in Open-Notebook?

Every workflow requires a `TypedDict` state definition, at least one async node function that receives state and `RunnableConfig`, and a compiled `StateGraph` instance. You must use `open_notebook.ai.provision.provision_langchain_model` to instantiate language models and handle errors through `classify_error` for consistency with the rest of the codebase.

### How do I provision different AI models in my custom workflow?

Pass the `model_id` parameter through the `config` dictionary in `RunnableConfig`. The `provision_langchain_model` function checks `config.get("configurable", {}).get("model_id")` and falls back to default settings when None is provided, allowing per-request model overrides.

### Where should I place my custom graph files?

Store all custom LangGraph workflows in the `open_notebook/graphs/` directory. This location follows the project's modular architecture and ensures imports remain consistent with existing patterns like `from open_notebook.graphs.transformation import graph`.

### Can I implement conditional branching and loops in Open-Notebook workflows?

Yes. The `StateGraph` class supports `add_conditional_edges()` for branching logic based on state evaluation. You can implement loops by wiring edges back to previous nodes or using the `add_conditional_edges` method with a router function that determines the next node based on state conditions, similar to patterns found in [`ask.py`](https://github.com/lfnovo/open-notebook/blob/main/ask.py).