LangChain and LangGraph Practical Tutorial - From Basics to Agent Workflows

pip install langchain langchain-core langchain-community
pip install langchain-openai langchain-anthropic
pip install langgraph
pip install langchain-chroma chromadb langchain-text-splitters
pip install langserve fastapi uvicorn

from langchain_openai import ChatOpenAI
 
model = ChatOpenAI(model="gpt-4o", temperature=0)
 
# invoke
response = model.invoke("Explain what LangChain is in one sentence.")
 
# stream
for chunk in model.stream("Explain what LangChain is in one sentence."):
    print(chunk.content, end="", flush=True)
 
# batch
responses = model.batch(["What is LangChain?", "What is LangGraph?"])

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, SystemMessage
 
model = ChatOpenAI(model="gpt-4o", temperature=0)
 
messages = [
    SystemMessage(content="You are a helpful coding assistant."),
    HumanMessage(content="Write a Python function that checks if a number is prime."),
]
response = model.invoke(messages)

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are an expert in {domain}. Answer concisely."),
    ("human", "{question}"),
])
 
# With conversation history support
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}"),
])

from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field
 
parser = StrOutputParser()  # Most common - extracts string content
 
class BookRecommendation(BaseModel):
    title: str = Field(description="Book title")
    author: str = Field(description="Author name")
    reason: str = Field(description="Why this book is recommended")
 
json_parser = JsonOutputParser(pydantic_object=BookRecommendation)

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 helpful assistant."),
    ("human", "{question}"),
])
 
chain = prompt | ChatOpenAI(model="gpt-4o") | StrOutputParser()
 
result = chain.invoke({"question": "What is LCEL?"})
 
for chunk in chain.stream({"question": "What is LCEL?"}):
    print(chunk, end="", flush=True)

translate_chain = (
    ChatPromptTemplate.from_messages([
        ("system", "Translate the following text to {language}."),
        ("human", "{text}"),
    ])
    | ChatOpenAI(model="gpt-4o", temperature=0)
    | StrOutputParser()
)
 
result = translate_chain.invoke({"language": "Korean", "text": "LangChain is great."})

from langchain_core.runnables import RunnablePassthrough
 
chain = (
    RunnablePassthrough.assign(word_count=lambda x: len(x["text"].split()))
    | ChatPromptTemplate.from_messages([
        ("system", "Summarize the following text ({word_count} words) in 2 sentences."),
        ("human", "{text}"),
    ])
    | ChatOpenAI(model="gpt-4o")
    | StrOutputParser()
)

from langchain_core.runnables import RunnableLambda
 
def preprocess(input_dict: dict) -> dict:
    return {"text": input_dict["text"].strip().lower()}
 
def postprocess(output: str) -> dict:
    return {"summary": output, "length": len(output)}
 
chain = (
    RunnableLambda(preprocess)
    | ChatPromptTemplate.from_messages([("human", "Summarize: {text}")])
    | ChatOpenAI(model="gpt-4o")
    | StrOutputParser()
    | RunnableLambda(postprocess)
)

from langchain_core.runnables import RunnableParallel
 
model = ChatOpenAI(model="gpt-4o", temperature=0)
 
analysis_chain = RunnableParallel(
    summary=(
        ChatPromptTemplate.from_messages([("human", "Summarize: {text}")])
        | model | StrOutputParser()
    ),
    keywords=(
        ChatPromptTemplate.from_messages([("human", "Extract 5 keywords from: {text}")])
        | model | StrOutputParser()
    ),
    sentiment=(
        ChatPromptTemplate.from_messages([("human", "Analyze the sentiment of: {text}")])
        | model | StrOutputParser()
    ),
)
 
result = analysis_chain.invoke({"text": "LangChain makes building LLM apps easy."})

from langchain_core.runnables import RunnableBranch
 
