LangChain & LangGraph

LangChain is the most widely used framework for building LLM-powered applications. LangGraph is its stateful, graph-based extension for building complex agents and multi-agent workflows.

LangChain Overview

LangChain provides:

• Integrations: 200+ LLM providers, vector DBs, document loaders, tools
• LCEL (LangChain Expression Language): compose chains declaratively
• Agents: ReAct, tool-calling, and custom agent loops
• Memory: conversation history management
• LangSmith: observability and evaluation platform

pip install langchain langchain-anthropic langchain-community langchain-text-splitters

Version note: LangChain 1.0 (late 2025) slimmed the langchain package down to the core building blocks and the new create_agent API. Legacy pieces such as LLMChain, ConversationChain, AgentExecutor, and the langchain.memory classes moved to the separate langchain-classic package. New code should use LCEL, create_agent, and LangGraph.

LCEL: LangChain Expression Language

LCEL uses the | pipe operator to compose chains:

from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatAnthropic(model="claude-sonnet-5")

# Simple chain: prompt → LLM → parse
chain = (
    ChatPromptTemplate.from_template("Explain {topic} in simple terms.")
    | llm
    | StrOutputParser()
)

result = chain.invoke({"topic": "vector databases"})
print(result)

# Streaming
for chunk in chain.stream({"topic": "Kubernetes"}):
    print(chunk, end="", flush=True)

# Batch
results = chain.batch([
    {"topic": "Docker"},
    {"topic": "Terraform"},
    {"topic": "dbt"},
])

RAG Chain

from langchain_anthropic import ChatAnthropic
from langchain_chroma import Chroma                    # pip install langchain-chroma
from langchain_huggingface import HuggingFaceEmbeddings  # pip install langchain-huggingface
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# Setup
embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")
vectorstore = Chroma(persist_directory="./db", embedding_function=embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

llm = ChatAnthropic(model="claude-sonnet-5")

prompt = ChatPromptTemplate.from_template("""
Answer the question based on the context below.

Context:
{context}

Question: {question}
""")

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# RAG chain
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

answer = rag_chain.invoke("What is dbt used for?")
print(answer)

Document Loaders & Text Splitters

from langchain_community.document_loaders import (
    PyPDFLoader, WebBaseLoader, DirectoryLoader,
    GitLoader, NotionDirectoryLoader
)
from langchain_text_splitters import RecursiveCharacterTextSplitter

# Load PDF
loader = PyPDFLoader("report.pdf")
docs = loader.load()

# Load all .md files in a directory
loader = DirectoryLoader("./docs", glob="**/*.md")
docs = loader.load()

# Load webpage
loader = WebBaseLoader("https://docs.dbt.dev/docs/introduction")
docs = loader.load()

# Split
splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
chunks = splitter.split_documents(docs)

# Add to vector store
vectorstore = Chroma.from_documents(chunks, embeddings, persist_directory="./db")

Agents in LangChain

from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langchain.agents import create_agent  # LangChain 1.0+

llm = ChatAnthropic(model="claude-sonnet-5")

# Define tools with @tool decorator
@tool
def get_weather(location: str) -> str:
    """Get current weather for a location."""
    return f"The weather in {location} is 22°C and sunny."

@tool
def run_sql(query: str) -> str:
    """Execute a SQL query and return results."""
    # Real implementation would query your DB
    return '[{"user_count": 1234}]'

@tool
def search_web(query: str) -> str:
    """Search the web for current information."""
    return f"Search results for: {query}"

tools = [get_weather, run_sql, search_web]

# Create agent (runs the tool-calling loop on LangGraph under the hood)
agent = create_agent(
    model=llm,
    tools=tools,
    system_prompt="You are a helpful data analyst assistant.",
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "How many users do we have and what's the weather in NYC?"}]}
)
# Claude Sonnet 5 thinks by default, so .content can be a list of blocks
# (thinking + text). .text joins just the text blocks.
print(result["messages"][-1].text)

# Pre-1.0 code used create_tool_calling_agent + AgentExecutor; those now live in langchain-classic.

LangGraph: Stateful Agent Graphs

LangGraph is the evolution beyond simple chains. It models workflows as graphs with nodes (LLM calls, tools) and edges (conditional routing, loops).

pip install langgraph

Basic Graph

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import BaseMessage, ToolMessage
import json

llm = ChatAnthropic(model="claude-sonnet-5")

# State schema
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

# Bind tools to LLM
llm_with_tools = llm.bind_tools([get_weather, run_sql])

