LangGraph - Practice by Agent 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
Loading Environment Variables
from dotenv import load_dotenv
load_dotenv()
Before we start, let's look at the overall structure
What's different from the previous graph is that the connections are no longer node-to-node, but between a supervisor (more specifically, a supervisor chain) and individual nodes. This also hints at how the structure of the Hierarchical Agent Teams example coming up later will change. Adding a rough comment about the graph: at the entry point, a node (lotto_agent) is connected that holds? a 'supervisor chain (group = Prompt + LLM + Parser)' and the entire tools list, and another node (code_agent) that holds? only '[python_repl_tool]' from the tools list assigned to it.
So let's start by defining the llm, which is the basic element of the supervisor.
Model
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4-1106-preview")
Tools
The process of using the PythonREPLTool() command to create a tool that runs code locally, a tool that returns a string in lowercase, and a tool that returns a random number between 0 and 100, then adding them to the tools list.
from typing import Annotated, List, Tuple, Union
from langchain.tools import BaseTool, StructuredTool, Tool
from langchain_experimental.tools import PythonREPLTool
from langchain_core.tools import tool
import random
python_repl_tool = PythonREPLTool()
@tool("lower_case", return_direct=False)
def to_lower_case(input:str) -> str:
"""Returns the input as all lower case."""
return input.lower()
@tool("random_number", return_direct=False)
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,python_repl_tool]
Helper Utils
As the name suggests, these are helper utilities. Nothing is mandatory; you can flesh out related functions to your taste. The example code below bundles together repetitive functionality that comes up while doing the work in this example. Specifically, it declares the most important and most frequently used functions: one for 'creating an agent', and another for 'composing a node for each agent' so that the agent can be used.
1. def create_agent(): declares a function that creates a conversational agent that can be used for natural language understanding and generation tasks
1) Set `ChatOpenAI` as the (fixed? always-on) value for the LLM (Language Model) parameter
2) Compose `prompt` — a ChatPromptTemplate object based on the input messages and a system prompt, including the agent's actions and scratchpad
3) Compose `agent` — use the `create_openai_tools_agent` function to specify the LLM, tools list, and system prompt
4) `AgentExecutor` — initialize an object together with the agent and the provided tools
2. def agent_node(): declares an `agent_node()` function that runs the agent's behavior based on a given state and returns the resulting message.
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
def create_agent(llm: ChatOpenAI, tools: list, system_prompt: str):
prompt = ChatPromptTemplate.from_messages(
[
(
"system",
system_prompt,
),
MessagesPlaceholder(variable_name="messages"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
]
)
agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools)
return executor
def agent_node(state, agent, name):
result = agent.invoke(state)
return {"messages": [HumanMessage(content=result["output"], name=name)]}
Create Agent Supervisor
In the diagram above, the part corresponding to the first rounded box is the top-level supervisor, which is responsible for managing the child nodes. The basic configuration consists of members, system_prompt, promptTemplate, and supervisor_chain.
For reference, this example code designates "Lotto_Manager" and "Coder" as child nodes under members. The system (supervisor) prompt is composed of 1) the point at which the work is complete (option definition) and 2) the structure for selecting the next agent to run (route function definition); the prompt template contains system_prompt and a MessagesPlaceholder; and the chain that determines the order of work is composed of the prompt, a bind function, and a JSON Parser.
from langchain.output_parsers.openai_functions import JsonOutputFunctionsParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
members = ["Lotto_Manager", "Coder"]
system_prompt = (
"You are a supervisor tasked with managing a conversation between the"
" following workers: {members}. Given the following user request,"
" respond with the worker to act next. Each worker will perform a"
" task and respond with their results and status. When finished,"
" respond with FINISH."
)
options = ["FINISH"] + members
function_def = {
"name": "route",
"description": "Select the next role.",
"parameters": {
"title": "routeSchema",
"type": "object",
"properties": {
"next": {
"title": "Next",
"anyOf": [
{"enum": options},
],
}
},
"required": ["next"],
},
}
prompt = ChatPromptTemplate.from_messages(
[
("system", system_prompt),
MessagesPlaceholder(variable_name="messages"),
(
"system",
"Given the conversation above, who should act next?"
" Or should we FINISH? Select one of: {options}",
),
]
).partial(options=str(options), members=", ".join(members))
supervisor_chain = (
prompt
| llm.bind_functions(functions=[function_def], function_call="route")
| JsonOutputFunctionsParser()
)Create the Agnet State and Graph
As in the previous practice (Agent Executor, Chat Executor), AgentState is a dictionary that defines the state structure of the agent.
- The 'messages' field stores a sequence of BaseMessage (previous conversation) instances, and the 'next' field points to the location where the route corresponding to the next state is specified.
Then, right above this, using the Helper Utils declared above, we create agents and nodes for the 'Lotto_Manager' and 'Coder' members (child nodes).
To break it down a bit more in detail;;
lotto:
- agent: create the lotto_agent object by passing the language model (llm), tools, and a role description as parameters
- node: create a partial function called lotto_node that stores the details of the agent node
code:
- Same as lotto, it creates the agent and node functions, but the only difference is that internally it uses
python_repl_tool (a package for executing Python code locally) instead of the tools we declared earlier.
Once this series of steps is finished, we create a workflow as a state graph of type AgentState, and add to this workflow the previously declared 'Lotto_Manager', 'Coder' and `supervisor_chain` (in a sense, all the nodes — including the supervisor — that we have declared so far).
import operator
from typing import Annotated, Any, Dict, List, Optional, Sequence, TypedDict
import functools
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langgraph.graph import StateGraph, END
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
next: str
lotto_agent = create_agent(llm, tools, "You are a senior lotto manager. you run the lotto and get random numbers")
lotto_node = functools.partial(agent_node, agent=lotto_agent, name="Lotto_Manager")
code_agent = create_agent(llm, [python_repl_tool], "You may generate safe python code to analyze data and generate charts using matplotlib.")
code_node = functools.partial(agent_node, agent=code_agent, name="Coder")
workflow = StateGraph(AgentState)
workflow.add_node("Lotto_Manager", lotto_node)
workflow.add_node("Coder", code_node)
workflow.add_node("supervisor", supervisor_chain)
Edges
Although it is not declared with explicit code, in LangGraph, Edges play the role of connecting node-to-node (or agent-to-agent, or supervisor-to-supervisor, or tool-to-tool — by now you should have a rough conceptual sense of what a 'node' means).
This example code first iterates through each member with a loop and adds a path between the member and the 'supervisor' node. Then, after creating a conditional_map containing the rule that determines the choices for moving on to the next step in the conversation, it adds the supervisor — which holds those conditions in the conditional_map — and sets it as the entry point. Finally, it expresses the process of compiling the workflow built up so far into the final graph.
for member in members:
workflow.add_edge(member, "supervisor")
conditional_map = {k: k for k in members}
conditional_map["FINISH"] = END
workflow.add_conditional_edges("supervisor", lambda x: x["next"], conditional_map)
workflow.set_entry_point("supervisor")
graph = workflow.compile()
Run it
The part that runs all the code so far is composed of two areas. To debug the execution step by step, there is code that iterates over graph.stream and prints the per-step results, and code that prints the LLM's final answer to the user's query.
# 'stream'을 사용하여 단계별로 결과를 확인
config = {"recursion_limit": 20}
for s in graph.stream(
{
"messages": [
HumanMessage(content="무작위 로또 번호 10개를 가져와서 히스토그램에 10개의 빈으로 표시하고 마지막에 10개의 숫자가 무엇인지 알려주세요.")
]
}, config=config
):
if "__end__" not in s:
print(s)
print("----")
# 'invoke'를 사용하여 최종 결과를 얻음
final_response = graph.invoke(
{
"messages": [
HumanMessage(content="무작위 로또 번호 10개를 가져와서 히스토그램에 10개의 빈으로 표시하고 마지막에 10개의 숫자가 무엇인지 알려주세요.")
]
}, config=config
)
final_response['messages'][1].content
Since this part personally tripped me up, let me add a more concrete explanation of graph.stream and graph.invoke;;
(To put the conclusion first, graph.stream and graph.invoke are not selected based on the type of a particular service (chatbot vs. QA, etc.); rather, the choice is determined by the kind of interaction that the service needs to provide, or that the developer building the service needs during development.
graph.stream
1. Purpose: Use it when you want to run each step of the state graph sequentially while inspecting intermediate results in real time.
2. Usage examples:
- Debugging: When you want to inspect the per-step execution result to find errors or identify points to improve.
- Real-time monitoring: When building an interactive system that provides intermediate results in real time as the user inputs.
3. Expected intermediate output:
- Lotto_Manager step: in this code, Lotto_Manager generates 10 random lotto numbers
{'messages': [HumanMessage(content='Generated 10 random lotto numbers: [12, 45, 78, 34, 23, 56, 89, 11, 90, 67]', name='Lotto_Manager')]}
- supervisor step: in this code, the supervisor decides the next action
{'messages': [HumanMessage(content='Lotto_Manager has completed generating random numbers. Now routing to Coder.', name='supervisor')]}
- Coder step: in this code, Coder uses the generated lotto numbers to plot a histogram.
{'messages': [HumanMessage(content='Plotted the histogram for the lotto numbers.', name='Coder')]}
- supervisor step (final): in this code, the supervisor wraps up the task and prepares the final result.
{'messages': [HumanMessage(content='All tasks are completed. Final response is ready.', name='supervisor')]}
graph.invoke
1. Purpose: Use it when you want to run all the steps of the state graph at once and obtain only the final result.
2. Usage examples:
- Deployment environment: when building a system that provides only the final result to the user.
- One-off tasks: when intermediate results don't matter and you only need the final result.
