[Agent Study] Multi-Agent , Multi-tool 만들기 - (1) : Custom Tool 만들기

오늘은 멀티 Agent 를 구축하는 과정을 포스팅해보려고 합니다. 아마 시리즈 별로 만들 것 같은데 최대한 쉽게 저도 이해하고 공부하면서 해보려고 합니다. 대 에이전트 시대에 다양한 툴들이 있습니다. 이미 Function Calling 을 평소에 잘쓰던 분들은 익숙 할 수 있지만, 저도 LangGraph 노드를 만들면서, Tavily 를 가져다 쓰는 것도 좋지만 적재 적소에 내가 Tool 을 만들고 쓴다면 얼마나 좋을까 생각을 했었습니다. 

 

Agent 도구에는 CrewAI, AutoGen, LangGraph 다양한게 존재하지만 저는 랭그래를 기준으로 실습하려고 합니다. 흐름 제어, 분기 처리에 강점이 있다고 생각했고, 노드간 개별 분리를 통해 상호의존성이 낮다고 생각했고, 랭그래프도 프로덕트 레벨에선 쓰기 어렵겠지만 그나마 셋중엔 제일 가깝지 않을까 생각했습니다.

 

Home

 

from dotenv import load_dotenv
from langchain_community.tools.tavily_search import TavilySearchResults

load_dotenv()

tool = TavilySearchResults(max_results=2)
tools = [tool]
result = tool.invoke("김덕배가 누구지?")
print(result)

 

일반적으로 Tool 을 사용할 때 많이들 보시게되는 예입니다. 이미 Tavily 라는 검색툴을 제공을 하고 있습니다. 저도 많이 활용을 했던 기억이나네요. 결과는 아래와 같이 json 타입으로 김덕배 관련 검색내용이 나옵니다 max_results=2 상위 2개의 내용을 출력시킵니다.

[{'url': 'https://thewiki.kr/w/김덕배', 'content': ', 마침 개콘 김덕배의 여파가 어느 정도는 남아 있던지라 이걸 더 토착화시킨 별명인 김덕배를 많은 팬들이 즐겨 사용하고 있다. 그리고, 2023년 한국의 한 프로그램에 출연하여 자신을 김덕배라고 소개했다. 김덕배가 김덕배를 소개 덕하 많관부 1.3.'}, 
{'url': 'https://www.fmkorea.com/3402812374', 'content': '케빈 데브라이너 별명이 김덕배가 된게 먼저임유튜버 김덕배가 먼저임? 에펨코리아 - 유머, 축구, 인터넷 방송, 게임, 풋볼매니저 종합 커뮤니티 포텐터짐'}]

 

이툴을 Chatgpt 라던지 LangChain 에서 정의한 LLM 과 연결하는 작업을 해야합니다.

 

from typing import Annotated
from langchain_openai import ChatOpenAI
from typing_extensions import TypedDict

from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages


class State(TypedDict):
    messages: Annotated[list, add_messages]


graph_builder = StateGraph(State)


llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools(tools)


def chatbot(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}


graph_builder.add_node("chatbot", chatbot)

 

 

class State(TypedDict)

• State는 상태를 정의한 TypedDict로, 챗봇이 처리할 메시지를 포함하는 messages 키를 가지고 있습니다.
• messages는 리스트 형태이며, add_messages 메타데이터가 추가되어 메시지 관리에 도움을 줍니다

graph_builder = StateGraph(State)

• StateGraph 객체를 생성하며, 위에서 정의한 State 구조를 기반으로 작동합니다.

llm_with_tools = llm.bind_tools(tools)

• llm 에 도구를 바인딩 시키는 작업

그래프는 State dict 기반으로 메시지들이 저장이되고, 그 걸 기반으로 노드, 엣지를 만들어서 흐름제어를 시킬 수 있습니다. 스테이트의 'message' 의 내용을 가져와서 invoke를 해서 다시 메시지에 저장 한 후 리턴하는 챗봇 방식입니다. 이렇게 [-1]  인덱싱을 통해서 최근 메시지 대화내역등을 통해서 챗봇을 구현합니다. 그 후 그래프를 빌더시켜서 노드를 만들어 줍니다.

 

class BasicToolNode:

    def __init__(self, tools: list) -> None:
        self.tools_by_name = {tool.name: tool for tool in tools}

    def __call__(self, inputs: dict):
        if messages := inputs.get("messages", []):
            message = messages[-1]
        else:
            raise ValueError("No message found in input")
        outputs = []
        for tool_call in message.tool_calls:
          # tool_call: {'name': 'tavily_search_results_json', 'args': {'query': '김덕배가 누구야?'}, 'id': 'call_m23dkockodwpclas', 'type': 'tool_call'}
            tool_result = self.tools_by_name[tool_call["name"]].invoke(
                tool_call["args"]
            )
            outputs.append(
                ToolMessage(
                    content=json.dumps(tool_result),
                    name=tool_call["name"],
                    tool_call_id=tool_call["id"],
                )
            )
        # outputs: [ToolMessage(content='[{"url": "https://thewiki.kr/w/김덕배",...]', name='tavily_search_results_json', tool_call_id='call_m23dkockodwpclas')]
        return {"messages": outputs}


