LangChain Recipes

 1# Core LangChain
 2pip install langchain langchain-community langchain-core
 3
 4# OpenAI
 5pip install langchain-openai
 6
 7# Other providers
 8pip install langchain-anthropic  # Claude
 9pip install langchain-google-genai  # Gemini
10
11# Vector stores
12pip install chromadb faiss-cpu  # or faiss-gpu
13
14# Document loaders
15pip install pypdf unstructured
16
17# Utilities
18pip install python-dotenv

1import os
2from dotenv import load_dotenv
3
4# Load environment variables
5load_dotenv()
6
7# Set API keys
8os.environ["OPENAI_API_KEY"] = "your-api-key"
9os.environ["ANTHROPIC_API_KEY"] = "your-api-key"

 1from langchain_openai import ChatOpenAI, OpenAI
 2
 3# Chat model (recommended)
 4llm = ChatOpenAI(
 5    model="gpt-4",
 6    temperature=0.7,
 7    max_tokens=1000
 8)
 9
10response = llm.invoke("What is LangChain?")
11print(response.content)
12
13# Completion model (legacy)
14llm = OpenAI(model="gpt-3.5-turbo-instruct")
15response = llm.invoke("What is LangChain?")

 1from langchain_anthropic import ChatAnthropic
 2
 3llm = ChatAnthropic(
 4    model="claude-3-opus-20240229",
 5    temperature=0.7,
 6    max_tokens=1000
 7)
 8
 9response = llm.invoke("What is LangChain?")
10print(response.content)

1from langchain_openai import ChatOpenAI
2
3llm = ChatOpenAI(model="gpt-4", streaming=True)
4
5for chunk in llm.stream("Write a short poem about AI"):
6    print(chunk.content, end="", flush=True)

 1from langchain.prompts import PromptTemplate
 2
 3# Create template
 4template = """
 5You are a helpful assistant. Answer the following question:
 6
 7Question: {question}
 8
 9Answer:
10"""
11
12prompt = PromptTemplate(
13    template=template,
14    input_variables=["question"]
15)
16
17# Format prompt
18formatted_prompt = prompt.format(question="What is machine learning?")
19print(formatted_prompt)
20
21# Use with LLM
22chain = prompt | llm
23response = chain.invoke({"question": "What is machine learning?"})

 1from langchain.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate
 2
 3# Create chat template
 4system_template = "You are a helpful assistant that {task}."
 5human_template = "{input}"
 6
 7chat_prompt = ChatPromptTemplate.from_messages([
 8    SystemMessagePromptTemplate.from_template(system_template),
 9    HumanMessagePromptTemplate.from_template(human_template)
10])
11
12# Format
13messages = chat_prompt.format_messages(
14    task="translates English to French",
15    input="Hello, how are you?"
16)
17
18# Use with LLM
19response = llm.invoke(messages)
20print(response.content)

 1from langchain.prompts import FewShotPromptTemplate
 2
 3# Examples
 4examples = [
 5    {"input": "happy", "output": "sad"},
 6    {"input": "tall", "output": "short"},
 7    {"input": "hot", "output": "cold"}
 8]
 9
10# Example template
11example_template = """
12Input: {input}
13Output: {output}
14"""
15
16example_prompt = PromptTemplate(
17    input_variables=["input", "output"],
18    template=example_template
19)
20
21# Few-shot template
22few_shot_prompt = FewShotPromptTemplate(
23    examples=examples,
24    example_prompt=example_prompt,
25    prefix="Give the antonym of the word:",
26    suffix="Input: {input}\nOutput:",
27    input_variables=["input"]
28)
29
30# Use
31chain = few_shot_prompt | llm
32response = chain.invoke({"input": "big"})

 1from langchain.chains import LLMChain, SimpleSequentialChain
 2
 3# First chain: generate topic
 4topic_template = "Suggest a topic about {subject}"
 5topic_prompt = PromptTemplate(
 6    input_variables=["subject"],
 7    template=topic_template
 8)
 9topic_chain = LLMChain(llm=llm, prompt=topic_prompt)
