AI Agent Tools 생태계: 실세계와 연결하는 도구들


AI Agent Tools란?

AI Agent가 실세계와 상호작용할 수 있게 해주는 기능들입니다.

Tool 없는 AI:

사용자: "서울 날씨 알려줘"
AI: "죄송합니다. 실시간 정보는 제공할 수 없습니다."

Tool을 가진 AI:

사용자: "서울 날씨 알려줘"
AI: [weather_api_tool 사용] → "서울은 현재 맑음, 20도입니다."

필수 Tools

1. 웹 검색

from langchain.tools import DuckDuckGoSearchRun

search = DuckDuckGoSearchRun()

result = search.run("2024년 노벨상 수상자")
print(result)
# "2024년 노벨 물리학상은..."

# Agent에 통합
from langchain.agents import initialize_agent, Tool

tools = [
    Tool(
        name="Web Search",
        func=search.run,
        description="인터넷에서 최신 정보를 검색합니다"
    )
]

2. 계산기

from langchain.tools import Tool
import numexpr

def calculator(expression):
    """수학 계산"""
    try:
        return str(numexpr.evaluate(expression))
    except:
        return "계산 오류"

calc_tool = Tool(
    name="Calculator",
    func=calculator,
    description="수학 계산을 수행합니다. 입력: '2+2', '(10*5)/2' 등"
)

# 사용
print(calc_tool.run("(100 + 50) * 0.9"))  # "135.0"

3. 날씨 API

import requests

def get_weather(location):
    """날씨 정보 조회"""
    api_key = "your-api-key"
    url = f"http://api.weather api.com/v1/current.json"

    response = requests.get(url, params={
        "key": api_key,
        "q": location
    })

    data = response.json()
    return f"{location}: {data['current']['condition']['text']}, {data['current']['temp_c']}°C"

weather_tool = Tool(
    name="Weather",
    func=get_weather,
    description="특정 지역의 현재 날씨를 조회합니다"
)

4. 파일 읽기/쓰기

def read_file(file_path):
    """파일 읽기"""
    try:
        with open(file_path, 'r') as f:
            return f.read()
    except:
        return "파일을 읽을 수 없습니다"

def write_file(args):
    """파일 쓰기"""
    file_path, content = args.split('|')
    with open(file_path, 'w') as f:
        f.write(content)
    return f"파일 저장 완료: {file_path}"

file_tools = [
    Tool(
        name="Read File",
        func=read_file,
        description="파일 내용을 읽습니다. 입력: 파일 경로"
    ),
    Tool(
        name="Write File",
        func=write_file,
        description="파일에 내용을 씁니다. 입력: '경로|내용'"
    )
]

고급 Tools

5. SQL 데이터베이스

from langchain.utilities import SQLDatabase
from langchain.tools import QuerySQLDataBaseTool

db = SQLDatabase.from_uri("sqlite:///sales.db")

sql_tool = QuerySQLDataBaseTool(db=db)

# Agent가 자동으로 SQL 생성 및 실행
tools.append(Tool(
    name="SQL Query",
    func=sql_tool.run,
    description="SQL 쿼리를 실행하여 데이터베이스에서 정보를 가져옵니다"
))

6. Python REPL

from langchain.utilities import PythonREPL

python_repl = PythonREPL()

repl_tool = Tool(
    name="Python REPL",
    func=python_repl.run,
    description="Python 코드를 실행합니다. 데이터 분석, 계산 등에 사용"
)

# 사용 예
result = repl_tool.run("""
import pandas as pd
df = pd.DataFrame({'a': [1, 2, 3], 'b': [4, 5, 6]})
df['c'] = df['a'] + df['b']
print(df)
""")

7. Gmail

from langchain.tools.gmail import GmailSendMessage

gmail_tool = GmailSendMessage(
    api_resource=gmail_api_resource
)

tools.append(Tool(
    name="Send Email",
    func=gmail_tool.run,
    description="이메일을 발송합니다"
))

# Agent가 자동으로 이메일 작성 및 발송
# "홍길동에게 내일 회의 리마인더 메일 보내줘"

