Orchestration in LangChain

The Runnable interface

Every LangChain component — prompts, models, parsers, retrievers, tools, even custom Python/JS functions — implements the same Runnable protocol. That uniform contract is what makes orchestration possible: anything Runnable can be composed with anything else Runnable.

• invoke(input) — single synchronous call.
• batch(inputs) — parallel execution with concurrency limits.
• stream(input) — token / chunk streaming.
• ainvoke / abatch / astream — async variants.
• with_config(...) — attach tags, metadata, callbacks, run names.

LCEL — LangChain Expression Language

LCEL uses the pipe operator to wire Runnables into a graph. The result is itself a Runnable, so chains compose recursively. LCEL gives you streaming, batching, async, retries, fallbacks, and tracing for free — without writing glue code.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a concise technical writer."),
    ("human", "Explain {topic} in two sentences."),
])

chain = prompt | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()

chain.invoke({"topic": "vector databases"})
# Stream tokens as they arrive
for chunk in chain.stream({"topic": "MoE models"}):
    print(chunk, end="")

Prompts & messages

Prompt templates turn variables into structured ChatMessages (system / human / ai / tool). MessagesPlaceholder slots in history. Few-shot templates inject curated examples. Partial variables let you pre-bind values like the current date.

Output parsers & structured output

Output parsers convert raw LLM text into typed values. StrOutputParser returns text; PydanticOutputParser validates against a schema; JsonOutputParser streams partial JSON. Modern models also expose with_structured_output(schema) which uses native tool / JSON mode under the hood.

from pydantic import BaseModel

class Ticket(BaseModel):
    title: str
    priority: int
    tags: list[str]

model = ChatOpenAI(model="gpt-4o-mini").with_structured_output(Ticket)
model.invoke("Login is broken on iOS, urgent for paying customers")

Parallel & conditional branching

RunnableParallel runs branches concurrently and merges results into a dict — ideal for fan-out / fan-in patterns like running a retriever and a query rewriter in parallel. RunnableBranch picks a branch based on a predicate, and RunnableLambda lifts any function into the graph.

from langchain_core.runnables import RunnableParallel, RunnablePassthrough

rag_inputs = RunnableParallel(
    context=retriever,
    question=RunnablePassthrough(),
)

rag_chain = rag_inputs | prompt | model | StrOutputParser()

Tools, tool-calling, and agents

A Tool is a Runnable with a name, description, and arg schema. bind_tools(tools) attaches them to a chat model so it can emit tool_calls. An agent is the loop: model → tool_calls → execute tools → feed results back → repeat until the model stops calling tools. For non-trivial control flow, graduate to LangGraph.

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    "Return the current weather for a city."
    return f"{city}: 22°C, clear"

llm_with_tools = ChatOpenAI(model="gpt-4o-mini").bind_tools([get_weather])
resp = llm_with_tools.invoke("What's the weather in Berlin?")
resp.tool_calls  # -> [{'name': 'get_weather', 'args': {'city': 'Berlin'}, ...}]

Memory & message history

Stateless chains become conversational via RunnableWithMessageHistory, which loads and persists messages per session_id from any BaseChatMessageHistory backend (in-memory, Redis, Postgres, DynamoDB). Trim or summarize history to stay within context windows.

Retrievers & RAG composition

Any Runnable that maps a query string to a list of Documents is a retriever — vector stores, BM25, hybrid, multi-query, parent-document, self-query. Compose them with rerankers and compressors via ContextualCompressionRetriever, then drop the retriever into an LCEL chain.

Streaming

stream() yields incremental output. astream_events() emits a typed event stream for every step in the graph (on_chat_model_stream, on_tool_start, on_retriever_end…), which is what powers token-by-token UIs and progress indicators.

Callbacks, tracing & LangSmith

Callbacks fire at the start, end, and error of every Runnable. LangSmith is the hosted callback handler: set LANGSMITH_TRACING=true and every chain run becomes a traced, replayable, evaluable record — inputs, outputs, latency, token usage, and nested spans.

Retries, fallbacks & timeouts

with_retry() adds exponential backoff for transient failures. with_fallbacks([cheaper_model]) swaps to a backup when the primary errors or rate-limits. with_config(run_name=..., tags=[...]) makes traces filterable in LangSmith.

robust = (
    primary_model
    .with_retry(stop_after_attempt=3)
    .with_fallbacks([backup_model])
)
chain = prompt | robust | StrOutputParser()

When to graduate to LangGraph

LCEL is a DAG — great for linear and parallel pipelines. As soon as you need cycles, conditional edges, durable state, human-in-the-loop, or long-running workflows, move the orchestration layer to LangGraph and keep LangChain Runnables as the nodes.