ZenML 集成 AWS Strands Agent 實(shí)戰(zhàn)從數(shù)學(xué)計(jì)算到實(shí)時(shí) HTTP 部署【免費(fèi)下載鏈接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.項(xiàng)目地址: https://gitcode.com/GitHub_Trending/ze/zenmlAWS Strands 是 Amazon 推出的輕量級(jí) Agent 框架以簡(jiǎn)潔的tool裝飾器、可直接調(diào)用的 Agent 接口著稱。本指南基于 ZenML 官方倉(cāng)庫(kù)中的 aws_strands 示例完整演示如何將 Strands Agent 封裝進(jìn) ZenML 流水線實(shí)現(xiàn)數(shù)學(xué)計(jì)算與天氣查詢等任務(wù)并借助 ZenML 的 Pipeline Deployment 功能將其一鍵部署為實(shí)時(shí) HTTP 服務(wù)。讀完本文你將掌握 Strands 工具定義、ZenML 步驟封裝、Docker 環(huán)境注入 API Key以及zenml pipeline deploy/zenml deployment invoke的完整部署與調(diào)用鏈路。一、示例概覽AWS Strands 與 ZenML 的集成價(jià)值在 ZenML 的 Agent 框架集成示例集中AWS Strands 被定位為「Simple」類型的框架——它的核心賣點(diǎn)是直接可調(diào)用的 Agent 接口與內(nèi)建工具。正如 examples/agent_framework_integrations/README.md 中的框架對(duì)比表所描述Strands 在 ZenML 生態(tài)中代表的是最簡(jiǎn)集成路徑無需多 Agent 協(xié)商、無需圖式工作流一個(gè)agent(query)調(diào)用即可完成推理。整個(gè)示例只有三個(gè)核心文件職責(zé)劃分非常清晰文件職責(zé)agent.py定義 Strands Agent、工具與模型配置run.py將 Agent 封裝進(jìn) ZenML 流水線配置 Docker 環(huán)境requirements.txt聲明運(yùn)行時(shí)依賴strands-agents、zenml 等這種「框架代碼與編排代碼分離」的結(jié)構(gòu)正是 ZenML 集成任何第三方 Agent 框架的通用模式框架負(fù)責(zé)智能ZenML 負(fù)責(zé)編排、追蹤與部署。二、環(huán)境準(zhǔn)備從零開始運(yùn)行 Strands 流水線2.1 依賴安裝示例要求 Python 3.11 與 uv 包管理器。在 requirements.txt 中聲明了以下核心依賴strands-agents[openai]1.0.0 strands-agents-tools0.0.1 zenml[local] jinja23.1.0 secure0.3.0安裝步驟與 ZenML 官方推薦的 Agent 框架快速啟動(dòng)流程一致參見 examples/agent_framework_integrations/README.mdexport OPENAI_API_KEYyour-api-key-here uv venv --python 3.11 source .venv/bin/activate uv pip install -r requirements.txt說明strands-agents[openai]使用 OpenAI 兼容接口因此需要提前申請(qǐng) OpenAI API Key 并通過環(huán)境變量注入密鑰不應(yīng)硬編碼在代碼中詳見下文模型配置部分。2.2 初始化 ZenML 并登錄zenml init zenml loginzenml init在當(dāng)前目錄初始化 ZenML 倉(cāng)庫(kù)建立本地存儲(chǔ)與配置。zenml login連接到 ZenML 服務(wù)端本地或遠(yuǎn)程之后所有流水線運(yùn)行記錄、產(chǎn)物artifacts都會(huì)寫入服務(wù)端可在 ZenML Dashboard 中查看。2.3 運(yùn)行流水線python run.py從 run.py 的入口邏輯可以看到腳本執(zhí)行后會(huì)打印運(yùn)行提示并調(diào)用agent_pipeline()流水線完成后會(huì)提示在 ZenML Dashboard 中查看詳細(xì)結(jié)果與產(chǎn)物。三、Agent 定義Strands 的工具與模型配置agent.py 是整個(gè)示例的智能核心它演示了 Strands 框架的兩個(gè)關(guān)鍵概念tool裝飾器與可調(diào)用的 Agent 對(duì)象。3.1 使用tool定義工具from strands import Agent, tool tool def get_weather(city: str) - str: Very naive weather lookup that *always* returns a sunny forecast. Args: city: Name of the city. Returns: A short weather blurb. return fIt’s always sunny in {city}!這是一個(gè)「樸素」的天氣查詢工具——為了演示目的它總是返回晴天預(yù)報(bào)。關(guān)鍵在于它展示了 Strands 的工具定義模式任意 Python 函數(shù)加上tool裝飾器即可成為 Agent 可調(diào)用的函數(shù)參數(shù)與返回值類型注解會(huì)被框架自動(dòng)解析并注入到 LLM 的 function calling 流程中。3.2 配置 OpenAI 模型import os from strands.models.openai import OpenAIModel model OpenAIModel( client_args{ api_key: os.getenv(OPENAI_API_KEY, ), }, model_idgpt-5-nano, params{temperature: 0.7}, )兩點(diǎn)值得注意密鑰從環(huán)境變量讀取os.getenv(OPENAI_API_KEY, )避免了在源碼中硬編碼憑據(jù)這與 ZenML 的安全實(shí)踐一致——敏感信息應(yīng)當(dāng)通過環(huán)境變量或 ZenML Secret 管理本示例采用環(huán)境變量方式見下文 Docker 配置中的environment注入。模型參數(shù)可調(diào)temperature: 0.7控制生成隨機(jī)性model_id可替換為其他 OpenAI 兼容模型。3.3 組裝全局 Agentagent: Agent Agent( modelmodel, tools[get_weather], system_promptYou are a helpful assistant that can check weather for cities., )示例以模塊級(jí)全局變量的形式導(dǎo)出agent對(duì)象并附帶一條關(guān)鍵注釋agent.pyPanAgent discovers theagentvariable at runtime; donotexecute the agent from within this module.這意味著不要在 agent.py 模塊內(nèi)部直接執(zhí)行agent()否則框架在運(yùn)行時(shí)動(dòng)態(tài)發(fā)現(xiàn)agent變量時(shí)會(huì)與已執(zhí)行的調(diào)用沖突。正確的做法是像 run.py 那樣從外部導(dǎo)入agent后再調(diào)用。四、流水線封裝把 Agent 編排成 ZenML Pipelinerun.py 演示了將 Strands Agent 封裝進(jìn) ZenML 的標(biāo)準(zhǔn)三步模式步驟step執(zhí)行 Agent → 步驟格式化結(jié)果 → 流水線pipeline串聯(lián)。4.1 配置 Docker 構(gòu)建環(huán)境from zenml import pipeline, step from zenml.config import DockerSettings, PythonPackageInstaller docker_settings DockerSettings( python_package_installerPythonPackageInstaller.UV, requirementsrequirements.txt, # relative to the pipeline directory environment{ OPENAI_API_KEY: os.getenv(OPENAI_API_KEY), }, )這里的配置非常關(guān)鍵python_package_installerPythonPackageInstaller.UV指定使用 uv 作為容器內(nèi) Python 包安裝器與本地環(huán)境的 uv 工作流保持一致可顯著加快依賴安裝速度。requirementsrequirements.txt聲明容器構(gòu)建時(shí)安裝的依賴文件路徑相對(duì)于流水線目錄確保容器內(nèi)strands-agents、zenml等包可用。environment{OPENAI_API_KEY: ...}將宿主環(huán)境變量注入容器運(yùn)行時(shí)使 Agent 在容器內(nèi)也能讀取到 API Key避免密鑰硬編碼。4.2 定義步驟第一步執(zhí)行 Agent 并捕獲異常step def run_strands_agent(query: str) - Annotated[str, agent_response]: Execute the Strands agent and return the response. try: response agent(query) return str(response) except Exception as e: return fAgent error: {str(e)}輸出使用Annotated[str, agent_response]命名產(chǎn)物ZenML 會(huì)將該輸出注冊(cè)為名為agent_response的 artifact供 Dashboard 追蹤與后續(xù)步驟消費(fèi)。通過 try/except 兜底Agent 失敗時(shí)返回錯(cuò)誤字符串而非中斷流水線——這正是示例集中強(qiáng)調(diào)的「優(yōu)雅降級(jí)」錯(cuò)誤處理模式見 examples/agent_framework_integrations/README.md。第二步格式化響應(yīng)step def format_weather_response( response: str, ) - Annotated[str, formatted_response]: Format the agent response into a readable summary. formatted f? WEATHER REPORT { * 30} {response} Powered by AWS Strands Agent return formatted.strip()此步驟將 Agent 的原始響應(yīng)包裝成帶標(biāo)題的分隔格式輸出產(chǎn)物命名為formatted_response。4.3 定義流水線pipeline(settings{docker: docker_settings}, enable_cacheFalse) def agent_pipeline(query: str Whats the weather like in Tokyo?) - str: ZenML pipeline that orchestrates the AWS Strands weather agent. Returns: Formatted weather response # Run the agent response run_strands_agent(queryquery) # Format the results summary format_weather_response(response) return summary兩個(gè)值得深挖的配置settings{docker: docker_settings}將前面定義的 Docker 構(gòu)建配置掛載到流水線實(shí)際運(yùn)行時(shí)每個(gè)步驟都會(huì)基于該配置構(gòu)建隔離容器環(huán)境。enable_cacheFalse禁用步驟緩存。Agent 調(diào)用具有非確定性LLM 輸出每次都可能不同關(guān)閉緩存可確保每次運(yùn)行都真實(shí)執(zhí)行推理而非復(fù)用舊產(chǎn)物。默認(rèn)查詢參數(shù)為「Tokyo 的天氣」。流水線返回str意味著整條流水線的最終產(chǎn)物可直接作為函數(shù)返回值傳遞這也是后續(xù)zenml pipeline deploy將其暴露為可調(diào)用 HTTP 接口的基礎(chǔ)。五、實(shí)時(shí)部署從批處理流水線到 HTTP 服務(wù)ZenML 的 Pipeline Deployment 特性可以將這類「批處理風(fēng)格」的 Agent 流水線直接轉(zhuǎn)換為持久的實(shí)時(shí) HTTP 服務(wù)。示例中的部署命令aws_strands/README.md與整套 Agent 集成示例的通用部署模式examples/agent_framework_integrations/README.md完全一致。5.1 部署流水線# Deploy the pipeline as an HTTP service zenml pipeline deploy run.agent_pipeline --name aws-strands-agent該命令對(duì)應(yīng)源碼中的 deploy_pipeline 函數(shù)其核心參數(shù)包括參數(shù)作用source位置參數(shù)可導(dǎo)入的流水線實(shí)例來源此處為run.agent_pipeline--name部署名稱如aws-strands-agent--config-path可選流水線配置文件路徑--stack-name-or-id可選指定部署目標(biāo) stack--timeout等待部署完成的最大秒數(shù)--update/--overtake同名部署已存在時(shí)的更新策略從源碼實(shí)現(xiàn)看zenml pipeline deploy會(huì)先導(dǎo)入流水線實(shí)例再基于當(dāng)前活躍 stack 構(gòu)建鏡像并啟動(dòng)常駐服務(wù)若在非 ZenML 倉(cāng)庫(kù)目錄下運(yùn)行CLI 會(huì)提示當(dāng)前目錄被作為源碼根目錄用于解析步驟類詳見 pipeline.py 的警告邏輯。5.2 通過 CLI 調(diào)用zenml deployment invoke aws-strands-agent --queryCalculate the square root of 144zenml deployment invoke的實(shí)現(xiàn)位于 src/zenml/cli/deployment.py它接收部署名稱/ID 與任意--keyvalue參數(shù)將其解析為流水線輸入并轉(zhuǎn)發(fā)給底層 invoke_deployment 工具。默認(rèn)超時(shí) 300 秒5 分鐘也可用--timeout覆蓋使用--no-wait可提交后臺(tái)執(zhí)行并立即返回運(yùn)行 ID。若傳入?yún)?shù)不合法CLI 會(huì)提示通過zenml deployment describe --schema name查看部署的輸入 schema。5.3 通過 HTTP API 調(diào)用curl -X POST http://localhost:8000/invoke \ -H Content-Type: application/json \ -d {parameters: {query: What is 15 * 23?}}HTTP 接口是 CLI 調(diào)用的等價(jià)形式POST /invoke端點(diǎn)接收{(diào)parameters: {...}}負(fù)載parameters中的字段即流水線的輸入?yún)?shù)此處是query。在源碼的 pretty_print_deployment 函數(shù) 中可以看到部署就緒后 CLI 還會(huì)給出配套運(yùn)維命令zenml deployment describe name查看部署詳情包含 Endpoint URL、Swagger 文檔地址與鑒權(quán)信息--show-secret可揭示密鑰zenml deployment logs name -f實(shí)時(shí)跟蹤部署日志zenml deployment invoke name args再次調(diào)用zenml deployment deprovision name/zenml deployment delete name下線或徹底刪除部署。得益于 Pipeline 的返回值類型str部署服務(wù)可以直接把格式化后的天氣/計(jì)算結(jié)果作為響應(yīng)返回實(shí)現(xiàn)真正的「實(shí)時(shí) Agent 服務(wù)」。六、實(shí)戰(zhàn)價(jià)值與適用場(chǎng)景6.1 從示例中可以遷移的模式框架無關(guān)的封裝step內(nèi)直接調(diào)用agent(query)、外部統(tǒng)一格式化這一模式可平移到 LangChain、CrewAI、PydanticAI 等任何框架參見 examples/agent_framework_integrations/README.md 的框架特性對(duì)比。環(huán)境注入Docker 容器內(nèi)依賴 API Key 的environment注入是遠(yuǎn)程執(zhí)行 Agent 流水線時(shí)避免密鑰硬編碼的標(biāo)準(zhǔn)做法。產(chǎn)物命名與追蹤Annotated[..., name]讓每次 Agent 推理的輸入輸出都成為可追溯的 artifact配合 ZenML Dashboard 可審計(jì)每一次調(diào)用。6.2 適用場(chǎng)景批處理夜間批量處理成百上千條查詢?nèi)缗繑?shù)學(xué)計(jì)算、批量城市天氣查詢實(shí)時(shí)推理服務(wù)將 Agent 作為微服務(wù)嵌入 Web 應(yīng)用或獨(dú)立部署通過 HTTP 獲得即時(shí)響應(yīng)Agent 評(píng)估與對(duì)比在 ZenML 中統(tǒng)一編排不同框架Autogen、LangGraph、Strands 等在測(cè)試數(shù)據(jù)集上系統(tǒng)化對(duì)比效果交互式 Demo一鍵部署可分享的 Agent 端點(diǎn)。七、快速參考完整命令清單# 1. 環(huán)境準(zhǔn)備 export OPENAI_API_KEYyour-api-key-here uv venv --python 3.11 source .venv/bin/activate uv pip install -r requirements.txt # 2. 初始化 ZenML zenml init zenml login # 3. 本地運(yùn)行批處理流水線 python run.py # 4. 部署為實(shí)時(shí) HTTP 服務(wù) zenml pipeline deploy run.agent_pipeline --name aws-strands-agent # 5. CLI 調(diào)用 zenml deployment invoke aws-strands-agent --queryCalculate the square root of 144 # 6. HTTP 調(diào)用 curl -X POST http://localhost:8000/invoke \ -H Content-Type: application/json \ -d {parameters: {query: What is 15 * 23?}} # 7. 運(yùn)維 zenml deployment describe aws-strands-agent zenml deployment logs aws-strands-agent -f zenml deployment deprovision aws-strands-agent八、深入閱讀示例源碼examples/agent_framework_integrations/aws_strands/含 run.py、agent.py、requirements.txt同類集成示例總覽examples/agent_framework_integrations/README.md部署命令實(shí)現(xiàn)src/zenml/cli/pipeline.pyzenml pipeline deploy調(diào)用命令實(shí)現(xiàn)src/zenml/cli/deployment.pyzenml deployment invoke部署信息展示src/zenml/cli/utils.pyEndpoint、Swagger、鑒權(quán)與調(diào)用示例【免費(fèi)下載鏈接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.項(xiàng)目地址: https://gitcode.com/GitHub_Trending/ze/zenml創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考