CrewAI는 처음부터 새로 구현한, 가볍고 매우 빠른 Python 프레임워크로, LangChain이나 다른 에이전트 프레임워크에 전혀 의존하지 않습니다. CrewAI는 높은 수준의 단순성(Crews)과 정밀한 저수준 제어(Flows)를 모두 제공하여, 어떤 시나리오에도 맞는 자율 AI 에이전트를 만들기에 이상적입니다. 더 자세한 내용은 CrewAI 문서를 참고하세요.
AI 에이전트로 작업할 때는 에이전트 간 상호작용을 디버깅하고 모니터링하는 것이 매우 중요합니다. CrewAI 애플리케이션은 종종 여러 에이전트가 함께 동작하므로, 이들이 어떻게 협업하고 소통하는지 이해하는 것이 필수적입니다. Weave는 CrewAI 애플리케이션에 대한 트레이스를 자동으로 캡처하여, 에이전트의 성능과 상호작용을 모니터링하고 분석하는 과정을 단순화합니다.
이 인테그레이션은 Crews와 Flows 둘 다를 지원합니다.
이 예제를 실행하려면 CrewAI(설치 방법) 및 Weave를 설치해야 합니다.
이제 CrewAI Crew를 생성하고 Weave를 사용해 실행을 추적해 보겠습니다. 시작하려면 스크립트의 처음 부분에서 weave.init()을 호출하면 됩니다. weave.init()의 인수는 트레이스를 기록할 프로젝트 이름입니다.
import weave
from crewai import Agent, Task, Crew, LLM, Process
# 프로젝트 이름으로 Weave 초기화
weave.init(project_name="crewai_demo")
# 결정론적 출력을 보장하기 위해 temperature를 0으로 설정한 LLM 생성
llm = LLM(model="gpt-4o-mini", temperature=0)
# 에이전트 생성
researcher = Agent(
role='Research Analyst',
goal='Find and analyze the best investment opportunities',
backstory='Expert in financial analysis and market research',
llm=llm,
verbose=True,
allow_delegation=False,
)
writer = Agent(
role='Report Writer',
goal='Write clear and concise investment reports',
backstory='Experienced in creating detailed financial reports',
llm=llm,
verbose=True,
allow_delegation=False,
)
# 태스크 생성
research_task = Task(
description='Deep research on the {topic}',
expected_output='Comprehensive market data including key players, market size, and growth trends.',
agent=researcher
)
writing_task = Task(
description='Write a detailed report based on the research',
expected_output='The report should be easy to read and understand. Use bullet points where applicable.',
agent=writer
)
# 크루 생성
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, writing_task],
verbose=True,
process=Process.sequential,
)
# 크루 실행
result = crew.kickoff(inputs={"topic": "AI in material science"})
print(result)
CrewAI는 kickoff 프로세스를 더 세밀하게 제어할 수 있도록 kickoff(), kickoff_for_each(), kickoff_async(), kickoff_for_each_async()와 같은 여러 메서드를 제공합니다. 이 인테그레이션은 이러한 모든 메서드에서의 트레이스 로깅을 지원합니다.
pip install 'crewai[tools]'
SerperDevTool을 사용해 ‘Research Analyst’ 에이전트가 인터넷에서 관련 정보를 검색할 수 있도록 합니다. 이 도구와 API 요구 사항에 대해 자세히 알아보려면 여기를 참조하세요.
# .... 기존 임포트 ....
from crewai_tools import SerperDevTool
# 에이전트에 도구를 제공합니다.
researcher = Agent(
role='Research Analyst',
goal='Find and analyze the best investment opportunities',
backstory='Expert in financial analysis and market research',
llm=llm,
verbose=True,
allow_delegation=False,
tools=[SerperDevTool()],
)
# .... 기존 코드 ....
import weave
# 프로젝트 이름으로 Weave 초기화
weave.init("crewai_demo")
from crewai.flow.flow import Flow, listen, router, start
from litellm import completion
class CustomerFeedbackFlow(Flow):
model = "gpt-4o-mini"
@start()
def fetch_feedback(self):
print("Fetching customer feedback")
# 실제 환경에서는 API 호출로 대체할 수 있습니다.
# 이 예제에서는 고객 피드백을 시뮬레이션합니다.
feedback = (
"I had a terrible experience with the product. "
"It broke after one use and customer service was unhelpful."
)
self.state["feedback"] = feedback
return feedback
@router(fetch_feedback)
def analyze_feedback(self, feedback):
# 언어 모델을 사용하여 감성 분석
prompt = (
f"Analyze the sentiment of this customer feedback and "
"return only 'positive' or 'negative':\n\n"
f"Feedback: {feedback}"
)
response = completion(
model=self.model,
messages=[{"role": "user", "content": prompt}],
)
sentiment = response["choices"][0]["message"]["content"].strip().lower()
# 응답이 모호한 경우 기본값을 negative로 설정
if sentiment not in ["positive", "negative"]:
sentiment = "negative"
return sentiment
@listen("positive")
def handle_positive_feedback(self):
# 긍정적인 피드백에 대한 감사 메시지 생성
prompt = "Generate a thank you message for a customer who provided positive feedback."
response = completion(
model=self.model,
messages=[{"role": "user", "content": prompt}],
)
thank_you_message = response["choices"][0]["message"]["content"].strip()
self.state["response"] = thank_you_message
return thank_you_message
@listen("negative")
def handle_negative_feedback(self):
# 부정적인 피드백에 대한 사과 메시지 및 서비스 개선 약속 생성
prompt = (
"Generate an apology message to a customer who provided negative feedback and offer assistance or a solution."
)
response = completion(
model=self.model,
messages=[{"role": "user", "content": prompt}],
)
apology_message = response["choices"][0]["message"]["content"].strip()
self.state["response"] = apology_message
return apology_message
# flow 인스턴스 생성 및 실행
flow = CustomerFeedbackFlow()
result = flow.kickoff()
print(result)
이 인테그레이션은 Flow.kickoff 엔트리 포인트와 @start, @listen, @router, @or_, @and_ 등 사용 가능한 모든 데코레이터를 자동으로 패치합니다.
Crew Guardrail - 나만의 ops 추적하기
@weave.op으로 감싸면 입력, 출력, 애플리케이션 로직을 캡처하므로 데이터가 에이전트를 통해 어떻게 검증되는지 디버깅할 수 있습니다. 또한 실험하는 동안 git에 커밋되지 않은 임시 정보까지 포착할 수 있도록 코드를 자동으로 버전 관리해 줍니다.
리서치 애널리스트와 작가의 예시를 살펴보겠습니다. 생성된 리포트의 길이를 검증하는 guardrail을 추가해 보겠습니다.
# .... 기존 임포트 및 weave 초기화 ....
# 가드레일 함수를 `@weave.op()`으로 데코레이트하세요
@weave.op(name="guardrail-validate_blog_content")
def validate_blog_content(result: TaskOutput) -> Tuple[bool, Any]:
# 원시 문자열 결과 가져오기
result = result.raw
"""블로그 콘텐츠가 요구사항을 충족하는지 검증합니다."""
try:
# 단어 수 확인
word_count = len(result.split())
if word_count > 200:
return (False, {
"error": "블로그 콘텐츠가 200단어를 초과합니다",
"code": "WORD_COUNT_ERROR",
"context": {"word_count": word_count}
})
# 추가 검증 로직을 여기에 작성하세요
return (True, result.strip())
except Exception as e:
return (False, {
"error": "검증 중 예기치 않은 오류 발생",
"code": "SYSTEM_ERROR"
})
# .... 기존 에이전트 및 리서치 애널리스트 태스크 ....
writing_task = Task(
description='리서치를 바탕으로 200단어 이내의 상세한 리포트를 작성하세요',
expected_output='리포트는 읽고 이해하기 쉬워야 합니다. 해당하는 경우 글머리 기호를 사용하세요.',
agent=writer,
guardrail=validate_blog_content,
)
# .... crew를 실행하는 기존 코드 ....
@weave.op 데코레이터를 guardrail 함수에 간단히 적용하기만 하면, 이 함수의 입력과 출력은 물론 실행 시간, 내부적으로 LLM이 사용될 경우의 토큰 정보, 코드 버전 등까지 추적할 수 있습니다.
이 인테그레이션을 어떻게 개선하면 좋을지 알려 주세요. 문제가 발생하면 여기에 이슈를 열어 주세요.
CrewAI의 다양한 예제와 문서를 통해 강력한 멀티에이전트 시스템을 구축하는 방법을 더 자세히 알아보세요.
플랫폼
W&B Models
W&B Weave
W&B Inference
W&B 트레이닝
릴리스 노트