[Feat] Support more running mode in workforce #3157

Ol1ver0413 · 2025-09-18T16:11:01Z

Description

Add pipeline running mode in workforce. #1663

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 · 2025-09-18T16:11:10Z

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
Commit unit tests in branch workforce_pipeline

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

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

Ol1ver0413 · 2025-09-21T14:55:38Z

graph TD
    A[Task: Generate 5 AI/ML Papers] --> B[Literature Researcher]
    B --> C{Fork: 5 Parallel Tasks}
    
    C --> D1[Summary Specialist 1<br/>Summarize Paper 1]
    C --> D2[Summary Specialist 2<br/>Summarize Paper 2]
    C --> D3[Summary Specialist 3<br/>Summarize Paper 3]
    C --> D4[Summary Specialist 4<br/>Summarize Paper 4]
    C --> D5[Summary Specialist 5<br/>Summarize Paper 5]
    
    D1 --> E{Join: Collect Summaries}
    D2 --> E
    D3 --> E
    D4 --> E
    D5 --> E
    
    E --> F[Research Synthesizer<br/>Analyze AI/ML Trends]
    F --> G[Final Result]
    
    style A fill:#e1f5fe
    style B fill:#fff3e0
    style D1 fill:#f3e5f5
    style D2 fill:#f3e5f5
    style D3 fill:#f3e5f5
    style D4 fill:#f3e5f5
    style D5 fill:#f3e5f5
    style F fill:#e8f5e8
    style G fill:#fff9c4

Ol1ver0413 · 2025-09-21T14:56:11Z

graph TD
    A[Task: Generate 4 RESTful API Types] --> B[API Researcher]
    B --> C{Fork: 4 Parallel Tasks}
    
    C --> D1[API Analyst 1<br/>Analyze API 1]
    C --> D2[API Analyst 2<br/>Analyze API 2]
    C --> D3[API Analyst 3<br/>Analyze API 3]
    C --> D4[API Analyst 4<br/>Analyze API 4]
    
    D1 --> E{Join: Collect Analyses}
    D2 --> E
    D3 --> E
    D4 --> E
    
    E --> F[Documentation Writer<br/>Generate API Usage Guide]
    F --> G[Final Result]
    
    style A fill:#e1f5fe
    style B fill:#fff3e0
    style D1 fill:#f3e5f5
    style D2 fill:#f3e5f5
    style D3 fill:#f3e5f5
    style D4 fill:#f3e5f5
    style F fill:#e8f5e8
    style G fill:#fff9c4

Ol1ver0413 · 2025-09-21T14:56:42Z

graph TD
    A[Task: Code Review Analysis] --> B[Code Scanner]
    B --> C{Fork: 3 Parallel Tasks}
    
    C --> D1[Code Reviewer 1<br/>Review File 1]
    C --> D2[Code Reviewer 2<br/>Review File 2]
    C --> D3[Code Reviewer 3<br/>Review File 3]
    
    D1 --> E{Join: Collect Reviews}
    D2 --> E
    D3 --> E
    
    E --> F[Review Summarizer<br/>Generate Comprehensive Report]
    F --> G[Final Result]
    
    style A fill:#e1f5fe
    style B fill:#fff3e0
    style D1 fill:#f3e5f5
    style D2 fill:#f3e5f5
    style D3 fill:#f3e5f5
    style F fill:#e8f5e8
    style G fill:#fff9c4

Ol1ver0413 · 2025-09-21T15:24:03Z

Hey @Wendong-Fan @fengju0213 ! I've already add pipeline mode in workforce, and maybe we can try more cases to test this mode.

Wendong-Fan · 2025-09-23T03:39:27Z

Hey @Wendong-Fan @fengju0213 ! I've already add pipeline mode in workforce, and maybe we can try more cases to test this mode.

thanks @Ol1ver0413 !

hesamsheikh

Thanks for the comprehensive PR @Ol1ver0413. I added a comment. Also I think the method namings could be a bit more consistent and remembrable.

  # before
  workforce.add_pipeline_task()
  workforce.fork_pipeline()
  workforce.add_parallel_pipeline_tasks()
  builder.fork()  
  builder.add_parallel_tasks() 

  # after
  workforce.pipeline_add()
  workforce.pipeline_fork()
  workforce.pipeline_join()
  workforce.pipeline_build()
  builder.add()
  builder.fork()
  builder.join()
  builder.build()

