这篇指南介绍如何通过 Pluto 将 LangServe 应用一键部署到 AWS 上，全程无需登陆 AWS 控制台，只需准备好 AWS 访问凭证即可。
LangServe 是 LangChain 的一个子项目，可以帮助开发者通过 REST API 部署 LangChain 的 Runnable 和 Chain。同时，它还提供了一个用于调用部署在服务器上的 Runnable 的客户端类，包括 Python、TypeScript 等多个版本，以及默认提供 Playground 供部署后直接在线试用。
可以从这里获取本示例的全部代码，这个链接提供了本示例应用的在线 IDE，点击右上角 Fork 按钮即可创建你自己的开发环境，然后你就可以直接在浏览器中修改代码并部署到 AWS 上了。
⚠️ 注意：
由于 Pluto 目前仅支持单文件，因此 LangServe 应用的代码需要放在一个文件中。
受限于 Pluto 现有的打包方式，目前还不支持 LangChain 的 Template 生态。快速支持中
环境准备
如果你还没有配置 Pluto 开发环境，请参考快速开始中的本地开发进行配置，或使用 Pluto 提供的在线 IDE 或容器体验。
开发 LangServe 应用
这里我们介绍两种不同的开发 LangServe 应用的方式：一种是 langserve 教程中提及的开发方式，使用 langchain app new 命令创建一个新的 LangChain 应用；另一种是使用 pluto new 命令创建一个新的 Pluto 应用。
方式 1： langchain app new
安装 LangChain CLI
pip install langchain-cli
复制代码
创建 LangServe 应用
使用 langchain app new 命令创建一个新的 LangChain 应用，这个命令会在当前目录下创建一个新的目录，目录名为你指定的应用名：
langchain app new --non-interactive my-appcd my-app
复制代码
注意：langchain app new 命令依赖 git，请确保你的环境中已经安装了 git。如果你在使用 Pluto 提供的容器环境，请先执行这条命令 apt-get update && apt-get install -y git 安装 git。
编写 LangServe 应用
你可以根据你的需求在 app/server.py 文件开发基于 LangChain 的 AI 应用，最后你应该会开发出 1 个或多个 LangChain 的 Agent、Chain 等 Runnable 实例。这些实例可以通过 LangServe 提供的 add_routes 方法添加到 FastAPI 中，然后以 HTTP 服务的形式提供给用户。
我们以 LangServe 首页的示例应用为例，该示例使用 add_routes 方法将多个 LangChain 的 Runnable 实例添加到 FastAPI 中。
修改代码适配 Pluto
接下来，我们需要将 LangServe 应用适配为 Pluto 应用，以便 Pluto 可以将其部署到 AWS 上。适配过程也非常简单，只需两步
首先，需要将与 FastAPI app 相关的代码放到一个函数中，并使这个函数返回 FastAPI app 实例，这里假设这个函数名为 return_fastapi_app。
然后，将代码中的 if __name__ == "__main__": 代码块整体替换为以下 4 条语句，你可以修改 router_name 为你的喜欢的名字，这个名字与最终在 AWS 创建的 Api Gateway 实例的名称相关。
from mangum import Mangumfrom pluto_client import Router
router = Router("router_name")router.all("/*", lambda *args, **kwargs: Mangum(return_fastapi_app(), api_gateway_base_path="/dev")(*args, **kwargs), raw=True)
复制代码
最终的代码如下：
由于 Pluto 还未支持传递环境变量，因此我们需要在代码中配置 OpenAI 和 Anthropic 的 API Key。
from fastapi import FastAPIfrom langchain.prompts import ChatPromptTemplate# from langchain.chat_models import ChatAnthropic, ChatOpenAIfrom langchain_anthropic import ChatAnthropicfrom langchain_openai import ChatOpenAIfrom langserve import add_routesfrom langchain.pydantic_v1 import SecretStr
from mangum import Mangumfrom pluto_client import Router
OPENAI_API_KEY = SecretStr("sk-EUk0Tal8cIkmG4vJF904F57a9eE241A8Ae72666fAxxxxxxx")ANTHROPIC_API_KEY = SecretStr("sk-EUk0Tal8cIkmG4vJF904F57a9eE241A8Ae72666fAxxxxxxx")
model = ChatAnthropic(api_key=ANTHROPIC_API_KEY)prompt = ChatPromptTemplate.from_template("tell me a joke about {topic}")
def return_fastapi_app(): # The langserve depends on this, but it may not come pre-installed. # So, we write it here to ensure it is installed. import sse_starlette
app = FastAPI( title="LangChain Server", version="1.0", description="A simple api server using Langchain's Runnable interfaces", )
add_routes( app, ChatOpenAI(api_key=OPENAI_API_KEY), path="/openai", )
add_routes( app, ChatAnthropic(api_key=ANTHROPIC_API_KEY), path="/anthropic", )
add_routes( app, prompt | model, path="/joke", )
return app
router = Router("router_name")router.all( "/*", lambda *args, **kwargs: Mangum(return_fastapi_app(), api_gateway_base_path="/dev")(*args, **kwargs), raw=True,)
复制代码
部署到 AWS
在正式部署之前，我们需要将该项目初始化为 Pluto 项目，这样 Pluto 才能识别并部署该项目。在项目根目录下运行以下命令，Pluto 会交互式的引导你初始化项目，编程语言请选择 Python：
初始化完成后，我们需要安装一些必需的依赖，执行以下两条命令：
npm install
pip install poetry# Python 版本不匹配时，请修改 pyproject.toml 中的 python 版本号poetry add pluto-client mangum langchain-openai langchain_anthropic sse_starlette
复制代码
最后，我们执行以下命令就可以将 LangServe 应用部署到 AWS 上：
poetry shellpluto deploy app/server.py
复制代码
注意：如果你的研发环境为 Arm64 架构，请在环境中安装并启动 docker。如果你使用的是 Pluto 提供的容器环境，环境中已经安装了 docker，但需要在启动时配置 --privileged 参数，然后在容器中手动启动 docker 服务，启动命令为:
dockerd > /dev/null 2>&1 &
复制代码
这条命令会把你的 LangServe 应用程序作为无服务器应用程序部署到 AWS 上，将创建一个 Api Gateway 实例和一个 Lambda 函数实例来处理请求。同时会在终端打印出 AWS 的 Api Gateway 的 URL，你可以通过访问这个 URL 来访问部署的应用程序。
方式 2： pluto new
创建 Pluto 应用
使用 pluto new 命令创建一个新的 Pluto 应用，这个命令会交互式地创建一个新的 Pluto 应用，并且会在当前目录下创建一个新的目录，目录名为你指定的应用名，编程语言请选择 Python：
创建完成后，进入到创建的应用目录，并安装必要的依赖：
cd <project name>npm installpip install -r requirements.txt
复制代码
编写 LangServe 应用
你可以根据你的需求在 app/main.py 文件开发基于 LangChain 的 AI 应用，最后你应该会开发出 1 个或多个 LangChain 的 Agent、Chain 等 Runnable 实例。这些实例可以通过 LangServe 提供的 add_routes 方法添加到 FastAPI 中，然后以 HTTP 服务的形式提供给用户。
但这里，我们需要将与 FastAPI app 相关的代码放到一个函数中，并使这个函数返回 FastAPI app 实例，最后将这个函数封装在 Router 的 all 方法中，以便 Pluto 可以将其部署到 AWS 上。
以 LangServe 首页的示例应用为例，最终的代码与上一种适配后的代码相同。
部署到 AWS
确保所有依赖都已安装完成后，执行下面这条命令就可以将 LangServe 应用部署到 AWS 上：
注意：如果你的研发环境为 Arm 架构，请在环境中安装并启动 docker。如果你使用的是 Pluto 提供的容器环境，环境中已经安装了 docker，但需要在启动时配置 --privileged 参数，然后在容器中手动启动 docker 服务，启动命令为:
dockerd > /dev/null 2>&1 &
复制代码
pluto deploy 会把你的 LangServe 应用程序作为无服务器应用程序部署到 AWS 上，将创建一个 Api Gateway 实例和一个 Lambda 函数实例来处理请求。同时会在终端打印出 AWS 的 Api Gateway 的 URL，你可以通过访问这个 URL 来访问部署的应用程序。
访问
部署完成后，你可以从终端看到 Pluto 输出的 URL，你可以通过这个 URL 访问你的 LangServe 应用程序。
⚠️ 注意：
Pluto 尚未支持 Stream 访问，在使用 LangServe 的 astream 方法时结果仍是一次性返回的。
由于第一次加载 LangChain 依赖库可能会比较慢，所以第一次调用 LangServe 服务或者访问 Playground 时可能会比较慢，超过 30 秒后会自动超时。因此，如果你在访问时遇到超时问题，请再次尝试。
每个 AWS Lambda 函数的实例只能同时处理一个请求，而每个 LangChain 的 Lambda 实例的初始化时间接近 2 分钟，因此在高并发情况下可能会出现请求超时问题。
通过 RemoteRunnable 调用
还是以 LangServe 首页示例应用提供的 Client 为例，你只需要将 LangServe 示例中的本地 URL 替换为 Pluto 输出的 URL 即可。
我们没有使用 Anthropic 模型，因此只保留了 OpenAI 和 Joke 模型的调用，修改后的 Python 客户端代码如下，请将代码中的 https://fcz1u130w3.execute-api.us-east-1.amazonaws.com/dev 替换为 Pluto 输出的 URL：
import asyncio
from langchain.schema import SystemMessage, HumanMessagefrom langchain.prompts import ChatPromptTemplatefrom langchain.schema.runnable import RunnableMapfrom langserve import RemoteRunnable
openai = RemoteRunnable( "https://fcz1u130w3.execute-api.us-east-1.amazonaws.com/dev/openai/")joke_chain = RemoteRunnable( "https://fcz1u130w3.execute-api.us-east-1.amazonaws.com/dev/joke/")
def sync_inoke(): result = joke_chain.invoke({"topic": "parrots"}) print( ">> The result of `joke_chain.invoke({'topic': 'parrots'})` is:\n", result.content, "\n", )
async def async_inoke(): result = await joke_chain.ainvoke({"topic": "parrots"}) print( ">> The result of `await joke_chain.ainvoke({'topic': 'parrots'})` is:\n", result.content, "\n", )
prompt = [ SystemMessage(content="Act like either a cat or a parrot."), HumanMessage(content="Hello!"), ]
# Supports astream print(">> The result of `openai.astream(prompt)` is:") async for msg in openai.astream(prompt): print(msg.content, end=" | ", flush=True) print()
def custom_chain(): prompt = ChatPromptTemplate.from_messages( [("system", "Tell me a long story about {topic}")] )
# Can define custom chains chain = prompt | RunnableMap( { "openai": openai, "anthropic": openai, } )
result = chain.batch([{"topic": "parrots"}, {"topic": "cats"}]) print( ">> The result of `chain.batch([{'topic': 'parrots'}, {'topic': 'cats'}])` is:\n", result, )
async def main(): sync_inoke() await async_inoke() custom_chain()
asyncio.run(main())
复制代码
下面这幅图展示了执行 Python 客户端代码的结果：
修改后的 TypeScript 客户端代码如下，请将代码中的 <your-api-gateway-url> 替换为 Pluto 输出的 URL：
import { RemoteRunnable } from "@langchain/core/runnables/remote";
const chain = new RemoteRunnable({ url: `<your-api-gateway-url>/joke/`,});const result = await chain.invoke({ topic: "cats",});
复制代码
通过 curl 访问
同样，你也只需要将示例中的 <your-api-gateway-url> 替换为 Pluto 输出的 URL：
curl --location --request POST '<your-api-gateway-url>/joke/invoke' \ --header 'Content-Type: application/json' \ --data-raw '{ "input": { "topic": "cats" } }'
复制代码
下面这幅图展示了执行 curl 命令的结果：
通过浏览器访问 Playground
受限于 LangServe 目前的路由策略，我们无法在不修改代码的情况下直接通过浏览器访问 LangServe 的 Playground，在这个 PR 合并之后，就可以直接支持通过浏览器访问 LangServe 的 Playground 了。
现在，我们需要对每一个 add_routes 方法，再额外添加一个 add_routes 方法，并在 path 参数前添加 /dev 前缀，这样就可以在浏览器中访问 LangServe 的 Playground 了。下面是一个示例代码：
add_routes( app, ChatOpenAI(api_key=OPENAI_API_KEY), path="/openai",)
add_routes( app, ChatOpenAI(api_key=OPENAI_API_KEY), path="/dev/openai",)
复制代码
修改部署后，可以通过以下 URL 访问示例应用的 Playground，注意的是，访问路径中需要额外添加 /dev，即路径中包含两个 /dev。注意，URL 可能会被重定向，如果被修改了，请重新调整路径并再次访问。
OpenAI: <your-api-gateway-url>/dev/openai/playground
Anthropic: <your-api-gateway-url>/dev/anthropic/playground
Joke: <your-api-gateway-url>/dev/joke/playground
下面两幅图分别展示了通过浏览器访问 OpenAI 和 Joke 的 Playground 的结果：
清理
如果你希望将部署的 LangServe 应用程序从 AWS 上下线，只需要执行以下命令即可：
总结
在本文中，我们详细探讨了如何使用 Pluto 将 LangServe 应用一键部署到 AWS 云平台。这种方式即使你不熟悉 AWS 的操作，也可以轻松地将 LangServe 应用部署到云端，实现远程调用和 Playground 的访问。
Pluto 还提供了自动创建 DynamoDB、SNS、SageMaker 等资源的能力，你只需要编写代码，pluto deploy 就会自动地在 AWS 上创建和配置这些资源，将云的计算、存储等能力更便捷地提供给你，帮助你更轻松地研发出功能强大的 AI 应用，实现你的 idea💡，你可以从更多资源中获取更多信息。
我们尽可能让本文的步骤简单易懂，即使你对 Pluto 或 AWS 不太熟悉，也可以轻松上手。如果你在阅读或实践时遇到问题，或者有新的想法与需求，请随时通过提交 issue 或加入 Pluto Slack 社区寻求帮助。
更多资源
极速体验版
将该脚本的 OPENAI_API_KEY、ANTHROPIC_API_KEY、AWS_ACCESS_KEY_ID、AWS_SECRET_ACCESS_KEY、AWS_REGION 等参数替换为你的真实值，然后将该脚本保存到你本地。
执行该脚本，将会自动创建一个 LangServe 示例应用，并将其部署到 AWS 上，最后输出部署的 URL，你可以参考上文中的 访问 部分来访问部署的应用。
执行结束后，会进入一个交互式命令行，方便你通过 pluto destroy 下线部署的应用。
OPENAI_API_KEY="<your-openai-api-key>"AWS_ACCESS_KEY_ID="<your-aws-access-key-id>"AWS_SECRET_ACCESS_KEY="<your-aws-secret-access-key>"AWS_REGION="us-east-1"
# Prepare the modified code of LangServe applicationMODIFIED_CODE=$(cat <<EOFfrom fastapi import FastAPIfrom langchain.prompts import ChatPromptTemplatefrom langchain_openai import ChatOpenAIfrom langserve import add_routesfrom langchain.pydantic_v1 import SecretStr
from mangum import Mangumfrom pluto_client import Router
OPENAI_API_KEY = SecretStr("${OPENAI_API_KEY}")
model = ChatOpenAI(api_key=OPENAI_API_KEY)prompt = ChatPromptTemplate.from_template("tell me a joke about {topic}")
def return_fastapi_app(): # The langserve depends on this, but it may not come pre-installed. # So, we write it here to ensure it is installed. import sse_starlette
app = FastAPI( title="LangChain Server", version="1.0", description="A simple api server using Langchain's Runnable interfaces", )
add_routes( app, ChatOpenAI(api_key=OPENAI_API_KEY), path="/openai", )
add_routes( app, ChatOpenAI(api_key=OPENAI_API_KEY), path="/dev/openai", )
add_routes( app, prompt | model, path="/joke", )
add_routes( app, prompt | model, path="/dev/joke", )
return app
router = Router("router_name")router.all( "/*", lambda *args, **kwargs: Mangum(return_fastapi_app(), api_gateway_base_path="/dev")(*args, **kwargs), raw=True,)EOF)
# Prepare the package.json file, used by the PlutoPACKAGE_JSON=$(cat <<EOF{ "name": "my-app", "private": true, "version": "0.0.0", "scripts": { "test:dev": "pluto test --sim", "test:prod": "pluto test", "deploy": "pluto deploy", "destroy": "pluto destroy" }, "dependencies": {}, "devDependencies": { "@types/node": "^20", "typescript": "^5.2.2", "@plutolang/base": "latest", "@plutolang/pluto-infra": "latest", "@pulumi/pulumi": "^3.88.0" }, "main": "dist/index.js"}EOF)
# Prepare the Pluto configuration filePLUTO_YML=$(cat <<EOFcurrent: awslanguage: pythonstacks: - configs: {} name: aws platformType: AWS provisionType: PulumiEOF)
# Prepare the AWS credentialsAWS_CREDENTIALS=$(cat <<EOF[default]aws_access_key_id = ${AWS_ACCESS_KEY_ID}aws_secret_access_key = ${AWS_SECRET_ACCESS_KEY}EOF)
# Prepare the AWS configurationAWS_CONFIG=$(cat <<EOF[default]region = ${AWS_REGION}EOF)
# Prepare the script to run inside the Docker containercat <<EOF1 > script.sh#!/bin/bash
apt updateapt install -y git
pip install langchain-cli poetry
langchain app new --non-interactive my-appcd my-app
cat << EOF2 > app/server.py${MODIFIED_CODE}EOF2
cat << EOF3 > package.json${PACKAGE_JSON}EOF3
mkdir -p .plutocat << EOF4 > .pluto/pluto.yml${PLUTO_YML}EOF4
npm installsed -i 's/\^3.11/\^3.10/' pyproject.tomlpoetry add pluto-client mangum langchain-openai sse_starlette
mkdir -p ~/.awscat << EOF5 > ~/.aws/credentials${AWS_CREDENTIALS}EOF5cat << EOF6 > ~/.aws/config${AWS_CONFIG}EOF6
source \$(poetry env info --path)/bin/activatepluto deploy -y --force app/server.py
bashEOF1
# Run the script inside the Docker containerdocker run -it --rm \ --platform linux/amd64 \ -v $(pwd)/script.sh:/script.sh \ plutolang/pluto:latest bash -c "bash /script.sh"
复制代码
评论