LangGraph로 EDA만들기 (2) - Supervisor 생성

 

먼저 Supervisor를 사용하여 멀티 에이전트로 개발 진행

 

제일 먼저 Supervisor를 만들고 테스트 하기 위해 data_loader_agent만 생성하여 테스트를 진행

 

Supervisor에서 입력으로 넣은 test_agent_sample.csv파일을 읽어서 df으로 가져왔음

 

from dotenv import load_dotenv
load_dotenv()

from typing import Annotated
from langchain_core.tools import tool, InjectedToolCallId
from langgraph.prebuilt import InjectedState, create_react_agent
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command

from eda_agent.agents.data_loader_agent import DataLoaderAgent
from eda_agent.state import EDAAgentState
from util import pretty_print_messages

def create_handoff_tool(*, agent_name: str, description: str | None = None):
    name = f"transfer_to_{agent_name}"
    description = description or f"Ask {agent_name} for help."

    @tool(name, description=description)
    def handoff_tool(
        state: Annotated[EDAAgentState, InjectedState],
        tool_call_id: Annotated[str, InjectedToolCallId],
    ) -> Command:
        tool_message = {
            "role": "tool",
            "content": f"Successfully transferred to {agent_name}",
            "name": name,
            "tool_call_id": tool_call_id,
        }
        return Command(
            goto=agent_name,  
            update={**state, "messages": state["messages"] + [tool_message]},  
            graph=Command.PARENT,  
        )

    return handoff_tool


# Handoffs
assign_to_data_loader_agent = create_handoff_tool(
    agent_name="data_loader_agent",
    description="Assign task to a data_loader_agent.",
)

supervisor_agent = create_react_agent(
    model="openai:gpt-4o-mini",
    tools=[assign_to_data_loader_agent],
    prompt=(
        "너는 agents를 관리하는 supervisor입니다.:\n"
        "- data_loader_agent는 경로에 있는 파일을 읽어 df으로 변환합니다. state의 file_path를 참고해서 파일을 읽어줘\n"
        "Assign work to one agent at a time, do not call agents in parallel.\n"
        "Do not do any work yourself."
    ),
    name="supervisor",
    state_schema=EDAAgentState,
)

data_loader_agent = DataLoaderAgent()

# Define the multi-agent supervisor graph
supervisor = (
    StateGraph(EDAAgentState)
    # NOTE: `destinations` is only needed for visualization and doesn't affect runtime behavior
    .add_node("supervisor", supervisor_agent, destinations=("data_loader_agent", END))
    .add_node("data_loader_agent", data_loader_agent)
    .add_edge(START, "supervisor")
    # always return back to the supervisor
    .add_edge("data_loader_agent", "supervisor")
    .compile()
)

# 초기 state 설정 - 빈 값들로 초기화
initial_state = {
    "messages": [
        {
            "role": "human",
            "content": "file_path: test_agent_sample.csv",
        }
    ],
    "remaining_steps": 25,
    "file_path": None,
    "data": None,
}

for chunk in supervisor.stream(initial_state):
    pretty_print_messages(chunk, last_message=True)

graph = supervisor

 

LangStudio에서 똑같은 graph를 실행하면 에러가 발생한다
LangSmith에서 데이터를 확인해봐도 InputData 가지 똑같은데
agent를 호출하는 부분에서 data가 df만 가능하다고 유효성 에러가 발생한다.

당분간은 터미널에서만 테스트 진행

langstudio 화면

langstudio에서 실행

 

터미널에서 실행