How to create a ReAct agent from scratch¶
Prerequisites
This guide assumes familiarity with the following:
Using the prebuilt ReAct agent (create_react_agent) is a great way to get started, but sometimes you might want more control and customization. In those cases, you can create a custom ReAct agent. This guide shows how to implement ReAct agent from scratch using LangGraph.
Setup¶
First, let's install the required packages and set our API keys:
import getpass
import os
def _set_env(var: str):
if not os.environ.get(var):
os.environ[var] = getpass.getpass(f"{var}: ")
_set_env("OPENAI_API_KEY")
Set up LangSmith for better debugging
Sign up for LangSmith to quickly spot issues and improve the performance of your LangGraph projects. LangSmith lets you use trace data to debug, test, and monitor your LLM aps built with LangGraph — read more about how to get started in the docs.
Create ReAct agent¶
Now that you have installed the required packages and set your environment variables, we can code our ReAct agent!
Define graph state¶
We are going to define the most basic ReAct state in this example, which will just contain a list of messages.
For your specific use case, feel free to add any other state keys that you need.
from typing import (
Annotated,
Sequence,
TypedDict,
)
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
"""The state of the agent."""
# add_messages is a reducer
# See https://langchain-ai.github.io/langgraph/concepts/low_level/#reducers
messages: Annotated[Sequence[BaseMessage], add_messages]
Define model and tools¶
Next, let's define the tools and model we will use for our example.
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
model = ChatOpenAI(model="gpt-4o-mini")
@tool
def get_weather(location: str):
"""Call to get the weather from a specific location."""
# This is a placeholder for the actual implementation
# Don't let the LLM know this though 😊
if any([city in location.lower() for city in ["sf", "san francisco"]]):
return "It's sunny in San Francisco, but you better look out if you're a Gemini 😈."
else:
return f"I am not sure what the weather is in {location}"
tools = [get_weather]
model = model.bind_tools(tools)
Define nodes and edges¶
Next let's define our nodes and edges. In our basic ReAct agent there are only two nodes, one for calling the model and one for using tools, however you can modify this basic structure to work better for your use case. The tool node we define here is a simplified version of the prebuilt ToolNode, which has some additional features.
Perhaps you want to add a node for adding structured output or a node for executing some external action (sending an email, adding a calendar event, etc.). Maybe you just want to change the way the call_model node works and how should_continue decides whether to call tools - the possibilities are endless and LangGraph makes it easy to customize this basic structure for your specific use case.
import json
from langchain_core.messages import ToolMessage, SystemMessage
from langchain_core.runnables import RunnableConfig
tools_by_name = {tool.name: tool for tool in tools}
# Define our tool node
def tool_node(state: AgentState):
outputs = []
for tool_call in state["messages"][-1].tool_calls:
tool_result = tools_by_name[tool_call["name"]].invoke(tool_call["args"])
outputs.append(
ToolMessage(
content=json.dumps(tool_result),
name=tool_call["name"],
tool_call_id=tool_call["id"],
)
)
return {"messages": outputs}
# Define the node that calls the model
def call_model(
state: AgentState,
config: RunnableConfig,
):
# this is similar to customizing the create_react_agent with state_modifier, but is a lot more flexible
system_prompt = SystemMessage(
"You are a helpful AI assistant, please respond to the users query to the best of your ability!"
)
response = model.invoke([system_prompt] + state["messages"], config)
# We return a list, because this will get added to the existing list
return {"messages": [response]}
# Define the conditional edge that determines whether to continue or not
def should_continue(state: AgentState):
messages = state["messages"]
last_message = messages[-1]
# If there is no function call, then we finish
if not last_message.tool_calls:
return "end"
# Otherwise if there is, we continue
else:
return "continue"
Define the graph¶
Now that we have defined all of our nodes and edges, we can define and compile our graph. Depending on if you have added more nodes or different edges, you will need to edit this to fit your specific use case.
from langgraph.graph import StateGraph, END
# Define a new graph
workflow = StateGraph(AgentState)
# Define the two nodes we will cycle between
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)
# Set the entrypoint as `agent`
# This means that this node is the first one called
workflow.set_entry_point("agent")
# We now add a conditional edge
workflow.add_conditional_edges(
# First, we define the start node. We use `agent`.
# This means these are the edges taken after the `agent` node is called.
"agent",
# Next, we pass in the function that will determine which node is called next.
should_continue,
# Finally we pass in a mapping.
# The keys are strings, and the values are other nodes.
# END is a special node marking that the graph should finish.
# What will happen is we will call `should_continue`, and then the output of that
# will be matched against the keys in this mapping.
# Based on which one it matches, that node will then be called.
{
# If `tools`, then we call the tool node.
"continue": "tools",
# Otherwise we finish.
"end": END,
},
)
# We now add a normal edge from `tools` to `agent`.
# This means that after `tools` is called, `agent` node is called next.
workflow.add_edge("tools", "agent")
# Now we can compile and visualize our graph
graph = workflow.compile()
from IPython.display import Image, display
try:
display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
# This requires some extra dependencies and is optional
pass
Use ReAct agent¶
Now that we have created our react agent, let's actually put it to the test!
# Helper function for formatting the stream nicely
def print_stream(stream):
for s in stream:
message = s["messages"][-1]
if isinstance(message, tuple):
print(message)
else:
message.pretty_print()
inputs = {"messages": [("user", "what is the weather in sf")]}
print_stream(graph.stream(inputs, stream_mode="values"))
================================[1m Human Message [0m=================================
what is the weather in sf
==================================[1m Ai Message [0m==================================
Tool Calls:
get_weather (call_azW0cQ4XjWWj0IAkWAxq9nLB)
Call ID: call_azW0cQ4XjWWj0IAkWAxq9nLB
Args:
location: San Francisco
=================================[1m Tool Message [0m=================================
Name: get_weather
"It's sunny in San Francisco, but you better look out if you're a Gemini \ud83d\ude08."
==================================[1m Ai Message [0m==================================
The weather in San Francisco is sunny! However, it seems there's a playful warning for Geminis. Enjoy the sunshine!
get_weather tool and responds to the user after receiving the information from the tool.