A shopping assistant agent langgraph implementation lets you combine retrieval, tool use, and multi-step reasoning without wiring a brittle prompt chain by hand. This tutorial builds a working agent that searches a product catalog and explains its recommendations, using LangGraph’s stateful graph and an OpenAI-compatible chat model.
Prerequisites
- Python 3.10 or newer.
pip install langgraph langchain-openai langchain-core- An API key for an OpenAI-compatible endpoint. You can use OpenAI directly, or point at n4n.ai’s single endpoint that addresses 240+ models with automatic fallback when a provider is degraded.
- Familiarity with Python typing and basic LangChain message objects.
Set the key in your environment:
export OPENAI_API_KEY=sk-...
export BASE_URL=https://api.openai.com/v1 # or https://api.n4n.ai/v1Design goals for the shopping assistant agent langgraph
The agent must accept natural language, call a search tool when it needs inventory data, and loop back to the model to synthesize a final answer. We keep the tool interface strict so the LLM can’t hallucinate fields, and we persist message history in graph state to avoid re-implementing conversation memory.
The product catalog and search tool
We’ll embed a tiny catalog as JSON. In production you’d query a vector store or SQL, but the tool interface stays identical.
[
{"id": 1, "name": "SoundCore Wireless Headphones", "category": "audio", "price": 79.99, "rating": 4.5},
{"id": 2, "name": "Studio Over-Ear Headphones", "category": "audio", "price": 149.00, "rating": 4.2},
{"id": 3, "name": "Travel Bluetooth Earbuds", "category": "audio", "price": 49.99, "rating": 4.0},
{"id": 4, "name": "Mechanical Keyboard", "category": "accessories", "price": 89.99, "rating": 4.7}
]Load it and wrap a search function as a LangChain tool. The decorator handles schema extraction.
import json
from langchain_core.tools import tool
CATALOG = json.loads(open("catalog.json").read())
@tool
def search_products(query: str, max_price: float = None) -> str:
"""Search the catalog for products by name or category, optionally capping price."""
matches = [p for p in CATALOG if query.lower() in p["name"].lower() or query.lower() in p["category"].lower()]
if max_price is not None:
matches = [p for p in matches if p["price"] <= max_price]
return json.dumps(matches)The tool returns a stringified list. Keeping the payload as JSON makes the model’s post-processing deterministic.
Agent state and graph scaffolding
LangGraph tracks conversation state as a typed dictionary. We use operator.add so each node appends messages rather than replacing them.
from typing import TypedDict, Annotated
import operator
class State(TypedDict):
messages: Annotated[list, operator.add]A StateGraph defines nodes and edges. Our nodes: model (calls LLM), tools (executes tool calls). The graph loops between them until the model stops requesting tools.
Model node and binding
Instantiate the chat model with bind_tools. Temperature zero keeps extraction factual.
from langchain_openai import ChatOpenAI
import os
model = ChatOpenAI(
model="gpt-4o-mini",
temperature=0,
base_url=os.environ["BASE_URL"],
api_key=os.environ["OPENAI_API_KEY"],
).bind_tools([search_products])
def model_node(state: State):
response = model.invoke(state["messages"])
return {"messages": [response]}Tool execution node
The model node may emit tool_calls. We map each call to a ToolMessage keyed by tool_call_id, which the model consumes on the next pass.
from langchain_core.messages import ToolMessage
def tool_node(state: State):
last_msg = state["messages"][-1]
tool_messages = []
for tc in last_msg.tool_calls:
if tc["name"] == "search_products":
result = search_products.invoke(tc["args"])
tool_messages.append(
ToolMessage(content=result, tool_call_id=tc["id"])
)
return {"messages": tool_messages}Conditional routing
After the model speaks, we check whether it requested a tool. If yes, run tools; otherwise end.
from langgraph.graph import StateGraph, END
def should_continue(state: State):
last = state["messages"][-1]
if getattr(last, "tool_calls", None):
return "tools"
return END
graph = StateGraph(State)
graph.add_node("model", model_node)
graph.add_node("tools", tool_node)
graph.add_conditional_edges("model", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "model")
app = graph.compile()Running the shopping assistant agent langgraph
Invoke with a user question. The agent will call search_products, receive JSON, and synthesize an answer.
from langchain_core.messages import HumanMessage
query = "Find wireless headphones under $100"
result = app.invoke({"messages": [HumanMessage(content=query)]})
for m in result["messages"]:
print(f"{m.type}: {m.content}")Expected output (abridged):
human: Find wireless headphones under $100
ai: (tool_call: search_products args={'query':'wireless headphones','max_price':100})
tool: [{"id":1,"name":"SoundCore Wireless Headphones","price":79.99,"rating":4.5},{"id":3,"name":"Travel Bluetooth Earbuds","price":49.99,"rating":4.0}]
ai: You have two options under $100: SoundCore Wireless Headphones at $79.99 (4.5★) and Travel Bluetooth Earbuds at $49.99 (4.0★). The SoundCore offers better rating for home use; the earbuds win on portability.The loop ran model → tools → model. No manual prompt engineering for the second turn; LangGraph persisted state.
Adding recommendation scoring
A pure search isn’t a shopping assistant agent langgraph strength unless it ranks. Extend the tool to weight rating and price:
@tool
def search_and_rank(query: str, max_price: float = None) -> str:
"""Return top products by rating/price value."""
matches = json.loads(search_products.invoke({"query": query, "max_price": max_price}))
for m in matches:
m["score"] = m["rating"] / (m["price"]/10)
matches.sort(key=lambda x: x["score"], reverse=True)
return json.dumps(matches[:2])Bind the new tool, swap search_products for search_and_rank in model.bind_tools, and the same graph works. The model now receives pre-ranked candidates and can focus on explaining trade-offs.
Handling empty results
Production catalogs return zero matches. Modify tool_node to emit a clear signal:
def tool_node(state: State):
last_msg = state["messages"][-1]
tool_messages = []
for tc in last_msg.tool_calls:
if tc["name"] == "search_products":
result = search_products.invoke(tc["args"])
if not json.loads(result):
result = json.dumps({"error": "no matches"})
tool_messages.append(ToolMessage(content=result, tool_call_id=tc["id"]))
return {"messages": tool_messages}The model will then tell the user no products fit, instead of inventing specs.
Streaming and debugging
For UX, stream tokens:
for chunk in app.stream({"messages": [HumanMessage(content=query)]}):
print(chunk)To debug tool calls, log last_msg.tool_calls inside tool_node. If the model repeatedly calls the tool with bad args, tighten the tool description or add Pydantic validation.
Production considerations
Streaming: wrap app.stream to yield tokens per node. Durability: LangGraph persists to a checkpointer; pass checkpointer=memory for dev, Postgres for prod. Observability: log state["messages"] lengths to catch runaway loops.
If you point BASE_URL at n4n.ai, the gateway forwards provider cache-control hints and honors client routing directives, so repeated catalog queries hit cache and cut cost. The automatic fallback also keeps the agent alive when a single model provider returns 429s.
Complete script
import json, os
from typing import TypedDict, Annotated
import operator
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
CATALOG = json.loads(open("catalog.json").read())
@tool
def search_products(query: str, max_price: float = None) -> str:
matches = [p for p in CATALOG if query.lower() in p["name"].lower() or query.lower() in p["category"].lower()]
if max_price is not None:
matches = [p for p in matches if p["price"] <= max_price]
return json.dumps(matches)
class State(TypedDict):
messages: Annotated[list, operator.add]
model = ChatOpenAI(
model="gpt-4o-mini",
temperature=0,
base_url=os.environ["BASE_URL"],
api_key=os.environ["OPENAI_API_KEY"],
).bind_tools([search_products])
def model_node(state: State):
return {"messages": [model.invoke(state["messages"])]}
def tool_node(state: State):
last = state["messages"][-1]
msgs = []
for tc in last.tool_calls:
if tc["name"] == "search_products":
res = search_products.invoke(tc["args"])
msgs.append(ToolMessage(content=res, tool_call_id=tc["id"]))
return {"messages": msgs}
def should_continue(state: State):
if getattr(state["messages"][-1], "tool_calls", None):
return "tools"
return END
g = StateGraph(State)
g.add_node("model", model_node)
g.add_node("tools", tool_node)
g.add_conditional_edges("model", should_continue, {"tools": "tools", END: END})
g.add_edge("tools", "model")
app = g.compile()
print(app.invoke({"messages": [HumanMessage(content="Find wireless headphones under $100")]})["messages"][-1].content)Where to take it next
Replace the JSON file with a real search backend (Elasticsearch, pgvector). Add a cart node that calls an order API. Use LangGraph’s subgraphs to isolate the shopping assistant agent langgraph from a larger support bot. The pattern—model proposes, tools act, graph loops—scales to most e-commerce copilots.