CAMEL 框架 SingleAgentWorker 与 AgentPool 深度解析：单代理工作节点的资源池化与结构化输出机制

CAMEL 框架 SingleAgentWorker 与 AgentPool 深度解析单代理工作节点的资源池化与结构化输出机制【免费下载链接】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导读SingleAgentWorker是 CAMEL 多智能体协作引擎 Workforce 中最核心、最常用的工作节点Worker类型它以单个ChatAgent为执行单元承接 Workforce 分解与分配下来的子任务。本文基于仓库 API 参考文档 docs/reference/camel.societies.workforce.single_agent_worker.md结合 single_agent_worker.py 源码系统讲解SingleAgentWorker与配套的AgentPool代理复用池的构造参数、内部机制、任务处理流程、结构化输出双通道以及工作流记忆存取。读完本文你将掌握如何在 Workforce 中装配高效、可弹性伸缩的单代理工作节点并理解其底层实现原理。一、定位Workforce 体系中的工作节点在 CAMEL 的 Workforce 多智能体体系中Worker见 worker.py是所有任务执行节点的抽象基类描述为workforce 系统中任务处理的基本单元。它通过TaskChannel获取分配给自己的任务、执行_process_task抽象方法、并把结果状态写回任务对象。SingleAgentWorker(Worker)是Worker最直接的实现一个节点 一个智能体。与成对智能体协作的RolePlayingWorker不同它内部只有单个ChatAgent适合专职单干的角色分工场景如研究者、分析师、写作者各司其职。从源码结构看SingleAgentWorker相比基类额外引入了两大能力AgentPool 代理复用池以克隆 回收的方式复用ChatAgent实例避免每个子任务都从头构建代理双通道结构化输出默认使用StructuredOutputHandler提示词引导 正则提取也可切换为模型原生response_format参数。二、AgentPool代理实例的弹性复用池2.1 构造参数AgentPool源码 single_agent_worker.py是用于高效复用的代理实例池管理一批预克隆的ChatAgent支持按需自动扩容auto-scaling与空闲回收。API 文档给出的构造参数如下参数类型默认值说明base_agentChatAgent必填池中代理的克隆模板base agent to clone frominitial_sizeint1池的初始代理数量max_sizeint10池中代理数量上限auto_scaleboolTrue是否自动扩容池大小idle_timeoutfloat180.0空闲代理被回收的秒数阈值源码中还额外暴露了一个未写入 API 文档的参数cleanup_interval默认60.0秒它是周期性清理检查的固定间隔见 single_agent_worker.py由内部定时任务在每次清理循环中await asyncio.sleep(cleanup_interval)后执行一次cleanup_idle_agents()L598-L611。2.2 初始化与克隆机制_initialize_pool(size)按min(size, max_size)的数量上限创建初始代理加入可用队列_available_agents并记录每个代理的最后使用时间L105-L110。_create_fresh_agent()通过self.base_agent.clone(with_memoryFalse)克隆出一个无记忆的全新代理实例并累加_total_clones_created计数L112-L116。不带记忆克隆保证了同一任务执行前后的内存状态隔离是池化复用的关键前提。2.3 借用、归还与空闲清理池的核心并发控制依赖asyncio.Lock与asyncio.Conditionget_agent()L118-L139优先从_available_agents队首取出代理并调用agent.reset()清理记忆状态若池空且auto_scaleTrue或未达max_size则新克隆一个否则在条件变量上等待其他协程归还代理。return_agent(agent)L141-L160任务结束后归还代理若可用队列未达max_size则重置并回队、唤醒一个等待协程超过上限的代理直接弃用释放内存。cleanup_idle_agents()L162-L183仅在auto_scaleTrue时生效扫描可用队列将当前时间 - 最后使用时间 idle_timeout的代理移除并计入_agents_cleaned。2.4 get_stats 统计指标get_stats()返回的字典L185-L197可用于观测池的运行状态与命中效率指标含义available_agents当前可用空闲代理数in_use_agents当前借用中的代理数pool_size池总规模可用 借用中total_borrows累计借用次数total_clones_created累计克隆创建的代理数pool_hits命中池复用已有代理的次数hit_rate命中率 pool_hits / max(total_borrows, 1)agents_cleaned_up因空闲超时被清理的代理数其中hit_rate是评估池化收益的核心指标命中率越高说明复用越充分克隆开销越低。三、SingleAgentWorker单代理工作节点3.1 构造参数API 文档列出的参数与默认值如下参数类型默认值说明descriptionstr必填节点描述workerChatAgent必填节点的工作代理单个智能体use_agent_poolboolTrue是否启用代理池以提升效率pool_initial_sizeint1代理池初始大小pool_max_sizeint10代理池最大大小auto_scale_poolboolTrue是否自动扩容代理池use_structured_output_handlerboolTrue是否使用结构化输出处理器见 3.3 节此外源码还支持两个未出现在 API 文档中的可选参数L222-L244context_utilityContextUtility默认None共享的上下文工具实例传入后所有工作流操作复用同一实例保证多个 worker 共享同一个会话目录enable_workflow_memorybool默认False是否在任务执行期间累积工作流记忆对话记录若计划调用save_workflow_memories()则需设为True。构造时node_id直接取自worker.agent_idL246并在use_agent_poolTrue时立即以传入的worker为模板构建AgentPoolL278-L284。3.2 任务处理主流程 _process_task_process_taskL349-L583是异步任务执行的核心流程如下取代理_get_worker_agent()从池中借用代理池关闭时退化为self.worker.clone(with_memoryFalse)的旧克隆方式L302-L308。组装提示词将PROCESS_TASK_PROMPT模板定义于 prompts.py格式化填充任务内容task.content、父任务内容与依赖任务信息dependency_tasks_info由基类的_get_dep_tasks_info生成见 worker.py。结构化输出根据use_structured_output_handler走两条路径见 3.3 节。执行与流式处理调用worker_agent.astep()流式响应时通过consume_response_content_async消费分块并转发stream_callback。Token 统计从响应info中读取usage/token_usage取total_tokens写入任务additional_infoL460-L474。工作流记忆收集仅enable_workflow_memoryTrue把工作代理记忆中的全部记录转存到对话累加器_conversation_accumulator由原始 worker 无记忆克隆而来供后续工作流保存使用L479-L501。执行记录把每次尝试写入task.additional_info[worker_attempts]包含agent_id、original_worker_id、时间戳、响应摘要、工具调用摘要与 token 用量L514-L544便于审计与失败排查。结果判定任务结果解析失败、task_result.failedTrue或is_task_result_insufficient(task)内容校验不足见 task.py时返回TaskState.FAILED否则返回TaskState.DONE。归还代理finally中通过_return_worker_agent将代理归还池中L510-L512。3.3 结构化输出的双通道这是SingleAgentWorker最重要的设计之一对应参数use_structured_output_handler默认True通道一StructuredOutputHandler默认开启StructuredOutputHandler实现于 structured_output_handler.py通过提示词引导 正则提取来获得结构化结果generate_structured_prompt()把 Pydantic schema此处为TaskResult定义于 utils.py的字段、类型、必填标记、JSON 示例与附加指令拼接到基础提示词后要求模型只返回 JSON 代码块L74-L155parse_structured_response()先用多级正则模式json代码块、裸 JSON、嵌套对象模式提取 JSON再经 Pydantic 校验校验失败时尝试_fix_common_issues修复仍失败则回退到fallback_values如{content: Task processing failed, failed: True}或 schema 默认实例L280-L342。该通道的兼容性收益显著即使底层模型不可靠地支持原生结构化输出也能通过提示与解析拿到符合TaskResultschema 的结果。通道二原生 response_format关闭时当use_structured_output_handlerFalse时直接调用worker_agent.astep(prompt, response_formatTaskResult)依赖模型侧的原生response_format能力L430-L434。流式场景下逐块从chunk.msg.parsed收集结果若最终未解析出结果则构造TaskResult(contentFailed to parse streaming response, failedTrue)兜底L437-L452。小结默认True优先保证兼容性如果你的模型原生结构化输出稳定可靠可关闭以获得更直接的解析路径。3.4 生命周期reset、通道监听与周期清理reset()L286-L300调用基类与self.worker.reset()重置状态并取消清理任务、以默认参数重建AgentPool使 worker 恢复到初始状态。_listen_to_channel()L585-L596启动监听时若启用了代理池则创建_periodic_cleanup后台任务监听结束await super()._listen_to_channel()返回时取消该任务。get_pool_stats()L613-L617代理池启用时返回AgentPool.get_stats()的统计字典否则返回None。Worker 基类的监听循环worker.py采用并发无人工上限策略不断从通道领取任务并asyncio.create_task派发执行支持并行任务处理stop()则会立即取消所有运行中任务并退出循环。四、工作流记忆持久化与智能复用对于周期性、重复性任务SingleAgentWorker支持把执行过的任务对话沉淀为可复用的工作流记忆save_workflow_memories()L619-L670同步版本已标记为 deprecated自 0.2.80 起建议改用异步版。通过WorkflowMemoryManager基于对话累加器的历史生成工作流摘要保存为 Markdown 文件成功后清理累加器返回status、summary、file_path、worker_description字段。save_workflow_memories_async()L672-L705异步版本内部使用asummarize()做非阻塞 LLM 调用支持多个 worker 并行保存。load_workflow_memories(patternNone, max_workflows3, session_idNone, use_smart_selectionTrue)L707-L747默认启用基于 Agent 的智能选择利用 worker 代理依据元数据标题、描述、标签挑选最相关的工作流use_smart_selectionFalse时退化为文件名pattern匹配。使用前提构造SingleAgentWorker时须将enable_workflow_memoryTrue或通过add_single_agent_worker的enable_workflow_memory参数开启否则不会累积对话记录。五、在 Workforce 中装配与实战5.1 通过 Workforce 添加单代理工作节点通常你不会直接实例化SingleAgentWorker而是通过Workforce.add_single_agent_worker()注册workforce.pydef add_single_agent_worker( self, description: str, worker: ChatAgent, pool_max_size: int DEFAULT_WORKER_POOL_SIZE, # 10 enable_workflow_memory: bool False, ) - Workforce:该方法把ChatAgent包装为SingleAgentWorkeruse_agent_pool、pool_initial_size、auto_scale_pool、use_structured_output_handler使用默认值并支持链式调用连续添加多个 worker也可在 Workforce 处于PAUSED状态时动态添加。5.2 完整示例三人研究团队参考官方示例 examples/workforce/multiple_single_agents.py组合研究者 分析师 写作者三个单代理工作节点from camel.agents.chat_agent import ChatAgent from camel.messages.base import BaseMessage from camel.models import ModelFactory from camel.societies.workforce import Workforce from camel.tasks.task import Task from camel.toolkits import SearchToolkit from camel.types import ModelPlatformType, ModelType # 1. 研究代理带网页检索工具 search_agent ChatAgent( system_messageBaseMessage.make_assistant_message( role_nameResearch Specialist, contentYou are a research specialist who excels at finding and gathering information from the web., ), modelModelFactory.create( model_platformModelPlatformType.DEFAULT, model_typeModelType.DEFAULT, ), tools[SearchToolkit().search_wiki], ) # 2. 分析代理 analyst_agent ChatAgent( system_messageBaseMessage.make_assistant_message( role_nameBusiness Analyst, contentYou are an expert business analyst. Your job is to analyze research findings, identify key insights, opportunities, and challenges., ), modelModelFactory.create( model_platformModelPlatformType.DEFAULT, model_typeModelType.DEFAULT, ), ) # 3. 写作代理 writer_agent ChatAgent( system_messageBaseMessage.make_assistant_message( role_nameReport Writer, contentYou are a professional report writer. You take analytical insights and synthesize them into a clear, concise, and well-structured final report., ), modelModelFactory.create( model_platformModelPlatformType.DEFAULT, model_typeModelType.DEFAULT, ), ) workforce Workforce(Business Analysis Team, graceful_shutdown_timeout30.0) workforce.add_single_agent_worker( A researcher who can search online for information., workersearch_agent, ).add_single_agent_worker( An analyst who can process research findings., workeranalyst_agent, ).add_single_agent_worker( A writer who can create a final report from the analysis., workerwriter_agent, ) # 4. 提交任务并执行 task Task(contentConduct a comprehensive market analysis for launching a new electric scooter in Berlin...) result workforce.process_task(task)任务执行过程中Workforce 的task_agent负责分解、coordinator_agent负责把子任务分派给最合适的 worker各SingleAgentWorker并行领取并处理任务结果可作为后续任务的依赖详见 docs/key_modules/workforce.md 中的任务生命周期图。5.3 调优建议高频小任务pool_initial_size可适当调大如 2~3减少并发高峰期的克隆开销内存敏感调小pool_max_size、idle_timeout如 60 秒配合cleanup_interval加快空闲代理回收多轮并发代理池基于asyncio并发模型在异步工作负载下收益最明显兼容性优先保持use_structured_output_handlerTrue若目标模型原生response_format稳定可关闭以简化解析链路。六、结语SingleAgentWorker与AgentPool是 CAMEL Workforce 体系中最基础的积木前者把单个ChatAgent包装成可参与任务分解、分配、并行执行与失败恢复的标准化节点后者通过无记忆克隆 借用归还 空闲回收 自动扩容把代理实例的开销降到最低而默认开启的StructuredOutputHandler则为不原生支持结构化输出的模型提供了可靠的结果解析兜底。需要进一步深入时可继续阅读 Workforce 主类文档、Worker 基类 以及 workforce 模块目录 下的其余实现与 examples/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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考