CAMEL 多智能体框架中的 RolePlayingWorker基于角色扮演的 Workforce 工作节点完全指南【免费下载链接】camel CAMEL: The first and the best multi-agent framework. Finding the Scaling Law of Agents. https://www.camel-ai.org项目地址: https://gitcode.com/GitHub_Trending/ca/camel导读RolePlayingWorker是 CAMEL 多智能体框架 Workforce 协作引擎中的一种特殊工作节点它不像SingleAgentWorker那样由一个智能体直接执行任务而是在一个节点内部启动一场「AI 助手 × AI 用户」的角色扮演RolePlaying对话让两个不同角色的智能体通过多轮交流共同求解任务最后由独立的 summarize 智能体总结对话产出最终结果。本文以 docs/reference/camel.societies.workforce.role_playing_worker.md 为骨架结合仓库源码、官方文档与可运行示例完整讲解该节点的全部构造参数、内部执行链路、结构化输出机制以及把它接入 Workforce 的实战方法。读完本文你将能独立创建、配置并调试一个基于角色扮演协作的 Workforce 工作节点用于头脑风暴、多方观点碰撞、方案评审等场景。RolePlayingWorker 的设计定位在 CAMEL 的 Workforce 体系中所有可执行任务的节点都继承自Worker抽象基类camel/societies/workforce/worker.py而 docs/key_modules/workforce.md 将工作节点划分为两类SingleAgentWorker最常见的节点内部是一个配置好工具与系统提示词的ChatAgent直接产出结果RolePlayingWorker内部维护一场RolePlaying会话由 assistant 与 user 两个智能体通过对话协作完成任务适合头脑风暴、辩论、多视角探索等场景。RolePlayingWorker的类定义位于 camel/societies/workforce/role_playing_worker.py官方 docstring 将其描述为 A worker node that contains a role playing即一个内含角色扮演会话的工作节点。这意味着对一个外部观察者而言它依然是一个普通的 Worker——通过统一的_process_task接口接收任务并返回TaskState.DONE或TaskState.FAILED但对任务本身而言它使用了一种截然不同的求解方式对话式协作。类签名与构造参数详解依据参考文档与源码RolePlayingWorker的完整构造签名如下class RolePlayingWorker(Worker): def __init__( self, description: str, assistant_role_name: str, user_role_name: str, assistant_agent_kwargs: Optional[Dict] None, user_agent_kwargs: Optional[Dict] None, summarize_agent_kwargs: Optional[Dict] None, chat_turn_limit: int 20, use_structured_output_handler: bool True ): ...各参数含义见 role_playing_worker.py参数类型默认值说明descriptionstr必填节点的描述信息用于在 Workforce 中标识该节点例如A role-playing session for system design.assistant_role_namestr必填角色扮演中 assistant 智能体的角色名例如Software Developeruser_role_namestr必填角色扮演中 user 智能体的角色名例如Solution Architectassistant_agent_kwargsOptional[Dict]None初始化 assistant 智能体的关键字参数如model、tools、system_message等user_agent_kwargsOptional[Dict]None初始化 user 智能体的关键字参数含义同上summarize_agent_kwargsOptional[Dict]None初始化 summarize 智能体的关键字参数不传时也会用默认系统提示词创建一个 summarize 智能体chat_turn_limitint20角色扮演的最大对话轮数上限防止对话无限进行use_structured_output_handlerboolTrue是否使用结构化输出处理器提示词引导 正则提取解析结果替代原生结构化输出注意区分两个默认值RolePlayingWorker类自身的chat_turn_limit默认为 20见 role_playing_worker.py而Workforce.add_role_playing_worker方法的默认值仅为 3见 workforce.py后文会详细说明。关于 summarize_agent 的隐式创建源码显示role_playing_worker.py即便不传summarize_agent_kwargs构造函数也会自动创建一个ChatAgent作为总结者并注入一个内置系统提示词You are a good summarizer. You will be presented with scenarios where an assistant and a user with specific roles are trying to solve a task. Your job is summarizing the result of the task based on the chat history.也就是说RolePlayingWorker内部实际运行着三个智能体assistant、user 与 summarizer。若你通过summarize_agent_kwargs传入自定义配置源码会将内置的system_message合并进你的参数字典summarize_agent_dict[system_message] summ_sys_msg因此你不需要也不应该重复指定 system_message。将 RolePlayingWorker 接入 Workforce通常情况下你不需要直接实例化RolePlayingWorker而是通过Workforce.add_role_playing_worker()注册节点workforce.py。该方法与类构造参数一一对应但有两处差异需要留意默认对话轮数不同add_role_playing_worker的chat_turn_limit默认是3传入后会原样传递给RolePlayingWorker结构化输出开关自动继承它会把 Workforce 自身的use_structured_output_handler透传给新创建的RolePlayingWorkerworkforce.py保证整支队伍行为一致。此外add_role_playing_worker会在 Workforce 处于RUNNING状态时抛出RuntimeError(Cannot add workers while workforce is running. Pause the workforce first.)即只能在空闲或暂停paused状态下动态添加节点——这一约束同样适用于add_single_agent_worker与add_workforce并在测试 test/workforce/test_workforce.py 的test_dynamic_worker_types用例中得到验证将 Workforce 置为PAUSED后可依次添加 RolePlayingWorker 与嵌套 Workforce。官方文档示例系统设计方案评审docs/key_modules/workforce.md 给出了一个典型的注册示例——让「解决方案架构师」与「软件开发者」通过角色扮演完成系统设计from camel.societies.workforce import Workforce workforce Workforce(System Design Team) workforce.add_role_playing_worker( descriptionA role-playing session for system design., assistant_role_nameSoftware Developer, user_role_nameSolution Architect, assistant_agent_kwargsdict( system_messageYou are a software developer responsible for implementing the system. ), user_agent_kwargsdict( system_messageYou are a solution architect responsible for the high-level design. ), chat_turn_limit5, ) # ... process a task with this workforce ...仓库完整示例旅游研究团队仓库中提供了可直接运行的完整示例 examples/workforce/role_playing_with_agents.py其核心片段展示了如何在角色扮演节点中为 assistant 挂载真实工具、并通过链式调用同时注册其他 Workerfrom camel.agents.chat_agent import ChatAgent from camel.models import ModelFactory from camel.societies.workforce import Workforce from camel.toolkits import GoogleMapsToolkit, SearchToolkit, WeatherToolkit from camel.types import ModelPlatformType, ModelType function_list [ *SearchToolkit().get_tools(), *WeatherToolkit().get_tools(), *GoogleMapsToolkit().get_tools(), ] assistant_agent_kwargs dict( modelModelFactory.create( model_platformModelPlatformType.DEFAULT, model_typeModelType.DEFAULT, ), toolsfunction_list, ) workforce Workforce(a travel group) workforce.add_role_playing_worker( descriptionresearch Group, assistant_role_nameSearcher, user_role_nameProfessor, assistant_agent_kwargsassistant_agent_kwargs, user_agent_kwargsuser_agent_kwargs, summarize_agent_kwargs{}, chat_turn_limit1, ).add_single_agent_worker( tour guide, guide_agent ).add_single_agent_worker(planner, planner_agent)这个示例有三点实战价值assistant_agent_kwargs中通过ModelFactory.create(...)显式指定模型并将SearchToolkit、WeatherToolkit、GoogleMapsToolkit的工具函数列表注入 assistant使角色扮演中的助理角色具备搜索、查天气、查地图的真实能力与ROLEPLAY_PROCESS_TASK_PROMPT中 It is recommended that tools be actively called when needed 的引导一致角色扮演节点research Group与两个SingleAgentWorkertour guide、planner可以共存于同一 Workforce 中说明RolePlayingWorker可以与普通单智能体节点混合编排、并行执行示例后半部分还调用get_workforce_log_tree()、get_workforce_kpis()与dump_workforce_logs()输出队伍日志与 KPI可用于观察角色扮演节点的运行过程。内部执行链路从任务到结果RolePlayingWorker的核心逻辑集中在_process_task方法role_playing_worker.py整条链路可分为四个阶段阶段一组装任务提示词。利用Worker._get_dep_tasks_infoworker.py把依赖任务dependencies的 id、content、result 拼接成文本然后填充ROLEPLAY_PROCESS_TASK_PROMPT模板prompts.py模板中包含任务内容、父任务内容、依赖任务结果与附加信息四部分。阶段二启动 RolePlaying 会话。使用任务提示词作为task_prompt创建RolePlaying会话with_task_specifyFalse不额外改写任务描述然后调用role_play_session.init_chat()得到初始输入消息。阶段三多轮对话直到终止。进入while n self.chat_turn_limit循环每轮调用role_play_session.astep(input_msg)推进一次对话并依次检查终止条件assistant 或 user 的响应中出现terminated标记例如触发了终止词、达到模型上下文上限打印终止原因并跳出循环user 回复内容中包含CAMEL_TASK_DONE标志立即终止这是 CAMEL RolePlaying 约定俗成的任务完成信号轮数达到chat_turn_limit上限强制结束。每轮对话内容AI User: .../AI Assistant: ...都会被追加进chat_history列表作为最终总结的素材。阶段四总结并产出结构化结果。将ROLEPLAY_SUMMARIZE_PROMPTprompts.py与聊天历史拼接交给 summarize 智能体要求其基于历史对话总结任务执行结果并输出包含content与failed两个字段的TaskResult。随后task.result task_result.content if is_task_insufficient(task): # 经 is_task_result_insufficient 校验 return TaskState.FAILED return TaskState.DONE源码role_playing_worker.py显示若总结失败或结果内容不充分is_task_result_insufficient校验会把任务标记为TaskState.FAILED供 Workforce 上层触发失败恢复策略。从更宏观的视角看_process_task由基类Worker._process_single_taskworker.py统一调度任务状态通过task.set_state(task_state)回写并通过TaskChannel.return_task归还任务若_process_task内部抛出异常也会被捕获并记录为TaskState.FAILED。Worker基类还提供了基于asyncio的无上限并行任务监听循环_listen_to_channelworker.py因此多个角色扮演节点可以被并行调度。use_structured_output_handler两条结果解析路径use_structured_output_handler是理解该节点输出可靠性的关键参数它决定总结结果的解析走哪条路径role_playing_worker.py路径 A结构化输出处理器默认开启True。使用 camel/societies/workforce/structured_output_handler.py 中的StructuredOutputHandlergenerate_structured_prompt根据TaskResult的 Pydantic schema 生成一段「结构化输出要求」把必填字段、字段类型、合法示例{content: ..., failed: false}与只输出 JSON、不得夹杂其他文字的强约束拼进原始总结提示词summarize 智能体按提示词输出 JSONparse_structured_response依次尝试多种提取策略标准 json 代码块正则、无代码块的纯 JSON 正则、针对嵌套对象的兜底正则再到 schema 专属正则SCHEMA_PATTERNS最后进行 Pydantic 校验全部失败时回退到fallback_values{content: Task summarization failed, failed: True}保证节点不会因解析异常而崩溃。路径 B原生结构化输出False。直接调用self.summarize_agent.step(prompt, response_formatTaskResult)依赖模型对原生response_format参数的支持若response.msg.parsed is None则构造一个failedTrue的兜底TaskResult。两条路径的取舍正如参考文档所述开启处理器默认时通过提示词引导 正则提取保证与不原生支持结构化输出的模型兼容关闭时则依赖模型的原生response_format能力。StructuredOutputHandler在整个 Workforce 中被复用任务分配、失败分析、Worker 创建等同样依赖它因此该开关实际上是全局一致的行为约定。参数调优建议与注意事项综合参考文档、源码与测试实践中有以下几点值得注意对话轮数按需设置RolePlayingWorker类默认chat_turn_limit20而add_role_playing_worker默认只有3。轮数越大对话越充分但成本与延迟越高若任务只需一次确认如示例中的chat_turn_limit1可显著降低成本。需要完全继承类默认行为时请显式传入chat_turn_limit20。不要重复传 system_message 给 summarize源码会强制覆盖 summarize 智能体的system_message自定义时只需提供model等其余参数。运行中不可动态增删节点添加RolePlayingWorker前需确保 Workforce 处于IDLE或PAUSED状态若需在运行中调整可先使用 Workforce 的pause()运行时控制 APIdocs/key_modules/workforce.md 中提到的pause()/resume()/stop_gracefully()等。结果失败会被上层接管总结失败或内容不充分时节点返回TaskState.FAILEDWorkforce 会根据failure_handling_config触发 retry / replan / decompose 等恢复策略因此 RolePlayingWorker 不必自行处理重试。测试覆盖佐证仓库测试 test/workforce/test_workforce.py 的test_dynamic_worker_types验证了在PAUSED状态下添加 RolePlayingWorker 的合法性可作为调试参考。小结RolePlayingWorker把 CAMEL 标志性的 RolePlaying 双角色协作范式封装成了 Workforce 中一个即插即用的标准节点对外保持统一的 Worker 接口对内则运行任务提示词组装 → 角色扮演多轮对话 → summarize 结构化总结的完整流水线。它既适用于头脑风暴、方案评审、多视角研究等天然适合对话协作的任务也可以与SingleAgentWorker混合组队、并行调度。理解其参数语义尤其是chat_turn_limit的两处默认值差异与use_structured_output_handler的双路径解析机制是在生产环境中高效编排这类节点的关键。【免费下载链接】camel CAMEL: The first and the best multi-agent framework. Finding the Scaling Law of Agents. https://www.camel-ai.org项目地址: https://gitcode.com/GitHub_Trending/ca/camel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考