branch_chain = RunnableBranch(
    (
        lambda x: x["language"] == "technical",
        ChatPromptTemplate.from_messages([
            ("system", "You are a technical expert."), ("human", "{question}"),
        ]) | model | StrOutputParser()
    ),
    (
        lambda x: x["language"] == "simple",
        ChatPromptTemplate.from_messages([
            ("system", "Explain as if talking to a 10-year-old."), ("human", "{question}"),
        ]) | model | StrOutputParser()
    ),
    # Default branch
    ChatPromptTemplate.from_messages([("human", "{question}")]) | model | StrOutputParser()
)

primary = ChatOpenAI(model="gpt-4o")
fallback = ChatAnthropic(model="claude-sonnet-4-20250514")
model_with_fallback = primary.with_fallbacks([fallback])
 
chain = (
    ChatPromptTemplate.from_messages([("human", "{question}")])
    | model_with_fallback
    | StrOutputParser()
)

from langchain_community.document_loaders import PyPDFLoader, WebBaseLoader, DirectoryLoader
 
pdf_docs = PyPDFLoader("data/report.pdf").load()
web_docs = WebBaseLoader("https://example.com/article").load()
all_docs = DirectoryLoader("data/docs/", glob="**/*.txt").load()

from langchain_text_splitters import RecursiveCharacterTextSplitter
 
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " ", ""],
)
chunks = splitter.split_documents(all_docs)

from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
 
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
 
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db",
    collection_name="my_documents",
)

retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 5},
)
 
# MMR retriever for diverse results
mmr_retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 5, "fetch_k": 20, "lambda_mult": 0.7},
)

from langchain_core.runnables import RunnablePassthrough
 
def format_docs(docs):
    return "\n\n---\n\n".join(doc.page_content for doc in docs)
 
rag_prompt = ChatPromptTemplate.from_messages([
    ("system", """Answer based only on the following context.
If the context is insufficient, say so.
 
Context:
{context}"""),
    ("human", "{question}"),
])
 
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | ChatOpenAI(model="gpt-4o", temperature=0)
    | StrOutputParser()
)
 
answer = rag_chain.invoke("What are the key findings?")

"""Complete RAG pipeline: load, split, embed, store, retrieve, generate."""
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
 
docs = WebBaseLoader("https://lilianweng.github.io/posts/2023-06-23-agent/").load()
chunks = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200).split_documents(docs)
vectorstore = Chroma.from_documents(chunks, OpenAIEmbeddings(model="text-embedding-3-small"))
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
 
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)
 
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | ChatPromptTemplate.from_messages([
        ("system", "Answer based on the context below.\n\nContext:\n{context}"),
        ("human", "{question}"),
    ])
    | ChatOpenAI(model="gpt-4o", temperature=0)
    | StrOutputParser()
)
 
print(rag_chain.invoke("What are the key components of an AI agent?"))

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from operator import add
 
class State(TypedDict):
    messages: Annotated[list[str], add]  # Accumulate via add reducer
    current_step: str
 
def step_one(state: State) -> dict:
    return {"messages": ["Step 1 completed"], "current_step": "step_one"}
 
def step_two(state: State) -> dict:
    return {"messages": ["Step 2 completed"], "current_step": "step_two"}
 
graph = StateGraph(State)
graph.add_node("step_one", step_one)
graph.add_node("step_two", step_two)
graph.add_edge(START, "step_one")
graph.add_edge("step_one", "step_two")
graph.add_edge("step_two", END)
 
app = graph.compile()
result = app.invoke({"messages": [], "current_step": ""})
print(result["messages"])  # ["Step 1 completed", "Step 2 completed"]

from typing import Literal
 
class State(TypedDict):
    query: str
    query_type: str
    response: str
 
def classify_query(state: State) -> dict:
    query = state["query"].lower()
    if "code" in query:
        return {"query_type": "coding"}
    elif "math" in query:
        return {"query_type": "math"}
    return {"query_type": "general"}
 
def handle_coding(state: State) -> dict:
    return {"response": f"[Coding Expert] {state['query']}"}
 
def handle_math(state: State) -> dict:
    return {"response": f"[Math Expert] {state['query']}"}
 
def handle_general(state: State) -> dict:
    return {"response": f"[General] {state['query']}"}
 
