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_for_each(), kickoff_async(), kickoff_for_each_async()와 같이 kickoff 프로세스를 보다 세밀하게 제어하기 위한 여러 메서드를 제공합니다. 이 통합에서는 이들 모든 메서드에서 생성되는 트레이스 로깅을 지원합니다.
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 - 나만의 op 추적하기
@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으로 가드레일 함수를 데코레이터로 감싸기만 하면, 이 함수의 입력과 출력은 물론 실행 시간, 내부적으로 LLM이 사용된 경우 토큰 정보, 코드 버전 등까지 모두 추적할 수 있습니다.
이 통합에서 개선이 필요하다고 생각되는 점이 있다면 꼭 알려 주세요. 문제가 발생하면 여기에 이슈를 생성해 주세요.
CrewAI를 사용해 강력한 멀티에이전트 시스템을 구축하는 방법은 해당 프로젝트의 다양한 예제와 문서를 통해 더 자세히 확인할 수 있습니다.
플랫폼
W&B Models
W&B Weave
W&B Inference
W&B 교육
릴리스 노트