LangGraph - Hands-on practice by agent type
intro
? Agent Executor
Chat Executor
Agent Supervisor
Hierarchical Agent Teams
Multi-agent Collaboration
[Terminal] Environment variables and GitHub setup
Create the env file and the git-push exclusion file in the root folder.
Write .env
Write .gitignore
[Terminal] Set up and run the virtual environment
(Optional) Set up and run a virtual environment ( LangGraph_Agents is just a personally chosen arbitrary name)
Setup : python -m venv LangGraph_Agents
Run (mac) : source LangGraph_Agents/bin/activate
Before we start, let's look at the overall structure
Agent Executor, literally an agent executor (or runner). The diagram below is the generated graph rendered through IPython.display.
A start point and an end point are declared, and the agent and action are arranged in a loop. From the diagram you can also see that, most importantly, 'conditional edges'are configured on the agent. As a side note, the agent itself can be composed in many different ways.
Load environment variables
from dotenv import load_dotenv
load_dotenv()
State
Create the state-management store : State, the feature that remembers (records the state of) the work performed by each node
from typing import TypedDict, Annotated, List, Union
from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.messages import BaseMessage
import operator
# GraphState - 각 노드가 수행한 작업들을 기억(상태 기록)하는 기능
class AgentState(TypedDict):
input: str
chat_history: list[BaseMessage] # 대화 내용 중 '이전 메시지' 목록
agent_outcome: Union[AgentAction, AgentFinish, None] # 유효한 유형으로 `None`이 필요
intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]
Tools
Create custom tools : to be nodes
Tools are interfaces that let the agent interact with the outside world. Therefore, so the agent (LLM) can understand them, write the tool's name, a description of the tool, how it runs, and what input it accepts and what output it produces.
## Custom Tools
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."""
return random.randint(0, 100)
tools = [to_lower_case,random_number_maker]
random_value = random_number_maker.run('random')
SAM_value = to_lower_case.run('SAM')
Agents
Configure a new agent: tobenode
Specify which LLM model the agent will use, which tools it can use, and under what conditions (prompt) it should respond. Also configure how it will handle the values it receives — how to process inputs, set up storage for chat history, set up storage for intermediate steps, and so on.
## Agent - with new create_open_ai
from langchain import hub
from langchain.agents import create_openai_functions_agent
from langchain_openai.chat_models import ChatOpenAI
prompt = hub.pull("hwchase17/openai-functions-agent")
llm = ChatOpenAI(model="gpt-3.5-turbo-1106", streaming=True)
agent_runnable = create_openai_functions_agent(llm,tools,prompt)
inputs = {"input": "give me a random number and then write in words and make it lower case.",
"chat_history": [],
"intermediate_steps":[]}
agent_outcome = agent_runnable.invoke(inputs)
Nodes
Create nodes : node
Define the list of tools you have, the agent, and the logic to be used for the 'conditional edges'.
## Nodes
from langchain_core.agents import AgentFinish
from langgraph.prebuilt.tool_executor import ToolExecutor
tool_executor = ToolExecutor(tools)
def run_agent(data):
agent_outcome = agent_runnable.invoke(data)
return {"agent_outcome": agent_outcome}
def execute_tools(data):
agent_action = data['agent_outcome']
output = tool_executor.invoke(agent_action)
return {"intermediate_steps": [(agent_action, str(output))]}
def should_continue(data):
if isinstance(data['agent_outcome'], AgentFinish):
return "end"
else:
return "continue"
Graph
Define the graph : Graph, a collection of nodes and edges
Based on the AgentState created earlier for state management, configure the nodes and put tools or agents inside them. Then set the entry point and, together with the 'conditional edges', define the end point. Finally, connect them with edges to complete the graph. And really finally, compile it.
## Define the graph
from langgraph.graph import END, StateGraph
workflow = StateGraph(AgentState)
workflow.add_node("agent", run_agent)
workflow.add_node("action", execute_tools)
workflow.set_entry_point("agent")
workflow.add_conditional_edges(
"agent",
should_continue,
{
"continue": "action",
"end": END
}
)
workflow.add_edge('action', 'agent') # really this is the graph.
app = workflow.compile() # to be like an app in here
Run It
Stream
## type 1
inputs = {"input": "give me a random number and then write in words and make it lower case.", "chat_history": []}
for s in app.stream(inputs):
print(list(s.values())[0])
print("----")
Invoke
## type 2
inputs = {"input": "give me a random number and then write in words and make it lower case", "chat_history": []}
output = app.invoke(inputs)
agent_outcome = output.get("agent_outcome").return_values['output']
# intermediate_steps = output.get("intermediate_steps"
Invoke (no tools used)
## type 3
inputs = {"input": "does it get cold in SF in Jan?", "chat_history": []}
output = app.invoke(inputs)
agent_outcome_no_Tools = output.get("agent_outcome").return_values['output']
print("\\n - agent_outcome_no_Tools :" ,agent_outcome_no_Tools)
# intermediate_steps_no_Tools = output.get("intermediate_steps")
