LangGraph - 실습 3. Agent Supervisor

LangGraph - 에이전트 유형별 실습

intro
Agent Executor
Chat Executor
👉  Agent Supervisor
Hierarchical Agent Teams
Multi-agent Collaboration

 
[터미널] 환경 변수 및 깃헙 설정 
root 폴더에 환경변수 파일과 깃푸시 배재 파일을 생성한다.

.env 작성
.gitignore 작성

 
[터미널] 가상환경 설정 및 실행
(옵션)가상 환경을 설정하고 실행한다( LangGraph_Agents는 개인적으로 작성한 임의의 이름)

설정 :  python -m venv LangGraph_Agents
실행(mac) :  source LangGraph_Agents/bin/activate

 
환경 변수 로드

from dotenv import load_dotenv
load_dotenv() 

 

---

시작에 앞서 전체적인 구조를 살펴보자

이전 그래프와 달라진 점은 노드와 노드가 아닌 슈퍼바이저(조금 더 구체적으로는 슈퍼바이저 체인)와 개별 노드간 연결이라는 점이다. 이 점은 뒤에 다룰 Hierarchical Agent Teams 예제의 구조가 어떻게 바뀔지 암시하는 부분이기도 하다. 그래프에 대해 대략적인 코멘트를 붙이자면, 시작점으로 '슈퍼바이저 체인(그룹 = Prompt + LLM + Parser)'과 tools 목록 전체를 들고? 있는 노드(lotto_agent)와 tools 목록 중 '[python_repl_tool]'만 들고? 할당받은 있는 노드(code_agent)가 연결되어 있다. 

Agent Supervisor Graph

 

 

그럼 슈퍼바이저의 기본 요소가 되는 llm부터 정의해보자,

Model

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4-1106-preview")

 

Tools

PythonREPLTool() 명령을 통해 로컬에서 코드를 실행 툴, 문자열을 소문자로 반환하는 툴과 0-100 사이 랜덤 수 반환하는 툴을 생성한 후 툴 목록에 넣는 과정

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

말 그대로 도우미 유틸이다. 명시적인 것은 아니고 작업자의 취향에 따라 관련 기능들을 구체화할 수 있다. 예시 코드는 이번 예제 중, 아래 작업을 수행하는 과정에서 반복되는 기능들을 묶어서 만들어 둔 함수로써 가장 중요하면서도 많이 사용되는 '에이전트를 생성'하고 사용할 수 있는 '각 에이전트 별 노드를 구성'하는 함수들을 선언하고 있다.

1. def create_agent() : 자연어 이해 및 생성 작업에 활용할 수 있는 대화형 에이전트 생성 함수를 선언한다
   1) `ChatOpenAI`를 LLM(Language Model) 파라미터의 (고정? 상시)인자 값으로 지정한다
   2) `prompt` 구성 - 입력 메시지와 시스템 프롬프트에 기반한 ChatPromptTemplate 객체와 에이전트의 행동과 스크래치패드를 포함
   3) `agent` 구성 - `create_openai_tools_agent` 함수를 사용하여 지정된 LLM, 도구 리스트 그리고 시스템 프롬프트를 지정 
   4) `AgentExecutor` 에이전트와 함께 제공된 도구로 객체를 초기화

2. def agent_node() : 주어진 상태를 기반으로 에이전트의 동작을 실행하고 결과 메시지를 반환할 수 있도록 하는 `agent_node()` 함수를 선언한다.

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

 

위 도식에서 첫 번째 라운딩 박스에 해당하는 부분 상위 슈퍼바이저로 하위 노드를 관리하는 역할을 담당한다. 기본적인 구성은 members, system_prompt, promptTemplate, supervisor_chain으로 구성된다. 
참고로 이 예시 코드에서는 "Lotto_Manager", "Coder"를 members의 하위 노드로 지정하고 있다. 시스템(슈퍼바이저)의 프롬프트는 1)작업 완료 시점(옵션 정의) 과 2)다음 수행할 에이전트 선택 구조(라우트 함수 정의) 를 구성했고 프롬프트 템플릿은 system_prompt 와 MessagesPlaceholder를 포함하고 있으며 작업 순서를 결정하는 체인은 프롬프트, bind 함수, 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

앞의 실습(Agent Executor, Chat Executor)에서와 같이 AgentState는 에이전트의 상태 구조를 정의하는 딕셔너리 정의하게 된다. 
- messages' 필드에는 BaseMessage(이전 대화) 인스턴스의 시퀀스를 저장하고, - 'next'필드는 다음 상태에 해당하는 경로(route)가 지정되어 있는 위치 가르킨다 .
그리고 바로 위에, 위에서 선언한 Helper Utils 를 사용하여 ‘Lotto_Manager’, ‘Coder’ 멤버(하위 노드)들에 대한 에이전트와 노드를 생성한다.

