refactor(workforce): Refactor WorkforceCallback and all related callbacks to async interface #3363

coolbeevip · 2025-11-03T13:00:46Z

Description

Fixes #3352

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-11-03T13:01:06Z

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 issue-3352

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.}

fengju0213

Everything else looks great—my only concern is that we should keep backward compatibility (or
provide sync wrappers) so existing code doesn’t break when upgrading.
ask for some comments cc @Wendong-Fan

fengju0213 · 2025-11-07T07:56:31Z

camel/societies/workforce/workforce.py

Thanks for the excellent contribution—the async-first design should definitely help with
throughput and allow callbacks to run I/O without blocking. One question: since the long-
standing public Workforce APIs (e.g., add_single_agent_worker, reset, clone) are now
asynchronous, do we plan to retain the former synchronous entry points (even if just as thin
wrappers) to preserve backward compatibility for existing scripts and examples?

Thanks for pointing this out — sync wrappers can introduce extra maintenance overhead and complexity. I’ll defer to the @Wendong-Fan , any thoughts or preferences on keeping them vs. deprecating?

Hey @coolbeevip,

Would you mind re-requesting a review after a decision is made on whether to include backwards compatibility.

@waleedalzarooni @fengju0213 Thanks for the thorough review and pointing out the backward compatibility concern!

I agree this is important to avoid breaking existing code. Since I'm not strongly opinionated here, I'd like to defer to the community's preference:

Options:

Add sync wrappers (e.g., add_single_agent_worker_sync() that calls the async version under the hood):

Pros: Minimal disruption for users.

Cons: Extra maintenance overhead (two APIs to keep in sync).

Deprecate old sync APIs (with warnings) and guide users to migrate to async:

Pros: Clean async-first design long-term.

Cons: Requires users to update code (but we can provide migration guide).

Force async-only (no wrappers):

Pros: Simplest codebase.

Cons: Breaks all existing sync usage immediately.

What do you think? Any preference? Once we decide, I'll implement and re-request reviews.

thanks @coolbeevip , @fengju0213 and @waleedalzarooni , i would prefer implement Option 2, keeping method() as deprecated sync wrappers and adding method_async() as the primary async API

i added another enhance PR #3464, feel free to review and leave your comments

Thanks @Wendong-Fan. I've merged this PR and updated the related code in examples.

fengju0213 · 2025-11-07T09:47:09Z

camel/benchmarks/browsecomp.py

                        input_message, response_format=QueryResponse
                    )
                elif isinstance(pipeline_template, Workforce):
                    pipeline = pipeline_template.clone()  # type: ignore[assignment]


still calls Workforce.clone() synchronously

Thanks for the reminder. This line uses '# type: ignore[...]' to suppress mypy checks. I will fix this issue.

waleedalzarooni

Great integration @coolbeevip!,

just a couple comments on my side, also additionally there is an example file that still uses the old synchronous syntax, here is some guidance on what to change:

The file examples/workforce/simple_workforce_mcp.py was NOT updated in this PR but calls the now-async method:

