十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

使用 ZenML 编排 CrewAI 多智能体 Crew:从本地快速运行到 HTTP 实时部署

使用 ZenML 编排 CrewAI 多智能体 Crew:从本地快速运行到 HTTP 实时部署 使用 ZenML 编排 CrewAI 多智能体 Crew从本地快速运行到 HTTP 实时部署【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenmlCrewAI 提供了一套基于角色分工Role-Based的多智能体协作框架而 ZenML 则负责将这套框架包装成可追踪、可版本化、可部署的管道。本文基于当前仓库中的 examples/agent_framework_integrations/crewai 示例讲解如何在 ZenML 管道中驱动 CrewAI 的 Agent / Task / Crew 三层结构完成从环境准备、本地运行到zenml pipeline deploy实时 HTTP 服务的完整链路帮助你在生产环境中把批量式智能体工作流升级为可被 Web 应用、微服务直接调用的在线 API。一、示例概览CrewAI 与 ZenML 的分工本示例的核心文件为 examples/agent_framework_integrations/crewai/run.py它实现了一个旅行助手Travel Assistant场景CrewAI 侧负责定义两个各司其职的智能体天气专员Weather Specialist与旅行顾问Travel Advisor、一个被智能体调用的天气查询工具以及由两个 Task 组成的协作链路。ZenML 侧通过step将 Crew 执行封装为一个 steprun_crewai_agents再通过pipeline把智能体执行与结果格式化format_travel_results串联成一条可复现、可部署的管道agent_pipeline。在 examples/agent_framework_integrations/README.md 的框架总览表中CrewAI 被归类为 Crews 类型关键技术特征为 Agent crews, Task delegation配套技术栈为crewai与openai。管道架构agent_pipeline(city: str Berlin) ├── run_crewai_agents(city) # step调用 crew.kickoff(inputs{city: city}) │ ├── Agent 1: Weather Specialisttools[get_weather] │ ├── Agent 2: Travel Advisor │ ├── Task 1: check_weather_task │ └── Task 2: packing_advice_taskcontext[check_weather_task] └── format_travel_results(crew_data) # step将结果整理为可读摘要二、快速开始环境准备与本地运行示例官方给出的 Quick Run 流程如下export OPENAI_API_KEYyour-api-key-here uv venv --python 3.11 source .venv/bin/activate uv pip install -r requirements.txt随后初始化 ZenML 并登录zenml init zenml login最后运行管道python run.py2.1 依赖说明examples/agent_framework_integrations/crewai/requirements.txt 中的依赖清单是理解版本约束的关键crewai1.2.0,1.6.1 zenml[local] jinja23.1.0 secure0.3.0文件顶部注释直接说明了版本锁定的原因CrewAI 的1.7.0版本将 opentelemetry-sdk 约束收窄到~1.34.0与 ZenML 使用的1.40.0冲突因此必须1.6.1。这说明把第三方 Agent 框架接入 ZenML 时依赖版本对齐是首要检查项。运行python run.py时程序入口会打印启动信息并执行agent_pipeline()默认参数cityBerlin完成后提示去 ZenML Dashboard 查看运行结果与产物。三、源码拆解CrewAI 三层结构如何在 ZenML 中落地下面按run.py的源码顺序拆解关键实现对应 run.py。3.1 自定义工具tool装饰器from crewai.tools import tool tool(Weather Checker Tool) def get_weather(city: str) - str: Get weather for a given city. return fCurrent weather in {city}: Sunny, 22°C (72°F), light breeze该工具接受城市名并返回模拟天气数据在示例中用于演示 CrewAI 的 Tool Calling 能力在生产中可替换为真实天气 API 或任何业务数据源。tools[get_weather]将工具注入对应 Agent。3.2 Agent 定义角色、目标与回溯weather_checker Agent( roleWeather Specialist, goalCheck weather conditions for a given city: {city}, backstoryYou are a meteorology expert who provides accurate weather updates. ..., tools[get_weather], verboseTrue, max_iter5, ) travel_advisor Agent( roleTravel Advisor, goalGive practical travel advice based on weather conditions for {city}, backstoryYou are an experienced travel advisor who helps people prepare for their trips., verboseTrue, max_iter5, )role/goal/backstory共同定义了智能体的人格与职责边界goal与backstory中的{city}占位符会在crew.kickoff(inputs...)时被实际参数填充max_iter5限制单智能体最大迭代次数防止失控循环verboseTrue便于在日志中观察智能体推理过程。3.3 Task 定义任务委托与上下文依赖check_weather_task Task( descriptionCheck the current weather in {city} and provide a weather report., expected_outputA clear statement of the current weather conditions in {city} including temperature and conditions, agentweather_checker, ) packing_advice_task Task( descriptionBased on the weather report for {city}, provide 3-5 specific items to pack for the trip., expected_outputA list of 3-5 specific items to pack based on the weather conditions, agenttravel_advisor, context[check_weather_task], )关键点在于packing_advice_task通过context[check_weather_task]声明对前序任务的依赖这正是 CrewAI Task Delegation 能力的体现第二个 Task 的输入来自第一个 Task 的输出形成接力式协作。3.4 Crew 组装PanAgent 的发现入口crew Crew( agents[weather_checker, travel_advisor], tasks[check_weather_task, packing_advice_task], verboseTrue, )源码注释明确指出PanAgent will discover this crew variable——即该模块级变量crew是后续 PanAgentZenML 的 Agent 层发现并调用的统一入口。这意味着命名约定crew具有框架层面的语义不应随意改名。3.5 ZenML Step把 Crew 执行包装成可追踪单元step def run_crewai_agents(city: str) - Annotated[Dict[str, Any], crew_results]: Execute the CrewAI crew and return results. try: result crew.kickoff(inputs{city: city}) return {city: city, result: str(result), status: success} except Exception as e: return { city: city, result: fCrew execution error: {str(e)}, status: error, }Annotated[Dict[str, Any], crew_results]为输出产物命名ZenML 会据此在 artifact store 中登记、版本化该产物采用 try/except 包裹整个 Crew 执行并将执行状态显式记录为success/error遵循了 examples/agent_framework_integrations/README.md 中总结的通用错误处理模式状态跟踪 优雅降级。3.6 ZenML Step结果格式化step def format_travel_results(crew_data: Dict[str, Any]) - Annotated[str, formatted_results]: ...该 step 读取上一步返回的 dict按状态生成成功旅行规划或错误报告两种格式的摘要文本并将格式化结果作为新产物formatted_results输出。两段式 step 结构执行 格式化也是整个 agent 示例集合的通用架构模式。3.7 管道定义与 Docker 设置pipeline(settings{docker: docker_settings}, enable_cacheFalse) def agent_pipeline(city: str Berlin) - str: crew_results run_crewai_agents(citycity) summary format_travel_results(crew_results) return summary这里有两个值得注意的细节enable_cacheFalse智能体执行具有随机性与外部依赖LLM 调用关闭缓存可确保每次运行都真实执行 Crew避免拿到过期缓存结果settings{docker: docker_settings}管道内置了容器化运行配置为后续远程栈运行与实时部署铺路。docker_settings定义如下对应 run.pydocker_settings DockerSettings( python_package_installerPythonPackageInstaller.UV, requirementsrequirements.txt, # relative to the pipeline directory environment{ OPENAI_API_KEY: os.getenv(OPENAI_API_KEY), # Set home directory to a writable location for CrewAI storage HOME: /tmp, XDG_DATA_HOME: /tmp/.local/share, }, )逐项解释python_package_installerPythonPackageInstaller.UV使用 UV 作为容器内 Python 包安装器加快依赖解析与安装requirementsrequirements.txt相对管道目录的依赖文件保证容器内复现与本地一致的环境含前述版本锁定environment中的OPENAI_API_KEY从宿主机环境变量透传进容器HOME/tmp与XDG_DATA_HOME/tmp/.local/share是为 CrewAI 的 appdirs 存储行为预留可写目录避免容器内默认 HOME 只读导致的写入失败——这是把 CrewAI 容器化的一个关键坑位。对应 ZenML 侧DockerSettings在 src/zenml/config/docker_settings.py 中定义其字段注释明确说明python_package_installer指定使用的包安装器、requirements接受相对路径或依赖列表、environment用于注入容器内环境变量。四、实时部署把管道变成 HTTP 服务示例 README 的 Pipeline Deployment 一节展示了把该智能体管道发布为实时 HTTP 服务的完整流程。4.1 部署管道zenml pipeline deploy run.agent_pipeline --name crewai-agentrun.agent_pipeline是 Python 模块与管道对象的导入路径source即run.py模块中定义的agent_pipeline--name crewai-agent为部署命名。该命令在 ZenML 源码中的实现位于 src/zenml/cli/pipeline.py 的deploy_pipeline函数。从源码可见其关键行为若当前目录没有 ZenML 仓库会给出警告并以当前工作目录作为源码解析根目录提示应先执行zenml init通过_import_pipeline(sourcesource)动态导入管道实例支持--config-path、--stack-name-or-id、--build-path-or-id、--update、--overtake、--attach、--timeout等选项若同名部署已存在会交互式确认是否覆盖可用--update跳过确认--overtake用于接管他人部署部署成功后打印部署信息并给出可在 ZenML UI 中查看的 Dashboard 链接。4.2 通过 CLI 调用zenml deployment invoke crewai-agent --topicAI in Healthcare注意README 中 CLI 示例使用的参数名是--topic而run.py中管道签名是city。实际参数名应与管道函数签名一致即本例应为--city以zenml deployment describe --schema name查看部署的输入 schema 为准。invoke_deployment在 src/zenml/cli/deployment.py 中实现其行为包括默认 5 分钟超时timeout or 300通过--no-wait可提交后台运行并立即返回 run ID参数非法时提示zenml deployment describe --schema查看 schema调用失败时提示zenml deployment logs查看日志。4.3 通过 HTTP API 调用curl -X POST http://localhost:8000/invoke \ -H Content-Type: application/json \ -d {parameters: {topic: The Future of Quantum Computing}}服务默认监听localhost:8000请求体为{parameters: {...}}结构其中字段名同样要与管道输入参数对应。这正是 examples/agent_framework_integrations/README.md 所强调的Pipeline Deployments把批量工作流转成在线服务模式。4.4 配套运维命令部署后常用运维命令包括均来自zenml deployment命令组zenml deployment list列出全部部署zenml deployment describe name查看部署详情--schema查看输入参数 schemazenml deployment logs name查看部署日志支持--follow/-f与--tailzenml deployment invoke name调用部署。五、该模式的价值与适用场景结合 examples/agent_framework_integrations/README.md 的说明这种框架智能体 ZenML 管道的组合在生产中的价值体现在批量执行可在夜间批量处理成百上千条查询例如对文档集合逐篇执行 Agent 分析实时服务将 Agent 以 HTTP API 形式部署供 Web 应用、微服务直接集成可观测与产物管理每次 Crew 执行都成为一条带元数据的管道运行输入输出作为版本化产物存入 artifact store可复现性DockerSettingsrequirements.txt保证了本地、远程与容器环境的一致性评估与 A/B 测试可以用同一套管道骨架在不同 Agent 配置间系统化对比。从源码结构看该示例与 examples/agent_framework_integrations 下的 LangChain、LangGraph、Autogen、PydanticAI 等示例遵循一致的骨架tool类装饰器定义工具、Agent/Crew 对象封装智能体、step包装执行、pipeline串联编排、Annotated命名产物、try/except 状态化错误处理。因此本文的通用模式可直接迁移到其他 Agent 框架的集成中。六、注意事项与限制版本兼容CrewAI 必须1.6.1否则 opentelemetry 依赖与 ZenML 冲突见 requirements.txt。API Key本地运行时通过OPENAI_API_KEY环境变量提供部署到远程环境时需确保密钥以安全方式注入容器环境变量或 ZenML Secret。容器 HOME 目录将 CrewAI 容器化时务必通过HOME/XDG_DATA_HOME指定可写路径否则智能体状态存储可能失败。缓存LLM 驱动的 Agent 执行具有不确定性建议保持enable_cacheFalse以保证运行的真实性。部署环境zenml pipeline deploy需要先完成zenml init与zenml login连接 ZenML Server并在所需 stack 上具备可用的部署器deployer组件。参数名一致性CLI / HTTP 调用传入的参数名必须与管道函数签名一致不确定时用zenml deployment describe --schema核对。七、继续探索完整示例代码examples/agent_framework_integrations/crewai/run.py依赖与版本锁定examples/agent_framework_integrations/crewai/requirements.txtAgent 框架集成集合总览examples/agent_framework_integrations/README.md部署命令实现src/zenml/cli/pipeline.py调用与运维命令实现src/zenml/cli/deployment.pyDocker 运行配置定义src/zenml/config/docker_settings.py【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
返回列表