[LLM 개발기초] 9. LangGraph 를 이용한 Multi-Agent 시스템 구성


Agent 와 도구(Tool)를 이용하면 LLM 의 추론능력으로 자동화된 작업 처리능력을 얻을 수 있습니다. 사용자의 문제를 해결하기 위하여 Agent는 자신에게 주어진 도구들을 이용해서 일련을 작업들을 구성해냅니다.

이런 강력한 Agent 도 만능은 아닙니다. 하나의 Agent 가 많은 도구를 사용하여 사이즈가 큰 작업을 처리해야하는 경우 Agent 의 성능이 저하됩니다. 따라서 Agent 를 기능/작업 단위로 구분하고 여러 Agent 가 상호 소통하면서 문제를 처리해야 할 필요가 있습니다. 이를 Multi-Agent 구조라고 합니다.

LLM 을 이용한 어플리케이션 구성에서 Multi-Agent 구조를 활용하는 이론은 아래 논문에서 제시되었으며, LangChain 라이브러리에서는 LangGraph 로 구현되었습니다.

LangGraph 가 Multi-Agent 구현만을 위한 도구는 아닙니다. LLM 을 이용한 코딩 작업을 모듈 단위로 나누어서 연동하는데 사용할 수도 있고, 비교적 큰 LLM 개발 프로젝트를 할 때 구조를 효율적으로 만들기 위한 도구로 사용할 수도 있습니다. 이를 위해 LangGraph 에서는 노드, 엣지, 워크플로우 등의 개념을 사용하여 작업을 모듈화 할 수 있도록 도와줍니다. 또한 사용자가 만든 구조를 시각화 하기 위한 도구도 제공됩니다.

보다 구체적인 LangGraph 의 기능은 다음과 같습니다.

  • 시각화
    • 에이전트와 도구, 데이터 흐름 등을 노드와 에지로 표현하여 시스템의 전체적인 구조를 시각적으로 보여줍니다.
  • 관리
    • 에이전트와 도구를 생성, 수정, 삭제하고, 이들 간의 연결 관계를 관리할 수 있습니다.
  • 디버깅
    • 시스템에서 발생하는 문제를 찾아내고 해결하는 데 도움을 줍니다.
  • 저장 및 로딩
    • 생성된 그래프를 저장하고 다시 로딩하여 재사용할 수 있습니다.

LangGraph 를 사용함으로써 다음과 같은 이점을 얻을 수 있습니다.

  • 복잡성 관리
    • 복잡한 에이전트 시스템을 간단한 그래프로 표현하여 시스템의 이해를 돕습니다.
  • 오류 감지
    • 그래프를 통해 데이터 흐름의 문제점이나 불필요한 연결을 쉽게 찾아낼 수 있습니다.
  • 시스템 설계
    • 새로운 에이전트를 추가하거나 기존 에이전트를 수정할 때 그래프를 기반으로 시스템 설계를 변경할 수 있습니다.
  • 팀 협업
    • 팀원들과 시스템 구조를 공유하고 함께 시스템을 개발할 수 있습니다.


LangGraph 실습을 위해 LangChain 에서 제공하는 official 예제 코드를 로컬 PC 환경에서 실행할 수 있도록 준비했습니다.

실습 코드를 분석하기 전에 LangGraph 가 사용하는 노드, 엣지, 빌더(워크플로우), 그래프의 개념을 먼저 이해해야합니다.

