Observability for CrewAI Multi-Agent Collaboration

Topic Input
    │
    ▼
┌─────────────────┐
│   Researcher    │ ──► Research findings
│   (GPT-4)       │
└─────────────────┘
    │
    ▼
┌─────────────────┐
│    Writer       │ ──► Draft article
│ (GPT-4/GPT-3.5) │
└─────────────────┘
    │
    ▼
┌─────────────────┐
│    Editor       │ ──► Polished article
│ (GPT-3.5/GPT-4) │
└─────────────────┘
    │
    ▼
┌─────────────────┐
│  SEO Specialist │ ──► Final optimized article
│   (GPT-3.5)     │
└─────────────────┘
    │
    ▼
Published Content

pip install netra-sdk crewai crewai-tools openai langchain-openai

export NETRA_API_KEY="your-netra-api-key"
export OPENAI_API_KEY="your-openai-api-key"

from crewai import Agent
from langchain_openai import ChatOpenAI

def create_agents(config: dict = None):
    """Create the content team agents with configurable models."""
    config = config or {
        "researcher": "gpt-4o",
        "writer": "gpt-4o",
        "editor": "gpt-3.5-turbo",
        "seo": "gpt-3.5-turbo",
    }

    researcher = Agent(
        role="Research Specialist",
        goal="Gather accurate facts, statistics, and expert opinions for the article",
        backstory=(
            "You are an expert researcher with 10 years of experience in content research. "
            "You excel at finding reliable sources, key statistics, and expert quotes "
            "that make articles authoritative and engaging."
        ),
        llm=ChatOpenAI(model=config["researcher"]),
        verbose=True,
    )

    writer = Agent(
        role="Content Writer",
        goal="Write engaging, well-structured blog articles that inform and captivate readers",
        backstory=(
            "You are a professional copywriter with expertise in creating compelling content. "
            "You know how to structure articles with clear introductions, informative body sections, "
            "and memorable conclusions that drive engagement."
        ),
        llm=ChatOpenAI(model=config["writer"]),
        verbose=True,
    )

    editor = Agent(
        role="Quality Editor",
        goal="Polish articles for clarity, grammar, flow, and readability",
        backstory=(
            "You are a senior editor with a keen eye for detail. "
            "You improve sentence structure, fix grammatical errors, enhance flow, "
            "and ensure the content is clear and professional."
        ),
        llm=ChatOpenAI(model=config["editor"]),
        verbose=True,
    )

    seo_specialist = Agent(
        role="SEO Optimizer",
        goal="Optimize content for search engines without sacrificing readability",
        backstory=(
            "You are an SEO expert who balances keyword optimization with user experience. "
            "You add meta descriptions, optimize headings, suggest internal links, "
            "and ensure content ranks well while remaining engaging."
        ),
        llm=ChatOpenAI(model=config["seo"]),
        verbose=True,
    )

    return {
        "researcher": researcher,
        "writer": writer,
        "editor": editor,
        "seo": seo_specialist,
    }

from crewai import Task

def create_tasks(agents: dict, topic: str):
    """Create the content pipeline tasks."""

    research_task = Task(
        description=(
            f"Research the topic: '{topic}'. "
            "Find 5-7 key facts, relevant statistics, and expert opinions. "
            "Include sources where possible. Focus on accuracy and relevance."
        ),
        expected_output=(
            "A research brief containing:\n"
            "- Key facts and statistics\n"
            "- Expert opinions or quotes\n"
            "- Source references\n"
            "- Main themes to cover"
        ),
        agent=agents["researcher"],
    )

    writing_task = Task(
        description=(
            "Write a 800-1000 word blog article based on the research provided. "
            "Include:\n"
            "- An engaging introduction that hooks the reader\n"
            "- 3-4 body sections with clear subheadings\n"
            "- A conclusion with key takeaways\n"
            "Format the article in markdown."
        ),
        expected_output="A draft blog article in markdown format with introduction, body sections, and conclusion",
        agent=agents["writer"],
        context=[research_task],
    )

    editing_task = Task(
        description=(
            "Edit the article for:\n"
            "- Grammar and spelling errors\n"
            "- Sentence structure and flow\n"
            "- Clarity and readability\n"
            "- Consistent tone and style\n"
            "Make improvements while preserving the author's voice."
        ),
        expected_output="A polished blog article with improved clarity, grammar, and flow",
        agent=agents["editor"],
        context=[writing_task],
    )

    seo_task = Task(
        description=(
            "Optimize the article for SEO:\n"
            "- Add a compelling meta description (150-160 characters)\n"
            "- Optimize the title and headings for keywords\n"
            "- Suggest 3-5 target keywords\n"
            "- Ensure proper heading hierarchy (H1, H2, H3)\n"
            "- Add a suggested slug for the URL\n"
            "Return the optimized article with SEO metadata."
        ),
        expected_output=(
            "SEO-optimized article with:\n"
            "- Meta description\n"
            "- Target keywords\n"
            "- Optimized headings\n"
            "- Suggested URL slug"
        ),
        agent=agents["seo"],
        context=[editing_task],
    )

    return [research_task, writing_task, editing_task, seo_task]