def route_query(state: State) -> Literal["handle_coding", "handle_math", "handle_general"]:
    return f"handle_{state['query_type']}"
 
graph = StateGraph(State)
graph.add_node("classify", classify_query)
graph.add_node("handle_coding", handle_coding)
graph.add_node("handle_math", handle_math)
graph.add_node("handle_general", handle_general)
 
graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_query)
graph.add_edge("handle_coding", END)
graph.add_edge("handle_math", END)
graph.add_edge("handle_general", END)
 
app = graph.compile()
result = app.invoke({"query": "Write a sorting algorithm", "query_type": "", "response": ""})

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
 
@tool
def search_web(query: str) -> str:
    """Search the web for current information."""
    return f"Search results for '{query}': LangChain is a framework for LLM apps..."
 
@tool
def calculator(expression: str) -> str:
    """Evaluate a mathematical expression."""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"Error: {e}"
 
@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"The weather in {city} is 22C and sunny."
 
agent = create_react_agent(
    model=ChatOpenAI(model="gpt-4o", temperature=0),
    tools=[search_web, calculator, get_weather],
    prompt="You are a helpful assistant. Use tools when needed.",
)
 
result = agent.invoke({
    "messages": [{"role": "user", "content": "What is 15 * 37 and what's the weather in Seoul?"}]
})
 
for msg in result["messages"]:
    print(f"[{msg.__class__.__name__}] {msg.content}")

from typing import TypedDict, Annotated, Literal
from langchain_core.messages import HumanMessage, ToolMessage, BaseMessage
from langgraph.graph import StateGraph, START, END
from operator import add
 
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add]
 
@tool
def lookup_database(query: str) -> str:
    """Look up information in the internal database."""
    db = {"langchain": "LangChain is a framework for LLM apps.",
          "langgraph": "LangGraph builds stateful agent workflows."}
    return db.get(query.lower().strip(), f"No results for '{query}'.")
 
tools = [lookup_database]
tool_map = {t.name: t for t in tools}
model = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(tools)
 
def call_model(state: AgentState) -> dict:
    return {"messages": [model.invoke(state["messages"])]}
 
def call_tools(state: AgentState) -> dict:
    last_message = state["messages"][-1]
    results = []
    for tc in last_message.tool_calls:
        result = tool_map[tc["name"]].invoke(tc["args"])
        results.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
    return {"messages": results}
 
def should_continue(state: AgentState) -> Literal["call_tools", "end"]:
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "call_tools"
    return "end"
 
graph = StateGraph(AgentState)
graph.add_node("call_model", call_model)
graph.add_node("call_tools", call_tools)
graph.add_edge(START, "call_model")
graph.add_conditional_edges("call_model", should_continue, {"call_tools": "call_tools", "end": END})
graph.add_edge("call_tools", "call_model")  # Loop back
 
agent = graph.compile()
result = agent.invoke({"messages": [HumanMessage(content="Look up 'langchain' in the database.")]})

from langgraph.checkpoint.memory import MemorySaver
 
memory = MemorySaver()
agent_with_memory = graph.compile(checkpointer=memory)
 
config = {"configurable": {"thread_id": "user-123"}}
result1 = agent_with_memory.invoke(
    {"messages": [HumanMessage(content="Look up langchain")]}, config=config,
)
# Second turn - agent remembers conversation
result2 = agent_with_memory.invoke(
    {"messages": [HumanMessage(content="Summarize what you found.")]}, config=config,
)

from typing import TypedDict, Annotated, Literal
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, BaseMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from operator import add
 
class MultiAgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add]
    next_agent: str
    research_result: str
    code_result: str
    final_answer: str
 
model = ChatOpenAI(model="gpt-4o", temperature=0)
 
def research_agent(state: MultiAgentState) -> dict:
    response = model.invoke([
        SystemMessage(content="You are a research specialist. Gather relevant information."),
        HumanMessage(content=f"Research: {state['messages'][-1].content}"),
    ])
    return {"research_result": response.content, "messages": [response]}
 