tool_node = BasicToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)

 

아래와 같이 tool node 를 정의해서도 만들 수 있다. 툴엔 tool 의 내용이 들어가고 먼저 llm 이 어떤 툴을 선택할지 고를때 호출이 일어나고, 그 후 툴을 활용해서 한번더 호출이 발생한다. tool_call 이 반복문을 순회하면서 그 안에 들어있는 name을 기반으로 output이 나오게 된다.

 

from typing import Literal

def route_tools(state: State):
    """
    메시지 상태를 기반으로, 도구 호출이 있는지 확인하여
    적절한 다음 노드로 라우팅(연결)하는 함수입니다.
    
    - 도구 호출이 있는 경우: "tools" 노드로 라우팅합니다.
    - 도구 호출이 없는 경우: "END"로 라우팅하여 대화 흐름을 종료합니다.
    
    이 함수는 LangGraph의 conditional_edge에서 사용됩니다.
    """
    # 상태가 리스트 형태일 경우, 마지막 메시지를 가져옵니다.
    if isinstance(state, list):
        ai_message = state[-1]
    # 상태가 딕셔너리 형태일 경우, "messages" 키를 통해 메시지를 가져옵니다.
    elif messages := state.get("messages", []):
        ai_message = messages[-1]
    else:
        # 메시지가 없는 경우 예외를 발생시킵니다.
        raise ValueError(f"No messages found in input state to tool_edge: {state}")
    
    # 메시지에 도구 호출(tool_calls)이 포함되어 있고, 해당 리스트가 비어 있지 않은 경우
    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
        # 도구 호출이 있으므로 "tools" 노드로 라우팅
        return "tools"
    
    # 도구 호출이 없으므로 "END" 노드로 라우팅
    return END


# Conditional Edge: 조건에 따라 그래프의 다음 노드를 결정하는 설정
graph_builder.add_conditional_edges(
    "chatbot",  # 현재 노드 이름
    route_tools,  # 조건을 평가하는 함수
    {
        # 조건 함수의 반환값에 따라 연결될 노드를 정의합니다.
        "tools": "tools",  # 도구 호출 시 "tools" 노드로 이동
        END: END           # "END"일 경우 종료
    },
)

# Graph Edges: 노드 간의 연결 설정
# "tools" 노드에서 작업이 끝난 후, 다시 "chatbot" 노드로 연결
graph_builder.add_edge("tools", "chatbot")

# START 노드에서 챗봇 노드로 첫 연결을 설정
graph_builder.add_edge(START, "chatbot")

# 그래프 컴파일: 설정한 노드와 연결을 기반으로 그래프를 완성
graph = graph_builder.compile()

 

 

이런식으로 하면 Chatbot 에서 tool 을 써야하는 경우는 Tool 로 컨디셔널 엣지가 발동이 되어 사용해야하면 tool 로, 안 써도 되는 경우 End 로 가서 답변을 출력하고 마무리한다. 

 

이제 커스텀 툴을 만들어 보겠다.

from langchain_core.tools import tool


@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b


# Let's inspect some of the attributes associated with the tool.
print(multiply.name)
print(multiply.description)
print(multiply.args)

 

이런식으로 함수 단위에  input 을 정의해주고, 함수 형태로 데코레이터를 써서 툴을 정의를 한다.

 

from pydantic import BaseModel, Field


class CalculatorInput(BaseModel):
    a: int = Field(description="first number")
    b: int = Field(description="second number")


@tool("multiplication-tool", args_schema=CalculatorInput, return_direct=True)
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b


# Let's inspect some of the attributes associated with the tool.
print(multiply.name)
print(multiply.description)
print(multiply.args)
print(multiply.return_direct)

 

또 pydantic 을 통해 데이터 형식을 정의해서 더욱 힌트를 준다는 개념이 될 수 있다.

 

@ 데코레이터가 정의된 주요 속성:

1. name: "multiplication-tool" -  도구의 이름입니다. 외부에서 호출할 때 이름으로 참조합니다.
2. args_schema: CalculatorInput - 도구가 기대하는 입력 스키마. Pydantic 모델을 통해 입력값의 타입과 유효성을 검사합니다.
3. return_direct: True - 결과값을 직접 반환합니다. (False로 설정하면, 추가적으로 메시지 생성 단계에서 결과가 가공될 수 있습니다.)

하위 클래스를 적용하는 방법이 있는데 나는 이걸 활용해서 Custom Tool 을 만들 생각이다.

 

from typing import Optional, Type

from langchain_core.callbacks import (
    AsyncCallbackManagerForToolRun,
    CallbackManagerForToolRun,
)
from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field


class CalculatorInput(BaseModel):
    a: int = Field(description="first number")
    b: int = Field(description="second number")


