AI 에이전트 Multi-Agent 협업 완벽 가이드

여러 AI 에이전트가 협업하여 복잡한 문제를 해결하는 Multi-Agent 시스템의 핵심 개념과 실무 구현 방법을 알아봅니다. CrewAI, AutoGen 등 최신 프레임워크 활용법과 실전 아키텍처 패턴을 다룹니다.

Multi-Agent 시스템 기본 개념
Agent Role과 Goal 설계
CrewAI 기본 구조
Sequential Task 실행
Parallel Task 실행
Agent 간 통신과 메모리 공유
Tool Sharing과 Delegation
AutoGen Framework
Consensus와 Voting
Multi-Agent 오케스트레이션
에러 핸들링과 복구 전략
실전 Multi-Agent 아키텍처

1. Multi-Agent 시스템 기본 개념

시작하며

여러분이 대규모 프로젝트를 혼자 진행할 때, 리서치도 하고 코드도 짜고 문서도 작성하느라 정신없었던 경험 있으시죠? 한 명이 모든 것을 처리하려니 효율도 떨어지고 전문성도 부족해집니다.

이런 문제는 AI 에이전트 시스템에서도 똑같이 발생합니다. 하나의 거대한 에이전트가 모든 작업을 처리하려면 프롬프트도 복잡해지고, 각 작업의 품질도 저하됩니다.

게다가 특정 작업이 실패하면 전체 프로세스가 중단되는 취약점도 생깁니다. 바로 이럴 때 필요한 것이 Multi-Agent 시스템입니다.

여러 전문 에이전트가 각자의 역할을 수행하면서 협업하여, 복잡한 문제를 효율적으로 해결할 수 있습니다.

개요

간단히 말해서, Multi-Agent 시스템은 각기 다른 역할과 전문성을 가진 여러 AI 에이전트가 협력하여 복잡한 작업을 수행하는 아키텍처입니다. 실무에서는 단일 LLM 호출로 해결할 수 없는 복잡한 워크플로우가 많습니다.

예를 들어, 시장 조사를 기반으로 블로그 글을 작성하고 SEO 최적화까지 해야 하는 경우, 리서처 에이전트가 데이터를 수집하고, 작가 에이전트가 글을 쓰고, 편집자 에이전트가 검토하는 식으로 역할을 분담할 수 있습니다. 기존에는 하나의 긴 프롬프트에 모든 지시사항을 넣어야 했다면, 이제는 각 에이전트가 자신의 전문 영역에만 집중하면서 이전 에이전트의 결과를 받아서 작업할 수 있습니다.

Multi-Agent 시스템의 핵심 특징은 모듈화, 전문화, 확장성입니다. 각 에이전트는 독립적으로 개선할 수 있고, 필요에 따라 새로운 에이전트를 추가하기도 쉽습니다.

이러한 특징들이 복잡한 AI 워크플로우를 관리 가능한 수준으로 만들어줍니다.

코드 예제

from crewai import Agent, Task, Crew

# 리서처 에이전트 - 정보 수집 전문
researcher = Agent(
    role='Research Analyst',
    goal='Find accurate and relevant information',
    backstory='Expert at gathering and analyzing data',
    verbose=True
)

# 작가 에이전트 - 콘텐츠 작성 전문
writer = Agent(
    role='Content Writer',
    goal='Create engaging content based on research',
    backstory='Skilled writer with SEO expertise',
    verbose=True
)

# Crew 구성 - 에이전트 팀 생성
crew = Crew(
    agents=[researcher, writer],
    verbose=True
)

설명

이것이 하는 일: 이 코드는 CrewAI 프레임워크를 사용하여 두 개의 전문 에이전트로 구성된 협업 시스템을 구축합니다. 첫 번째로, 각 Agent 객체를 생성할 때 role, goal, backstory를 정의합니다.

role은 에이전트의 직책이나 전문 분야를 나타내고, goal은 이 에이전트가 달성하고자 하는 목표입니다. backstory는 에이전트의 배경 설정으로, LLM이 이 맥락을 이해하고 해당 역할에 맞게 행동하도록 돕습니다.

verbose=True는 에이전트의 사고 과정을 로그로 출력하여 디버깅을 쉽게 만듭니다. 그 다음으로, Crew 객체를 생성하면서 agents 리스트에 여러 에이전트를 등록합니다.

Crew는 에이전트들의 오케스트레이터 역할을 하며, 작업을 분배하고 결과를 취합합니다. 각 에이전트는 독립적으로 LLM을 호출할 수 있으며, 이전 에이전트의 출력을 입력으로 받을 수 있습니다.

마지막으로, 이렇게 구성된 Crew는 나중에 Task들을 할당받아 실행하게 됩니다. researcher가 먼저 정보를 수집하고, 그 결과를 writer가 받아서 콘텐츠를 작성하는 파이프라인이 자동으로 형성됩니다.

여러분이 이 코드를 사용하면 각 에이전트의 프롬프트를 분리하여 관리할 수 있고, 특정 단계만 개선하거나 교체하기가 쉬워집니다. 또한 각 에이전트의 출력을 개별적으로 모니터링하고 평가할 수 있어 전체 시스템의 품질 관리가 용이합니다.

실전 팁

💡 각 에이전트의 role과 goal은 구체적이고 명확하게 작성하세요. 모호한 역할 정의는 에이전트의 출력 품질을 저하시킵니다.

💡 초기 개발 시에는 반드시 verbose=True로 설정하여 각 에이전트의 사고 과정을 확인하세요. 예상치 못한 동작을 빠르게 발견할 수 있습니다.

💡 에이전트의 backstory는 단순한 설명이 아니라 LLM의 행동을 유도하는 프롬프트 엔지니어링 요소입니다. 원하는 출력 스타일이나 전문성 수준을 backstory에 녹여내세요.

💡 너무 많은 에이전트를 한 Crew에 넣으면 조율 비용이 증가합니다. 처음에는 2-3개의 핵심 에이전트로 시작하여 검증한 후 확장하세요.

💡 각 에이전트를 독립적으로 테스트할 수 있는 단위 테스트를 작성하세요. 전체 Crew를 실행하기 전에 개별 에이전트의 동작을 검증하면 디버깅 시간을 크게 줄일 수 있습니다.

2. Agent Role과 Goal 설계

시작하며

여러분이 팀 프로젝트에서 각 팀원의 역할이 불명확해서 업무가 중복되거나 누락되는 상황을 겪어본 적 있나요? "이거 누가 하는 거였지?"라는 질문이 나오는 순간, 프로젝트는 이미 비효율의 늪에 빠진 것입니다.

AI 에이전트 시스템도 마찬가지입니다. 각 에이전트의 역할과 목표가 명확하지 않으면 에이전트들이 같은 작업을 반복하거나, 중요한 작업을 아무도 하지 않는 문제가 발생합니다.

특히 LLM 기반 에이전트는 명시적인 지시가 없으면 자신의 역할을 벗어나 다른 일을 시도하기도 합니다. 바로 이럴 때 필요한 것이 명확한 Role과 Goal 설계입니다.

각 에이전트의 책임 범위와 달성 목표를 구체적으로 정의하면, 협업의 효율성이 극대화됩니다.

개요

간단히 말해서, Agent의 Role은 "무엇을 하는 사람인가"를 정의하고, Goal은 "무엇을 달성해야 하는가"를 명시하는 것입니다. 실무에서는 에이전트의 역할 설계가 전체 시스템의 성공을 좌우합니다.

예를 들어, 고객 지원 시스템을 구축할 때 "문의 분류 에이전트", "기술 지원 에이전트", "에스컬레이션 에이전트"처럼 각자의 전문 영역을 명확히 나누면, 각 에이전트가 자신의 강점에 집중할 수 있습니다. 기존에는 "고객 지원 AI"라는 하나의 범용 에이전트를 만들었다면, 이제는 각 단계별로 최적화된 전문 에이전트를 배치하여 전체 품질을 높일 수 있습니다.

Role 설계의 핵심은 단일 책임 원칙(Single Responsibility Principle)을 따르는 것이고, Goal 설계의 핵심은 측정 가능하고 구체적인 목표를 설정하는 것입니다. 또한 에이전트의 제약 사항(constraints)도 함께 정의하여 역할을 벗어나지 않도록 해야 합니다.

이러한 명확한 정의가 에이전트 간 협업의 품질을 결정합니다.

코드 예제

from crewai import Agent

# 데이터 분석 전문 에이전트
data_analyst = Agent(
    role='Senior Data Analyst',
    goal='Extract insights from data and identify trends',
    backstory="""You are an expert data analyst with 10 years of experience.
    You excel at finding patterns in complex datasets and presenting
    actionable insights. You never make assumptions without data.""",
    verbose=True,
    allow_delegation=False  # 다른 에이전트에게 작업 위임 금지
)

# 비즈니스 전략 에이전트
strategist = Agent(
    role='Business Strategist',
    goal='Create actionable business recommendations based on data insights',
    backstory="""You are a strategic thinker who transforms data insights
    into concrete business actions. You consider market conditions, risks,
    and implementation feasibility.""",
    verbose=True,
    allow_delegation=True  # 필요시 다른 에이전트에게 위임 가능
)

설명

이것이 하는 일: 이 코드는 두 개의 전문 에이전트를 생성하면서 각자의 역할, 목표, 배경을 명확히 구분하여 정의합니다. 첫 번째로, data_analyst는 데이터 분석에만 집중하도록 설계되었습니다.

role을 'Senior Data Analyst'로 설정하여 LLM이 시니어 수준의 분석 능력을 발휘하도록 유도하고, goal을 "Extract insights from data and identify trends"로 명시하여 데이터에서 인사이트를 추출하는 것이 주 임무임을 분명히 합니다. backstory에서 "never make assumptions without data"라고 명시하여 데이터 기반 의사결정만 하도록 제약을 걸었고, allow_delegation=False로 설정하여 이 에이전트가 다른 에이전트에게 작업을 넘기지 못하도록 했습니다.

그 다음으로, strategist는 비즈니스 전략 수립에 특화되어 있습니다. 이 에이전트는 data_analyst의 분석 결과를 받아서 실행 가능한 비즈니스 액션으로 전환하는 역할을 합니다.

backstory에서 "market conditions, risks, and implementation feasibility"를 고려하도록 명시하여 단순한 제안이 아닌 실현 가능한 전략을 만들도록 유도합니다. allow_delegation=True로 설정하여 필요시 다른 에이전트(예: 시장 조사 에이전트)에게 추가 작업을 요청할 수 있습니다.

마지막으로, 이 두 에이전트의 역할 분리는 분석과 전략이라는 서로 다른 사고방식을 요구하는 작업을 각각의 전문 에이전트가 처리하도록 합니다. 데이터 분석은 객관적이고 사실 기반이어야 하고, 전략 수립은 창의적이고 미래 지향적이어야 하므로, 이 둘을 분리하면 각 단계의 품질이 높아집니다.

여러분이 이 패턴을 사용하면 각 에이전트의 프롬프트를 최적화하기 쉬워지고, 특정 단계에서 문제가 발생했을 때 해당 에이전트만 수정하면 됩니다. 또한 allow_delegation을 적절히 설정하여 에이전트 간 작업 흐름을 제어할 수 있습니다.

실전 팁

💡 backstory는 단순한 설명이 아니라 에이전트의 행동 패턴을 정의하는 프롬프트입니다. "You never...", "You always..."와 같은 제약 조건을 명시하여 일관된 출력을 보장하세요.

💡 allow_delegation은 신중하게 설정하세요. 무분별한 위임은 작업 흐름을 예측 불가능하게 만들고, 과도한 LLM 호출로 비용을 증가시킵니다.

💡 role 이름에 시니어티 레벨(Junior, Senior, Lead 등)을 포함하면 LLM의 출력 품질과 자신감 수준이 달라집니다. 실험을 통해 최적의 레벨을 찾으세요.

💡 goal은 동사로 시작하는 구체적인 행동 문장으로 작성하세요. "Analyze data" 대신 "Extract actionable insights from sales data to identify growth opportunities"처럼 구체적으로 작성하면 더 나은 결과를 얻습니다.

💡 여러 에이전트가 비슷한 역할을 하고 있다면 역할 설계를 재검토하세요. 각 에이전트는 명확히 구분되는 고유한 가치를 제공해야 합니다.

3. CrewAI 기본 구조

시작하며

여러분이 여러 마이크로서비스를 관리할 때, 각 서비스는 잘 작동하는데 이들을 연결하고 조율하는 게 더 어려웠던 경험 있으신가요? 각 서비스의 입출력 형식을 맞추고, 실행 순서를 정하고, 에러를 처리하는 통합 레이어가 필요합니다.

Multi-Agent 시스템도 똑같은 문제를 겪습니다. 각 에이전트는 잘 만들어졌지만, 이들을 어떤 순서로 실행할지, 어떻게 데이터를 전달할지, 에러가 발생하면 어떻게 처리할지 등 오케스트레이션이 복잡합니다.

직접 이 모든 로직을 구현하면 비즈니스 로직보다 인프라 코드가 더 많아지죠. 바로 이럴 때 필요한 것이 CrewAI 같은 Multi-Agent 프레임워크입니다.

에이전트 정의, 작업 할당, 실행 흐름 제어를 선언적으로 관리할 수 있어 복잡도를 크게 줄여줍니다.

개요

간단히 말해서, CrewAI는 Agent, Task, Crew라는 세 가지 핵심 개념으로 Multi-Agent 시스템을 구조화하는 프레임워크입니다. 실무에서는 복잡한 AI 워크플로우를 안정적으로 운영하기 위해 검증된 프레임워크가 필수적입니다.

예를 들어, 콘텐츠 생성 파이프라인을 구축할 때 리서치, 작성, 편집, SEO 최적화 단계를 각각의 Task로 정의하고, 각 Task를 담당할 Agent를 할당한 후, Crew가 전체 실행을 관리하도록 할 수 있습니다. 기존에는 각 에이전트 호출을 직접 코딩하고 결과를 수동으로 전달해야 했다면, 이제는 Task 간 의존성만 선언하면 CrewAI가 자동으로 실행 순서를 결정하고 데이터를 전달합니다.

CrewAI의 핵심 특징은 선언적 구성(declarative configuration), 자동 오케스트레이션, 내장된 에러 처리입니다. Task 간 의존성 그래프를 자동으로 해석하여 최적의 실행 순서를 결정하고, 특정 Task가 실패해도 재시도하거나 우회할 수 있습니다.

이러한 추상화가 개발자가 비즈니스 로직에 집중할 수 있게 해줍니다.

코드 예제

from crewai import Agent, Task, Crew, Process

# 에이전트 정의
researcher = Agent(
    role='Researcher',
    goal='Gather comprehensive information',
    backstory='Expert research analyst',
    verbose=True
)

writer = Agent(
    role='Writer',
    goal='Create engaging content',
    backstory='Professional content writer',
    verbose=True
)

# 작업 정의
research_task = Task(
    description='Research the topic: {topic}',
    agent=researcher,
    expected_output='Detailed research report'
)

writing_task = Task(
    description='Write an article based on the research',
    agent=writer,
    expected_output='Well-written article',
    context=[research_task]  # research_task의 출력을 입력으로 사용
)

# Crew 구성 및 실행
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,  # 순차 실행
    verbose=True
)

# 실행
result = crew.kickoff(inputs={'topic': 'AI Agent Collaboration'})

설명

이것이 하는 일: 이 코드는 CrewAI의 핵심 구조를 사용하여 리서치부터 글쓰기까지의 완전한 워크플로우를 정의하고 실행합니다. 첫 번째로, Agent 객체들을 생성하여 각 작업을 담당할 에이전트를 정의합니다.

이들은 독립적인 LLM 호출자로서 각자의 role과 goal에 맞게 작동합니다. 여기서는 researcher와 writer 두 명의 전문가를 정의했습니다.

그 다음으로, Task 객체들을 생성하여 실제 수행할 작업을 정의합니다. 각 Task는 description(무엇을 할지), agent(누가 할지), expected_output(어떤 형식으로 출력할지)을 명시합니다.

핵심은 writing_task의 context=[research_task] 부분인데, 이것이 Task 간 의존성을 선언하는 방법입니다. writing_task는 research_task가 완료된 후에 실행되며, research_task의 출력을 자동으로 입력으로 받습니다.

세 번째로, Crew 객체를 생성하여 전체 팀을 구성합니다. agents 리스트에 모든 에이전트를, tasks 리스트에 모든 작업을 등록합니다.

process=Process.sequential은 Task들을 순차적으로 실행하라는 의미입니다(나중에 병렬 실행도 가능합니다). CrewAI는 이 정보를 바탕으로 실행 계획을 자동으로 수립합니다.

마지막으로, crew.kickoff()를 호출하여 전체 워크플로우를 시작합니다. inputs 딕셔너리로 전달한 값들은 Task의 description에서 {topic} 같은 플레이스홀더에 자동으로 주입됩니다.

CrewAI가 각 Task를 순서대로 실행하고, 각 에이전트에게 적절한 프롬프트를 생성하여 LLM을 호출하고, 결과를 다음 Task로 전달합니다. 여러분이 이 구조를 사용하면 복잡한 오케스트레이션 로직을 직접 작성할 필요가 없습니다.

Task 간 의존성만 선언하면 CrewAI가 알아서 실행 순서를 결정하고, 에러가 발생하면 재시도하거나 로그를 남깁니다. 또한 각 Task의 입출력이 자동으로 연결되어 수동 데이터 전달 코드가 필요 없습니다.

실전 팁

💡 expected_output을 명확히 정의하면 LLM이 더 일관된 형식으로 출력합니다. "JSON format with keys: title, content, tags"처럼 구체적으로 작성하세요.

💡 context는 리스트이므로 여러 Task의 출력을 하나의 Task가 입력으로 받을 수 있습니다. 예: context=[research_task, market_analysis_task]

💡 kickoff의 inputs는 모든 Task의 description에서 사용할 수 있습니다. 공통 파라미터는 여기에 정의하여 중복을 피하세요.

💡 개발 중에는 verbose=True로 설정하여 각 단계의 진행 상황을 확인하고, 프로덕션에서는 False로 설정하여 불필요한 로그를 줄이세요.

💡 Process.sequential은 디버깅이 쉽지만 느립니다. 워크플로우가 안정화되면 Process.hierarchical이나 병렬 실행을 고려하여 성능을 개선하세요.

4. Sequential Task 실행

시작하며

여러분이 데이터 파이프라인을 구축할 때, 데이터 수집 → 전처리 → 분석 → 시각화 순서로 반드시 진행해야 하는 경우가 있죠? 이전 단계의 출력이 다음 단계의 입력이 되기 때문에 순서를 바꾸면 전체 파이프라인이 망가집니다.

AI 에이전트 워크플로우도 마찬가지입니다. 리서치 없이 글을 쓸 수 없고, 글을 쓰지 않았는데 편집할 수 없습니다.

각 단계가 이전 단계의 결과에 의존하는 순차적 프로세스가 필요한데, 이를 수동으로 관리하면 if문과 변수 전달 코드로 가득 차게 됩니다. 바로 이럴 때 필요한 것이 Sequential Task 실행 패턴입니다.

CrewAI가 Task 간 의존성을 자동으로 파악하고 순서대로 실행하며 데이터를 전달해줍니다.

개요

간단히 말해서, Sequential Task 실행은 여러 Task를 정해진 순서대로 하나씩 실행하면서, 각 Task의 출력을 다음 Task의 입력으로 자동 전달하는 패턴입니다. 실무에서는 대부분의 복잡한 워크플로우가 순차적 구조를 가집니다.

예를 들어, 고객 피드백 분석 시스템을 만들 때 "피드백 수집 → 감정 분석 → 카테고리 분류 → 우선순위 결정 → 액션 아이템 생성" 같은 명확한 순서가 있습니다. 각 단계를 독립적인 Task로 만들고 순차 실행하면, 각 단계를 개별적으로 테스트하고 개선할 수 있습니다.

기존에는 step1_output = agent1.run(), step2_output = agent2.run(step1_output) 식으로 수동 체이닝을 했다면, 이제는 context 속성으로 의존성만 선언하면 CrewAI가 자동으로 데이터를 흘려보냅니다. Sequential 실행의 핵심은 데이터 흐름의 명확성과 디버깅 용이성입니다.

각 단계의 입출력을 로그로 확인할 수 있어 어디서 문제가 생겼는지 쉽게 파악할 수 있고, 특정 단계만 재실행하거나 건너뛸 수도 있습니다. 순차 실행은 병렬 실행보다 느리지만, 복잡한 의존성이 있는 워크플로우에서는 가장 안전하고 예측 가능한 방법입니다.

