CrewAI は、LangChain や他のエージェントフレームワークに完全に依存せず、一から構築された軽量かつ非常に高速な Python フレームワークです。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 = Crew(
agents=[researcher, writer],
tasks=[research_task, writing_task],
verbose=True,
process=Process.sequential,
)
# Crewを実行する
result = crew.kickoff(inputs={"topic": "AI in material science"})
print(result)
CrewAI には、kickoff プロセスをより細かく制御するための複数の method (kickoff(), kickoff_for_each(), kickoff_async(), kickoff_for_each_async()) があります。このインテグレーションは、これらすべての method からのトレースのログをサポートしています。
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 callに置き換えることができます。
# この例では、顧客フィードバックをシミュレートします。
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 = CustomerFeedbackFlow()
result = flow.kickoff()
print(result)
このインテグレーションでは、Flow.kickoff のエントリポイントと、使用可能なすべてのデコレータ (@start、@listen、@router、@or_、@and_) に自動的にパッチが適用されます。
Crew Guardrail - 独自の ops をトラッキングする
@weave.op でラップすると、inputs、outputs、アプリケーションロジックのキャプチャが始まり、エージェント を通じてデータがどのように検証されるかをデバッグできるようになります。さらに、実験を進める中で git にコミットされていないアドホックな詳細も記録できるよう、コードのバージョン管理も自動的に開始されます。
リサーチアナリストとライターの例を見てみましょう。ここでは、生成されたレポートの長さを検証するためのガードレールを追加します。
# .... 既存のインポートと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,
)
# .... クルーを実行する既存のコード ....
@weave.op でデコレートするだけで、この関数の入力と出力に加え、実行時間、内部で LLM を使用している場合のトークン情報、コード version などをトラッキングできます。
このインテグレーションの改善点について、ぜひご意見をお寄せください。問題が発生した場合は、こちら から issue を作成してください。
CrewAI を使って強力なマルチエージェントシステムを構築する方法については、同社の豊富なサンプルやドキュメントをご覧ください。