LangGraph 의 역할을 단순화하면, 작업을 모듈 단위로 나누어서 관리하고 서로간에 유기적으로 연동해서 문제를 해결할 수 있도록 하는 것임을 기억하세요.

  • 노드 (Node)
    • 특정 작업을 모듈화 한 구현체로 Agent, 함수 또는 클래스를 말합니다.
    • 따라서 작업이 실행되는 단위이기도 합니다.
    • 위 이미지에서 a, b, c, d 에 해당
  • 엣지 (Edge)
    • 각 노드들을 이어주는 라인에 해당하며 노드간 작업의 전환을 의미합니다.
    • 노드가 작업을 마치면 실행해야할 다음 작업을 나타내는데 사용됩니다
    • 정적인 실행(add_edge), 조건부 실행(add_conditional_edges)을 사용할 수 있습니다
  • 빌더, 워크플로우 (Builder, Workflow)
    • 위 이미지 좌측 코드에서 빌더로 표시되는 변수
    • 실제로 사용할 그래프의 타입에 따라 만들어지는 그래프 builder 입니다.
    • 노드나 엣지가 빌더에 등록됨으로써 서로 결합하여 실행될 준비가 됩니다
  • 그래프 (Graph)
    • 빌더를 컴파일하면 생성되는 인스턴스입니다.
    • graph.invoke() 를 통해 그래프를 실행하면 설정한 노드와 엣지의 구조대로 작업을 수행합니다.
  • 라우터 (Router)
    • add_conditional_edges() 를 통해 조건부로 노드간 연결을 할 때, 어떤 조건에서 어떤 노드가 연결되어야 할지에 대한 코드를 담고있는 함수가 라우터입니다.
    • 위 이미지에는 표시되지 않았지만, 실습 코드에서는 라우터 함수를 정의해서 사용합니다.

LangGraph에서는 주로 StateGraph와 MessageGraph를 사용하지만, 다른 유형의 그래프도 존재합니다.

  • StateGraph
    • 상태 기반 워크플로우를 모델링하는 데 사용
    • 복잡한 의사결정 로직과 조건부 흐름을 구현
  • MessageGraph
    • 메시지 기반 시스템을 모델링하는 데 사용
    • 메시지의 흐름과 변환을 추적하는 데 유용
  • DAG (Directed Acyclic Graph)
    • 비순환 방향 그래프
    • 작업의 순서와 의존성을 모델링
  • Channel
    • 비동기 통신을 위한 단순한 그래프 구조
  • Pydantic Graph
    • Pydantic 모델을 사용하여 그래프의 노드와 엣지를 정의
    • 데이터 유효성 검사와 직렬화를 그래프 구조에 통합
  • TypedGraph
    • 강력한 타입 검사를 제공하는 그래프 구조
    • 복잡한 시스템에서 타입 안전성을 보장



실습 1. 인터넷 검색과 차트 작성을 해주는 Multi-Agent 예제

LangChain 사이트에서 제공하는 Multi-Agent 예제는 사용자의 요구사항을 분석해서 필요한 자료를 검색하고, 자료를 차트로 만들어 이미지로 저장해주는 예제입니다.

이를 위해 작업을 처리하는 2개의 Agent – Researcher, ChartGenerator 를 생성합니다.

  • Researcher Agent
    • 질문에 대한 응답을 만들기 위해 필요한 데이터를 만드는 Agent 입니다. Tavily search AI 검색 도구를 사용합니다.
  • ChartGenerator Agent
    • 차트를 만들기 위해 파이썬 코드를 만들고, 코드를 실행하는 Agent 입니다. PythonREPL 을 이용해서 파이썬 코드를 사용자의 PC에서 실행합니다.

실습 예제는 Researcher, ChartGenerator – 2개의 Agent 를 만드는데, Agent 가 직접 도구를 실행하지는 않습니다. 대신 도구들을 실행하게 해주는 call_tool (ToolExecutor)가 노드로 추가되어 있습니다. Agent 에서 도구를 사용할 때 router 를 거쳐 ToolExecutor 를 실행합니다.

Router 는 노드 간 작업을 전환하기 위한 조건식을 담고 있기 때문에 작업을 조율하는 역할을 합니다.

실습 코드를 실행해보면 아래와 같은 순서로 실행됩니다.

  1. START
  2. Researcher (검색 도구를 호출하면서 데이터 수집)
    • Router
  3. ToolExecutor
    • Tavily 검색 수행
  4. [2-3] 과정을 반복
  5. Researcher (데이터 수집 완료)
    • Router
  6. ChartGenerator (파이썬 코드 제작)
    • Router
  7. ToolExecutor
    • PythonREPL 실행
  8. END

