This is a Jupyter notebook

Cookbook: crewAI Integration

This is a cookbook with examples of the Langfuse Integration for crewAI (opens in a new tab) (GitHub (opens in a new tab)).
It utilizes the Langfuse OpenAI Integration (opens in a new tab) and the @observe() decorator (opens in a new tab). The OpenAI Integration ensures that each step executed by your crew is logged as a new generation in the system. The @observe() decorator automatically and asynchronously groups these logs into a single trace.

Note: crewAI is compatible with Python >=3.10, <=3.13.

Setup

!pip install crewai langfuse

Initialize the Langfuse client with your API keys from the project settings in the Langfuse UI and add them to your environment.

import os
 
os.environ["LANGFUSE_PUBLIC_KEY"] = ""
os.environ["LANGFUSE_SECRET_KEY"] = ""
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com" # For US data region, set this to "https://us.cloud.langfuse.com"
 
os.environ["OPENAI_API_KEY"] = ""

Import the Langfuse OpenAI integration and check the server connection.

from langfuse.openai import auth_check
 
auth_check()

Create Custom Tool

For more information on how to create custom tools for the crewAI framework, please visit the crewAI docs (opens in a new tab).

from crewai_tools import tool
 
@tool
def multiplication_tool(first_number: int, second_number: int) -> str:
    """Useful for when you need to multiply two numbers together."""
    return first_number * second_number

Assemble Crew

For detailed instructions on setting up your crew, please refer to the crewAI documentation (opens in a new tab).

from crewai import Agent, Task, Crew
 
writer = Agent(
        role="Writer",
        goal="You make math engaging and understandable for young children through poetry",
        backstory="You're an expert in writing haikus but you know nothing of math.",
        tools=[multiplication_tool],  
    )
 
task = Task(description=("What is {multiplication}?"),
            expected_output=("Compose a haiku that includes the answer."),
            agent=writer)
 
crew = Crew(
  agents=[writer],
  tasks=[task],
  share_crew=False
)

Examples

Simple Example using OpenAI Integration and @observe() decorator

You can use this integration in combination with the @observe() decorator from the Langfuse Python SDK. Thereby, you can trace non-Langchain code, combine multiple Langchain invocations in a single trace, and use the full functionality of the Langfuse Python SDK.

Learn more about Langfuse Tracing here (opens in a new tab).

from langfuse.decorators import observe
 
@observe()
def create_simple_haiku(input):
    result = crew.kickoff(inputs={"multiplication": input})
    return result
 
create_simple_haiku("3 * 3")

Example with additional trace attributes

from langfuse.decorators import langfuse_context
 
@observe()
def create_custom_haiku(input):
    result = crew.kickoff(inputs={"multiplication": input})
    langfuse_context.update_current_observation(
        name="Crew AI Trace",
        session_id="session_id",
        user_id="user_id",
        tags=["haiku"],
        metadata={"env": "prod"}
    )
    return result
 
create_custom_haiku("5 * 3")