코드 예제

from crewai import Agent, Task, Crew, Process

# 단계별 에이전트 정의
collector = Agent(role='Data Collector', goal='Gather raw data', backstory='Expert at data collection')
analyzer = Agent(role='Data Analyzer', goal='Analyze collected data', backstory='Skilled data analyst')
reporter = Agent(role='Report Writer', goal='Create actionable reports', backstory='Professional report writer')

# 순차적 Task 체인 정의
task1 = Task(
    description='Collect customer feedback from {source}',
    agent=collector,
    expected_output='Raw feedback data in JSON format'
)

task2 = Task(
    description='Analyze sentiment and categorize feedback',
    agent=analyzer,
    expected_output='Analyzed data with sentiment scores',
    context=[task1]  # task1의 출력을 입력으로 사용
)

task3 = Task(
    description='Generate executive summary and action items',
    agent=reporter,
    expected_output='Executive report with recommendations',
    context=[task2]  # task2의 출력을 입력으로 사용
)

# Sequential 실행
crew = Crew(
    agents=[collector, analyzer, reporter],
    tasks=[task1, task2, task3],
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff(inputs={'source': 'customer_survey_2024'})

설명

이것이 하는 일: 이 코드는 데이터 수집부터 리포트 생성까지 3단계로 이루어진 순차적 워크플로우를 구현합니다. 첫 번째로, 각 단계를 담당할 전문 에이전트를 정의합니다.

collector는 데이터 수집만, analyzer는 분석만, reporter는 리포트 작성만 담당하도록 역할을 분리했습니다. 이렇게 하면 각 에이전트의 프롬프트를 해당 작업에 최적화할 수 있고, 특정 단계에 문제가 있을 때 해당 에이전트만 수정하면 됩니다.

그 다음으로, Task들을 정의하면서 context 속성으로 의존성을 명시합니다. task2는 context=[task1]로 설정되어 있어서 task1이 완료되기 전까지 실행되지 않으며, task1의 출력(Raw feedback data)을 자동으로 입력으로 받습니다.

마찬가지로 task3은 task2가 완료될 때까지 대기하고, task2의 분석 결과를 받아서 리포트를 작성합니다. CrewAI는 이 context 정보를 바탕으로 실행 순서를 자동으로 결정합니다.

세 번째로, Crew를 생성할 때 process=Process.sequential로 설정하여 순차 실행 모드를 활성화합니다. 이 모드에서는 tasks 리스트의 순서와 각 Task의 context를 종합하여 실행 계획을 수립합니다.

만약 context가 없더라도 리스트 순서대로 실행되지만, context가 있으면 의존성을 우선합니다. 마지막으로, kickoff()를 호출하면 CrewAI가 task1부터 시작합니다.

task1이 완료되면 그 출력을 task2의 프롬프트에 자동으로 포함시켜 analyzer 에이전트를 호출하고, task2가 완료되면 그 결과를 task3에 전달합니다. 각 단계의 진행 상황은 verbose=True 덕분에 콘솔에 출력되어 실시간으로 모니터링할 수 있습니다.

여러분이 이 패턴을 사용하면 복잡한 데이터 파이프라인을 선언적으로 정의할 수 있고, 각 단계를 독립적으로 테스트하거나 교체할 수 있습니다. 또한 특정 단계에서 에러가 발생해도 해당 Task부터 재시작할 수 있어 전체 프로세스를 처음부터 다시 실행할 필요가 없습니다.

실전 팁

💡 각 Task의 expected_output을 다음 Task에서 필요한 형식으로 명시하세요. "JSON with keys: sentiment, category, priority"처럼 구체적으로 정의하면 데이터 전달이 원활합니다.

💡 Sequential 실행은 각 단계를 기다려야 하므로 느릴 수 있습니다. 병렬화할 수 있는 Task가 있는지 검토하여 성능을 개선하세요.

💡 context 리스트에 여러 Task를 넣으면 여러 이전 단계의 출력을 한 번에 받을 수 있습니다. 예: context=[task1, task2]는 두 Task의 출력을 모두 입력으로 사용합니다.

💡 중간 단계의 출력을 로깅하거나 저장하려면 Task에 콜백 함수를 추가하세요. 디버깅과 모니터링에 매우 유용합니다.

💡 Task 간 데이터 전달 시 토큰 제한을 고려하세요. 이전 Task의 출력이 너무 길면 요약하거나 핵심 정보만 추출하는 중간 Task를 추가하는 것이 좋습니다.

5. Parallel Task 실행

시작하며

여러분이 여러 API를 동시에 호출해서 결과를 취합해야 할 때, 순차적으로 하나씩 호출하면 너무 느려서 답답했던 경험 있으시죠? 각 API 호출이 2초씩 걸린다면, 5개를 순차 호출하면 10초가 걸리지만 병렬로 호출하면 2초면 끝납니다.

AI 에이전트 워크플로우에서도 똑같은 문제가 있습니다. 여러 독립적인 리서치 작업을 순차적으로 하면 시간이 너무 오래 걸립니다.

예를 들어, "기술 트렌드 조사", "경쟁사 분석", "고객 리뷰 수집"은 서로 의존하지 않으므로 동시에 진행할 수 있는데, Sequential 방식으로만 하면 비효율적입니다. 바로 이럴 때 필요한 것이 Parallel Task 실행 패턴입니다.

서로 독립적인 Task들을 동시에 실행하여 전체 워크플로우 시간을 크게 단축할 수 있습니다.

개요

간단히 말해서, Parallel Task 실행은 서로 의존하지 않는 여러 Task를 동시에 실행하여 전체 처리 시간을 최소화하는 패턴입니다. 실무에서는 데이터 수집 단계처럼 여러 독립적인 소스에서 정보를 가져올 때 병렬화가 필수적입니다.

예를 들어, 시장 조사 시스템에서 "뉴스 기사 수집", "소셜 미디어 트렌드 분석", "경쟁사 웹사이트 크롤링"은 서로 독립적이므로 동시에 진행하고, 모든 결과가 모이면 다음 단계(종합 분석)로 넘어가는 구조가 효율적입니다. 기존에는 asyncio나 threading을 사용해 직접 병렬 처리를 구현해야 했다면, 이제는 CrewAI가 Task 간 의존성 그래프를 분석하여 자동으로 병렬화 가능한 Task를 식별하고 동시에 실행합니다.

Parallel 실행의 핵심은 독립성 식별과 결과 동기화입니다. 어떤 Task들이 서로 의존하지 않는지 정확히 파악하는 것이 중요하고, 모든 병렬 Task가 완료될 때까지 다음 단계가 기다려야 합니다.

잘못 설계하면 race condition이나 데이터 불일치가 발생할 수 있으므로 주의가 필요합니다.

코드 예제

from crewai import Agent, Task, Crew, Process

# 독립적인 리서치 에이전트들
tech_researcher = Agent(role='Tech Researcher', goal='Research technology trends', backstory='Tech expert')
market_researcher = Agent(role='Market Researcher', goal='Analyze market data', backstory='Market analyst')
competitor_researcher = Agent(role='Competitor Researcher', goal='Analyze competitors', backstory='Competition analyst')

# 병렬로 실행될 독립적인 Task들
task_tech = Task(
    description='Research latest AI technology trends',
    agent=tech_researcher,
    expected_output='Technology trend report'
)

task_market = Task(
    description='Analyze current market size and growth',
    agent=market_researcher,
    expected_output='Market analysis report'
)

task_competitor = Task(
    description='Analyze top 3 competitors',
    agent=competitor_researcher,
    expected_output='Competitor analysis report'
)

# 병렬 Task의 결과를 종합하는 Task
synthesis_agent = Agent(role='Strategy Synthesizer', goal='Create comprehensive strategy', backstory='Strategic planner')
task_synthesis = Task(
    description='Synthesize all research into actionable strategy',
    agent=synthesis_agent,
    expected_output='Comprehensive strategy document',
    context=[task_tech, task_market, task_competitor]  # 3개 Task 완료 후 실행
)

# Crew 구성 - Process.hierarchical은 병렬+순차 혼합 지원
crew = Crew(
    agents=[tech_researcher, market_researcher, competitor_researcher, synthesis_agent],
    tasks=[task_tech, task_market, task_competitor, task_synthesis],
    process=Process.hierarchical,  # 의존성 기반 자동 병렬화
    verbose=True
)

result = crew.kickoff()

설명

이것이 하는 일: 이 코드는 3개의 독립적인 리서치 Task를 병렬로 실행하고, 모든 결과를 종합하는 워크플로우를 구현합니다. 첫 번째로, 서로 다른 영역을 담당하는 3개의 전문 에이전트를 정의합니다.

tech_researcher, market_researcher, competitor_researcher는 각자의 전문 분야에서 독립적으로 작업하므로 서로의 결과를 기다릴 필요가 없습니다. 이런 독립성이 병렬화의 핵심 조건입니다.

그 다음으로, 각 에이전트에 대응하는 Task를 생성합니다. task_tech, task_market, task_competitor는 모두 context 속성이 없거나 비어있습니다.

이것이 "이 Task들은 다른 Task에 의존하지 않는다"는 신호이며, CrewAI는 이를 보고 병렬 실행 가능하다고 판단합니다. 각 Task는 자신의 에이전트를 사용하여 독립적으로 LLM을 호출합니다.

세 번째로, task_synthesis는 context=[task_tech, task_market, task_competitor]로 설정되어 있어서 3개의 병렬 Task가 모두 완료될 때까지 대기합니다. CrewAI는 이 의존성을 보고 "먼저 3개를 병렬로 실행하고, 모두 완료되면 synthesis를 실행한다"는 실행 계획을 수립합니다.

task_synthesis는 3개 Task의 출력을 모두 입력으로 받아서 종합 분석을 수행합니다. 마지막으로, process=Process.hierarchical은 Task 간 의존성 그래프를 분석하여 최적의 실행 전략을 자동으로 결정합니다.

의존성이 없는 Task들은 병렬로 실행하고, 의존성이 있는 Task는 순차적으로 실행합니다. 이 예제에서는 1단계(3개 병렬) → 2단계(1개 순차) 구조가 자동으로 형성됩니다.

여러분이 이 패턴을 사용하면 전체 실행 시간을 크게 단축할 수 있습니다. 3개 Task가 각각 30초 걸린다면, 순차 실행은 90초가 걸리지만 병렬 실행은 30초면 끝납니다.

특히 LLM API 호출처럼 I/O bound 작업에서 병렬화 효과가 큽니다.

실전 팁

💡 병렬 Task들이 같은 외부 리소스(DB, API)에 접근한다면 rate limit이나 동시성 제한을 고려하세요. 너무 많은 병렬 호출은 오히려 에러를 유발할 수 있습니다.

💡 각 병렬 Task의 실행 시간이 비슷하도록 작업을 분배하세요. 하나가 10초, 다른 하나가 60초 걸리면 병렬화 효과가 줄어듭니다.

💡 LLM API에는 보통 RPM(Requests Per Minute) 제한이 있습니다. 병렬 Task 수가 이 제한을 초과하지 않도록 주의하고, 필요하면 배치로 나누어 실행하세요.

💡 병렬 Task 중 하나가 실패해도 다른 Task는 계속 실행되도록 에러 핸들링 전략을 설정하세요. 부분적인 결과라도 활용할 수 있습니다.

💡 debugging 시에는 병렬 실행을 일시적으로 비활성화하고 순차 실행으로 전환하면 로그를 읽기 쉽습니다. Process.sequential로 바꾸면 됩니다.

6. Agent 간 통신과 메모리 공유

시작하며

여러분이 팀 프로젝트를 할 때, 각자 독립적으로 일하다가 나중에 통합하려니 이미 다른 팀원이 해결한 문제를 또 풀고 있거나, 중요한 결정 사항을 공유받지 못해 잘못된 방향으로 작업한 경험 있으신가요? 팀 협업의 핵심은 정보 공유입니다.

Multi-Agent 시스템도 똑같습니다. 각 에이전트가 고립되어 작동하면 중복 작업이 발생하고, 이전 단계에서 발견한 중요한 정보를 다음 에이전트가 활용하지 못합니다.

특히 긴 워크플로우에서는 초반 에이전트의 발견이 후반 에이전트에게 매우 중요할 수 있습니다. 바로 이럴 때 필요한 것이 Agent 간 메모리 공유와 통신 메커니즘입니다.

에이전트들이 공통 지식을 축적하고 참조할 수 있게 하여 협업의 품질을 높입니다.

개요

간단히 말해서, Agent 간 메모리 공유는 여러 에이전트가 공통의 지식 저장소를 읽고 쓰면서 정보를 축적하고 재사용하는 메커니즘입니다. 실무에서는 장기 실행되는 워크플로우나 반복적인 작업에서 메모리가 필수적입니다.

예를 들어, 고객 지원 시스템에서 "문의 분류 에이전트"가 파악한 고객의 기술 수준을 "기술 지원 에이전트"가 참조하여 답변의 난이도를 조절할 수 있습니다. 또한 이전 대화 내용을 기억하여 맥락을 유지하는 것도 메모리 공유의 중요한 사용 사례입니다.

기존에는 각 에이전트 호출마다 전체 대화 기록을 프롬프트에 포함시켜야 했다면, 이제는 공유 메모리에서 관련 정보만 검색하여 토큰을 절약하면서도 맥락을 유지할 수 있습니다. 메모리 공유의 핵심 특징은 지속성(persistence), 선택적 검색(selective retrieval), 컨텍스트 관리입니다.

모든 정보를 매번 전달하면 토큰 낭비가 심하므로, 현재 Task와 관련된 정보만 메모리에서 검색하여 제공하는 것이 중요합니다. CrewAI는 short-term memory(현재 세션), long-term memory(영구 저장), entity memory(특정 엔티티 관련 정보)를 지원합니다.

코드 예제

from crewai import Agent, Task, Crew, Process
from crewai.memory import ShortTermMemory, LongTermMemory, EntityMemory

# 메모리 활성화된 Crew 설정
crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    memory=True,  # 메모리 활성화
    verbose=True
)

# 메모리에 정보를 명시적으로 저장하는 에이전트
researcher = Agent(
    role='Researcher',
    goal='Research and store findings in memory',
    backstory="""You are a researcher who carefully documents findings.
    Store important discoveries in memory for other agents to use.""",
    verbose=True,
    memory=True  # 에이전트별 메모리 활성화
)

# 메모리를 참조하는 에이전트
writer = Agent(
    role='Writer',
    goal='Write content using researched information from memory',
    backstory="""You are a writer who checks memory for research findings
    before writing. Use stored information to create accurate content.""",
    verbose=True,
    memory=True
)

research_task = Task(
    description='Research {topic} and store key findings',
    agent=researcher,
    expected_output='Research summary with key facts'
)

writing_task = Task(
    description='Write article about {topic} using memory',
    agent=writer,
    expected_output='Well-researched article',
    context=[research_task]
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    memory=True,  # Crew 레벨 메모리 공유
    verbose=True
)

result = crew.kickoff(inputs={'topic': 'Multi-Agent AI Systems'})

설명

이것이 하는 일: 이 코드는 CrewAI의 메모리 시스템을 활성화하여 에이전트 간 정보 공유를 가능하게 합니다. 첫 번째로, Crew 생성 시 memory=True로 설정하면 CrewAI가 자동으로 메모리 시스템을 초기화합니다.

이 메모리는 모든 에이전트가 접근할 수 있는 공유 저장소로 작동하며, 기본적으로 short-term memory(현재 실행 세션 동안 유지)가 활성화됩니다. 필요하면 long-term memory를 설정하여 데이터베이스에 영구 저장할 수도 있습니다.

그 다음으로, 각 Agent에도 memory=True를 설정하여 에이전트가 메모리를 읽고 쓸 수 있게 합니다. researcher 에이전트의 backstory에 "Store important discoveries in memory"라고 명시하면, LLM이 중요한 발견을 메모리에 저장하도록 유도됩니다.

CrewAI는 에이전트의 출력을 자동으로 파싱하여 메모리에 저장하거나, 에이전트가 명시적으로 메모리 write 명령을 사용할 수도 있습니다. 세 번째로, writer 에이전트의 backstory에 "checks memory for research findings"라고 명시하면, 이 에이전트가 작업을 시작하기 전에 메모리를 검색합니다.

CrewAI는 현재 Task의 description과 관련된 메모리 항목을 자동으로 검색하여 프롬프트에 포함시킵니다. 이렇게 하면 research_task의 결과가 context로 명시적으로 전달되는 것 외에도, 메모리에 저장된 세부 정보들도 활용할 수 있습니다.

마지막으로, 여러 번 kickoff()를 호출해도 메모리는 유지됩니다(long-term memory를 사용하는 경우). 예를 들어, 첫 번째 실행에서 "Multi-Agent AI Systems"에 대해 리서치한 내용이 메모리에 저장되고, 나중에 관련 주제로 다시 실행하면 이전 정보를 재사용할 수 있습니다.

이것이 반복 작업의 효율성을 크게 높입니다. 여러분이 이 메커니즘을 사용하면 에이전트들이 "팀 지식"을 축적하면서 점점 더 똑똑해집니다.

또한 긴 워크플로우에서 초반 에이전트의 발견을 후반 에이전트가 자연스럽게 참조할 수 있어, 수동으로 모든 정보를 전달할 필요가 없습니다.

실전 팁

💡 메모리가 계속 쌓이면 검색 품질이 저하될 수 있습니다. 주기적으로 메모리를 정리하거나, 관련성 낮은 오래된 항목을 제거하는 전략을 수립하세요.

💡 프로덕션 환경에서는 long-term memory를 데이터베이스(PostgreSQL + pgvector 등)에 연결하여 세션 간 지식을 유지하세요. 개발 중에는 short-term memory로 충분합니다.

💡 민감한 정보가 메모리에 저장되지 않도록 주의하세요. 개인정보나 API 키 같은 데이터는 메모리에서 제외하거나 암호화해야 합니다.

💡 메모리 검색은 임베딩 기반 유사도 검색을 사용하므로, Task description을 명확하고 구체적으로 작성하면 관련 메모리를 더 잘 찾습니다.

💡 memory=True를 활성화하면 추가 LLM 호출(임베딩 생성)이 발생하여 비용과 지연이 증가합니다. 메모리가 정말 필요한 워크플로우에만 사용하세요.

7. Tool Sharing과 Delegation

시작하며

여러분이 개발팀에서 일할 때, 각자 자기만의 도구를 사용하다가 팀원이 유용한 라이브러리를 발견하면 공유해서 모두가 생산성을 높인 경험 있으시죠? 또한 자신이 감당하기 어려운 작업은 전문가에게 위임하는 것도 효율적인 협업의 핵심입니다.

Multi-Agent 시스템에서도 마찬가지입니다. 특정 에이전트가 웹 검색 도구를 가지고 있다면 다른 에이전트도 사용할 수 있어야 하고, 한 에이전트가 자신의 역량을 벗어난 작업을 만나면 적절한 전문가 에이전트에게 위임할 수 있어야 합니다.

그렇지 않으면 각 에이전트가 모든 도구를 중복으로 가지거나, 부적절한 에이전트가 작업을 억지로 수행하게 됩니다. 바로 이럴 때 필요한 것이 Tool Sharing과 Delegation 메커니즘입니다.

도구를 공유하여 중복을 줄이고, 작업을 적절한 에이전트에게 위임하여 품질을 높입니다.

개요

간단히 말해서, Tool Sharing은 여러 에이전트가 공통 도구(함수, API)를 공유하는 것이고, Delegation은 한 에이전트가 다른 에이전트에게 하위 작업을 위임하는 메커니즘입니다. 실무에서는 복잡한 작업을 수행할 때 에이전트가 모든 것을 혼자 할 수 없습니다.

예를 들어, 콘텐츠 작성 에이전트가 최신 통계 데이터가 필요하면 리서치 에이전트에게 "2024년 AI 시장 규모 조사" 작업을 위임하고, 그 결과를 받아서 글을 완성할 수 있습니다. 이렇게 하면 각 에이전트가 자신의 전문 영역에만 집중하면서도 필요할 때 협력할 수 있습니다.

기존에는 모든 에이전트에게 모든 도구를 할당하거나, 에이전트가 직접 다른 에이전트를 호출하는 복잡한 로직을 작성해야 했다면, 이제는 allow_delegation=True 설정만으로 에이전트가 자율적으로 작업을 위임할 수 있습니다. Tool Sharing의 핵심은 재사용성이고, Delegation의 핵심은 전문성 활용입니다.