10
11# Second chain: write about topic
12write_template = "Write a short paragraph about: {topic}"
13write_prompt = PromptTemplate(
14    input_variables=["topic"],
15    template=write_template
16)
17write_chain = LLMChain(llm=llm, prompt=write_prompt)
18
19# Combine chains
20overall_chain = SimpleSequentialChain(
21    chains=[topic_chain, write_chain],
22    verbose=True
23)
24
25result = overall_chain.invoke("artificial intelligence")
26print(result)

 1from langchain.chains import SequentialChain
 2
 3# Chain 1: Translate
 4translate_template = "Translate this to {language}: {text}"
 5translate_prompt = PromptTemplate(
 6    input_variables=["language", "text"],
 7    template=translate_template
 8)
 9translate_chain = LLMChain(
10    llm=llm,
11    prompt=translate_prompt,
12    output_key="translated_text"
13)
14
15# Chain 2: Summarize
16summarize_template = "Summarize this text in one sentence: {translated_text}"
17summarize_prompt = PromptTemplate(
18    input_variables=["translated_text"],
19    template=summarize_template
20)
21summarize_chain = LLMChain(
22    llm=llm,
23    prompt=summarize_prompt,
24    output_key="summary"
25)
26
27# Combine
28overall_chain = SequentialChain(
29    chains=[translate_chain, summarize_chain],
30    input_variables=["language", "text"],
31    output_variables=["translated_text", "summary"],
32    verbose=True
33)
34
35result = overall_chain.invoke({
36    "language": "French",
37    "text": "LangChain is a framework for developing applications powered by language models."
38})
39
40print("Translation:", result["translated_text"])
41print("Summary:", result["summary"])

 1from langchain.chains.router import MultiPromptChain
 2from langchain.chains.router.llm_router import LLMRouterChain, RouterOutputParser
 3
 4# Define prompts for different topics
 5physics_template = """You are a physics expert. Answer this question:
 6
 7{input}"""
 8
 9math_template = """You are a math expert. Answer this question:
10
11{input}"""
12
13prompt_infos = [
14    {
15        "name": "physics",
16        "description": "Good for answering physics questions",
17        "prompt_template": physics_template
18    },
19    {
20        "name": "math",
21        "description": "Good for answering math questions",
22        "prompt_template": math_template
23    }
24]
25
26# Create destination chains
27destination_chains = {}
28for p_info in prompt_infos:
29    name = p_info["name"]
30    prompt = PromptTemplate(
31        template=p_info["prompt_template"],
32        input_variables=["input"]
33    )
34    chain = LLMChain(llm=llm, prompt=prompt)
35    destination_chains[name] = chain
36
37# Default chain
38default_chain = LLMChain(
39    llm=llm,
40    prompt=PromptTemplate(
41        template="Answer this question: {input}",
42        input_variables=["input"]
43    )
44)
45
46# Router chain
47destinations = [f"{p['name']}: {p['description']}" for p in prompt_infos]
48router_template = f"""Given a user question, choose which expert should answer it.
49
50Experts:
51{chr(10).join(destinations)}
52
53Question: {{input}}
54
55Expert:"""
56
57router_prompt = PromptTemplate(
58    template=router_template,
59    input_variables=["input"]
60)
61
62router_chain = LLMRouterChain.from_llm(llm, router_prompt)
63
64# Multi-prompt chain
65chain = MultiPromptChain(
66    router_chain=router_chain,
67    destination_chains=destination_chains,
68    default_chain=default_chain,
69    verbose=True
70)
71
72result = chain.invoke("What is Newton's second law?")
73print(result)

 1from langchain.memory import ConversationBufferMemory
 2from langchain.chains import ConversationChain
 3
 4# Create memory
 5memory = ConversationBufferMemory()
 6
 7# Create conversation chain
 8conversation = ConversationChain(
 9    llm=llm,
10    memory=memory,
11    verbose=True
12)
13
14# Have conversation
15response1 = conversation.invoke("Hi, my name is Alice")
16print(response1)
17
18response2 = conversation.invoke("What's my name?")
19print(response2)
20
21# View conversation history
22print(memory.load_memory_variables({}))

 1from langchain.memory import ConversationBufferWindowMemory
 2
 3# Keep only last k interactions
 4memory = ConversationBufferWindowMemory(k=2)
 5
 6conversation = ConversationChain(
 7    llm=llm,
 8    memory=memory,
 9    verbose=True
10)
11
12conversation.invoke("Hi, I'm Alice")
13conversation.invoke("I like pizza")
14conversation.invoke("I have a cat")
15conversation.invoke("What do I like?")  # Remembers pizza
16conversation.invoke("What's my name?")  # Forgets Alice (outside window)

 1from langchain.memory import ConversationSummaryMemory
 2
 3# Summarize conversation history
 4memory = ConversationSummaryMemory(llm=llm)
 5
 6conversation = ConversationChain(
 7    llm=llm,
 8    memory=memory,
 9    verbose=True
10)
11
12conversation.invoke("Hi, I'm working on a machine learning project")
13conversation.invoke("I'm using Python and TensorFlow")
14conversation.invoke("The model accuracy is 95%")
15
16# View summary
17print(memory.load_memory_variables({}))

 1from langchain.memory import ConversationSummaryBufferMemory
 2
 3# Hybrid: keep recent messages, summarize old ones
 4memory = ConversationSummaryBufferMemory(
 5    llm=llm,
 6    max_token_limit=100  # Summarize when exceeds limit
 7)
 8
 9conversation = ConversationChain(
10    llm=llm,
11    memory=memory,
12    verbose=True
13)

 1from langchain_community.document_loaders import (
 2    TextLoader, PyPDFLoader, DirectoryLoader,
 3    WebBaseLoader, CSVLoader
 4)
 5
 6# Text file
 7loader = TextLoader("document.txt")
 8docs = loader.load()
 9
