CrewAI 애플리케이션 추적하기

목차

CrewAI와 주요 기능 이해

CrewAI로 모니터링과 분석을 수행하는 방법

그림: W&B Weave의 트레이스 인터페이스가 CrewAI 실행을 캡처한 모습. 스크린샷에는 crew.kickoff() 실행에 대한 계층적 트레이스가 표시되며, 두 개의 주요 태스크(Research Analyst와 Report Writer)가 포함되어 있습니다. 각 태스크 아래에는 LLM 호출(예: llm.complete)과 그 파라미터 같은 하위 작업이 나타납니다. 오른쪽 패널에는 선택한 작업의 입력과 출력이 표시되며, 이 경우 “AI in Material Science” 주제로 생성된 투자 보고서의 발췌본이 보여집니다. 이러한 상세 트레이스는 개발자가 에이전트 간 상호작용을 모니터링하고 각 단계에서의 성능을 분석하는 데 도움이 됩니다.

CrewAI 애플리케이션 트레이싱 시작하기

Weave 시작 단계

CrewAI와 Weave 설치하기: Python 3.10–3.13이 설치되어 있는지 확인한 다음, pip을 통해 필요한 패키지를 설치하세요. 예시는 다음과 같습니다: pip install crewai weaveCrewAI의 추가 도구(예: 웹 브라우징 또는 기타 통합)를 사용할 계획이라면 다음으로 설치할 수 있습니다 pip install 'crewai[tools]' 이렇게 하면 환경에 CrewAI 프레임워크와 W&B Weave 라이브러리 둘 다 설정됩니다.
아직 가입하지 않았다면 무료 Weights & Biases 계정에 가입하세요이는 Weave가 추적 로그를 W&B 클라우드에 기록하고, 그곳에서 확인할 수 있도록 하기 위해 필요합니다. 계정을 만들고 로그인한 뒤(예를 들어 다음 명령을 실행하여) wandb login 터미널에서 또는 API 키를 사용해 로그인하면, 추적을 기록하고 시각화할 준비가 됩니다.
애플리케이션에서 Weave를 초기화하세요: Python 스크립트나 Jupyter 노트북에서 다음을 임포트하세요 weave 그리고 호출하세요 weave.init(...) 코드의 시작 부분에서 추적을 초기화하세요. 프로젝트 이름을 지정해야 합니다 weave.init (예: weave.init(project_name="crewai_demo")) — 이 이름은 추적 로그를 정리하는 데 사용됩니다. 초기화가 완료되면, 콘솔에 보통 이 프로젝트의 Weave 대시보드 URL이 출력되며, 브라우저에서 해당 링크를 열 수 있습니다.
에이전트, 태스크, 그리고 크루를 정의하세요: CrewAI API를 사용해 평소처럼 에이전트와 태스크를 설정하세요. 이는 다음을 포함합니다 LLM 인스턴스(예: GPT-4 기반 모델), 그리고 다음으로 생성하기 Agent 특정 역할과 목표를 가진 객체, 그리고 Task 각 에이전트가 수행해야 할 작업을 설명하는 객체입니다. 예를 들어, 정보를 수집하는 것을 목표로 하는 Researcher 에이전트와 보고서를 작성하는 것을 목표로 하는 Writer 에이전트를 둘 수 있습니다. 각 Task는 에이전트에 할당됩니다(예: Researcher 에이전트를 사용하는 연구 태스크, Writer 에이전트를 사용하는 작성 태스크). 마지막으로, 다음을 생성합니다 Crew 해당 에이전트와 태스크를 포함하고, 태스크를 어떤 방식으로 실행할지(순차 또는 병렬) 프로세스를 결정하세요. 예를 들어, 연구 태스크가 완료된 뒤에 작성 태스크가 시작되도록 순차 프로세스를 사용할 수 있습니다(이 예시에서는 일반적으로 기대되는 흐름입니다).
크루를 실행하세요: 모든 구성이 끝났다면, 다음을 호출하여 멀티 에이전트 워크플로를 시작하세요 crew.kickoff(inputs={...}) 필요한 입력 매개변수와 함께. 진행 중인 예시에서는 다음과 같은 입력을 전달할 수 있습니다 {"topic": "AI in material science"} Researcher에게 무엇을 조사해야 하는지 알려주기 위해서입니다. 호출을 완료하면 kickoff()CrewAI가 에이전트를 시작하면, 에이전트들은 자신의 태스크를 실행하기 시작합니다. 이 실행 동안 Weave는 최상위 시작 지점부터 에이전트가 수행하는 각 LLM 호출까지 모든 작업을 자동으로 캡처합니다. 코드를 실행하면서 최종 결과나 출력을 출력할 수도 있지만, 진짜 핵심은 전체 프로세스의 트레이스가 백그라운드에서 기록되고 있다는 점입니다.
Weave에서 트레이스를 확인하기: 실행이 완료된 후(또는 실행 중에도, Weave가 데이터를 스트리밍할 수 있으므로), 프로젝트를 초기화할 때 제공된 Weave 대시보드 URL을 열세요. 웹 인터페이스에서 프로젝트와 트레이스 실행 목록을 확인할 수 있습니다(각 실행은 일반적으로 호출한 함수명으로 식별됩니다. 예: crewai.Crew.kickoff). 최신 실행 항목을 클릭해 트레이스 시각화를 열세요. 여기에서 다음 내용을 포함한 트레이스를 탐색할 수 있습니다: 모든 LLM 호출과 해당 메타데이터, 태스크 실행과 에이전트 동작의 순서, 실행 시간과 토큰 수와 같은 성능 지표, 그리고 발생한 오류나 예외. 사이드바나 타임라인을 사용해 특정 작업을 자세히 들여다볼 수 있습니다. 예를 들어, 에이전트가 LLM에 어떤 프롬프트를 보냈고 어떤 응답을 받았는지 확인할 수 있습니다. 이 트레이스 뷰는 CrewAI 애플리케이션의 동작을 디버깅하고 이해하기 위한 기본 도구입니다.

