The Ultimate LangChain Cheatsheet

# Core (always install)
pip install langchain langchain-core

# Partner packages (pick what you need)
pip install langchain-openai           # OpenAI / Azure OpenAI
pip install langchain-anthropic         # Anthropic Claude
pip install langchain-google-genai      # Google Gemini

# Community integrations
pip install langchain-community

# Common add-ons
pip install langchain-chroma            # Chroma vector store
pip install langchain-text-splitters    # Text splitting utilities
pip install langchainhub                # Shared prompts from LangSmith Hub

# Full install (everything)
pip install langchain langchain-community langchain-core

# LangSmith (tracing & evaluation)
pip install langsmith

# Core
npm install langchain @langchain/core

# Partner packages
npm install @langchain/openai
npm install @langchain/anthropic
npm install @langchain/google-genai

# Community
npm install @langchain/community

# Vector stores & tools
npm install @langchain/chroma
npm install langchain-chroma

# Full
npm install langchain @langchain/core @langchain/openai

# Required environment variables
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
GOOGLE_API_KEY=...
LANGCHAIN_API_KEY=...           # LangSmith
LANGCHAIN_TRACING_V2=true       # Enable tracing
LANGCHAIN_PROJECT=my-project    # Project name

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI

# OpenAI
openai = ChatOpenAI(model="gpt-4o", temperature=0.7)
openai_mini = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Anthropic
claude = ChatAnthropic(model="claude-sonnet-4-20250514", temperature=0.7)
claude_haiku = ChatAnthropic(model="claude-3-5-haiku-20241022")

# Google Gemini
gemini = ChatGoogleGenerativeAI(model="gemini-2.5-pro-preview-05-06")

# Streaming
async for chunk in openai.astream("Hello!"):
    print(chunk.content, end="", flush=True)

from pydantic import BaseModel, Field
from langchain_core.output_parsers import PydanticOutputParser

class MovieReview(BaseModel):
    title: str = Field(description="Movie title")
    rating: float = Field(description="Rating 1-10")
    summary: str = Field(description="One paragraph summary")
    recommended: bool = Field(description="Would you recommend it?")

parser = PydanticOutputParser(pydantic_object=MovieReview)

prompt = ChatPromptTemplate.from_messages([
    ("system", "Extract movie review info.\n{format_instructions}"),
    ("human", "Review: {review}"),
]).partial(format_instructions=parser.get_format_instructions())

chain = prompt | model | parser
result = chain.invoke({"review": "Inception is a 10/10 masterpiece..."})
print(result.title)      # "Inception"
print(result.rating)     # 10.0
print(result.recommended) # True

from langchain_core.tools import tool, StructuredTool
from pydantic import BaseModel, Field

# Method 1: @tool decorator (simplest)
@tool
def search_web(query: str) -> str:
    """Search the web for information about a topic."""
    # Call your search API here
    return f"Search results for: {query}"

@tool
def calculate(expression: str) -> float:
    """Evaluate a mathematical expression."""
    return eval(expression)

# Method 2: Pydantic schema for complex tools
class SearchInput(BaseModel):
    query: str = Field(description="Search query")
    num_results: int = Field(default=5, description="Number of results")

@tool(args_schema=SearchInput)
def search_advanced(query: str, num_results: int = 5) -> str:
    """Search with configurable result count."""
    return f"Top {num_results} results for: {query}"

# Method 3: StructuredTool
from langchain_core.tools import StructuredTool
weather_tool = StructuredTool.from_function(
    func=get_weather,
    name="get_weather",
    description="Get current weather for a city",
    args_schema=WeatherInput,
)

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")
tools = [search_web, calculate, search_advanced]

# Bind tools to model
llm_with_tools = model.bind_tools(tools)

# Invoke — LLM decides which tools to call
response = llm_with_tools.invoke("What is 25 * 47?")
# AIMessage with tool_calls: [
#   {"name": "calculate", "args": {"expression": "25 * 47"}}
# ]

