ZenML 编排 LangGraph ReAct Agent从本地管道到实时 HTTP 部署【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml本指南基于 ZenML 官方示例 examples/agent_framework_integrations/langgraph完整讲解如何将一个 LangGraph ReActReasoning ActingAgent 接入 ZenML 管道实现消息驱动的智能体状态管理、内置工具调用与制品管理并进一步将批量式 Agent 工作流部署为可实时响应的 HTTP 服务。读完本文你将掌握从环境搭建、管道编码、CLI 部署到 HTTP API 调用的完整实战链路。示例概览LangGraph Agent 为什么需要 ZenMLLangGraph 提供基于图的工作流抽象非常适合实现 ReAct 模式——让大模型在推理与行动之间循环自主决定何时调用工具、何时给出最终答案。但单个 Agent 脚本难以回答如下生产问题每次执行是否有可复现的记录输入输出制品如何版本化如何把交互式 Agent 变成可供 Web 应用或微服务调用的在线接口ZenML 给出的答案是把 Agent 执行包装成管道pipeline步骤step让 ZenML 承担编排、跟踪、制品管理与部署职责。在 agent_framework_integrations/README.md 列出的 12 个框架集成示例中LangGraph 被定位为Graphs / ReAct agents, Graph workflows类型是高级编排类集成之一。整个示例目录同时展示了统一的最佳实践uv管理环境、Annotated标注制品、try-catch兜底错误处理、以及Agent 执行 → 响应格式化的两段式管道架构。环境准备基于 uv 的一键式依赖安装示例统一要求Python 3.11并使用uv作为包管理器以获得更快的安装体验export OPENAI_API_KEYyour-api-key-here uv venv --python 3.11 source .venv/bin/activate uv pip install -r requirements.txt依赖清单见 requirements.txt其中与 Agent 相关的核心依赖为langgraph0.5.3提供图式 Agent 运行时langchain0.2.0与langchain-openai0.2.0提供 Agent 构造器与 OpenAI 模型接入zenml[local]本地模式的 ZenML SDK含本地存储与服务器jinja23.1.0、secure0.3.0ZenML 本地运行所需的间接依赖。初始化 ZenML 仓库并登录本地模式可直接通过zenml init快速起步无需远端服务器zenml init zenml login之后即可直接运行管道python run.py管道源码剖析Agent 执行与格式化两个步骤管道定义在 run.py其核心思路非常清晰一个执行步骤 一个格式化步骤二者通过管道返回值串联形成可复现、可跟踪的 Agent 工作流。步骤一执行 LangGraph Agentrun_langgraph_agent是一个标准的 ZenML 步骤接收用户查询字符串返回一个带制品名agent_results的字典step def run_langgraph_agent( query: str, ) - Annotated[Dict[str, Any], agent_results]: Execute the LangGraph ReAct agent and return results. try: # LangGraph agents expect messages in a specific format messages [{role: user, content: query}] result agent.invoke({messages: messages}) # Extract the response from the result if messages in result and result[messages]: last_message result[messages][-1] if hasattr(last_message, content): response last_message.content else: response str(last_message) else: response str(result) return {query: query, response: response, status: success} except Exception as e: return { query: query, response: fAgent error: {str(e)}, status: error, }这段代码体现了两点关键实现细节消息式通信LangGraph 的invoke以{messages: [...]}字典为输入与 Chat 模型的对话历史结构一致这正是 README 中Message-Based Communication: Structured agent state management所描述的状态管理方式响应提取的健壮性结果可能包含多条消息代码取最后一条消息并兼容content属性与非结构化对象两种返回类型try-catch兜底保证 Agent 异常时管道仍能产出结构化错误信息status: error实现优雅降级。步骤二格式化响应format_langgraph_response将上一步的字典渲染成人类可读的摘要文本并对成功/失败两种状态分别输出不同模板step def format_langgraph_response( agent_data: Dict[str, Any], ) - Annotated[str, formatted_response]: Format the LangGraph agent results into a readable summary. query agent_data[query] response agent_data[response] status agent_data[status] if status error: formatted f❌ LANGGRAPH AGENT ERROR { * 40} Query: {query} Error: {response} else: formatted f LANGGRAPH REACT AGENT RESPONSE { * 40} Query: {query} Response: {response} Powered by LangGraph (ReAct Agent Tools) return formatted.strip()管道定义Docker 设置与缓存控制agent_pipeline通过pipeline装饰器组装两个步骤并显式配置了两项重要设置docker_settings DockerSettings( python_package_installerPythonPackageInstaller.UV, requirementsrequirements.txt, # relative to the pipeline directory environment{ OPENAI_API_KEY: os.getenv(OPENAI_API_KEY), }, ) pipeline(settings{docker: docker_settings}, enable_cacheFalse) def agent_pipeline( query: str What is the weather in San Francisco?, ) - str: ZenML pipeline that orchestrates the LangGraph ReAct agent. agent_results run_langgraph_agent(query) summary format_langgraph_response(agent_results) return summaryDockerSettings声明了远程/容器化运行时所需的依赖安装方式PythonPackageInstaller.UV、依赖清单路径以及通过环境变量注入的OPENAI_API_KEY——这样密钥不会硬编码进代码而是来自当前进程环境enable_cacheFalse关闭步骤缓存确保每次调用都真实执行 AgentLLM 调用不可缓存且查询是动态的默认查询为What is the weather in San Francisco?运行python run.py时无需传参即可验证整条链路也可通过管道参数传入任意问题。Agent 定义LangGraph ReAct 的最小可用形态Agent 本体位于 langgraph_agent.py展示了如何用 LangChain 的create_agent快速构建一个带工具能力的 ReAct Agentfrom langchain.agents import create_agent def get_weather(city: str) - str: Get weather for a given city. return fIts always sunny in {city}! agent create_agent( modelopenai:gpt-5-nano, tools[get_weather], system_promptYou are a helpful assistant, )要点说明模型示例配置为openai:gpt-5-nanoOpenAI 模型需要OPENAI_API_KEY环境变量工具get_weather是一个普通 Python 函数直接作为工具注入tools列表。LangGraph 的 ReAct 循环会在推理过程中自主决定是否调用它——这正是Reasoning Acting模式的体现系统提示词通过system_prompt设定 Agent 的角色基调文件末尾保留了agent.invoke({messages: [...]})的调用示例注释状态说明 Agent 的调用契约就是消息字典与上面管道步骤中的调用方式完全一致。需要说明的是示例中的天气工具返回的是固定字符串Its always sunny in {city}!属于演示用的占位实现。在实际项目中你可以将其替换为真实的天气 API、搜索服务或计算器LangGraph 的 ReAct 循环会自动适配。部署为实时 HTTP 服务三步走README 展示了 ZenML Pipeline Deployment 能力——将上述批量式 Agent 管道原地转化为常驻的实时 HTTP 服务这一能力同样适用于该目录下的其他框架示例Autogen、CrewAI、LlamaIndex 等。第一步部署管道# 将管道部署为 HTTP 服务 zenml pipeline deploy run.agent_pipeline --name langgraph-agent这里run.agent_pipeline是可导入的管道实例引用source参数来自run.py中的agent_pipeline--name指定部署名。从部署命令的 CLI 实现src/zenml/cli/pipeline.py可以看到更多可用选项--config指定管道配置文件路径--stack指定部署所用的 stack默认使用当前活跃 stack--build复用已有的构建接受 build UUID 或 YAML 文件路径--update/--overtake同名部署已存在时的更新策略--overtake可接管其他用户拥有的部署--attach部署后挂接日志--timeout等待部署完成的最大秒数。第二步通过 CLI 调用zenml deployment invoke langgraph-agent --queryResearch the population of Tokyo and calculate its densityinvoke命令将--query等额外参数透传给部署的管道见 src/zenml/cli/deployment.py支持--timeout控制最长等待时间以及--no-wait提交后台执行并立即返回运行 ID若参数不合法命令会提示用zenml deployment describe --schema name查看部署参数 Schema出错时也可用zenml deployment logs name排查日志。第三步通过 HTTP API 调用部署成功后服务默认监听本地 8000 端口可直接用curl以 JSON 负载调用curl -X POST http://localhost:8000/invoke \ -H Content-Type: application/json \ -d {parameters: {query: What is the square root of 144 and why is it useful?}}请求体中的parameters即管道入参此处为agent_pipeline的query参数响应返回 Agent 的格式化结果。这一模式让 Agent 能力可以无缝嵌入 Web 应用、微服务、交互式 Demo 等实时场景而无需改动任何管道代码。核心特性与适用场景总结对照 README 中列出的特性清单本示例的工程价值可归纳为ReAct Pattern推理与行动循环适合需要多步工具调用的复杂任务如研究东京人口并计算密度这类组合型问题Message-Based Communication以消息字典作为 Agent 状态与输入输出的统一契约结构清晰、易于扩展多轮对话Built-in Tools普通 Python 函数即工具接入成本极低Real-time Deploymentzenml pipeline deployzenml deployment invoke HTTP API 三步即可上线ZenML Orchestration每次 Agent 执行都作为一次可跟踪的管道运行查询与响应以agent_results、formatted_response等命名制品落盘可在 ZenML Dashboard 中回溯实现完整的制品管理与执行记录。在官方列出的生产场景中agent_framework_integrations/README.md这类 Agent 管道既适合批量处理一夜处理成百上千条查询也适合实时服务HTTP API 即时响应既可用于 Agent 评测在测试数据集上对比不同框架也可作为数据处理流水线的一环。可以说LangGraph 解决了Agent 怎么想、怎么做的问题而 ZenML 解决了Agent 怎么跑、怎么管、怎么上线的问题——两者结合构成了一个从开发到生产的最小闭环。扩展阅读框架集成总览与 12 种 Agent 框架对比examples/agent_framework_integrations/README.md管道部署 CLI 的完整参数实现src/zenml/cli/pipeline.py部署调用invoke命令的实现src/zenml/cli/deployment.py本示例 Agent 定义langgraph_agent.py管道定义run.py依赖清单requirements.txt【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 ZenML 编排 LangGraph ReAct Agent：从本地管道到实时 HTTP 部署
ZenML 编排 LangGraph ReAct Agent从本地管道到实时 HTTP 部署【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml本指南基于 ZenML 官方示例 examples/agent_framework_integrations/langgraph完整讲解如何将一个 LangGraph ReActReasoning ActingAgent 接入 ZenML 管道实现消息驱动的智能体状态管理、内置工具调用与制品管理并进一步将批量式 Agent 工作流部署为可实时响应的 HTTP 服务。读完本文你将掌握从环境搭建、管道编码、CLI 部署到 HTTP API 调用的完整实战链路。示例概览LangGraph Agent 为什么需要 ZenMLLangGraph 提供基于图的工作流抽象非常适合实现 ReAct 模式——让大模型在推理与行动之间循环自主决定何时调用工具、何时给出最终答案。但单个 Agent 脚本难以回答如下生产问题每次执行是否有可复现的记录输入输出制品如何版本化如何把交互式 Agent 变成可供 Web 应用或微服务调用的在线接口ZenML 给出的答案是把 Agent 执行包装成管道pipeline步骤step让 ZenML 承担编排、跟踪、制品管理与部署职责。在 agent_framework_integrations/README.md 列出的 12 个框架集成示例中LangGraph 被定位为Graphs / ReAct agents, Graph workflows类型是高级编排类集成之一。整个示例目录同时展示了统一的最佳实践uv管理环境、Annotated标注制品、try-catch兜底错误处理、以及Agent 执行 → 响应格式化的两段式管道架构。环境准备基于 uv 的一键式依赖安装示例统一要求Python 3.11并使用uv作为包管理器以获得更快的安装体验export OPENAI_API_KEYyour-api-key-here uv venv --python 3.11 source .venv/bin/activate uv pip install -r requirements.txt依赖清单见 requirements.txt其中与 Agent 相关的核心依赖为langgraph0.5.3提供图式 Agent 运行时langchain0.2.0与langchain-openai0.2.0提供 Agent 构造器与 OpenAI 模型接入zenml[local]本地模式的 ZenML SDK含本地存储与服务器jinja23.1.0、secure0.3.0ZenML 本地运行所需的间接依赖。初始化 ZenML 仓库并登录本地模式可直接通过zenml init快速起步无需远端服务器zenml init zenml login之后即可直接运行管道python run.py管道源码剖析Agent 执行与格式化两个步骤管道定义在 run.py其核心思路非常清晰一个执行步骤 一个格式化步骤二者通过管道返回值串联形成可复现、可跟踪的 Agent 工作流。步骤一执行 LangGraph Agentrun_langgraph_agent是一个标准的 ZenML 步骤接收用户查询字符串返回一个带制品名agent_results的字典step def run_langgraph_agent( query: str, ) - Annotated[Dict[str, Any], agent_results]: Execute the LangGraph ReAct agent and return results. try: # LangGraph agents expect messages in a specific format messages [{role: user, content: query}] result agent.invoke({messages: messages}) # Extract the response from the result if messages in result and result[messages]: last_message result[messages][-1] if hasattr(last_message, content): response last_message.content else: response str(last_message) else: response str(result) return {query: query, response: response, status: success} except Exception as e: return { query: query, response: fAgent error: {str(e)}, status: error, }这段代码体现了两点关键实现细节消息式通信LangGraph 的invoke以{messages: [...]}字典为输入与 Chat 模型的对话历史结构一致这正是 README 中Message-Based Communication: Structured agent state management所描述的状态管理方式响应提取的健壮性结果可能包含多条消息代码取最后一条消息并兼容content属性与非结构化对象两种返回类型try-catch兜底保证 Agent 异常时管道仍能产出结构化错误信息status: error实现优雅降级。步骤二格式化响应format_langgraph_response将上一步的字典渲染成人类可读的摘要文本并对成功/失败两种状态分别输出不同模板step def format_langgraph_response( agent_data: Dict[str, Any], ) - Annotated[str, formatted_response]: Format the LangGraph agent results into a readable summary. query agent_data[query] response agent_data[response] status agent_data[status] if status error: formatted f❌ LANGGRAPH AGENT ERROR { * 40} Query: {query} Error: {response} else: formatted f LANGGRAPH REACT AGENT RESPONSE { * 40} Query: {query} Response: {response} Powered by LangGraph (ReAct Agent Tools) return formatted.strip()管道定义Docker 设置与缓存控制agent_pipeline通过pipeline装饰器组装两个步骤并显式配置了两项重要设置docker_settings DockerSettings( python_package_installerPythonPackageInstaller.UV, requirementsrequirements.txt, # relative to the pipeline directory environment{ OPENAI_API_KEY: os.getenv(OPENAI_API_KEY), }, ) pipeline(settings{docker: docker_settings}, enable_cacheFalse) def agent_pipeline( query: str What is the weather in San Francisco?, ) - str: ZenML pipeline that orchestrates the LangGraph ReAct agent. agent_results run_langgraph_agent(query) summary format_langgraph_response(agent_results) return summaryDockerSettings声明了远程/容器化运行时所需的依赖安装方式PythonPackageInstaller.UV、依赖清单路径以及通过环境变量注入的OPENAI_API_KEY——这样密钥不会硬编码进代码而是来自当前进程环境enable_cacheFalse关闭步骤缓存确保每次调用都真实执行 AgentLLM 调用不可缓存且查询是动态的默认查询为What is the weather in San Francisco?运行python run.py时无需传参即可验证整条链路也可通过管道参数传入任意问题。Agent 定义LangGraph ReAct 的最小可用形态Agent 本体位于 langgraph_agent.py展示了如何用 LangChain 的create_agent快速构建一个带工具能力的 ReAct Agentfrom langchain.agents import create_agent def get_weather(city: str) - str: Get weather for a given city. return fIts always sunny in {city}! agent create_agent( modelopenai:gpt-5-nano, tools[get_weather], system_promptYou are a helpful assistant, )要点说明模型示例配置为openai:gpt-5-nanoOpenAI 模型需要OPENAI_API_KEY环境变量工具get_weather是一个普通 Python 函数直接作为工具注入tools列表。LangGraph 的 ReAct 循环会在推理过程中自主决定是否调用它——这正是Reasoning Acting模式的体现系统提示词通过system_prompt设定 Agent 的角色基调文件末尾保留了agent.invoke({messages: [...]})的调用示例注释状态说明 Agent 的调用契约就是消息字典与上面管道步骤中的调用方式完全一致。需要说明的是示例中的天气工具返回的是固定字符串Its always sunny in {city}!属于演示用的占位实现。在实际项目中你可以将其替换为真实的天气 API、搜索服务或计算器LangGraph 的 ReAct 循环会自动适配。部署为实时 HTTP 服务三步走README 展示了 ZenML Pipeline Deployment 能力——将上述批量式 Agent 管道原地转化为常驻的实时 HTTP 服务这一能力同样适用于该目录下的其他框架示例Autogen、CrewAI、LlamaIndex 等。第一步部署管道# 将管道部署为 HTTP 服务 zenml pipeline deploy run.agent_pipeline --name langgraph-agent这里run.agent_pipeline是可导入的管道实例引用source参数来自run.py中的agent_pipeline--name指定部署名。从部署命令的 CLI 实现src/zenml/cli/pipeline.py可以看到更多可用选项--config指定管道配置文件路径--stack指定部署所用的 stack默认使用当前活跃 stack--build复用已有的构建接受 build UUID 或 YAML 文件路径--update/--overtake同名部署已存在时的更新策略--overtake可接管其他用户拥有的部署--attach部署后挂接日志--timeout等待部署完成的最大秒数。第二步通过 CLI 调用zenml deployment invoke langgraph-agent --queryResearch the population of Tokyo and calculate its densityinvoke命令将--query等额外参数透传给部署的管道见 src/zenml/cli/deployment.py支持--timeout控制最长等待时间以及--no-wait提交后台执行并立即返回运行 ID若参数不合法命令会提示用zenml deployment describe --schema name查看部署参数 Schema出错时也可用zenml deployment logs name排查日志。第三步通过 HTTP API 调用部署成功后服务默认监听本地 8000 端口可直接用curl以 JSON 负载调用curl -X POST http://localhost:8000/invoke \ -H Content-Type: application/json \ -d {parameters: {query: What is the square root of 144 and why is it useful?}}请求体中的parameters即管道入参此处为agent_pipeline的query参数响应返回 Agent 的格式化结果。这一模式让 Agent 能力可以无缝嵌入 Web 应用、微服务、交互式 Demo 等实时场景而无需改动任何管道代码。核心特性与适用场景总结对照 README 中列出的特性清单本示例的工程价值可归纳为ReAct Pattern推理与行动循环适合需要多步工具调用的复杂任务如研究东京人口并计算密度这类组合型问题Message-Based Communication以消息字典作为 Agent 状态与输入输出的统一契约结构清晰、易于扩展多轮对话Built-in Tools普通 Python 函数即工具接入成本极低Real-time Deploymentzenml pipeline deployzenml deployment invoke HTTP API 三步即可上线ZenML Orchestration每次 Agent 执行都作为一次可跟踪的管道运行查询与响应以agent_results、formatted_response等命名制品落盘可在 ZenML Dashboard 中回溯实现完整的制品管理与执行记录。在官方列出的生产场景中agent_framework_integrations/README.md这类 Agent 管道既适合批量处理一夜处理成百上千条查询也适合实时服务HTTP API 即时响应既可用于 Agent 评测在测试数据集上对比不同框架也可作为数据处理流水线的一环。可以说LangGraph 解决了Agent 怎么想、怎么做的问题而 ZenML 解决了Agent 怎么跑、怎么管、怎么上线的问题——两者结合构成了一个从开发到生产的最小闭环。扩展阅读框架集成总览与 12 种 Agent 框架对比examples/agent_framework_integrations/README.md管道部署 CLI 的完整参数实现src/zenml/cli/pipeline.py部署调用invoke命令的实现src/zenml/cli/deployment.py本示例 Agent 定义langgraph_agent.py管道定义run.py依赖清单requirements.txt【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考