튜토리얼: CrewAI와 W&B Weave로 고객 지원 분류 시스템 트레이싱

설치 및 설정

pip install crewai wandb weave datasets openai

import os
import wandb
import weave
from crewai import Agent, Task, Crew, LLM, Process

# Set your OpenAI API key (ensure this is configured; you can also use os.environ)
os.environ["OPENAI_API_KEY"] = "<YOUR_OPENAI_KEY>"

# Initialize Weave for tracing (creates a project for logs)
wandb.login()
weave.init(project_name="customer_support_triage")

지원 문의용 샘플 데이터셋 로드하기

from datasets import load_dataset

# Load the CLINC150 out-of-scope dataset (use the 'imbalanced' version for full label set)
data = load_dataset('DeepPavlov/clinc_oos', 'imbalanced', split='test')
# Choose an example customer query from the dataset
sample = data[42]
user_query = sample['text']
true_category = sample['label_text']
print(f"Sample user query: {user_query}")
print(f"True intent (for reference): {true_category}")



CrewAI 에이전트와 태스크 정의하기

분류 에이전트: 문의의 범주(의도)를 판별합니다. 단순화를 위해 LLM을 사용해 문의를 알려진 지원 의도 중 하나로 분류합니다(예: 직접 입금, 카드 활성화, 사기 신고, 등), 또는 어떤 범주에도 맞지 않으면 “범위 외”로 표시합니다.
응답 에이전트: 분류된 범주를 바탕으로 유용한 응답을 생성합니다. 문의가 알려진 범주에 속하면, 이 에이전트가 답변을 시도합니다. 범주가 범위 외 (또는 분류 신뢰도가 낮은 경우) 사람 상담원에게 이관하는 것을 모사하여, 에스컬레이션 안내 메시지로 응답합니다.

# Create a deterministic LLM instance (using OpenAI's GPT-3.5 Turbo model)
llm = LLM(model="gpt-4o", temperature=0)

