Introduction to Tool Use Patterns with LangGraph
LangGraph is a framework built on top of LangChain that allows developers to create stateful, multi-actor applications using graphs. One of its most powerful capabilities is orchestrating tool use — the ability for language models to invoke external functions, APIs, databases, and other utilities as part of a reasoning loop. In this guide, we'll explore the most common and advanced tool use patterns with LangGraph, complete with runnable code examples.
What Is Tool Use in LangGraph?
Tool use (also called function calling) is the mechanism by which a language model decides to call an external function with structured arguments, receives the result, and incorporates that result into its reasoning. LangGraph extends this concept by letting you model tool use as a graph: nodes represent steps (the LLM, individual tools, validators), and edges represent the flow of control and state between them.
This graph-based approach is what separates LangGraph from a simple "call the model, call the tool, repeat" loop. You get fine-grained control over branching, parallelism, retries, human approval, and memory.
Why Tool Use Patterns Matter
- Reliability: Structured patterns make tool calls predictable and testable.
- Control: Graphs let you insert validation, retries, and human-in-the-loop checkpoints.
- Scalability: Patterns like parallel tool execution reduce latency for multi-step tasks.
- Observability: Each node and edge can be logged, traced, and debugged independently.
- Composability: Reusable subgraphs let you build complex agents from simple building blocks.
Prerequisites and Setup
Before diving in, install the required packages and set up your environment. The examples below assume Python 3.10+.
pip install langgraph langchain-openai langchain-core
Set your OpenAI API key as an environment variable:
export OPENAI_API_KEY="your-key-here"
Pattern 1: The Basic Tool Calling Loop
The foundational pattern is a two-node graph: one node calls the LLM, and another node executes whatever tools the LLM requests. A conditional edge routes back to the LLM if tools were called, or to an end node if the LLM produced a final answer.
from typing import Annotated, TypedDict
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
# 1. Define tools
@tool
def add(a: int, b: int) -> int:
"""Add two numbers together."""
return a + b
@tool
def multiply(a: int, b: int) -> int:
"""Multiply two numbers."""
return a * b
tools = [add, multiply]
# 2. Bind tools to the model
llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools(tools)
# 3. Define state
class State(TypedDict):
messages: Annotated[list, add_messages]
# 4. Define the assistant node
def assistant(state: State) -> dict:
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
# 5. Build the graph
builder = StateGraph(State)
builder.add_node("assistant", assistant)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "assistant")
def should_use_tools(state: State) -> str:
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tools"
return END
builder.add_conditional_edges("assistant", should_use_tools)
builder.add_edge("tools", "assistant")
graph = builder.compile()
# 6. Run it
result = graph.invoke({
"messages": [{"role": "user", "content": "What is 3 + 5 multiplied by 2?"}]
})
print(result["messages"][-1].content)
This is the canonical ReAct-style loop, but expressed as a graph. The ToolNode helper from langgraph.prebuilt handles executing tool calls and appending ToolMessage results to state.
Pattern 2: Parallel Tool Execution
Modern models like GPT-4o can emit multiple tool calls in a single response. LangGraph's ToolNode executes these in parallel by default, which dramatically reduces latency when tools are independent.
import time
from langchain_core.tools import tool
@tool
def fetch_weather(city: str) -> str:
"""Get the current weather for a city."""
time.sleep(1) # simulate network latency
return f"Sunny, 72F in {city}"
@tool
def fetch_stock(ticker: str) -> str:
"""Get the current price of a stock."""
time.sleep(1)
return f"{ticker}: $150.25"
@tool
def fetch_news(topic: str) -> str:
"""Get the latest news headline for a topic."""
time.sleep(1)
return f"Top {topic} story: Markets rally on tech earnings"
tools = [fetch_weather, fetch_stock, fetch_news]
llm = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)
class State(TypedDict):
messages: Annotated[list, add_messages]
def assistant(state: State) -> dict:
return {"messages": [llm.invoke(state["messages"])]}
builder = StateGraph(State)
builder.add_node("assistant", assistant)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "assistant")
def route(state: State) -> str:
return "tools" if state["messages"][-1].tool_calls else END
builder.add_conditional_edges("assistant", route)
builder.add_edge("tools", "assistant")
graph = builder.compile()
start = time.time()
result = graph.invoke({
"messages": [{
"role": "user",
"content": "Give me the weather in Tokyo, the price of AAPL, and the latest AI news."
}]
})
print(f"Elapsed: {time.time() - start:.2f}s")
print(result["messages"][-1].content)
Even though each tool sleeps for one second, the total wall-clock time for the tool execution step is roughly one second rather than three, because ToolNode dispatches all three calls concurrently.
Pattern 3: Human-in-the-Loop Approval
Some tools have real-world side effects — sending emails, executing trades, modifying databases. For these, you want a human to approve before execution. LangGraph supports this with an interrupt mechanism that pauses the graph and resumes after human input.
from typing import Annotated, TypedDict
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaver
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""Send an email to a recipient."""
# In production, integrate with an email provider
return f"Email sent to {to} with subject '{subject}'"
@tool
def search_contacts(name: str) -> str:
"""Search for a contact by name."""
return f"Found: {name} <{name.lower().replace(' ', '.')}@example.com>"
tools = [send_email, search_contacts]
llm = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)
class State(TypedDict):
messages: Annotated[list, add_messages]
def assistant(state: State) -> dict:
return {"messages": [llm.invoke(state["messages"])]}
builder = StateGraph(State)
builder.add_node("assistant", assistant)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "assistant")
builder.add_conditional_edges("assistant", tools_condition)
builder.add_edge("tools", "assistant")
# Compile with an interrupt before the "tools" node executes
graph = builder.compile(
checkpointer=MemorySaver(),
interrupt_before=["tools"]
)
config = {"configurable": {"thread_id": "thread-1"}}
# First invocation — pauses before tool execution
result = graph.invoke(
{"messages": [{"role": "user", "content": "Send an email to Jane Doe saying hello."}]},
config=config
)
# Inspect the pending tool call
last_msg = result["messages"][-1]
print("Pending tool calls:", [tc["name"] for tc in last_msg.tool_calls])
# Human reviews and approves — resume execution
result = graph.invoke(None, config=config)
print(result["messages"][-1].content)
The key is interrupt_before=["tools"]. The graph halts right before the tools node, giving your application a chance to surface the pending tool call to a human. Calling graph.invoke(None, config) resumes from the checkpoint.
Pattern 4: Dynamic Tool Selection
In complex agents, you may have dozens of tools but only want to expose a relevant subset at each step. You can dynamically rebind tools based on the current state, reducing token usage and improving accuracy.
from typing import Annotated, TypedDict
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
@tool
def query_orders(user_id: str) -> str:
"""Query a user's order history."""
return f"Orders for {user_id}: [#1001, #1002]"
@tool
def query_billing(user_id: str) -> str:
"""Query a user's billing info."""
return f"Billing for {user_id}: balance $0.00"
@tool
def reset_password(user_id: str) -> str:
"""Reset a user's password."""
return f"Password reset link sent to {user_id}"
@tool
def escalate_to_human(reason: str) -> str:
"""Escalate the conversation to a human agent."""
return f"Escalated: {reason}"
all_tools = [query_orders, query_billing, reset_password, escalate_to_human]
llm = ChatOpenAI(model="gpt-4o-mini")
class State(TypedDict):
messages: Annotated[list, add_messages]
available_tools: list
def select_tools(state: State) -> dict:
"""Dynamically choose which tools to expose based on conversation context."""
last_user_msg = ""
for m in reversed(state["messages"]):
if m.type == "human":
last_user_msg = m.content.lower()
break
if "password" in last_user_msg:
chosen = [reset_password, escalate_to_human]
elif "bill" in last_user_msg or "payment" in last_user_msg:
chosen = [query_billing, escalate_to_human]
else:
chosen = [query_orders, query_billing, escalate_to_human]
return {"available_tools": chosen}
def assistant(state: State) -> dict:
tools = state["available_tools"]
bound_llm = llm.bind_tools(tools)
response = bound_llm.invoke(state["messages"])
return {"messages": [response]}
def execute_tools(state: State) -> dict:
last_msg = state["messages"][-1]
tool_map = {t.name: t for t in all_tools}
results = []
for tc in last_msg.tool_calls:
tool_fn = tool_map[tc["name"]]
results.append(tool_fn.invoke(tc["args"]))
return {"messages": [{"role": "tool", "content": str(r), "tool_call_id": tc["id"]}
for tc, r in zip(last_msg.tool_calls, results)]}
builder = StateGraph(State)
builder.add_node("select_tools", select_tools)
builder.add_node("assistant", assistant)
builder.add_node("tools", execute_tools)
builder.add_edge(START, "select_tools")
builder.add_edge("select_tools", "assistant")
def route(state: State) -> str:
return "tools" if state["messages"][-1].tool_calls else END
builder.add_conditional_edges("assistant", route)
builder.add_edge("tools", "assistant")
graph = builder.compile()
result = graph.invoke({
"messages": [{"role": "user", "content": "I forgot my password, can you help?"}],
"available_tools": []
})
print(result["messages"][-1].content)
By inserting a select_tools node before the assistant, you ensure the model only sees tools relevant to the current intent. This pattern is especially useful in customer support, where tool catalogs can be large.
Pattern 5: Error Handling and Retries
Tools fail. APIs time out, arguments are malformed, rate limits get hit. A robust agent must handle these gracefully. LangGraph lets you catch tool errors, feed them back to the model, and let the model decide whether to retry with corrected arguments.
from typing import Annotated, TypedDict
from langchain_core.tools import tool, ToolException
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
@tool
def get_user_profile(user_id: int) -> dict:
"""Fetch a user profile by numeric ID."""
if not isinstance(user_id, int) or user_id < 0:
raise ToolException(f"Invalid user_id: {user_id}. Must be a positive integer.")
# Simulate a lookup
if user_id == 42:
return {"id": 42, "name": "Alice", "email": "alice@example.com"}
raise ToolException(f"User {user_id} not found.")
tools = [get_user_profile]
llm = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)
class State(TypedDict):
messages: Annotated[list, add_messages]
def assistant(state: State) -> dict:
return {"messages": [llm.invoke(state["messages"])]}
# ToolNode with handle_tool_errors=True converts exceptions into ToolMessages
tool_node = ToolNode(tools, handle_tool_errors=True)
builder = StateGraph(State)
builder.add_node("assistant", assistant)
builder.add_node("tools", tool_node)
builder.add_edge(START, "assistant")
def route(state: State) -> str:
return "tools" if state["messages"][-1].tool_calls else END
builder.add_conditional_edges("assistant", route)
builder.add_edge("tools", "assistant")
graph = builder.compile()
# The model initially passes a string; the error feedback lets it self-correct
result = graph.invoke({
"messages": [{"role": "user", "content": "Get the profile for user forty-two."}]
})
print(result["messages"][-1].content)
With handle_tool_errors=True, the exception message is returned to the model as a ToolMessage. The model reads the error, realizes it needs a numeric ID, and retries with 42. This self-correcting loop is one of the most valuable patterns in production agents.
Pattern 6: Stateful Tools with Shared Memory
Sometimes tools need to read or write to a shared state that persists across the conversation — a shopping cart, a scratchpad, or accumulated research notes. You can store this in the graph state and have tools access it through a closure or dependency injection.
from typing import Annotated, TypedDict
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages]
cart: dict # product_name -> quantity
def make_cart_tools(cart_ref: dict):
@tool
def add_to_cart(product: str, quantity: int = 1) -> str:
"""Add a product to the shopping cart."""
cart_ref[product] = cart_ref.get(product, 0) + quantity
return f"Added {quantity} x {product} to cart. Cart: {cart_ref}"
@tool
def view_cart() -> str:
"""View the current contents of the cart."""
if not cart_ref:
return "Your cart is empty."
items = [f"{q} x {p}" for p, q in cart_ref.items()]
return "Cart contents: " + ", ".join(items)
@tool
def checkout() -> str:
"""Checkout and place the order."""
if not cart_ref:
return "Cannot checkout: cart is empty."
total_items = sum(cart_ref.values())
cart_ref.clear()
return f"Order placed for {total_items} item(s). Cart is now empty."
return [add_to_cart, view_cart, checkout]
# Shared mutable cart
cart = {}
tools = make_cart_tools(cart)
llm = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)
def assistant(state: State) -> dict:
return {"messages": [llm.invoke(state["messages"])]}
def execute_tools(state: State) -> dict:
last_msg = state["messages"][-1]
tool_map = {t.name: t for t in tools}
results = []
for tc in last_msg.tool_calls:
results.append({
"role": "tool",
"content": tool_map[tc["name"]].invoke(tc["args"]),
"tool_call_id": tc["id"]
})
return {"messages": results, "cart": cart}
builder = StateGraph(State)
builder.add_node("assistant", assistant)
builder.add_node("tools", execute_tools)
builder.add_edge(START, "assistant")
def route(state: State) -> str:
return "tools" if state["messages"][-1].tool_calls else END
builder.add_conditional_edges("assistant", route)
builder.add_edge("tools", "assistant")
graph = builder.compile()
result = graph.invoke({
"messages": [{"role": "user", "content": "Add 2 laptops to my cart, then add a mouse, then checkout."}],
"cart": {}
})
print(result["messages"][-1].content)
print("Final cart state:", result["cart"])
By closing over the cart dictionary, the tools share mutable state across invocations. The graph state also tracks cart, so you can persist it in a checkpointer for long-running sessions.
Pattern 7: Multi-Agent Tool Orchestration
For complex systems, you can compose multiple agents — each with its own toolset — into a single graph. A supervisor agent routes requests to specialist sub-agents, each of which has its own tool-calling loop.
from typing import Annotated, TypedDict, Literal
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
llm = ChatOpenAI(model="gpt-4o-mini")
# --- Research Agent Tools ---
@tool
def web_search(query: str) -> str:
"""Search the web for information."""
return f"Search results for '{query}': AI adoption grew 40% in 2024."
# --- Code Agent Tools ---
@tool
def run_python(code: str) -> str:
"""Execute Python code and return the output."""
return f"Executed. Output: 42"
research_tools = [web_search]
code_tools = [run_python]
class State(TypedDict):
messages: Annotated[list, add_messages]
next: str
def supervisor(state: State) -> dict:
"""Decide which agent should handle the request."""
last_msg = state["messages"][-1].content.lower()
if "code" in last_msg or "python" in last_msg or "calculate" in last_msg:
next_agent = "code_agent"
elif "search" in last_msg or "research" in last_msg or "find" in last_msg:
next_agent = "research_agent"
else:
next_agent = END
return {"next": next_agent}
def research_agent(state: State) -> dict:
bound = llm.bind_tools(research_tools)
response = bound.invoke(state["messages"])
# Execute any tool calls inline for simplicity
if response.tool_calls:
for tc in response.tool_calls:
result = {t.name: t for t in research_tools}[tc["name"]].invoke(tc["args"])
response = llm.invoke(state["messages"] + [response, {
"role": "tool", "content": result, "tool_call_id": tc["id"]
}])
return {"messages": [response]}
def code_agent(state: State) -> dict:
bound = llm.bind_tools(code_tools)
response = bound.invoke(state["messages"])
if response.tool_calls:
for tc in response.tool_calls:
result = {t.name: t for t in code_tools}[tc["name"]].invoke(tc["args"])
response = llm.invoke(state["messages"] + [response, {
"role": "tool", "content": result, "tool_call_id": tc["id"]
}])
return {"messages": [response]}
builder = StateGraph(State)
builder.add_node("supervisor", supervisor)
builder.add_node("research_agent", research_agent)
builder.add_node("code_agent", code_agent)
builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", lambda s: s["next"])
builder.add_edge("research_agent", END)
builder.add_edge("code_agent", END)
graph = builder.compile()
result = graph.invoke({
"messages": [{"role": "user", "content": "Search for the latest AI adoption statistics."}]
})
print(result["messages"][-1].content)
This supervisor pattern scales well: add new specialist agents by adding nodes and updating the supervisor's routing logic. Each sub-agent can itself be a full tool-calling graph.
Best Practices
Write Excellent Tool Descriptions
The tool's docstring is the model's only guide for when and how to use it. Be explicit about arguments, expected formats, and edge cases. A vague description like "Gets data" will lead to poor tool selection.
Keep Tools Focused and Composable
Prefer many small, single-purpose tools over one large tool with complex arguments. Small tools are easier for the model to call correctly and easier for you to test.
Always Use Structured Return Types
Return strings, dicts, or Pydantic models from tools — never raw objects. The return value becomes a ToolMessage that the model must parse. Structured, predictable output improves reasoning quality.
Set a Recursion Limit
Tool-calling loops can run indefinitely if the model keeps calling tools without converging. Always set a recursion limit when compiling your graph:
graph = builder.compile()
result = graph.invoke(
{"messages": [{"role": "user", "content": "..."}]},
config={"recursion_limit": 25}
)
Log and Trace Every Tool Call
Use LangSmith or a custom callback handler to log every tool invocation, its arguments, and its result. This is essential for debugging unexpected agent behavior in production.
Validate Tool Arguments
Even though the model produces structured arguments, validate them before execution. Use Pydantic models or explicit checks to catch malformed inputs before they reach external systems.
Handle Rate Limits and Timeouts
Wrap external API calls in retry logic with exponential backoff. Consider using a library like tenacity for robust retry policies on individual tools.
Conclusion
Tool use is the bridge between the reasoning power of language models and the concrete capabilities of your software systems. LangGraph's graph-based approach elevates tool use from a simple request-response loop into a rich, controllable, and observable workflow. By mastering the patterns in this guide — the basic loop, parallel execution, human-in-the-loop approval, dynamic tool selection, error recovery, stateful tools, and multi-agent orchestration — you can build agents that are not only capable but also reliable, safe, and production-ready. Start with the basic loop, add complexity incrementally, and always test your tool descriptions and error paths as rigorously as you would test any other critical code path.