10# PDF
11loader = PyPDFLoader("document.pdf")
12pages = loader.load_and_split()
13
14# Directory of files
15loader = DirectoryLoader("./docs", glob="**/*.txt")
16docs = loader.load()
17
18# Web page
19loader = WebBaseLoader("https://example.com")
20docs = loader.load()
21
22# CSV
23loader = CSVLoader("data.csv")
24docs = loader.load()

 1from langchain.text_splitter import (
 2    RecursiveCharacterTextSplitter,
 3    CharacterTextSplitter,
 4    TokenTextSplitter
 5)
 6
 7# Recursive splitter (recommended)
 8text_splitter = RecursiveCharacterTextSplitter(
 9    chunk_size=1000,
10    chunk_overlap=200,
11    length_function=len
12)
13
14chunks = text_splitter.split_documents(docs)
15
16# Character splitter
17text_splitter = CharacterTextSplitter(
18    separator="\n\n",
19    chunk_size=1000,
20    chunk_overlap=200
21)
22
23# Token splitter
24text_splitter = TokenTextSplitter(
25    chunk_size=500,
26    chunk_overlap=50
27)

 1from langchain_openai import OpenAIEmbeddings
 2from langchain_community.embeddings import HuggingFaceEmbeddings
 3
 4# OpenAI embeddings
 5embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
 6
 7# HuggingFace embeddings (free, local)
 8embeddings = HuggingFaceEmbeddings(
 9    model_name="sentence-transformers/all-MiniLM-L6-v2"
10)
11
12# Embed text
13text = "This is a test document"
14vector = embeddings.embed_query(text)
15print(f"Vector dimension: {len(vector)}")

 1from langchain_community.vectorstores import Chroma
 2
 3# Create vector store
 4vectorstore = Chroma.from_documents(
 5    documents=chunks,
 6    embedding=embeddings,
 7    persist_directory="./chroma_db"
 8)
 9
10# Similarity search
11query = "What is LangChain?"
12docs = vectorstore.similarity_search(query, k=3)
13
14for doc in docs:
15    print(doc.page_content)
16    print("---")
17
18# Similarity search with scores
19docs_with_scores = vectorstore.similarity_search_with_score(query, k=3)
20
21for doc, score in docs_with_scores:
22    print(f"Score: {score}")
23    print(doc.page_content)
24    print("---")
25
26# Load existing vector store
27vectorstore = Chroma(
28    persist_directory="./chroma_db",
29    embedding_function=embeddings
30)

 1from langchain_community.vectorstores import FAISS
 2
 3# Create FAISS index
 4vectorstore = FAISS.from_documents(chunks, embeddings)
 5
 6# Save
 7vectorstore.save_local("faiss_index")
 8
 9# Load
10vectorstore = FAISS.load_local(
11    "faiss_index",
12    embeddings,
13    allow_dangerous_deserialization=True
14)
15
16# Search
17docs = vectorstore.similarity_search(query, k=3)

 1from langchain.chains import RetrievalQA
 2
 3# Create retriever
 4retriever = vectorstore.as_retriever(
 5    search_type="similarity",
 6    search_kwargs={"k": 3}
 7)
 8
 9# Create QA chain