도구를 중앙에서 관리하면 버전 관리와 업데이트가 쉬워지고, 위임을 활용하면 각 에이전트의 프롬프트를 단순하게 유지하면서도 복잡한 작업을 수행할 수 있습니다.

코드 예제

from crewai import Agent, Task, Crew
from crewai_tools import SerperDevTool, ScrapeWebsiteTool

# 공유 도구 정의
search_tool = SerperDevTool()  # 웹 검색 도구
scrape_tool = ScrapeWebsiteTool()  # 웹 스크래핑 도구

# 도구를 공유하는 여러 에이전트
researcher = Agent(
    role='Researcher',
    goal='Find accurate information',
    backstory='Expert researcher',
    tools=[search_tool, scrape_tool],  # 두 도구 모두 사용 가능
    verbose=True,
    allow_delegation=False  # 위임 불가
)

writer = Agent(
    role='Writer',
    goal='Write compelling content',
    backstory='Professional writer',
    tools=[search_tool],  # 검색 도구만 사용 가능
    verbose=True,
    allow_delegation=True  # 다른 에이전트에게 위임 가능
)

editor = Agent(
    role='Editor',
    goal='Ensure content quality and accuracy',
    backstory='Experienced editor',
    tools=[],  # 도구 없음, 위임만 사용
    verbose=True,
    allow_delegation=True  # 필요시 리서치나 수정 작업 위임
)

task_write = Task(
    description='Write an article about {topic}. If you need additional research, delegate to researcher.',
    agent=writer,
    expected_output='Article draft'
)

task_edit = Task(
    description='Edit and fact-check the article. Delegate verification to researcher if needed.',
    agent=editor,
    expected_output='Final polished article',
    context=[task_write]
)

crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[task_write, task_edit],
    verbose=True
)

result = crew.kickoff(inputs={'topic': 'AI Agent Collaboration'})

설명

이것이 하는 일: 이 코드는 여러 에이전트가 도구를 공유하고, 필요시 서로에게 작업을 위임하는 협업 시스템을 구현합니다. 첫 번째로, search_tool과 scrape_tool을 Crew 외부에서 정의하여 여러 에이전트가 같은 도구 인스턴스를 공유합니다.

이렇게 하면 도구의 설정이나 API 키를 한 곳에서 관리할 수 있고, 도구를 업그레이드할 때도 한 번만 수정하면 됩니다. CrewAI는 에이전트가 도구를 호출할 때 자동으로 프롬프트에 도구 사용 지침을 포함시킵니다.

그 다음으로, 각 에이전트의 tools 리스트에 필요한 도구만 할당합니다. researcher는 검색과 스크래핑 모두 필요하므로 두 도구를 가지고, writer는 검색만 필요하므로 search_tool만 가집니다.

editor는 도구 없이 위임만 사용하도록 설계했습니다. 이렇게 최소 권한 원칙을 적용하면 에이전트가 불필요한 도구를 오용하는 것을 방지할 수 있습니다.

세 번째로, allow_delegation 속성을 선택적으로 활성화합니다. writer와 editor는 allow_delegation=True로 설정되어 있어서, 작업 중에 자신이 처리하기 어려운 부분을 만나면 다른 에이전트에게 위임할 수 있습니다.

예를 들어, writer가 글을 쓰다가 "최신 통계가 필요하다"고 판단하면, researcher에게 "2024 AI market statistics 조사" 작업을 자동으로 위임하고 결과를 기다립니다. CrewAI가 이 위임 프로세스를 자동으로 관리합니다.

마지막으로, Task의 description에 "delegate to researcher if needed"라고 명시하여 에이전트가 위임 옵션을 인지하도록 합니다. LLM은 이 지시를 보고 필요시 위임 명령을 생성합니다.

editor가 fact-check을 하다가 의심스러운 정보를 발견하면, researcher에게 검증 작업을 위임하여 정확성을 높일 수 있습니다. 여러분이 이 패턴을 사용하면 각 에이전트를 단순하게 유지하면서도 복잡한 작업을 수행할 수 있습니다.

또한 도구를 중앙 관리하여 일관성을 유지하고, 위임을 통해 동적으로 협업 구조를 형성할 수 있습니다.

실전 팁

💡 allow_delegation=True는 강력하지만 예측 불가능성을 증가시킵니다. 프로덕션에서는 위임 로그를 모니터링하여 불필요한 위임이 발생하지 않는지 확인하세요.

💡 위임이 무한 루프에 빠지지 않도록 최대 위임 깊이를 설정하세요. A가 B에게 위임하고, B가 다시 A에게 위임하는 상황을 방지해야 합니다.

💡 도구는 비용이 발생할 수 있으므로(API 호출 등), 각 에이전트에 정말 필요한 도구만 할당하세요. 불필요한 도구 접근을 제한하면 오용과 비용을 줄입니다.

💡 위임 가능한 에이전트와 위임 불가능한 에이전트를 명확히 구분하세요. 리프 노드 에이전트(실제 작업 수행)는 위임 불가로, 관리자 에이전트는 위임 가능으로 설정하는 것이 일반적입니다.

💡 도구 사용 로그를 남겨서 어떤 에이전트가 어떤 도구를 얼마나 사용하는지 추적하세요. 이 데이터를 바탕으로 도구 할당을 최적화할 수 있습니다.

8. AutoGen Framework

시작하며

여러분이 코드 리뷰를 할 때, 작성자와 리뷰어가 여러 번 주고받으며 개선해나가는 과정 겪어보셨죠? "이 부분 개선이 필요합니다" → "이렇게 수정했습니다" → "좋네요, 이제 이 부분은?" 같은 대화형 협업이 품질을 높입니다.

CrewAI는 주로 정해진 워크플로우를 실행하는데 최적화되어 있지만, 때로는 에이전트들이 자유롭게 대화하며 문제를 해결해야 할 때가 있습니다. 특정 주제에 대해 여러 관점을 가진 에이전트들이 토론하거나, 한 에이전트의 출력을 다른 에이전트가 반복적으로 개선하는 시나리오에서는 대화형 접근이 효과적입니다.

바로 이럴 때 필요한 것이 Microsoft의 AutoGen 프레임워크입니다. 에이전트 간 대화를 중심으로 설계되어 동적이고 유연한 협업을 지원합니다.

개요

간단히 말해서, AutoGen은 여러 AI 에이전트가 자연어 대화를 통해 협업하여 문제를 해결하는 대화형 Multi-Agent 프레임워크입니다. 실무에서는 정형화되지 않은 복잡한 문제를 해결할 때 AutoGen이 유용합니다.

예를 들어, 소프트웨어 개발 시나리오에서 "개발자 에이전트"가 코드를 작성하면 "테스터 에이전트"가 버그를 찾고, 개발자가 수정하고, 테스터가 재검증하는 반복적인 대화 과정을 통해 코드 품질을 높일 수 있습니다. 기존 CrewAI가 "작업 A → 작업 B → 작업 C"의 파이프라인 구조라면, AutoGen은 "에이전트들이 목표 달성까지 자유롭게 대화"하는 구조입니다.

언제 대화를 종료할지, 누가 다음에 말할지도 동적으로 결정됩니다. AutoGen의 핵심 특징은 대화 기반 협업, 인간 개입 지원(human-in-the-loop), 코드 실행 능력입니다.

AssistantAgent는 LLM 기반으로 작동하고, UserProxyAgent는 코드를 실행하거나 인간의 피드백을 대신할 수 있습니다. 또한 GroupChat으로 여러 에이전트의 토론을 조율할 수 있습니다.

코드 예제

import autogen

# LLM 설정
config_list = [{
    "model": "gpt-4",
    "api_key": "your-api-key"
}]

# Assistant 에이전트 - LLM 기반
assistant = autogen.AssistantAgent(
    name="Assistant",
    llm_config={"config_list": config_list},
    system_message="You are a helpful AI assistant skilled at solving problems."
)

# UserProxy 에이전트 - 코드 실행 및 인간 역할
user_proxy = autogen.UserProxyAgent(
    name="User",
    human_input_mode="NEVER",  # 자동 모드 (인간 개입 없음)
    max_consecutive_auto_reply=10,
    code_execution_config={"work_dir": "coding", "use_docker": False}
)

# 대화 시작 - User가 Assistant에게 요청
user_proxy.initiate_chat(
    assistant,
    message="Write a Python function to calculate fibonacci numbers and test it."
)

# GroupChat으로 여러 에이전트 토론
coder = autogen.AssistantAgent(name="Coder", llm_config={"config_list": config_list})
tester = autogen.AssistantAgent(name="Tester", llm_config={"config_list": config_list})
reviewer = autogen.AssistantAgent(name="Reviewer", llm_config={"config_list": config_list})

groupchat = autogen.GroupChat(
    agents=[user_proxy, coder, tester, reviewer],
    messages=[],
    max_round=12
)
manager = autogen.GroupChatManager(groupchat=groupchat, llm_config={"config_list": config_list})

user_proxy.initiate_chat(manager, message="Develop and test a sorting algorithm.")

설명

이것이 하는 일: 이 코드는 AutoGen 프레임워크를 사용하여 LLM 에이전트와 실행 에이전트가 대화하며 코드를 작성하고 테스트하는 시스템을 구현합니다. 첫 번째로, AssistantAgent는 LLM(GPT-4 등)을 사용하여 사용자의 요청을 이해하고 응답을 생성하는 에이전트입니다.

system_message로 에이전트의 역할을 정의하며, 이것이 LLM의 system prompt로 사용됩니다. assistant는 코드를 작성하거나, 설명을 제공하거나, 질문에 답변하는 등 범용적인 작업을 수행합니다.

그 다음으로, UserProxyAgent는 사용자를 대신하거나, 코드를 실행하는 특수한 에이전트입니다. human_input_mode="NEVER"는 인간의 입력 없이 자동으로 작동하라는 의미이고(개발/테스트용), "ALWAYS"로 설정하면 매번 인간에게 확인을 받습니다.

code_execution_config는 에이전트가 생성한 Python 코드를 실제로 실행할 수 있게 하며, work_dir에서 코드 파일을 관리합니다. 이것이 AutoGen의 강력한 특징으로, LLM이 코드를 작성하면 바로 실행하고 결과를 확인할 수 있습니다.

세 번째로, initiate_chat()으로 대화를 시작합니다. user_proxy가 assistant에게 메시지를 보내면, assistant가 응답(코드 작성)하고, user_proxy가 그 코드를 실행하고, 결과를 assistant에게 전달하고, assistant가 결과를 보고 수정하는 식으로 대화가 이어집니다.

max_consecutive_auto_reply=10은 최대 10번까지 자동 응답하라는 의미로, 무한 루프를 방지합니다. 마지막으로, GroupChat은 3명 이상의 에이전트가 토론하는 시나리오를 지원합니다.

coder, tester, reviewer가 각자의 역할로 대화에 참여하며, GroupChatManager가 누가 다음에 말할지 결정합니다(LLM이 맥락을 보고 판단). 예를 들어, coder가 코드를 작성하면 tester가 테스트하고, 문제를 발견하면 coder에게 피드백하고, 최종적으로 reviewer가 검토하는 자연스러운 협업 흐름이 형성됩니다.

여러분이 AutoGen을 사용하면 에이전트들이 스스로 대화를 이어가며 문제를 해결하므로, 모든 단계를 미리 정의할 필요가 없습니다. 특히 코드 생성과 테스트처럼 반복적인 개선이 필요한 작업에 매우 효과적입니다.

실전 팁

💡 human_input_mode를 "TERMINATE"로 설정하면 에이전트가 작업을 완료했다고 판단할 때 인간에게 확인을 요청합니다. 중요한 결정이 필요한 워크플로우에 유용합니다.

💡 code_execution_config에서 use_docker=True로 설정하면 코드를 Docker 컨테이너에서 실행하여 보안을 강화할 수 있습니다. 프로덕션에서는 필수입니다.

💡 GroupChat의 max_round를 적절히 설정하세요. 너무 적으면 문제를 해결하기 전에 종료되고, 너무 많으면 비용이 과도하게 증가합니다.

💡 각 에이전트의 system_message를 명확히 구분하여 역할을 명시하세요. "You are a Python expert who writes clean code" vs "You are a tester who finds bugs"처럼 역할을 구분하면 협업 품질이 높아집니다.

💡 AutoGen은 대화 기록을 자동으로 관리하므로 토큰 사용량이 빠르게 증가합니다. 긴 대화는 주기적으로 요약하거나, 중요한 부분만 유지하는 전략을 사용하세요.

9. Consensus와 Voting

시작하며

여러분이 팀에서 중요한 기술 결정을 내릴 때, 여러 팀원의 의견을 듣고 투표나 합의를 통해 최선의 선택을 한 경험 있으시죠? 한 사람의 판단보다 여러 전문가의 집단 지성이 더 나은 결과를 만들어냅니다.

Multi-Agent 시스템에서도 마찬가지입니다. 하나의 에이전트가 내린 결정은 편향되거나 오류가 있을 수 있지만, 여러 에이전트가 독립적으로 분석하고 투표하여 결정하면 신뢰성이 높아집니다.

특히 중요한 의사결정이나 모호한 상황에서는 합의 메커니즘이 필수적입니다. 바로 이럴 때 필요한 것이 Consensus와 Voting 알고리즘입니다.

여러 에이전트의 의견을 체계적으로 수집하고 집계하여 최종 결정을 내립니다.

개요

간단히 말해서, Consensus는 여러 에이전트가 같은 문제를 독립적으로 분석하고, 그 결과를 투표나 가중치 합산으로 집계하여 최종 결정을 내리는 메커니즘입니다. 실무에서는 불확실성이 높은 판단이나 중요한 의사결정에 합의 방식을 사용합니다.

예를 들어, 고객 피드백의 감정 분석에서 "긍정인지 부정인지 애매한 경우" 3개의 독립적인 감정 분석 에이전트를 실행하고 다수결로 결정하면 정확도가 높아집니다. 또한 투자 결정, 리스크 평가, 콘텐츠 필터링 같은 중요한 작업에도 적합합니다.

기존에는 하나의 LLM 호출 결과를 그대로 사용했다면, 이제는 여러 번 호출하거나 여러 모델을 사용하여 결과의 일관성과 신뢰성을 검증할 수 있습니다. Consensus의 핵심 방법은 다수결 투표(Majority Voting), 가중치 투표(Weighted Voting), 앙상블 평균(Ensemble Averaging)입니다.

각 에이전트에 동등한 투표권을 주거나, 전문성에 따라 가중치를 달리 줄 수 있습니다. 또한 수치 결과의 경우 평균이나 중앙값을 사용할 수도 있습니다.

코드 예제

from crewai import Agent, Task, Crew
from collections import Counter

# 독립적인 감정 분석 에이전트 3명
analyst1 = Agent(
    role='Sentiment Analyst 1',
    goal='Analyze sentiment accurately',
    backstory='Conservative analyst who tends to be cautious',
    verbose=True
)

analyst2 = Agent(
    role='Sentiment Analyst 2',
    goal='Analyze sentiment accurately',
    backstory='Balanced analyst with neutral perspective',
    verbose=True
)

analyst3 = Agent(
    role='Sentiment Analyst 3',
    goal='Analyze sentiment accurately',
    backstory='Optimistic analyst who looks for positive aspects',
    verbose=True
)

# 각 에이전트가 독립적으로 분석하는 Task
task1 = Task(
    description='Analyze sentiment of: {text}. Output only: positive, negative, or neutral',
    agent=analyst1,
    expected_output='Sentiment label'
)

task2 = Task(
    description='Analyze sentiment of: {text}. Output only: positive, negative, or neutral',
    agent=analyst2,
    expected_output='Sentiment label'
)

task3 = Task(
    description='Analyze sentiment of: {text}. Output only: positive, negative, or neutral',
    agent=analyst3,
    expected_output='Sentiment label'
)

# 병렬 실행으로 독립성 보장
crew = Crew(
    agents=[analyst1, analyst2, analyst3],
    tasks=[task1, task2, task3],
    verbose=True
)

result = crew.kickoff(inputs={'text': 'The product is okay but could be better'})

# 결과 수집 및 다수결 투표
votes = [task1.output.raw_output, task2.output.raw_output, task3.output.raw_output]
vote_counts = Counter(votes)
final_decision = vote_counts.most_common(1)[0][0]

print(f"Votes: {votes}")
print(f"Final decision by majority: {final_decision}")

설명

이것이 하는 일: 이 코드는 3개의 독립적인 에이전트가 같은 텍스트의 감정을 분석하고, 다수결 투표로 최종 결과를 결정하는 시스템을 구현합니다. 첫 번째로, 3개의 감정 분석 에이전트를 생성하되, 각자 다른 backstory를 부여하여 관점을 다르게 만듭니다.

analyst1은 보수적(부정적 측면에 민감), analyst2는 균형잡힌, analyst3은 낙관적(긍정적 측면을 강조)으로 설정했습니다. 이렇게 다양한 관점을 가진 에이전트를 사용하면 편향을 줄이고 다각도로 분석할 수 있습니다.

그 다음으로, 각 에이전트에 동일한 작업을 할당하지만, 독립적인 Task 객체로 생성합니다. task1, task2, task3은 모두 같은 텍스트를 분석하지만, 서로 다른 에이전트를 사용하므로 독립적인 판단이 나옵니다.

expected_output을 "Sentiment label"로 명시하여 출력 형식을 통일하고, description에서 "Output only: positive, negative, or neutral"로 제한하여 파싱을 쉽게 만듭니다. 세 번째로, Crew를 실행하면 3개의 Task가 병렬로(또는 순차로) 실행되어 각자의 결과를 생성합니다.

중요한 점은 각 에이전트가 서로의 결과를 보지 못하고 독립적으로 판단한다는 것입니다. context를 설정하지 않았으므로 서로 영향을 주지 않습니다.

마지막으로, kickoff() 완료 후 각 Task의 output.raw_output에서 결과를 수집하여 투표를 진행합니다. Counter를 사용하여 각 라벨의 개수를 세고, most_common(1)로 가장 많이 나온 라벨을 최종 결정으로 선택합니다.

예를 들어, votes가 ['positive', 'neutral', 'positive']라면 'positive'가 2표로 최종 결정됩니다. 여러분이 이 패턴을 사용하면 단일 에이전트의 오판을 보정할 수 있습니다.

3개 중 1개가 잘못 판단해도 나머지 2개가 맞으면 올바른 결과를 얻을 수 있습니다. 특히 LLM의 비결정성(같은 프롬프트에도 다른 응답)을 고려하면, 여러 번 실행하여 안정적인 결과를 얻는 것이 중요합니다.

실전 팁

💡 투표 참여자 수는 홀수로 설정하여 동점을 방지하세요. 3명, 5명, 7명이 일반적이며, 너무 많으면 비용이 증가합니다.

💡 각 에이전트의 backstory나 temperature를 다르게 설정하여 다양성을 확보하세요. 모두 똑같은 설정이면 같은 결과만 반복될 수 있습니다.

💡 가중치 투표를 구현하려면 특정 에이전트의 투표에 더 높은 가중치를 부여하세요. 예: 전문가 에이전트는 2표, 일반 에이전트는 1표.

💡 투표 결과가 너무 분산되면(예: 3명이 모두 다른 답) 신뢰도가 낮다는 신호입니다. 이 경우 더 많은 에이전트를 추가하거나, 문제를 재검토해야 합니다.

💡 수치 결과(예: 점수)의 경우 평균이나 중앙값을 사용하세요. 이상치(outlier) 영향을 줄이려면 중앙값이 더 안정적입니다.

10. Multi-Agent 오케스트레이션

시작하며

여러분이 마이크로서비스 아키텍처를 운영할 때, 각 서비스는 잘 작동하지만 전체 시스템의 흐름을 제어하는 오케스트레이터(Kubernetes, 서비스 메시)가 없으면 혼란스러웠던 경험 있으시죠? 누가 언제 무엇을 실행할지, 에러가 나면 어떻게 복구할지 중앙에서 관리해야 합니다.

Multi-Agent 시스템도 마찬가지입니다. 각 에이전트는 자신의 작업을 잘 수행하지만, 전체 워크플로우를 조율하고 모니터링하며 에러를 처리하는 오케스트레이션 레이어가 필요합니다.

특히 복잡한 의존성과 조건부 분기가 있는 워크플로우에서는 중앙 제어가 필수적입니다. 바로 이럴 때 필요한 것이 Multi-Agent 오케스트레이션 패턴입니다.

전체 에이전트 시스템의 실행 흐름을 제어하고, 상태를 관리하며, 예외를 처리합니다.

개요

