How LangGraph Workflows Manage Chat, Ask, and Source Processing Pipelines in Open Notebook

Open Notebook leverages LangGraph state-machine orchestration to implement three distinct AI pipelines—Chat, Ask, and Source Processing—where each workflow defines a typed StateDict, connects specialized async node functions, and executes through directed graphs that handle everything from text extraction to LLM invocation.

The lfnovo/open-notebook repository structures its core AI capabilities through LangGraph workflows, decomposing complex multi-step operations into manageable, stateful pipelines. These workflows orchestrate the entire lifecycle of user interactions, from ingesting raw documents via open_notebook/graphs/source.py to generating contextual chat responses through open_notebook/graphs/chat.py. By implementing asynchronous node functions and resilient error handling, the system maintains responsive performance under concurrent user demands while ensuring data flows correctly between extraction, embedding, and inference stages.

The Three Core LangGraph Workflows

Open Notebook defines three primary workflow graphs in the open_notebook/graphs package, each handling a distinct user interaction pattern.

Chat Workflow (chat.py)

The Chat workflow manages interactive conversations while maintaining message history and retrieving relevant context. Located in open_notebook/graphs/chat.py, this graph processes incoming messages, performs semantic searches over stored sources to augment context, and invokes the LLM to generate replies. The workflow maintains state through a StateDict containing messages, retrieved_sources, and llm_response fields.

Ask Workflow (ask.py)

The Ask workflow implements a "search-then-synthesize" pattern for direct questions. Implemented in open_notebook/graphs/ask.py, this pipeline receives a user query, executes a vector search to fetch the top-k relevant source snippets, formats these into a specialized prompt via prompt.py, and returns a concise synthesized answer. Unlike the Chat workflow, this pipeline focuses on single-turn information retrieval rather than multi-turn conversation.

Source Processing Pipeline (source.py and source_chat.py)

The Source workflows handle content ingestion and immediate interaction. The primary ingestion graph in open_notebook/graphs/source.py extracts raw text from files, URLs, PDFs, and audio using the content-core library, generates embeddings through the Esperanto multi-provider wrapper, and persists data to SurrealDB. When immediate interaction is required, open_notebook/graphs/source_chat.py chains the ingestion pipeline directly into the Chat workflow, allowing users to query newly added content without separate API calls.

State Management and Node Architecture

Each LangGraph workflow relies on typed state definitions and reusable node functions that perform discrete operations.

Defining State with StateDict

Every workflow begins with a StateDict declaration—a typed dictionary that describes data flowing between nodes. Common state keys include:

• messages: Conversation history for chat contexts
• retrieved_sources: Vector search results from SurrealDB
• embedding_vectors: Float arrays generated by the embedding service
• uploaded_file: Raw file handles for source ingestion
• llm_response: Final generated text from the model

Node Function Implementation

Core node functions are defined as async operations and wired into the graph:

Extraction: The extract_text() function in open_notebook/graphs/source.py interfaces with the content-core library to pull normalized text from PDFs, audio files, or web pages.

Embedding: The embed() node uses the Esperanto wrapper to obtain vector embeddings from configured AI providers, preparing data for vector storage.

Persistence: The save_source() function writes sources and their embeddings to SurrealDB via the repository layer, returning a source_id for future retrieval.

Retrieval: The semantic_search() function (utilized by ask.py) performs vector similarity searches against stored embeddings to return relevant snippets.

LLM Invocation: The call_llm() function—common across all three workflows—selects models via ModelManager, injects system prompts from prompt.py, and streams responses back to the state object.

Graph Construction and Edge Routing

Workflows assemble nodes into directed graphs using LangGraph's builder pattern:

# From open_notebook/graphs/source.py (conceptual structure)

graph = Graph()
graph.add_node("extract", extract_text)
graph.add_node("embed", embed)
graph.add_node("save", save_source)
graph.add_edge("extract", "embed")
graph.add_edge("embed", "save")

The Ask graph extends this pattern by adding search and synthesis nodes between retrieval and LLM invocation. The Chat graph includes a context-merging node that combines historical messages with newly retrieved sources before calling call_llm().

Error Handling and Asynchronous Execution

Production AI pipelines require resilience and non-blocking operations.

Resilient Pipeline Execution with Fallbacks

Nodes are wrapped with LangGraph's catch mechanism to prevent workflow failures from crashing the API. If an LLM call fails during call_llm(), the graph automatically falls back to a cheaper provider or returns a cached response. This ensures that source ingestion jobs and chat sessions remain available even when primary model services experience latency or outages.

Async Event Loop Integration

