LangChain & LangGraph — Prepflix AI Notes

LangChain Expression Language (LCEL)

LCEL lets you compose LangChain components using the pipe operator |. Every component (prompts, LLMs, parsers, retrievers) implements the Runnable interface.

from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser # Build a chain with the pipe operator prompt = ChatPromptTemplate.from_template("Explain {topic} in 2 sentences.") llm = ChatOpenAI(model="gpt-4o-mini") parser = StrOutputParser() chain = prompt | llm | parser # This is LCEL # Three ways to invoke: result = chain.invoke({"topic": "RAG"}) # blocking for token in chain.stream({"topic": "RAG"}): # streaming print(token, end="") results = chain.batch(["RAG", "agents"]) # parallel batch

Runnable method	What it does
`.invoke(input)`	Single call, returns output
`.stream(input)`	Returns generator of tokens/chunks
`.batch([inputs])`	Parallel execution on a list
`.ainvoke()`	Async version of invoke

Streaming is built-in: because every component in LCEL implements .stream(), streaming works end-to-end automatically. You don't need extra code to stream a multi-step chain.

LangChain RAG Stack

from langchain_community.document_loaders import PyPDFLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_community.vectorstores import Chroma from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser # 1. Load documents loader = PyPDFLoader("docs.pdf") docs = loader.load() # 2. Split into chunks splitter = RecursiveCharacterTextSplitter( chunk_size=512, chunk_overlap=50 ) chunks = splitter.split_documents(docs) # 3. Embed and store vectorstore = Chroma.from_documents(chunks, OpenAIEmbeddings()) retriever = vectorstore.as_retriever(search_kwargs={"k": 5}) # 4. RAG chain prompt = ChatPromptTemplate.from_template(""" Answer using ONLY the context below. Context: {context} Question: {question}""") rag_chain = ( {"context": retriever, "question": RunnablePassthrough()} | prompt | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser() ) answer = rag_chain.invoke("What is the refund policy?")

LangGraph — State Machines for AI

LangGraph models workflows as a directed graph. Use it when you need loops, branches, or multi-agent coordination.

LangChain (Sequential)

A → B → C → done. Fixed order, no loops, no branches. Good for simple pipelines.

LangGraph (Graph)

A → B → (check) → C or retry A. Cycles allowed. Human-in-the-loop. Multi-agent coordination.

Minimal LangGraph Agent

from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] # append-only tool_calls: int def call_llm(state: AgentState) -> AgentState: response = llm_with_tools.invoke(state["messages"]) return {"messages": [response]} def call_tools(state: AgentState) -> AgentState: results = tool_executor.invoke(state["messages"][-1].tool_calls) return {"messages": results, "tool_calls": state["tool_calls"] + 1} def should_continue(state: AgentState) -> str: last = state["messages"][-1] if state["tool_calls"] >= 10: return "end" # safety guard if hasattr(last, "tool_calls") and last.tool_calls: return "tools" return "end" graph = StateGraph(AgentState) graph.add_node("llm", call_llm) graph.add_node("tools", call_tools) graph.set_entry_point("llm") graph.add_conditional_edges("llm", should_continue, {"tools": "tools", "end": END}) graph.add_edge("tools", "llm") # loop back to LLM after tools app = graph.compile()

Persistent Memory with Checkpointers

from langgraph.checkpoint.sqlite import SqliteSaver memory = SqliteSaver.from_conn_string("agent_memory.db") app = graph.compile(checkpointer=memory) # thread_id = unique ID per user/session config = {"configurable": {"thread_id": "user_42"}} # First turn result1 = app.invoke( {"messages": [HumanMessage("My name is Pranjal")]}, config=config ) # Second turn — LangGraph restores full prior state result2 = app.invoke( {"messages": [HumanMessage("What's my name?")]}, config=config # same thread_id → continues conversation )

Checkpointer	Use Case
SqliteSaver	Local dev, single instance
PostgresSaver	Production, multi-node deployments
MemorySaver	In-memory testing only, not persistent

Framework Decision: When to Use What

Scenario	Best Choice	Why
Simple one-shot LLM call	Raw API	No abstraction needed — direct is simpler
Multi-step pipeline with streaming	LCEL	Streaming built in, composable, LangSmith tracing
RAG over documents	LangChain RAG stack	Loaders, splitters, vectorstore already built
ReAct agent with tool loop	LangGraph	Needs cycles — LCEL can't loop
Multi-agent workflow	LangGraph	Supervisor pattern, conditional routing
Human-in-the-loop	LangGraph	Interrupt/resume built into StateGraph
Complex logic or custom control	Raw API	Framework adds overhead when you need full control

Real-world reality: most production AI codebases use LangChain for document loading/splitting and LangGraph for agent loops. Raw API calls for anything where the framework adds more confusion than it saves.

Interview tip: "I use LCEL for composing multi-step pipelines, LangGraph for anything that needs a loop or conditional branching — like a ReAct agent — and raw API calls for simple, one-shot features where I don't need the overhead."

Interview Cheat Sheet

LCEL pipe operatorprompt | llm | parser — Runnable sequence

RunnablePassthroughPasses input through unchanged — used in RAG chains

LangGraph StateGraphState = TypedDict shared across all nodes

Conditional edgesBranch based on state content — enables ReAct loop

CheckpointerPersists state between sessions — key for memory

thread_idUnique per user/session — routes to correct stored state

END nodeImported from langgraph.graph — marks graph completion

Safety guardAlways limit tool_calls count — prevents infinite loops