LangGraph

LangGraph is the closest peer Flux has on the durability axis. Both treat execution as a sequence of state transitions. Both can crash mid-run and resume. The difference is granularity: LangGraph checkpoints at graph nodes; Flux checkpoints at @task boundaries. They compose well — a LangGraph agent runs as one Flux task, and each side handles its own crash recovery.

This page covers the integration pattern, the comparison (so you can decide which framework to build on), and how to use both together when one tool fits the small-scale agent loop and the other fits the larger workflow.

Prerequisites

pip install langgraph langchain-core langchain-openai flux-core

LangGraph builds on langchain-core’s message types but does not require the full LangChain bundle. Add a chat-model package (langchain-openai, langchain-anthropic, etc.) for the LLM nodes.

Running a StateGraph inside a Flux task

A LangGraph StateGraph is a directed graph of nodes plus typed state. The compiled graph is callable: pass an initial state, get a final state. Wrap that call in a @task:

import operator
from typing import Annotated, TypedDict

from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph
from flux import ExecutionContext, task, workflow


class ReviewState(TypedDict):
    code: str
    reviews: Annotated[list[dict], operator.add]


async def reviewer(specialty: str, state: ReviewState) -> dict:
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    response = await llm.ainvoke([
        ("system", f"You are a {specialty} code reviewer."),
        ("human", f"Review:\n```\n{state['code']}\n```\nReturn findings as JSON."),
    ])
    return {"reviews": [{"agent": specialty, "content": response.content}]}


@task.with_options(
    retry_max_attempts=2,
    retry_delay=5,
    timeout=300,
    secret_requests=["openai_api_key"],
)
async def run_review_graph(code: str, secrets: dict) -> list[dict]:
    import os
    os.environ["OPENAI_API_KEY"] = secrets["openai_api_key"]

    builder = StateGraph(ReviewState)
    for name in ("security", "performance", "style"):
        builder.add_node(name, lambda s, n=name: reviewer(n, s))
        builder.add_edge(START, name)
        builder.add_edge(name, END)

    graph = builder.compile()
    final_state = await graph.ainvoke({"code": code, "reviews": []})
    return final_state["reviews"]


@workflow
async def code_review(ctx: ExecutionContext[dict]):
    code = (ctx.input or {}).get("code")
    if not code:
        return {"error": "code required"}
    reviews = await run_review_graph(code)
    return {"reviews": reviews, "execution_id": ctx.execution_id}

The three nodes run in parallel because each has its own edge from START. LangGraph handles the fan-out. Flux sees one task call.

Using a LangGraph checkpointer

LangGraph has its own checkpointing system, distinct from Flux’s. Pass a checkpointer to builder.compile() and the graph records state after every node:

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver


@task.with_options(timeout=600, secret_requests=["pg_url"])
async def run_with_checkpoints(thread_id: str, code: str, secrets: dict) -> list[dict]:
    async with AsyncPostgresSaver.from_conn_string(secrets["pg_url"]) as checkpointer:
        await checkpointer.setup()

        builder = StateGraph(ReviewState)
        # ... add nodes and edges
        graph = builder.compile(checkpointer=checkpointer)

        config = {"configurable": {"thread_id": thread_id}}
        final_state = await graph.ainvoke({"code": code, "reviews": []}, config=config)
        return final_state["reviews"]

Now you have two layers of durability:

• LangGraph’s checkpointer records state after each graph node. If the graph itself raises mid-run, you can resume from the last completed node by re-invoking with the same thread_id.
• Flux’s event log records the task call. If the Flux worker crashes before the task returns, Flux retries the task from the top, which re-invokes the LangGraph graph with the same thread_id, and LangGraph resumes from its last checkpoint.

The composition matters when graph nodes are expensive (long LLM calls, document retrieval) and you want crash recovery at sub-task granularity. For cheap nodes, skip the checkpointer — the Flux retry will re-run the whole graph and that is usually fine.

LangGraph vs Flux: when to pick which

Both frameworks provide durable execution. The framing differs.

Pick LangGraph when:

• The work is a visually stateful agent loop with conditional branching that you want to draw as a graph.
• You need fine-grained human-in-the-loop interrupts mid-graph (LangGraph’s interrupt() is more ergonomic than serializing state through pause()).
• You want LangSmith’s UI for inspecting agent traces.

Pick Flux when:

• The agent loop is one step in a larger workflow that also calls non-LLM code (database queries, file uploads, external HTTP calls).
• You need scheduling, secrets, multi-worker distribution, or RBAC.
• You want plain Python control flow rather than a graph definition.

The most common pattern is to use both: LangGraph for the agent loops, Flux for everything around them. The example above does exactly that.

Conditional edges and Flux task boundaries

LangGraph’s conditional edges let you route based on state:

def should_run_testing(state: ReviewState) -> str:
    return "skip" if state.get("skip_testing", False) else "run"

builder.add_conditional_edges(START, should_run_testing, {"run": "testing", "skip": END})

This kind of branching is fine inside a Flux task. Flux does not see it; Flux only sees the task call’s input and output. If you want Flux to see the branching (so each branch shows up in the event log and gets its own retry policy), pull the branching up to the workflow level:

@workflow
async def review(ctx: ExecutionContext[dict]):
    code = ctx.input["code"]
    skip_testing = ctx.input.get("skip_testing", False)

    primary = await run_primary_review(code)
    if not skip_testing:
        testing = await run_testing_review(code)
        return {"primary": primary, "testing": testing}
    return {"primary": primary}

Now run_primary_review and run_testing_review are each Flux tasks. They have independent retry policies, independent event log entries, and independent timeouts. The tradeoff is that you have given up LangGraph’s parallel-edge handling — for parallelism, use parallel(...):

from flux.tasks import parallel

primary, testing = await parallel(
    run_primary_review(code),
    run_testing_review(code),
)

Tracing

LangGraph emits its own spans (via LangSmith if configured). Flux emits flux.workflow.execute and flux.task.execute spans. The two streams are independent.

If both are configured, you get two pictures of the same run:

• LangSmith shows the graph-internal flow: which nodes ran, what state was passed, how each LLM call went.
• Flux’s OTel export shows the workflow-level flow: which tasks ran, how long the LangGraph task took, what other tasks ran before and after.

Correlate by tagging LangGraph runs with the Flux execution ID, the same trick from the LangChain page.

Error handling

A LangGraph node failure raises out of graph.ainvoke() as a normal Python exception. Flux’s retry kicks in for the whole task. If you are using a LangGraph checkpointer, the retry resumes from the last checkpointed node — otherwise the graph starts over.

@task.with_options(retry_max_attempts=3, retry_delay=5, timeout=600)
async def safe_graph_run(code: str) -> list[dict]:
    try:
        return await graph.ainvoke({"code": code, "reviews": []})
    except Exception as e:
        raise RuntimeError(f"LangGraph review failed: {e}") from e

Derived against LangGraph 0.2 and Flux 0.56.0, 2026-07.