examples/workforce/pipeline_workflow_example.py

hesamsheikh · 2025-09-29T16:05:19Z

camel/societies/workforce/workforce.py

+        expected_task_ids = {task.id for task in self._pending_tasks}
+        expected_task_ids.update(task.id for task in self._completed_tasks)
+
+        completed_successful_ids = {
+            task.id for task in self._completed_tasks 
+            if task.state == TaskState.DONE
+        }
+


doesn't this create a situation if which if any of the branched tasks fails, the whole pipeline is marked as failed and the join task never executes?

doesn't this create a situation if which if any of the branched tasks fails, the whole pipeline is marked as failed and the join task never executes?

Thanks @hesamsheikh ! Maybe I need to add some branch task failure handling mechanisms into the pipeline.

Hi, @hesamsheikh. Since the original design follows a pipeline pattern, I didn’t perform task restructuring or decomposition as in the decompose mode. Therefore, I decided that if a task fails, it would still be considered successful, and the error message would be passed to the join stage. We can discuss whether there might be a better approach.

Well this method seems a bit confusing and hard to understand anyway. Could you maybe simplify it, or explain it a bit in the comments? I know how it works and what it does, but not the intent of it.

doesn't this create a situation if which if any of the branched tasks fails, the whole pipeline is marked as failed and the join task never executes?

Yeah, I'll add more comments.

a7m-1st · 2025-10-02T18:05:31Z

camel/societies/workforce/workforce.py

+
+        # Clear existing tasks and dependencies
+        self._pending_tasks.clear()
+        self._task_dependencies.clear()
+        self._assignees.clear()
+


@Ol1ver0413 , thanks for the PR, but as of now, Is the pipeline designed that it could interrupt running tasks or is it only with predefined tasks only? i.e. before workforce.start()

@Ol1ver0413 , thanks for the PR, but as of now, Is the pipeline designed that it could interrupt running tasks or is it only with predefined tasks only? i.e. before workforce.start()

Thanks @a7m-1st. The pipeline mode was designed for sequential predefined tasks, which cannot be interrupted by human. But maybe we can have a further discussion for it. Thanks again!

Wendong-Fan · 2025-10-19T13:11:13Z

hi @Ol1ver0413

Hope you're doing well!

Just wanted to check in on this PR. It looks like there's some feedback that needs to be addressed, along with some failing in pre-commit checks.

Please let us know if you have any questions or need a hand with anything. We're looking forward to getting this merged!

Cheers

Ol1ver0413 · 2025-10-19T14:42:00Z

hi @Ol1ver0413

Hope you're doing well!

Just wanted to check in on this PR. It looks like there's some feedback that needs to be addressed, along with some failing in pre-commit checks.

Please let us know if you have any questions or need a hand with anything. We're looking forward to getting this merged!

Cheers

Hi, @Wendong-Fan!
I’ve been handling some work transitions lately, which may have caused me to overlook enhancing the running mode in the workforce. I’ll make sure to complete it soon.
Thanks again.

review-notebook-app · 2025-10-20T15:18:16Z

Check out this pull request on

See visual diffs & provide feedback on Jupyter Notebooks.

Powered by ReviewNB

…into workforce_pipeline

a7m-1st · 2025-10-21T19:53:41Z

@Ol1ver0413 There is merge conflicts I guess as I am seeing 140+ files changes.
cc: @Wendong-Fan

Ol1ver0413 · 2025-10-22T01:26:38Z

@Ol1ver0413 There is merge conflicts I guess as I am seeing 140+ files changes. cc: @Wendong-Fan

Yeah, I see that too. It's quite weird because I only modified workforce.py and example file. Let me take a check.

…orce_pipeline

a7m-1st · 2025-10-22T18:03:12Z

Cool. all good now, thanks for the update will review it asap !

hesamsheikh

Thanks for your updates. I checked the PR again and added a few more comments.

hesamsheikh · 2025-10-23T12:25:15Z