간단히 말해서, Multi-Agent 오케스트레이션은 여러 에이전트의 실행 순서, 데이터 흐름, 에러 처리를 중앙에서 제어하고 모니터링하는 아키텍처 패턴입니다. 실무에서는 복잡한 비즈니스 로직을 AI 에이전트로 구현할 때 오케스트레이션이 필수입니다.

예를 들어, 고객 주문 처리 시스템에서 "재고 확인 → 결제 처리 → 배송 준비 → 알림 발송" 각 단계를 다른 에이전트가 담당하는데, 재고가 없으면 다른 공급자 조회로 분기하고, 결제 실패하면 재시도하는 등의 복잡한 로직을 오케스트레이터가 관리합니다. 기존에는 각 에이전트 호출을 코드로 직접 체이닝하고 if-else로 분기 처리했다면, 이제는 LangGraph나 CrewAI의 hierarchical process로 선언적으로 정의할 수 있습니다.

오케스트레이션의 핵심 요소는 상태 관리(state management), 조건부 라우팅(conditional routing), 에러 복구(error recovery), 모니터링(monitoring)입니다. 전체 워크플로우의 현재 상태를 추적하고, 조건에 따라 다른 경로로 분기하며, 실패 시 재시도하거나 보상 트랜잭션을 실행합니다.

코드 예제

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

# 워크플로우 상태 정의
class WorkflowState(TypedDict):
    input: str
    research_result: str
    draft: str
    final_output: str
    error: str
    retry_count: Annotated[int, operator.add]

# 각 노드(에이전트 작업)를 함수로 정의
def research_node(state: WorkflowState):
    try:
        # researcher 에이전트 실행
        result = researcher.run(state['input'])
        return {"research_result": result, "error": ""}
    except Exception as e:
        return {"error": str(e), "retry_count": 1}

def write_node(state: WorkflowState):
    try:
        # writer 에이전트 실행
        draft = writer.run(state['research_result'])
        return {"draft": draft, "error": ""}
    except Exception as e:
        return {"error": str(e), "retry_count": 1}

def edit_node(state: WorkflowState):
    try:
        # editor 에이전트 실행
        final = editor.run(state['draft'])
        return {"final_output": final, "error": ""}
    except Exception as e:
        return {"error": str(e), "retry_count": 1}

# 조건부 라우팅 - 에러 여부에 따라 분기
def should_continue(state: WorkflowState):
    if state.get('error'):
        if state.get('retry_count', 0) < 3:
            return "retry"
        else:
            return "fail"
    return "continue"

# 워크플로우 그래프 구성
workflow = StateGraph(WorkflowState)

# 노드 추가
workflow.add_node("research", research_node)
workflow.add_node("write", write_node)
workflow.add_node("edit", edit_node)

# 엣지(연결) 정의
workflow.set_entry_point("research")
workflow.add_conditional_edges(
    "research",
    should_continue,
    {
        "continue": "write",
        "retry": "research",
        "fail": END
    }
)
workflow.add_edge("write", "edit")
workflow.add_edge("edit", END)

# 컴파일 및 실행
app = workflow.compile()
result = app.invoke({"input": "AI Agent Collaboration", "retry_count": 0})

설명

이것이 하는 일: 이 코드는 LangGraph를 사용하여 Multi-Agent 워크플로우를 상태 머신으로 구현하고, 에러 처리와 재시도 로직을 포함합니다. 첫 번째로, WorkflowState라는 TypedDict로 전체 워크플로우에서 공유되는 상태를 정의합니다.

input은 초기 입력, research_result, draft, final_output은 각 단계의 출력이고, error는 에러 메시지, retry_count는 재시도 횟수입니다. Annotated[int, operator.add]는 이 필드가 누적(accumulate)된다는 의미로, 여러 노드에서 증가시킬 수 있습니다.

그 다음으로, 각 에이전트의 작업을 노드 함수로 정의합니다. research_node는 researcher 에이전트를 실행하고, 성공하면 research_result를 상태에 저장하고, 실패하면 error와 retry_count를 설정합니다.

모든 노드는 상태를 입력으로 받아서 새로운 상태 값을 반환하며, LangGraph가 자동으로 상태를 병합합니다. try-except로 에러를 포착하여 워크플로우가 중단되지 않도록 합니다.

세 번째로, should_continue라는 조건부 라우팅 함수를 정의합니다. 이 함수는 현재 상태를 보고 다음에 어느 노드로 갈지 결정합니다.

error가 있으면 retry_count를 확인하여 3회 미만이면 "retry"(같은 노드 재실행), 3회 이상이면 "fail"(워크플로우 종료)를 반환합니다. error가 없으면 "continue"를 반환하여 다음 단계로 진행합니다.

마지막으로, StateGraph로 워크플로우를 구성합니다. add_node로 각 에이전트 작업을 노드로 등록하고, add_edge로 고정된 연결(write → edit)을 정의하고, add_conditional_edges로 조건부 분기(research 이후)를 정의합니다.

set_entry_point("research")는 시작 노드를 지정합니다. compile() 후 invoke()로 실행하면, LangGraph가 상태를 관리하며 노드들을 순서대로 실행하고, 조건에 따라 분기하며, 에러 시 재시도합니다.

여러분이 이 패턴을 사용하면 복잡한 워크플로우를 시각적으로 이해할 수 있는 그래프로 표현하고, 상태를 명시적으로 관리하며, 에러 처리와 재시도 로직을 선언적으로 정의할 수 있습니다. 또한 각 노드를 독립적으로 테스트하고 교체할 수 있어 유지보수가 쉽습니다.

실전 팁

💡 상태가 너무 커지면 메모리와 직렬화 비용이 증가합니다. 각 단계에서 필요한 최소한의 정보만 상태에 저장하세요.

💡 조건부 라우팅을 과도하게 사용하면 워크플로우가 복잡해지고 디버깅이 어려워집니다. 가능하면 단순한 선형 구조를 유지하고, 정말 필요한 곳에만 분기를 사용하세요.

💡 재시도 로직에 exponential backoff(지수 백오프)를 추가하면 일시적인 에러(네트워크 등)에 더 효과적으로 대응할 수 있습니다.

💡 각 노드의 실행 시간과 성공/실패를 로깅하여 모니터링 대시보드를 구축하세요. 어느 단계에서 병목이나 에러가 자주 발생하는지 파악할 수 있습니다.

💡 프로덕션에서는 워크플로우 상태를 데이터베이스에 저장하여 중단된 워크플로우를 재개하거나, 실패한 워크플로우를 분석할 수 있도록 하세요.

11. 에러 핸들링과 복구 전략

시작하며

여러분이 분산 시스템을 운영할 때, 네트워크 장애나 서비스 다운타임이 발생해도 시스템이 계속 작동하도록 재시도, 폴백, 서킷 브레이커 같은 패턴을 적용한 경험 있으시죠? 완벽한 시스템은 없으므로, 에러를 어떻게 처리하느냐가 시스템의 안정성을 결정합니다.

Multi-Agent 시스템도 똑같습니다. LLM API가 일시적으로 실패할 수 있고, 특정 에이전트가 잘못된 출력을 생성할 수 있으며, 외부 도구 호출이 타임아웃될 수 있습니다.

이런 상황에서 전체 워크플로우가 중단되면 사용자 경험이 나빠지고, 비용도 낭비됩니다. 바로 이럴 때 필요한 것이 체계적인 에러 핸들링과 복구 전략입니다.

예상 가능한 에러를 포착하고, 재시도하거나 대안 경로를 실행하여 시스템의 회복탄력성을 높입니다.

개요

간단히 말해서, 에러 핸들링은 Multi-Agent 워크플로우에서 발생하는 다양한 에러를 포착하고, 재시도, 폴백, 부분 성공 처리 등의 전략으로 복구하는 메커니즘입니다. 실무에서는 프로덕션 환경의 Multi-Agent 시스템이 다양한 에러에 직면합니다.

LLM API의 rate limit 초과, 네트워크 타임아웃, 잘못된 출력 형식, 외부 도구 실패 등을 적절히 처리해야 시스템이 안정적으로 작동합니다. 예를 들어, 리서치 에이전트가 웹 검색에 실패하면 캐시된 데이터를 사용하거나, 다른 검색 도구로 대체하는 폴백 전략이 필요합니다.

기존에는 try-except로 간단히 처리하고 로그만 남겼다면, 이제는 에러 유형별로 다른 복구 전략(재시도, 폴백, 스킵, 인간 개입 요청)을 적용하여 가용성을 높일 수 있습니다. 에러 핸들링의 핵심 전략은 재시도(Retry with backoff), 폴백(Fallback to alternative), 부분 성공(Partial success), 서킷 브레이커(Circuit breaker)입니다.

일시적 에러는 재시도하고, 재시도가 실패하면 대안을 사용하며, 일부 작업이 실패해도 나머지는 계속 진행하고, 반복적인 실패는 빠르게 차단하여 리소스 낭비를 방지합니다.

코드 예제

from crewai import Agent, Task, Crew
import time
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

# 재시도 데코레이터 - API 에러에 대한 자동 재시도
@retry(
    stop=stop_after_attempt(3),  # 최대 3회 재시도
    wait=wait_exponential(multiplier=1, min=2, max=10),  # 2초, 4초, 8초 대기
    retry=retry_if_exception_type(Exception)  # 모든 예외에 대해 재시도
)
def call_agent_with_retry(agent, task_description):
    """에이전트를 재시도 로직과 함께 호출"""
    return agent.execute_task(task_description)

# 폴백 전략 - 주 도구 실패 시 대안 도구 사용
def research_with_fallback(topic):
    try:
        # 주 검색 도구 시도
        result = primary_search_tool.run(topic)
        return result
    except Exception as e:
        print(f"Primary tool failed: {e}, trying fallback...")
        try:
            # 대안 검색 도구 시도
            result = fallback_search_tool.run(topic)
            return result
        except Exception as e2:
            print(f"Fallback also failed: {e2}, using cached data...")
            # 최종 폴백: 캐시된 데이터
            return get_cached_data(topic)

# 부분 성공 처리 - 일부 에이전트 실패해도 계속 진행
def run_parallel_with_partial_success(tasks):
    results = []
    errors = []
    for task in tasks:
        try:
            result = task.execute()
            results.append(result)
        except Exception as e:
            errors.append({"task": task.description, "error": str(e)})
            print(f"Task {task.description} failed, continuing with others...")

    # 최소 하나 이상 성공하면 진행
    if results:
        return {"results": results, "errors": errors, "status": "partial_success"}
    else:
        raise Exception(f"All tasks failed: {errors}")

# 에러 핸들링이 통합된 Crew
class ResilientCrew:
    def __init__(self, agents, tasks):
        self.agents = agents
        self.tasks = tasks
        self.failure_count = 0
        self.circuit_open = False

    def execute(self):
        # 서킷 브레이커 체크
        if self.circuit_open:
            raise Exception("Circuit breaker is open due to repeated failures")

        try:
            results = []
            for task in self.tasks:
                result = call_agent_with_retry(task.agent, task.description)
                results.append(result)

            # 성공 시 실패 카운트 리셋
            self.failure_count = 0
            return results

        except Exception as e:
            self.failure_count += 1
            # 5회 연속 실패 시 서킷 오픈
            if self.failure_count >= 5:
                self.circuit_open = True
                print("Circuit breaker opened after 5 failures")
            raise e

설명

이것이 하는 일: 이 코드는 Multi-Agent 워크플로우에서 발생하는 다양한 에러를 처리하는 여러 전략을 구현합니다. 첫 번째로, tenacity 라이브러리의 @retry 데코레이터를 사용하여 자동 재시도를 구현합니다.

stop_after_attempt(3)은 최대 3회까지 재시도하고, wait_exponential은 재시도 간격을 지수적으로 증가시켜 일시적 에러(네트워크 혼잡 등)가 해소될 시간을 줍니다. 예를 들어, LLM API의 rate limit에 걸리면 2초 후 재시도, 또 실패하면 4초 후 재시도하는 식으로 점진적으로 대기 시간을 늘립니다.

그 다음으로, research_with_fallback 함수는 다단계 폴백 전략을 구현합니다. 먼저 primary_search_tool(예: Google Search API)을 시도하고, 실패하면 fallback_search_tool(예: Bing Search API)을 시도하고, 그것도 실패하면 get_cached_data로 이전에 캐시된 결과를 반환합니다.

이렇게 하면 외부 의존성이 실패해도 시스템이 계속 작동할 수 있습니다. 세 번째로, run_parallel_with_partial_success는 여러 병렬 작업 중 일부가 실패해도 성공한 결과만 가지고 계속 진행하는 패턴입니다.

예를 들어, 5개의 리서치 소스 중 2개가 실패해도 3개의 결과로 분석을 진행할 수 있습니다. results와 errors를 모두 반환하여 호출자가 부분 성공 상황을 인지하고 적절히 처리할 수 있게 합니다.

마지막으로, ResilientCrew 클래스는 서킷 브레이커 패턴을 구현합니다. failure_count를 추적하여 연속 5회 실패하면 circuit_open을 True로 설정하고, 이후 요청을 즉시 거부합니다.

이렇게 하면 반복적으로 실패하는 작업으로 인한 리소스 낭비(비용, 시간)를 방지하고, 시스템이 빠르게 실패(fail-fast)하여 다른 복구 수단을 시도할 수 있습니다. 여러분이 이런 전략들을 조합하여 사용하면 프로덕션 환경에서 Multi-Agent 시스템의 안정성을 크게 높일 수 있습니다.

일시적 에러는 재시도로 해결하고, 지속적 에러는 폴백으로 우회하며, 치명적 에러는 서킷 브레이커로 격리합니다.

실전 팁

💡 재시도 전략은 멱등성(idempotency)이 보장될 때만 안전합니다. 같은 작업을 여러 번 실행해도 결과가 같은지 확인하세요. 결제 같은 작업은 재시도 시 중복 처리를 방지해야 합니다.

💡 에러 로그에 충분한 컨텍스트(어느 에이전트, 어느 Task, 입력 데이터)를 포함하여 디버깅을 쉽게 하세요. Sentry나 CloudWatch 같은 모니터링 도구와 통합하면 더 좋습니다.

💡 서킷 브레이커는 일정 시간 후 자동으로 half-open 상태로 전환하여 회복 여부를 테스트해야 합니다. 영구적으로 닫혀있으면 시스템이 복구되어도 사용할 수 없습니다.

💡 각 에러 유형별로 다른 처리 전략을 적용하세요. rate limit은 재시도, invalid input은 스킵, critical failure는 알림 발송 등 세분화된 처리가 필요합니다.

💡 테스트 환경에서 의도적으로 에러를 주입(chaos engineering)하여 에러 핸들링 로직이 제대로 작동하는지 검증하세요. 실제 장애 상황에서 처음 테스트하면 늦습니다.

12. 실전 Multi-Agent 아키텍처

시작하며

여러분이 지금까지 배운 모든 개념들을 실제 프로덕션 시스템에 어떻게 통합할지 고민하고 계시죠? 각 패턴은 이해했지만, 이들을 조합하여 확장 가능하고 유지보수 가능한 실전 아키텍처를 설계하는 것은 또 다른 차원의 문제입니다.

실무에서는 단순한 데모가 아니라 수천 명의 사용자를 처리하고, 비용을 최적화하며, 장애에 대응하고, 지속적으로 개선할 수 있는 시스템이 필요합니다. 지금까지 배운 개념들을 어떻게 조합하고, 어떤 아키텍처 결정을 내려야 할까요?

바로 이럴 때 필요한 것이 검증된 Multi-Agent 아키텍처 패턴입니다. 모범 사례를 따르고, 확장성과 관찰 가능성을 고려하여 프로덕션 레벨 시스템을 설계합니다.

개요

간단히 말해서, 실전 Multi-Agent 아키텍처는 여러 에이전트, 오케스트레이션, 메모리, 도구, 모니터링을 통합하여 확장 가능하고 안정적인 프로덕션 시스템을 구축하는 설계 패턴입니다. 실무에서는 다음과 같은 요구사항을 모두 충족해야 합니다: 수평 확장 가능, 비용 효율적, 관찰 가능(로그, 메트릭, 추적), 테스트 가능, 버전 관리 가능, 보안(API 키, 데이터 격리).

예를 들어, 대규모 콘텐츠 생성 플랫폼을 구축한다면 수백 개의 동시 요청을 처리하고, 각 요청의 진행 상황을 추적하며, 실패한 작업을 재시도하고, 비용을 모니터링해야 합니다. 기존의 단순한 스크립트나 프로토타입과 달리, 프로덕션 아키텍처는 계층화(presentation, orchestration, agent, tool), 관심사 분리(각 레이어의 책임), 인터페이스 추상화(에이전트 교체 가능)가 필요합니다.

실전 아키텍처의 핵심 원칙은 모듈화(각 에이전트/도구는 독립적 모듈), 관찰 가능성(모든 단계를 로깅/추적), 확장성(수평 확장 가능한 설계), 비용 최적화(캐싱, 배칭, 모델 선택)입니다. 이런 원칙들을 구체적인 아키텍처 패턴으로 구현해야 합니다.

코드 예제

# 프로덕션 Multi-Agent 아키텍처 예시
from crewai import Agent, Task, Crew
from langchain.cache import RedisCache
from langchain.callbacks import get_openai_callback
import structlog
from prometheus_client import Counter, Histogram

# 구조화된 로깅 설정
logger = structlog.get_logger()

# 메트릭 정의
agent_calls = Counter('agent_calls_total', 'Total agent calls', ['agent_name', 'status'])
agent_duration = Histogram('agent_duration_seconds', 'Agent execution time', ['agent_name'])

# 캐싱 레이어 - 중복 LLM 호출 방지
from langchain.globals import set_llm_cache
set_llm_cache(RedisCache(redis_url="redis://localhost:6379"))

class ProductionAgent:
    """프로덕션용 에이전트 래퍼"""
    def __init__(self, agent: Agent):
        self.agent = agent
        self.name = agent.role

    def execute(self, task_description: str) -> dict:
        logger.info("agent_execution_started", agent=self.name, task=task_description)

        with agent_duration.labels(agent_name=self.name).time():
            try:
                with get_openai_callback() as cb:
                    result = self.agent.execute_task(task_description)

                    # 비용 및 토큰 로깅
                    logger.info("agent_execution_completed",
                                agent=self.name,
                                tokens=cb.total_tokens,
                                cost=cb.total_cost)

                    agent_calls.labels(agent_name=self.name, status='success').inc()
                    return {"status": "success", "result": result, "cost": cb.total_cost}

            except Exception as e:
                logger.error("agent_execution_failed", agent=self.name, error=str(e))
                agent_calls.labels(agent_name=self.name, status='error').inc()
                return {"status": "error", "error": str(e)}

class MultiAgentPipeline:
    """프로덕션 Multi-Agent 파이프라인"""
    def __init__(self):
        self.agents = self._initialize_agents()
        self.cache = RedisCache(redis_url="redis://localhost:6379")

    def _initialize_agents(self):
        """에이전트 초기화"""
        researcher = ProductionAgent(Agent(
            role='Researcher',
            goal='Research information',
            backstory='Expert researcher',
            verbose=False  # 프로덕션에서는 verbose 비활성화
        ))

        writer = ProductionAgent(Agent(
            role='Writer',
            goal='Write content',
            backstory='Professional writer',
            verbose=False
        ))

        return {"researcher": researcher, "writer": writer}

    async def process(self, topic: str, user_id: str):
        """비동기 처리 파이프라인"""
        logger.info("pipeline_started", topic=topic, user_id=user_id)

        # 캐시 확인
        cache_key = f"content:{topic}"
        cached = self.cache.lookup(topic, "")
        if cached:
            logger.info("cache_hit", topic=topic)
            return cached

        # 단계별 실행
        research_result = self.agents["researcher"].execute(f"Research: {topic}")
        if research_result["status"] == "error":
            return research_result

#AI#MultiAgent#CrewAI#AutoGen#AgentCollaboration

AI 에이전트 Multi-Agent 협업 완벽 가이드