실습 코드는 아래와 같습니다. LangChain 에서 제공하는 예제 코드에서 문제점들을 수정하고 정리한 코드입니다. [09_01_LangGraph_multi_agent.py]

import os
import functools
import operator
import subprocess
from typing import Annotated, Sequence, TypedDict
from typing import Literal
from dotenv import load_dotenv
from langchain_core.messages import AIMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
from langchain_core.messages import (
    BaseMessage,
    HumanMessage,
    ToolMessage,
)
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langgraph.graph import END, StateGraph, START
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.tools import tool
from langchain_core.runnables import RunnableConfig
from langgraph.prebuilt import ToolNode
from langchain_experimental.utilities import PythonREPL

# 프로세스 실행을 위한 환경설정 및 파라미터 준비
# LangSmith 사이트에서 아래 내용을 복사해서 .env 파일에 입력
# TAVILY_API_KEY=<your_api_key>
# LANGCHAIN_TRACING_V2=true
# LANGCHAIN_ENDPOINT=https://api.smith.langchain.com
# LANGCHAIN_API_KEY=<your_api_key>
# LANGCHAIN_PROJECT="<your_project_name>"
load_dotenv()


#################################################################
# 1. Utility 함수 및 LangChain/Agent/LangGraph 관련 함수 정의
#################################################################
# 1-1. 입력한 파라미터를 이용해서 Agent 를 생성하는 함수
def create_agent(llm, tools, system_message: str):
    """Create Researcher agent."""
    prompt = ChatPromptTemplate.from_messages(
        [
            (
                "system",
                "You are a helpful AI assistant, collaborating with other assistants."
                " Use the provided tools to progress towards answering the question."
                " If you are unable to fully answer, that's OK, another assistant with different tools "
                " will help where you left off. Execute what you can to make progress."
                " If you or any of the other assistants have the final answer or deliverable,"
                " prefix your response with FINAL ANSWER so the team knows to stop."
                " You have access to the following tools: {tool_names}.\n{system_message}",
            ),
            MessagesPlaceholder(variable_name="messages"),
        ]
    )
    prompt = prompt.partial(system_message=system_message)
    prompt = prompt.partial(tool_names=", ".join([tool.name for tool in tools]))
    return prompt | llm.bind_tools(tools)


# 1-2. Graph node 사이에 전달되는 state 데이터 클래스를 정의
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    sender: str


# 1-3. Agent 에 해당하는 node 를 생성
#      이 함수 자체가 node 에 해당하며, agent 를 실행해주는 executor 이다
#      AgentState 객체인 state 를 입력받고 새로운 state 를 반환한다
#      즉, Node 의 agent 실행 결과는 다음 실행할 node 정보를 담고 있어야 한다.
def agent_node(state, agent, name):
    result = agent.invoke(state)
    if isinstance(result, ToolMessage):
        pass
    else:
        result = AIMessage(**result.dict(exclude={"type", "name"}), name=name)

    print(f"Current agent name = {name}")
    return {
        "messages": [result],
        "sender": name,
    }


# 1-4. Node 간 작업을 조율할 Router 생성
#      Router 에 전달된 state 를 통해 현재 상태를 파악
#      다음 수행할 작업을 string 으로 리턴 (Literal 로 미리 지정된 문자열 중 하나를 사용)
def router(state) -> Literal["call_tool", "__end__", "continue", "to_chart_generator"]:
    messages = state["messages"]
    last_message = messages[-1]
    if last_message.tool_calls:
        print(f"-----> Router: Calling ToolExecutor, sender={state["sender"]}\n----->")
        return "call_tool"
    if "FINAL ANSWER" in last_message.content:
        print(f"-----> Router: __end__, sender={state["sender"]}\n----->")
        return "__end__"
    if "to_chart_generator" in last_message.content:
        print(f"-----> Router: Move to ChartGenerator, sender={state["sender"]}\n----->")
        return "to_chart_generator"
    if state["sender"] == "Researcher":
        print(f"-----> Router: Move to ChartGenerator, sender={state["sender"]}\n----->")
        return "to_chart_generator"
    print(f"-----> Router: continue, sender={state["sender"]}\n----->")
    return "continue"


