How to Add a Custom LangGraph Workflow to Open-Notebook

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 and 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 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, 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 (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, 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:

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

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:

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

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 – Multi-step Q&A with strategy generation and vector search integration.
• open_notebook/graphs/source_chat.py – Conversational interface with context building and checkpointing.
• open_notebook/graphs/transformation.py – Simple linear transformation pipeline (lines 45-55, 71-76).
• open_notebook/graphs/prompt.py – Minimal test-suite pattern for basic chain validation.
• 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.

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