LangChain: Complete Interactive Notes

pip install -U langchain langchain-openai langchain-text-splitters
export OPENAI_API_KEY="sk-..."          # Windows PowerShell: $env:OPENAI_API_KEY="sk-..."

"""Stand-ins so every example runs without an API key.
In real code replace ScriptedModel(...) with e.g. init_chat_model("openai:gpt-4o-mini")."""
import re, math
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.embeddings import Embeddings
from langchain_core.messages import AIMessage

class ScriptedModel(GenericFakeChatModel):
    """Replies with the messages you give it, one per call (text or tool calls)."""
    def __init__(self, replies):
        super().__init__(messages=iter([r if isinstance(r, AIMessage) else AIMessage(content=r) for r in replies]))
    def bind_tools(self, tools, **kwargs):
        return self

class KeywordEmbeddings(Embeddings):
    """A tiny bag-of-words embedding: similar words -> similar vectors. Real apps use OpenAIEmbeddings etc."""
    def __init__(self, dim=256): self.dim = dim
    def _vec(self, text):
        v = [0.0] * self.dim
        for w in re.findall(r"[a-z]+", text.lower()):
            w = w[:5]                                   # crude stemming: 'withdrawal' ~ 'withdraw'
            v[sum(map(ord, w)) % self.dim] += 1
        n = math.sqrt(sum(x * x for x in v)) or 1
        return [x / n for x in v]
    def embed_documents(self, texts): return [self._vec(t) for t in texts]
    def embed_query(self, text): return self._vec(text)

from langchain_core.messages import SystemMessage, HumanMessage
from fake_llm import ScriptedModel
# real:  from langchain.chat_models import init_chat_model
#        model = init_chat_model("openai:gpt-4o-mini", temperature=0)
model = ScriptedModel(["UPI lets you send money instantly from your phone.", "Use NEFT for scheduled transfers.", "IMPS works 24x7."])

messages = [SystemMessage("You are a helpful banking tutor. Answer in one line."),
            HumanMessage("What is UPI?")]
reply = model.invoke(messages)                 # 1. invoke: one input -> one AIMessage
print(type(reply).__name__, "->", reply.content)

for chunk in model.stream("When should I use NEFT?"):   # 2. stream: tokens as they arrive
    print(chunk.content, end="|")
print()

print([r.content for r in model.batch(["What is IMPS?"])])   # 3. batch: many inputs at once

from langchain.chat_models import init_chat_model

gpt    = init_chat_model("openai:gpt-4o-mini", temperature=0)
claude = init_chat_model("anthropic:claude-sonnet-4-5")        # needs langchain-anthropic
local  = init_chat_model("ollama:llama3.1")                    # free, runs on your laptop
# all three have the same invoke / stream / batch methods

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a {bank} support assistant. Reply in {language}."),
    MessagesPlaceholder("history"),                       # slot for earlier chat messages
    ("human", "{question}"),
])
print("variables:", sorted(prompt.input_variables))

hindi = prompt.partial(language="Hindi")                  # fill some blanks now, the rest later
value = hindi.invoke({"bank": "NovaBank", "history": [("human", "Hi"), ("ai", "Hello! How can I help?")],
                      "question": "How do I block my card?"})
for m in value.to_messages():
    print(f"{m.type:6s}| {m.content}")

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda, RunnableParallel, RunnablePassthrough
from fake_llm import ScriptedModel

model = ScriptedModel(["Fixed deposits pay a fixed interest rate for a fixed period."])
explain = ChatPromptTemplate.from_template("Explain {topic} in one line.") | model | StrOutputParser()
print("1.", explain.invoke({"topic": "FD"}))

# RunnableParallel: run branches on the same input at the same time
sentiment = ChatPromptTemplate.from_template("Sentiment of: {text}") | ScriptedModel(["positive"]) | StrOutputParser()
simple = ChatPromptTemplate.from_template("Explain {text} to a child") | ScriptedModel(["An FD is like a locked savings box that pays you extra."]) | StrOutputParser()
both = RunnableParallel(sentiment=sentiment, for_kids=simple)
print("2.", both.invoke({"text": "FD"}))

# RunnablePassthrough.assign adds new keys; RunnableLambda wraps any Python function
pipeline = RunnablePassthrough.assign(words=RunnableLambda(lambda d: len(d["text"].split())))
print("3.", pipeline.invoke({"text": "my fixed deposit matured today"}))
print("4. input keys:", list(explain.input_schema.model_json_schema()["properties"]))

from typing import Literal
from pydantic import BaseModel, Field
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.exceptions import OutputParserException
from fake_llm import ScriptedModel

class Ticket(BaseModel):
    category: Literal["card", "upi", "loan", "other"]
    priority: Literal["low", "medium", "high"]
    summary: str = Field(description="one short sentence")

parser = PydanticOutputParser(pydantic_object=Ticket)
prompt = ChatPromptTemplate.from_template("Classify the complaint.\n{format}\n\nComplaint: {text}").partial(
    format=parser.get_format_instructions())

good = prompt | ScriptedModel(['{"category": "upi", "priority": "high", "summary": "UPI debit without credit."}']) | parser
t = good.invoke({"text": "Money left my account via UPI but the shop never got it!"})
print(type(t).__name__, t.category, t.priority, "|", t.summary)