camel/societies/workforce/utils.py

+            if hasattr(self, '_last_parallel_tasks') and self._last_parallel_tasks:
+                wait_for = self._last_parallel_tasks
+                # Clear the parallel tasks after using them
+                delattr(self, '_last_parallel_tasks')


why delete the whole attribute? consider emptying the attribute or setting it to none.

hesamsheikh · 2025-10-23T12:25:57Z

camel/societies/workforce/utils.py

+        self.task_counter = 0
+        self._last_task_id = None
+        if hasattr(self, '_last_parallel_tasks'):
+            delattr(self, '_last_parallel_tasks')


again, pretty unusual to delete the attribute

hesamsheikh · 2025-10-23T12:27:27Z

camel/societies/workforce/workforce.py

+    async def _process_task_with_hybrid(self, task: Task) -> Task:
+        """Process task using hybrid approach (partial auto-decomposition)."""
+        # For now, fall back to auto-decompose mode
+        # This can be extended to support more sophisticated hybrid logic
+        return await self._process_task_with_auto_decompose(task)


is this for future use? it's unused

hesamsheikh · 2025-10-23T12:36:29Z

camel/societies/workforce/workforce.py

+        expected_task_ids = {task.id for task in self._pending_tasks}
+        expected_task_ids.update(task.id for task in self._completed_tasks)


why are pending tasks counted as completed ones? seems like an issue.

Ol1ver0413 · 2025-10-24T14:54:25Z

Thanks for your updates. I checked the PR again and added a few more comments.

Thanks @hesamsheikh for reviewing. Let me take a check.

…into workforce_pipeline

a7m-1st

Thanks @Ol1ver0413 for your work. All worked well with my side, except some small points I would like your opinion on, and a small change perhaps overlooked when merging conflicts 🙌

a7m-1st · 2025-10-27T14:57:58Z

camel/societies/workforce/workforce.py

+        auto_depend: bool = True,
+    ) -> 'Workforce':
+        """Add a task to the pipeline with support for chaining.


Maybe we don't need to wrap the return in quotes?

a7m-1st · 2025-10-27T15:06:34Z

camel/societies/workforce/workforce.py

+
+    def pipeline_fork(self, task_contents: List[str]) -> 'Workforce':
+        """Create parallel branches from the current task.
+        


I can see that those parallel tasks only support text, would have been nice to perhaps support list of Tasks. Or is that not viable?

a7m-1st · 2025-10-27T15:16:08Z

camel/societies/workforce/workforce.py

+        # Collect results from all pipeline tasks
+        task.result = self._collect_pipeline_results()
+        task.state = (
+            TaskState.DONE if self._all_pipeline_tasks_successful() 
+            else TaskState.FAILED
+        )
+


Maybe we can reset the mode to AUTO_DECOMPOSE after completing? Perhaps another function to change the mode of workforce after init()

As when using any of pipeline's functions it switches mode automatically in _ensure_pipeline_builder() to PIPELINE

a7m-1st · 2025-10-27T15:20:39Z

camel/societies/workforce/workforce.py

-                and (
-                    task.additional_info is None
-                    or not task.additional_info.get(
-                        "_needs_decomposition", False
-                    )
-                )


Can we revert this @Ol1ver0413?
More details on this on #3238.

This fix was added as we are mixing independent & subtasks in the same _pending_queue differentiated by _needs_decomposition flag. So we delay positing the independent tasks.

Ol1ver0413 · 2025-10-28T06:14:42Z

Thanks @Ol1ver0413 for your work. All worked well with my side, except some small points I would like your opinion on, and a small change perhaps overlooked when merging conflicts 🙌

Thanks @a7m-1st . I'll take a review asap.

hesamsheikh

Please re-request a review after making the necessary changes @Ol1ver0413

wk_pipline_0.1

2d33864

Ol1ver0413 self-assigned this Sep 18, 2025

wk_pipeline_0.2

9ce04ee

Ol1ver0413 marked this pull request as ready for review September 21, 2025 15:01

Ol1ver0413 requested review from Wendong-Fan and fengju0213 September 21, 2025 15:02

