LangChain

LangChain is the broadest framework on this list. The surface includes chat models, prompt templates, LCEL (the | operator that pipes components together), output parsers, retrievers, document loaders, vector store adapters, and memory abstractions. Flux does not replace any of it. The integration model is the same as every Python library that runs inside Flux: build the chain, call .ainvoke() inside a @task, return the result.

The Phase 4 page Connecting LangChain walks through three patterns (conversational chain, RAG pipeline, LangGraph multi-agent). This page is the deeper guide: where the layers divide, how LangSmith tracing coexists with Flux tracing, what LangChain memory means inside a Flux workflow, and when you should reach for LangChain instead of Flux primitives.

Prerequisites

pip install langchain-core langchain-openai flux-core

Pull in langchain-community for document loaders, langchain-chroma for a vector store, langchain-text-splitters for chunking. Each is a separate package — install only what your workflow uses. The full bundle has heavy transitive dependencies (Chroma pulls in C++ build tools on some platforms), so prefer the narrow imports.

Wrapping an LCEL pipeline

An LCEL pipeline is a chain assembled with the pipe operator:

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from flux import ExecutionContext, task, workflow


@task.with_options(
    retry_max_attempts=3,
    retry_delay=2,
    timeout=60,
    secret_requests=["openai_api_key"],
)
async def summarize_chain(article: str, secrets: dict) -> str:
    import os
    os.environ["OPENAI_API_KEY"] = secrets["openai_api_key"]

    prompt = ChatPromptTemplate.from_messages([
        ("system", "Summarize the article in three bullet points."),
        ("human", "{article}"),
    ])
    model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    parser = StrOutputParser()

    chain = prompt | model | parser
    return await chain.ainvoke({"article": article})


@workflow
async def article_summarizer(ctx: ExecutionContext[dict]):
    article = (ctx.input or {}).get("article")
    if not article:
        return {"error": "article required"}
    summary = await summarize_chain(article)
    return {"summary": summary, "execution_id": ctx.execution_id}

The chain is rebuilt every time the task runs. That is fine — building an LCEL chain is cheap (it allocates a few Python objects). The expensive part is the LLM call, which happens inside ainvoke.

Use .ainvoke() rather than .invoke() whenever the chain has any async-capable component. Mixing sync .invoke() inside an async Flux task blocks the worker’s event loop and prevents other workflows on the same worker from progressing.

LangSmith tracing

LangChain ships with a tracing client (LangSmith). Setting LANGSMITH_API_KEY and LANGSMITH_TRACING=true makes every .ainvoke() call send spans to LangSmith. This works inside Flux without any special setup.

@task.with_options(secret_requests=["openai_api_key", "langsmith_api_key"])
async def traced_chain(article: str, secrets: dict) -> str:
    import os
    os.environ["OPENAI_API_KEY"] = secrets["openai_api_key"]
    os.environ["LANGSMITH_API_KEY"] = secrets["langsmith_api_key"]
    os.environ["LANGSMITH_TRACING"] = "true"
    # ... build and run chain

The two tracing systems are independent. LangSmith records the chain’s internal steps (prompt rendering, model call, parser invocation). Flux records the task-level event in its own event log and, if OpenTelemetry is configured, emits a flux.task.execute span. You can correlate them by tagging LangSmith runs with the Flux execution ID:

from langsmith import trace as langsmith_trace

@task.with_options(...)
async def chain_with_correlation(article: str) -> str:
    ctx = await ExecutionContext.get()
    with langsmith_trace("summarize", metadata={"flux_execution_id": ctx.execution_id}):
        return await chain.ainvoke({"article": article})

In LangSmith, filter by metadata.flux_execution_id to jump from a Flux execution to its LangSmith trace.

Memory: LangChain vs Flux

Both LangChain and Flux have something called “memory,” but they mean different things.

• LangChain memory (InMemoryChatMessageHistory, ConversationBufferMemory) holds chat history inside a chain. It is per-chain, per-call. When the Flux task returns, the LangChain memory object goes out of scope.
• Flux WorkingMemory is for the in-workflow agent() task. It holds the agent’s tool-call history across iterations of the agent loop inside a single workflow execution.
• Flux LongTermMemory (in flux.tasks.ai.memory) persists facts across executions, optionally backed by a vector store.

To make a LangChain chain remember across workflow turns, serialize the message list and pass it through pause()/resume():

from flux.tasks import pause

@workflow
async def conversational(ctx: ExecutionContext[dict]):
    messages: list[dict] = []
    user_msg = ctx.input["message"]

    while True:
        messages, reply = await call_langchain_chain(messages, user_msg)
        next_input = await pause("waiting_for_user")
        if not next_input or not next_input.get("message"):
            return {"history": messages}
        user_msg = next_input["message"]

The list is plain JSON-serializable data, so it travels through Flux’s checkpoint mechanism with no special handling. Flux owns the durability; LangChain owns the chain semantics.

When to use LangChain inside Flux (and when not to)

Use LangChain when:

• You need document loaders or retrievers it already wraps (Notion, Confluence, Slack, dozens of databases).
• You want LCEL’s composition style: a chain you can read top-to-bottom as prompt | model | parser.
• Your team already has chains you do not want to rewrite.

Reach for Flux primitives instead when:

• You need durable multi-step orchestration. A LangChain chain is one synchronous unit; Flux workflows have checkpoints between every @task call. If a worker crashes between steps two and three of a five-task workflow, Flux resumes at step three. A chain crash starts over.
• You need typed structured output with validation. Flux’s built-in agent() accepts a response_format=PydanticModel and validates the LLM response against it.
• You need scheduling, retries, or secrets. LangChain has neither a scheduler nor a secrets layer.

Do not try to put a Flux workflow inside a LangChain chain. The directions are wrong: LangChain components are called from Python code, and Flux workflows are top-level entry points. Wrap the chain in a Flux task and let the workflow orchestrate.

What each side handles

Error handling

Wrap the chain call in try/except and re-raise as RuntimeError with a message that identifies the failure mode. Flux retries on any exception, but a clear message saves debugging time:

@task.with_options(retry_max_attempts=3, retry_delay=2, timeout=60)
async def call_chain(article: str) -> str:
    try:
        return await chain.ainvoke({"article": article})
    except Exception as e:
        raise RuntimeError(
            f"LangChain summarize failed: {e}. "
            "Check OpenAI status and the prompt template."
        ) from e

On the final retry, Flux marks the task and the workflow execution as failed. Inspect with flux execution show <execution-id>.

Derived against LangChain 0.3 (langchain-core, langchain-openai, langchain-community) and Flux 0.56.0, 2026-07.