from crewai import Crew, Process

def create_content_crew(config: dict = None):
    """Create a content creation crew with the specified configuration."""
    agents = create_agents(config)

    # Tasks will be created when running
    return {
        "agents": agents,
        "config": config or {
            "researcher": "gpt-4o",
            "writer": "gpt-4o",
            "editor": "gpt-3.5-turbo",
            "seo": "gpt-3.5-turbo",
        },
    }

def run_content_crew(crew_data: dict, topic: str):
    """Execute the content creation pipeline."""
    agents = crew_data["agents"]
    tasks = create_tasks(agents, topic)

    crew = Crew(
        agents=list(agents.values()),
        tasks=tasks,
        process=Process.sequential,
        verbose=True,
    )

    result = crew.kickoff()
    return result

# Create crew with default configuration
crew_data = create_content_crew()

# Run a test article
result = run_content_crew(
    crew_data,
    topic="The Future of AI in Healthcare"
)

print("Article created successfully!")
print(result.raw[:500] + "...")

from netra import Netra
from netra.instrumentation.instruments import InstrumentSet

# Initialize Netra with CrewAI and OpenAI instrumentation
Netra.init(
    app_name="contentcraft",
    instruments=set([InstrumentSet.CREWAI, InstrumentSet.OPENAI]),
    trace_content=True,
)

from netra.decorators import workflow

@workflow(name="content-pipeline")
def create_article(topic: str, config_name: str = "default", config: dict = None):
    """Run the content creation pipeline with full tracing."""

    # Set custom attributes for filtering and analysis
    Netra.set_custom_attributes(key="topic", value=topic)
    Netra.set_custom_attributes(key="config_name", value=config_name)

    # Create and run the crew
    crew_data = create_content_crew(config)
    result = run_content_crew(crew_data, topic)

    return {
        "topic": topic,
        "config": config_name,
        "output": result.raw,
        "token_usage": getattr(result, "token_usage", None),
    }

from netra import Netra, SpanType

@workflow(name="content-pipeline-detailed")
def create_article_detailed(topic: str, config_name: str, config: dict):
    """Run pipeline with detailed custom tracing."""

    with Netra.start_span("pipeline-setup") as setup_span:
        setup_span.set_attribute("topic", topic)
        setup_span.set_attribute("config_name", config_name)
        setup_span.set_attribute("model.researcher", config["researcher"])
        setup_span.set_attribute("model.writer", config["writer"])
        setup_span.set_attribute("model.editor", config["editor"])
        setup_span.set_attribute("model.seo", config["seo"])

        crew_data = create_content_crew(config)

    with Netra.start_span("pipeline-execution", as_type=SpanType.AGENT) as exec_span:
        result = run_content_crew(crew_data, topic)
        exec_span.set_attribute("output_length", len(result.raw))

    return {
        "topic": topic,
        "config": config_name,
        "output": result.raw,
    }

# Model configurations to test
CONFIGS = {
    "premium": {
        "name": "Premium (All GPT-4)",
        "researcher": "gpt-4o",
        "writer": "gpt-4o",
        "editor": "gpt-4o",
        "seo": "gpt-4o",
    },
    "budget": {
        "name": "Budget (Mixed)",
        "researcher": "gpt-4o",
        "writer": "gpt-4o",
        "editor": "gpt-3.5-turbo",
        "seo": "gpt-3.5-turbo",
    },
    "economy": {
        "name": "Economy (Minimal GPT-4)",
        "researcher": "gpt-4o",
        "writer": "gpt-3.5-turbo",
        "editor": "gpt-3.5-turbo",
        "seo": "gpt-3.5-turbo",
    },
}

print("Running Premium configuration (all GPT-4)...")

result_premium = create_article(
    topic="The Future of AI in Healthcare",
    config_name="premium-all-gpt4",
    config=CONFIGS["premium"],
)

print(f"Premium article created: {len(result_premium['output'])} characters")

print("Running Budget configuration (mixed models)...")

result_budget = create_article(
    topic="The Future of AI in Healthcare",
    config_name="budget-mixed",
    config=CONFIGS["budget"],
)

print(f"Budget article created: {len(result_budget['output'])} characters")

print("Running Economy configuration (minimal GPT-4)...")

result_economy = create_article(
    topic="The Future of AI in Healthcare",
    config_name="economy-minimal",
    config=CONFIGS["economy"],
)

print(f"Economy article created: {len(result_economy['output'])} characters")

TEST_TOPICS = [
    "The Future of AI in Healthcare",
    "Remote Work Best Practices for 2026",
    "Sustainable Investing for Beginners",
    "How to Build a Personal Brand Online",
    "The Rise of Electric Vehicles",
]

def run_config_comparison():
    """Run all configurations across all topics."""
    results = []

    for config_key, config in CONFIGS.items():
        for topic in TEST_TOPICS:
            print(f"Running {config_key} for: {topic[:30]}...")

            result = create_article(
                topic=topic,
                config_name=config_key,
                config=config,
            )

            results.append({
                "config": config_key,
                "topic": topic,
                "output": result["output"],
                "output_length": len(result["output"]),
            })

    return results