bad = prompt | ScriptedModel(['{"category": "atm", "priority": "urgent"}']) | parser
try:
    bad.invoke({"text": "ATM ate my card"})
except OutputParserException as e:
    print("rejected:", str(e).split(". Got:")[1].split("\n")[0].strip()[:80])

# with a real model you can skip the parser:  model.with_structured_output(Ticket)

from langchain_core.tools import tool
from langchain_core.messages import AIMessage
from langchain.agents import create_agent
from fake_llm import ScriptedModel

@tool
def get_balance(account: str) -> str:
    """Get the current balance of a bank account (last 4 digits)."""
    return {"1234": "₹12,500", "9876": "₹3,200"}.get(account, "account not found")

@tool
def convert_currency(amount_inr: float, to: str) -> str:
    """Convert an amount in Indian rupees to another currency."""
    rates = {"USD": 0.0105, "EUR": 0.0098}          # example rates
    return f"{amount_inr * rates[to]:.2f} {to}"

print("tool schema:", get_balance.name, get_balance.args)

# The scripted model plays the LLM: call a tool, call another, then answer.
model = ScriptedModel([
    AIMessage("", tool_calls=[{"name": "get_balance", "args": {"account": "1234"}, "id": "c1"}]),
    AIMessage("", tool_calls=[{"name": "convert_currency", "args": {"amount_inr": 12500, "to": "USD"}, "id": "c2"}]),
    "Your balance is ₹12,500, which is about 131.25 USD.",
])
agent = create_agent(model, tools=[get_balance, convert_currency],
                     system_prompt="You are NovaBank's assistant. Use tools; never guess numbers.")

result = agent.invoke({"messages": [{"role": "user", "content": "How much is my 1234 account worth in dollars?"}]})
for m in result["messages"]:
    detail = m.tool_calls[0]["name"] + str(m.tool_calls[0]["args"]) if getattr(m, "tool_calls", None) else m.content
    print(f"{m.type:5s} -> {detail}")

from langchain_core.documents import Document
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_text_splitters import RecursiveCharacterTextSplitter
from fake_llm import KeywordEmbeddings, ScriptedModel
# real: from langchain_openai import OpenAIEmbeddings; from langchain_chroma import Chroma

policy = Document(page_content="""Fixed Deposit Policy.
Eligibility: any resident above 18 years can open an FD.
Premature withdrawal: breaking an FD early reduces the interest rate by 1%.
Senior citizens get an extra 0.50% interest on all tenures.
Loan against FD: you can borrow up to 90% of the FD amount.""", metadata={"source": "fd_policy.pdf"})

chunks = RecursiveCharacterTextSplitter(chunk_size=90, chunk_overlap=0).split_documents([policy])
print(len(chunks), "chunks")
store = InMemoryVectorStore.from_documents(chunks, KeywordEmbeddings())
retriever = store.as_retriever(search_kwargs={"k": 2})

for d in retriever.invoke("What happens if I withdraw my FD early?"):
    print("  found:", d.page_content)

prompt = ChatPromptTemplate.from_template("Answer only from the context.\nContext:\n{context}\n\nQuestion: {question}")
format_docs = lambda docs: "\n".join(d.page_content for d in docs)
rag = ({"context": retriever | format_docs, "question": RunnablePassthrough()}
       | prompt | ScriptedModel(["Breaking an FD early reduces your interest rate by 1% [fd_policy.pdf]."]) | StrOutputParser())
print("answer:", rag.invoke("What happens if I withdraw my FD early?"))

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from fake_llm import ScriptedModel

agent = create_agent(ScriptedModel(["Nice to meet you, Asha!", "Your name is Asha.", "I don't know your name yet."]),
                     tools=[], checkpointer=InMemorySaver())       # the checkpointer saves each conversation

asha = {"configurable": {"thread_id": "asha-1"}}                   # one thread_id per conversation
agent.invoke({"messages": [{"role": "user", "content": "Hi, I am Asha"}]}, asha)
r = agent.invoke({"messages": [{"role": "user", "content": "What is my name?"}]}, asha)
print("asha thread:", len(r["messages"]), "messages ->", r["messages"][-1].content)

ravi = {"configurable": {"thread_id": "ravi-1"}}                   # a new thread starts empty
r = agent.invoke({"messages": [{"role": "user", "content": "What is my name?"}]}, ravi)
print("ravi thread:", len(r["messages"]), "messages ->", r["messages"][-1].content)

from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware, ModelCallLimitMiddleware
from fake_llm import ScriptedModel

agent = create_agent(
    ScriptedModel(["I have noted your request."]), tools=[],
    middleware=[
        PIIMiddleware("email", strategy="redact"),              # hide emails before the model sees them
        ModelCallLimitMiddleware(run_limit=5),                  # stop runaway loops
    ])
r = agent.invoke({"messages": [{"role": "user", "content": "Send the statement to priya@example.com"}]})
print("model saw:", r["messages"][0].content)

for step in agent.stream({"messages": [("user", "Balance in USD?")]}, stream_mode="updates"):
    print(step)          # one dict per finished step: model call, tool call, final answer

LANGSMITH_TRACING=true
LANGSMITH_API_KEY=lsv2_...
LANGSMITH_PROJECT=novabank-assistant