# AI 에이전트 Multi-Agent 협업 완벽 가이드 여러 AI 에이전트가 협업하여 복잡한 문제를 해결하는 Multi-Agent 시스템의 핵심 개념과 실무 구현 방법을 알아봅니다. CrewAI, AutoGen 등 최신 프레임워크 활용법과 실전 아키텍처 패턴을 다룹니다. --- ## 목차 1. [Multi-Agent 시스템 기본 개념](#multi-agent-시스템-기본-개념) 2. [Agent Role과 Goal 설계](#agent-role과-goal-설계) 3. [CrewAI 기본 구조](#crewai-기본-구조) 4. [Sequential Task 실행](#sequential-task-실행) 5. [Parallel Task 실행](#parallel-task-실행) 6. [Agent 간 통신과 메모리 공유](#agent-간-통신과-메모리-공유) 7. [Tool Sharing과 Delegation](#tool-sharing과-delegation) 8. [AutoGen Framework](#autogen-framework) 9. [Consensus와 Voting](#consensus와-voting) 10. [Multi-Agent 오케스트레이션](#multi-agent-오케스트레이션) 11. [에러 핸들링과 복구 전략](#에러-핸들링과-복구-전략) 12. [실전 Multi-Agent 아키텍처](#실전-multi-agent-아키텍처) --- ## 1. Multi-Agent 시스템 기본 개념 ### 시작하며 여러분이 대규모 프로젝트를 혼자 진행할 때, 리서치도 하고 코드도 짜고 문서도 작성하느라 정신없었던 경험 있으시죠? 한 명이 모든 것을 처리하려니 효율도 떨어지고 전문성도 부족해집니다. 이런 문제는 AI 에이전트 시스템에서도 똑같이 발생합니다. 하나의 거대한 에이전트가 모든 작업을 처리하려면 프롬프트도 복잡해지고, 각 작업의 품질도 저하됩니다. 게다가 특정 작업이 실패하면 전체 프로세스가 중단되는 취약점도 생깁니다. 바로 이럴 때 필요한 것이 Multi-Agent 시스템입니다. 여러 전문 에이전트가 각자의 역할을 수행하면서 협업하여, 복잡한 문제를 효율적으로 해결할 수 있습니다. ### 개요 간단히 말해서, Multi-Agent 시스템은 각기 다른 역할과 전문성을 가진 여러 AI 에이전트가 협력하여 복잡한 작업을 수행하는 아키텍처입니다. 실무에서는 단일 LLM 호출로 해결할 수 없는 복잡한 워크플로우가 많습니다. 예를 들어, 시장 조사를 기반으로 블로그 글을 작성하고 SEO 최적화까지 해야 하는 경우, 리서처 에이전트가 데이터를 수집하고, 작가 에이전트가 글을 쓰고, 편집자 에이전트가 검토하는 식으로 역할을 분담할 수 있습니다. 기존에는 하나의 긴 프롬프트에 모든 지시사항을 넣어야 했다면, 이제는 각 에이전트가 자신의 전문 영역에만 집중하면서 이전 에이전트의 결과를 받아서 작업할 수 있습니다. Multi-Agent 시스템의 핵심 특징은 모듈화, 전문화, 확장성입니다. 각 에이전트는 독립적으로 개선할 수 있고, 필요에 따라 새로운 에이전트를 추가하기도 쉽습니다. 이러한 특징들이 복잡한 AI 워크플로우를 관리 가능한 수준으로 만들어줍니다. ### 코드 예제 ```python from crewai import Agent, Task, Crew # 리서처 에이전트 - 정보 수집 전문 researcher = Agent( role='Research Analyst', goal='Find accurate and relevant information', backstory='Expert at gathering and analyzing data', verbose=True ) # 작가 에이전트 - 콘텐츠 작성 전문 writer = Agent( role='Content Writer', goal='Create engaging content based on research', backstory='Skilled writer with SEO expertise', verbose=True ) # Crew 구성 - 에이전트 팀 생성 crew = Crew( agents=[researcher, writer], verbose=True ) ``` ### 설명 이것이 하는 일: 이 코드는 CrewAI 프레임워크를 사용하여 두 개의 전문 에이전트로 구성된 협업 시스템을 구축합니다. 첫 번째로, 각 Agent 객체를 생성할 때 role, goal, backstory를 정의합니다. role은 에이전트의 직책이나 전문 분야를 나타내고, goal은 이 에이전트가 달성하고자 하는 목표입니다. backstory는 에이전트의 배경 설정으로, LLM이 이 맥락을 이해하고 해당 역할에 맞게 행동하도록 돕습니다. verbose=True는 에이전트의 사고 과정을 로그로 출력하여 디버깅을 쉽게 만듭니다. 그 다음으로, Crew 객체를 생성하면서 agents 리스트에 여러 에이전트를 등록합니다. Crew는 에이전트들의 오케스트레이터 역할을 하며, 작업을 분배하고 결과를 취합합니다. 각 에이전트는 독립적으로 LLM을 호출할 수 있으며, 이전 에이전트의 출력을 입력으로 받을 수 있습니다. 마지막으로, 이렇게 구성된 Crew는 나중에 Task들을 할당받아 실행하게 됩니다. researcher가 먼저 정보를 수집하고, 그 결과를 writer가 받아서 콘텐츠를 작성하는 파이프라인이 자동으로 형성됩니다. 여러분이 이 코드를 사용하면 각 에이전트의 프롬프트를 분리하여 관리할 수 있고, 특정 단계만 개선하거나 교체하기가 쉬워집니다. 또한 각 에이전트의 출력을 개별적으로 모니터링하고 평가할 수 있어 전체 시스템의 품질 관리가 용이합니다. ### 실전 팁 💡 각 에이전트의 role과 goal은 구체적이고 명확하게 작성하세요. 모호한 역할 정의는 에이전트의 출력 품질을 저하시킵니다. 💡 초기 개발 시에는 반드시 verbose=True로 설정하여 각 에이전트의 사고 과정을 확인하세요. 예상치 못한 동작을 빠르게 발견할 수 있습니다. 💡 에이전트의 backstory는 단순한 설명이 아니라 LLM의 행동을 유도하는 프롬프트 엔지니어링 요소입니다. 원하는 출력 스타일이나 전문성 수준을 backstory에 녹여내세요. 💡 너무 많은 에이전트를 한 Crew에 넣으면 조율 비용이 증가합니다. 처음에는 2-3개의 핵심 에이전트로 시작하여 검증한 후 확장하세요. 💡 각 에이전트를 독립적으로 테스트할 수 있는 단위 테스트를 작성하세요. 전체 Crew를 실행하기 전에 개별 에이전트의 동작을 검증하면 디버깅 시간을 크게 줄일 수 있습니다. --- ## 2. Agent Role과 Goal 설계 ### 시작하며 여러분이 팀 프로젝트에서 각 팀원의 역할이 불명확해서 업무가 중복되거나 누락되는 상황을 겪어본 적 있나요? "이거 누가 하는 거였지?"라는 질문이 나오는 순간, 프로젝트는 이미 비효율의 늪에 빠진 것입니다. AI 에이전트 시스템도 마찬가지입니다. 각 에이전트의 역할과 목표가 명확하지 않으면 에이전트들이 같은 작업을 반복하거나, 중요한 작업을 아무도 하지 않는 문제가 발생합니다. 특히 LLM 기반 에이전트는 명시적인 지시가 없으면 자신의 역할을 벗어나 다른 일을 시도하기도 합니다. 바로 이럴 때 필요한 것이 명확한 Role과 Goal 설계입니다. 각 에이전트의 책임 범위와 달성 목표를 구체적으로 정의하면, 협업의 효율성이 극대화됩니다. ### 개요 간단히 말해서, Agent의 Role은 "무엇을 하는 사람인가"를 정의하고, Goal은 "무엇을 달성해야 하는가"를 명시하는 것입니다. 실무에서는 에이전트의 역할 설계가 전체 시스템의 성공을 좌우합니다. 예를 들어, 고객 지원 시스템을 구축할 때 "문의 분류 에이전트", "기술 지원 에이전트", "에스컬레이션 에이전트"처럼 각자의 전문 영역을 명확히 나누면, 각 에이전트가 자신의 강점에 집중할 수 있습니다. 기존에는 "고객 지원 AI"라는 하나의 범용 에이전트를 만들었다면, 이제는 각 단계별로 최적화된 전문 에이전트를 배치하여 전체 품질을 높일 수 있습니다. Role 설계의 핵심은 단일 책임 원칙(Single Responsibility Principle)을 따르는 것이고, Goal 설계의 핵심은 측정 가능하고 구체적인 목표를 설정하는 것입니다. 또한 에이전트의 제약 사항(constraints)도 함께 정의하여 역할을 벗어나지 않도록 해야 합니다. 이러한 명확한 정의가 에이전트 간 협업의 품질을 결정합니다. ### 코드 예제 ```python from crewai import Agent # 데이터 분석 전문 에이전트 data_analyst = Agent( role='Senior Data Analyst', goal='Extract insights from data and identify trends', backstory="""You are an expert data analyst with 10 years of experience. You excel at finding patterns in complex datasets and presenting actionable insights. You never make assumptions without data.""", verbose=True, allow_delegation=False # 다른 에이전트에게 작업 위임 금지 ) # 비즈니스 전략 에이전트 strategist = Agent( role='Business Strategist', goal='Create actionable business recommendations based on data insights', backstory="""You are a strategic thinker who transforms data insights into concrete business actions. You consider market conditions, risks, and implementation feasibility.""", verbose=True, allow_delegation=True # 필요시 다른 에이전트에게 위임 가능 ) ``` ### 설명 이것이 하는 일: 이 코드는 두 개의 전문 에이전트를 생성하면서 각자의 역할, 목표, 배경을 명확히 구분하여 정의합니다. 첫 번째로, data_analyst는 데이터 분석에만 집중하도록 설계되었습니다. role을 'Senior Data Analyst'로 설정하여 LLM이 시니어 수준의 분석 능력을 발휘하도록 유도하고, goal을 "Extract insights from data and identify trends"로 명시하여 데이터에서 인사이트를 추출하는 것이 주 임무임을 분명히 합니다. backstory에서 "never make assumptions without data"라고 명시하여 데이터 기반 의사결정만 하도록 제약을 걸었고, allow_delegation=False로 설정하여 이 에이전트가 다른 에이전트에게 작업을 넘기지 못하도록 했습니다. 그 다음으로, strategist는 비즈니스 전략 수립에 특화되어 있습니다. 이 에이전트는 data_analyst의 분석 결과를 받아서 실행 가능한 비즈니스 액션으로 전환하는 역할을 합니다. backstory에서 "market conditions, risks, and implementation feasibility"를 고려하도록 명시하여 단순한 제안이 아닌 실현 가능한 전략을 만들도록 유도합니다. allow_delegation=True로 설정하여 필요시 다른 에이전트(예: 시장 조사 에이전트)에게 추가 작업을 요청할 수 있습니다. 마지막으로, 이 두 에이전트의 역할 분리는 분석과 전략이라는 서로 다른 사고방식을 요구하는 작업을 각각의 전문 에이전트가 처리하도록 합니다. 데이터 분석은 객관적이고 사실 기반이어야 하고, 전략 수립은 창의적이고 미래 지향적이어야 하므로, 이 둘을 분리하면 각 단계의 품질이 높아집니다. 여러분이 이 패턴을 사용하면 각 에이전트의 프롬프트를 최적화하기 쉬워지고, 특정 단계에서 문제가 발생했을 때 해당 에이전트만 수정하면 됩니다. 또한 allow_delegation을 적절히 설정하여 에이전트 간 작업 흐름을 제어할 수 있습니다. ### 실전 팁 💡 backstory는 단순한 설명이 아니라 에이전트의 행동 패턴을 정의하는 프롬프트입니다. "You never...", "You always..."와 같은 제약 조건을 명시하여 일관된 출력을 보장하세요. 💡 allow_delegation은 신중하게 설정하세요. 무분별한 위임은 작업 흐름을 예측 불가능하게 만들고, 과도한 LLM 호출로 비용을 증가시킵니다. 💡 role 이름에 시니어티 레벨(Junior, Senior, Lead 등)을 포함하면 LLM의 출력 품질과 자신감 수준이 달라집니다. 실험을 통해 최적의 레벨을 찾으세요. 💡 goal은 동사로 시작하는 구체적인 행동 문장으로 작성하세요. "Analyze data" 대신 "Extract actionable insights from sales data to identify growth opportunities"처럼 구체적으로 작성하면 더 나은 결과를 얻습니다. 💡 여러 에이전트가 비슷한 역할을 하고 있다면 역할 설계를 재검토하세요. 각 에이전트는 명확히 구분되는 고유한 가치를 제공해야 합니다. --- ## 3. CrewAI 기본 구조 ### 시작하며 여러분이 여러 마이크로서비스를 관리할 때, 각 서비스는 잘 작동하는데 이들을 연결하고 조율하는 게 더 어려웠던 경험 있으신가요? 각 서비스의 입출력 형식을 맞추고, 실행 순서를 정하고, 에러를 처리하는 통합 레이어가 필요합니다. Multi-Agent 시스템도 똑같은 문제를 겪습니다. 각 에이전트는 잘 만들어졌지만, 이들을 어떤 순서로 실행할지, 어떻게 데이터를 전달할지, 에러가 발생하면 어떻게 처리할지 등 오케스트레이션이 복잡합니다. 직접 이 모든 로직을 구현하면 비즈니스 로직보다 인프라 코드가 더 많아지죠. 바로 이럴 때 필요한 것이 CrewAI 같은 Multi-Agent 프레임워크입니다. 에이전트 정의, 작업 할당, 실행 흐름 제어를 선언적으로 관리할 수 있어 복잡도를 크게 줄여줍니다. ### 개요 간단히 말해서, CrewAI는 Agent, Task, Crew라는 세 가지 핵심 개념으로 Multi-Agent 시스템을 구조화하는 프레임워크입니다. 실무에서는 복잡한 AI 워크플로우를 안정적으로 운영하기 위해 검증된 프레임워크가 필수적입니다. 예를 들어, 콘텐츠 생성 파이프라인을 구축할 때 리서치, 작성, 편집, SEO 최적화 단계를 각각의 Task로 정의하고, 각 Task를 담당할 Agent를 할당한 후, Crew가 전체 실행을 관리하도록 할 수 있습니다. 기존에는 각 에이전트 호출을 직접 코딩하고 결과를 수동으로 전달해야 했다면, 이제는 Task 간 의존성만 선언하면 CrewAI가 자동으로 실행 순서를 결정하고 데이터를 전달합니다. CrewAI의 핵심 특징은 선언적 구성(declarative configuration), 자동 오케스트레이션, 내장된 에러 처리입니다. Task 간 의존성 그래프를 자동으로 해석하여 최적의 실행 순서를 결정하고, 특정 Task가 실패해도 재시도하거나 우회할 수 있습니다. 이러한 추상화가 개발자가 비즈니스 로직에 집중할 수 있게 해줍니다. ### 코드 예제 ```python from crewai import Agent, Task, Crew, Process # 에이전트 정의 researcher = Agent( role='Researcher', goal='Gather comprehensive information', backstory='Expert research analyst', verbose=True ) writer = Agent( role='Writer', goal='Create engaging content', backstory='Professional content writer', verbose=True ) # 작업 정의 research_task = Task( description='Research the topic: {topic}', agent=researcher, expected_output='Detailed research report' ) writing_task = Task( description='Write an article based on the research', agent=writer, expected_output='Well-written article', context=[research_task] # research_task의 출력을 입력으로 사용 ) # Crew 구성 및 실행 crew = Crew( agents=[researcher, writer], tasks=[research_task, writing_task], process=Process.sequential, # 순차 실행 verbose=True ) # 실행 result = crew.kickoff(inputs={'topic': 'AI Agent Collaboration'}) ``` ### 설명 이것이 하는 일: 이 코드는 CrewAI의 핵심 구조를 사용하여 리서치부터 글쓰기까지의 완전한 워크플로우를 정의하고 실행합니다. 첫 번째로, Agent 객체들을 생성하여 각 작업을 담당할 에이전트를 정의합니다. 이들은 독립적인 LLM 호출자로서 각자의 role과 goal에 맞게 작동합니다. 여기서는 researcher와 writer 두 명의 전문가를 정의했습니다. 그 다음으로, Task 객체들을 생성하여 실제 수행할 작업을 정의합니다. 각 Task는 description(무엇을 할지), agent(누가 할지), expected_output(어떤 형식으로 출력할지)을 명시합니다. 핵심은 writing_task의 context=[research_task] 부분인데, 이것이 Task 간 의존성을 선언하는 방법입니다. writing_task는 research_task가 완료된 후에 실행되며, research_task의 출력을 자동으로 입력으로 받습니다. 세 번째로, Crew 객체를 생성하여 전체 팀을 구성합니다. agents 리스트에 모든 에이전트를, tasks 리스트에 모든 작업을 등록합니다. process=Process.sequential은 Task들을 순차적으로 실행하라는 의미입니다(나중에 병렬 실행도 가능합니다). CrewAI는 이 정보를 바탕으로 실행 계획을 자동으로 수립합니다. 마지막으로, crew.kickoff()를 호출하여 전체 워크플로우를 시작합니다. inputs 딕셔너리로 전달한 값들은 Task의 description에서 {topic} 같은 플레이스홀더에 자동으로 주입됩니다. CrewAI가 각 Task를 순서대로 실행하고, 각 에이전트에게 적절한 프롬프트를 생성하여 LLM을 호출하고, 결과를 다음 Task로 전달합니다. 여러분이 이 구조를 사용하면 복잡한 오케스트레이션 로직을 직접 작성할 필요가 없습니다. Task 간 의존성만 선언하면 CrewAI가 알아서 실행 순서를 결정하고, 에러가 발생하면 재시도하거나 로그를 남깁니다. 또한 각 Task의 입출력이 자동으로 연결되어 수동 데이터 전달 코드가 필요 없습니다. ### 실전 팁 💡 expected_output을 명확히 정의하면 LLM이 더 일관된 형식으로 출력합니다. "JSON format with keys: title, content, tags"처럼 구체적으로 작성하세요. 💡 context는 리스트이므로 여러 Task의 출력을 하나의 Task가 입력으로 받을 수 있습니다. 예: context=[research_task, market_analysis_task] 💡 kickoff의 inputs는 모든 Task의 description에서 사용할 수 있습니다. 공통 파라미터는 여기에 정의하여 중복을 피하세요. 💡 개발 중에는 verbose=True로 설정하여 각 단계의 진행 상황을 확인하고, 프로덕션에서는 False로 설정하여 불필요한 로그를 줄이세요. 💡 Process.sequential은 디버깅이 쉽지만 느립니다. 워크플로우가 안정화되면 Process.hierarchical이나 병렬 실행을 고려하여 성능을 개선하세요. --- ## 4. Sequential Task 실행 ### 시작하며 여러분이 데이터 파이프라인을 구축할 때, 데이터 수집 → 전처리 → 분석 → 시각화 순서로 반드시 진행해야 하는 경우가 있죠? 이전 단계의 출력이 다음 단계의 입력이 되기 때문에 순서를 바꾸면 전체 파이프라인이 망가집니다. AI 에이전트 워크플로우도 마찬가지입니다. 리서치 없이 글을 쓸 수 없고, 글을 쓰지 않았는데 편집할 수 없습니다. 각 단계가 이전 단계의 결과에 의존하는 순차적 프로세스가 필요한데, 이를 수동으로 관리하면 if문과 변수 전달 코드로 가득 차게 됩니다. 바로 이럴 때 필요한 것이 Sequential Task 실행 패턴입니다. CrewAI가 Task 간 의존성을 자동으로 파악하고 순서대로 실행하며 데이터를 전달해줍니다. ### 개요 간단히 말해서, Sequential Task 실행은 여러 Task를 정해진 순서대로 하나씩 실행하면서, 각 Task의 출력을 다음 Task의 입력으로 자동 전달하는 패턴입니다. 실무에서는 대부분의 복잡한 워크플로우가 순차적 구조를 가집니다. 예를 들어, 고객 피드백 분석 시스템을 만들 때 "피드백 수집 → 감정 분석 → 카테고리 분류 → 우선순위 결정 → 액션 아이템 생성" 같은 명확한 순서가 있습니다. 각 단계를 독립적인 Task로 만들고 순차 실행하면, 각 단계를 개별적으로 테스트하고 개선할 수 있습니다. 기존에는 step1_output = agent1.run(), step2_output = agent2.run(step1_output) 식으로 수동 체이닝을 했다면, 이제는 context 속성으로 의존성만 선언하면 CrewAI가 자동으로 데이터를 흘려보냅니다. Sequential 실행의 핵심은 데이터 흐름의 명확성과 디버깅 용이성입니다. 각 단계의 입출력을 로그로 확인할 수 있어 어디서 문제가 생겼는지 쉽게 파악할 수 있고, 특정 단계만 재실행하거나 건너뛸 수도 있습니다. 순차 실행은 병렬 실행보다 느리지만, 복잡한 의존성이 있는 워크플로우에서는 가장 안전하고 예측 가능한 방법입니다. ### 코드 예제 ```python from crewai import Agent, Task, Crew, Process # 단계별 에이전트 정의 collector = Agent(role='Data Collector', goal='Gather raw data', backstory='Expert at data collection') analyzer = Agent(role='Data Analyzer', goal='Analyze collected data', backstory='Skilled data analyst') reporter = Agent(role='Report Writer', goal='Create actionable reports', backstory='Professional report writer') # 순차적 Task 체인 정의 task1 = Task( description='Collect customer feedback from {source}', agent=collector, expected_output='Raw feedback data in JSON format' ) task2 = Task( description='Analyze sentiment and categorize feedback', agent=analyzer, expected_output='Analyzed data with sentiment scores', context=[task1] # task1의 출력을 입력으로 사용 ) task3 = Task( description='Generate executive summary and action items', agent=reporter, expected_output='Executive report with recommendations', context=[task2] # task2의 출력을 입력으로 사용 ) # Sequential 실행 crew = Crew( agents=[collector, analyzer, reporter], tasks=[task1, task2, task3], process=Process.sequential, verbose=True ) result = crew.kickoff(inputs={'source': 'customer_survey_2024'}) ``` ### 설명 이것이 하는 일: 이 코드는 데이터 수집부터 리포트 생성까지 3단계로 이루어진 순차적 워크플로우를 구현합니다. 첫 번째로, 각 단계를 담당할 전문 에이전트를 정의합니다. collector는 데이터 수집만, analyzer는 분석만, reporter는 리포트 작성만 담당하도록 역할을 분리했습니다. 이렇게 하면 각 에이전트의 프롬프트를 해당 작업에 최적화할 수 있고, 특정 단계에 문제가 있을 때 해당 에이전트만 수정하면 됩니다. 그 다음으로, Task들을 정의하면서 context 속성으로 의존성을 명시합니다. task2는 context=[task1]로 설정되어 있어서 task1이 완료되기 전까지 실행되지 않으며, task1의 출력(Raw feedback data)을 자동으로 입력으로 받습니다. 마찬가지로 task3은 task2가 완료될 때까지 대기하고, task2의 분석 결과를 받아서 리포트를 작성합니다. CrewAI는 이 context 정보를 바탕으로 실행 순서를 자동으로 결정합니다. 세 번째로, Crew를 생성할 때 process=Process.sequential로 설정하여 순차 실행 모드를 활성화합니다. 이 모드에서는 tasks 리스트의 순서와 각 Task의 context를 종합하여 실행 계획을 수립합니다. 만약 context가 없더라도 리스트 순서대로 실행되지만, context가 있으면 의존성을 우선합니다. 마지막으로, kickoff()를 호출하면 CrewAI가 task1부터 시작합니다. task1이 완료되면 그 출력을 task2의 프롬프트에 자동으로 포함시켜 analyzer 에이전트를 호출하고, task2가 완료되면 그 결과를 task3에 전달합니다. 각 단계의 진행 상황은 verbose=True 덕분에 콘솔에 출력되어 실시간으로 모니터링할 수 있습니다. 여러분이 이 패턴을 사용하면 복잡한 데이터 파이프라인을 선언적으로 정의할 수 있고, 각 단계를 독립적으로 테스트하거나 교체할 수 있습니다. 또한 특정 단계에서 에러가 발생해도 해당 Task부터 재시작할 수 있어 전체 프로세스를 처음부터 다시 실행할 필요가 없습니다. ### 실전 팁 💡 각 Task의 expected_output을 다음 Task에서 필요한 형식으로 명시하세요. "JSON with keys: sentiment, category, priority"처럼 구체적으로 정의하면 데이터 전달이 원활합니다. 💡 Sequential 실행은 각 단계를 기다려야 하므로 느릴 수 있습니다. 병렬화할 수 있는 Task가 있는지 검토하여 성능을 개선하세요. 💡 context 리스트에 여러 Task를 넣으면 여러 이전 단계의 출력을 한 번에 받을 수 있습니다. 예: context=[task1, task2]는 두 Task의 출력을 모두 입력으로 사용합니다. 💡 중간 단계의 출력을 로깅하거나 저장하려면 Task에 콜백 함수를 추가하세요. 디버깅과 모니터링에 매우 유용합니다. 💡 Task 간 데이터 전달 시 토큰 제한을 고려하세요. 이전 Task의 출력이 너무 길면 요약하거나 핵심 정보만 추출하는 중간 Task를 추가하는 것이 좋습니다. --- ## 5. Parallel Task 실행 ### 시작하며 여러분이 여러 API를 동시에 호출해서 결과를 취합해야 할 때, 순차적으로 하나씩 호출하면 너무 느려서 답답했던 경험 있으시죠? 각 API 호출이 2초씩 걸린다면, 5개를 순차 호출하면 10초가 걸리지만 병렬로 호출하면 2초면 끝납니다. AI 에이전트 워크플로우에서도 똑같은 문제가 있습니다. 여러 독립적인 리서치 작업을 순차적으로 하면 시간이 너무 오래 걸립니다. 예를 들어, "기술 트렌드 조사", "경쟁사 분석", "고객 리뷰 수집"은 서로 의존하지 않으므로 동시에 진행할 수 있는데, Sequential 방식으로만 하면 비효율적입니다. 바로 이럴 때 필요한 것이 Parallel Task 실행 패턴입니다. 서로 독립적인 Task들을 동시에 실행하여 전체 워크플로우 시간을 크게 단축할 수 있습니다. ### 개요 간단히 말해서, Parallel Task 실행은 서로 의존하지 않는 여러 Task를 동시에 실행하여 전체 처리 시간을 최소화하는 패턴입니다. 실무에서는 데이터 수집 단계처럼 여러 독립적인 소스에서 정보를 가져올 때 병렬화가 필수적입니다. 예를 들어, 시장 조사 시스템에서 "뉴스 기사 수집", "소셜 미디어 트렌드 분석", "경쟁사 웹사이트 크롤링"은 서로 독립적이므로 동시에 진행하고, 모든 결과가 모이면 다음 단계(종합 분석)로 넘어가는 구조가 효율적입니다. 기존에는 asyncio나 threading을 사용해 직접 병렬 처리를 구현해야 했다면, 이제는 CrewAI가 Task 간 의존성 그래프를 분석하여 자동으로 병렬화 가능한 Task를 식별하고 동시에 실행합니다. Parallel 실행의 핵심은 독립성 식별과 결과 동기화입니다. 어떤 Task들이 서로 의존하지 않는지 정확히 파악하는 것이 중요하고, 모든 병렬 Task가 완료될 때까지 다음 단계가 기다려야 합니다. 잘못 설계하면 race condition이나 데이터 불일치가 발생할 수 있으므로 주의가 필요합니다. ### 코드 예제 ```python from crewai import Agent, Task, Crew, Process # 독립적인 리서치 에이전트들 tech_researcher = Agent(role='Tech Researcher', goal='Research technology trends', backstory='Tech expert') market_researcher = Agent(role='Market Researcher', goal='Analyze market data', backstory='Market analyst') competitor_researcher = Agent(role='Competitor Researcher', goal='Analyze competitors', backstory='Competition analyst') # 병렬로 실행될 독립적인 Task들 task_tech = Task( description='Research latest AI technology trends', agent=tech_researcher, expected_output='Technology trend report' ) task_market = Task( description='Analyze current market size and growth', agent=market_researcher, expected_output='Market analysis report' ) task_competitor = Task( description='Analyze top 3 competitors', agent=competitor_researcher, expected_output='Competitor analysis report' ) # 병렬 Task의 결과를 종합하는 Task synthesis_agent = Agent(role='Strategy Synthesizer', goal='Create comprehensive strategy', backstory='Strategic planner') task_synthesis = Task( description='Synthesize all research into actionable strategy', agent=synthesis_agent, expected_output='Comprehensive strategy document', context=[task_tech, task_market, task_competitor] # 3개 Task 완료 후 실행 ) # Crew 구성 - Process.hierarchical은 병렬+순차 혼합 지원 crew = Crew( agents=[tech_researcher, market_researcher, competitor_researcher, synthesis_agent], tasks=[task_tech, task_market, task_competitor, task_synthesis], process=Process.hierarchical, # 의존성 기반 자동 병렬화 verbose=True ) result = crew.kickoff() ``` ### 설명 이것이 하는 일: 이 코드는 3개의 독립적인 리서치 Task를 병렬로 실행하고, 모든 결과를 종합하는 워크플로우를 구현합니다. 첫 번째로, 서로 다른 영역을 담당하는 3개의 전문 에이전트를 정의합니다. tech_researcher, market_researcher, competitor_researcher는 각자의 전문 분야에서 독립적으로 작업하므로 서로의 결과를 기다릴 필요가 없습니다. 이런 독립성이 병렬화의 핵심 조건입니다. 그 다음으로, 각 에이전트에 대응하는 Task를 생성합니다. task_tech, task_market, task_competitor는 모두 context 속성이 없거나 비어있습니다. 이것이 "이 Task들은 다른 Task에 의존하지 않는다"는 신호이며, CrewAI는 이를 보고 병렬 실행 가능하다고 판단합니다. 각 Task는 자신의 에이전트를 사용하여 독립적으로 LLM을 호출합니다. 세 번째로, task_synthesis는 context=[task_tech, task_market, task_competitor]로 설정되어 있어서 3개의 병렬 Task가 모두 완료될 때까지 대기합니다. CrewAI는 이 의존성을 보고 "먼저 3개를 병렬로 실행하고, 모두 완료되면 synthesis를 실행한다"는 실행 계획을 수립합니다. task_synthesis는 3개 Task의 출력을 모두 입력으로 받아서 종합 분석을 수행합니다. 마지막으로, process=Process.hierarchical은 Task 간 의존성 그래프를 분석하여 최적의 실행 전략을 자동으로 결정합니다. 의존성이 없는 Task들은 병렬로 실행하고, 의존성이 있는 Task는 순차적으로 실행합니다. 이 예제에서는 1단계(3개 병렬) → 2단계(1개 순차) 구조가 자동으로 형성됩니다. 여러분이 이 패턴을 사용하면 전체 실행 시간을 크게 단축할 수 있습니다. 3개 Task가 각각 30초 걸린다면, 순차 실행은 90초가 걸리지만 병렬 실행은 30초면 끝납니다. 특히 LLM API 호출처럼 I/O bound 작업에서 병렬화 효과가 큽니다. ### 실전 팁 💡 병렬 Task들이 같은 외부 리소스(DB, API)에 접근한다면 rate limit이나 동시성 제한을 고려하세요. 너무 많은 병렬 호출은 오히려 에러를 유발할 수 있습니다. 💡 각 병렬 Task의 실행 시간이 비슷하도록 작업을 분배하세요. 하나가 10초, 다른 하나가 60초 걸리면 병렬화 효과가 줄어듭니다. 💡 LLM API에는 보통 RPM(Requests Per Minute) 제한이 있습니다. 병렬 Task 수가 이 제한을 초과하지 않도록 주의하고, 필요하면 배치로 나누어 실행하세요. 💡 병렬 Task 중 하나가 실패해도 다른 Task는 계속 실행되도록 에러 핸들링 전략을 설정하세요. 부분적인 결과라도 활용할 수 있습니다. 💡 debugging 시에는 병렬 실행을 일시적으로 비활성화하고 순차 실행으로 전환하면 로그를 읽기 쉽습니다. Process.sequential로 바꾸면 됩니다. --- ## 6. Agent 간 통신과 메모리 공유 ### 시작하며 여러분이 팀 프로젝트를 할 때, 각자 독립적으로 일하다가 나중에 통합하려니 이미 다른 팀원이 해결한 문제를 또 풀고 있거나, 중요한 결정 사항을 공유받지 못해 잘못된 방향으로 작업한 경험 있으신가요? 팀 협업의 핵심은 정보 공유입니다. Multi-Agent 시스템도 똑같습니다. 각 에이전트가 고립되어 작동하면 중복 작업이 발생하고, 이전 단계에서 발견한 중요한 정보를 다음 에이전트가 활용하지 못합니다. 특히 긴 워크플로우에서는 초반 에이전트의 발견이 후반 에이전트에게 매우 중요할 수 있습니다. 바로 이럴 때 필요한 것이 Agent 간 메모리 공유와 통신 메커니즘입니다. 에이전트들이 공통 지식을 축적하고 참조할 수 있게 하여 협업의 품질을 높입니다. ### 개요 간단히 말해서, Agent 간 메모리 공유는 여러 에이전트가 공통의 지식 저장소를 읽고 쓰면서 정보를 축적하고 재사용하는 메커니즘입니다. 실무에서는 장기 실행되는 워크플로우나 반복적인 작업에서 메모리가 필수적입니다. 예를 들어, 고객 지원 시스템에서 "문의 분류 에이전트"가 파악한 고객의 기술 수준을 "기술 지원 에이전트"가 참조하여 답변의 난이도를 조절할 수 있습니다. 또한 이전 대화 내용을 기억하여 맥락을 유지하는 것도 메모리 공유의 중요한 사용 사례입니다. 기존에는 각 에이전트 호출마다 전체 대화 기록을 프롬프트에 포함시켜야 했다면, 이제는 공유 메모리에서 관련 정보만 검색하여 토큰을 절약하면서도 맥락을 유지할 수 있습니다. 메모리 공유의 핵심 특징은 지속성(persistence), 선택적 검색(selective retrieval), 컨텍스트 관리입니다. 모든 정보를 매번 전달하면 토큰 낭비가 심하므로, 현재 Task와 관련된 정보만 메모리에서 검색하여 제공하는 것이 중요합니다. CrewAI는 short-term memory(현재 세션), long-term memory(영구 저장), entity memory(특정 엔티티 관련 정보)를 지원합니다. ### 코드 예제 ```python from crewai import Agent, Task, Crew, Process from crewai.memory import ShortTermMemory, LongTermMemory, EntityMemory # 메모리 활성화된 Crew 설정 crew = Crew( agents=[agent1, agent2], tasks=[task1, task2], memory=True, # 메모리 활성화 verbose=True ) # 메모리에 정보를 명시적으로 저장하는 에이전트 researcher = Agent( role='Researcher', goal='Research and store findings in memory', backstory="""You are a researcher who carefully documents findings. Store important discoveries in memory for other agents to use.""", verbose=True, memory=True # 에이전트별 메모리 활성화 ) # 메모리를 참조하는 에이전트 writer = Agent( role='Writer', goal='Write content using researched information from memory', backstory="""You are a writer who checks memory for research findings before writing. Use stored information to create accurate content.""", verbose=True, memory=True ) research_task = Task( description='Research {topic} and store key findings', agent=researcher, expected_output='Research summary with key facts' ) writing_task = Task( description='Write article about {topic} using memory', agent=writer, expected_output='Well-researched article', context=[research_task] ) crew = Crew( agents=[researcher, writer], tasks=[research_task, writing_task], memory=True, # Crew 레벨 메모리 공유 verbose=True ) result = crew.kickoff(inputs={'topic': 'Multi-Agent AI Systems'}) ``` ### 설명 이것이 하는 일: 이 코드는 CrewAI의 메모리 시스템을 활성화하여 에이전트 간 정보 공유를 가능하게 합니다. 첫 번째로, Crew 생성 시 memory=True로 설정하면 CrewAI가 자동으로 메모리 시스템을 초기화합니다. 이 메모리는 모든 에이전트가 접근할 수 있는 공유 저장소로 작동하며, 기본적으로 short-term memory(현재 실행 세션 동안 유지)가 활성화됩니다. 필요하면 long-term memory를 설정하여 데이터베이스에 영구 저장할 수도 있습니다. 그 다음으로, 각 Agent에도 memory=True를 설정하여 에이전트가 메모리를 읽고 쓸 수 있게 합니다. researcher 에이전트의 backstory에 "Store important discoveries in memory"라고 명시하면, LLM이 중요한 발견을 메모리에 저장하도록 유도됩니다. CrewAI는 에이전트의 출력을 자동으로 파싱하여 메모리에 저장하거나, 에이전트가 명시적으로 메모리 write 명령을 사용할 수도 있습니다. 세 번째로, writer 에이전트의 backstory에 "checks memory for research findings"라고 명시하면, 이 에이전트가 작업을 시작하기 전에 메모리를 검색합니다. CrewAI는 현재 Task의 description과 관련된 메모리 항목을 자동으로 검색하여 프롬프트에 포함시킵니다. 이렇게 하면 research_task의 결과가 context로 명시적으로 전달되는 것 외에도, 메모리에 저장된 세부 정보들도 활용할 수 있습니다. 마지막으로, 여러 번 kickoff()를 호출해도 메모리는 유지됩니다(long-term memory를 사용하는 경우). 예를 들어, 첫 번째 실행에서 "Multi-Agent AI Systems"에 대해 리서치한 내용이 메모리에 저장되고, 나중에 관련 주제로 다시 실행하면 이전 정보를 재사용할 수 있습니다. 이것이 반복 작업의 효율성을 크게 높입니다. 여러분이 이 메커니즘을 사용하면 에이전트들이 "팀 지식"을 축적하면서 점점 더 똑똑해집니다. 또한 긴 워크플로우에서 초반 에이전트의 발견을 후반 에이전트가 자연스럽게 참조할 수 있어, 수동으로 모든 정보를 전달할 필요가 없습니다. ### 실전 팁 💡 메모리가 계속 쌓이면 검색 품질이 저하될 수 있습니다. 주기적으로 메모리를 정리하거나, 관련성 낮은 오래된 항목을 제거하는 전략을 수립하세요. 💡 프로덕션 환경에서는 long-term memory를 데이터베이스(PostgreSQL + pgvector 등)에 연결하여 세션 간 지식을 유지하세요. 개발 중에는 short-term memory로 충분합니다. 💡 민감한 정보가 메모리에 저장되지 않도록 주의하세요. 개인정보나 API 키 같은 데이터는 메모리에서 제외하거나 암호화해야 합니다. 💡 메모리 검색은 임베딩 기반 유사도 검색을 사용하므로, Task description을 명확하고 구체적으로 작성하면 관련 메모리를 더 잘 찾습니다. 💡 memory=True를 활성화하면 추가 LLM 호출(임베딩 생성)이 발생하여 비용과 지연이 증가합니다. 메모리가 정말 필요한 워크플로우에만 사용하세요. --- ## 7. Tool Sharing과 Delegation ### 시작하며 여러분이 개발팀에서 일할 때, 각자 자기만의 도구를 사용하다가 팀원이 유용한 라이브러리를 발견하면 공유해서 모두가 생산성을 높인 경험 있으시죠? 또한 자신이 감당하기 어려운 작업은 전문가에게 위임하는 것도 효율적인 협업의 핵심입니다. Multi-Agent 시스템에서도 마찬가지입니다. 특정 에이전트가 웹 검색 도구를 가지고 있다면 다른 에이전트도 사용할 수 있어야 하고, 한 에이전트가 자신의 역량을 벗어난 작업을 만나면 적절한 전문가 에이전트에게 위임할 수 있어야 합니다. 그렇지 않으면 각 에이전트가 모든 도구를 중복으로 가지거나, 부적절한 에이전트가 작업을 억지로 수행하게 됩니다. 바로 이럴 때 필요한 것이 Tool Sharing과 Delegation 메커니즘입니다. 도구를 공유하여 중복을 줄이고, 작업을 적절한 에이전트에게 위임하여 품질을 높입니다. ### 개요 간단히 말해서, Tool Sharing은 여러 에이전트가 공통 도구(함수, API)를 공유하는 것이고, Delegation은 한 에이전트가 다른 에이전트에게 하위 작업을 위임하는 메커니즘입니다. 실무에서는 복잡한 작업을 수행할 때 에이전트가 모든 것을 혼자 할 수 없습니다. 예를 들어, 콘텐츠 작성 에이전트가 최신 통계 데이터가 필요하면 리서치 에이전트에게 "2024년 AI 시장 규모 조사" 작업을 위임하고, 그 결과를 받아서 글을 완성할 수 있습니다. 이렇게 하면 각 에이전트가 자신의 전문 영역에만 집중하면서도 필요할 때 협력할 수 있습니다. 기존에는 모든 에이전트에게 모든 도구를 할당하거나, 에이전트가 직접 다른 에이전트를 호출하는 복잡한 로직을 작성해야 했다면, 이제는 allow_delegation=True 설정만으로 에이전트가 자율적으로 작업을 위임할 수 있습니다. Tool Sharing의 핵심은 재사용성이고, Delegation의 핵심은 전문성 활용입니다. 도구를 중앙에서 관리하면 버전 관리와 업데이트가 쉬워지고, 위임을 활용하면 각 에이전트의 프롬프트를 단순하게 유지하면서도 복잡한 작업을 수행할 수 있습니다. ### 코드 예제 ```python from crewai import Agent, Task, Crew from crewai_tools import SerperDevTool, ScrapeWebsiteTool # 공유 도구 정의 search_tool = SerperDevTool() # 웹 검색 도구 scrape_tool = ScrapeWebsiteTool() # 웹 스크래핑 도구 # 도구를 공유하는 여러 에이전트 researcher = Agent( role='Researcher', goal='Find accurate information', backstory='Expert researcher', tools=[search_tool, scrape_tool], # 두 도구 모두 사용 가능 verbose=True, allow_delegation=False # 위임 불가 ) writer = Agent( role='Writer', goal='Write compelling content', backstory='Professional writer', tools=[search_tool], # 검색 도구만 사용 가능 verbose=True, allow_delegation=True # 다른 에이전트에게 위임 가능 ) editor = Agent( role='Editor', goal='Ensure content quality and accuracy', backstory='Experienced editor', tools=[], # 도구 없음, 위임만 사용 verbose=True, allow_delegation=True # 필요시 리서치나 수정 작업 위임 ) task_write = Task( description='Write an article about {topic}. If you need additional research, delegate to researcher.', agent=writer, expected_output='Article draft' ) task_edit = Task( description='Edit and fact-check the article. Delegate verification to researcher if needed.', agent=editor, expected_output='Final polished article', context=[task_write] ) crew = Crew( agents=[researcher, writer, editor], tasks=[task_write, task_edit], verbose=True ) result = crew.kickoff(inputs={'topic': 'AI Agent Collaboration'}) ``` ### 설명 이것이 하는 일: 이 코드는 여러 에이전트가 도구를 공유하고, 필요시 서로에게 작업을 위임하는 협업 시스템을 구현합니다. 첫 번째로, search_tool과 scrape_tool을 Crew 외부에서 정의하여 여러 에이전트가 같은 도구 인스턴스를 공유합니다. 이렇게 하면 도구의 설정이나 API 키를 한 곳에서 관리할 수 있고, 도구를 업그레이드할 때도 한 번만 수정하면 됩니다. CrewAI는 에이전트가 도구를 호출할 때 자동으로 프롬프트에 도구 사용 지침을 포함시킵니다. 그 다음으로, 각 에이전트의 tools 리스트에 필요한 도구만 할당합니다. researcher는 검색과 스크래핑 모두 필요하므로 두 도구를 가지고, writer는 검색만 필요하므로 search_tool만 가집니다. editor는 도구 없이 위임만 사용하도록 설계했습니다. 이렇게 최소 권한 원칙을 적용하면 에이전트가 불필요한 도구를 오용하는 것을 방지할 수 있습니다. 세 번째로, allow_delegation 속성을 선택적으로 활성화합니다. writer와 editor는 allow_delegation=True로 설정되어 있어서, 작업 중에 자신이 처리하기 어려운 부분을 만나면 다른 에이전트에게 위임할 수 있습니다. 예를 들어, writer가 글을 쓰다가 "최신 통계가 필요하다"고 판단하면, researcher에게 "2024 AI market statistics 조사" 작업을 자동으로 위임하고 결과를 기다립니다. CrewAI가 이 위임 프로세스를 자동으로 관리합니다. 마지막으로, Task의 description에 "delegate to researcher if needed"라고 명시하여 에이전트가 위임 옵션을 인지하도록 합니다. LLM은 이 지시를 보고 필요시 위임 명령을 생성합니다. editor가 fact-check을 하다가 의심스러운 정보를 발견하면, researcher에게 검증 작업을 위임하여 정확성을 높일 수 있습니다. 여러분이 이 패턴을 사용하면 각 에이전트를 단순하게 유지하면서도 복잡한 작업을 수행할 수 있습니다. 또한 도구를 중앙 관리하여 일관성을 유지하고, 위임을 통해 동적으로 협업 구조를 형성할 수 있습니다. ### 실전 팁 💡 allow_delegation=True는 강력하지만 예측 불가능성을 증가시킵니다. 프로덕션에서는 위임 로그를 모니터링하여 불필요한 위임이 발생하지 않는지 확인하세요. 💡 위임이 무한 루프에 빠지지 않도록 최대 위임 깊이를 설정하세요. A가 B에게 위임하고, B가 다시 A에게 위임하는 상황을 방지해야 합니다. 💡 도구는 비용이 발생할 수 있으므로(API 호출 등), 각 에이전트에 정말 필요한 도구만 할당하세요. 불필요한 도구 접근을 제한하면 오용과 비용을 줄입니다. 💡 위임 가능한 에이전트와 위임 불가능한 에이전트를 명확히 구분하세요. 리프 노드 에이전트(실제 작업 수행)는 위임 불가로, 관리자 에이전트는 위임 가능으로 설정하는 것이 일반적입니다. 💡 도구 사용 로그를 남겨서 어떤 에이전트가 어떤 도구를 얼마나 사용하는지 추적하세요. 이 데이터를 바탕으로 도구 할당을 최적화할 수 있습니다. --- ## 8. AutoGen Framework ### 시작하며 여러분이 코드 리뷰를 할 때, 작성자와 리뷰어가 여러 번 주고받으며 개선해나가는 과정 겪어보셨죠? "이 부분 개선이 필요합니다" → "이렇게 수정했습니다" → "좋네요, 이제 이 부분은?" 같은 대화형 협업이 품질을 높입니다. CrewAI는 주로 정해진 워크플로우를 실행하는데 최적화되어 있지만, 때로는 에이전트들이 자유롭게 대화하며 문제를 해결해야 할 때가 있습니다. 특정 주제에 대해 여러 관점을 가진 에이전트들이 토론하거나, 한 에이전트의 출력을 다른 에이전트가 반복적으로 개선하는 시나리오에서는 대화형 접근이 효과적입니다. 바로 이럴 때 필요한 것이 Microsoft의 AutoGen 프레임워크입니다. 에이전트 간 대화를 중심으로 설계되어 동적이고 유연한 협업을 지원합니다. ### 개요 간단히 말해서, AutoGen은 여러 AI 에이전트가 자연어 대화를 통해 협업하여 문제를 해결하는 대화형 Multi-Agent 프레임워크입니다. 실무에서는 정형화되지 않은 복잡한 문제를 해결할 때 AutoGen이 유용합니다. 예를 들어, 소프트웨어 개발 시나리오에서 "개발자 에이전트"가 코드를 작성하면 "테스터 에이전트"가 버그를 찾고, 개발자가 수정하고, 테스터가 재검증하는 반복적인 대화 과정을 통해 코드 품질을 높일 수 있습니다. 기존 CrewAI가 "작업 A → 작업 B → 작업 C"의 파이프라인 구조라면, AutoGen은 "에이전트들이 목표 달성까지 자유롭게 대화"하는 구조입니다. 언제 대화를 종료할지, 누가 다음에 말할지도 동적으로 결정됩니다. AutoGen의 핵심 특징은 대화 기반 협업, 인간 개입 지원(human-in-the-loop), 코드 실행 능력입니다. AssistantAgent는 LLM 기반으로 작동하고, UserProxyAgent는 코드를 실행하거나 인간의 피드백을 대신할 수 있습니다. 또한 GroupChat으로 여러 에이전트의 토론을 조율할 수 있습니다. ### 코드 예제 ```python import autogen # LLM 설정 config_list = [{ "model": "gpt-4", "api_key": "your-api-key" }] # Assistant 에이전트 - LLM 기반 assistant = autogen.AssistantAgent( name="Assistant", llm_config={"config_list": config_list}, system_message="You are a helpful AI assistant skilled at solving problems." ) # UserProxy 에이전트 - 코드 실행 및 인간 역할 user_proxy = autogen.UserProxyAgent( name="User", human_input_mode="NEVER", # 자동 모드 (인간 개입 없음) max_consecutive_auto_reply=10, code_execution_config={"work_dir": "coding", "use_docker": False} ) # 대화 시작 - User가 Assistant에게 요청 user_proxy.initiate_chat( assistant, message="Write a Python function to calculate fibonacci numbers and test it." ) # GroupChat으로 여러 에이전트 토론 coder = autogen.AssistantAgent(name="Coder", llm_config={"config_list": config_list}) tester = autogen.AssistantAgent(name="Tester", llm_config={"config_list": config_list}) reviewer = autogen.AssistantAgent(name="Reviewer", llm_config={"config_list": config_list}) groupchat = autogen.GroupChat( agents=[user_proxy, coder, tester, reviewer], messages=[], max_round=12 ) manager = autogen.GroupChatManager(groupchat=groupchat, llm_config={"config_list": config_list}) user_proxy.initiate_chat(manager, message="Develop and test a sorting algorithm.") ``` ### 설명 이것이 하는 일: 이 코드는 AutoGen 프레임워크를 사용하여 LLM 에이전트와 실행 에이전트가 대화하며 코드를 작성하고 테스트하는 시스템을 구현합니다. 첫 번째로, AssistantAgent는 LLM(GPT-4 등)을 사용하여 사용자의 요청을 이해하고 응답을 생성하는 에이전트입니다. system_message로 에이전트의 역할을 정의하며, 이것이 LLM의 system prompt로 사용됩니다. assistant는 코드를 작성하거나, 설명을 제공하거나, 질문에 답변하는 등 범용적인 작업을 수행합니다. 그 다음으로, UserProxyAgent는 사용자를 대신하거나, 코드를 실행하는 특수한 에이전트입니다. human_input_mode="NEVER"는 인간의 입력 없이 자동으로 작동하라는 의미이고(개발/테스트용), "ALWAYS"로 설정하면 매번 인간에게 확인을 받습니다. code_execution_config는 에이전트가 생성한 Python 코드를 실제로 실행할 수 있게 하며, work_dir에서 코드 파일을 관리합니다. 이것이 AutoGen의 강력한 특징으로, LLM이 코드를 작성하면 바로 실행하고 결과를 확인할 수 있습니다. 세 번째로, initiate_chat()으로 대화를 시작합니다. user_proxy가 assistant에게 메시지를 보내면, assistant가 응답(코드 작성)하고, user_proxy가 그 코드를 실행하고, 결과를 assistant에게 전달하고, assistant가 결과를 보고 수정하는 식으로 대화가 이어집니다. max_consecutive_auto_reply=10은 최대 10번까지 자동 응답하라는 의미로, 무한 루프를 방지합니다. 마지막으로, GroupChat은 3명 이상의 에이전트가 토론하는 시나리오를 지원합니다. coder, tester, reviewer가 각자의 역할로 대화에 참여하며, GroupChatManager가 누가 다음에 말할지 결정합니다(LLM이 맥락을 보고 판단). 예를 들어, coder가 코드를 작성하면 tester가 테스트하고, 문제를 발견하면 coder에게 피드백하고, 최종적으로 reviewer가 검토하는 자연스러운 협업 흐름이 형성됩니다. 여러분이 AutoGen을 사용하면 에이전트들이 스스로 대화를 이어가며 문제를 해결하므로, 모든 단계를 미리 정의할 필요가 없습니다. 특히 코드 생성과 테스트처럼 반복적인 개선이 필요한 작업에 매우 효과적입니다. ### 실전 팁 💡 human_input_mode를 "TERMINATE"로 설정하면 에이전트가 작업을 완료했다고 판단할 때 인간에게 확인을 요청합니다. 중요한 결정이 필요한 워크플로우에 유용합니다. 💡 code_execution_config에서 use_docker=True로 설정하면 코드를 Docker 컨테이너에서 실행하여 보안을 강화할 수 있습니다. 프로덕션에서는 필수입니다. 💡 GroupChat의 max_round를 적절히 설정하세요. 너무 적으면 문제를 해결하기 전에 종료되고, 너무 많으면 비용이 과도하게 증가합니다. 💡 각 에이전트의 system_message를 명확히 구분하여 역할을 명시하세요. "You are a Python expert who writes clean code" vs "You are a tester who finds bugs"처럼 역할을 구분하면 협업 품질이 높아집니다. 💡 AutoGen은 대화 기록을 자동으로 관리하므로 토큰 사용량이 빠르게 증가합니다. 긴 대화는 주기적으로 요약하거나, 중요한 부분만 유지하는 전략을 사용하세요. --- ## 9. Consensus와 Voting ### 시작하며 여러분이 팀에서 중요한 기술 결정을 내릴 때, 여러 팀원의 의견을 듣고 투표나 합의를 통해 최선의 선택을 한 경험 있으시죠? 한 사람의 판단보다 여러 전문가의 집단 지성이 더 나은 결과를 만들어냅니다. Multi-Agent 시스템에서도 마찬가지입니다. 하나의 에이전트가 내린 결정은 편향되거나 오류가 있을 수 있지만, 여러 에이전트가 독립적으로 분석하고 투표하여 결정하면 신뢰성이 높아집니다. 특히 중요한 의사결정이나 모호한 상황에서는 합의 메커니즘이 필수적입니다. 바로 이럴 때 필요한 것이 Consensus와 Voting 알고리즘입니다. 여러 에이전트의 의견을 체계적으로 수집하고 집계하여 최종 결정을 내립니다. ### 개요 간단히 말해서, Consensus는 여러 에이전트가 같은 문제를 독립적으로 분석하고, 그 결과를 투표나 가중치 합산으로 집계하여 최종 결정을 내리는 메커니즘입니다. 실무에서는 불확실성이 높은 판단이나 중요한 의사결정에 합의 방식을 사용합니다. 예를 들어, 고객 피드백의 감정 분석에서 "긍정인지 부정인지 애매한 경우" 3개의 독립적인 감정 분석 에이전트를 실행하고 다수결로 결정하면 정확도가 높아집니다. 또한 투자 결정, 리스크 평가, 콘텐츠 필터링 같은 중요한 작업에도 적합합니다. 기존에는 하나의 LLM 호출 결과를 그대로 사용했다면, 이제는 여러 번 호출하거나 여러 모델을 사용하여 결과의 일관성과 신뢰성을 검증할 수 있습니다. Consensus의 핵심 방법은 다수결 투표(Majority Voting), 가중치 투표(Weighted Voting), 앙상블 평균(Ensemble Averaging)입니다. 각 에이전트에 동등한 투표권을 주거나, 전문성에 따라 가중치를 달리 줄 수 있습니다. 또한 수치 결과의 경우 평균이나 중앙값을 사용할 수도 있습니다. ### 코드 예제 ```python from crewai import Agent, Task, Crew from collections import Counter # 독립적인 감정 분석 에이전트 3명 analyst1 = Agent( role='Sentiment Analyst 1', goal='Analyze sentiment accurately', backstory='Conservative analyst who tends to be cautious', verbose=True ) analyst2 = Agent( role='Sentiment Analyst 2', goal='Analyze sentiment accurately', backstory='Balanced analyst with neutral perspective', verbose=True ) analyst3 = Agent( role='Sentiment Analyst 3', goal='Analyze sentiment accurately', backstory='Optimistic analyst who looks for positive aspects', verbose=True ) # 각 에이전트가 독립적으로 분석하는 Task task1 = Task( description='Analyze sentiment of: {text}. Output only: positive, negative, or neutral', agent=analyst1, expected_output='Sentiment label' ) task2 = Task( description='Analyze sentiment of: {text}. Output only: positive, negative, or neutral', agent=analyst2, expected_output='Sentiment label' ) task3 = Task( description='Analyze sentiment of: {text}. Output only: positive, negative, or neutral', agent=analyst3, expected_output='Sentiment label' ) # 병렬 실행으로 독립성 보장 crew = Crew( agents=[analyst1, analyst2, analyst3], tasks=[task1, task2, task3], verbose=True ) result = crew.kickoff(inputs={'text': 'The product is okay but could be better'}) # 결과 수집 및 다수결 투표 votes = [task1.output.raw_output, task2.output.raw_output, task3.output.raw_output] vote_counts = Counter(votes) final_decision = vote_counts.most_common(1)[0][0] print(f"Votes: {votes}") print(f"Final decision by majority: {final_decision}") ``` ### 설명 이것이 하는 일: 이 코드는 3개의 독립적인 에이전트가 같은 텍스트의 감정을 분석하고, 다수결 투표로 최종 결과를 결정하는 시스템을 구현합니다. 첫 번째로, 3개의 감정 분석 에이전트를 생성하되, 각자 다른 backstory를 부여하여 관점을 다르게 만듭니다. analyst1은 보수적(부정적 측면에 민감), analyst2는 균형잡힌, analyst3은 낙관적(긍정적 측면을 강조)으로 설정했습니다. 이렇게 다양한 관점을 가진 에이전트를 사용하면 편향을 줄이고 다각도로 분석할 수 있습니다. 그 다음으로, 각 에이전트에 동일한 작업을 할당하지만, 독립적인 Task 객체로 생성합니다. task1, task2, task3은 모두 같은 텍스트를 분석하지만, 서로 다른 에이전트를 사용하므로 독립적인 판단이 나옵니다. expected_output을 "Sentiment label"로 명시하여 출력 형식을 통일하고, description에서 "Output only: positive, negative, or neutral"로 제한하여 파싱을 쉽게 만듭니다. 세 번째로, Crew를 실행하면 3개의 Task가 병렬로(또는 순차로) 실행되어 각자의 결과를 생성합니다. 중요한 점은 각 에이전트가 서로의 결과를 보지 못하고 독립적으로 판단한다는 것입니다. context를 설정하지 않았으므로 서로 영향을 주지 않습니다. 마지막으로, kickoff() 완료 후 각 Task의 output.raw_output에서 결과를 수집하여 투표를 진행합니다. Counter를 사용하여 각 라벨의 개수를 세고, most_common(1)로 가장 많이 나온 라벨을 최종 결정으로 선택합니다. 예를 들어, votes가 ['positive', 'neutral', 'positive']라면 'positive'가 2표로 최종 결정됩니다. 여러분이 이 패턴을 사용하면 단일 에이전트의 오판을 보정할 수 있습니다. 3개 중 1개가 잘못 판단해도 나머지 2개가 맞으면 올바른 결과를 얻을 수 있습니다. 특히 LLM의 비결정성(같은 프롬프트에도 다른 응답)을 고려하면, 여러 번 실행하여 안정적인 결과를 얻는 것이 중요합니다. ### 실전 팁 💡 투표 참여자 수는 홀수로 설정하여 동점을 방지하세요. 3명, 5명, 7명이 일반적이며, 너무 많으면 비용이 증가합니다. 💡 각 에이전트의 backstory나 temperature를 다르게 설정하여 다양성을 확보하세요. 모두 똑같은 설정이면 같은 결과만 반복될 수 있습니다. 💡 가중치 투표를 구현하려면 특정 에이전트의 투표에 더 높은 가중치를 부여하세요. 예: 전문가 에이전트는 2표, 일반 에이전트는 1표. 💡 투표 결과가 너무 분산되면(예: 3명이 모두 다른 답) 신뢰도가 낮다는 신호입니다. 이 경우 더 많은 에이전트를 추가하거나, 문제를 재검토해야 합니다. 💡 수치 결과(예: 점수)의 경우 평균이나 중앙값을 사용하세요. 이상치(outlier) 영향을 줄이려면 중앙값이 더 안정적입니다. --- ## 10. Multi-Agent 오케스트레이션 ### 시작하며 여러분이 마이크로서비스 아키텍처를 운영할 때, 각 서비스는 잘 작동하지만 전체 시스템의 흐름을 제어하는 오케스트레이터(Kubernetes, 서비스 메시)가 없으면 혼란스러웠던 경험 있으시죠? 누가 언제 무엇을 실행할지, 에러가 나면 어떻게 복구할지 중앙에서 관리해야 합니다. Multi-Agent 시스템도 마찬가지입니다. 각 에이전트는 자신의 작업을 잘 수행하지만, 전체 워크플로우를 조율하고 모니터링하며 에러를 처리하는 오케스트레이션 레이어가 필요합니다. 특히 복잡한 의존성과 조건부 분기가 있는 워크플로우에서는 중앙 제어가 필수적입니다. 바로 이럴 때 필요한 것이 Multi-Agent 오케스트레이션 패턴입니다. 전체 에이전트 시스템의 실행 흐름을 제어하고, 상태를 관리하며, 예외를 처리합니다. ### 개요 간단히 말해서, Multi-Agent 오케스트레이션은 여러 에이전트의 실행 순서, 데이터 흐름, 에러 처리를 중앙에서 제어하고 모니터링하는 아키텍처 패턴입니다. 실무에서는 복잡한 비즈니스 로직을 AI 에이전트로 구현할 때 오케스트레이션이 필수입니다. 예를 들어, 고객 주문 처리 시스템에서 "재고 확인 → 결제 처리 → 배송 준비 → 알림 발송" 각 단계를 다른 에이전트가 담당하는데, 재고가 없으면 다른 공급자 조회로 분기하고, 결제 실패하면 재시도하는 등의 복잡한 로직을 오케스트레이터가 관리합니다. 기존에는 각 에이전트 호출을 코드로 직접 체이닝하고 if-else로 분기 처리했다면, 이제는 LangGraph나 CrewAI의 hierarchical process로 선언적으로 정의할 수 있습니다. 오케스트레이션의 핵심 요소는 상태 관리(state management), 조건부 라우팅(conditional routing), 에러 복구(error recovery), 모니터링(monitoring)입니다. 전체 워크플로우의 현재 상태를 추적하고, 조건에 따라 다른 경로로 분기하며, 실패 시 재시도하거나 보상 트랜잭션을 실행합니다. ### 코드 예제 ```python from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated import operator # 워크플로우 상태 정의 class WorkflowState(TypedDict): input: str research_result: str draft: str final_output: str error: str retry_count: Annotated[int, operator.add] # 각 노드(에이전트 작업)를 함수로 정의 def research_node(state: WorkflowState): try: # researcher 에이전트 실행 result = researcher.run(state['input']) return {"research_result": result, "error": ""} except Exception as e: return {"error": str(e), "retry_count": 1} def write_node(state: WorkflowState): try: # writer 에이전트 실행 draft = writer.run(state['research_result']) return {"draft": draft, "error": ""} except Exception as e: return {"error": str(e), "retry_count": 1} def edit_node(state: WorkflowState): try: # editor 에이전트 실행 final = editor.run(state['draft']) return {"final_output": final, "error": ""} except Exception as e: return {"error": str(e), "retry_count": 1} # 조건부 라우팅 - 에러 여부에 따라 분기 def should_continue(state: WorkflowState): if state.get('error'): if state.get('retry_count', 0) < 3: return "retry" else: return "fail" return "continue" # 워크플로우 그래프 구성 workflow = StateGraph(WorkflowState) # 노드 추가 workflow.add_node("research", research_node) workflow.add_node("write", write_node) workflow.add_node("edit", edit_node) # 엣지(연결) 정의 workflow.set_entry_point("research") workflow.add_conditional_edges( "research", should_continue, { "continue": "write", "retry": "research", "fail": END } ) workflow.add_edge("write", "edit") workflow.add_edge("edit", END) # 컴파일 및 실행 app = workflow.compile() result = app.invoke({"input": "AI Agent Collaboration", "retry_count": 0}) ``` ### 설명 이것이 하는 일: 이 코드는 LangGraph를 사용하여 Multi-Agent 워크플로우를 상태 머신으로 구현하고, 에러 처리와 재시도 로직을 포함합니다. 첫 번째로, WorkflowState라는 TypedDict로 전체 워크플로우에서 공유되는 상태를 정의합니다. input은 초기 입력, research_result, draft, final_output은 각 단계의 출력이고, error는 에러 메시지, retry_count는 재시도 횟수입니다. Annotated[int, operator.add]는 이 필드가 누적(accumulate)된다는 의미로, 여러 노드에서 증가시킬 수 있습니다. 그 다음으로, 각 에이전트의 작업을 노드 함수로 정의합니다. research_node는 researcher 에이전트를 실행하고, 성공하면 research_result를 상태에 저장하고, 실패하면 error와 retry_count를 설정합니다. 모든 노드는 상태를 입력으로 받아서 새로운 상태 값을 반환하며, LangGraph가 자동으로 상태를 병합합니다. try-except로 에러를 포착하여 워크플로우가 중단되지 않도록 합니다. 세 번째로, should_continue라는 조건부 라우팅 함수를 정의합니다. 이 함수는 현재 상태를 보고 다음에 어느 노드로 갈지 결정합니다. error가 있으면 retry_count를 확인하여 3회 미만이면 "retry"(같은 노드 재실행), 3회 이상이면 "fail"(워크플로우 종료)를 반환합니다. error가 없으면 "continue"를 반환하여 다음 단계로 진행합니다. 마지막으로, StateGraph로 워크플로우를 구성합니다. add_node로 각 에이전트 작업을 노드로 등록하고, add_edge로 고정된 연결(write → edit)을 정의하고, add_conditional_edges로 조건부 분기(research 이후)를 정의합니다. set_entry_point("research")는 시작 노드를 지정합니다. compile() 후 invoke()로 실행하면, LangGraph가 상태를 관리하며 노드들을 순서대로 실행하고, 조건에 따라 분기하며, 에러 시 재시도합니다. 여러분이 이 패턴을 사용하면 복잡한 워크플로우를 시각적으로 이해할 수 있는 그래프로 표현하고, 상태를 명시적으로 관리하며, 에러 처리와 재시도 로직을 선언적으로 정의할 수 있습니다. 또한 각 노드를 독립적으로 테스트하고 교체할 수 있어 유지보수가 쉽습니다. ### 실전 팁 💡 상태가 너무 커지면 메모리와 직렬화 비용이 증가합니다. 각 단계에서 필요한 최소한의 정보만 상태에 저장하세요. 💡 조건부 라우팅을 과도하게 사용하면 워크플로우가 복잡해지고 디버깅이 어려워집니다. 가능하면 단순한 선형 구조를 유지하고, 정말 필요한 곳에만 분기를 사용하세요. 💡 재시도 로직에 exponential backoff(지수 백오프)를 추가하면 일시적인 에러(네트워크 등)에 더 효과적으로 대응할 수 있습니다. 💡 각 노드의 실행 시간과 성공/실패를 로깅하여 모니터링 대시보드를 구축하세요. 어느 단계에서 병목이나 에러가 자주 발생하는지 파악할 수 있습니다. 💡 프로덕션에서는 워크플로우 상태를 데이터베이스에 저장하여 중단된 워크플로우를 재개하거나, 실패한 워크플로우를 분석할 수 있도록 하세요. --- ## 11. 에러 핸들링과 복구 전략 ### 시작하며 여러분이 분산 시스템을 운영할 때, 네트워크 장애나 서비스 다운타임이 발생해도 시스템이 계속 작동하도록 재시도, 폴백, 서킷 브레이커 같은 패턴을 적용한 경험 있으시죠? 완벽한 시스템은 없으므로, 에러를 어떻게 처리하느냐가 시스템의 안정성을 결정합니다. Multi-Agent 시스템도 똑같습니다. LLM API가 일시적으로 실패할 수 있고, 특정 에이전트가 잘못된 출력을 생성할 수 있으며, 외부 도구 호출이 타임아웃될 수 있습니다. 이런 상황에서 전체 워크플로우가 중단되면 사용자 경험이 나빠지고, 비용도 낭비됩니다. 바로 이럴 때 필요한 것이 체계적인 에러 핸들링과 복구 전략입니다. 예상 가능한 에러를 포착하고, 재시도하거나 대안 경로를 실행하여 시스템의 회복탄력성을 높입니다. ### 개요 간단히 말해서, 에러 핸들링은 Multi-Agent 워크플로우에서 발생하는 다양한 에러를 포착하고, 재시도, 폴백, 부분 성공 처리 등의 전략으로 복구하는 메커니즘입니다. 실무에서는 프로덕션 환경의 Multi-Agent 시스템이 다양한 에러에 직면합니다. LLM API의 rate limit 초과, 네트워크 타임아웃, 잘못된 출력 형식, 외부 도구 실패 등을 적절히 처리해야 시스템이 안정적으로 작동합니다. 예를 들어, 리서치 에이전트가 웹 검색에 실패하면 캐시된 데이터를 사용하거나, 다른 검색 도구로 대체하는 폴백 전략이 필요합니다. 기존에는 try-except로 간단히 처리하고 로그만 남겼다면, 이제는 에러 유형별로 다른 복구 전략(재시도, 폴백, 스킵, 인간 개입 요청)을 적용하여 가용성을 높일 수 있습니다. 에러 핸들링의 핵심 전략은 재시도(Retry with backoff), 폴백(Fallback to alternative), 부분 성공(Partial success), 서킷 브레이커(Circuit breaker)입니다. 일시적 에러는 재시도하고, 재시도가 실패하면 대안을 사용하며, 일부 작업이 실패해도 나머지는 계속 진행하고, 반복적인 실패는 빠르게 차단하여 리소스 낭비를 방지합니다. ### 코드 예제 ```python from crewai import Agent, Task, Crew import time from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type # 재시도 데코레이터 - API 에러에 대한 자동 재시도 @retry( stop=stop_after_attempt(3), # 최대 3회 재시도 wait=wait_exponential(multiplier=1, min=2, max=10), # 2초, 4초, 8초 대기 retry=retry_if_exception_type(Exception) # 모든 예외에 대해 재시도 ) def call_agent_with_retry(agent, task_description): """에이전트를 재시도 로직과 함께 호출""" return agent.execute_task(task_description) # 폴백 전략 - 주 도구 실패 시 대안 도구 사용 def research_with_fallback(topic): try: # 주 검색 도구 시도 result = primary_search_tool.run(topic) return result except Exception as e: print(f"Primary tool failed: {e}, trying fallback...") try: # 대안 검색 도구 시도 result = fallback_search_tool.run(topic) return result except Exception as e2: print(f"Fallback also failed: {e2}, using cached data...") # 최종 폴백: 캐시된 데이터 return get_cached_data(topic) # 부분 성공 처리 - 일부 에이전트 실패해도 계속 진행 def run_parallel_with_partial_success(tasks): results = [] errors = [] for task in tasks: try: result = task.execute() results.append(result) except Exception as e: errors.append({"task": task.description, "error": str(e)}) print(f"Task {task.description} failed, continuing with others...") # 최소 하나 이상 성공하면 진행 if results: return {"results": results, "errors": errors, "status": "partial_success"} else: raise Exception(f"All tasks failed: {errors}") # 에러 핸들링이 통합된 Crew class ResilientCrew: def __init__(self, agents, tasks): self.agents = agents self.tasks = tasks self.failure_count = 0 self.circuit_open = False def execute(self): # 서킷 브레이커 체크 if self.circuit_open: raise Exception("Circuit breaker is open due to repeated failures") try: results = [] for task in self.tasks: result = call_agent_with_retry(task.agent, task.description) results.append(result) # 성공 시 실패 카운트 리셋 self.failure_count = 0 return results except Exception as e: self.failure_count += 1 # 5회 연속 실패 시 서킷 오픈 if self.failure_count >= 5: self.circuit_open = True print("Circuit breaker opened after 5 failures") raise e ``` ### 설명 이것이 하는 일: 이 코드는 Multi-Agent 워크플로우에서 발생하는 다양한 에러를 처리하는 여러 전략을 구현합니다. 첫 번째로, tenacity 라이브러리의 @retry 데코레이터를 사용하여 자동 재시도를 구현합니다. stop_after_attempt(3)은 최대 3회까지 재시도하고, wait_exponential은 재시도 간격을 지수적으로 증가시켜 일시적 에러(네트워크 혼잡 등)가 해소될 시간을 줍니다. 예를 들어, LLM API의 rate limit에 걸리면 2초 후 재시도, 또 실패하면 4초 후 재시도하는 식으로 점진적으로 대기 시간을 늘립니다. 그 다음으로, research_with_fallback 함수는 다단계 폴백 전략을 구현합니다. 먼저 primary_search_tool(예: Google Search API)을 시도하고, 실패하면 fallback_search_tool(예: Bing Search API)을 시도하고, 그것도 실패하면 get_cached_data로 이전에 캐시된 결과를 반환합니다. 이렇게 하면 외부 의존성이 실패해도 시스템이 계속 작동할 수 있습니다. 세 번째로, run_parallel_with_partial_success는 여러 병렬 작업 중 일부가 실패해도 성공한 결과만 가지고 계속 진행하는 패턴입니다. 예를 들어, 5개의 리서치 소스 중 2개가 실패해도 3개의 결과로 분석을 진행할 수 있습니다. results와 errors를 모두 반환하여 호출자가 부분 성공 상황을 인지하고 적절히 처리할 수 있게 합니다. 마지막으로, ResilientCrew 클래스는 서킷 브레이커 패턴을 구현합니다. failure_count를 추적하여 연속 5회 실패하면 circuit_open을 True로 설정하고, 이후 요청을 즉시 거부합니다. 이렇게 하면 반복적으로 실패하는 작업으로 인한 리소스 낭비(비용, 시간)를 방지하고, 시스템이 빠르게 실패(fail-fast)하여 다른 복구 수단을 시도할 수 있습니다. 여러분이 이런 전략들을 조합하여 사용하면 프로덕션 환경에서 Multi-Agent 시스템의 안정성을 크게 높일 수 있습니다. 일시적 에러는 재시도로 해결하고, 지속적 에러는 폴백으로 우회하며, 치명적 에러는 서킷 브레이커로 격리합니다. ### 실전 팁 💡 재시도 전략은 멱등성(idempotency)이 보장될 때만 안전합니다. 같은 작업을 여러 번 실행해도 결과가 같은지 확인하세요. 결제 같은 작업은 재시도 시 중복 처리를 방지해야 합니다. 💡 에러 로그에 충분한 컨텍스트(어느 에이전트, 어느 Task, 입력 데이터)를 포함하여 디버깅을 쉽게 하세요. Sentry나 CloudWatch 같은 모니터링 도구와 통합하면 더 좋습니다. 💡 서킷 브레이커는 일정 시간 후 자동으로 half-open 상태로 전환하여 회복 여부를 테스트해야 합니다. 영구적으로 닫혀있으면 시스템이 복구되어도 사용할 수 없습니다. 💡 각 에러 유형별로 다른 처리 전략을 적용하세요. rate limit은 재시도, invalid input은 스킵, critical failure는 알림 발송 등 세분화된 처리가 필요합니다. 💡 테스트 환경에서 의도적으로 에러를 주입(chaos engineering)하여 에러 핸들링 로직이 제대로 작동하는지 검증하세요. 실제 장애 상황에서 처음 테스트하면 늦습니다. --- ## 12. 실전 Multi-Agent 아키텍처 ### 시작하며 여러분이 지금까지 배운 모든 개념들을 실제 프로덕션 시스템에 어떻게 통합할지 고민하고 계시죠? 각 패턴은 이해했지만, 이들을 조합하여 확장 가능하고 유지보수 가능한 실전 아키텍처를 설계하는 것은 또 다른 차원의 문제입니다. 실무에서는 단순한 데모가 아니라 수천 명의 사용자를 처리하고, 비용을 최적화하며, 장애에 대응하고, 지속적으로 개선할 수 있는 시스템이 필요합니다. 지금까지 배운 개념들을 어떻게 조합하고, 어떤 아키텍처 결정을 내려야 할까요? 바로 이럴 때 필요한 것이 검증된 Multi-Agent 아키텍처 패턴입니다. 모범 사례를 따르고, 확장성과 관찰 가능성을 고려하여 프로덕션 레벨 시스템을 설계합니다. ### 개요 간단히 말해서, 실전 Multi-Agent 아키텍처는 여러 에이전트, 오케스트레이션, 메모리, 도구, 모니터링을 통합하여 확장 가능하고 안정적인 프로덕션 시스템을 구축하는 설계 패턴입니다. 실무에서는 다음과 같은 요구사항을 모두 충족해야 합니다: 수평 확장 가능, 비용 효율적, 관찰 가능(로그, 메트릭, 추적), 테스트 가능, 버전 관리 가능, 보안(API 키, 데이터 격리). 예를 들어, 대규모 콘텐츠 생성 플랫폼을 구축한다면 수백 개의 동시 요청을 처리하고, 각 요청의 진행 상황을 추적하며, 실패한 작업을 재시도하고, 비용을 모니터링해야 합니다. 기존의 단순한 스크립트나 프로토타입과 달리, 프로덕션 아키텍처는 계층화(presentation, orchestration, agent, tool), 관심사 분리(각 레이어의 책임), 인터페이스 추상화(에이전트 교체 가능)가 필요합니다. 실전 아키텍처의 핵심 원칙은 모듈화(각 에이전트/도구는 독립적 모듈), 관찰 가능성(모든 단계를 로깅/추적), 확장성(수평 확장 가능한 설계), 비용 최적화(캐싱, 배칭, 모델 선택)입니다. 이런 원칙들을 구체적인 아키텍처 패턴으로 구현해야 합니다. ### 코드 예제 ```python # 프로덕션 Multi-Agent 아키텍처 예시 from crewai import Agent, Task, Crew from langchain.cache import RedisCache from langchain.callbacks import get_openai_callback import structlog from prometheus_client import Counter, Histogram # 구조화된 로깅 설정 logger = structlog.get_logger() # 메트릭 정의 agent_calls = Counter('agent_calls_total', 'Total agent calls', ['agent_name', 'status']) agent_duration = Histogram('agent_duration_seconds', 'Agent execution time', ['agent_name']) # 캐싱 레이어 - 중복 LLM 호출 방지 from langchain.globals import set_llm_cache set_llm_cache(RedisCache(redis_url="redis://localhost:6379")) class ProductionAgent: """프로덕션용 에이전트 래퍼""" def __init__(self, agent: Agent): self.agent = agent self.name = agent.role def execute(self, task_description: str) -> dict: logger.info("agent_execution_started", agent=self.name, task=task_description) with agent_duration.labels(agent_name=self.name).time(): try: with get_openai_callback() as cb: result = self.agent.execute_task(task_description) # 비용 및 토큰 로깅 logger.info("agent_execution_completed", agent=self.name, tokens=cb.total_tokens, cost=cb.total_cost) agent_calls.labels(agent_name=self.name, status='success').inc() return {"status": "success", "result": result, "cost": cb.total_cost} except Exception as e: logger.error("agent_execution_failed", agent=self.name, error=str(e)) agent_calls.labels(agent_name=self.name, status='error').inc() return {"status": "error", "error": str(e)} class MultiAgentPipeline: """프로덕션 Multi-Agent 파이프라인""" def __init__(self): self.agents = self._initialize_agents() self.cache = RedisCache(redis_url="redis://localhost:6379") def _initialize_agents(self): """에이전트 초기화""" researcher = ProductionAgent(Agent( role='Researcher', goal='Research information', backstory='Expert researcher', verbose=False # 프로덕션에서는 verbose 비활성화 )) writer = ProductionAgent(Agent( role='Writer', goal='Write content', backstory='Professional writer', verbose=False )) return {"researcher": researcher, "writer": writer} async def process(self, topic: str, user_id: str): """비동기 처리 파이프라인""" logger.info("pipeline_started", topic=topic, user_id=user_id) # 캐시 확인 cache_key = f"content:{topic}" cached = self.cache.lookup(topic, "") if cached: logger.info("cache_hit", topic=topic) return cached # 단계별 실행 research_result = self.agents["researcher"].execute(f"Research: {topic}") if research_result["status"] == "error": return research_result ``` ---