# Define the Classifier agent
classifier = Agent(
    role="Support Classifier",
    goal="Identify the category of the customer's query (e.g., billing, account, card issue, or out-of-scope)",
    backstory="A customer support AI specialized in triaging requests by intent.",
    llm=llm,
    verbose=True
)

# Define the Responder agent
responder = Agent(
    role="Support Agent",
    goal="Provide a helpful response to the customer based on the query category, or escalate if out-of-scope.",
    backstory="An AI assistant that answers customer questions or escalates them to a human if it cannot handle them.",
    llm=llm,
    verbose=True
)

# Task 1: Classification Task

classification_task = Task(
    description=(
        "Classify the customer query into a support category. Query: \"{query}\"\n"
        "Return ONLY the category name from this list: direct_deposit, card_issue, billing, fraud, atm_support, transfer, oos"
    ),
    expected_output="A single category name only (e.g., 'transfer' or 'billing' or 'oos').",
    agent=classifier
)

# Task 2: Response Generation Task
response_task = Task(
    description=(
        "You have the user's query and the classification result in your context. "
        "Based on that, write either a helpful answer or an escalation notice."
    ),
    expected_output="A support reply or escalation message.",
    agent=responder,
    context=[classification_task],
)

폴백 로직을 위한 가드레일 추가

from typing import Tuple, Any

@weave.op(name="guardrail_validate_classification")
def validate_classification(result: Any) -> Tuple[bool, Any]:
    """
    Guardrail check for classification task.
    Returns (True, cleaned_result) if category is valid, or (False, error_info) if invalid.
    """
    try:
        # Extract the raw output from the result
        if hasattr(result, 'raw'):
            category = str(result.raw).strip()
        else:
            # Fallback if structure is different
            category = str(result).strip()
            
        # List of known categories
        known_categories = ["direct_deposit", "card_issue", "billing", "fraud", "atm_support", "transfer", "oos"]
        if category not in known_categories:
            # If the classifier gave something unexpected, flag it
            return (False, {"error": f"Unknown category '{category}'", "code": "CLASSIFICATION_ERROR"})
        if category == "oos":
            # Out-of-scope category triggers fallback/escalation
            return (False, {"error": "Query is out-of-scope", "code": "ESCALATION_TRIGGER"})
        # If valid and in-scope, return cleaned category
        return (True, category)
    except Exception as e:
        return (False, {"error": "Exception in guardrail", "code": "SYSTEM_ERROR", "details": str(e)})

# Attach the guardrail to the classification task
classification_task.guardrail = validate_classification

Crew 초기화 및 워크플로 시작

# Create the crew with the two agents and their tasks in order
crew = Crew(
    agents=[classifier, responder],
    tasks=[classification_task, response_task],
    process=Process.sequential,  # tasks will run sequentially
    verbose=True
)

# Run the crew on the sample query
result = crew.kickoff(inputs={"query": user_query})
print("\nFinal output from Responder agent:\n", result)

Weave에서 트레이스 관찰하기



실패 상태를 활용해 프롬프트를 디버깅하고 문제를 해결했으며, 분류기가 정상적으로 실행되도록 할 수 있었습니다.

