AI/LLM Cheatsheet 07 — LangChain Patterns

pip install langchain langchain-openai langchain-community langchain-anthropic

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 helpful."),
    ("human", "{question}"),
])

llm = ChatOpenAI(model="gpt-5")
parser = StrOutputParser()

chain = prompt | llm | parser

result = chain.invoke({"question": "Hello"})
# async
result = await chain.ainvoke({"question": "Hello"})
# batch
results = chain.batch([{"question": "Hi"}, {"question": "Hey"}])
# stream
for chunk in chain.stream({"question": "Hi"}): print(chunk, end="")

from langchain_core.runnables import RunnableLambda, RunnablePassthrough

upper = RunnableLambda(lambda x: x.upper())
chain = upper | (lambda x: x + "!")

# Parallel
from langchain_core.runnables import RunnableParallel

chain = RunnableParallel(
    upper=upper,
    lower=RunnableLambda(str.lower),
)

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

vectorstore = Chroma.from_texts(texts, OpenAIEmbeddings())
retriever = vectorstore.as_retriever(k=5)

prompt = ChatPromptTemplate.from_template("""
Answer using context:
{context}

Question: {question}
""")

chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

answer = chain.invoke("What is X?")

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

class Result(BaseModel):
    name: str
    score: int

parser = PydanticOutputParser(pydantic_object=Result)

prompt = ChatPromptTemplate.from_messages([
    ("system", "Output as JSON."),
    ("human", "{q}\n\n{format_instructions}"),
]).partial(format_instructions=parser.get_format_instructions())

chain = prompt | llm | parser
result: Result = chain.invoke({"q": "..."})

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

store = {}

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

with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="question",
    history_messages_key="history",
)

with_history.invoke(
    {"question": "Hi, my name is Alice"},
    config={"configurable": {"session_id": "user-1"}},
)

from langchain_community.document_loaders import (
    TextLoader, PyPDFLoader, WebBaseLoader, DirectoryLoader,
    UnstructuredMarkdownLoader, GitLoader,
)

docs = TextLoader("file.txt").load()
docs = PyPDFLoader("file.pdf").load_and_split()
docs = WebBaseLoader("https://example.com").load()

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", " ", ""],
)
chunks = splitter.split_documents(docs)

from langchain_community.vectorstores import Chroma, Qdrant, FAISS, Pinecone
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Persistent
db = Chroma.from_documents(chunks, embeddings, persist_directory="./db")
db = Chroma(persist_directory="./db", embedding_function=embeddings)

from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

@tool
def search(query: str) -> str:
    """Search the web."""
    return "results..."

agent = create_react_agent(model=llm, tools=[search])
result = agent.invoke({"messages": [{"role": "user", "content": "Find info on X"}]})

async for event in chain.astream_events(input, version="v2"):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="")

from langchain_core.runnables import ConfigurableField

llm = ChatOpenAI().configurable_fields(
    model_name=ConfigurableField(id="model"),
    temperature=ConfigurableField(id="temperature"),
)

# Override at runtime
result = chain.invoke(input, config={"configurable": {"model": "gpt-5", "temperature": 0}})

export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY=...

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader

docs = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(docs)
engine = index.as_query_engine()
print(engine.query("What is X?"))