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 ReAct（Reasoning + Acting）Agent 接入 ZenML 管道，实现消息驱动的智能体状态管理、内置工具调用与制品管理，并进一步将批量式 Agent 工作流部署为可实时响应的 HTTP 服务。读完本文，你将掌握从环境搭建、管道编码、CLI 部署到 HTTP API 调用的完整实战链路。

示例概览：LangGraph Agent 为什么需要 ZenML

LangGraph 提供基于图的工作流抽象，非常适合实现 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_KEY="your-api-key-here" uv venv --python 3.11 source .venv/bin/activate uv pip install -r requirements.txt

依赖清单见 requirements.txt，其中与 Agent 相关的核心依赖为：

• langgraph>=0.5.3：提供图式 Agent 运行时；
• langchain>=0.2.0与langchain-openai>=0.2.0：提供 Agent 构造器与 OpenAI 模型接入；
• zenml[local]：本地模式的 ZenML SDK（含本地存储与服务器）；
• jinja2>=3.1.0、secure>=0.3.0：ZenML 本地运行所需的间接依赖。

初始化 ZenML 仓库并登录（本地模式可直接通过zenml init快速起步，无需远端服务器）：

zenml init zenml login

之后即可直接运行管道：

python run.py

管道源码剖析：Agent 执行与格式化两个步骤

管道定义在 run.py，其核心思路非常清晰：一个执行步骤 + 一个格式化步骤，二者通过管道返回值串联，形成可复现、可跟踪的 Agent 工作流。

步骤一：执行 LangGraph Agent

run_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": f"Agent error: {str(e)}", "status": "error", }

这段代码体现了两点关键实现细节：

1. 消息式通信：LangGraph 的invoke以{"messages": [...]}字典为输入，与 Chat 模型的对话历史结构一致，这正是 README 中"Message-Based Communication: Structured agent state management"所描述的状态管理方式；
2. 响应提取的健壮性：结果可能包含多条消息，代码取最后一条消息，并兼容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_installer=PythonPackageInstaller.UV, requirements="requirements.txt", # relative to the pipeline directory environment={ "OPENAI_API_KEY": os.getenv("OPENAI_API_KEY"), }, ) @pipeline(settings={"docker": docker_settings}, enable_cache=False) 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 summary

• DockerSettings声明了远程/容器化运行时所需的依赖安装方式（PythonPackageInstaller.UV）、依赖清单路径，以及通过环境变量注入的OPENAI_API_KEY——这样密钥不会硬编码进代码，而是来自当前进程环境；
• enable_cache=False关闭步骤缓存，确保每次调用都真实执行 Agent（LLM 调用不可缓存，且查询是动态的）；
• 默认查询为"What is the weather in San Francisco?"，运行python run.py时无需传参即可验证整条链路；也可通过管道参数传入任意问题。

Agent 定义：LangGraph ReAct 的最小可用形态

Agent 本体位于 langgraph_agent.py，展示了如何用 LangChain 的create_agent快速构建一个带工具能力的 ReAct Agent：

from langchain.agents import create_agent def get_weather(city: str) -> str: """Get weather for a given city.""" return f"It's always sunny in {city}!" agent = create_agent( model="openai:gpt-5-nano", tools=[get_weather], system_prompt="You are a helpful assistant", )

要点说明：

• 模型：示例配置为openai:gpt-5-nano（OpenAI 模型），需要OPENAI_API_KEY环境变量；
• 工具：get_weather是一个普通 Python 函数，直接作为工具注入tools列表。LangGraph 的 ReAct 循环会在推理过程中自主决定是否调用它——这正是"Reasoning + Acting"模式的体现；
• 系统提示词：通过system_prompt设定 Agent 的角色基调；
• 文件末尾保留了agent.invoke({"messages": [...]})的调用示例（注释状态），说明 Agent 的调用契约就是消息字典，与上面管道步骤中的调用方式完全一致。

需要说明的是：示例中的天气工具返回的是固定字符串（"It's 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 --query="Research the population of Tokyo and calculate its density"

invoke命令将--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 Deployment：zenml pipeline deploy+zenml 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