feat: Support task group when decomposing in workforce

[Zephyroam]

Description

Describe your changes in detail (optional if the linked issue already contains a detailed description of the changes).

Closes #3580

• Introduce TaskGroup in camel.tasks to support grouped tasks.
• Introduce a new workforce mode AUTO_DECOMPOSE_GROUPED.

Checklist

Go over all the following points, and put an x in all the boxes that apply.

• I have read the CONTRIBUTION guide (required)
• I have linked this PR to an issue using the Development section on the right sidebar or by adding Fixes #issue-number in the PR description (required)
• I have checked if any dependencies need to be added or updated in pyproject.toml and uv lock
• I have updated the tests accordingly (required for a bug fix or a new feature)
• I have updated the documentation if needed:
• I have added examples if this is a new feature

If you are unsure about any of these, don't hesitate to ask. We are here to help!

[coderabbitai]

Important

Review skipped

Auto reviews are disabled on this repository.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

✨ Finishing touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[Wendong-Fan]

Hi @Zephyroam thank you for your contribution to this PR! I've done review of the implementation and would like to share some concerns about the design approach.

While the TaskGroup concept is interesting for organizing related tasks, I have concerns about whether this approach provides meaningful benefits over the existing batch assignment mechanism.

1. Current Batch Processing is Already Efficient

The existing AUTO_DECOMPOSE mode already performs batch assignment via _call_coordinator_for_assignment(). The
coordinator processes all tasks in a single call and determines both worker assignments and dependencies. The new AUTO_DECOMPOSE_GROUPED mode doesn't fundamentally reduce coordinator workload, it just shifts the granularity from tasks to groups.

2. Reduced Assignment Flexibility

In the current approach, each task can be assigned to the most suitable worker:
Task 0.1 (Apple financials) -> financial_analyst_1
Task 0.2 (Google financials) -> financial_analyst_2

With task groups, all tasks within a group must be assigned to the same worker:
Group 0.1 (Financial Analysis) -> worker_A // All tasks go to one worker

This loses the ability to leverage specialized workers for individual tasks within the same logical phase.

3. Coarser Dependency Control

The current system allows precise task-level dependencies:
{"task_id": "0.5", "dependencies": ["0.1", "0.3"]} // Only specific tasks

The grouped approach only supports group-level dependencies, which then get expanded:

• "ALL" mode: Every task in the downstream group depends on ALL tasks in the upstream group

  • "ANY" mode: Only works correctly when group sizes match; otherwise falls back to "ALL" behavior

  This is less flexible and can create unnecessary blocking dependencies.

  4. Increased Code Complexity

  This PR adds significant complexity:

  • New TaskGroup class and related models (TaskGroupAssignment, TaskGroupAssignResult)
  • New parsing logic (parse_response_grouped)
  • New decomposition method (decompose_grouped)
  • New assignment method (_call_coordinator_for_assignment_grouped)
  • Dependency expansion logic (_expand_group_to_task_dependencies)
  • 230+ lines of new prompt templates

  This parallel code path increases maintenance burden without clear benefits.

  Questions for Consideration

  1. What specific problem is this PR trying to solve that the current batch assignment doesn't address? Did you test current approach and saw the bottleneck?
  2. Are there change showing improved performance?

  I'd be happy to discuss this further

[Zephyroam]

Thank you! I think your concern is true. That is the reason why I put it in a new mode.

To demonstrate the possible improvement, I test multiple_single_agents.py with the following task:

    human_task = Task(
        content=(
            "Conduct a comprehensive cross-country market analysis for launching a new "
            "electric scooter service across 20 European countries.\n\n"
            "Countries (20): Germany, France, Italy, Spain, Netherlands, Belgium, Sweden, "
            "Norway, Denmark, Finland, Poland, Czechia, Austria, Switzerland, Portugal, "
            "Ireland, Greece, Hungary, Romania, Bulgaria.\n\n"
            "The analysis should cover: "
            "1. Current market size and key competitors. "
            "2. Target audience and their preferences. "
            "3. Local regulations and potential challenges. "
            "Finally, synthesize all findings into a summary report."
        ),
        id='0',
    )

The task agent splits the task into 41 tasks:

• 20 tasks, gather all the information for each country (task group 1 when grouping is enabled)
• 20 tasks, produce a summary for each country (task group 2 when grouping is enabled)
• 1 task, compile into one summary (task group 2 when grouping is enabled)

The task related input and output length for the coordinator when grouping is disabled: 15881, 4366.

The task related input and output length for the coordinator when grouping is enabled: 775, 532.

The input and output length got 20x and 8x reduction, respectively.

Logs are provided in the following:

20_countries_analysis_group_disabled.log
20_countries_analysis_group_enabled.log

[Zephyroam]

I also did the following to simplify the changes:

• Move _expand_group_to_task_dependencies to utils.py
• Merge func Task.decompose with Task.decompose_grouped
• Merge func _call_coordinator_for_assignment with _call_coordinator_for_assignment_grouped

The design principle of the new mode is not to introduce a totally new pipeline, but just adding a bypass based on the current design.