[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 는 노드 간 작업을 전환하기 위한 조건식을 담고 있기 때문에 작업을 조율하는 역할을 합니다.
실습 코드를 실행해보면 아래와 같은 순서로 실행됩니다.
- START
- Researcher (검색 도구를 호출하면서 데이터 수집)
- Router
- ToolExecutor
- Tavily 검색 수행
- [2-3] 과정을 반복
- Researcher (데이터 수집 완료)
- Router
- ChartGenerator (파이썬 코드 제작)
- Router
- ToolExecutor
- PythonREPL 실행
- 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
코드가 상당히 길기 때문에 코드에 달려있는 주석을 확인하세요.
- 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 로 미리 지정된 문자열 중 하나를 사용)
- (Literal 로 미리 지정된 문자열 중 하나를 사용)
- Agent 에서 사용할 도구들 (Tools)
- 2-1. PythonREPL 인스턴스 생성
- 메인 스레드에서 생성해야 하므로 코드 위치를 바꾸면 안될 수 있음
- 2-2. 파이썬 코드 실행 도구 정의
- 메인 스레드 외부에서 실행 가능하도록 설정 추가
- 2-3. AI 검색 도구 정의
- .env 파일에 TAVILY_API_KEY 환경설정 필요
- 2-4. ToolExecutor 에서 사용할 Tool list 정의
- 2-1. PythonREPL 인스턴스 생성
- 그래프 구조 생성 (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. 그래프 생성
- Graph 구조 이미지 저장 및 출력, 그래프 실행
- 4-1. 그래프 구조 이미지 출력
- 그래프에 설정된 노드와 엣지가 PNG 이미지로 저장, 출력됩니다.
- 4-2. 그래프 실행
- 4-3. 발생하는 이벤트를 모두 출력
- 출력 내용이 무척 많기 때문에 콘솔 로그보다는 LangSmith 사이트에서 확인하는 편이 좋습니다.
- 4-1. 그래프 구조 이미지 출력
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 기록으로 확인하세요.