# Line 47 in examples/workforce/simple_workforce_mcp.py
def main():  #  Synchronous function
    workforce = Workforce(description="Simple Analysis Team")
    analyst = ChatAgent(system_message=analyst_msg)
    workforce.add_single_agent_worker(  #  Calling async method without await
        description="General purpose analyst", worker=analyst

waleedalzarooni · 2025-11-14T08:29:29Z

camel/societies/workforce/workforce.py

-        self._child_listening_tasks.append(
-            asyncio.create_task(new_node.start())
-        )
+        self._child_listening_tasks.append(await new_node.start())


Suggested change

self._child_listening_tasks.append(await new_node.start())

self._child_listening_tasks.append(asyncio.create_task(new_node.start()))

The start() method is an async coroutine that runs indefinitely - it's a long-running background listener that processes messages from a channel. When you write await new_node.start():

The execution blocks at this line

It waits for start() to complete

But start() never completes - it runs forever!

The function never returns, so the worker is never added to the list

waleedalzarooni · 2025-11-14T08:46:17Z

camel/societies/workforce/workforce.py


        for child in self._children:
-            self._notify_worker_created(child)
+            asyncio.run(self._notify_worker_created(child))


Suggested change

asyncio.run(self._notify_worker_created(child))

self._notify_worker_created(child)

asyncio.run() cannot be called from a running event loop

illustration of issue:

async def main(): # This creates an event loop and runs main() workforce = Workforce(...) #PROBLEM HAPPENS HERE asyncio.run(main()) # Event loop is already running here When Workforce.__init__() is called: def __init__(self): # ... initialization code ... for child in self._children: asyncio.run(self._notify_worker_created(child)) RuntimeError: asyncio.run() cannot be called from a running event loop

Thank you for that.

The __init__ function only uses asyncio.run for callbacks when a child object is provided. I'll first add test cases for this part.

Hi @waleedalzarooni, could you provide a complete test case that demonstrates how to pass the children parameter when constructing Workforce? I couldn’t find any usage of the children parameter in the codebase. A minimal, reproducible example would be helpful. Thanks!

No problem

async def test_workforce_with_children(): worker = SingleAgentWorker( description="Test worker", worker=ChatAgent("You are a test agent."), ) workforce = Workforce( description="Test workforce", children=[worker], ) assert len(workforce._children) == 1 assert isinstance(workforce._children[0], SingleAgentWorker)

coolbeevip · 2025-11-14T14:45:35Z

Great integration @coolbeevip!,

just a couple comments on my side, also additionally there is an example file that still uses the old synchronous syntax, here is some guidance on what to change:

The file examples/workforce/simple_workforce_mcp.py was NOT updated in this PR but calls the now-async method:
# Line 47 in examples/workforce/simple_workforce_mcp.py
def main():  #  Synchronous function
    workforce = Workforce(description="Simple Analysis Team")
    analyst = ChatAgent(system_message=analyst_msg)
    workforce.add_single_agent_worker(  #  Calling async method without await
        description="General purpose analyst", worker=analyst

Thanks for the reminder. done

…acks to async interface

#3464)

…ing in async context

waleedalzarooni

LGTM

github-actions bot added the Review Required PR need to be reviewed label Nov 3, 2025

coolbeevip marked this pull request as ready for review November 4, 2025 00:50

coolbeevip changed the title ~~WIP: refactor(workforce): Refactor WorkforceCallback and all related callbacks to async interface~~ refactor(workforce): Refactor WorkforceCallback and all related callbacks to async interface Nov 4, 2025

Wendong-Fan assigned coolbeevip Nov 5, 2025

Wendong-Fan added this to Project Camel Nov 5, 2025

Wendong-Fan added this to the Sprint 41 milestone Nov 5, 2025

waleedalzarooni requested review from fengju0213 and waleedalzarooni November 7, 2025 06:22

fengju0213 reviewed Nov 7, 2025

View reviewed changes

waleedalzarooni reviewed Nov 14, 2025

View reviewed changes

coolbeevip force-pushed the issue-3352 branch from 151326a to a88f9ce Compare November 14, 2025 14:17

coolbeevip requested review from fengju0213 and waleedalzarooni November 14, 2025 15:47

coolbeevip force-pushed the issue-3352 branch 3 times, most recently from 29e6b91 to c55c18c Compare November 26, 2025 15:42

LIHUA919 approved these changes Nov 29, 2025

View reviewed changes

coolbeevip and others added 3 commits December 5, 2025 11:06

refactor(workforce): Refactor WorkforceCallback and all related callb…

75b52cc

…acks to async interface

enhance: Refactor WorkforceCallback and all related callbacks to asyn… (

82425a5

#3464)

refactor(workforce): Enhance worker initialization and callback handl…

f45163f

…ing in async context

coolbeevip force-pushed the issue-3352 branch from a96e1d5 to f45163f Compare December 5, 2025 03:07

Merge branch 'master' into issue-3352

d9bad64

waleedalzarooni approved these changes Dec 5, 2025

View reviewed changes

coolbeevip requested a review from Wendong-Fan December 5, 2025 14:20

	self._child_listening_tasks.append(await new_node.start())
	self._child_listening_tasks.append(asyncio.create_task(new_node.start()))

	asyncio.run(self._notify_worker_created(child))
	self._notify_worker_created(child)

refactor(workforce): Refactor WorkforceCallback and all related callbacks to async interface #3363

Are you sure you want to change the base?

refactor(workforce): Refactor WorkforceCallback and all related callbacks to async interface #3363

Uh oh!

Conversation

coolbeevip commented Nov 3, 2025

Description

Checklist

Uh oh!

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

Uh oh!

Review skipped

Uh oh!

fengju0213 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

Options:

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!

waleedalzarooni 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!

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!

coolbeevip commented Nov 14, 2025

Uh oh!

waleedalzarooni left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

coderabbitai bot commented Nov 3, 2025 •

edited

Loading