관련해서 조금 더 상세히 풀어보면;;
lotto : 
   - agent : 언어 모델(llm), 도구, 역할 설명을 매개변수 lotto_agent 객체를 생성
   - node : 에이전트 노드의 자세한 사항을 저장하는 lotto_node라는 부분 함수를 생성 
code :
   - lotto와 동일하게 agent, node 함수를 생성하는데 단지 다른 점은, 내부적으로 선언한 tools가 아닌
     python_repl_tool(로컬에서 파이썬 코드를 수행하기 위한 패키지)를 사용하고 있다는 점이다. 

위의 일련의 과정이 끝나면, AgentState 타입의 상태 그래프로 workflow를 생성하고 이 workflow에 앞에 선언했던  ‘Lotto_Manager’, ‘Coder’ 와 `supervisor_chain`을 (어찌보면 그동안 선언되었던 -슈퍼바이저 포함- 노드들)추가한다.

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

명시적인 코드로 선언되지는 않지만,, LangGraph에서의 Edges는 노드와 노드(또는 에이전트와 에이전트, 또는 슈퍼바이저와 슈퍼바이더, 또는 툴과 툴을 - 이쯤하면 대략적인 '노드'에 대한 개념적 의미에 대해 이해가 되었을 것이다)를 연결하는 역할을 수행한다. 

이 예시 코드는 먼저, 반복문을 통해 각 멤버를 순회?하면서, 멤버와 'supervisor' 노드 사이에 경로를 추가하고 있다. 그리고 대화 중 다음 단계로 넘어갈 수 있는 선택지를 결정하는 규칙을 담은 conditional_map 생성한 후 conditional_map에 조건을 포함하는 supervisor를 추가하고 시작점으로 세팅한다. 마지막으로 그동안 생성했던 워크플로우를 컴파일한 최종 그래프를 생성하는 과정을 표현하고 있다.

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

그동안의 코드를 실행하는 부분은 두 가지 영역으로 구성된다. 코드 수행 과정을 단계별로 디버깅하기 위해 graph.stream을 순회하며 단계별 결과를 출력하는 코드와 사용자의 질의에 대해 LLM의 최종 답변을 출력하는 코드로 구성되어 있다. 

# '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

 

 

---

개인적으로 헷갈렸던 부분이라 graph.stream와 graph.invoke에 대해 조금 더 구체적인 설명을 덧붙인다면 ;; 
(결론부터 정리하면, graph.stream과 graph.invoke은 특정 서비스(이를테면 챗봇이냐, QA냐)의 타입에 따라 선택해서 사용하는 것이 아니라, 그 서비스가 제공해야 하는 또는 서비스를 개발하는 담당자가 개발 과정에서 필요한 상호작용의 종류와 필요에 따라 결정한다.)

graph.stream

1. 목적: 상태 그래프의 각 단계를 순차적으로 실행하면서 실시간으로 중간 결과를 확인하고 싶을 때 사용

2. 사용 예시:
    - 디버깅: 각 단계의 실행 결과를 확인하여 오류를 찾거나 개선점을 파악할 때.
    - 실시간 모니터링: 사용자가 입력할 때마다 중간 결과를 실시간으로 제공하는 인터랙티브한 시스템을 만들 때.

3. 예상되는 중간 결과 출력 :
    - Lotto_Manager 단계 : 이 코드에서는 Lotto_Manager가 10개의 랜덤 로또 번호를 생성한다

{'messages': [HumanMessage(content='Generated 10 random lotto numbers: [12, 45, 78, 34, 23, 56, 89, 11, 90, 67]', name='Lotto_Manager')]}

    - supervisor 단계: 이 코드에서는 supervisor가 다음 작업을 결정한다 

{'messages': [HumanMessage(content='Lotto_Manager has completed generating random numbers. Now routing to Coder.', name='supervisor')]}

    - Coder 단계: 이 코드에서는 Coder가 생성된 로또 번호를 사용하여 히스토그램을 플로팅한다.

{'messages': [HumanMessage(content='Plotted the histogram for the lotto numbers.', name='Coder')]}

    - supervisor 단계 (마지막): 이 코드에서는supervisor가 작업을 마무리하고 최종 결과를 준비한다.

{'messages': [HumanMessage(content='All tasks are completed. Final response is ready.', name='supervisor')]}

 

graph.invoke

1. 목적: 상태 그래프의 모든 단계를 한 번에 실행하고 최종 결과를 얻고 싶을 때 사용

2. 사용 예시:
    - 배포 환경: 사용자에게 최종 결과만을 제공하는 시스템을 만들 때.
    - 일회성 작업: 중간 결과가 중요하지 않고 최종 결과만 필요한 작업을 수행할 때.