10qa_chain = RetrievalQA.from_chain_type(
11    llm=llm,
12    chain_type="stuff",  # "stuff", "map_reduce", "refine", "map_rerank"
13    retriever=retriever,
14    return_source_documents=True,
15    verbose=True
16)
17
18# Ask question
19result = qa_chain.invoke("What is LangChain?")
20print("Answer:", result["result"])
21print("\nSources:")
22for doc in result["source_documents"]:
23    print(doc.page_content[:200])

 1from langchain.chains import ConversationalRetrievalChain
 2
 3# Create conversational chain with memory
 4qa_chain = ConversationalRetrievalChain.from_llm(
 5    llm=llm,
 6    retriever=retriever,
 7    memory=ConversationBufferMemory(
 8        memory_key="chat_history",
 9        return_messages=True
10    ),
11    verbose=True
12)
13
14# Have conversation
15response1 = qa_chain.invoke({"question": "What is LangChain?"})
16print(response1["answer"])
17
18response2 = qa_chain.invoke({"question": "What are its main features?"})
19print(response2["answer"])

 1from langchain_core.output_parsers import StrOutputParser
 2from langchain_core.runnables import RunnablePassthrough
 3
 4# Define prompt
 5template = """Answer the question based only on the following context:
 6
 7{context}
 8
 9Question: {question}
10
11Answer:"""
12
13prompt = ChatPromptTemplate.from_template(template)
14
15# Create RAG chain
16def format_docs(docs):
17    return "\n\n".join(doc.page_content for doc in docs)
18
19rag_chain = (
20    {"context": retriever | format_docs, "question": RunnablePassthrough()}
21    | prompt
22    | llm
23    | StrOutputParser()
24)
25
26# Use
27answer = rag_chain.invoke("What is LangChain?")
28print(answer)

 1from langchain.agents import AgentExecutor, create_react_agent
 2from langchain import hub
 3from langchain_community.tools import DuckDuckGoSearchRun
 4
 5# Create tools
 6search = DuckDuckGoSearchRun()
 7
 8tools = [
 9    search
10]
11
12# Get prompt
13prompt = hub.pull("hwchase17/react")
14
15# Create agent
16agent = create_react_agent(llm, tools, prompt)
17
18# Create executor
19agent_executor = AgentExecutor(
20    agent=agent,
21    tools=tools,
22    verbose=True,
23    handle_parsing_errors=True
24)
25
26# Run
27result = agent_executor.invoke({
28    "input": "What is the current weather in San Francisco?"
29})
30print(result["output"])

 1from langchain.tools import Tool, StructuredTool
 2from langchain.pydantic_v1 import BaseModel, Field
 3
 4# Simple tool
 5def multiply(a: float, b: float) -> float:
 6    """Multiply two numbers"""
 7    return a * b
 8
 9multiply_tool = Tool(
10    name="Multiply",
11    func=multiply,
12    description="Multiply two numbers together"
13)
14
15# Structured tool with validation
16class CalculatorInput(BaseModel):
17    a: float = Field(description="First number")
18    b: float = Field(description="Second number")
19
20def calculator(a: float, b: float) -> float:
21    """Add two numbers"""
22    return a + b
23
24calculator_tool = StructuredTool.from_function(
25    func=calculator,
26    name="Calculator",
27    description="Add two numbers",
28    args_schema=CalculatorInput
29)
30
31# Use tools
32tools = [multiply_tool, calculator_tool]

 1from langchain.agents import AgentExecutor, create_react_agent
 2from langchain.memory import ConversationBufferMemory
 3
 4# Create memory
 5memory = ConversationBufferMemory(
 6    memory_key="chat_history",
 7    return_messages=True
 8)
 9
10# Create agent with memory
11agent_executor = AgentExecutor(
12    agent=agent,
13    tools=tools,
14    memory=memory,
15    verbose=True
16)
17
18# Have conversation
19response1 = agent_executor.invoke({"input": "My name is Alice"})
20response2 = agent_executor.invoke({"input": "What's my name?"})

 1from langchain.output_parsers import PydanticOutputParser
 2from langchain.pydantic_v1 import BaseModel, Field
 3
 4# Define schema
 5class Person(BaseModel):
 6    name: str = Field(description="Person's name")
 7    age: int = Field(description="Person's age")
 8    occupation: str = Field(description="Person's occupation")
 9
10# Create parser
11parser = PydanticOutputParser(pydantic_object=Person)
12
13# Create prompt with format instructions
14template = """Extract information about the person.
15
16{format_instructions}
17
18Text: {text}
19
20Output:"""
21
22prompt = PromptTemplate(
23    template=template,
24    input_variables=["text"],
25    partial_variables={"format_instructions": parser.get_format_instructions()}
26)
27
28# Create chain
29chain = prompt | llm | parser
30
31# Use
32result = chain.invoke({
33    "text": "John is a 30-year-old software engineer"
34})
35
36print(result.name)  # John
37print(result.age)   # 30
38print(result.occupation)  # software engineer

 1from langchain.output_parsers import ResponseSchema, StructuredOutputParser
 2
 3# Define schema
 4response_schemas = [
 5    ResponseSchema(name="answer", description="Answer to the question"),
 6    ResponseSchema(name="confidence", description="Confidence score 0-100")
 7]
 8
 9parser = StructuredOutputParser.from_response_schemas(response_schemas)
