CrewAI Integration

Full observability for CrewAI multi-agent systems. Track crew executions, agent tasks, and inter-agent communication.

crewai >= 0.30.0CrewsAgentsTasks

Installation

Terminal
pip install turingpulse_sdk turingpulse_sdk_crewai crewai

Quick Start

1. Initialize & Instrument

setup.py
from turingpulse_sdk import init, TuringPulseConfig
from turingpulse_sdk_crewai import instrument_crewaiai

# Initialize TuringPulse
init(TuringPulseConfig(
    api_key="sk_live_your_api_key",
    workflow_name="my-project",
))

# Enable auto-instrumentation for CrewAI
instrument_crewaiai()

2. Define Your Crew

crew.py
from crewai import Agent, Task, Crew

# Define agents
researcher = Agent(
    role="Researcher",
    goal="Find relevant information",
    backstory="You are an expert researcher...",
    llm=llm,
)

writer = Agent(
    role="Writer",
    goal="Write compelling content",
    backstory="You are a skilled content writer...",
    llm=llm,
)

# Define tasks
research_task = Task(
    description="Research the topic: {topic}",
    agent=researcher,
    expected_output="Research findings",
)

writing_task = Task(
    description="Write an article based on research",
    agent=writer,
    expected_output="Final article",
)

# Create crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
)

3. Run Your Crew

run.py
# Run the crew - fully traced automatically
result = crew.kickoff(inputs={"topic": "AI in healthcare"})
print(result)

# Each agent execution, task, and tool call is captured

What Gets Traced

  • Crew Execution — Full trace for each kickoff
  • Agent Tasks — Each agent's task execution as a span
  • LLM Calls — Model, tokens, latency for each call
  • Tool Usage — Tool invocations with inputs/outputs
  • Agent Delegation — When agents delegate to others
  • Context Sharing — Information passed between agents
  • Errors — Exceptions with full stack traces

Manual Crew Instrumentation

manual.py
from turingpulse_sdk_crewai import instrument_crewai

# Wrap a specific crew
instrumented_crew = instrument_crewai(
    crew,
    name="research-crew",
    labels={"department": "content", "priority": "high"},
)

result = instrumented_crew.kickoff(inputs={"topic": "AI"})

With Governance

governance.py
from turingpulse_sdk import GovernanceDirective
from turingpulse_sdk_crewai import instrument_crewai

instrumented_crew = instrument_crewai(
    crew,
    name="publishing-crew",
    governance=GovernanceDirective(
        hitl=True,
        hatl=True,
        reviewers=["editor@company.com"],
        escalation_channels=["slack://content-review"],
        severity="high",
    ),
)

With KPIs

kpis.py
from turingpulse_sdk import KPIConfig
from turingpulse_sdk_crewai import instrument_crewai

instrumented_crew = instrument_crewai(
    crew,
    name="research-crew",
    kpis=[
        KPIConfig(
            kpi_id="total_latency_ms",
            use_duration=True,
            alert_threshold=60000,  # Alert if > 60 seconds
            comparator="gt",
        ),
        KPIConfig(
            kpi_id="total_cost",
            value=lambda ctx: ctx.metadata.get("total_cost", 0),
            alert_threshold=1.00,  # Alert if > $1
            comparator="gt",
        ),
        KPIConfig(
            kpi_id="agent_count",
            value=lambda ctx: len(ctx.metadata.get("agents_used", [])),
            alert_threshold=5,
            comparator="gt",
        ),
    ],
)

Agent-Level Configuration

governance.py
from turingpulse_sdk import GovernanceDirective
from turingpulse_sdk_crewai import instrument_crewai

run = instrument_crewai(
    crew,
    name="content-crew",
    governance=GovernanceDirective(
        hatl=True,
        reviewers=["qa@company.com"],
    ),
    metadata={"team": "content"},
)

Hierarchical Process

hierarchical.py
from crewai import Crew, Process

crew = Crew(
    agents=[manager, researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.hierarchical,  # Manager delegates tasks
    manager_llm=manager_llm,
)

# Hierarchical delegation is fully traced
# See which agent delegated to which
💡
Multi-Agent Visibility
The TuringPulse dashboard shows a visual timeline of all agent interactions, making it easy to debug complex multi-agent workflows.

Next Steps