# Execute tool calls
for tool_call in response.tool_calls:
    tool_name = tool_call["name"]
    tool_args = tool_call["args"]
    selected_tool = {t.name: t for t in tools}[tool_name]
    result = selected_tool.invoke(tool_args)
    print(f"{tool_name}({tool_args}) = {result}")

# Built-in LangChain tools
from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
wiki_tool = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

store = {}  # session_id -> ChatMessageHistory

def get_session_history(session_id):
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}"),
])

chain = prompt | model | StrOutputParser()
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

# Multi-turn with memory
chain_with_history.invoke(
    {"input": "Hi, I'm John. I love Python."},
    config={"configurable": {"session_id": "user-123"}}
)
chain_with_history.invoke(
    {"input": "What's my name and favorite language?"},
    config={"configurable": {"session_id": "user-123"}}
)
# "Your name is John and your favorite language is Python."

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

# ─── Async Streaming ───
async for chunk in chain.astream({"topic": "AI"}):
    print(chunk, end="", flush=True)

# ─── Stream with events (detailed) ───
async for event in chain.astream_events(
    {"topic": "AI"}, version="v2"
):
    kind = event["event"]
    if kind == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="", flush=True)
    elif kind == "on_tool_start":
        print(f"\nCalling tool: {event['name']}")

# ─── Stream tokens only ───
for token in model.stream("Tell me about AI"):
    print(token.content, end="", flush=True)

# RAG with streaming
async for chunk in rag_chain.astream("What is Python?"):
    print(chunk, end="", flush=True)

# Stream both retrieved docs AND response
async for event in rag_chain.astream_events(
    {"topic": "Python"}, version="v2"
):
    kind = event["event"]
    if kind == "on_retriever_end":
        docs = event["data"]["output"]
        print(f"Retrieved {len(docs)} documents")
    elif kind == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="")

# FastAPI streaming endpoint
from fastapi.responses import StreamingResponse

async def stream_response(query: str):
    async for chunk in rag_chain.astream(query):
        yield chunk

@app.get("/chat")
async def chat(q: str):
    return StreamingResponse(stream_response(q))

# ─── Environment Variables ───
export LANGCHAIN_API_KEY="lsv2_..."
export LANGCHAIN_TRACING_V2="true"
export LANGCHAIN_PROJECT="my-project"

# ─── Auto-tracing (recommended) ───
# Just set the env vars — all LCEL chains
# are automatically traced!

# ─── Manual tracing ───
from langsmith import traceable

@traceable(name="my_custom_function", tags=["custom"])
def my_function(query: str):
    result = chain.invoke({"input": query})
    return result

# ─── Nested traces ───
@traceable(name="rag_pipeline")
def rag_query(query: str):
    docs = retriever.invoke(query)
    answer = chain.invoke({"context": docs, "question": query})
    return answer

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

primary = ChatOpenAI(model="gpt-4o")
fallback = ChatAnthropic(model="claude-sonnet-4-20250514")
backup = ChatOpenAI(model="gpt-4o-mini")

prompt = ChatPromptTemplate.from_template("{input}")

# Try primary, fall back on error
from langchain_core.runnables import RunnableWithFallbacks

primary_chain = prompt | primary | StrOutputParser()
fallback_chain = prompt | fallback | StrOutputParser()
backup_chain = prompt | backup | StrOutputParser()

chain = primary_chain.with_fallbacks([fallback_chain, backup_chain])

# Automatically retries with fallback on rate limit / error
result = chain.invoke({"input": "Hello!"})

from langchain_core.rate_limiters import InMemoryRateLimiter

# Rate limiting (requests per minute)
rate_limiter = InMemoryRateLimiter(
    requests_per_second=5,
    max_retries=3,
    check_every_n_seconds=0.1,
)

model = ChatOpenAI(
    model="gpt-4o",
    rate_limiter=rate_limiter,
    max_retries=3,  # Built-in retry on transient errors
)

# Retry with exponential backoff
model_with_retry = model.with_retry(
    stop_after_attempt=3,
    retry_if_exception_type=(RateLimitError, APIConnectionError),
)

# Parsing error handling
chain = (
    prompt
    | model
    | output_parser.with_fallbacks([
        StrOutputParser(),  # fallback to raw string
    ])
)