#################################################################
# 2. Agent 에서 사용할 도구들 (Tools)
#################################################################
# 2-1. PythonREPL 인스턴스 생성
# Warning: This executes code locally, which can be unsafe when not sandboxed
repl = PythonREPL()

# 2-2. 파이썬 코드 실행 도구 정의
@tool
def python_repl(code):
    """Use this to execute python code. If you want to see the output of a value,
    you should print it out with `print(...)`. This is visible to the user."""
    try:
        # Matplotlib 백엔드 설정 추가 (메인 스레드 외부에서 실행 가능하도록)
        setup_code = """
import matplotlib
import os
import sys

if sys.platform.startswith('win'):
    matplotlib.use('TkAgg')  # Windows
elif sys.platform.startswith('darwin'):
    matplotlib.use('MacOSX')  # macOS
else:
    matplotlib.use('TkAgg')  # Linux and others

import matplotlib.pyplot as plt
"""
        # 파이썬 코드 실행
        repl.run(setup_code)
        result = repl.run(code)
    except BaseException as ex:
        print(f"Execution failed. Error: {repr(ex)}")
        return f"Failed to execute. Error: {repr(ex)}"

    result_str = f"Successfully executed:\n```python\n{code}\n```\nStdout: {result}"
    return (
        result_str + "\n\nIf you have completed all tasks, respond with FINAL ANSWER."
    )


# 2-3. AI 검색 도구 정의, [.env] 파일에 TAVILY_API_KEY 환경설정 필요
tavily_tool = TavilySearchResults(max_results=5)

# 2-4. ToolExecutor 에서 사용할 Tool list
tools = [tavily_tool, python_repl]


#################################################################
# 3. 그래프 구조 생성 (LLM, agent, node, router, workflow)
#################################################################
# 3-1. LLM 초기화
llm = ChatOpenAI(model="gpt-4o-mini")

# 3-2. Researcher Agent 생성
research_agent = create_agent(
    llm,
    [tavily_tool],
    # Agent 의 역할과 tool/agent 와의 연동이 원활하도록 충분히 설명 필요
    system_message="""You are a Research agent.
Your role is to gather accurate and concise data using search tools.
(Use tavily_tool for this job)
You should provide accurate data for the ChartGenerator to use.
When you are ready to launch ChartGenerator, include below string at your response.
'to_chart_generator'
""",
)
research_node = functools.partial(agent_node, agent=research_agent, name="Researcher")

# 3-3. ChartGenerator Agent 생성
chart_agent = create_agent(
    llm,
    [python_repl],
    # Agent 의 역할과 tool/agent 와의 연동이 원활하도록 충분히 설명 필요
    system_message="""You are a ChartGenerator agent.
When you receive data from Researcher, create python codes to draw a chart.
Always use the python_repl tool to execute your code.
After generating the chart, respond with FINAL ANSWER.""",
)
chart_node = functools.partial(agent_node, agent=chart_agent, name="ChartGenerator")

# 3-4. Tool node
tool_node = ToolNode(tools)

# 3-5. Workflow(builder) 생성, Node 설정
workflow = StateGraph(AgentState)

workflow.add_node("Researcher", research_node)
workflow.add_node("ChartGenerator", chart_node)
workflow.add_node("ToolExecutor", tool_node)

# 3-6. Conditional Edge 설정
#      conditional edge 는 시작 Node 와 Router, Path map 을 설정
#      시작 노드의 실행 결과가 router 로 전달됨
#      router 에서 다음 실행할 노드를 찾음 (next node string 리턴)
#      path map 에서 다음 실행할 노드를 확인
workflow.add_conditional_edges(
    "Researcher",
    router,
    {
        "continue": "ChartGenerator", # 데이터 작업이 끝나면 chart 를 작성
        "to_chart_generator": "ChartGenerator", # 데이터 작업이 끝나면 chart 를 작성
        "call_tool": "ToolExecutor",
        "__end__": END,
    },
)
workflow.add_conditional_edges(
    "ChartGenerator",
    router,
    {"continue": "ChartGenerator", "call_tool": "ToolExecutor", "__end__": END},
)
workflow.add_conditional_edges(
    "ToolExecutor",
    # Agent 노드는 'sender' 필드값을 state 데이터에 담아서 전송
    # -> ToolExecutor 는 작업이 끝나면 자신을 호출한 sender 를 다시 호출함
    lambda x: x["sender"],
    {
        "Researcher": "Researcher",
        "ChartGenerator": "ChartGenerator",
    },
)
# 3-7. 그래프의 시작 node 를 지정
workflow.add_edge(START, "Researcher")