10
11# Create prompt
12template = """Answer the question and provide confidence.
13
14{format_instructions}
15
16Question: {question}
17
18Output:"""
19
20prompt = PromptTemplate(
21    template=template,
22    input_variables=["question"],
23    partial_variables={"format_instructions": parser.get_format_instructions()}
24)
25
26# Use
27chain = prompt | llm | parser
28result = chain.invoke({"question": "What is 2+2?"})
29
30print(result["answer"])
31print(result["confidence"])

 1from langchain_core.output_parsers import StrOutputParser
 2
 3# Simple chain
 4chain = prompt | llm | StrOutputParser()
 5
 6# Invoke
 7result = chain.invoke({"question": "What is AI?"})
 8
 9# Batch
10results = chain.batch([
11    {"question": "What is AI?"},
12    {"question": "What is ML?"}
13])
14
15# Stream
16for chunk in chain.stream({"question": "What is AI?"}):
17    print(chunk, end="", flush=True)

 1from langchain_core.runnables import RunnableParallel
 2
 3# Run multiple chains in parallel
 4chain = RunnableParallel(
 5    summary=prompt1 | llm | StrOutputParser(),
 6    translation=prompt2 | llm | StrOutputParser()
 7)
 8
 9result = chain.invoke({"text": "Some text"})
10print("Summary:", result["summary"])
11print("Translation:", result["translation"])

 1from langchain_core.runnables import RunnableBranch
 2
 3# Route based on input
 4branch = RunnableBranch(
 5    (lambda x: "python" in x["question"].lower(), python_chain),
 6    (lambda x: "javascript" in x["question"].lower(), js_chain),
 7    default_chain
 8)
 9
10result = branch.invoke({"question": "How to use Python?"})

1from langchain.callbacks import get_openai_callback
2
3try:
4    with get_openai_callback() as cb:
5        result = chain.invoke({"question": "What is AI?"})
6        print(f"Tokens used: {cb.total_tokens}")
7        print(f"Cost: ${cb.total_cost}")
8except Exception as e:
9    print(f"Error: {e}")

 1from langchain.cache import InMemoryCache, SQLiteCache
 2from langchain.globals import set_llm_cache
 3
 4# In-memory cache
 5set_llm_cache(InMemoryCache())
 6
 7# SQLite cache
 8set_llm_cache(SQLiteCache(database_path=".langchain.db"))
 9
10# Now LLM calls are cached
11response1 = llm.invoke("What is AI?")  # Makes API call
12response2 = llm.invoke("What is AI?")  # Uses cache

1from langchain.callbacks import StdOutCallbackHandler
2
3# Add callback for debugging
4chain = prompt | llm
5result = chain.invoke(
6    {"question": "What is AI?"},
7    config={"callbacks": [StdOutCallbackHandler()]}
8)

 1from langchain_openai import ChatOpenAI
 2from pydantic import BaseSettings
 3
 4class Settings(BaseSettings):
 5    openai_api_key: str
 6    model_name: str = "gpt-4"
 7    temperature: float = 0.7
 8    
 9    class Config:
10        env_file = ".env"
11
12settings = Settings()
13
14llm = ChatOpenAI(
15    api_key=settings.openai_api_key,
16    model=settings.model_name,
17    temperature=settings.temperature
18)

 1from langchain.llms import OpenAI
 2from langchain.callbacks import get_openai_callback
 3import time
 4
 5def rate_limited_invoke(chain, input_data, max_retries=3):
 6    for attempt in range(max_retries):
 7        try:
 8            return chain.invoke(input_data)
 9        except Exception as e:
10            if "rate_limit" in str(e).lower() and attempt < max_retries - 1:
11                wait_time = 2 ** attempt  # Exponential backoff
12                print(f"Rate limited. Waiting {wait_time}s...")
13                time.sleep(wait_time)
14            else:
15                raise

 1import logging
 2from langchain.callbacks import FileCallbackHandler
 3
 4# Setup logging
 5logging.basicConfig(level=logging.INFO)
 6logger = logging.getLogger(__name__)
 7
 8# File callback
 9handler = FileCallbackHandler("langchain.log")
10
11# Use with chain
12result = chain.invoke(
13    {"question": "What is AI?"},
14    config={"callbacks": [handler]}
15)