LangGraph - 2. Practice by Type
intro
Agent Executor
? Chat Executor
Agent Supervisor
Hierarchical Agent Teams
Multi-agent Collaboration
[Terminal] Environment Variables and GitHub Setup
Create an environment variable file and a git push exclusion file in the root folder.
Write .env
Write .gitignore
[Terminal] Setting Up and Running a Virtual Environment
(Optional) Set up and run a virtual environment ( LangGraph_Agents is an arbitrary name I personally chose)
Setup: python -m venv LangGraph_Agents
Run (mac): source LangGraph_Agents/bin/activate
Before we start, let's look at the overall structure
The structure of the Chat Executor graph is identical to the Agent Executor graph.
Loading Environment Variables
from dotenv import load_dotenv
load_dotenv()
State
Creating a state-management store: State, the feature that remembers (records the state of) the work each node has performed
Unlike the Agent Executor example, we don't need input, chat_history, agent_outcome, or intermediate_steps. Because this is a chat form, metadata (operator.add) and the chat history are managed directly inside chat (messages) rather than in intermediate_steps, chat_history, and so on.
from typing import TypedDict, Annotated, Sequence
import operator
from langchain_core.messages import BaseMessage
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
Model
Unlike the Agent Executor example, we configure a Model instead of Agents. To do this, instead of 'the createOpenAI function agent', we import the ChatOpenAI class for OpenAI chat models from langchain_openai.
from langchain_openai import ChatOpenAI
model = ChatOpenAI(temperature=0, streaming=True)
Tools
Creating custom tools : to be node
As explained in the previous example, tools are interfaces that allow an agent to interact with the outside world. The difference from the Agent Executor example is that since we already declared the Model as a Class rather than a function, we only need to bundle the defined tools together as a list.
from langchain.tools import BaseTool, StructuredTool, Tool, tool
import random
@tool("lower_case", return_direct=True)
def to_lower_case(input:str) -> str:
"""Returns the input as all lower case."""
return input.lower()
@tool("random_number", return_direct=True)
def random_number_maker(input:str) -> str:
"""Returns a random number between 0-100. input the word 'random'"""
return random.randint(0, 100)
tools = [to_lower_case,random_number_maker]
Binding tools to models
This is the process of converting langchain tools into a format compatible with OpenAI chat functions. The chat model can call these functions to use them, so we convert the tools defined above into the OpenAI function format, build a list of those functions, and bind these functions to the chat model.
from langchain.tools.render import format_tool_to_openai_function
functions = [format_tool_to_openai_function(t) for t in tools]
model = model.bind_functions(functions)
Nodes
Creating nodes: node
We define the list of tools we have, the agent, and the logic used in the 'conditional edges'. The differences from the Agent Executor example are that call_model replaces call_agent, and for the call_tool function, whereas previously it acted as a kind of data shortcut by directly fetching 'agent_outcome' from a dictionary called data and using tool_executor to invoke the corresponding tool, here it focuses on tool calls based on a particular state.
Specifically, after retrieving the messages from the state dictionary, it extracts the information needed to call a tool from the last message, creates a ToolInvocation object, executes the corresponding tool via tool_executor, and then creates a FunctionMessage containing the response and returns the result in message form.
## Nodes
from langchain_core.agents import AgentFinish
from langgraph.prebuilt import ToolInvocation
import json
from langchain_core.messages import FunctionMessage
from langgraph.prebuilt.tool_executor import ToolExecutor
tool_executor = ToolExecutor(tools)
def call_model(state):
messages = state['messages']
response = model.invoke(messages)
return {"messages": [response]}
def call_tool(state):
messages = state['messages']
last_message = messages[-1]
action = ToolInvocation(
tool=last_message.additional_kwargs["function_call"]["name"],
tool_input=json.loads(last_message.additional_kwargs["function_call"]["arguments"]),
)
# print(f"The agent action is {action}")
response = tool_executor.invoke(action)
# print(f"The tool result is: {response}")
function_message = FunctionMessage(content=str(response), name=action.tool)
return {"messages": [function_message]}
def should_continue(state):
messages = state['messages']
last_message = messages[-1]
if "function_call" not in last_message.additional_kwargs:
return "end"
else:
return "continue"
Graph
Defining the graph : Graph, a collection of nodes and edges
Compared to the Agent Executor example, the only difference is that call_model is used in place of run_agent. It is the process of building an application that runs the workflow back and forth between the "agent" and "action" nodes, calling the call_model and call_tool functions; starting from "agent", it moves to "action" or terminates depending on a condition, and from "action" it can return to "agent" again, forming a cyclic structure.
from langgraph.graph import StateGraph, END
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("action", call_tool)
workflow.set_entry_point("agent")
workflow.add_conditional_edges(
"agent",
should_continue,
{
"continue": "action",
"end": END
}
)
workflow.add_edge('action', 'agent')
app = workflow.compile()
Run It with LangSmith
type 1: When using both tools
from langchain_core.messages import HumanMessage, SystemMessage
system_message = SystemMessage(content="you are a helpful assistant")
user_01 = HumanMessage(content="give me a random number and then write in words and make it lower case")
inputs = {"messages": [system_message,user_01]}
app.invoke(inputs)
type 2: When using only one tool
from langchain_core.messages import HumanMessage, SystemMessage
system_message = SystemMessage(content="you are a helpful assistant")
user_01 = HumanMessage(content="plear write 'Merlion' in lower case")
inputs = {"messages": [system_message,user_01]}
app.invoke(inputs)
type 3: When tool use is not needed
from langchain_core.messages import HumanMessage, SystemMessage
system_message = SystemMessage(content="you are a helpful assistant")
user_01 = HumanMessage(content="what is a Merlion?")
inputs = {"messages": [system_message,user_01]}
app.invoke(inputs)