Wendong-Fan requested review from a7m-1st, hesamsheikh and nitpicker55555 September 29, 2025 07:04

Wendong-Fan added the Review Required PR need to be reviewed label Sep 29, 2025

Wendong-Fan added this to Project Camel Sep 29, 2025

Wendong-Fan added this to the Sprint 38 milestone Sep 29, 2025

Wendong-Fan linked an issue Sep 29, 2025 that may be closed by this pull request

[Feature Request] Support more running mode in workforce #1663

Open

2 tasks

hesamsheikh reviewed Sep 29, 2025

View reviewed changes

a7m-1st reviewed Oct 2, 2025

View reviewed changes

Wendong-Fan added Waiting for Update PR has been reviewed, need to be updated based on review comment and removed Review Required PR need to be reviewed labels Oct 19, 2025

merge and resolve conflict

56527f6

Ol1ver0413 added 3 commits October 20, 2025 21:33

resolve conflict

061aa18

Merge branch 'workforce_pipeline' of https://github.com/camel-ai/camel …

17bfac8

…into workforce_pipeline

wk_pipeline_0.3

9648550

Ol1ver0413 requested review from a7m-1st and hesamsheikh October 21, 2025 16:15

Ol1ver0413 added 3 commits October 21, 2025 21:32

Merge remote-tracking branch 'upstream/workforce_pipeline' into workf…

604699f

…orce_pipeline

solve merge conflict

ad771b6

del unnecessary file

3bc8da5

hesamsheikh requested changes Oct 23, 2025

View reviewed changes

Ol1ver0413 added 2 commits October 26, 2025 22:36

wk_pipeline_0.4

4386e93

Merge branch 'workforce_pipeline' of https://github.com/camel-ai/camel …

92804d6

…into workforce_pipeline

Ol1ver0413 requested a review from hesamsheikh October 26, 2025 14:38

Wendong-Fan added Review Required PR need to be reviewed and removed Waiting for Update PR has been reviewed, need to be updated based on review comment labels Oct 26, 2025

fengju0213 modified the milestones: Sprint 38, Sprint 41 Oct 27, 2025

Merge branch 'master' into workforce_pipeline

c248666

a7m-1st requested changes Oct 27, 2025

View reviewed changes

hesamsheikh requested changes Oct 28, 2025

View reviewed changes

		expected_task_ids = {task.id for task in self._pending_tasks}
		expected_task_ids.update(task.id for task in self._completed_tasks)


		def pipeline_fork(self, task_contents: List[str]) -> 'Workforce':
		"""Create parallel branches from the current task.

[Feat] Support more running mode in workforce #3157

Are you sure you want to change the base?

[Feat] Support more running mode in workforce #3157

Uh oh!

Conversation

Ol1ver0413 commented Sep 18, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Checklist

Uh oh!

coderabbitai bot commented Sep 18, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Review skipped

Uh oh!

Ol1ver0413 commented Sep 21, 2025

Uh oh!

Ol1ver0413 commented Sep 21, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Ol1ver0413 commented Sep 21, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Ol1ver0413 commented Sep 21, 2025

Uh oh!

Wendong-Fan commented Sep 23, 2025

Uh oh!

hesamsheikh left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

a7m-1st Oct 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Wendong-Fan commented Oct 19, 2025

Uh oh!

Ol1ver0413 commented Oct 19, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

review-notebook-app bot commented Oct 20, 2025

Uh oh!

a7m-1st commented Oct 21, 2025

Uh oh!

Ol1ver0413 commented Oct 22, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

a7m-1st commented Oct 22, 2025

Uh oh!

hesamsheikh left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Ol1ver0413 commented Oct 24, 2025

Uh oh!

a7m-1st left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Ol1ver0413 commented Sep 18, 2025 •

edited

Loading

coderabbitai bot commented Sep 18, 2025 •

edited

Loading

Ol1ver0413 commented Sep 21, 2025 •

edited

Loading

Ol1ver0413 commented Sep 21, 2025 •

edited

Loading

a7m-1st Oct 2, 2025 •

edited

Loading

Ol1ver0413 commented Oct 19, 2025 •

edited

Loading

Ol1ver0413 commented Oct 22, 2025 •

edited

Loading