---

목차

1. 작성 개요
2. 상태 정의
3. 에이전트 생성
4. 노드 정의
5. 그래프 구성 및 실행

---

1. 작성 개요

Multi Agent 아키텍처 중 이번에는 Supervisor를 사용하는 아키텍처에 대해 정리해 보려고 한다. 이전에 업로드 했던 Colaboration 구조와는 약간의 장단점 차이가 있다.

Supervisor가 Colaboration에 비해 갖는 장점

• 노드 및 Agent 수가 많아질 경우 효율성을 갖는다.
  - Supervisor가 직접 동작 순서를 할당하는 라우터 역할을 하기 때문에 순서가 유동적인 경우 오류 발생 가능성을 줄이거나 토큰 사용을 줄일 수 있다.
• 노드 혹은 Agent가 많아지는 경우에 개발하는 사람들의 관리가 용이해진다.

따라서 Supervisor 형태가 어떻게 구성되는지 해당 게시물에서 정리하려고 한다.

해당 게시물은 테디 노트님의 유료 강의를 듣고 복습 차원으로 정리하는 게시물이기도 하다. 게시물이 부족하다고 여겨진다면 유료 강의를 구매해 들을 수 있기를 권장한다.

2. 상태 정의

import operator
from typing import Sequence, Annotated
from typing_extensions import TypedDict

from langchain_core.messages import BaseMessage


# 상태 정의
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]  # 메시지
    next: str  # 다음으로 라우팅할 에이전트

그래프를 만들어야 하니 우선 Agent들이 공유할 수 있는 AgentState를 정의한다. AgentState의 경우 Colaboration과 다른 지점은 next 부분이다. 어디에서 결과를 받았는지 보다, 어디로 작업물을 전달해야 하는 지가 중요하니까 바뀌었다고 생각한다.

3. 에이전트 생성

• 도구 생성

  from langchain_teddynote.tools.tavily import TavilySearch
  from langchain_experimental.tools import PythonREPLTool
  
  # 최대 5개의 검색 결과를 반환하는 Tavily 검색 도구 초기화
  tavily_tool = TavilySearch(max_results=5)
  
  # 로컬에서 코드를 실행하는 Python REPL 도구 초기화 (안전하지 않을 수 있음)
  python_repl_tool = PythonREPLTool()

  하위 에이전트들이 사용할 도구들을 생성한다

• Agent를 생성하는 Utility

  from langchain_core.messages import HumanMessage


  # 지정한 agent와 name을 사용하여 agent 노드를 생성
  def agent_node(state, agent, name):
      # agent 호출
      agent_response = agent.invoke(state)
      # agent의 마지막 메시지를 HumanMessage로 변환하여 반환
      return {
          "messages": [
              HumanMessage(content=agent_response["messages"][-1].content, name=name)
          ]
      }

Agent의 Node를 만들때 편의성을 추구하기 위해 다음과 같은 함수를 만들 수 있다. 이졸의 도우미 함수인데, 이전과 다른 점은 노드에 State를 인풋으로 받는 것이 아니라 agent와 name을 같이 받는 형태를 가지고 있는 것이다. 우선 굳이 이렇게 하는 것의 이유는 아래와 같다.

    1.에이전트 노드 생성에의 편의성 확보
    2. 작업 흐름 관리의 편의성 => 에이전트간 작업 흐름을 조정하는데의 편의성 확복
    3. 에러 처리 메커니즘 포함 용이성 

다만 인풋으로 agent와 node를 같이 넣는 것에 대한 불편함이 있을 수 있는데, 이를 해결하기 위한 방법이 functools.partial이다.

• functools.partial

  research_node = functools.partial(agent_node, agent=research_agent, names="Researcher")

다음 기능은 기존의 함수의 일부 인자 또는 키워드 인자를 미리 고정하여 새함수를 생성하는데 사용된다. 이를 통해 함수 호출 패턴의 간소화를 추구할 수 있다.

• Supervisor Agent 생성

  from pydantic import BaseModel
  from typing import Literal

  # 멤버 Agent 목록 정의
  members = ["Researcher", "Coder"]

  # 다음 작업자 선택 옵션 목록 정의
  options_for_next = ["FINISH"] + members


  # 작업자 선택 응답 모델 정의: 다음 작업자를 선택하거나 작업 완료를 나타냄
  class RouteResponse(BaseModel):
      next: Literal[*options_for_next]

   from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
  from langchain_openai import ChatOpenAI


  # 시스템 프롬프트 정의: 작업자 간의 대화를 관리하는 감독자 역할
  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."
  )

  # ChatPromptTemplate 생성
  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_for_next), members=", ".join(members))


  # LLM 초기화
  llm = ChatOpenAI(model=MODEL_NAME, temperature=0)


  # Supervisor Agent 생성
  def supervisor_agent(state):
      # 프롬프트와 LLM을 결합하여 체인 구성
      supervisor_chain = prompt | llm.with_structured_output(RouteResponse)
      # Agent 호출
      return supervisor_chain.invoke(state)