# Node: call the LLM
def call_llm(state: AgentState) -> AgentState:
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# Node: execute tool calls
def execute_tools(state: AgentState) -> AgentState:
    last_message = state["messages"][-1]
    tool_results = []

    for tool_call in last_message.tool_calls:
        # Find and execute the tool
        tool_map = {"get_weather": get_weather, "run_sql": run_sql}
        result = tool_map[tool_call["name"]].invoke(tool_call["args"])

        tool_results.append(ToolMessage(
            content=str(result),
            tool_call_id=tool_call["id"]
        ))

    return {"messages": tool_results}

# Routing: should we call tools or end?
def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"   # must match the node name registered below
    return END

# Build the graph
graph = StateGraph(AgentState)
graph.add_node("agent", call_llm)
graph.add_node("tools", execute_tools)

graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")   # loop back after tool execution

app = graph.compile()

# Run
result = app.invoke({"messages": [("user", "What's our user count and NYC weather?")]})
print(result["messages"][-1].text)

Multi-Agent Graph

from langgraph.graph import StateGraph, END
from typing import TypedDict, List

class ResearchState(TypedDict):
    task: str
    research: str
    draft: str
    review: str
    final: str

# Specialized agents
def researcher(state: ResearchState) -> ResearchState:
    """Searches and collects information."""
    response = llm.invoke(f"Research: {state['task']}")
    return {"research": response.text}

def writer(state: ResearchState) -> ResearchState:
    """Writes a draft based on research."""
    response = llm.invoke(f"Write a report based on:\n{state['research']}")
    return {"draft": response.text}

def reviewer(state: ResearchState) -> ResearchState:
    """Reviews and improves the draft."""
    response = llm.invoke(f"Review and improve:\n{state['draft']}")
    return {"review": response.text, "final": response.text}

# Linear pipeline
workflow = StateGraph(ResearchState)
workflow.add_node("researcher", researcher)
workflow.add_node("writer", writer)
workflow.add_node("reviewer", reviewer)

workflow.set_entry_point("researcher")
workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", "reviewer")
workflow.add_edge("reviewer", END)

pipeline = workflow.compile()
result = pipeline.invoke({"task": "Current state of vector databases"})
print(result["final"])

LangSmith (Observability)

import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "lsv2_..."
os.environ["LANGSMITH_PROJECT"] = "my-rag-app"
# Older LANGCHAIN_TRACING_V2 / LANGCHAIN_API_KEY / LANGCHAIN_PROJECT names are still read for backward compatibility

# All LangChain calls are now traced automatically in LangSmith
# → View traces, latency, token usage, errors at smith.langchain.com

Memory Management

The classic memory classes below are legacy: in LangChain 1.0 they live in langchain-classic (pip install langchain-classic, import from langchain_classic). You will still see them in older codebases and interviews, but new code should keep history in LangGraph state with a checkpointer (next section).

# Legacy API (pre-1.0 imports were langchain.memory / langchain.chains)
from langchain_classic.memory import ConversationBufferMemory, ConversationSummaryMemory
from langchain_anthropic import ChatAnthropic
from langchain_classic.chains import ConversationChain

llm = ChatAnthropic(model="claude-sonnet-5")

# Buffer memory: keeps all messages (good for short conversations)
memory = ConversationBufferMemory()

# Summary memory: summarizes older history (good for long conversations)
memory_summary = ConversationSummaryMemory(llm=llm)

chain = ConversationChain(llm=llm, memory=memory)
response1 = chain.invoke({"input": "My name is Alex."})
response2 = chain.invoke({"input": "What's my name?"})   # Correctly recalls "Alex"
print(response2["response"])

LangGraph Persistence (Production Memory)

# pip install langgraph-checkpoint-sqlite
from langgraph.checkpoint.sqlite import SqliteSaver

# Persist state between runs with a checkpointer
# (from_conn_string is a context manager in current releases)
with SqliteSaver.from_conn_string("./memory.db") as memory_saver:
    app = graph.compile(checkpointer=memory_saver)

    # Thread-based sessions (each thread_id = one conversation)
    config = {"configurable": {"thread_id": "user-123"}}
    result = app.invoke({"messages": [("user", "Hello!")]}, config=config)

Output Parsers & Structured Output

from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field  # langchain_core.pydantic_v1 is deprecated; use Pydantic v2 directly
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate

# Define your schema
class ProductReview(BaseModel):
    sentiment: str = Field(description="positive, negative, or neutral")
    score: int = Field(description="Score from 1-10")
    summary: str = Field(description="One-sentence summary")
    key_points: list[str] = Field(description="List of key points")

llm = ChatAnthropic(model="claude-sonnet-5")

# Structured output via Pydantic
structured_llm = llm.with_structured_output(ProductReview)

