Crewai

CrewAIは、LangChainや他のエージェントフレームワークから完全に独立した、一から構築された軽量で超高速なPythonフレームワークです。CrewAIは開発者に高レベルのシンプルさ（Crews）と精密な低レベル制御（Flows）の両方を提供し、あらゆるシナリオに合わせた自律型AIエージェントの作成に最適です。CrewAIについてはこちら。

AIエージェントを扱う際、そのインタラクションのデバッグとモニタリングは非常に重要です。CrewAIアプリケーションは多くの場合、複数のエージェントが協力して動作するため、それらがどのように連携し、コミュニケーションを取るかを理解することが不可欠です。Weaveはこのプロセスを簡素化し、CrewAIアプリケーションのトレースを自動的に取得することで、エージェントのパフォーマンスとインタラクションを監視・分析できるようにします。

この統合はCrewsとFlowsの両方をサポートしています。

Crewを使い始める

この例を実行するには、CrewAI（詳細はこちら）とweaveをインストールする必要があります：

pip install crewai weave

ここでCrewAI Crewを作成し、Weaveを使用して実行をトレースします。始めるには、スクリプトの先頭で単にweave.init()を呼び出すだけです。weave.init()の引数は、トレースがログに記録されるプロジェクト名です。

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

# Initialize Weave with your project name
# highlight-next-line
weave.init(project_name="crewai_demo")

# Create an LLM with a temperature of 0 to ensure deterministic outputs
llm = LLM(model="gpt-4o-mini", temperature=0)

# Create agents
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,
)

# Create tasks
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
)

# Create a crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    verbose=True,
    process=Process.sequential,
)

# Run the crew
result = crew.kickoff(inputs={"topic": "AI in material science"})
print(result)

Weaveは、エージェントのインタラクション、タスクの実行、LLM呼び出しなど、CrewAIライブラリを通じて行われるすべての呼び出しを追跡し、ログに記録します。Weaveウェブインターフェースでトレースを確認できます。

CrewAIは、キックオフプロセスをより適切に制御するためのいくつかのメソッドを提供しています：kickoff()、kickoff_for_each()、kickoff_async()、およびkickoff_for_each_async()。この統合は、これらすべてのメソッドからのトレースのログ記録をサポートしています。

ツールの追跡

CrewAIツールは、ウェブ検索やデータ分析から、同僚間のコラボレーションやタスク委任まで、さまざまな機能をエージェントに提供します。この統合はそれらもトレースすることができます。

上記の例で生成されるレポートの品質を向上させるために、インターネットを検索して最も関連性の高い結果を返すツールへのアクセスを提供します。

まず、追加の依存関係をインストールしましょう。

pip install 'crewai[tools]'

この例では、SerperDevToolを使用して、「リサーチアナリスト」エージェントがインターネット上の関連情報を検索できるようにしています。このツールとAPI要件の詳細についてはこちらをご覧ください。

# .... existing imports ....
from crewai_tools import SerperDevTool

# We provide the agent with the tool.
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,
    # highlight-next-line
    tools=[SerperDevTool()],
)

# .... existing code ....

インターネットにアクセスできるエージェントを持つこのCrewを実行すると、より良く関連性の高い結果が得られます。下の画像に示すように、ツールの使用を自動的にトレースします。

この統合は、crewAI-toolsリポジトリで利用可能なすべてのツールを自動的にパッチします。

Flowを使い始める

import weave
# Initialize Weave with your project name
# highlight-next-line
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")
        # In a real-world scenario, this could be replaced by an API call.
        # For this example, we simulate customer feedback.
        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):
        # Use the language model to analyze sentiment
        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()
        # If the response is ambiguous, default to negative
        if sentiment not in ["positive", "negative"]:
            sentiment = "negative"
        return sentiment

    @listen("positive")
    def handle_positive_feedback(self):
        # Generate a thank you message for positive feedback
        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):
        # Generate an apology message with a promise to improve service for negative feedback
        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

# Instantiate and kickoff the flow
flow = CustomerFeedbackFlow()
result = flow.kickoff()
print(result)

この統合はFlow.kickoffエントリーポイントと利用可能なすべてのデコレーター —@start、@listen、@router、@or_および@and_を自動的にパッチします。

Crewガードレール - 独自のオペレーションを追跡する

タスクガードレールは、タスク出力を次のタスクに渡す前に検証および変換する方法を提供します。シンプルなPython関数を使用して、エージェントの実行をリアルタイムで検証できます。

この関数を@weave.opでラップすると、入力、出力、アプリケーションロジックの取得が開始され、データがエージェントを通じてどのように検証されるかをデバッグできます。これにより、実験時にコードの自動バージョン管理も開始され、gitにコミットされていないアドホックな詳細が取得されます。

リサーチアナリストとライターの例を見てみましょう。生成されたレポートの長さを検証するガードレールを追加します。

# .... existing imports and weave initialization ....

# Decorate your guardrail function with `@weave.op()`
# highlight-next-line
@weave.op(name="guardrail-validate_blog_content")
def validate_blog_content(result: TaskOutput) -> Tuple[bool, Any]:
    # Get raw string result
    result = result.raw

    """Validate blog content meets requirements."""
    try:
        # Check word count
        word_count = len(result.split())

        if word_count > 200:
            return (False, {
                "error": "Blog content exceeds 200 words",
                "code": "WORD_COUNT_ERROR",
                "context": {"word_count": word_count}
            })

        # Additional validation logic here
        return (True, result.strip())
    except Exception as e:
        return (False, {
            "error": "Unexpected error during validation",
            "code": "SYSTEM_ERROR"
        })


# .... existing agents and research analyst task ....

writing_task = Task(
    description='Write a detailed report based on the research under 200 words',
    expected_output='The report should be easy to read and understand. Use bullet points where applicable.',
    agent=writer,
    # highlight-next-line
    guardrail=validate_blog_content,
)

# .... existing code to run crew ....

ガードレール関数を単に@weave.opでデコレートするだけで、この関数への入力と出力を、実行時間、LLMが内部で使用されている場合のトークン情報、コードバージョンなどと共に追跡できます。

結論

この統合について改善すべき点があれば、ぜひお知らせください。問題が発生した場合は、こちらで問題を開いてください。

CrewAIを使用して強力なマルチエージェントシステムを構築する方法の詳細については、多くの例やドキュメントをご覧ください。