尧图网页设计 
CAMEL Workforce 工具层源码解析WorkerConf、任务分配模型与故障恢复机制【免费下载链接】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导读camel.societies.workforce.utils是 CAMEL 多智能体协作框架Workforce中负责数据结构定义与通用工具的核心模块。本文以该模块的 API 参考文档docs/reference/camel.societies.workforce.utils.md为骨架结合其底层源码 camel/societies/workforce/utils.py系统讲解 Worker 配置模型WorkerConf、任务结果与分配模型TaskResult、TaskAssignment、TaskAssignResult、故障恢复策略RecoveryStrategy、FailureContext、TaskAnalysisResult以及运行状态守卫装饰器check_if_running的设计原理与实际用法。读完本文你将能够独立阅读 CAMEL Workforce 源码中的数据结构定义理解任务依赖解析、故障恢复决策和并发安全控制的底层实现并能在自己的 Workforce 应用中正确使用这些模型与工具函数。模块定位Workforce 工具的数据契约层CAMEL 的 Workforce 是一个可编排多个 Worker智能体并行/串行协作完成复杂任务的框架。在 camel/societies/workforce/utils.py 中定义了一组Pydantic 数据模型作为 Worker 与协调器之间传递的结构化契约与通用工具函数如运行状态检查装饰器、流水线任务构建器等。从源码结构看该模块承担三类职责配置契约WorkerConf定义了一个工作节点Work Node被创建时需要的角色、系统提示词与描述信息任务数据流契约TaskResult、TaskAssignment、TaskAssignResult定义了任务执行结果与任务分配的结构故障恢复契约RecoveryStrategy、FailureContext、TaskAnalysisResult定义了任务失败时的分析与恢复决策结构。这些模型被 camel/societies/workforce/workforce.py、camel/societies/workforce/worker.py、camel/societies/workforce/structured_output_handler.py 以及 camel/societies/workforce/base.py 共同引用是理解整个 Workforce 运行机制的入口。说明该模块还包含WorkflowMetadata、WorkflowConfig、QualityEvaluation、FailureHandlingConfig、PipelineTaskBuilder等源码类它们是上述文档提及的核心模型的延伸实现本文一并覆盖。WorkerConf工作节点的配置契约WorkerConf是 API 文档中首先列出的类定义了一个 Worker工作节点的配置class WorkerConf(BaseModel): rThe configuration of a worker. role: str Field( descriptionThe role of the agent working in the work node. ) sys_msg: str Field( descriptionThe system message that will be sent to the agent in the node. ) description: str Field( descriptionThe description of the new work node itself. )三个字段均为必填字符串含义如下字段类型必填含义rolestr是工作节点中智能体扮演的角色名例如Data Analystsys_msgstr是发送给该节点智能体的系统提示词System Message例如You are an expert data analyst.descriptionstr是新工作节点自身的描述用于解释该节点能胜任的工作实际调用场景协调器创建 Worker 时的结构化输出WorkerConf在 workforce.py 的 worker 创建流程中被反复使用见第 4231、4251、4253、4272、4281 行附近。以使用结构化输出处理器use_structured_output_handlerTrue的路径为例协调器智能体会在提示词中带上WorkerConf的 schema 与示例enhanced_prompt ( self.structured_handler.generate_structured_prompt( base_promptprompt, schemaWorkerConf, examples[ { description: Data analysis specialist, role: Data Analyst, sys_msg: You are an expert data analyst., } ], ) ) response self.coordinator_agent.step(enhanced_prompt) result self.structured_handler.parse_structured_response( response_content, schemaWorkerConf, fallback_values{ description: fWorker for task: {task.content}, role: Task Specialist, sys_msg: fYou are a specialist for: {task.content}, }, )如果协调器返回空响应代码会兜底创建一个通用配置roleGeneral Assistant如果返回的是 dict则通过WorkerConf(**result)重建实例workforce.py 第 4251 行。最终new_node_conf.role与new_node_conf.sys_msg被传给_create_new_agent创建真正的智能体节点workforce.py 第 4293-4296 行。同时structured_output_handler.py 在第 420、490 行也为WorkerConf提供了默认实例与失败回退实例保证 LLM 输出不符合 schema 时流程仍可继续。TaskResult 与 TaskAssignResult任务结果与分配的数据结构TaskResult任务执行结果class TaskResult(BaseModel): rThe result of a task. content: str Field(descriptionThe result of the task.) failed: bool Field( defaultFalse, descriptionFlag indicating whether the task processing failed., )TaskResult有两个字段content必填任务的执行结果内容failed可选默认False标记任务处理是否失败。TaskAssignment单条任务分配class TaskAssignment(BaseModel): rAn individual task assignment within a batch. task_id: str Field(descriptionThe ID of the task to be assigned.) assignee_id: str Field( descriptionThe ID of the worker/workforce to assign the task to. ) dependencies: List[str] Field( default_factorylist, descriptionList of task IDs that must complete before this task. This is critical for the task decomposition and execution., )三个字段中task_id是被分配任务的 IDassignee_id是承接任务的 Worker 或子 Workforce 的 IDdependencies是前置任务 ID 列表——这一字段对任务分解Task Decomposition与执行顺序至关重要。_split_and_stripLLM 输出的容错解析源码中通过staticmethod定义了_split_and_stripstaticmethod def _split_and_strip(dep_str: str) - List[str]: rUtility to split a comma separated string and strip whitespace. return [d.strip() for d in dep_str.split(,) if d.strip()]它把一个以逗号分隔的字符串按,切分并去除空白同时过滤掉空串。validate_dependencies字段校验器validate_dependencies是dependencies字段的modebefore校验器会在字段赋值前执行field_validator(dependencies, modebefore) def validate_dependencies(cls, v) - List[str]: if v is None: return [] # Handle empty string or comma-separated string from LLM if isinstance(v, str): return TaskAssignment._split_and_strip(v) return v设计意图很明确允许 LLM 将dependencies输出为逗号分隔字符串甚至空字符串或 None然后统一转换为List[str]避免下游逻辑因校验错误而中断。例如 LLM 输出task_1, task_2, task_3会被规整为[task_1, task_2, task_3]输出空字符串则得到[]。TaskAssignResult单批分配的整体结果class TaskAssignResult(BaseModel): rThe result of task assignment for both single and batch assignments. assignments: List[TaskAssignment] Field( descriptionList of task assignments. )TaskAssignResult只包含一个assignments字段是TaskAssignment的列表同时覆盖单任务与批量任务的分配场景。在 structured_output_handler.py 第 418、482 行中当 LLM 输出无法解析时会返回TaskAssignResult(assignments[])作为回退值。RecoveryStrategy任务失败的五种恢复策略RecoveryStrategy继承自str, Enum定义了任务失败后的五种恢复策略class RecoveryStrategy(str, Enum): rStrategies for handling failed tasks. RETRY retry REPLAN replan DECOMPOSE decompose CREATE_WORKER create_worker REASSIGN reassign def __str__(self): return self.value def __repr__(self): return fRecoveryStrategy.{self.name}枚举成员值含义RETRYretry直接重试失败的任务REPLANreplan重新规划任务内容后再次执行DECOMPOSEdecompose将失败任务拆解为更小的子任务CREATE_WORKERcreate_worker创建新的专用 Worker 来处理该任务REASSIGNreassign将任务重新分配给其他 Worker由于继承自str枚举值可以直接与字符串比较或序列化__str__返回其值如retry__repr__返回形如RecoveryStrategy.RETRY的可读形式。FailureHandlingConfig恢复策略的开关配置源码中与RecoveryStrategy配套的是FailureHandlingConfig它允许用户按需启用或禁用恢复策略class FailureHandlingConfig(BaseModel): max_retries: int Field( default3, ge1, descriptionMaximum retry attempts before giving up on a task, ) enabled_strategies: Optional[List[RecoveryStrategy]] Field( defaultNone, descriptionList of enabled recovery strategies. None means all enabled. Empty list means no recovery (immediate failure). Can be strings like [retry, replan] or RecoveryStrategy enums., ) halt_on_max_retries: bool Field( defaultTrue, descriptionWhether to halt workforce when max retries exceeded, )配置语义如下max_retries默认 3最小值 1放弃任务前的最大重试次数enabled_strategies允许使用的恢复策略列表。None表示全部启用配合 LLM 分析空列表[]表示不采取任何恢复策略、失败任务立即标记为失败仅[retry]时使用无 LLM 分析的简单重试。可以传字符串如[retry, replan]或RecoveryStrategy枚举halt_on_max_retries默认True当任务超过最大重试次数时是否暂停整个 Workforce为False时任务标记失败、工作流继续类似 PIPELINE 模式行为。validate_enabled_strategies校验器会把字符串列表小写化后转换为RecoveryStrategy枚举非法策略会抛出带合法选项列表的ValueError。源码中的 docstring 给出了四种典型用法# 使用字符串列表简单方式 config FailureHandlingConfig( enabled_strategies[retry, replan, decompose], ) # 使用枚举列表 config FailureHandlingConfig( enabled_strategies[ RecoveryStrategy.RETRY, RecoveryStrategy.REPLAN, ] ) # 仅简单重试 config FailureHandlingConfig( enabled_strategies[retry], max_retries2, ) # 不恢复——失败任务立即标记为失败 config FailureHandlingConfig( enabled_strategies[], )FailureContext 与 TaskAnalysisResult失败上下文与统一分析结果FailureContext任务失败的上下文信息class FailureContext(BaseModel): rContext information about a task failure. task_id: str Field(descriptionID of the failed task) task_content: str Field(descriptionContent of the failed task) failure_count: int Field( descriptionNumber of times this task has failed ) error_message: str Field(descriptionDetailed error message) worker_id: Optional[str] Field( defaultNone, descriptionID of the worker that failed ) task_depth: int Field( descriptionDepth of the task in the decomposition hierarchy ) additional_info: Optional[str] Field( defaultNone, descriptionAdditional context about the task )该模型把一次任务失败的全部上下文打包失败任务 ID 与内容、累计失败次数、错误信息、失败 Worker ID可空、任务在分解层级中的深度task_depth以及附加信息可空。这些字段足以支撑 LLM 或规则引擎做出合理的恢复决策。RecoveryDecision 的演变已被 TaskAnalysisResult 取代API 文档中列出了RecoveryDecisionDecision on how to recover from a task failure.但当前源码中并不存在这个类。从仓库证据看其职责已被 utils.py 中的TaskAnalysisResult完全承担RecoveryDecision名称仅残留在历史文档中。阅读源码时请以TaskAnalysisResult为准。TaskAnalysisResult故障分析与质量评估的统一结果class TaskAnalysisResult(BaseModel): # 公共字段——始终填充 reasoning: str Field( descriptionExplanation for the analysis result or recovery decision ) recovery_strategy: Optional[RecoveryStrategy] Field( defaultNone, descriptionRecommended recovery strategy: retry, replan, decompose, create_worker, or reassign. None indicates no recovery needed (quality sufficient)., ) modified_task_content: Optional[str] Field( defaultNone, descriptionModified task content if strategy requires replan, ) # 质量评估专属字段——仅质量评估时填充 quality_score: Optional[int] Field( defaultNone, ge0, le100, descriptionQuality score from 0 to 100 (only for quality evaluation). None indicates this is a failure analysis., ) issues: List[str] Field( default_factorylist, descriptionList of issues found. For failures: error details. For quality evaluation: quality issues., )TaskAnalysisResult将失败恢复决策与质量评估结果统一到一个结构中做失败分析时只填充reasoning与recovery_strategy做质量评估时额外填充quality_score0-100与issues。它还提供两个只读属性property def is_quality_evaluation(self) - bool: return self.quality_score is not None property def quality_sufficient(self) - bool: return ( self.quality_score is not None and self.quality_score 60 and self.recovery_strategy is None )is_quality_evaluation通过quality_score是否非空判断这是质量评估还是失败分析quality_sufficient质量评估中当quality_score 60且无推荐恢复策略时视为质量达标对失败分析结果恒为False。在 test/workforce/test_workforce.py 中第 559-673 行大量测试用例验证了_analyze_task返回的TaskAnalysisResult实例及其recovery_strategy字段如RecoveryStrategy.RETRY、RecoveryStrategy.REPLAN说明该模型是 Workforce 内部故障恢复分析的标准输出格式。此外structured_output_handler.py 第 426-430、498-503 行在解析失败时会默认返回RecoveryStrategy.RETRY的TaskAnalysisResult兜底实例保证恢复决策链路不中断。check_if_runningWorkforce 运行状态的守卫装饰器check_if_running是文档中唯一以函数形式出现的工具它是一个装饰器工厂用于校验 Workforce/Worker 是否处于预期的运行状态并提供自动重试与异常处理容错。def check_if_running( running: bool, max_retries: int 3, retry_delay: float 1.0, handle_exceptions: bool False, ) - Callable:参数说明参数类型默认值含义runningbool必填期望的运行状态True表示期望正在运行False表示期望未在运行max_retriesint3操作失败时的最大重试次数设为0可禁用重试retry_delayfloat1.0每次重试之间等待的秒数handle_exceptionsboolFalse为True时捕获并记录异常而不向上抛出为False时异常会继续传播异常行为RuntimeError当 Workforce 不处于期望状态、且重试已耗尽或已禁用时抛出例如期望未运行却检测到正在运行或反之错误消息形如The workforce is running. Cannot perform the operation reset.Exception被装饰函数本身抛出的任何异常在handle_exceptionsFalse且重试耗尽时会被重新抛出。实现原理装饰器内部通过wraps(func)保留原函数元信息wrapper中维护重试计数器检查self._running ! running状态不符时若有剩余重试次数则记录 warning 日志、time.sleep(retry_delay)后重试否则抛出RuntimeError被装饰函数抛出其他异常时同样按剩余重试次数决定重试或抛出handle_exceptionsTrue时记录 error 日志并返回None对于消息中包含workforce is的RuntimeError即状态校验失败会立即重新抛出而不重试——因为等待也无法改变状态错误避免无意义的重试循环。实际使用位置该装饰器被广泛用于 Workforce 与 Worker 的公开方法上实现并发安全的生命周期控制camel/societies/workforce/base.py 第 38 行check_if_running(False)装饰reset方法确保节点在未运行时才能重置camel/societies/workforce/worker.py 第 93、150、207 行check_if_running(False)装饰set_channel等操作第 212 行用check_if_running(True)装饰需要正在运行前置条件的方法camel/societies/workforce/workforce.py 第 2668、3208、3671、5281、5821 行使用check_if_running(False)第 5836 行使用check_if_running(True)。典型场景当 Workforce 正在处理任务时调用方试图reset()会先收到 warning 日志并按retry_delay间隔重试若状态始终不符则最终抛出RuntimeError从而避免在运行中重置导致的数据损坏。延伸PipelineTaskBuilder——带依赖的流水线任务构建虽然 API 文档未单列但PipelineTaskBuilder是 utils.py 中与任务依赖紧密相关的辅助类对理解TaskAssignment.dependencies的语义很有帮助。它支持链式 API 构建带依赖关系的任务序列# 串行链式Step 2 依赖 Step 1Step 3 依赖 Step 2 builder.add(Step 1).add(Step 2).add(Step 3) # 并行分支 汇合 builder.add(Collect Data).add_parallel_tasks([ Technical Analysis, Fundamental Analysis ]).add_sync_task(Generate Report) # fork/join 别名 builder.add(Collect Data).fork([ Technical Analysis, Fundamental Analysis ]).join(Generate Report)核心方法add(content, task_idNone, dependenciesNone, additional_infoNone, auto_dependTrue)添加任务auto_dependTrue且未指定dependencies时自动依赖最近添加的任务add_parallel_tasks(task_contents, ...)批量添加可并行执行的任务任务 ID 形如parallel_0_counter添加后_last_task_id置空要求后续任务显式声明依赖add_sync_task(content, wait_forNone, ...)添加同步任务自动等待最近一批并行任务wait_for缺省且无并行任务时抛出ValueErrorbuild()返回任务列表前会用 DFS 检测循环依赖发现环时抛出ValueErrorfork()/join()分别是add_parallel_tasks与add_sync_task的语义化别名get_task_info()返回{task_count: ..., tasks: [{id, content, dependencies}]}形式的流水线概览。内部通过_task_registrytask_id → Task 映射实现快速查重与依赖校验并把依赖解析为真正的Task对象列表传入Task构造函数源码中from camel.tasks import Task。模块全貌utils.py 中的其他模型除上述核心模型外utils.py 还定义了以下辅助模型API 文档未单独列出属于源码级补充GENERIC_ROLE_NAMES与is_generic_role_name判断角色名是否为assistant、agent、user、system、worker、helper等通用名大小写不敏感。通用角色名无法体现智能体的实际用途触发回退逻辑——从 LLM 生成的agent_title或description中寻找更具体的标识同时避免用通用名作为工作流文件夹名WorkflowMetadata工作流元数据session_id、working_directory、created_at、updated_at、workflow_version、agent_id、message_count用于工作流历史与版本管理WorkflowConfig工作流内存管理配置max_workflows_per_role100、文件名后缀_workflow、存储目录workforce_workflows、enable_versioningTrue、default_max_files_to_load3集中管理配置项避免散落各处QualityEvaluation质量评估结果quality_sufficient、quality_score、issues、recovery_strategy、modified_task_content其 docstring 明确标注deprecated建议改用TaskAnalysisResult保留仅为向后兼容。小结camel.societies.workforce.utils模块是整个 Workforce 框架的数据契约层WorkerConf驱动协调器创建 Worker 的结构化输出TaskResult/TaskAssignment/TaskAssignResult承载任务结果与分配信息validate_dependencies保证了 LLM 输出的容错性RecoveryStrategy/FailureContext/TaskAnalysisResult构建了任务失败的恢复决策体系而check_if_running则通过重试与异常处理的容错设计守护着 Workforce 的生命周期并发安全。阅读本文后你可以顺着 camel/societies/workforce/utils.py 继续深入 workforce.py 的 worker 创建与失败恢复流程或通过 test/workforce/test_workforce.py 验证上述模型的实际行为。【免费下载链接】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），仅供参考 CAMEL Workforce 工具层源码解析：WorkerConf、任务分配模型与故障恢复机制
CAMEL Workforce 工具层源码解析WorkerConf、任务分配模型与故障恢复机制【免费下载链接】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导读camel.societies.workforce.utils是 CAMEL 多智能体协作框架Workforce中负责数据结构定义与通用工具的核心模块。本文以该模块的 API 参考文档docs/reference/camel.societies.workforce.utils.md为骨架结合其底层源码 camel/societies/workforce/utils.py系统讲解 Worker 配置模型WorkerConf、任务结果与分配模型TaskResult、TaskAssignment、TaskAssignResult、故障恢复策略RecoveryStrategy、FailureContext、TaskAnalysisResult以及运行状态守卫装饰器check_if_running的设计原理与实际用法。读完本文你将能够独立阅读 CAMEL Workforce 源码中的数据结构定义理解任务依赖解析、故障恢复决策和并发安全控制的底层实现并能在自己的 Workforce 应用中正确使用这些模型与工具函数。模块定位Workforce 工具的数据契约层CAMEL 的 Workforce 是一个可编排多个 Worker智能体并行/串行协作完成复杂任务的框架。在 camel/societies/workforce/utils.py 中定义了一组Pydantic 数据模型作为 Worker 与协调器之间传递的结构化契约与通用工具函数如运行状态检查装饰器、流水线任务构建器等。从源码结构看该模块承担三类职责配置契约WorkerConf定义了一个工作节点Work Node被创建时需要的角色、系统提示词与描述信息任务数据流契约TaskResult、TaskAssignment、TaskAssignResult定义了任务执行结果与任务分配的结构故障恢复契约RecoveryStrategy、FailureContext、TaskAnalysisResult定义了任务失败时的分析与恢复决策结构。这些模型被 camel/societies/workforce/workforce.py、camel/societies/workforce/worker.py、camel/societies/workforce/structured_output_handler.py 以及 camel/societies/workforce/base.py 共同引用是理解整个 Workforce 运行机制的入口。说明该模块还包含WorkflowMetadata、WorkflowConfig、QualityEvaluation、FailureHandlingConfig、PipelineTaskBuilder等源码类它们是上述文档提及的核心模型的延伸实现本文一并覆盖。WorkerConf工作节点的配置契约WorkerConf是 API 文档中首先列出的类定义了一个 Worker工作节点的配置class WorkerConf(BaseModel): rThe configuration of a worker. role: str Field( descriptionThe role of the agent working in the work node. ) sys_msg: str Field( descriptionThe system message that will be sent to the agent in the node. ) description: str Field( descriptionThe description of the new work node itself. )三个字段均为必填字符串含义如下字段类型必填含义rolestr是工作节点中智能体扮演的角色名例如Data Analystsys_msgstr是发送给该节点智能体的系统提示词System Message例如You are an expert data analyst.descriptionstr是新工作节点自身的描述用于解释该节点能胜任的工作实际调用场景协调器创建 Worker 时的结构化输出WorkerConf在 workforce.py 的 worker 创建流程中被反复使用见第 4231、4251、4253、4272、4281 行附近。以使用结构化输出处理器use_structured_output_handlerTrue的路径为例协调器智能体会在提示词中带上WorkerConf的 schema 与示例enhanced_prompt ( self.structured_handler.generate_structured_prompt( base_promptprompt, schemaWorkerConf, examples[ { description: Data analysis specialist, role: Data Analyst, sys_msg: You are an expert data analyst., } ], ) ) response self.coordinator_agent.step(enhanced_prompt) result self.structured_handler.parse_structured_response( response_content, schemaWorkerConf, fallback_values{ description: fWorker for task: {task.content}, role: Task Specialist, sys_msg: fYou are a specialist for: {task.content}, }, )如果协调器返回空响应代码会兜底创建一个通用配置roleGeneral Assistant如果返回的是 dict则通过WorkerConf(**result)重建实例workforce.py 第 4251 行。最终new_node_conf.role与new_node_conf.sys_msg被传给_create_new_agent创建真正的智能体节点workforce.py 第 4293-4296 行。同时structured_output_handler.py 在第 420、490 行也为WorkerConf提供了默认实例与失败回退实例保证 LLM 输出不符合 schema 时流程仍可继续。TaskResult 与 TaskAssignResult任务结果与分配的数据结构TaskResult任务执行结果class TaskResult(BaseModel): rThe result of a task. content: str Field(descriptionThe result of the task.) failed: bool Field( defaultFalse, descriptionFlag indicating whether the task processing failed., )TaskResult有两个字段content必填任务的执行结果内容failed可选默认False标记任务处理是否失败。TaskAssignment单条任务分配class TaskAssignment(BaseModel): rAn individual task assignment within a batch. task_id: str Field(descriptionThe ID of the task to be assigned.) assignee_id: str Field( descriptionThe ID of the worker/workforce to assign the task to. ) dependencies: List[str] Field( default_factorylist, descriptionList of task IDs that must complete before this task. This is critical for the task decomposition and execution., )三个字段中task_id是被分配任务的 IDassignee_id是承接任务的 Worker 或子 Workforce 的 IDdependencies是前置任务 ID 列表——这一字段对任务分解Task Decomposition与执行顺序至关重要。_split_and_stripLLM 输出的容错解析源码中通过staticmethod定义了_split_and_stripstaticmethod def _split_and_strip(dep_str: str) - List[str]: rUtility to split a comma separated string and strip whitespace. return [d.strip() for d in dep_str.split(,) if d.strip()]它把一个以逗号分隔的字符串按,切分并去除空白同时过滤掉空串。validate_dependencies字段校验器validate_dependencies是dependencies字段的modebefore校验器会在字段赋值前执行field_validator(dependencies, modebefore) def validate_dependencies(cls, v) - List[str]: if v is None: return [] # Handle empty string or comma-separated string from LLM if isinstance(v, str): return TaskAssignment._split_and_strip(v) return v设计意图很明确允许 LLM 将dependencies输出为逗号分隔字符串甚至空字符串或 None然后统一转换为List[str]避免下游逻辑因校验错误而中断。例如 LLM 输出task_1, task_2, task_3会被规整为[task_1, task_2, task_3]输出空字符串则得到[]。TaskAssignResult单批分配的整体结果class TaskAssignResult(BaseModel): rThe result of task assignment for both single and batch assignments. assignments: List[TaskAssignment] Field( descriptionList of task assignments. )TaskAssignResult只包含一个assignments字段是TaskAssignment的列表同时覆盖单任务与批量任务的分配场景。在 structured_output_handler.py 第 418、482 行中当 LLM 输出无法解析时会返回TaskAssignResult(assignments[])作为回退值。RecoveryStrategy任务失败的五种恢复策略RecoveryStrategy继承自str, Enum定义了任务失败后的五种恢复策略class RecoveryStrategy(str, Enum): rStrategies for handling failed tasks. RETRY retry REPLAN replan DECOMPOSE decompose CREATE_WORKER create_worker REASSIGN reassign def __str__(self): return self.value def __repr__(self): return fRecoveryStrategy.{self.name}枚举成员值含义RETRYretry直接重试失败的任务REPLANreplan重新规划任务内容后再次执行DECOMPOSEdecompose将失败任务拆解为更小的子任务CREATE_WORKERcreate_worker创建新的专用 Worker 来处理该任务REASSIGNreassign将任务重新分配给其他 Worker由于继承自str枚举值可以直接与字符串比较或序列化__str__返回其值如retry__repr__返回形如RecoveryStrategy.RETRY的可读形式。FailureHandlingConfig恢复策略的开关配置源码中与RecoveryStrategy配套的是FailureHandlingConfig它允许用户按需启用或禁用恢复策略class FailureHandlingConfig(BaseModel): max_retries: int Field( default3, ge1, descriptionMaximum retry attempts before giving up on a task, ) enabled_strategies: Optional[List[RecoveryStrategy]] Field( defaultNone, descriptionList of enabled recovery strategies. None means all enabled. Empty list means no recovery (immediate failure). Can be strings like [retry, replan] or RecoveryStrategy enums., ) halt_on_max_retries: bool Field( defaultTrue, descriptionWhether to halt workforce when max retries exceeded, )配置语义如下max_retries默认 3最小值 1放弃任务前的最大重试次数enabled_strategies允许使用的恢复策略列表。None表示全部启用配合 LLM 分析空列表[]表示不采取任何恢复策略、失败任务立即标记为失败仅[retry]时使用无 LLM 分析的简单重试。可以传字符串如[retry, replan]或RecoveryStrategy枚举halt_on_max_retries默认True当任务超过最大重试次数时是否暂停整个 Workforce为False时任务标记失败、工作流继续类似 PIPELINE 模式行为。validate_enabled_strategies校验器会把字符串列表小写化后转换为RecoveryStrategy枚举非法策略会抛出带合法选项列表的ValueError。源码中的 docstring 给出了四种典型用法# 使用字符串列表简单方式 config FailureHandlingConfig( enabled_strategies[retry, replan, decompose], ) # 使用枚举列表 config FailureHandlingConfig( enabled_strategies[ RecoveryStrategy.RETRY, RecoveryStrategy.REPLAN, ] ) # 仅简单重试 config FailureHandlingConfig( enabled_strategies[retry], max_retries2, ) # 不恢复——失败任务立即标记为失败 config FailureHandlingConfig( enabled_strategies[], )FailureContext 与 TaskAnalysisResult失败上下文与统一分析结果FailureContext任务失败的上下文信息class FailureContext(BaseModel): rContext information about a task failure. task_id: str Field(descriptionID of the failed task) task_content: str Field(descriptionContent of the failed task) failure_count: int Field( descriptionNumber of times this task has failed ) error_message: str Field(descriptionDetailed error message) worker_id: Optional[str] Field( defaultNone, descriptionID of the worker that failed ) task_depth: int Field( descriptionDepth of the task in the decomposition hierarchy ) additional_info: Optional[str] Field( defaultNone, descriptionAdditional context about the task )该模型把一次任务失败的全部上下文打包失败任务 ID 与内容、累计失败次数、错误信息、失败 Worker ID可空、任务在分解层级中的深度task_depth以及附加信息可空。这些字段足以支撑 LLM 或规则引擎做出合理的恢复决策。RecoveryDecision 的演变已被 TaskAnalysisResult 取代API 文档中列出了RecoveryDecisionDecision on how to recover from a task failure.但当前源码中并不存在这个类。从仓库证据看其职责已被 utils.py 中的TaskAnalysisResult完全承担RecoveryDecision名称仅残留在历史文档中。阅读源码时请以TaskAnalysisResult为准。TaskAnalysisResult故障分析与质量评估的统一结果class TaskAnalysisResult(BaseModel): # 公共字段——始终填充 reasoning: str Field( descriptionExplanation for the analysis result or recovery decision ) recovery_strategy: Optional[RecoveryStrategy] Field( defaultNone, descriptionRecommended recovery strategy: retry, replan, decompose, create_worker, or reassign. None indicates no recovery needed (quality sufficient)., ) modified_task_content: Optional[str] Field( defaultNone, descriptionModified task content if strategy requires replan, ) # 质量评估专属字段——仅质量评估时填充 quality_score: Optional[int] Field( defaultNone, ge0, le100, descriptionQuality score from 0 to 100 (only for quality evaluation). None indicates this is a failure analysis., ) issues: List[str] Field( default_factorylist, descriptionList of issues found. For failures: error details. For quality evaluation: quality issues., )TaskAnalysisResult将失败恢复决策与质量评估结果统一到一个结构中做失败分析时只填充reasoning与recovery_strategy做质量评估时额外填充quality_score0-100与issues。它还提供两个只读属性property def is_quality_evaluation(self) - bool: return self.quality_score is not None property def quality_sufficient(self) - bool: return ( self.quality_score is not None and self.quality_score 60 and self.recovery_strategy is None )is_quality_evaluation通过quality_score是否非空判断这是质量评估还是失败分析quality_sufficient质量评估中当quality_score 60且无推荐恢复策略时视为质量达标对失败分析结果恒为False。在 test/workforce/test_workforce.py 中第 559-673 行大量测试用例验证了_analyze_task返回的TaskAnalysisResult实例及其recovery_strategy字段如RecoveryStrategy.RETRY、RecoveryStrategy.REPLAN说明该模型是 Workforce 内部故障恢复分析的标准输出格式。此外structured_output_handler.py 第 426-430、498-503 行在解析失败时会默认返回RecoveryStrategy.RETRY的TaskAnalysisResult兜底实例保证恢复决策链路不中断。check_if_runningWorkforce 运行状态的守卫装饰器check_if_running是文档中唯一以函数形式出现的工具它是一个装饰器工厂用于校验 Workforce/Worker 是否处于预期的运行状态并提供自动重试与异常处理容错。def check_if_running( running: bool, max_retries: int 3, retry_delay: float 1.0, handle_exceptions: bool False, ) - Callable:参数说明参数类型默认值含义runningbool必填期望的运行状态True表示期望正在运行False表示期望未在运行max_retriesint3操作失败时的最大重试次数设为0可禁用重试retry_delayfloat1.0每次重试之间等待的秒数handle_exceptionsboolFalse为True时捕获并记录异常而不向上抛出为False时异常会继续传播异常行为RuntimeError当 Workforce 不处于期望状态、且重试已耗尽或已禁用时抛出例如期望未运行却检测到正在运行或反之错误消息形如The workforce is running. Cannot perform the operation reset.Exception被装饰函数本身抛出的任何异常在handle_exceptionsFalse且重试耗尽时会被重新抛出。实现原理装饰器内部通过wraps(func)保留原函数元信息wrapper中维护重试计数器检查self._running ! running状态不符时若有剩余重试次数则记录 warning 日志、time.sleep(retry_delay)后重试否则抛出RuntimeError被装饰函数抛出其他异常时同样按剩余重试次数决定重试或抛出handle_exceptionsTrue时记录 error 日志并返回None对于消息中包含workforce is的RuntimeError即状态校验失败会立即重新抛出而不重试——因为等待也无法改变状态错误避免无意义的重试循环。实际使用位置该装饰器被广泛用于 Workforce 与 Worker 的公开方法上实现并发安全的生命周期控制camel/societies/workforce/base.py 第 38 行check_if_running(False)装饰reset方法确保节点在未运行时才能重置camel/societies/workforce/worker.py 第 93、150、207 行check_if_running(False)装饰set_channel等操作第 212 行用check_if_running(True)装饰需要正在运行前置条件的方法camel/societies/workforce/workforce.py 第 2668、3208、3671、5281、5821 行使用check_if_running(False)第 5836 行使用check_if_running(True)。典型场景当 Workforce 正在处理任务时调用方试图reset()会先收到 warning 日志并按retry_delay间隔重试若状态始终不符则最终抛出RuntimeError从而避免在运行中重置导致的数据损坏。延伸PipelineTaskBuilder——带依赖的流水线任务构建虽然 API 文档未单列但PipelineTaskBuilder是 utils.py 中与任务依赖紧密相关的辅助类对理解TaskAssignment.dependencies的语义很有帮助。它支持链式 API 构建带依赖关系的任务序列# 串行链式Step 2 依赖 Step 1Step 3 依赖 Step 2 builder.add(Step 1).add(Step 2).add(Step 3) # 并行分支 汇合 builder.add(Collect Data).add_parallel_tasks([ Technical Analysis, Fundamental Analysis ]).add_sync_task(Generate Report) # fork/join 别名 builder.add(Collect Data).fork([ Technical Analysis, Fundamental Analysis ]).join(Generate Report)核心方法add(content, task_idNone, dependenciesNone, additional_infoNone, auto_dependTrue)添加任务auto_dependTrue且未指定dependencies时自动依赖最近添加的任务add_parallel_tasks(task_contents, ...)批量添加可并行执行的任务任务 ID 形如parallel_0_counter添加后_last_task_id置空要求后续任务显式声明依赖add_sync_task(content, wait_forNone, ...)添加同步任务自动等待最近一批并行任务wait_for缺省且无并行任务时抛出ValueErrorbuild()返回任务列表前会用 DFS 检测循环依赖发现环时抛出ValueErrorfork()/join()分别是add_parallel_tasks与add_sync_task的语义化别名get_task_info()返回{task_count: ..., tasks: [{id, content, dependencies}]}形式的流水线概览。内部通过_task_registrytask_id → Task 映射实现快速查重与依赖校验并把依赖解析为真正的Task对象列表传入Task构造函数源码中from camel.tasks import Task。模块全貌utils.py 中的其他模型除上述核心模型外utils.py 还定义了以下辅助模型API 文档未单独列出属于源码级补充GENERIC_ROLE_NAMES与is_generic_role_name判断角色名是否为assistant、agent、user、system、worker、helper等通用名大小写不敏感。通用角色名无法体现智能体的实际用途触发回退逻辑——从 LLM 生成的agent_title或description中寻找更具体的标识同时避免用通用名作为工作流文件夹名WorkflowMetadata工作流元数据session_id、working_directory、created_at、updated_at、workflow_version、agent_id、message_count用于工作流历史与版本管理WorkflowConfig工作流内存管理配置max_workflows_per_role100、文件名后缀_workflow、存储目录workforce_workflows、enable_versioningTrue、default_max_files_to_load3集中管理配置项避免散落各处QualityEvaluation质量评估结果quality_sufficient、quality_score、issues、recovery_strategy、modified_task_content其 docstring 明确标注deprecated建议改用TaskAnalysisResult保留仅为向后兼容。小结camel.societies.workforce.utils模块是整个 Workforce 框架的数据契约层WorkerConf驱动协调器创建 Worker 的结构化输出TaskResult/TaskAssignment/TaskAssignResult承载任务结果与分配信息validate_dependencies保证了 LLM 输出的容错性RecoveryStrategy/FailureContext/TaskAnalysisResult构建了任务失败的恢复决策体系而check_if_running则通过重试与异常处理的容错设计守护着 Workforce 的生命周期并发安全。阅读本文后你可以顺着 camel/societies/workforce/utils.py 继续深入 workforce.py 的 worker 创建与失败恢复流程或通过 test/workforce/test_workforce.py 验证上述模型的实际行为。【免费下载链接】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），仅供参考