Multimodal agents are where LangGraph’s stateful graph architecture shines. Unlike linear chains, a graph lets you branch on image content, loop for clarification, and persist context across turns — exactly what a vision-enabled assistant needs. This tutorial builds a complete agent that accepts images, reasons over them with GPT-4o, and uses tools to act on what it sees.
Prerequisites
You need Python 3.10+, an OpenAI API key with GPT-4o access, and these packages:
pip install langgraph langchain-openai langchain-core pydantic pillow python-dotenvCreate a .env file with your key:
OPENAI_API_KEY=sk-...The code below assumes a project structure like:
multimodal-agent/
├── .env
├── agent.py
├── tools.py
├── state.py
└── main.pyDefine the state
LangGraph’s power comes from a typed state schema that flows through every node. For a vision agent, we need the conversation history, the current image (if any), and a scratchpad for tool results.
# state.py
from typing import Annotated, Sequence, Optional
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langgraph.graph.message import add_messages
from pydantic import BaseModel, Field
class AgentState(BaseModel):
messages: Annotated[Sequence[BaseMessage], add_messages] = Field(default_factory=list)
current_image_b64: Optional[str] = None
image_media_type: Optional[str] = None
tool_scratchpad: dict = Field(default_factory=dict)add_messages handles message concatenation automatically. The image fields let us pass a base64-encoded frame between nodes without stuffing it into the message list every turn.
Build the vision-capable LLM node
GPT-4o accepts images as base64 data URLs or raw base64 in the content array. We’ll write a node that injects the current image into the message payload when present.
# agent.py
import base64
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from state import AgentState
llm = ChatOpenAI(model="gpt-4o", temperature=0)
VISION_SYSTEM_PROMPT = """You are a multimodal assistant. You can see images the user provides.
When an image is present, describe what you see concisely, then answer the user's question.
If you need to act (search, calculate, read a file), call the appropriate tool.
Never hallucinate tool outputs — only use what the tools return."""
def build_vision_messages(state: AgentState) -> list:
"""Construct the message list for the LLM, injecting the current image if present."""
messages = [SystemMessage(content=VISION_SYSTEM_PROMPT)]
messages.extend(state.messages)
if state.current_image_b64:
# Replace the last human message with a multimodal version
last_human = None
for msg in reversed(state.messages):
if isinstance(msg, HumanMessage):
last_human = msg
break
if last_human:
# Remove the text-only version
messages = [m for m in messages if m is not last_human]
# Add multimodal version
content = [
{"type": "text", "text": last_human.content},
{
"type": "image_url",
"image_url": {
"url": f"data:{state.image_media_type};base64,{state.current_image_b64}",
"detail": "high"
}
}
]
messages.append(HumanMessage(content=content))
return messages
async def vision_node(state: AgentState) -> AgentState:
response = await llm.ainvoke(build_vision_messages(state))
return AgentState(
messages=state.messages + [response],
current_image_b64=state.current_image_b64,
image_media_type=state.image_media_type,
tool_scratchpad=state.tool_scratchpad
)Key detail: we reconstruct the last HumanMessage as a multimodal message only when an image exists. This keeps the conversation history clean — earlier turns without images stay text-only.
Add tools the agent can call
A vision agent needs tools to act on what it sees. We’ll implement three: web search for context, OCR for text extraction, and a calculator for measurements.
# tools.py
import base64
import io
import requests
from typing import Annotated
from langchain_core.tools import tool
from pydantic import BaseModel, Field
from PIL import Image
import pytesseract
class SearchInput(BaseModel):
query: str = Field(description="Search query")
@tool(args_schema=SearchInput)
def web_search(query: str) -> str:
"""Search the web for current information. Returns summarized results."""
# In production, use a real search API (SerpAPI, Tavily, etc.)
# This stub returns a deterministic response for tutorial purposes.
return f"Search results for '{query}': [simulated result — replace with real API]"
class OCRInput(BaseModel):
image_b64: str = Field(description="Base64-encoded image")
media_type: str = Field(description="MIME type, e.g., image/png")
@tool(args_schema=OCRInput)
def extract_text(image_b64: str, media_type: str) -> str:
"""Extract text from an image using OCR."""
try:
img_data = base64.b64decode(image_b64)
image = Image.open(io.BytesIO(img_data))
text = pytesseract.image_to_string(image)
return text.strip() or "No text detected"
except Exception as e:
return f"OCR failed: {e}"
class CalculatorInput(BaseModel):
expression: str = Field(description="Math expression to evaluate")
@tool(args_schema=CalculatorInput)
def calculate(expression: str) -> str:
"""Evaluate a mathematical expression safely."""
allowed = set("0123456789+-*/(). ")
if not set(expression).issubset(allowed):
return "Invalid characters in expression"
try:
result = eval(expression, {"__builtins__": {}}, {})
return str(result)
except Exception as e:
return f"Calculation error: {e}"
TOOLS = [web_search, extract_text, calculate]
TOOL_MAP = {t.name: t for t in TOOLS}The OCR tool uses pytesseract — install tesseract-ocr system package if you run this locally. The calculator uses a restricted eval for simplicity; production code should use a proper expression parser.
Wire the tool-calling node
LangGraph’s ToolNode handles tool execution, but we need to bind tools to the model and route based on whether the model requested a tool call.
# agent.py (continued)
from langgraph.prebuilt import ToolNode
from langchain_core.messages import ToolMessage
llm_with_tools = llm.bind_tools(TOOLS)
tool_node = ToolNode(TOOLS)
async def vision_node_with_tools(state: AgentState) -> AgentState:
messages = build_vision_messages(state)
response = await llm_with_tools.ainvoke(messages)
return AgentState(
messages=state.messages + [response],
current_image_b64=state.current_image_b64,
image_media_type=state.image_media_type,
tool_scratchpad=state.tool_scratchpad
)
def should_continue(state: AgentState) -> str:
last_message = state.messages[-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
return "end"Build the graph
Now compose the nodes into a graph with conditional edges. The flow: vision node → (tools → vision node)* → end.
# agent.py (continued)
from langgraph.graph import StateGraph, END
workflow = StateGraph(AgentState)
workflow.add_node("vision", vision_node_with_tools)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("vision")
workflow.add_conditional_edges(
"vision",
should_continue,
{"tools": "tools", "end": END}
)
workflow.add_edge("tools", "vision")
app = workflow.compile()That’s the entire graph. The conditional edge loops back to vision after each tool call, letting the model reason over tool results and decide whether to call another tool or respond.
Encode images and run the agent
A helper to load images from disk or URL, then a simple REPL to test.
# main.py
import base64
import mimetypes
import asyncio
from pathlib import Path
from PIL import Image
import io
from agent import app
from state import AgentState
from langchain_core.messages import HumanMessage
def encode_image(path_or_url: str) -> tuple[str, str]:
"""Return (base64_string, media_type) for a local file or HTTP URL."""
if path_or_url.startswith(("http://", "https://")):
import httpx
resp = httpx.get(path_or_url, timeout=30)
resp.raise_for_status()
data = resp.content
media_type = resp.headers.get("content-type", "image/png")
else:
path = Path(path_or_url)
data = path.read_bytes()
media_type, _ = mimetypes.guess_type(str(path))
media_type = media_type or "image/png"
return base64.b64encode(data).decode(), media_type
async def run_turn(state: AgentState, user_text: str, image_path: str | None = None) -> AgentState:
if image_path:
b64, mime = encode_image(image_path)
state.current_image_b64 = b64
state.image_media_type = mime
else:
state.current_image_b64 = None
state.image_media_type = None
state.messages.append(HumanMessage(content=user_text))
result = await app.ainvoke(state)
return AgentState(**result)
async def main():
state = AgentState()
print("Multimodal agent ready. Type 'quit' to exit.")
print("Prefix image path with '@' to attach: 'What is this? @/path/to/image.png'")
while True:
user_input = input("\n> ").strip()
if user_input.lower() in ("quit", "exit"):
break
image_path = None
text = user_input
if "@" in user_input:
parts = user_input.split("@", 1)
text = parts[0].strip()
image_path = parts[1].strip()
state = await run_turn(state, text, image_path)
last_msg = state.messages[-1]
print(f"\nAssistant: {last_msg.content}")
if __name__ == "__main__":
asyncio.run(main())Run it:
python main.pyExpected output at checkpoints
Checkpoint 1 — Image description without tools
> What's in this image? @./receipt.png
Assistant: The image shows a receipt from "Green Valley Market" dated 2024-01-15.
Items: Organic apples ($3.99), Whole milk ($4.49), Sourdough bread ($5.99).
Subtotal: $14.47, Tax: $1.16, Total: $15.63. Paid with Visa ending in 4242.Checkpoint 2 — OCR tool invocation
> Extract all text from this receipt @./receipt.png
Assistant: I'll extract the text for you.
[Tool call: extract_text]
Assistant: GREEN VALLEY MARKET
123 Main St, Springfield
Date: 01/15/2024 14:32
--------------------------------
Organic Apples $3.99
Whole Milk $4.49
Sourdough Bread $5.99
--------------------------------
Subtotal $14.47
Tax $1.16
Total $15.63
Visa ****4242 $15.63
--------------------------------
Thank you for shopping!Checkpoint 3 — Multi-step reasoning with search
> This receipt is from a store in Springfield. What's the sales tax rate there? @./receipt.png
Assistant: The receipt shows tax of $1.16 on a $14.47 subtotal, which is 8%.
Let me verify the current Springfield sales tax rate.
[Tool call: web_search with query "Springfield sales tax rate 2024"]
Assistant: According to current data, Springfield's combined sales tax rate is 8.0%
(state 6.25% + city 1.75%), which matches the receipt calculation.Handling conversation memory
The AgentState.messages list accumulates every turn. For long conversations, you’ll want a summarization node that triggers after N messages. Add this to the graph:
# agent.py (addition)
from langchain_core.messages import RemoveMessage
SUMMARIZATION_THRESHOLD = 20
async def maybe_summarize(state: AgentState) -> AgentState:
if len(state.messages) <= SUMMARIZATION_THRESHOLD:
return state
# Keep system prompt + last 10 messages, summarize the rest
to_summarize = state.messages[1:-10] # skip system, keep recent
summary_prompt = [
SystemMessage(content="Summarize the following conversation concisely."),
*to_summarize
]
summary = await llm.ainvoke(summary_prompt)
# Remove summarized messages, insert summary
remove_ids = [m.id for m in to_summarize]
new_messages = [
SystemMessage(content=VISION_SYSTEM_PROMPT),
AIMessage(content=f"[Conversation summary: {summary.content}]"),
*state.messages[-10:]
]
return AgentState(
messages=new_messages,
current_image_b64=state.current_image_b64,
image_media_type=state.image_media_type,
tool_scratchpad=state.tool_scratchpad
)Insert it before the vision node:
workflow.add_node("summarize", maybe_summarize)
workflow.set_entry_point("summarize")
workflow.add_edge("summarize", "vision")Production considerations
Streaming responses — Replace ainvoke with astream in the vision node to yield tokens as they arrive. The graph supports streaming natively via app.astream().
Structured outputs — Use llm.with_structured_output(PydanticModel) for the final answer node if you need guaranteed JSON.
Image size limits — GPT-4o accepts up to 20MB per image. Downscale large uploads server-side:
def downscale_if_needed(image_b64: str, max_dim: int = 2048) -> str:
img_data = base64.b64decode(image_b64)
image = Image.open(io.BytesIO(img_data))
if max(image.size) > max_dim:
image.thumbnail((max_dim, max_dim))
buf = io.BytesIO()
image.save(buf, format=image.format or "PNG")
return base64.b64encode(buf.getvalue()).decode()
return image_b64Provider fallback — If you route through a gateway like n4n.ai, you can specify fallback models in the routing header when the primary provider is degraded, keeping the agent available without code changes.
Observability — Add LangSmith tracing by setting LANGCHAIN_TRACING_V2=true and LANGCHAIN_API_KEY. Every node execution, tool call, and LLM invocation appears in the trace tree.
Extending the agent
Three high-value additions:
- Document QA node — Add a RAG tool that indexes PDFs/images and retrieves relevant chunks before the vision node runs.
- Human-in-the-loop — Insert an interrupt node before destructive actions (sending email, placing orders) using
graph.update_statewith aNodeInterrupt. - Video frame sampling — For video input, extract frames at 1fps, run the vision node on each, then aggregate with a summarization node.
The graph structure stays the same — you only add nodes and edges. That’s the point of LangGraph: the control flow is data, not code.