Supervisor Agent를 만들 때, 해당 Agent가 작업 관리를 대부분 담당하기 때문에 시스템 프롬프트에 작업이 끝나면 FINISH를 반환하게 해 탈출 요건을 만들어야 한다.

여기서 주의 깊게 봐야 하는 부분은 ChatPromptTemplate의 구조인데, 굳이 AI메시지를 담는게 아니라 시스템 메시지를 중간에 끼워 AI 메시지는 반환하게 하는 구조이다.

4. 노드 정의

import functools
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI


# Research Agent 생성
research_agent = create_react_agent(ChatOpenAI(model="gpt-4o"), tools=[tavily_tool])

# research node 생성
research_node = functools.partial(agent_node, agent=research_agent, name="Researcher")

# 노드의 사용 방법
research_node(
    {
        "messages": [
            HumanMessage(content="Code hello world and print it to the terminal")
        ]
    }
)

노드는 기본적으로 총 3개가 필요할 것으로 보이는데 위와 같이 partial을 이용하여 노드를 정의하는 방법이 있다.

5. 그래프 구성 및 실행

import functools
from langgraph.prebuilt import create_react_agent


# Research Agent 생성
research_agent = create_react_agent(llm, tools=[tavily_tool])
research_node = functools.partial(agent_node, agent=research_agent, name="Researcher")

code_system_prompt = """
Be sure to use the following font in your code for visualization.

##### 폰트 설정 #####
import platform

# OS 판단
current_os = platform.system()

if current_os == "Windows":
    # Windows 환경 폰트 설정
    font_path = "C:/Windows/Fonts/malgun.ttf"  # 맑은 고딕 폰트 경로
    fontprop = fm.FontProperties(fname=font_path, size=12)
    plt.rc("font", family=fontprop.get_name())
elif current_os == "Darwin":  # macOS
    # Mac 환경 폰트 설정
    plt.rcParams["font.family"] = "AppleGothic"
else:  # Linux 등 기타 OS
    # 기본 한글 폰트 설정 시도
    try:
        plt.rcParams["font.family"] = "NanumGothic"
    except:
        print("한글 폰트를 찾을 수 없습니다. 시스템 기본 폰트를 사용합니다.")

##### 마이너스 폰트 깨짐 방지 #####
plt.rcParams["axes.unicode_minus"] = False  # 마이너스 폰트 깨짐 방지
"""


# Coder Agent 생성
coder_agent = create_react_agent(
    llm,
    tools=[python_repl_tool],
    state_modifier=code_system_prompt,
)
coder_node = functools.partial(agent_node, agent=coder_agent, name="Coder")

from langgraph.graph import END, StateGraph, START
from langgraph.checkpoint.memory import MemorySaver

# 그래프 생성
workflow = StateGraph(AgentState)

# 그래프에 노드 추가
workflow.add_node("Researcher", research_node)
workflow.add_node("Coder", coder_node)
workflow.add_node("Supervisor", supervisor_agent)


# 멤버 노드 > Supervisor 노드로 엣지 추가
for member in members:
    workflow.add_edge(member, "Supervisor")

# 조건부 엣지 추가 (
conditional_map = {k: k for k in members}
conditional_map["FINISH"] = END


def get_next(state):
    return state["next"]


# Supervisor 노드에서 조건부 엣지 추가
workflow.add_conditional_edges("Supervisor", get_next, conditional_map)

# 시작점
workflow.add_edge(START, "Supervisor")

# 그래프 컴파일
graph = workflow.compile(checkpointer=MemorySaver())

from langchain_core.runnables import RunnableConfig
from langchain_teddynote.messages import random_uuid, invoke_graph

# config 설정(재귀 최대 횟수, thread_id)
config = RunnableConfig(recursion_limit=10, configurable={"thread_id": random_uuid()})

# 질문 입력
inputs = {
    "messages": [
        HumanMessage(
            content="2010년 ~ 2024년까지의 대한민국의 1인당 GDP 추이를 그래프로 시각화 해주세요."
        )
    ],
}

# 그래프 실행
invoke_graph(graph, inputs, config)

그래프는 3개의 노드로 구성하고 조건부 엣지와 for 문을 사용하여 코드를 간소화하는 부분을 잘 익혀부면 좋을 것 같다.