# Run comparison (this will take a while)
# comparison_results = run_config_comparison()

Evaluate the quality of this blog article draft.

Score from 0 to 1 based on:
- Coherence: Does the article flow logically? (0.3)
- Coverage: Does it cover the topic comprehensively? (0.3)
- Engagement: Is it interesting to read? (0.2)
- Structure: Does it have clear intro, body, conclusion? (0.2)

Article:
{output}

Topic:
{input}

function handler(input, output, expectedOutput) {
    // input.draft = the original draft
    // output = the edited version

    const draft = input?.draft || "";
    const edited = output || "";

    // No changes = no improvement
    if (draft === edited) {
        return 0;
    }

    // Calculate similarity (simple length-based heuristic)
    const lengthRatio = edited.length / Math.max(draft.length, 1);

    // Edited should be similar length (0.8 to 1.2x)
    if (lengthRatio < 0.7 || lengthRatio > 1.3) {
        return 0.5; // Major length change might indicate issues
    }

    // Check for improvements
    let score = 0.5; // Base score for making changes

    // Reward similar length (tight editing)
    if (lengthRatio >= 0.9 && lengthRatio <= 1.1) {
        score += 0.3;
    }

    // Reward if output is valid markdown
    if (edited.includes("# ") || edited.includes("## ")) {
        score += 0.2;
    }

    return Math.min(score, 1);
}

function handler(input, output, expectedOutput) {
    const article = output || "";
    const keywords = expectedOutput?.keywords || [];

    let score = 0;

    // Check for meta description (30%)
    const hasMetaDesc = article.toLowerCase().includes("meta description") ||
                        article.toLowerCase().includes("meta:") ||
                        article.includes("**Meta Description**");
    if (hasMetaDesc) {
        score += 0.3;
    }

    // Check for keyword usage (40%)
    if (keywords.length > 0) {
        const articleLower = article.toLowerCase();
        const keywordsFound = keywords.filter(kw =>
            articleLower.includes(kw.toLowerCase())
        );
        score += (keywordsFound.length / keywords.length) * 0.4;
    } else {
        score += 0.2; // Partial credit if no keywords specified
    }

    // Check for proper heading structure (30%)
    const hasH1 = article.includes("# ") && !article.startsWith("## ");
    const hasH2 = article.includes("## ");
    const hasH3 = article.includes("### ");

    if (hasH1) score += 0.1;
    if (hasH2) score += 0.1;
    if (hasH1 && hasH2) score += 0.1;

    return Math.min(score, 1);
}

Evaluate this blog article for publication readiness.

Score from 0 to 1 based on:
- Informativeness: Does it provide valuable information? (0.25)
- Readability: Is it easy to read and understand? (0.25)
- SEO Optimization: Does it have proper structure and keywords? (0.25)
- Professionalism: Is it publication-ready? (0.25)

Article:
{output}

Original Topic:
{input}

TEST_CASES = [
    {
        "id": "TC-001",
        "topic": "The Future of AI in Healthcare",
        "target_keywords": ["AI healthcare", "medical AI", "diagnosis", "treatment"],
        "expected_sections": ["introduction", "benefits", "challenges", "future"],
    },
    {
        "id": "TC-002",
        "topic": "Remote Work Best Practices for 2026",
        "target_keywords": ["remote work", "productivity", "work from home", "hybrid"],
        "expected_sections": ["introduction", "tools", "communication", "conclusion"],
    },
    {
        "id": "TC-003",
        "topic": "Sustainable Investing for Beginners",
        "target_keywords": ["ESG investing", "sustainable", "green funds", "portfolio"],
        "expected_sections": ["introduction", "what is ESG", "how to start", "risks"],
    },
]

def run_quality_evaluation():
    """Run all test cases across all configurations."""
    results = []

    for config_key, config in CONFIGS.items():
        for test_case in TEST_CASES:
            print(f"Running {config_key} for {test_case['id']}...")

            result = create_article(
                topic=test_case["topic"],
                config_name=f"{config_key}-{test_case['id']}",
                config=config,
            )

            results.append({
                "test_id": test_case["id"],
                "config": config_key,
                "topic": test_case["topic"],
                "output": result["output"],
                "keywords": test_case["target_keywords"],
            })

    return results

# Run evaluation
# evaluation_results = run_quality_evaluation()

# Example: Improved editor agent after observing poor edit quality
editor_improved = Agent(
    role="Quality Editor",
    goal="Polish articles for clarity, grammar, flow, and readability",
    backstory=(
        "You are a senior editor with 15 years of experience at major publications. "
        "You have a keen eye for detail and always improve content while preserving "
        "the author's voice. You focus on:\n"
        "1. Fixing grammatical errors and typos\n"
        "2. Improving sentence structure for better flow\n"
        "3. Ensuring consistent tone throughout\n"
        "4. Making complex ideas more accessible\n"
        "You make targeted improvements, not wholesale rewrites."
    ),
    llm=ChatOpenAI(model="gpt-3.5-turbo"),
    verbose=True,
)