As Large Language Model applications evolved from single-prompt generation to complex, multi-turn autonomous agents, standard imperative scripting became inadequate for managing asynchronous streaming, token callbacks, retry policies, and cyclical state transitions.
LangChain Expression Language (LCEL) and LangGraph provide the composable runtime abstractions necessary to build deterministic, stateful, and observable agentic workflows. This guide covers LCEL execution mechanics, LangGraph state channels, cyclic graph compilation, Human-in-the-Loop (HITL) checkpoints, and production observability.
+-----------------------------------------------------------------------------------------------------------------------+
| AGENT RUNTIME COMPARISON |
+-----------------------------------------------------------------------------------------------------------------------+
| Framework | Core Paradigm | Control Flow | State Management | Best Use Case |
+-----------------+--------------------------------+--------------------+--------------------+--------------------------+
| Direct SDK | Imperative Python / Node | Procedural code | Manual variables | Simple 1-step scripts |
| LangChain LCEL | Declarative Runnable DAG | Linear / Fork-Join | Stateless / Pass-thru| Deterministic RAG chains |
| LangGraph | Pregel-inspired State Graph | Cyclic / Loops | Channel Reducers | Autonomous agent loops |
| AutoGen / Crew | Multi-agent conversational | Actor messaging | Conversation queue | Role-playing simulations |
+-----------------------------------------------------------------------------------------------------------------------+
LCEL unifies disparate LLM operations (prompt templates, model invocation, output parsing, tool dispatching) under the Runnable interface. Every component in LCEL implements standard synchronous, asynchronous, batch, and streaming primitives:
invoke() / ainvoke(): Single input to single output.stream() / astream(): Streams output chunks as they are generated by the underlying model token stream.batch() / abatch(): Concurrent parallel execution over a list of inputs.from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnablePassthrough, RunnableParallel
# Declare prompt and model
prompt = ChatPromptTemplate.from_template(
"Given the context: {context}\nAnswer the technical question: {question}"
)
model = ChatOpenAI(model="gpt-4o", temperature=0.0)
parser = StrOutputParser()
# Construct declarative pipeline
rag_chain = (
RunnableParallel(
context=retriever_tool | (lambda docs: "\n".join(d.page_content for d in docs)),
question=RunnablePassthrough()
)
| prompt
| model
| parser
)
# Stream completion tokens asynchronously
async for token in rag_chain.astream("Explain two-phase commit protocol failure modes"):
print(token, end="", flush=True)
While LCEL is fundamentally a Directed Acyclic Graph (DAG), autonomous agents require cyclical loops: reasoning ightarrow action ightarrow environment observation ightarrow reflection ightarrow revised reasoning.
LangGraph implements the Pregel bulk-synchronous graph computation model. An agent is modeled as a state machine where nodes represent Python functions (agents or tools) and edges represent conditional routing logic.
+-----------------------+
| [START] |
+-----------+-----------+
|
v
+-----------------------+
+-----------------> | Agent Node | <-----------------+
| | (Reasoning / LLM) | |
| +-----------+-----------+ |
| | |
| v |
| +-----------------------+ |
| | Conditional Edge | |
| | (tools_condition) | |
| +-----------+-----------+ |
| | |
| +------------------+------------------+ |
| | | |
| Tool Call Needed? Final Answer? |
| | | |
| v v |
| +-------------------+ +-------------------+ |
| | Tool Node | | [END] | |
| | (Execute Tools) | +-------------------+ |
| +---------+---------+ |
| | |
| +-------------------------------------------------+
from typing import Annotated, TypedDict, List
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_core.messages import BaseMessage, HumanMessage
import operator
# State dictionary with additive message channel
class AgentState(TypedDict):
messages: Annotated[List[BaseMessage], operator.add]
iteration_count: int
def call_model(state: AgentState):
response = model_with_tools.invoke(state["messages"])
return {
"messages": [response],
"iteration_count": state.get("iteration_count", 0) + 1
}
# Build graph topology
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", ToolNode(tools=[query_knowledge_graph, run_sql]))
workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", tools_condition)
workflow.add_edge("tools", "agent")
# Compile with persistent checkpointing (SQLite / Postgres)
app = workflow.compile(checkpointer=MemorySaver())
In mission-critical enterprise workflows (e.g., executing database migrations, dispatching financial transactions), an agent must not run entirely unconstrained.
LangGraph provides native Interrupts and State Time-Travel:
execute_fund_transfer), persisting complete execution state to PostgreSQL.resume().