LangChain & LangGraph

Architecture selection first, then implementation + memory

Prerequisites: Agent Core. Next: MCP & A2A.

Which framework? Compare first

Five options developers actually use

LangChain vs LangGraph (decision table)

LangChain in practice

Step 1 — Install & model

pip install langchain langchain-openai langchain-community chromadb

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

Step 2 — RAG chain (most common use)

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

docs = [...]  # your chunked documents
retriever = Chroma.from_documents(docs, OpenAIEmbeddings()).as_retriever(k=4)

prompt = ChatPromptTemplate.from_messages([
    ("system", "Answer ONLY from context. Cite chunk id.\n\n{context}"),
    ("human", "{question}"),
])

rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt | llm | StrOutputParser()
)
rag_chain.invoke("What is the refund window?")

Step 3 — Add a tool (single shot, not loop)

from langchain_core.tools import tool

@tool
def lookup_order(order_id: str) -> str:
    # Return JSON string for order status
    return '{"status":"shipped","eta":"2025-06-01"}'

llm_tools = llm.bind_tools([lookup_order])
msg = llm_tools.invoke("Status of ORD-99?")
# If msg.tool_calls: execute lookup_order(**args) then call llm again with result

LangChain stops here when you need automatic multi-step tool loops — use LangGraph below.

LangGraph in practice

Step 1 — State schema

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

Step 2 — Nodes: model + tools

from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.graph import StateGraph, START, END

def call_model(state: AgentState):
    response = llm_tools.invoke(state["messages"])
    return {"messages": [response]}

builder = StateGraph(AgentState)
builder.add_node("agent", call_model)
builder.add_node("tools", ToolNode([lookup_order, search_docs]))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)  # tools or END
builder.add_edge("tools", "agent")
graph = builder.compile()

Step 3 — Intent router (conditional entry)

def route_intent(state):
    last = state["messages"][-1].content
    if "order" in last.lower():
        return "order_agent"
    return "rag_agent"

builder.add_conditional_edges(START, route_intent)

Step 4 — Human approval before dangerous tool

graph = builder.compile(interrupt_before=["tools"])
# First invoke pauses before tools; user approves; then graph.invoke(None, config) continues

SqliteSaver — short-term session memory

Stores full graph state per thread_id. User closes tab, comes back — conversation continues.

from langgraph.checkpoint.sqlite import SqliteSaver

checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "support-user-42"}}
graph.invoke({"messages": [("user", "Refund for ORD-8821")]}, config)
# ... later same day, same thread_id ...
graph.invoke({"messages": [("user", "Yes item was damaged")]}, config)
# Agent still knows ORD-8821 without re-asking

Debug: time travel

for snap in graph.get_state_history(config):
    print(snap.metadata["step"], snap.values["messages"][-1])

Store — long-term cross-session memory

Checkpoint = one chat thread. Store = user facts across new threads.

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
graph = builder.compile(checkpointer=checkpointer, store=store)
NS = ("users", "alice")

def agent_with_memory(state, *, store):
    prefs = store.search(NS, query=state["messages"][-1].content, limit=3)
    mem = "\n".join(p.value.get("text","") for p in prefs)
    sys = f"User memory:\n{mem}"
    ...

# After user says "I prefer email contact":
store.put(NS, "contact", {"text": "Prefers email over phone"})

Lab: LangGraph support agent

1. LangChain RAG chain for FAQ only (baseline).
2. Rebuild as LangGraph: agent + search_docs tool.
3. Add SqliteSaver — same thread_id, verify no re-fetch of order ID.
4. Add Store — new thread_id next day, preference still loaded.
5. Add interrupt_before=["tools"] on refund tool; demo approve/deny.

Complete minimal project (copy-paste start)

# support_agent.py — run after pip install langgraph langgraph-checkpoint-sqlite langchain-openai
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def search_docs(q: str) -> str:
    return "Refund window: 30 days if unopened."

@tool
def refund(order_id: str) -> str:
    return f"Refund initiated for {order_id}"

llm = ChatOpenAI(model="gpt-4o-mini").bind_tools([search_docs, refund])

class S(TypedDict):
    messages: Annotated[list, add_messages]

def agent(s: S):
    return {"messages": [llm.invoke(s["messages"])]}

g = StateGraph(S)
g.add_node("agent", agent)
g.add_node("tools", ToolNode([search_docs, refund]))
g.add_edge(START, "agent")
g.add_conditional_edges("agent", tools_condition)
g.add_edge("tools", "agent")
app = g.compile(checkpointer=SqliteSaver.from_conn_string("ckpt.db"),
                interrupt_before=["tools"])
cfg = {"configurable": {"thread_id": "u1"}}
app.invoke({"messages": [("user", "Refund ORD-99?")]}, cfg)