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 요구 사항에 대한 자세한 내용은 여기에서 확인하세요.
# .... existing imports ....
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()],
)
# .... existing code ....
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("고객 피드백 가져오는 중")
# 실제 시나리오에서는 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 데코레이터만 추가하면 이 함수의 입력과 출력은 물론, 실행 시간, 내부적으로 LLM을 사용하는 경우의 토큰 정보, 코드 버전 등도 추적할 수 있습니다.
이 인테그레이션에서 개선이 필요한 점이 있다면 알려주세요. 문제가 발생했다면 여기에서 이슈를 등록해 주세요.
CrewAI의 다양한 예제와 문서에서 강력한 멀티에이전트 시스템을 구축하는 방법을 더 자세히 알아보세요.