# Note: It's important that every field has type hints. BaseTool is a
# Pydantic class and not having type hints can lead to unexpected behavior.
class CustomCalculatorTool(BaseTool):
    name: str = "Calculator"
    description: str = "useful for when you need to answer questions about math"
    args_schema: Type[BaseModel] = CalculatorInput
    return_direct: bool = True

    def _run(
        self, a: int, b: int, run_manager: Optional[CallbackManagerForToolRun] = None
    ) -> str:
        """Use the tool."""
        return a * b

    async def _arun(
        self,
        a: int,
        b: int,
        run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
    ) -> str:
        """Use the tool asynchronously."""
        # If the calculation is cheap, you can just delegate to the sync implementation
        # as shown below.
        # If the sync calculation is expensive, you should delete the entire _arun method.
        # LangChain will automatically provide a better implementation that will
        # kick off the task in a thread to make sure it doesn't block other async code.
        return self._run(a, b, run_manager=run_manager.get_sync())

 

입력정의를 한 후 tool 클래스를 선언하고 BaseTool 을 상속받아서 이름, 주석, args_schema 를 입력정의 한것에 넣어주고, 동기, 비동기 부분에서 사용자 함수를 적용하면 된다. 너무나도 편하고 직관적인 Tool 사용법이다.

 

사용 예제

 

import yfinance as yf
from datetime import datetime
from typing import Dict, Any, Optional, Type, Union
from pydantic import BaseModel, Field
from langchain_core.tools import BaseTool
from langchain_core.callbacks import AsyncCallbackManagerForToolRun, CallbackManagerForToolRun
import asyncio

class StockPriceInput(BaseModel):
    """Stock Price Tool의 입력 모델"""
    symbol: str = Field(..., description="주식 심볼 (예: AAPL)")
    company_name: Optional[str] = Field(default="", description="회사명 (예: 'Apple')")

class StockPriceAnalyzer:
    """주식 가격 정보 분석 클래스"""
    async def get_stock_price(self, symbol: str, company_name: str = "") -> Dict[str, Any]:
        """주식 기본 가격 정보 조회"""
        try:
            ticker = yf.Ticker(symbol)
            info = ticker.info
            hist = ticker.history(period="1d")
            
            if hist.empty:
                return {"error": "주가 데이터를 가져올 수 없습니다."}
            
            current_price = float(hist['Close'].iloc[-1])
            open_price = float(hist['Open'].iloc[-1])
            
            return {
                "stock_info": {
                    "symbol": symbol,
                    "company_name": company_name or info.get('longName', 'N/A'),
                },
                "price_data": {
                    "current_price": round(current_price, 2),
                    "open": round(open_price, 2),
                    "high": round(float(hist['High'].iloc[-1]), 2),
                    "low": round(float(hist['Low'].iloc[-1]), 2),
                    "volume": int(hist['Volume'].iloc[-1]),
                    "day_change": round(float(((current_price - open_price) / open_price) * 100), 2),
                    "timestamp": datetime.now().isoformat()
                }
            }
        except Exception as e:
            return {"error": f"주가 데이터 조회 중 오류 발생: {str(e)}"}

class StockPriceTool(BaseTool):
    """주식 가격 조회를 위한 Tool"""
    
    name: str = "stock_price"
    description: str = "기본적인 주가 정보를 조회합니다. 주가 분석에는 활용하지 않습니다. 주가 분석에는 MarketAnalysisTool을 사용합니다."
    args_schema: Type[BaseModel] = StockPriceInput
    
    def _run(
        self,
        symbol: str,
        company_name: str = "",
        run_manager: Optional[CallbackManagerForToolRun] = None,
    ) -> Dict[str, Any]:
        """동기 실행 메서드"""
        return asyncio.run(self._arun(symbol=symbol, company_name=company_name))

    async def _arun(
        self,
        symbol: str,
        company_name: str = "",
        run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
    ) -> Dict[str, Any]:
        """비동기 실행 메서드"""
        try:
            analyzer = StockPriceAnalyzer()
            result = await analyzer.get_stock_price(
                symbol=symbol,
                company_name=company_name
            )
            return result
        except Exception as e:
            return {'error': f'가격 데이터 조회 중 오류 발생: {str(e)}'}

 

 

먼저 입력 정의에서 2개를 받아준다 심볼과 회사이름 그 후 Class 선언을해서 가격정보를 위한 tool 을 만드는데 이건 야후 파이낸스 ticker 를 활용한 함수다. 이코드는 ticker 가 인자로 들어가고 이름은 안들어가도 상관없다. 티커가 들어가서 OHLC 데이터를 가져오는 방식이다. Class Tool 호출에 LLM 이 꼭 작동해야 하는 경우라던지, 이런경우에 이 툴을 썼으면 좋겠다 싶은걸 정확히 명시해주면 툴을 적재적소에 잘 사용할 수 있고, 비동기 처리를 위해 동기 처리도 비동기를 받아서 처리했다.  끝이다. 매우 쉽다.

 

 

 

 

다음에는 이툴을 직접 MultiTool 호출 하는 방식에 대해 알아볼 예정입니다. 사실 이미 코드는 다 구현해놨지만 심층적으로 하나씩 연결해서 멀티에이전트를 구현해보겠습니다.