# 3-8. 그래프 생성
graph = workflow.compile()

#################################################################
# 4. Graph 구조 이미지 저장 및 출력, 그래프 실행
#################################################################
# 4-1. 그래프 구조 이미지 출력
try:
    # 그래프를 PNG 파일로 저장
    png_data = graph.get_graph(xray=True).draw_mermaid_png()

    # 현재 작업 디렉토리에 'graph.png' 파일로 저장
    file_path = os.path.join(os.getcwd(), "graph.png")
    with open(file_path, "wb") as f:
        f.write(png_data)

    print(f"Graph saved as {file_path}")

    # Windows의 기본 이미지 뷰어로 파일 열기
    subprocess.run(["start", file_path], shell=True, check=True)
except Exception as e:
    print(f"An error occurred: {e}")

# 4-2. 그래프 실행
events = graph.stream(
    {
        "messages": [
            HumanMessage(
                content="""Fetch the UK's GDP data over the past 3 years, 
                then create a line graph of it using Python. 
                Use the Researcher to get the data and call the ChartGenerator to create the graph. 
                The ChartGenerator MUST use the python_repl tool to execute the code 
                and create the chart."""
            )
        ],
    },
    RunnableConfig(recursion_limit=20),  # Maximum number of steps
)

# 4-3. 발생하는 이벤트를 모두 출력 (graph 실행 완료까지 대기하기 위해 아래 코드가 필요)
for s in events:
    # 출력 내용이 무척 많기 때문에 상세 내용은 LangSmith 사이트에서 확인하는 편이 좋다
    # print(s)
    # print("-------------------")
    pass

코드가 상당히 길기 때문에 코드에 달려있는 주석을 확인하세요.

  1. Utility 함수 및 LangChain/Agent/LangGraph 설정
    • 1-1. 입력한 파라미터를 이용해서 Agent 를 생성하는 함수
    • 1-2. node 사이에 전달되는 state 데이터 클래스를 정의
      • sender : 작업을 끝낸 node 를 나타내며 state 를 생성-전달하는 주체
      • messages : node 의 작업 결과를 담는 배열. 다음 실행할 노드를 나타내는 문자열이 포함되어 있어야한다.
    • 1-3. Agent 에 해당하는 node 를 생성
      • 이 함수 자체가 node 에 해당하며, agent 를 실행해주는 executor 이다
      • AgentState 객체인 state 를 입력받고 새로운 state 를 반환한다
      • Node 의 agent 실행 결과는 다음 실행할 node 정보를 담고 있어야 한다
    • 1-4. Node 간 작업을 조율할 Router 생성
      • Router 에 전달된 state 를 통해 현재 상태를 파악
      • 다음 수행할 작업을 string 으로 리턴
        • (Literal 로 미리 지정된 문자열 중 하나를 사용)
  2. Agent 에서 사용할 도구들 (Tools)
    • 2-1. PythonREPL 인스턴스 생성
      • 메인 스레드에서 생성해야 하므로 코드 위치를 바꾸면 안될 수 있음
    • 2-2. 파이썬 코드 실행 도구 정의
      • 메인 스레드 외부에서 실행 가능하도록 설정 추가
    • 2-3. AI 검색 도구 정의
      • .env 파일에 TAVILY_API_KEY 환경설정 필요
    • 2-4. ToolExecutor 에서 사용할 Tool list 정의
  3. 그래프 구조 생성 (LLM, agent, node, router, workflow)
    • 3-1. LLM 초기화
    • 3-2. Researcher Agent 생성
      • Agent의 역할과 tool/agent 와의 연동이 원활하도록 system_message에 충분히 설명 필요
    • 3-3. ChartGenerator Agent 생성
      • Agent의 역할과 tool/agent 와의 연동이 원활하도록 system_message에 충분히 설명 필요
    • 3-4. Tool node
    • 3-5. workflow(builder) 생성, Node 설정
      • 2개의 agent 노드와 1개의 tool 노드를 등록
    • 3-6. Conditional Edge 설정
      • conditional edge 는 시작 Node 와 Router, Path map 을 설정
      • 시작 노드의 실행 결과가 router 로 전달됨
      • router 에서 다음 실행할 노드를 찾음 (node string 리턴)
      • path map 에서 다음 실행할 노드를 확인
    • 3-7. 그래프의 시작 node 를 지정
    • 3-8. 그래프 생성
  4. Graph 구조 이미지 저장 및 출력, 그래프 실행
    • 4-1. 그래프 구조 이미지 출력
      • 그래프에 설정된 노드와 엣지가 PNG 이미지로 저장, 출력됩니다.
    • 4-2. 그래프 실행
    • 4-3. 발생하는 이벤트를 모두 출력
      • 출력 내용이 무척 많기 때문에 콘솔 로그보다는 LangSmith 사이트에서 확인하는 편이 좋습니다.