prompt = ChatPromptTemplate.from_template("Analyze this review: {review}")
chain = prompt | structured_llm

result = chain.invoke({"review": "Great product but shipping was slow."})
print(result.sentiment)   # "positive"
print(result.score)       # e.g. 7

Few-Shot Prompting

from langchain_core.prompts import FewShotChatMessagePromptTemplate, ChatPromptTemplate

examples = [
    {"input": "What is 2+2?", "output": "4"},
    {"input": "What is 3*5?", "output": "15"},
]

example_prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"),
    ("ai", "{output}"),
])

few_shot_prompt = FewShotChatMessagePromptTemplate(
    examples=examples,
    example_prompt=example_prompt,
)

final_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a math assistant."),
    few_shot_prompt,
    ("human", "{input}"),
])

chain = final_prompt | llm
result = chain.invoke({"input": "What is 7+8?"})

Interview Q&A

Q1: What is LCEL and why was it introduced? LCEL (LangChain Expression Language) is a declarative way to compose chains using the | pipe operator. It was introduced to replace the older LLMChain/SequentialChain classes. Benefits: built-in streaming, async support, batch processing, and easier observability via LangSmith.

Q2: What is the difference between LangChain and LangGraph? LangChain provides components and LCEL for building linear chains (prompt → LLM → parser). LangGraph is built on top of LangChain to support stateful, cyclic workflows (agents with loops, conditional branching, multi-agent coordination). Use LangChain for simple RAG pipelines; use LangGraph for agents that need to loop, branch, or maintain complex state.

Q3: How does LangGraph handle state persistence across sessions? LangGraph uses checkpointers (e.g., SqliteSaver, PostgresSaver, RedisSaver) to persist graph state between invocations. Each conversation is identified by a thread_id in the config. This enables resuming interrupted workflows and multi-turn conversations without loading full history each time.

Q4: How would you implement a ReAct agent in LangChain? In LangChain 1.0+, use create_agent(model, tools, system_prompt=...) with a model that supports tool calling (Claude, GPT, Gemini, etc.); it runs the ReAct loop on LangGraph. Older code used create_tool_calling_agent + AgentExecutor, now in langchain-classic. For more control, use LangGraph to manually implement the ReAct loop: call LLM → check for tool calls → execute tools → feed results back → repeat until no more tool calls.

Q5: What are the main memory types in LangChain and when would you use each? The first three are the classic memory classes (legacy since LangChain 1.0, now in langchain-classic); they still come up in interviews.

• ConversationBufferMemory: stores all messages verbatim. Use for short conversations.
• ConversationSummaryMemory: summarizes older turns using an LLM. Use for long sessions where context window is a concern.
• ConversationBufferWindowMemory: keeps the last N messages. Simple and predictable.
• LangGraph checkpointing: best for production: persists full graph state (not just messages) to a database, enabling true session continuity.

Q6: What is LangSmith and how do you use it? LangSmith is LangChain's observability platform. You enable it by setting LANGSMITH_TRACING=true and LANGSMITH_API_KEY (the older LANGCHAIN_TRACING_V2 / LANGCHAIN_API_KEY names still work). It automatically captures every LangChain call (LLM inputs/outputs, tool calls, chain steps, latency, token usage). It also provides an evaluation framework to run evals on datasets and compare prompt/model versions.

Q7: When would you choose LangChain over calling the Anthropic API directly? Use LangChain when: you need many integrations quickly (vector DBs, document loaders), want LCEL's streaming/batch/async built-in, need agent orchestration (LangGraph), or want LangSmith tracing. Use the raw API when: you need maximum control, minimum dependencies, or you're building a production system where simplicity is more important than framework features.

Q8: How does LangChain's RunnablePassthrough work? RunnablePassthrough passes input through unchanged. It's useful in parallel chains where you need to pass the original query alongside retrieved context:

chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt | llm | StrOutputParser()
)

Here the question key gets the raw user input while context gets the retrieved documents.

Q9: How would you handle errors and retries in LangChain? Use .with_retry() on any runnable:

llm_with_retry = llm.with_retry(stop_after_attempt=3, wait_exponential_jitter=True)

For structured handling use .with_fallbacks() to specify backup models/chains when the primary fails.

Q10: What are the main challenges with LangChain in production?

1. Version instability: the library moves fast; pin versions carefully
2. Debugging complexity: deep chains can be hard to trace without LangSmith
3. Over-abstraction: sometimes easier to call APIs directly for simple use cases
4. Memory management: default in-memory state is not suitable for multi-user production; need external checkpointing

Common Pitfalls

LangChain vs LlamaIndex vs Raw API

Related Topics