8. 파일 다운로드/업로드

import requests
from pathlib import Path

def download_file(url):
    """URL에서 파일 다운로드"""
    response = requests.get(url)
    filename = Path(url).name

    with open(filename, 'wb') as f:
        f.write(response.content)

    return f"다운로드 완료: {filename}"

def upload_to_s3(args):
    """S3에 파일 업로드"""
    import boto3

    file_path, bucket_name = args.split('|')

    s3 = boto3.client('s3')
    s3.upload_file(file_path, bucket_name, Path(file_path).name)

    return f"업로드 완료: {file_path}{bucket_name}"

file_management_tools = [
    Tool(name="Download", func=download_file, description="..."),
    Tool(name="Upload S3", func=upload_to_s3, description="...")
]

Tool 조합

멀티 도구 Agent

from langchain.agents import initialize_agent, AgentType
from langchain.chat_models import ChatOpenAI

llm = ChatOpenAI(model="gpt-4", temperature=0)

tools = [
    search_tool,
    calc_tool,
    weather_tool,
    sql_tool,
    repl_tool
]

agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# 복잡한 작업 수행
result = agent.run("""
1. 오늘 서울 날씨 확인
2. 비가 오면 우산 재고 데이터베이스에서 조회
3. 재고가 100개 이하면 발주 필요 수량 계산
4. 결과를 요약해서 보고
""")

순차 도구 사용

class SequentialToolAgent:
    def __init__(self, tools):
        self.tools = {tool.name: tool for tool in tools}

    def execute_plan(self, plan):
        """단계별 도구 실행"""
        results = {}

        for step in plan:
            tool_name = step['tool']
            args = step['args']

            # 이전 결과 참조
            if 'depends_on' in step:
                args = results[step['depends_on']]

            # 도구 실행
            result = self.tools[tool_name].run(args)
            results[step['name']] = result

        return results

# 사용
plan = [
    {"name": "weather", "tool": "Weather", "args": "서울"},
    {"name": "search", "tool": "Web Search", "args": "우산 추천", "depends_on": "weather"}
]

agent = SequentialToolAgent(tools)
results = agent.execute_plan(plan)

커스텀 Tool 만들기

기본 Tool

from langchain.tools import BaseTool

class CustomSearchTool(BaseTool):
    name = "Custom Search"
    description = "특정 웹사이트에서만 검색합니다"

    def _run(self, query: str) -> str:
        # 실제 검색 로직
        results = custom_search_api(query, domain="example.com")
        return results

    async def _arun(self, query: str) -> str:
        # 비동기 버전
        return self._run(query)

Structured Tool

from langchain.tools import StructuredTool
from pydantic import BaseModel, Field

class EmailInput(BaseModel):
    to: str = Field(description="수신자 이메일")
    subject: str = Field(description="제목")
    body: str = Field(description="본문")

def send_email_func(to: str, subject: str, body: str) -> str:
    # 이메일 발송 로직
    send_email(to, subject, body)
    return f"이메일 발송 완료: {to}"

email_tool = StructuredTool.from_function(
    func=send_email_func,
    name="Send Email",
    description="이메일을 발송합니다",
    args_schema=EmailInput
)

Tool with State

class StatefulTool(BaseTool):
    name = "Session Manager"
    description = "세션 정보를 관리합니다"

    def __init__(self):
        super().__init__()
        self.session_data = {}

    def _run(self, action: str, key: str = None, value: str = None) -> str:
        if action == "set":
            self.session_data[key] = value
            return f"저장됨: {key} = {value}"
        elif action == "get":
            return self.session_data.get(key, "없음")
        elif action == "list":
            return str(self.session_data)

실전 사용 사례

1. 개인 비서 Agent

tools = [
    Tool(name="Email", func=send_email, description="이메일 발송"),
    Tool(name="Calendar", func=add_calendar_event, description="일정 추가"),
    Tool(name="Reminder", func=set_reminder, description="리마인더 설정"),
    Tool(name="Search", func=web_search, description="웹 검색")
]

agent = initialize_agent(tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION)