def code_agent(state: MultiAgentState) -> dict:
    response = model.invoke([
        SystemMessage(content="You are a coding specialist. Write clean code."),
        HumanMessage(content=f"Based on research:\n{state.get('research_result', '')}\n\nImplement."),
    ])
    return {"code_result": response.content, "messages": [response]}
 
def writing_agent(state: MultiAgentState) -> dict:
    response = model.invoke([
        SystemMessage(content="You are a technical writer."),
        HumanMessage(content=f"Research:\n{state.get('research_result', '')}\n"
                     f"Code:\n{state.get('code_result', '')}\n\nWrite documentation."),
    ])
    return {"final_answer": response.content, "messages": [response]}
 
def supervisor(state: MultiAgentState) -> dict:
    response = model.invoke([
        SystemMessage(content="""Decide which agent to call next.
Agents: research, code, writing, FINISH. Respond with only the name."""),
        HumanMessage(content=f"Task: {state['messages'][0].content}\n"
                     f"Research: {'done' if state.get('research_result') else 'pending'}\n"
                     f"Code: {'done' if state.get('code_result') else 'pending'}\n"
                     f"Final: {'done' if state.get('final_answer') else 'pending'}"),
    ])
    return {"next_agent": response.content.strip().lower()}
 
def route_supervisor(state: MultiAgentState) -> Literal["research", "code", "writing", "end"]:
    n = state.get("next_agent", "")
    if "research" in n: return "research"
    elif "code" in n: return "code"
    elif "writing" in n: return "writing"
    return "end"
 
graph = StateGraph(MultiAgentState)
graph.add_node("supervisor", supervisor)
graph.add_node("research", research_agent)
graph.add_node("code", code_agent)
graph.add_node("writing", writing_agent)
 
graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", route_supervisor,
    {"research": "research", "code": "code", "writing": "writing", "end": END})
graph.add_edge("research", "supervisor")
graph.add_edge("code", "supervisor")
graph.add_edge("writing", "supervisor")
 
multi_agent = graph.compile()
result = multi_agent.invoke({
    "messages": [HumanMessage(content="Create a Python rate limiter with token bucket")],
    "next_agent": "", "research_result": "", "code_result": "", "final_answer": "",
})

graph = StateGraph(MultiAgentState)
graph.add_node("research", research_agent)
graph.add_node("code", code_agent)
graph.add_node("writing", writing_agent)
graph.add_edge(START, "research")
graph.add_edge("research", "code")
graph.add_edge("code", "writing")
graph.add_edge("writing", END)
pipeline = graph.compile()

from langchain_core.runnables import RunnableParallel
 
parallel_research = RunnableParallel(
    web=web_research_chain,
    db=database_research_chain,
    api=api_research_chain,
)
results = parallel_research.invoke({"query": "LangChain best practices"})

from langgraph.checkpoint.memory import MemorySaver
 
agent = graph.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["call_tools"],  # Pause before tool execution
)
 
config = {"configurable": {"thread_id": "review-123"}}
 
# First invocation - pauses before tools
result = agent.invoke(
    {"messages": [HumanMessage(content="Delete all old records")]}, config=config,
)
 
# Inspect planned action
print(f"Agent wants to call: {result['messages'][-1].tool_calls}")
 
# Approve and resume
result = agent.invoke(None, config=config)
 
# Or modify before resuming
agent.update_state(config, {"messages": [HumanMessage(content="Only delete records > 1 year old")]})
result = agent.invoke(None, config=config)

pip install langsmith
export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY="ls__your_api_key_here"
export LANGCHAIN_PROJECT="my-langchain-project"

# Automatic tracing - just invoke your chain
result = chain.invoke(
    {"question": "What is LangSmith?"},
    config={"run_name": "langsmith-demo"},  # Custom name for identification
)

from langsmith import traceable
 
@traceable(name="my-rag-pipeline")
def rag_query(question: str) -> str:
    docs = retriever.invoke(question)
    context = "\n".join(doc.page_content for doc in docs)
    return chain.invoke({"context": context, "question": question})

