CrewAI Integration

pip install turingpulse crewai

from turingpulse import init
from turingpulse.integrations.crewai import instrument_crewai

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

# Enable auto-instrumentation for CrewAI
instrument_crewai()

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],
)

# 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

from turingpulse.integrations.crewai import instrument_crew

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

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

from turingpulse import GovernanceDirective
from turingpulse.integrations.crewai import instrument_crew

instrumented_crew = instrument_crew(
    crew,
    agent_id="publishing-crew",
    governance=GovernanceDirective(
        # Review before publishing
        hitl=True,
        hitl_condition=lambda ctx: ctx.task_name == "publish",
        reviewers=["editor@company.com"],
        
        # Review all crew outputs
        hatl=True,
        hatl_sample_rate=1.0,
    ),
)

from turingpulse import KPIConfig
from turingpulse.integrations.crewai import instrument_crew

instrumented_crew = instrument_crew(
    crew,
    agent_id="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",
        ),
    ],
)

from turingpulse.integrations.crewai import instrument_agent

# Configure tracing for specific agents
instrumented_researcher = instrument_agent(
    researcher,
    labels={"role": "research"},
    capture_reasoning=True,  # Capture agent's internal reasoning
)

instrumented_writer = instrument_agent(
    writer,
    labels={"role": "writing"},
    governance=GovernanceDirective(
        hatl=True,  # Review all writing outputs
    ),
)

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