분류 단계: 분류기의 프롬프트(사용자의 쿼리가 포함되어 있어야 합니다)와 예측한 카테고리를 확인하세요. 예시에서는 다음과 같은 값을 기대합니다. "direct_deposit" 출력으로서
가드레일 검사: 가드레일 함수 validate_classification 단계로 표시됩니다. 카테고리가 유효하고 "oos"가 아니면 True와 함께 해당 카테고리를 반환합니다. False를 반환한 경우(예: 카테고리가 "oos")에서 오류 코드(ESCALATION_TRIGGER트레이스에 표시됩니다. 이렇게 하면 시스템이 에스컬레이션을 결정하는 시점을 쉽게 파악할 수 있습니다.
응답 생성: 응답자 에이전트의 단계에는 최종 답변이 표시됩니다. 쿼리를 적절히 처리했는지 확인하세요. 예를 들어, 카테고리가 직접 예치의 경우, 답변에는 자동이체 설정 단계가 나열될 수 있습니다. 카테고리가 범위 밖이었다면, 답변은 우리 에이전트의 목표와 로직에 따라 정중한 에스컬레이션 메시지가 되었을 것입니다.

CrewAI로 멀티 에이전트 워크플로를 구축하고 오케스트레이션하기

멀티 에이전트 워크플로의 활용 사례

자동화된 리서치와 리포팅 – 에이전트 팀이 여러 출처에서 정보를 협업으로 수집하고 데이터를 분석한 뒤, 최소한의 인간 개입으로 종합 보고서를 만들어낼 수 있습니다. 예를 들어 한 에이전트는 문헌과 뉴스를 검색하고, 다른 에이전트는 결과를 해석해 요약하며, 세 번째 에이전트가 최종 보고서를 편집·작성합니다. 이는 시장 조사, 학술 연구 리뷰, 경쟁 정보 수집 같은 분야에서 유용합니다.
콘텐츠 제작 파이프라인 – 모든 작업을 단일 AI가 처리하게 하기보다, 콘텐츠 제작의 각 단계에 특화된 에이전트를 배정할 수 있습니다. 예를 들어, a 연구자 에이전트가 사실을 수집하고, 개요 작성기 글의 구조를 잡고, 작가 콘텐츠를 초안 작성하고, 그리고 한 에디터 에이전트가 초안을 다듬고 검수합니다. 이러한 역할 분담은 단일 에이전트 방식보다 훨씬 효율적으로 더 높은 품질의 기사, 마케팅 카피, 문서를 만들어낼 수 있습니다.
고객 지원 자동화 – 멀티 에이전트 구성은 복잡한 고객 서비스 업무를 처리할 수 있습니다. 예를 들어, 당신은 a 분류 담당자 수신되는 지원 문의를 분류하는 에이전트가 먼저 작업한 뒤, 청구 문제나 기술 문제 해결에 특화된 에이전트 등에게 적절히 위임합니다. 필요하면 에이전트가 더 고급 에이전트나 사람에게 문의를 에스컬레이션할 수도 있습니다. 이러한 크루는 24시간 365일 운영될 수 있어 전자상거래나 IT 서비스 같은 업계에서 응답 시간과 지원의 일관성을 향상시킵니다.
비즈니스 인텔리전스와 데이터 분석 – 비즈니스 인텔리전스 시나리오에서는 서로 다른 에이전트가 각기 다른 데이터 소스나 분석 방법에 집중한 뒤, 그 인사이트를 결합할 수 있습니다. 예를 들어, 한 에이전트는 시장 동향 데이터를 분석하고, 다른 에이전트는 고객 피드백을 파싱하며, 또 다른 에이전트는 매출 지표를 검토합니다. 이후 매니저 에이전트가 이러한 결과를 취합해 의사결정자에게 제공할 일관된 분석으로 정리할 수 있습니다. 이러한 멀티 에이전트 접근은 분석 팀이 업무를 분담하는 방식을 반영하며, 더 빠르게 포괄적인 인텔리전스 보고서를 도출할 수 있습니다.
소프트웨어 개발 지원 – AI 에이전트가 협업하여 소프트웨어 개발 작업을 가속할 수 있습니다. 명세를 기반으로 코드를 생성하는 에이전트, 코드를 리뷰하거나 테스트하는 에이전트, 그리고 해당 코드의 문서를 작성하는 에이전트를 구성할 수 있습니다. 이 에이전트들을 오케스트레이션하면 요구사항부터 테스트된 코드까지 이어지는 자동화된 워크플로를 만들 수 있습니다. 이는 빠른 프로토타이핑이나, 반복적인 코딩·문서화 작업을 대신 처리해 개발자를 보조하는 데에 유용합니다.

결론

에이전트 상호작용 (프롬프트, 응답, 토큰 사용량)
가드레일 점검 (검증, 폴백, 에스컬레이션)
작업 오케스트레이션 (크루 전반의 작업 흐름)