Libraries & Frameworks

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, SystemMessage

# Instantiate models from different providers
openai_model = ChatOpenAI(model="gpt-4o", temperature=0.7)
anthropic_model = ChatAnthropic(model="claude-sonnet-4-20250514", temperature=0.7)

# Both accept the same message format
messages = [
    SystemMessage(content="You are a helpful coding assistant."),
    HumanMessage(content="Explain Python list comprehensions in two sentences.")
]

# invoke() returns an AIMessage
response = openai_model.invoke(messages)
print(response.content)

# Swap to Anthropic with no other code changes
response = anthropic_model.invoke(messages)
print(response.content)

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Streaming: tokens arrive incrementally
for chunk in model.stream([HumanMessage(content="Write a haiku about Python.")]):
    print(chunk.content, end="", flush=True)
print()

# Batch: process multiple inputs concurrently
questions = [
    [HumanMessage(content="What is a decorator?")],
    [HumanMessage(content="What is a generator?")],
    [HumanMessage(content="What is a context manager?")],
]
responses = model.batch(questions, config={"max_concurrency": 3})
for resp in responses:
    print(resp.content[:80], "...")

from langchain_core.prompts import ChatPromptTemplate, PromptTemplate

# Simple string template (useful for completion-style models)
string_template = PromptTemplate.from_template(
    "Translate the following text to {language}: {text}"
)
print(string_template.format(language="French", text="Hello, world!"))

# Chat prompt template (recommended for chat models)
chat_template = ChatPromptTemplate.from_messages([
    ("system", "You are a {role} who explains concepts at a {level} level."),
    ("human", "{question}")
])

# format_messages returns a list of Message objects
messages = chat_template.format_messages(
    role="computer science professor",
    level="beginner",
    question="What is recursion?"
)
for msg in messages:
    print(f"{msg.__class__.__name__}: {msg.content}")

from langchain.chains import LLMChain
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# Legacy approach (still functional, but not recommended for new code)
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("human", "Explain {topic} in {num_sentences} sentences.")
])

model = ChatOpenAI(model="gpt-4o-mini")
chain = LLMChain(llm=model, prompt=prompt)

result = chain.invoke({"topic": "gradient descent", "num_sentences": "3"})
print(result["text"])

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

# Define components
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a concise technical writer."),
    ("human", "Explain {concept} in exactly {sentences} sentences.")
])
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = StrOutputParser()

# Compose with the pipe operator
chain = prompt | model | parser

# invoke() flows data through: prompt -> model -> parser
result = chain.invoke({"concept": "MapReduce", "sentences": "3"})
print(result)  # A plain string (parser extracts .content from AIMessage)

# Streaming works automatically through the entire chain
for token in chain.stream({"concept": "Docker containers", "sentences": "2"}):
    print(token, end="", flush=True)

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel, RunnablePassthrough

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = StrOutputParser()

# Two independent chains
pros_chain = (
    ChatPromptTemplate.from_template("List 3 pros of {technology}.")
    | model | parser
)
cons_chain = (
    ChatPromptTemplate.from_template("List 3 cons of {technology}.")
    | model | parser
)

# RunnableParallel runs both chains concurrently
parallel = RunnableParallel(pros=pros_chain, cons=cons_chain)
result = parallel.invoke({"technology": "microservices"})
print("PROS:", result["pros"])
print("CONS:", result["cons"])

from langchain_core.runnables import RunnablePassthrough, RunnableParallel

# Toy in-memory keyword "retriever". Replace with a real retriever
# (FAISS, Chroma, pgvector, Pinecone, ...) in production; the point of
# this example is the RunnableParallel pattern, not a useful retriever.
FACTS = {
    "python": "Python was created by Guido van Rossum in 1991.",
    "django": "Django is a Python web framework released in 2005.",
    "pytorch": "PyTorch is a deep-learning library released by Meta in 2016.",
    "transformer": "The Transformer architecture was introduced by Vaswani et al. in 2017.",
}

def keyword_retriever(query: str) -> str:
    """Naive keyword match: return the first fact whose key appears in the query."""
    q = query.lower()
    for key, fact in FACTS.items():
        if key in q:
            return fact
    return "No relevant context in knowledge base."

# RunnableParallel runs retrieval and pass-through in parallel: the user
# question flows to BOTH the prompt template (as `question`) and the
# retriever (as `context`), so we never re-fetch and never lose the query.
setup = RunnableParallel(
    context=keyword_retriever,
    question=RunnablePassthrough(),
)

rag_prompt = ChatPromptTemplate.from_template(
    "Context: {context}\n\nAnswer this question: {question}"
)

rag_chain = setup | rag_prompt | model | parser

# Three queries hit three different facts; one falls through.
for q in ["Who created Python?", "When was PyTorch released?",
          "What is the Transformer architecture?", "How fast is Rust?"]:
    print(f"Q: {q}\nA: {rag_chain.invoke(q)}\n")

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import ConfigurableField

model = ChatOpenAI(model="gpt-4o-mini", temperature=0).configurable_fields(
    model_name=ConfigurableField(
        id="model_name",
        name="Model Name",
        description="The OpenAI model to use"
    )
)

chain = (
    ChatPromptTemplate.from_template("Summarize: {text}")
    | model
    | StrOutputParser()
)

# Use default model
result1 = chain.invoke({"text": "LangChain is a framework for LLM apps."})

# Override model at runtime
result2 = chain.with_config(
    configurable={"model_name": "gpt-4o"}
).invoke({"text": "LangChain is a framework for LLM apps."})

from pydantic import BaseModel, Field
from typing import Literal
from langchain_openai import ChatOpenAI