3번 단계에서 그래프에 설정한 노드와 엣지는 [4-1] 코드에서 get_graph() 를 실행하면 이미지로 확인할 수 있습니다.

# 4-1. 그래프 구조 이미지 출력
try:
    # 그래프를 PNG 파일로 저장
    png_data = graph.get_graph(xray=True).draw_mermaid_png()


코드를 실행해보면 아래처럼 로그가 출력됩니다.

$ python -u "c:\Workspace\......\test_multi_agent_bug_fix.py"

Graph saved as C:\Workspace\......\graph.png
Current agent name = Researcher
-----> Router: Calling ToolExecutor, sender=Researcher
----->
Current agent name = Researcher
-----> Router: Calling ToolExecutor, sender=Researcher
----->
Current agent name = Researcher
-----> Router: Move to ChartGenerator, sender=Researcher
----->
Current agent name = ChartGenerator
-----> Router: Calling ToolExecutor, sender=ChartGenerator
----->
Python REPL can execute arbitrary code. Use with caution.
<string>:8: UserWarning: Starting a Matplotlib GUI outside of the main thread will likely fail.
<string>:15: UserWarning: Starting a Matplotlib GUI outside of the main thread will likely fail.

-----> (여기서 파이썬 코드가 실행되고 그래프의 동작은 멈춤)

Current agent name = ChartGenerator
-----> Router: __end__, sender=ChartGenerator
----->

$

Researcher 에서 답변을 만들기 위해 반복해서 ToolExecutor 를 호출합니다. 데이터 수집 작업이 끝나면 Researcher 에서 ChartGenerator 로 전환되었습니다.

ChartGenerator 에서 차트를 그리는 파이썬 코드를 만들고 ToolExecutor – PythonREPL() 을 이용해 파이썬 코드를 실행합니다.

파이썬 코드가 실행되면 그래프의 동작은 멈춥니다. 파이썬 실행창을 닫으면 ToolExecutor 가 종료되고, ToolExecutor 를 호출한 ChartGenerator 가 다시 실행됩니다.

ChartGenerator 는 종료 시그널(FINAL ANSWER)을 만들고, 앱 실행도 종료됩니다.

이상의 동작 과정은 콘솔로 확인하기보다는 LangSmith – Project – trace 기록으로 확인하세요.

LangGraph 실행 프로세스 (1)
LangGraph 실행 프로세스 (2)




참고자료


You may also like...

답글 남기기

이메일 주소는 공개되지 않습니다. 필수 필드는 *로 표시됩니다

이 사이트는 스팸을 줄이는 아키스밋을 사용합니다. 댓글이 어떻게 처리되는지 알아보십시오.