[AI] CrewAI에서의 Tool, Agent, Task 개념
이번 맛집 추천 시스템을 구현하며 각 개념을 어떻게 적용했는지 정리해 보았다.
CrewAI란?
Agent(에이전트) 기반 자동화 시스템을 설계하는 강력한 프레임워크로, 다양한 Tool(툴), Agent(에이전트), Task(태스크) 요소를 활용하여 작업을 분산하고 실행할 수 있다.
Tool(툴)이란?
에이전트가 특정 작업을 수행하는 데 사용하는 기능(도구)이다.
CrewAI에서는 BaseTool 클래스를 상속하여 원하는 툴을 정의할 수 있으며, 실제로 API 요청을 보내거나 데이터 가공을 수행하는 역할을 한다.
내가 작성한 Tool은 이렇게 활용했다.
| Tool | 역할 |
| GeocodingTool | Google Geocoding API를 활용하여 위치(도시명)의 좌표(위도, 경도)를 조회 |
| RestaurantBasicSearchTool | Google Maps API를 활용하여 특정 위치(좌표)의 맛집 리스트를 가져옴 (가게명, 평점, 리뷰수) |
| RestaurantFilterTool | 평점 4.0 이상 & 리뷰 500개 이상인 맛집을 필터링 |
| RestaurantDetailTool | Google Places API를 이용해 특정 식당의 세부 정보를 조회 |
| FinalRecommendationTool | 필터링된 맛집과 세부 정보를 조합하여 최종 추천 리스트를 생성 |
GeocodingTool
- 사용자 입력 (location)을 Google Geocoding API에 전달하여 해당 지역의 좌표(latitude, longitude)를 반환한다.
- geocoding_agent가 GeocodingTool을 사용하여 좌표를 조회한다.
class GeocodingTool(BaseTool):
name: str = "GeocodingTool"
description: str = (
"Google Geocoding API를 사용하여 주어진 위치의 위도와 경도를 조회합니다. "
"입력된 location 값은 변경 없이 그대로 반환합니다."
)
def _run(self, location: str) -> Dict:
url = "https://maps.googleapis.com/maps/api/geocode/json"
params = {"address": location, "key": GOOGLE_MAP_API_KEY}
try:
response = requests.get(url, params=params)
response.raise_for_status()
data = response.json()
if data.get("results"):
loc = data["results"][0]["geometry"]["location"]
coordinates = f"{loc['lat']},{loc['lng']}"
else:
coordinates = ""
except Exception as e:
coordinates = f"[GeocodingTool] Error: {str(e)}"
return {"location": location, "coordinates": coordinates}
Agent(에이전트)란?
특정 역할을 수행하는 인공지능 엔티티로, 하나 이상의 Tool을 사용하여 주어진 목표를 달성한다.
각 에이전트는 CrewAI의 핵심 요소이며, role(역할), goal(목표), backstory(설명), tools(사용할 툴), llm(사용할 AI 모델) 등의 속성을 가진다.
내가 작성한 Agent는 이렇게 활용했다.
| Agent | 역할 |
| geocoding_agent | 사용자 입력을 기반으로 좌표 조회 |
| restaurant_basic_search_agent | 특정 좌표 기반으로 맛집 리스트 가져오기 |
| restaurant_filter_agent | 평점과 리뷰 수 기준으로 맛집 필터링 |
| restaurant_detail_agent | 특정 식당의 세부 정보를 가져오기 |
| final_recommendation_agent | 최종 맛집 추천 리스트 생성 |
geocoding_agent
- Google Maps API를 이용하여 특정 위치(예: 부산)의 맛집 리스트(이름, 평점, 리뷰 수, place_id)를 가져온다.
- Task("맛집 기본 조회")에서 restaurant_basic_search_agent가 실행되며, 맛집 리스트를 가져오는 역할을 수행한다.
geocoding_agent = Agent(
role="좌표 조회 전문가",
goal="사용자가 입력한 location(예: '부산광역시')의 위도와 경도를 조회하며, location 값은 그대로 유지한다.",
backstory="""
나는 위치 데이터 전문가로, 입력된 location 값을 변경하지 않고 Google Geocoding API를 통해 좌표를 조회한다.
""",
tools=[GeocodingTool()],
llm=llm,
verbose=True,
)
Task(태스크)란?
CrewAI에서 특정 Agent가 수행해야 할 작업을 정의하는 요소이다.
각 Task는 어떤 Agent가 수행할지, 무엇을 기대하는지 (expected_output) 등의 정보를 포함한다.
내가 작성한 Task은 이렇게 활용했다.
| Task | 수행하는 Agent |
| 좌표 조회 | geocoding_agent |
| 맛집 기본 조회 | restaurant_basic_search_agent |
| 맛집 필터링 | restaurant_filter_agent |
| 세부 정보 조회 | restaurant_detail_agent |
| 최종 추천 생성 | final_recommendation_agent |
geocoding_agent
- geocoding_agent가 GeocodingTool을 사용하여 좌표를 조회하는 작업을 수행한다.
- Crew.kickoff() 실행 시 좌표 조회 → 맛집 조회 → 필터링 → 세부 정보 조회 → 최종 추천 순서대로 Task가 수행된다.
Task(
description="좌표 조회",
agent=geocoding_agent,
expected_output="{'location': '부산광역시', 'coordinates': '...'}",
)
CrewAI 전체 실행 흐름
1. 사용자가 입력한 위치(도시명)를 좌표로 변환 (geocoding_agent)
2. 해당 좌표를 이용해 맛집 기본 리스트 가져오기 (restaurant_basic_search_agent)
3. 평점 & 리뷰 수 기준으로 필터링 (restaurant_filter_agent)
4. 필터링된 식당의 세부 정보를 추가 요청 (restaurant_detail_agent)
5. 사용자 여행 일정 & 필터링 결과를 기반으로 최종 맛집 추천 생성
crew = Crew(
agents=[
geocoding_agent,
restaurant_basic_search_agent,
restaurant_filter_agent,
restaurant_detail_agent,
final_recommendation_agent,
],
tasks=[
Task("좌표 조회", agent=geocoding_agent, expected_output="{location, coordinates}"),
Task("맛집 기본 조회", agent=restaurant_basic_search_agent, expected_output="맛집 리스트"),
Task("맛집 필터링", agent=restaurant_filter_agent, expected_output="필터링된 리스트"),
Task("세부 정보 조회", agent=restaurant_detail_agent, expected_output="세부 정보"),
Task("최종 추천 생성", agent=final_recommendation_agent, expected_output="추천 리스트"),
],
verbose=True,
)
crew_result = crew.kickoff()
정리하자면 다음과 같다.
- Tool: API 요청 및 데이터 가공
- Agent: 특정 역할을 수행하는 AI 전문가
- Task: CrewAI에서 실행될 개별 작업 단위