All node functions are implemented as async def operations, allowing the FastAPI application layer to await graph.ainvoke(state) without blocking the event loop. This architecture enables concurrent processing of multiple chat sessions, parallel source ingestion for bulk uploads, and background podcast generation jobs. The asynchronous design is critical for maintaining responsive API endpoints under load.

FastAPI Integration and Workflow Invocation

FastAPI routers delegate directly to LangGraph instances, mapping HTTP endpoints to specific workflow files.

# api/routers/chat.py

@router.post("/chat")
async def chat_endpoint(payload: ChatRequest):
    state = {"messages": payload.messages}
    result = await chat_graph.ainvoke(state)
    return {
        "response": result["llm_response"], 
        "messages": result["messages"]
    }

# api/routers/ask.py

@router.post("/ask")
async def ask_endpoint(payload: AskRequest):
    state = {"question": payload.question}
    result = await ask_graph.ainvoke(state)
    return {"answer": result["llm_response"]}

# api/routers/source.py

@router.post("/sources")
async def source_endpoint(file: UploadFile):
    state = {"uploaded_file": file}
    result = await source_graph.ainvoke(state)
    return {"source_id": result["source_id"]}

The /sources/chat endpoint (handled by source_chat.py) chains both graphs sequentially, first invoking the source ingestion pipeline and then immediately executing the chat workflow with the newly created source ID.

End-to-End Pipeline Flows

Understanding how data flows through complete executions helps clarify the system design.

Chat with Context Retrieval

When a client sends a message to /chat, the chat.py graph first executes semantic_search() to retrieve relevant sources from SurrealDB. These snippets are merged with the conversation history in the state, then call_llm() generates a contextual response. The updated message list and LLM output are returned to the client, with the new state persisted for subsequent turns.

Ask Search-Then-Synthesize

A query to /ask triggers the ask.py graph to perform a vector search against stored embeddings, retrieving the top-k most relevant text snippets. The graph formats these into a concise prompt template defined in open_notebook/graphs/prompt.py, invokes the LLM for synthesis, and returns only the generated answer without exposing the intermediate retrieval steps to the client.

Source Ingestion to Immediate Chat

Uploading a PDF to /sources/chat initiates a two-phase workflow. First, source.py nodes extract text via content-core, generate embeddings through Esperanto, and store the result in SurrealDB via save_source(). Then, source_chat.py transitions the state to the Chat graph, pre-loading the new source into the context and allowing immediate questioning about the uploaded content.

Summary

• LangGraph workflows in Open Notebook structure AI operations into three distinct pipelines: Chat (chat.py), Ask (ask.py), and Source Processing (source.py/source_chat.py).
• StateDict objects manage typed data flow between async node functions handling extraction, embedding, persistence, retrieval, and LLM invocation.
• Error resilience is achieved through LangGraph's catch mechanism and automatic fallback to alternative providers when primary LLM calls fail.
• FastAPI integration uses await graph.ainvoke(state) to execute workflows asynchronously, preventing event loop blocking during concurrent sessions.
• End-to-end flows demonstrate how chat maintains history, ask performs search-then-synthesize, and source processing enables immediate content interaction through graph composition.

Frequently Asked Questions

What is LangGraph and why does Open Notebook use it?

LangGraph is a state-machine orchestration library that structures complex AI operations as directed graphs. Open Notebook uses it to manage dependencies between extraction, embedding, and inference steps while maintaining type-safe state management through StateDict objects. This approach provides clearer error handling, supports asynchronous execution, and allows the system to resume or retry specific nodes without reprocessing entire pipelines.

How does the Ask workflow differ from the Chat workflow?

The Ask workflow (ask.py) implements a single-turn, search-then-synthesize pattern that retrieves relevant snippets via semantic_search(), formats them into a prompt, and returns a concise answer without maintaining conversation history. The Chat workflow (chat.py) maintains persistent messages state across multiple turns, merges retrieved context with historical conversation, and supports ongoing dialogue. Ask optimizes for quick information retrieval, while Chat optimizes for contextual conversation.

What happens if an LLM call fails during pipeline execution?

LangGraph's catch mechanism wraps node functions to intercept failures. If call_llm() encounters an error or timeout, the graph automatically falls back to a cheaper provider configuration or returns a cached response. This ensures that source ingestion completes successfully and chat endpoints remain available even when primary AI services experience outages, preventing the entire API from crashing on model failures.

Can source processing handle multiple file types simultaneously?

Yes. The extract_text() node in open_notebook/graphs/source.py uses the content-core library to normalize diverse formats—including PDFs, audio files, web URLs, and text documents—into unified text representations. The graph processes these through the same embedding and persistence nodes regardless of original format, enabling batch uploads of mixed media types to be processed concurrently through the async pipeline.

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