카테고리: AI

언어: Python

태그: AI, MultiAgent, CrewAI, AutoGen, AgentCollaboration

작성자: AI Generated

프리미엄 콘텐츠 - 3개월 무료 체험 가능

AI 에이전트 Multi-Agent 협업 완벽 가이드

목차

1. Multi-Agent 시스템 기본 개념

시작하며

개요

코드 예제

설명

실전 팁

2. Agent Role과 Goal 설계

시작하며

개요

코드 예제

설명

실전 팁

3. CrewAI 기본 구조

시작하며

개요

코드 예제

설명

실전 팁

4. Sequential Task 실행

시작하며

개요

코드 예제

설명

실전 팁

5. Parallel Task 실행

시작하며

개요

코드 예제

설명

실전 팁

6. Agent 간 통신과 메모리 공유

시작하며

개요

코드 예제

설명

실전 팁

7. Tool Sharing과 Delegation

시작하며

개요

코드 예제

설명

실전 팁

8. AutoGen Framework

시작하며

개요

코드 예제

설명

실전 팁

9. Consensus와 Voting

시작하며

개요

코드 예제

설명

실전 팁

10. Multi-Agent 오케스트레이션

시작하며

개요

코드 예제

설명

실전 팁

11. 에러 핸들링과 복구 전략

시작하며

개요

코드 예제

설명

실전 팁

12. 실전 Multi-Agent 아키텍처

시작하며

개요

코드 예제

댓글 (0)

함께 보면 좋은 카드 뉴스

Whisper 모델 완벽 가이드

AI 음성 인식 기술 완벽 가이드 - 음성 인식의 발전과 실전 활용

AI 파인튜닝 실전 사례 연구 완벽 가이드

AI 파인튜닝 모델 병합과 배포 완벽 가이드

AI 파인튜닝 평가와 벤치마킹 완벽 가이드

AI 파인튜닝 하이퍼파라미터 튜닝 완벽 가이드