협업 지능의 구현 – 순차적/계층적 실행
1. 실습 목표 및 개요
목표: 이번 실습의 목표는 여러 에이전트들이 협업하여 작업을 수행하는 협업 지능을 구현해 보는 것입니다. 특히 순차적 실행(Sequential Execution)과 계층적 실행(Hierarchical Execution)의 두 가지 방식을 다뤄보며, 각 방식의 개념과 구현 방법을 이해합니다.
개요: 사람의 팀워크처럼, 복잡한 과제를 해결할 때 하나의 AI보다 여러 역할의 AI 에이전트들이 협력하면 더 효과적입니다. 예를 들어, 여행 계획을 세우는 문제를 생각해봅시다. 한 에이전트가 최신 여행 뉴스를 조사하고, 다른 에이전트가 그 정보를 바탕으로 여행 일정을 만들고, 또 다른 에이전트는 여행지의 맛집 정보를 찾아 레시피를 추천한다면 각자의 전문성을 살려 더 풍부한 최종 결과를 얻을 수 있습니다. 이처럼 협업 지능은 개별 AI의 한계를 보완하고, 단계별로 일을 나눔으로써 정확하고 창의적인 결과를 끌어내는 데 중요합니다.
본 Lab에서는 CrewAI 프레임워크를 활용하여 이러한 협업 지능을 구현합니다. CrewAI를 통해 여러 **Agent(에이전트)**들이 Task(태스크) 단위로 나뉜 작업을 수행하고, 이를 **Crew(크루)**가 조율하게 됩니다. 우리는 두 가지 프로세스(작업 진행 방식)를 실습할 것입니다: 순차적 프로세스는 파이프라인처럼 작업을 차례대로 실행하고, 계층적 프로세스는 관리자 에이전트가 하위 에이전트들에게 작업을 위임하여 수행합니다.
2. 필수 라이브러리 설치 및 VS Code 실행 준비
실습을 시작하기 전에 환경을 준비하겠습니다. 이번 실습에 필요한 파이썬 라이브러리와 도구, 설정은 다음과 같습니다:
- CrewAI 프레임워크 설치: 터미널에서 pip install crewai 명령으로 CrewAI를 설치합니다. (또는 프로젝트 환경에 이미 포함되어 있다면 생략 가능)CrewAI는 다중 에이전트 협업을 지원하는 오픈소스 프레임워크입니다.
- pip install crewai
- 기타 라이브러리 설치: 뉴스 검색 등을 위해 python-dotenv (환경변수 관리)와 OpenAI API 클라이언트 등이 필요합니다. 터미널에서 다음을 실행하세요:만약 DuckDuckGoSearchRun이나 웹 스크래핑 툴을 사용하려면 해당 패키지가 crewai에 포함되어 있으므로 추가 설치는 필요하지 않습니다.
- pip install python-dotenv openai
- API 키 설정: OpenAI의 GPT API를 사용하므로, OpenAI API 키를 준비합니다. 프로젝트 루트에 .env 파일을 만들고 다음 내용을 추가하세요 (자신의 키로 대체):.env 파일은 dotenv를 통해 로드되며, CrewAI의 LLM 에이전트가 이 키를 사용해 OpenAI 모델(gpt-4o-mini)을 호출합니다. 주의: API 키는 유출되지 않도록 관리합니다.
- OPENAI_API_KEY=sk-*******************
- VS Code 에서 실행 준비: VS Code로 이 프로젝트 디렉토리를 열고, 각 실습용 파이썬 파일(05-1.scenario_sequential.py, 05-2.scenario_sequential_verbose.py, 05-3.scenario_hierarchical.py)을 작성하거나 제공된 코드를 불러옵니다.
- 파일을 열고, 주어진 코드를 그대로 붙여넣습니다 (아래 실습 절차에 코드 제공).
- 터미널을 열고 python 파일명.py 으로 실행하거나, VS Code의 Run 기능(▶️ 버튼)을 사용해 실행합니다.
- 환경변수 로드 확인: 코드를 실행하면 dotenv가 .env의 API 키를 불러옵니다. 키가 없거나 잘못되면 오류가 발생하므로, 실행 전에 .env 설정이 제대로 되었는지 확인하세요.
이제 환경이 준비되었습니다. 다음으로 실습에 등장하는 핵심 개념을 알아보고, 실제 코드를 실행하며 순차적/계층적 협업이 어떻게 동작하는지 체험해보겠습니다.
3. 핵심 개념 설명: Agent, Task, Crew, Process (순차적/계층적)
실습에 등장하는 주요 개념들을 간략히 정리합니다. 이전 4교시에서 tools.py에 대한 설명을 이미 했으므로 여기서는 도구 툴에 대해서는 짧게 언급하고 넘어가겠습니다. 이번 초점은 Agent, Task, Crew, Process 네 가지입니다:
- Agent (에이전트): 특정 역할과 목표를 가진 AI 개체를 말합니다. Agent는 자신의 역할에 맞는 전문 지식/기능과 **LLM(대형 언어 모델)**을 갖추고 있습니다. 예를 들어 뉴스 분석가 에이전트, 여행 전문가 에이전트, 요리 전문가 에이전트처럼 역할을 부여할 수 있습니다. 각 Agent에는 해결하려는 **goal(목표)**과 배경지식 역할인 backstory, 그리고 사용할 수 있는 tools(도구) 목록을 지정합니다. 도구에는 웹 검색, 웹 스크래핑, 계산기 등 작업에 도움이 되는 유틸리티가 포함됩니다. Agent는 주어진 Task를 수행하기 위해 LLM에게 프롬프트를 생성하고, 필요하면 도구를 사용해 정보를 얻어옵니다.
- Task (태스크): 에이전트가 수행할 구체적인 작업 단위를 의미합니다. Task는 사람에게 주는 지시와 비슷하며, **description(설명)**과 **expected_output(예상 출력)**으로 구성됩니다. 또한 이 Task를 담당할 **agent(에이전트)**가 지정됩니다. 예를 들어 *"최신 여행 뉴스를 찾아 요약한다"*라는 Task를 뉴스 분석가 에이전트에 할당할 수 있습니다. Task는 순서를 가지고 여러 개 구성되며, 각 Task 완료 시 결과가 생성됩니다. (필요하다면 콜백 함수를 통해 결과를 저장하거나 후속 처리할 수 있습니다 – 실습 2에서 다룸)
- Crew (크루): 여러 에이전트를 조직하고 Task들을 관리하는 컨테이너입니다. Crew는 팀 단위로 에이전트 목록과 수행할 Task 목록을 보유합니다. 또한 **Process(프로세스)**라는 속성을 통해 작업 실행 방식을 결정합니다. 쉽게 말해, Crew는 "어떤 에이전트들이 어떤 순서/방법으로 작업을 진행할 것인지"를 한 곳에서 관리합니다. crew.kickoff() 메소드를 호출하면 Crew가 정의된 프로세스에 따라 에이전트들을 동원하여 작업을 착수하고 완료합니다.
- Process (프로세스): Crew에서 작업을 진행하는 전략/방식을 의미합니다. CrewAI에서는 대표적으로 두 가지 프로세스 모드를 지원합니다:
- 순차적 프로세스 (Process.sequential) – 미리 정의된 순서대로 Task를 하나씩 실행합니다. 각 태스크가 완료된 후 다음 태스크로 넘어가며, 직렬 파이프라인 형태로 작업이 진행됩니다. 이 방식은 단계별로 결과를 이어서 사용해야 하는 경우나, 절차가 명확히 정해진 경우에 적합합니다. 예를 들어: Agent A가 정보를 수집하고 -> Agent B가 이를 분석하고 -> Agent C가 보고서를 작성하는 연속적인 흐름입니다.
- 계층적 프로세스 (Process.hierarchical) – 관리자 역할의 에이전트가 존재하며, 이 관리 에이전트가 작업을 진행하면서 필요에 따라 다른 하위 에이전트들에게 업무를 위임합니다. 즉, 상위-하위 관계의 구조로 작업이 수행됩니다. 관리자 에이전트는 전체 목표를 알고 있고, 일부 세부 작업을 자신이 직접 하기보다 적절한 다른 에이전트에게 맡깁니다. 이 방식은 문제가 복잡하여 하나의 태스크를 다시 세분화해야 할 때 유용합니다. 예를 들어: 총괄 기획자 에이전트가 프로젝트 계획 Task를 받으면, 그 안에서 조사 작업은 연구원 에이전트에게, 데이터 수집은 분석가 에이전트에게 각각 맡긴 후 결과를 취합하여 완성하는 식입니다.
도구 (Tools)에 대하여: 각 에이전트가 사용할 수 있는 도구들은 tools.py에 정의되어 있습니다. 지난 실습에서 구현한 주요 도구로는 DuckDuckGoSearchTool (웹 검색을 통해 최신 정보 찾기), WebScraperTool (URL로 주어진 웹페이지 내용 가져오기), CalculatorTool (간단한 수식 계산 수행)이 있습니다. 에이전트는 내부적으로 필요 시 이러한 도구를 호출하여 (예: 뉴스를 검색하거나, 웹페이지의 내용을 스크래핑하거나, 비용 계산을 수행) Task를 완수합니다. 이번 실습에서는 도구 구현 상세보다 에이전트 간 협업 흐름에 집중하므로, 도구 설명은 이 정도로만 하겠습니다.
위 개념들을 염두에 두고, 다음 섹션부터 실제 코드를 작성하고 실행하면서 순차적/계층적 협업 지능이 어떻게 작동하는지 단계별로 살펴보겠습니다.
4. 실습 1: scenario_sequential.py – 기본적인 순차 실행
첫 번째 실습에서는 순차적 프로세스를 사용하여 협업 에이전트들을 실행해보겠습니다. 예시 시나리오는 **"여행 계획 수립"**입니다. 세 명의 에이전트가 역할을 분담해 순차적으로 작업합니다:
- 뉴스 분석가 에이전트: 최신 여행 뉴스를 검색하고 요약 제공 (새로운 아이디어나 트렌드 수집)
- 여행 전문가 에이전트: 뉴스 요약을 참고하여 현실적인 3일 여행 일정을 계획
- 요리 전문가 에이전트: 여행지에 어울리는 현지 음식 레시피를 추천
이 세 에이전트의 작업은 앞선 결과를 다음이 활용하는 구조로 연결됩니다. 이제 05-1.scenario_sequential.py 파일의 전체 코드를 살펴보고, 순차적 실행을 구현하는 방법을 이해해봅시다.
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process
from tools import search_tool, scrape_tool, calculator_tool
load_dotenv() # .env 파일에서 API 키 등 환경변수 로드
# 뉴스 분석 Agent 정의
news_agent = Agent(
role="뉴스 분석가",
goal="최신 여행 관련 뉴스를 검색하고 요약 제공",
backstory="뉴스 분석 전문 AI 기자",
tools=[search_tool, scrape_tool], # 웹 검색 및 스크래핑 도구 사용
llm="gpt-4o-mini",
verbose=True
)
# 여행 계획 Agent 정의
travel_agent = Agent(
role="여행 전문가",
goal="뉴스 내용을 바탕으로 현실적이고 매력적인 여행 일정 구성",
backstory="다년간의 여행 플래너 경험 보유",
tools=[search_tool, scrape_tool, calculator_tool], # 필요한 경우 계산기 도구도 사용
llm="gpt-4o-mini",
verbose=True
)
# 요리 추천 Agent 정의
recipe_agent = Agent(
role="요리 전문가",
goal="여행지에서 즐길 수 있는 적절한 현지 요리 레시피 추천",
backstory="글로벌 요리 전문가 AI 셰프",
tools=[search_tool, scrape_tool], # 웹 검색 및 스크래핑 도구 사용
llm="gpt-4o-mini",
verbose=True
)
# Task 정의: 각 에이전트에게 할 일 할당
news_task = Task(
description="최근 일주일 내 국내 여행과 관련된 중요한 뉴스 1건을 한국어로 요약해줘.",
expected_output="한국어로 작성된 여행 관련 뉴스 1건의 제목, 요약, 링크",
agent=news_agent
)
travel_task = Task(
description="위에서 제공된 최신 여행 뉴스를 참고하여 3일 동안의 여행 일정을 한국어로 작성해줘.",
expected_output="한국어로 작성된 상세한 3일 여행 일정",
agent=travel_agent
)
recipe_task = Task(
description="여행 일정에 포함된 지역에서 추천할 만한 현지 음식 레시피 1개를 한국어로 제공해줘.",
expected_output="한국어로 작성된 현지 음식 이름, 재료 목록, 상세한 조리법",
agent=recipe_agent
)
# Crew 생성 및 실행 (순차적 프로세스 지정)
crew = Crew(
agents=[news_agent, travel_agent, recipe_agent],
tasks=[news_task, travel_task, recipe_task],
process=Process.sequential, # 순차적으로 Task들을 처리
verbose=True
)
def run_sequential_scenario():
result = crew.kickoff() # 순차적 프로세스 시작
print("[순차적 시나리오 최종 결과]\n", result)
if __name__ == "__main__":
run_sequential_scenario()
코드 설명: 위 코드에서는 3개의 Agent와 3개의 Task를 정의하고, 하나의 Crew에 묶었습니다. Crew(process=Process.sequential)로 설정했기 때문에 crew.kickoff()를 호출하면 다음과 같은 순서로 진행됩니다:
- 뉴스 분석가 에이전트가 news_task를 수행합니다. (description: 최신 여행 뉴스 요약)
- 에이전트는 DuckDuckGo 검색(search_tool)으로 최신 여행 뉴스를 찾아 내용을 요약합니다.
- verbose=True 설정 덕분에, 에이전트의 내부 수행 과정(프롬프트 구성이나 중간 결과)을 콘솔에 자세히 출력하게 됩니다.
- 이 Task의 결과로 예를 들어 *"최근 일주일 내 국내 여행 관련 뉴스 1건의 제목, 요약, 링크"*가 생성됩니다.
- 여행 전문가 에이전트가 travel_task를 수행합니다. (앞에서 얻은 뉴스 정보를 참고하여 3일 여행 일정 작성)
- 이 단계에서는 이전 news_task의 출력 내용을 자동으로 참조할 수 있도록, CrewAI가 context를 전달하거나 에이전트가 출력 텍스트를 인지하고 있습니다. (프레임워크가 Task 간 데이터를 전달)
- Travel 에이전트는 예를 들어 뉴스에서 언급된 관광 트렌드를 반영하여 3일치 여행 코스를 만들어냅니다.
- 여기서 calculator_tool도 할당돼 있지만, 3일 일정 작성에는 꼭 필요하지 않을 수 있습니다. (도구는 선택적으로 사용)
- 요리 전문가 에이전트가 recipe_task를 수행합니다. (여행 일정에 포함된 지역 중 하나의 현지 음식 레시피 추천)
- 이 에이전트는 여행 일정 결과를 입력 받아 해당 지역의 유명한 요리를 검색하고(search_tool), 레시피를 요약하여 제공합니다.
- 최종 출력으로는 *"현지 음식 이름 및 레시피 (재료와 조리법)"*이 한국어로 작성됩니다.
모든 Task가 끝나면 crew.kickoff()의 반환값 result에 최종 결과가 들어갑니다. 현재 구현에서는 print("[순차적 시나리오 최종 결과]\n", result)로 한꺼번에 출력하고 있습니다.
실행하기: VS Code에서 05-1.scenario_sequential.py 코드를 작성한 후 실행해보세요. 처음엔 시간이 다소 소요될 수 있습니다 (세 단계의 작업 동안 API 호출 및 웹 검색이 수행됨). 콘솔 출력에는 각 에이전트의 활동 내역이 verbose 모드로 표시되고, 마지막에 "[순차적 시나리오 최종 결과]" 이후 종합된 결과가 나타날 것입니다.
예상 출력: (실제 내용은 실행 시점의 뉴스와 에이전트 응답에 따라 달라집니다.) 결과는 뉴스 요약, 여행 일정, 요리 레시피 세 부분으로 구성됩니다. 예를 들어 요약하면 다음과 같습니다:
[순차적 시나리오 최종 결과]
제목: ○○○ 뉴스 제목
요약: 이 뉴스는 최근 국내 여행 동향으로 ○○에 대한 내용입니다...
링크: http://... (뉴스 출처 링크)
Day 1: ○○ 지역 관광지 방문 – 아침에 ○○, 점심에 ○○ 방문, 저녁에 ○○ 야경 감상
Day 2: ○○ 박물관 견학 및 ○○ 시장 투어 ...
Day 3: ... (이와 같이 3일치 상세 일정이 나열됨)
추천 음식: ○○ (해당 지역의 명물 요리)
재료: ○○, ○○, ...
조리법: 1) ○○를 손질한다. 2) ○○와 함께 볶는다. ... (이러한 형식으로 레시피 제공)
위와 같이 세 단계 결과물이 하나의 흐름으로 연결되어 최종 출력된 것을 볼 수 있습니다. 순차적 실행의 장점은 단계별로 명확한 흐름을 가지므로, 각 단계에서 이전 단계의 결과를 활용하여 응답의 일관성과 연결성을 확보할 수 있다는 점입니다. 뉴스 -> 일정 -> 레시피의 맥락 연결을 확인해보세요.
참고: 현재 scenario_sequential.py에서는 각 단계 출력이 프로그램 내 변수에 명시적으로 저장되지 않고 바로 다음 Task에 활용되고 있습니다. 만약 프로그램 내에서 각 결과를 개별적으로 다루고 싶다면 다음 실습에서 소개할 콜백 함수나 다른 저장 방법을 사용할 수 있습니다.
5. 실습 2: scenario_sequential_verbose.py – 출력 결과 저장 및 확인 (콜백 활용)
두 번째 실습에서는 순차적 실행 프로세스는 그대로 유지하면서, 중간 결과를 저장하고 활용하는 방법을 연습해봅니다. 이를 위해 Task의 콜백(callback) 함수를 사용하여 각 Task 완료 시의 출력을 파이썬 변수에 기록하도록 합니다. 이렇게 하면 모든 결과를 구조화된 형태로 저장해두고 나중에 사용할 수 있으며, 프로그램 상에서 결과를 조작하거나 검증하기에도 편리합니다.
05-2.scenario_sequential_verbose.py의 코드를 살펴보겠습니다:
from crewai import Agent, Task, Crew, Process
from tools import DuckDuckGoSearchTool, WebScraperTool, CalculatorTool
# 도구 인스턴스 생성 (tools.py의 클래스 이용)
search_tool = DuckDuckGoSearchTool()
scrape_tool = WebScraperTool()
calculator_tool = CalculatorTool()
# 결과 저장용 딕셔너리 생성
agent_results = {}
# 뉴스 분석 Agent
news_agent = Agent(
role="뉴스 분석가",
goal="최신 여행 관련 뉴스를 요약하여 전달합니다.",
backstory="다양한 언론사의 뉴스를 분석하여 최신 여행 정보를 제공합니다.",
tools=[search_tool, scrape_tool],
llm="gpt-4o-mini",
verbose=True
)
# 여행 전문가 Agent
travel_agent = Agent(
role="여행 전문가",
goal="여행 계획을 세우고 일정을 제안합니다.",
backstory="오랜 여행 기획 경력을 가진 여행 전문가입니다.",
tools=[search_tool, scrape_tool],
llm="gpt-4o-mini",
verbose=True
)
# 요리 전문가 Agent
recipe_agent = Agent(
role="요리 전문가",
goal="여행지의 현지 음식을 찾아 레시피를 제공합니다.",
backstory="전 세계의 다양한 음식을 연구하고 요리법을 전파하는 요리 연구가입니다.",
tools=[search_tool, scrape_tool],
llm="gpt-4o-mini",
verbose=True
)
# Task 정의 (각 Task에 콜백 함수를 지정하여 결과 저장)
news_task = Task(
description="최신 여행 뉴스를 검색하여 한국어로 3가지 주요 뉴스를 요약해 제공합니다.",
expected_output="최신 여행 뉴스 요약 3가지 (한국어)",
agent=news_agent,
callback=lambda output: agent_results.update({"뉴스 요약": str(output)})
)
travel_task = Task(
description="앞에서 제공된 최신 여행 뉴스 요약을 바탕으로 한국에서 3일간의 여행 일정을 상세히 작성합니다.",
expected_output="3일 여행 일정 상세 안내 (한국어)",
agent=travel_agent,
callback=lambda output: agent_results.update({"여행 일정": str(output)})
)
recipe_task = Task(
description="앞서 제시된 여행 일정의 지역 중 한 곳의 대표적인 현지 음식을 선정하고, "
"웹 검색으로 해당 음식의 레시피를 검색한 후, 한국어로 상세한 레시피를 제공합니다.",
expected_output="선택된 지역의 대표 현지 음식 레시피 (한국어)",
agent=recipe_agent,
callback=lambda output: agent_results.update({"레시피": str(output)})
)
# Crew 정의 (순차적 프로세스)
crew = Crew(
agents=[news_agent, travel_agent, recipe_agent],
tasks=[news_task, travel_task, recipe_task],
process=Process.sequential,
verbose=True
)
# Crew 실행 및 결과 출력
def run_sequential_scenario():
final_result = crew.kickoff()
print("\n--- 최종 결과 출력 ---\n")
for key, value in agent_results.items():
print(f"\n✅ {key}\n{value}\n")
if __name__ == "__main__":
run_sequential_scenario()
코드 설명: 이번 코드에서는 몇 가지 변화를 주었습니다.
- agent_results = {} 딕셔너리를 만들어 놓고, 각 Task 정의 시 callback 매개변수를 설정했습니다. Task(..., callback=lambda output: agent_results.update({...})) 형태로 람다 함수를 지정하면, 해당 Task를 에이전트가 완료할 때 자동으로 이 콜백 함수가 호출되어 output 결과를 우리 딕셔너리에 저장하게 됩니다.
예를 들어 news_task에는 "뉴스 요약" 키로 결과 문자열을 저장하고, travel_task는 "여행 일정", recipe_task는 "레시피"라는 키로 각각 저장하도록 했습니다. (str(output)로 형 변환하는 이유는 혹시 output이 객체일 경우 문자열로 저장하기 위함입니다.) - 각 에이전트 정의에서 tools 목록에 계산기가 빠져있는데, 이는 이번 Task에서는 계산 작업이 없어 간소화한 것입니다. 필요에 따라 tools는 가감해도 무방합니다. (뉴스, 여행 일정, 레시피 추천에는 검색과 스크래핑 정도면 충분하다는 판단)
- crew.kickoff() 후에 이전처럼 바로 결과를 출력하는 대신, agent_results 딕셔너리에 모인 결과를 순회하며 예쁘게 출력합니다. print(f"\n✅ {key}\n{value}\n")를 통해 각 섹션별로 결과를 구분해 보여줍니다. 앞에 ✅ 아이콘을 넣어 가독성을 높였습니다.
이제 이 코드를 실행하면, 이전 실습과 기능적으로는 동일한 작업을 수행하지만 프로그램 내에 모든 중간 산출물이 기록됩니다. 그리고 최종적으로 아래와 같이 결과를 구분하여 확인할 수 있습니다:
실행 및 결과: 05-2.scenario_sequential_verbose.py 파일을 실행하세요. 완료되면 콘솔에서 최종 결과 부분이 다음과 같이 출력됩니다:
--- 최종 결과 출력 ---
✅ 뉴스 요약
... (최신 여행 뉴스 3건에 대한 요약 텍스트) ...
✅ 여행 일정
... (뉴스를 반영한 3일간의 여행 일정 텍스트) ...
✅ 레시피
... (여행지 관련 현지 음식 한 가지에 대한 레시피 텍스트) ...
위와 같이 각 결과에 제목을 붙여 구분해서 볼 수 있으므로, 순차적 협업 결과물을 한눈에 파악하기 쉽습니다. 예를 들어 뉴스 요약 섹션에는 3개의 주요 뉴스 제목과 요약이 목록 형태로 나올 것이고, 여행 일정 섹션에는 Day 1, Day 2, Day 3로 구분된 계획이, 레시피 섹션에는 음식 이름과 조리법 등이 나올 것입니다.
이 접근의 장점은 각 단계의 출력을 개별 변수로 활용할 수 있다는 것입니다. 예를 들어 나중에 여행 일정 내용(agent_results["여행 일정"])을 다른 함수에 전달하거나, 레시피(agent_results["레시피"])를 파일로 저장하는 등 응용이 가능합니다. 또한 콜백 함수를 사용하면 Task가 끝날 때마다 후처리를 자동으로 할당할 수 있어 유용합니다. (로그 저장, 추가 가공 등)
추가 팁: CrewAI의 verbose=True 설정으로 인해 에이전트 내부의 진행 로그도 출력됩니다. 이 로그에는 에이전트가 만든 프롬프트나 도구 사용 내역 등이 포함될 수 있어 디버깅에 도움이 됩니다. 필요에 따라 verbose를 False로 하면 출력이 조용해집니다. 또한 final_result = crew.kickoff()의 final_result에는 마지막 태스크의 결과가 저장되어 있습니다. 여기서는 모든 결과를 개별로 보고자 딕셔너리를 썼지만, final_result를 출력하면 마지막 레시피 에이전트의 출력(레시피 텍스트)만 볼 수 있다는 점 참고하세요.
지금까지 순차적 실행을 기본과 콜백 활용 두 가지 방식으로 실습해보았습니다. 다음으로는 보다 복잡한 계층적 실행 방식을 적용한 예제를 실습해보겠습니다.
6. 실습 3: scenario_hierarchical.py – 계층적 실행의 개념과 적용
이제 계층적 프로세스를 사용하는 예제로 넘어가겠습니다. 계층적 협업에서는 한 에이전트가 **매니저(관리자)**처럼 동작하여, 필요한 경우 다른 에이전트들에게 작업 일부를 **위임(delegation)**합니다. 이번 시나리오에서도 여행 계획 주제를 이어가되, 설정을 조금 바꾸어 보겠습니다:
- 총괄 여행 플래너 에이전트: 관리 역할. 최종 목표는 *"최신 트렌드를 반영한 서울 근교 1박 2일 여행 일정과 각 여행지에 어울리는 현지 음식 레시피를 모두 포함한 계획서"*를 작성하는 것입니다. 이 에이전트는 직접 전체 일정을 작성하면서도, 세부 사항(여행지 추천이나 음식 추천)은 다른 전문가들의 도움을 받을 수 있습니다.
- 여행 전문가 에이전트: 서브 역할. 최신 여행 트렌드를 조사하고, 서울 근교의 인기 여행지와 일정을 제안합니다.
- 요리 전문가 에이전트: 서브 역할. 여행 전문가가 추천한 장소에 어울리는 현지 음식 및 레시피를 제안합니다.
위 구조에서 총괄 여행 플래너가 **관리자(agent_manager)**로 지정되며, allow_delegation=True 속성을 줘서 다른 에이전트를 호출할 권한을 부여합니다. 이렇게 하면 플래너 에이전트는 자신의 Task를 수행하는 도중 필요 시 travel_agent나 culinary_agent를 활용할 수 있습니다. (CrewAI 프레임워크가 이러한 상호작용을 지원)
05-3.scenario_hierarchical.py 코드를 살펴보겠습니다:
from crewai import Agent, Task, Crew, Process
from tools import DuckDuckGoSearchTool, WebScraperTool, CalculatorTool
# 도구 인스턴스 생성
search_tool = DuckDuckGoSearchTool()
scrape_tool = WebScraperTool()
calculator_tool = CalculatorTool()
# 총괄 여행 플래너 (관리자 Agent)
planner_agent = Agent(
role="총괄 여행 플래너",
goal="최적의 서울 근교 1박 2일 여행 일정과 추천 음식을 종합하여 최종 여행 계획서를 작성",
backstory="10년 경력의 베테랑 여행 컨설턴트로 다양한 분야의 의견을 취합하여 여행 계획을 완성합니다.",
allow_delegation=True, # 다른 에이전트에게 업무 위임을 허용
llm="gpt-4o-mini",
verbose=True
)
# 여행 전문가 Agent (여행 일정 추천 담당)
travel_agent = Agent(
role="여행 전문가",
goal="최신 여행 트렌드를 조사하여 서울 근교의 인기 있는 여행지와 일정을 제안",
backstory="국내 여행지에 대해 잘 알고 있는 전문가로, 최근의 여행 트렌드를 바탕으로 관광지를 추천합니다.",
tools=[search_tool],
llm="gpt-4o-mini",
verbose=True
)
# 요리 전문가 Agent (음식 추천 담당)
culinary_agent = Agent(
role="요리 전문가",
goal="추천된 여행지와 잘 어울리는 현지 음식 및 레시피를 추천",
backstory="국내 각 지역의 음식 문화와 레시피에 능통한 전문가로, 여행지에 어울리는 음식을 추천합니다.",
tools=[search_tool],
llm="gpt-4o-mini",
verbose=True
)
# Task 정의 (총괄 플래너에게 최종 여행 계획서 작성 지시)
planner_task = Task(
description=(
"국내 최신 여행 트렌드가 반영된 서울 근교의 1박 2일 여행 일정을 작성하고, "
"각 여행지와 잘 어울리는 현지 음식과 레시피를 포함하여 여행 계획서를 한국어로 작성해주세요."
),
expected_output=(
"최신 여행 트렌드를 반영한 서울 근교 1박 2일 여행 일정과 "
"각 여행지의 현지 음식 및 간단한 레시피를 포함한 한국어 여행 계획서"
),
agent=planner_agent
)
# Crew 구성 (계층적 프로세스 사용)
crew = Crew(
agents=[travel_agent, culinary_agent], # 하위 실행에 참여할 에이전트들 (관리자 제외)
tasks=[planner_task],
process=Process.hierarchical,
manager_agent=planner_agent, # 총괄 여행 플래너를 매니저로 지정
verbose=True
)
# Crew 실행
if __name__ == "__main__":
result = crew.kickoff()
print("\n\n📗 최종 여행 계획서:\n")
print(result)
코드 설명: 이 코드에서는 핵심적으로 manager_agent(관리자 에이전트) 설정과 allow_delegation 사용에 주목해야 합니다.
- planner_agent (총괄 여행 플래너)를 정의할 때 allow_delegation=True로 설정했습니다. 이것이 계층적 협업의 열쇠입니다. 이 속성이 True이면, 이 에이전트는 자신이 맡은 Task를 수행하는 중간에 다른 에이전트들에게 도움을 요청할 수 있습니다. (CrewAI 내부적으로 프롬프트를 해석해 delegation이 필요하면 travel_agent나 culinary_agent에게 sub-task를 전달하고, 그 결과를 받아 이어서 진행합니다.)
- Crew 생성 시 process=Process.hierarchical로 지정하고, manager_agent=planner_agent로 설정했습니다. 이렇게 하면 Crew는 planner_agent를 중심으로 작업을 수행합니다. agents=[travel_agent, culinary_agent] 리스트에는 플래너가 활용할 부하(agent)들을 넣습니다. (플래너 자신은 manager_agent로 따로 지정했으므로, agents 리스트에 중복으로 넣지 않습니다.)
- 오직 하나의 Task인 planner_task만 정의했습니다. 이 Task가 바로 플래너에게 내려지는 최종 목표입니다. description을 보면 1박 2일 여행 일정과 각 장소의 음식/레시피까지 통합한 계획서를 작성하라고 요구하고 있습니다. 한 에이전트가 처리하기에는 광범위하지만, 플래너 에이전트는 다른 에이전트들과 협력하여 이 Task를 완수할 것입니다.
실제 실행 시에는 다음과 같은 협업 흐름이 일어납니다 (내부적으로):
- 플래너 에이전트가 Task를 받고 우선 전체 구조를 잡습니다. 예를 들어 "Day1, Day2 여행 코스, 그리고 각 Day에 음식 추천을 포함해야겠다"고 계획할 것입니다. 그러다가 각 Day에 어떤 여행지를 넣을지 고민되면, travel_agent를 호출하여 최신 트렌드 기반 추천을 얻을 수 있습니다. (플래너의 LLM은 allow_delegation=True이므로 답변 중에 travel_agent에게 질문하는 형식으로 지시를 내릴 수 있고, CrewAI가 이를 인식해 travel_agent를 실행합니다.)
- 여행 전문가 에이전트는 플래너로부터 요청을 받아 (예: "서울 근교 인기 여행지 추천") 자신의 역할에 맞게 정보를 수집/생성합니다. 아마 search_tool을 사용해 최신 인기 장소를 찾고 1박2일 코스를 제안해 플래너에게 반환할 것입니다.
- 플래너 에이전트는 받은 여행지 제안들을 토대로 일정의 뼈대를 작성합니다. 이제 각 여행지에 맞는 음식 정보를 채우기 위해 요리 전문가 에이전트에게 문의할 것입니다. (예: "이런 장소들이 있는데, 각각 어울리는 지역 음식과 간단한 레시피 알려줘.")
- 요리 전문가 에이전트는 역시 search_tool 등을 활용해 각 장소별 대표 음식을 찾아내고, 간략한 조리법이나 설명을 만듭니다. 이를 플래너에게 전달합니다.
- 플래너 에이전트는 최종적으로 여행 일정과 음식 정보를 모두 모아 종합적인 여행 계획서를 완성합니다.
이 모든 과정이 crew.kickoff() 한 번으로 계층적으로 이루어집니다. 중간에 플래너 ↔ 여행 전문가, 플래너 ↔ 요리 전문가 간의 상호작용이 일어나며, verbose 모드를 켰다면 이 과정에서 주고받는 메시지 등이 콘솔에 표시될 것입니다.
실행하기: 05-3.scenario_hierarchical.py 파일을 실행해보세요. 수행 시간은 순차적 사례보다 길 수 있습니다. (플래너 에이전트가 내부적으로 여러 번 서브에이전트를 호출하며 대화할 수 있으므로) 실행이 완료되면 최종적으로 플래너가 작성한 완성된 여행 계획서가 출력됩니다.
예상 출력: 결과는 하나의 완결된 여행 계획 문서 형태입니다. 예를 들어 구조는 다음과 같을 수 있습니다 (내용은 생략):
📗 최종 여행 계획서:
1일차 - 최신 트렌드를 반영한 여행 코스
- 방문지: ○○ (설명: ... 최신 인기 장소로 ... )
- 추천 음식: ○○ (해당 지역의 명물 음식)
- 간략한 레시피: ... (어떻게 만드는지 한두 문장으로)
2일차 - 이어지는 여행 코스
- 방문지: △△ (설명: ... 전날과 다른 매력의 장소 ...)
- 추천 음식: △△ (두번째 날 장소에 어울리는 음식)
- 간략한 레시피: ...
(마무리 인사말 혹은 여행 팁 등이 포함될 수도 있음)
위 결과는 플래너 에이전트가 여러 전문가의 도움을 받아 작성했지만, 겉보기에는 하나의 AI가 작성한 것처럼 일관된 문서입니다. 내용 중에 각 일차별로 여행지 설명과 음식 정보가 함께 어우러져 있어, 순차적 프로세스로는 얻기 어려운 교차 통합 정보를 보여줍니다. 예컨대 1일차 여행지 설명 바로 아래에 그 장소의 음식 이야기가 붙어있는 등, 협업을 통해 얻은 정보를 한 데 모아 표현합니다.
계층적 실행의 장점은 유연성에 있습니다. 매니저 에이전트의 판단으로 작업 순서를 유동적으로 조절하고, 필요한 정보를 적시에 요청하여 받을 수 있습니다. 또한 하위 에이전트들을 병렬적으로도 활용할 수 있으므로 (예: Day1, Day2 음식 정보를 동시에 요청) 효율적입니다. 다만 구현이 복잡해지고, 에이전트 사이의 커뮤니케이션 로직을 잘 설계해야 한다는 점이 도전입니다. CrewAI가 이러한 부분을 프레임워크 차원에서 지원해주고 있으므로, 우리는 역할과 목표만 잘 설정하면 나머지는 AI들 간 대화로 풀어나갈 수 있습니다.
비고: 계층적 프로세스에서는 디버깅을 위해 에이전트 간 주고받는 메시지 로그를 확인하는 것이 중요합니다. verbose 모드를 True로 했을 때 출력되는 내용을 살펴보면, 플래너 에이전트가 어떤 프롬프트로 travel_agent를 호출했고 travel_agent의 응답이 무엇이었는지 등을 볼 수 있습니다. 이를 통해 에이전트들이 의도대로 협력하고 있는지 검증할 수 있습니다.
7. 수업 마무리 및 확장 아이디어
이번 Lab에서는 협업 지능을 구현하는 두 가지 방법, 순차적 실행과 계층적 실행을 실습했습니다. 복습해보면:
- 순차적 프로세스: 정해진 순서에 따라 에이전트들이 차례로 작업하여 최종 결과를 도출. (뉴스 -> 일정 -> 레시피 같이 일렬진행)
- 계층적 프로세스: 한 에이전트가 조율자가 되어 필요에 따라 다른 에이전트를 동시에 혹은 부분적으로 투입하여 작업을 완수. (플래너가 여행/요리 전문가와 동시협업)
두 방식 모두 장단점이 있으며, 해결하려는 문제의 성격에 맞게 선택하거나 혼합할 수 있습니다. 중요한 것은, 역할을 나누고 협력함으로써 개별 LLM의 한계를 극복하고 보다 복잡한 요구사항을 만족시킬 수 있다는 점입니다.
확장 아이디어: 배운 내용을 바탕으로, 여러분만의 협업 시나리오를 확장해보는 것도 권장합니다. 예를 들어, 우리가 다룬 여행 계획 예시에 다음과 같은 기능을 추가해볼 수 있습니다:
- 날씨 정보 통합: 여행 일정을 제시할 때 해당 날짜의 날씨 예보를 함께 알려주도록 해볼 수 있습니다. 이를 위해 날씨 정보를 제공하는 새로운 에이전트를 추가하거나, 기존 여행 플래너/전문가 에이전트가 날씨 API를 호출하는 도구(예: Weather API Tool)를 사용하게 할 수 있습니다. 이렇게 하면 "Day1: ○○ 관광 (맑음, 최고 25℃)" 처럼 더 현실감 있는 일정표를 만들 수 있습니다.
- 비용 계산 추가: 여행 일정에 예상 비용 산출 단계를 넣을 수 있습니다. 예를 들어 숙박비, 교통비, 식비 등을 계산해주는 예산 담당 에이전트를 만들고, CalculatorTool을 활용하여 총합을 계산하게 할 수 있습니다. 순차적 프로세스에서는 뉴스 -> 일정 -> 예산 계산 -> 음식 추천 순으로 Task를 구성할 수 있고, 계층적 프로세스에서는 플래너가 일정 작성 후 예산 에이전트에 비용 계산을 위임받아 다시 결과를 통합하는 방식도 가능하겠습니다.
- 기타 전문 에이전트: 여행 주제 이외에도 응용 가능성이 무궁무진합니다. 예를 들어 개인 비서 시나리오를 만든다면 일정 관리 에이전트, 이메일 초안 작성 에이전트, 데이터 검색 에이전트 등이 협업하도록 할 수 있습니다. 또는 소프트웨어 개발을 예로 들어 문제 분석 에이전트, 코드 생성 에이전트, 테스트 에이전트가 순차/계층적으로 같이 일할 수도 있습니다.
마지막으로, 협업 지능 시스템을 설계할 때는 프롬프트 설계와 역할 분담이 매우 중요함을 기억하세요. 에이전트들에게 적절한 역할과 목표를 주고, 상호 작용을 잘 설정하면 기대 이상의 강력한 결과를 얻을 수 있지만, 반대로 역할이 모호하거나 프로세스 구조가 맞지 않으면 비효율이 발생할 수 있습니다. 이번 실습의 경험을 토대로 작은 실험들을 계속 해보시면 감을 잡는 데 도움이 될 것입니다.
이상으로 실습을 마칩니다. 순차적/계층적 협업 지능 구현을 직접 해보았으므로, 이를 응용하여 더욱 흥미롭고 유용한 AI 에이전트 협업 시나리오를 만들어보세요! 🚀
'HRDI_AI > [인공지능] AI 멀티 에이전트 서비스 개발을 위한 CrewAI' 카테고리의 다른 글
| 4. 외부 도구(Tools) 연동하기 (0) | 2025.05.30 |
|---|---|
| 3. Agent 및 Task 설계, 프롬프트 작성법 (0) | 2025.05.30 |
| 2. CrewAI 핵심 개념 (Agent, Task, Crew) (0) | 2025.05.30 |
| 1. 개발 환경 준비 (0) | 2025.05.30 |