from langsmith import Client
from langsmith.evaluation import evaluate
 
client = Client()
 
dataset = client.create_dataset("qa-eval-dataset")
client.create_examples(
    inputs=[{"question": "What is LangChain?"}, {"question": "What is LangGraph?"}],
    outputs=[
        {"answer": "A framework for building LLM applications."},
        {"answer": "A library for stateful agent workflows."},
    ],
    dataset_id=dataset.id,
)
 
def predict(inputs: dict) -> dict:
    return {"answer": chain.invoke(inputs)}
 
results = evaluate(predict, data=dataset.name, evaluators=["qa", "relevance"],
                   experiment_prefix="rag-v1")

from langchain import hub
 
prompt = hub.pull("rlm/rag-prompt")               # Pull shared prompt
hub.push("my-org/my-prompt", prompt, new_repo_is_public=False)  # Push your own

# server.py
from fastapi import FastAPI
from langserve import add_routes
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
chain = (
    ChatPromptTemplate.from_messages([
        ("system", "You are a helpful assistant."),
        ("human", "{question}"),
    ])
    | ChatOpenAI(model="gpt-4o")
    | StrOutputParser()
)
 
app = FastAPI(title="LangChain API", version="1.0")
add_routes(app, chain, path="/chat")
# Run: uvicorn server:app --host 0.0.0.0 --port 8000

# Auto-generated endpoints:
# POST /chat/invoke     - Single invocation
# POST /chat/batch      - Batch invocation
# POST /chat/stream     - Streaming
# GET  /chat/playground - Interactive UI

from langserve import RemoteRunnable
 
chain = RemoteRunnable("http://localhost:8000/chat")
result = chain.invoke({"question": "What is LangChain?"})

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
 
# Retry with backoff
model = ChatOpenAI(model="gpt-4o", max_retries=3, request_timeout=30)
 
# Fallback models
resilient_model = ChatOpenAI(model="gpt-4o").with_fallbacks([
    ChatAnthropic(model="claude-sonnet-4-20250514")
])
 
# Error handling in LangGraph nodes
def safe_node(state: dict) -> dict:
    try:
        result = model.invoke(state["messages"])
        return {"messages": [result]}
    except Exception as e:
        logger.error(f"Node failed: {e}")
        return {"messages": [AIMessage(content=f"Error: {e}. Please try again.")]}

from fastapi.responses import StreamingResponse
import json
 
@app.post("/stream")
async def stream_response(request: dict):
    async def generate():
        async for chunk in chain.astream(request):
            yield f"data: {json.dumps({'content': chunk})}\n\n"
        yield "data: [DONE]\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream")
 
# LangGraph event streaming
async for event in agent.astream_events(
    {"messages": [HumanMessage(content="Search for LangChain")]}, version="v2",
):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="")

from langchain_core.globals import set_llm_cache
from langchain_community.cache import InMemoryCache, SQLiteCache
 
set_llm_cache(InMemoryCache())                              # Fast, in-memory
set_llm_cache(SQLiteCache(database_path=".langchain.db"))   # Persistent
 
# Semantic cache for similar queries
from langchain_community.cache import RedisSemanticCache
set_llm_cache(RedisSemanticCache(
    redis_url="redis://localhost:6379",
    embedding=OpenAIEmbeddings(),
    score_threshold=0.95,
))

import asyncio
from collections import deque
from time import time
 
class RateLimiter:
    def __init__(self, max_calls: int, time_window: float = 60.0):
        self.max_calls = max_calls
        self.time_window = time_window
        self.calls = deque()
 
    async def acquire(self):
        now = time()
        while self.calls and self.calls[0] < now - self.time_window:
            self.calls.popleft()
        if len(self.calls) >= self.max_calls:
            await asyncio.sleep(self.calls[0] + self.time_window - now)
        self.calls.append(time())
 
rate_limiter = RateLimiter(max_calls=50, time_window=60)
 
async def rate_limited_invoke(chain, input_data):
    await rate_limiter.acquire()
    return await chain.ainvoke(input_data)