LangChain and LangGraph

LangChain: Building Blocks

LangChain organizes LLM applications around five primitive types:

The simplest possible chain connects these with the pipe operator `|`: ```python 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 a helpful assistant."), ("user", "{question}") ]) model = ChatOpenAI(model="gpt-4o-mini") parser = StrOutputParser()

# LCEL chain: prompt → model → parser chain = prompt | model | parser result = chain.invoke({"question": "What is RAG in AI?"}) print(result) ``` This uses LangChain Expression Language (LCEL), where `|` composes runnables left-to-right, analogous to Unix pipes.

Giving the LLM Tools

A tool is just a Python function decorated with `@tool`. The LLM decides whether and when to call it — you never hard-code that decision: ```python from langchain_core.tools import tool from langchain_openai import ChatOpenAI @tool def multiply(a: float, b: float) -> float: """Multiply two numbers together.""" return a * b llm = ChatOpenAI(model="gpt-4o-mini") agent = llm.bind_tools([multiply]) response = agent.invoke("What is 17.5 times 4.2?")

# The model returns a tool_call dict, not a plain string print(response.tool_calls)

# [{'name': 'multiply', 'args': {'a': 17.5, 'b': 4.2}, 'id': '...'}] ``` The key concept is the ReAct loop (Reason + Act): the LLM reasons about what to do, calls a tool, observes the result, and reasons again. Formally, if we denote the agent's state at step $t$ as the tuple of observations so far, the next action is:

Explanation: The agent policy $\pi$ (parameterized by the LLM weights $\theta$) maps the full history of observations to the next action — meaning every previous tool result is visible when the agent decides what to do next. Here, $a_t$ is the chosen action (tool call or final answer), and $o_t$ is the observation returned by the environment (tool output or user message).

LangGraph: State as a Graph

LangGraph models the entire agentic workflow as a StateGraph. You define: 1. State — a typed Python dict that flows through every node 2. Nodes — Python functions that read state and return updates 3. Edges — unconditional or conditional transitions between nodes ```python from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated import operator

# 1. Define state schema class AgentState(TypedDict): messages: Annotated[list, operator.add] # messages accumulate tool_calls_remaining: int

# 2. Define nodes def call_llm(state: AgentState) -> dict:

# ... invoke LLM, return new messages return {"messages": [llm_response]} def call_tool(state: AgentState) -> dict:

# ... execute the tool call return {"messages": [tool_result]}

# 3. Conditional edge: should we loop back or finish? def should_continue(state: AgentState) -> str: last = state["messages"][-1] if last.tool_calls and state["tool_calls_remaining"] > 0: return "call_tool" # → loop return END # → finish

# 4. Build the graph graph = StateGraph(AgentState) graph.add_node("call_llm", call_llm) graph.add_node("call_tool", call_tool) graph.set_entry_point("call_llm") graph.add_conditional_edges("call_llm", should_continue) graph.add_edge("call_tool", "call_llm") # tool result feeds back to LLM app = graph.compile() result = app.invoke({"messages": [user_msg], "tool_calls_remaining": 5}) ``` The graph state update at each node follows an accumulation rule. If a node returns a partial state update $\Delta s$, the new state is:

Explanation: The $\oplus$ operator represents a merge — LangGraph applies each node's returned dict as a partial update to the shared state, using reducer functions (like `operator.add` for lists) to combine old and new values.

Retrieval-Augmented Generation (RAG) with LangChain

One of the most common LangChain patterns is RAG: fetching relevant documents from a vector store and injecting them into the prompt before the LLM answers. ```python from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_community.vectorstores import FAISS from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough

# Build a tiny vector store docs = ["LangGraph uses nodes and edges.", "FAISS is a vector similarity library."] store = FAISS.from_texts(docs, embedding=OpenAIEmbeddings()) retriever = store.as_retriever(search_kwargs={"k": 2}) prompt = ChatPromptTemplate.from_template( "Answer using only this context:\n{context}\n\nQuestion: {question}" ) rag_chain = ( {"context": retriever, "question": RunnablePassthrough()} | prompt | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser() ) print(rag_chain.invoke("What does LangGraph use?")) ``` The retrieval step selects the top-$k$ documents by cosine similarity between the query embedding $\mathbf{q}$ and each document embedding $\mathbf{d}_i$:

Explanation: Cosine similarity measures the angle between two embedding vectors — a score of 1 means identical direction (highly relevant), 0 means orthogonal (unrelated), and the retriever returns the documents with the highest scores.

Retrieval-Augmented Generation · Transformer Architecture · Prompt Engineering · Vector Databases · Autonomous AI Agents