class TicketClassification(BaseModel):
    """Classification of a customer support ticket."""
    category: Literal["billing", "technical", "general", "account"] = Field(
        description="The primary category of the ticket"
    )
    priority: Literal["low", "medium", "high", "critical"] = Field(
        description="The urgency level"
    )
    summary: str = Field(
        description="A one-sentence summary of the issue"
    )
    requires_human: bool = Field(
        description="Whether the ticket needs human escalation"
    )

model = ChatOpenAI(model="gpt-4o", temperature=0)
structured_model = model.with_structured_output(TicketClassification)

ticket_text = """
I've been charged twice for my subscription this month.
The first charge was on the 1st and the second on the 15th.
I need a refund for the duplicate charge immediately.
"""

result = structured_model.invoke(
    f"Classify this support ticket:\n\n{ticket_text}"
)

# result is a TicketClassification instance, not a string
print(f"Category: {result.category}")      # "billing"
print(f"Priority: {result.priority}")      # "high"
print(f"Summary: {result.summary}")
print(f"Needs human: {result.requires_human}")  # True
print(f"Type: {type(result)}")  # <class 'TicketClassification'>

from pydantic import BaseModel, Field
from typing import List, Optional

class Entity(BaseModel):
    """A named entity extracted from text."""
    name: str = Field(description="The entity name")
    entity_type: str = Field(description="Type: person, organization, location, date")
    context: str = Field(description="The sentence where this entity appears")

class DocumentAnalysis(BaseModel):
    """Complete analysis of a document."""
    title: str = Field(description="A suitable title for the document")
    language: str = Field(description="The primary language of the text")
    entities: List[Entity] = Field(description="All named entities found")
    key_topics: List[str] = Field(description="3 to 5 main topics")
    sentiment: Literal["positive", "negative", "neutral", "mixed"] = Field(
        description="Overall sentiment"
    )
    word_count_estimate: int = Field(description="Approximate word count")

structured_model = model.with_structured_output(DocumentAnalysis)
analysis = structured_model.invoke("Analyze this text: " + some_text)

# Access nested objects
for entity in analysis.entities:
    print(f"  {entity.name} ({entity.entity_type})")

from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from typing import List

class Recipe(BaseModel):
    name: str = Field(description="Name of the recipe")
    ingredients: List[str] = Field(description="List of ingredients")
    prep_time_minutes: int = Field(description="Preparation time in minutes")
    difficulty: str = Field(description="easy, medium, or hard")

parser = PydanticOutputParser(pydantic_object=Recipe)

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful cooking assistant.\n{format_instructions}"),
    ("human", "Give me a recipe for {dish}.")
])

# The parser generates instructions like:
# "The output should be formatted as a JSON instance..."
chain = prompt.partial(
    format_instructions=parser.get_format_instructions()
) | model | parser

recipe = chain.invoke({"dish": "pasta carbonara"})
print(f"{recipe.name}: {recipe.prep_time_minutes} min, {recipe.difficulty}")
print(f"Ingredients: {', '.join(recipe.ingredients)}")

from langchain_core.output_parsers import JsonOutputParser

json_parser = JsonOutputParser()

chain = (
    ChatPromptTemplate.from_template(
        "Return a JSON object with keys 'city', 'country', and 'population' "
        "for: {query}\n{format_instructions}"
    ).partial(format_instructions=json_parser.get_format_instructions())
    | model
    | json_parser
)

result = chain.invoke({"query": "Tokyo"})
print(result)  # {'city': 'Tokyo', 'country': 'Japan', 'population': 13960000}
print(type(result))  # <class 'dict'>

from langchain.output_parsers import OutputFixingParser

# Wrap the Pydantic parser with auto-fixing
fixing_parser = OutputFixingParser.from_llm(
    parser=parser,  # The PydanticOutputParser from above
    llm=model
)

# Even if the model returns slightly malformed JSON, the fixer will retry
malformed = '{"name": "Carbonara", "ingredients": ["pasta", "eggs"], "prep_time_minutes": "thirty", "difficulty": "medium"}'
# "thirty" is a string but the schema expects int
# The fixing parser will ask the LLM to correct it

fixed = fixing_parser.parse(malformed)
print(f"Prep time: {fixed.prep_time_minutes}")  # 30 (corrected to int)

from langchain.output_parsers import RetryOutputParser

retry_parser = RetryOutputParser.from_llm(
    parser=parser,
    llm=model,
    max_retries=2  # Try up to 2 times before raising an error
)

# The retry parser needs access to the original prompt
# so it can re-ask the model with context
from langchain_core.prompt_values import StringPromptValue

prompt_value = StringPromptValue(text="Give me a recipe for pasta carbonara.")
completion = "Here is a great carbonara recipe..."  # No JSON at all

try:
    result = retry_parser.parse_with_prompt(completion, prompt_value)
    print(result)
except Exception as e:
    print(f"Failed after retries: {e}")

from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from typing import List

class MovieReview(BaseModel):
    title: str = Field(description="Movie title")
    year: int = Field(description="Release year")
    rating: float = Field(description="Rating from 0 to 10")
    pros: List[str] = Field(description="What worked well")
    cons: List[str] = Field(description="What could be improved")
    verdict: str = Field(description="One sentence verdict")

model = ChatOpenAI(model="gpt-4o", temperature=0)
structured_model = model.with_structured_output(MovieReview)

# Stream partial objects
for partial in structured_model.stream("Review the movie Inception (2010)"):
    print(partial)
    # Early chunks: MovieReview(title='Inception', year=None, ...)
    # Later chunks fill in more fields progressively