In this post, you are going to learn how we can create this chatbot in LangGraph, Gemini Pro or any model you like, Custom Function and Streamlit that can respond to user’s support requests
LangGraph is a library for building stateful “ multi-actor applications “ with LLM. LangChain Expression Language can be extended to allow multiple chains (or actors) to work together cyclically over multiple steps.
One of the great values of “LangChain” is that you can easily create custom chains.
We have provided LangChain Expression Language as a function for this purpose, but there was no easy way to introduce the cycle. LangGraph makes it easy to introduce cycles into LLM applications.
LangChain Expression Language is not suitable for describing cycles (loops), but by using LangGraph, it becomes possible to describe and introduce the cycles necessary for the agent.
Before we start! 🦸🏻♀️
If you like this topic and you want to support me:
- Share my article; that will really help me out.👏
- Follow me on Medium and subscribe to get my latest article🫶
- Follow me on YouTube Gao Dalie (高達烈)
Let’s Start Coding
Before we dive into our application, we will create an ideal environment for the code to work. For this, we need to install the necessary Python libraries needed. Firstly, we will start by installing the libraries that support the model. For this, we will do a pip install of the below libraries.
pip install streamlit
pip install langchainhub
pip install langgraph
pip install langchain_google_genai
pip install -U langchain-openai langchain
Once installed we import Langchain, Langchain Google, langchain community, os, typing, langchain core, operator, langchian prebuilt, langGraph and streamlit
from langchain import hub
from langchain.agents import Tool, create_react_agent
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_community.utilities import GoogleSerperAPIWrapper
import os
from typing import TypedDict, Annotated, Union
from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.messages import BaseMessage
import operator
from typing import TypedDict, Annotated
from langchain_core.agents import AgentFinish
from langgraph.prebuilt.tool_executor import ToolExecutor
from langgraph.prebuilt import ToolInvocation
from langgraph.graph import END, StateGraph
from langchain_core.agents import AgentActionMessageLog
import streamlit as st
we will set up the page title and layout style for a streamlit web application
st.set_page_config(page_title="LangChain Agent", layout="wide")
We start by defining a function named main
, which will serve as the central feature of our Streamlit application. Next, we set up a title for the web page, followed by the creation of a text area. This text area is designed for user input. Additionally, we incorporate a button on the page. When a user clicks this button, the code within the corresponding if
the block gets executed.
def main():
# Streamlit UI elements
st.title("LangGraph Agent + Gemini Pro + Custom Tool + Streamlit")
# Input from user
input_text = st.text_area("Enter your text:")
if st.button("Run Agent"):
we set a variable named SERPER_API_KEY
with the value "YOUR_KEY_API"
Next, we use the GoogleSerperAPIWrapper
module to retrieve relevant news search results using Serper API,
os.environ["SERPER_API_KEY"] = "YOUR-KEY-API"
search = GoogleSerperAPIWrapper()
Let’s add a `Sort Tool` to sort a list of words alphabetically or a `Toggle Case Tool` to convert letters to lowercase or uppercase.
Next, We are going to create functions:
First: we define the Sort function which sorts the words alphabetically
Second: we define the case toggle function which converts the letter to lowercase or uppercase
def toggle_case(word):
toggled_word = ""
for char in word:
if char.islower():
toggled_word += char.upper()
elif char.isupper():
toggled_word += char.lower()
else:
toggled_word += char
return toggled_word
def sort_string(string):
return ''.join(sorted(string))
So far we have defined all our custom tools and now it’s time to add them to our agent
In our tools array, we have defined the list of tools. The search, Sort and Convert letter tools have only one parameter so you can define them directly. We have used the name, func and description attributes where we have defined the name of the tool, the function to be called by the Agent when it wants to use it, and then the description, which the Agent reads to make a judgment on when to use this tool.
tools = [
Tool(
name = "Search",
func=search.run,
description="useful for when you need to answer questions about current events",
),
Tool(
name = "Toogle_Case",
func = lambda word: toggle_case(word),
description = "use when you want covert the letter to uppercase or lowercase",
),
Tool(
name = "Sort String",
func = lambda string: sort_string(string),
description = "use when you want sort a string alphabetically",
),
]
- The ChatGoogleGenerativeAI is the class that is worked with to get the Gemini LLM working, Then we create the LLM class by passing the Gemini Model that we want to work with to the ChatGoogleGeneraativeAI class. Next, we use the ReAct approach to create agents, which are AI systems that can interact with tools and tools.
prompt = hub.pull("hwchase17/react")
llm = ChatGoogleGenerativeAI(model="gemini-pro",
google_api_key="Your_API_KEY",
convert_system_message_to_human = True,
verbose = True,
)
agent_runnable = create_react_agent(llm, tools, prompt)
First, define a class to maintain the state within the Agent.
The internal state during Agent processing is maintained in an instance of this class.
This class will be used inside LangGraph later.
The information held as the status is as follows.
input
: Input content from the userchat_history
: Conversation history before agent executionintermediate_steps
: Intermediate execution details and results during Agent processingagent_outcome
: Agent response result, AgentAction or AgentFinish instance is stored. If the response result is AgentFinish, the process should be terminated; otherwise, the handling will be to run the tool.
class AgentState(TypedDict):
input: str
chat_history: list[BaseMessage]
agent_outcome: Union[AgentAction, AgentFinish, None]
return_direct: bool
intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]
Define the process to be executed by the Agent.
These functions correspond to nodes in the Graph.
The functions for the nodes defined here are as follows.
run_agent
: Process of executing the agent chain defined in Step 4
tool_executor = ToolExecutor(tools)
def run_agent(state):
"""
#if you want to better manages intermediate steps
inputs = state.copy()
if len(inputs['intermediate_steps']) > 5:
inputs['intermediate_steps'] = inputs['intermediate_steps'][-5:]
"""
agent_outcome = agent_runnable.invoke(state)
return {"agent_outcome": agent_outcome}
We define function execute_tools which takes a parameter state and Retrieving Tool Name and Arguments we set if statement checks if tool_name
is “Search”, “Sort”, or “Toggle_Case”, If return_direct
is a key in arguments
, it is deleted.
Then we prepare tool invocation for invoking a tool with specified inputs. and return statement.
def execute_tools(state):
messages = [state['agent_outcome'] ]
last_message = messages[-1]
######### human in the loop ###########
# human input y/n
# Get the most recent agent_outcome - this is the key added in the `agent` above
# state_action = state['agent_outcome']
# human_key = input(f"[y/n] continue with: {state_action}?")
# if human_key == "n":
# raise ValueError
tool_name = last_message.tool
arguments = last_message
if tool_name == "Search" or tool_name == "Sort" or tool_name == "Toggle_Case":
if "return_direct" in arguments:
del arguments["return_direct"]
action = ToolInvocation(
tool=tool_name,
tool_input= last_message.tool_input,
)
response = tool_executor.invoke(action)
return {"intermediate_steps": [(state['agent_outcome'],response)]}
should_continue
The function checks the contents of the previous response result and uses it to determine whether to terminate or continue Agent processing.
Else See: What is Langchain 0.1.0? Explained Simply
def should_continue(state):
messages = [state['agent_outcome'] ]
last_message = messages[-1]
if "Action" not in last_message.log:
return "end"
else:
arguments = state["return_direct"]
if arguments is True:
return "final"
else:
return "continue"
Normally, this is all you need, but since the model we’re using this time doesn’t seem to use the Tool much, we’ll first first_agent
define an additional node function to make sure it uses the Tool.
def first_agent(inputs):
action = AgentActionMessageLog(
tool="Search",
tool_input=inputs["input"],
log="",
message_log=[]
)
return {"agent_outcome": action}
Define a graph of Agent processing. Graphs are roughly composed of nodes and edges. You can get a good idea of this by looking at Wikipedia’s graph theory.
workflow = StateGraph(AgentState)
workflow.add_node("agent", run_agent)
workflow.add_node("action", execute_tools)
workflow.add_node("final", execute_tools)
# uncomment if you want to always calls a certain tool first
# workflow.add_node("first_agent", first_agent)
workflow.set_entry_point("agent")
# uncomment if you want to always calls a certain tool first
# workflow.set_entry_point("first_agent")
workflow.add_conditional_edges(
"agent",
should_continue,
{
"continue": "action",
"final": "final",
"end": END
}
)
workflow.add_edge('action', 'agent')
workflow.add_edge('final', END)
# uncomment if you want to always calls a certain tool first
# workflow.add_edge('first_agent', 'action')
app = workflow.compile()
The Agent is now ready to run. Execute the graph for the Agent defined so far. After compilation, it is a LangChain Runnable object, so it can be used in the same way as LangChain Expression Language or LCEL Chain.
So let’s run it.
inputs = {"input": input_text, "chat_history": [], "return_direct": False}
results = []
for s in app.stream(inputs):
result = list(s.values())[0]
results.append(result)
st.write(result) # Display each step's output
The results of Agent processing (each node in the graph) will be output in order. The contents of the last line agent_outcome
will be the final result obtained. Especially if you don’t need the execution log, invoke
you can only get the last result. Let’s run another query and print only the final results.
result = app.invoke({"input": input_text, "chat_history": [], "return_direct": False})
print(result["agent_outcome"].return_values["output"])
Please If you want to check the result of the code feel free to check my video
Conclusion :
It was an Agent implementation using LangGraph.In this execution example, the process does not cycle (proceeds in one direction and ends as it is), so there is little point in using LangGraph. Still, the LLM response → tool execution → LLM response using the result → I think it will be a very powerful framework when building an agent to run another tool.
Else See LangGraph: Create Your Hyper AI Agent