---

목차

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

---

1. 작성 개요

LangGraph는 멀티 에이전트 시스템을 구축하기 위한 프레임워크로도 활용될 수 있다. 그 첫번째 구조로 MAC(Multi Agent Collaboration)을 제안하려고 한다. 멀티 에이전트라고 해서 막 어려운 개념이 되는 것이 아니라 복수의 에이전트를 사용하게 되는 개념으로 이해하면 된다.

다만 단일 에이전트를 사용하는 것보다, 복수의 에이전트를 사용하는 것에는 몇가지 메리트가 있다. 그것을 제안해 보자면

1. 복수의 에이전트를 활용함으로써, 복잡한 문제 혹은 Task를 해결하는데 도움이 된다. => 복잡한 문제를 나누어 수행하기 때문
2. 에이전트로 하여금 특정 과업의 Task를 전담하게 하여 프롬프트의 효율성과 효과성을 확보한다.
3. 혹시 에이전트가 복수의 도구를 바탕으로 작업하게 할 때, 이로 인해 발생하는 로직의 복잡도를 낮춰 수행 가능성을 높인다.

다음 게시물은 테디노트님의 유료 강의를 듣고 복습하는 차원에서 작성되는 것이기도 하다. 이에 따라 특정 주제에 대한 시각화를 수행하는 MAC를 다룰 것이고, 웹 검색을 하는 에이전트와 시각화를 하는 에이전트를 이용한다. 혹시 해당 글이 부족하다고 여겨진다면, 테디 노트님의 유료 강의를 수강해 볼 것을 권장한다.

2. 상태 정의

우선 상태 먼저 정의 해보자

import operator
from typing import Annotated, Sequence
from typing_extensions import TypedDict
from langchain_core.messages import BaseMessage


# 상태 정의
class AgentState(TypedDict):
    messages: Annotated[
        Sequence[BaseMessage], operator.add
    ]  # Agent 간 공유하는 메시지 목록
    sender: Annotated[str, "The sender of the last message"]  # 마지막 메시지의 발신자

여기서 일단 주목해 볼만한 포인트는 두가지 정도가 있을 것 같다.
1. 두개의 에이전트가 만들어지지만, 상태는 서로 기능적으로 너무 어긋나지 않는다면, 하나의 State를 두개의 에이전트가 공유할 수 도 있다는 부분이다.
2. sender 부분은 집중해서 볼만 하다. 다음 MAC의 구조는 웹서칭을 하는 에이전트와 시각화 하는 에이전트가 핑퐁을 하는 것이 기본 골자다. 다만 웹서칭하는 에이전트의 경우 유저에게서 쿼리도 받아야 한다. 즉 복수의 sender가 존재하는 것. 이와 같은 경우에 어디서 부터 받은 메시지인지 확인할 수 있는 장치가 있어야 하기 때문에, sender라는 데이터를 상태에서 계속 수집해야 한다.

3. 도구 정의

from typing import Annotated

from langchain_teddynote.tools.tavily import TavilySearch
from langchain_core.tools import tool
from langchain_experimental.utilities import PythonREPL

# Tavily 검색 도구 정의
tavily_tool = TavilySearch(max_results=5)

# Python 코드를 실행하는 도구 정의
python_repl = PythonREPL()


# Python 코드를 실행하는 도구 정의
@tool
def python_repl_tool(
    code: Annotated[str, "The python code to execute to generate your chart."],
):
    """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:
        # 주어진 코드를 Python REPL에서 실행하고 결과 반환
        result = python_repl.run(code)
    except BaseException as e:
        return f"Failed to execute code. Error: {repr(e)}"
    # 실행 성공 시 결과와 함께 성공 메시지 반환
    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."
    )

MAC에서 각 에이전트들이 사용할 도구를 정의해야 한다.

4. 에이전트 정의

여기서는 두개의 에이전트를 다루어야 하니 기본적으로는 나눠서 설명하는데, 프롬프트의 효율화를 위해 어느 정도 공통적으로 전달해야하는 시스템 프롬프트의 영역이 있다면 동시에 관리할 수 있도록 함수를 만들어 놓는 것도 좋은 방법이 될 수 있을 것 같다.

def make_system_prompt(suffix: str) -> str:
    return (
        "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."
        f"\n{suffix}"
    )

여기서 가장 중요한 부분은 FINAL ANSWER를 뱉으라는 부분이다. 해당 부분이 존재하기 때문에, 두 에이전트가 핑퐁을 주고 받다가도, 더 작업할 게 없으면 FINAL ANSWER라는 답을 뱉게 되고, 시스템은 종료된다.

1. 웹서칭 에이전트

from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langgraph.graph import MessagesState
from langchain_together import ChatTogether

# LLM 정의
# llm = ChatOpenAI(model=MODEL_NAME)

llm = ChatTogether(model="meta-llama/Llama-3.3-70B-Instruct-Turbo")

# Research Agent 생성
research_agent = create_react_agent(
    llm,
    tools=[tavily_tool],
    state_modifier=make_system_prompt(
        "You can only do research. You are working with a chart generator colleague."
    ),
)


# Research Agent 노드 정의
def research_node(state: MessagesState) -> MessagesState:
    result = research_agent.invoke(state)

    # 마지막 메시지를 HumanMessage 로 변환
    last_message = HumanMessage(
        content=result["messages"][-1].content, name="researcher"
    )
    return {
        # Research Agent 의 메시지 목록 반환
        "messages": [last_message],
    }

2. 차트 생성 에이전트 (시각화 에이전트)

chart_generator_system_prompt = """
You can only generate charts. You are working with a researcher colleague.
Be sure to use the following font code in your code when generating charts.
"""

# Chart Generator Agent 생성
chart_agent = create_react_agent(
    llm,
    [python_repl_tool],
    state_modifier=make_system_prompt(chart_generator_system_prompt),
)

def chart_node(state: MessagesState) -> MessagesState:
    result = chart_agent.invoke(state)

    # 마지막 메시지를 HumanMessage 로 변환
    last_message = HumanMessage(
        content=result["messages"][-1].content, name="chart_generator"
    )
    return {
        # share internal message history of chart agent with other agents
        "messages": [last_message],
    }

다음 부분이 무사히 잘 끝나면 나중에 시스템이 끝날 때 이용할 수 있도록 라우터 함수를 만들어준다.

from langgraph.graph import END


def router(state: MessagesState):
    # This is the router
    messages = state["messages"]
    last_message = messages[-1]
    if "FINAL ANSWER" in last_message.content:
        # Any agent decided the work is done
        return END
    return "continue"

다음과 같은 라우터를 통해 FINAL ANSWER가 나오면 시스템을 종료시킬 수 있다.

5. 그래프 생성 및 실행

from langchain_core.messages import HumanMessage, ToolMessage
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

workflow = StateGraph(MessagesState)
workflow.add_node("researcher", research_node)
workflow.add_node("chart_generator", chart_node)

workflow.add_conditional_edges(
    "researcher",
    router,
    {"continue": "chart_generator", END: END},
)
workflow.add_conditional_edges(
    "chart_generator",
    router,
    {"continue": "researcher", END: END},
)

workflow.add_edge(START, "researcher")
app = 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(app, inputs, config, node_names=["researcher", "chart_generator", "agent"])

생각보다 두개가 핑퐁을 주고 받는 정도의 구조라 쉬워보이긴 한다. ㅇㅅㅇ