# 사용
agent.run("내일 오후 3시에 팀 회의 일정 추가하고, 1시간 전에 리마인더 설정해줘")

2. 데이터 분석 Agent

tools = [
    Tool(name="SQL", func=query_database, description="DB 쿼리"),
    Tool(name="Python", func=python_repl.run, description="Python 실행"),
    Tool(name="Plot", func=create_plot, description="그래프 생성"),
    Tool(name="Export", func=export_csv, description="CSV 내보내기")
]

# "2024년 월별 매출을 분석하고 그래프로 보여줘"
# 1. SQL로 데이터 추출
# 2. Python으로 분석
# 3. Plot으로 시각화
# 4. 결과 저장

3. 콘텐츠 생성 Agent

tools = [
    Tool(name="Research", func=web_search, description="리서치"),
    Tool(name="Image", func=dalle_generate, description="이미지 생성"),
    Tool(name="Save", func=save_file, description="파일 저장"),
    Tool(name="Publish", func=publish_blog, description="블로그 발행")
]

# "AI 트렌드에 대한 블로그 글 작성해줘"
# 1. 최신 정보 검색
# 2. 글 작성
# 3. 이미지 생성
# 4. 발행

Tool 최적화

1. Tool 선택 최적화

class SmartToolSelector:
    def __init__(self, tools):
        self.tools = tools
        self.usage_stats = {tool.name: 0 for tool in tools}

    def select_tool(self, query):
        # LLM으로 적절한 도구 선택
        prompt = f"""
사용 가능한 도구:
{[f"{t.name}: {t.description}" for t in self.tools]}

질문: {query}

가장 적절한 도구는?
"""
        tool_name = llm.predict(prompt)

        self.usage_stats[tool_name] += 1
        return next(t for t in self.tools if t.name == tool_name)

2. 병렬 Tool 실행

import asyncio

async def parallel_tools(tools_and_args):
    """여러 도구를 동시에 실행"""
    tasks = [
        tool.arun(args)
        for tool, args in tools_and_args
    ]

    results = await asyncio.gather(*tasks)
    return results

# 사용
results = await parallel_tools([
    (weather_tool, "서울"),
    (weather_tool, "부산"),
    (weather_tool, "제주")
])
# 3배 빠른 실행

3. Tool 캐싱

from functools import lru_cache
import hashlib

class CachedTool(BaseTool):
    @lru_cache(maxsize=100)
    def _run(self, query: str) -> str:
        # 동일 쿼리는 캐시에서 반환
        return self.actual_run(query)

    def actual_run(self, query: str) -> str:
        # 실제 실행 로직
        return expensive_api_call(query)

보안 고려사항

1. Tool 권한 관리

class SecureTool(BaseTool):
    def __init__(self, required_permissions):
        self.required_permissions = required_permissions

    def _run(self, query: str, user_permissions: set) -> str:
        # 권한 확인
        if not self.required_permissions.issubset(user_permissions):
            return "권한 없음"

        return self.execute(query)

# 사용
admin_tool = SecureTool(required_permissions={"admin"})

2. 입력 검증

def safe_tool_wrapper(tool, input_validator):
    def wrapper(query):
        # 입력 검증
        if not input_validator(query):
            return "잘못된 입력"

        return tool.run(query)

    return wrapper

# SQL 인젝션 방지
def validate_sql_input(query):
    dangerous_keywords = ['DROP', 'DELETE', 'UPDATE']
    return not any(kw in query.upper() for kw in dangerous_keywords)

safe_sql_tool = safe_tool_wrapper(sql_tool, validate_sql_input)

결론

AI Agent Tools는:

  • 🔌 AI를 실세계와 연결
  • 🛠️ 무한한 확장 가능성
  • 🤖 자율적인 작업 수행
  • 💼 업무 자동화

시작하기:

  1. 기본 도구부터 (검색, 계산)
  2. 커스텀 도구 추가
  3. 조합하여 복잡한 작업
  4. 보안 및 최적화

Tools를 마스터하면 단순한 챗봇을 넘어 진정한 AI Agent를 만들 수 있습니다!