使用 LangServe 构建和部署 MinIO 支持的 LangChain 代理 API

• 使用 LangChain 创新 S3 存储桶检索: 本指南介绍如何利用 LangChain 的 S3 加载器和 OpenAI API 创建自定义对话式 AI 代理，建立一种简化的数据管理方法。
• 使用 MinIO 增强 LangChain 代理: 深入探讨如何利用 MinIO 的强大功能，展示 LangChain 与 OpenAI 的 GPT 如何在 AI 和 ML 数据处理方面开拓新领域。

基于这些见解，我们现在将注意力转向 LangServe，它是将 LangChain 应用程序从开发过渡到部署的关键工具，简化了启动生产就绪 API 的过程。

LangServe：简化 LangChain 应用程序的部署

LangServe 是开发人员的基石，消除了传统上与 API 部署相关的复杂性。它能够将与 MinIO 集成的 LangChain 应用程序顺利转换为可访问、易于使用的 API。以下是 LangServe 如何重新定义部署环境

• 自动 API 端点创建：LangServe 的自动化功能可以轻松生成必要的 API 端点，简化开发工作并显着缩短部署时间。
• 模式生成和验证：LangServe 凭借其智能模式推断功能，确保 API 提供定义明确的接口，从而简化集成并提供无缝的用户体验。
• 可自定义的端点配置：LangServe 提供各种端点以满足不同的应用程序需求，从同步操作到实时更新，为开发人员提供无与伦比的灵活性。
• 轻松集成：LangServe 最引人注目的功能可能在于它能够与现有 LangChain 代码无缝集成，这意味着开发人员可以利用其当前的代码库和专业知识，而无需进行重大修改。

深入了解 LangChain 和 Langserve

我们将深入探讨将 MinIO 与 LangChain 集成的过程，具体步骤如下。

1. 使用 langchain-cli 创建一个 LangChain 应用程序。
2. 在 agent.py 文件中开发一个自定义的 LangChain 代理。
3. 在 server.py 中实现我们的代理，以便作为 LangServe API 运行。

使用 LangChain 的命令行界面创建应用程序

使用 LangServe 部署 LangChain 应用程序带来了无缝的集成体验，弥合了复杂 AI 功能和 RESTful API 暴露之间的差距，使开发人员能够高效地利用 LangChain 的全部功能，为在当今快节奏的数字环境中部署智能应用程序设定了新标准。

LangChain 提供了一种便捷而简单的方法来使用其 langchain-cli 库 创建应用程序，该库可以使用 pip 安装。该软件包提供了一个界面，允许用户通过利用现有的 LangChain 应用程序模板 或创建自己的模板来轻松创建新的应用程序。 

注意：所有必要文件都位于 MinIO 的“blog-assets”存储库中，位于名为“minio-langserve-deployment”的目录下。 

要创建一个新的 LangChain 应用程序，我们可以从以下命令开始，以创建一个 虚拟环境 并安装 langchain-cli 软件包

mkdir minio-langserve-testing
cd minio-Langserve-testing


python -m venv .myenv
source .myenv/bin/activate


pip install langchain-cli

为了使用 langchain-cli 创建一个新的应用程序，我们可以在终端中输入 langchain，以下命令用于创建一个名为 my-app 的新应用程序目录。

langchain app new my-app

使用上述命令创建的 langchain 应用程序执行了所有繁重的工作，为开发创建了一个一致的环境。一个全新的 LangChain 应用程序的结构如下所示

./my-app
├── Dockerfile
├── README.md
├── app
│   ├── __init__.py
│   └── server.py ⇐ (This is where we will import our agent into)
├── packages ⇐  (This directory is where we will write our agent)
│   └── README.md
└── pyproject.toml

在接下来的步骤中，我们将通过编写一个名为 packages/agent.py 的新文件，并修改 app/server.py 来修改新创建的 LangChain 应用程序 (my-app)。 

以下是我们将在本文中讨论的文件

• my-app/packages/agent.py
• my-app/app/server.py

开发一个使用 LangServe 部署的 LangChain MinIO 代理

为了说明如何使用 LangServe 部署与 MinIO 集成的 LangChain 代理，我们将从在 agent.py 中保存代理链代码开始。

首先，让我们初始化一个连接到 "play.min.io:443" 公共服务器的 minio_client。此文件最终将调用 LangChain 的 agent_executor，使我们能够将其传递给 LangServe 的 add_route 包装器。

注意：阅读之前的出版物“MinIO Langchain 工具”将为您提供有关使用 LangChain 和 MinIO 协同开发的宝贵见解。我们将遵循类似的概念方法，但将包含额外的 MinIO 工具逻辑。

要开始，请使用文本编辑器打开 agent.py 文件

sudo nano packages/agent.py

在文件开头，导入必要的软件包，例如 os、io、minio 和 ChatOpenAI

import os
import io
from minio import Minio
from minio.error import S3Error
from langchain_openai import ChatOpenAI


os.environ["OPENAI_API_KEY"] = "<<Your API Key Here>>"

# Initialize llm
llm = ChatOpenAI(api_key=os.environ["OPENAI_API_KEY"]) 

# Initialize MinIO client
minio_client = Minio('play.min.io:443',
                     access_key='minioadmin',
                     secret_key='minioadmin',
                     secure=True)

在此代码片段中，我们导入了所需的软件包，并使用存储在 OPENAI_API_KEY 环境变量中的 OpenAI API 密钥初始化了 ChatOpenAI 语言模型。我们还通过向 "play.min.io" 公共服务器提供必要的连接详细信息来初始化 minio_client。

接下来，让我们定义 MinIO 存储桶，如果它不存在，则创建它

# This variable will check if bucket exists
bucket_name = "test"


try:
    # Check if bucket exists
    if not minio_client.bucket_exists(bucket_name):
        # Create the bucket because it does not exist
        minio_client.make_bucket(bucket_name)
        print(f"Bucket '{bucket_name}' created successfully.")
    else:
        print(f"Bucket '{bucket_name}' already exists.")
except S3Error as err:
    print(f"Error encountered: {err}")

在这里，我们将 bucket_name 定义为 "test"，并使用 minio_client.bucket_exists() 方法检查它是否已经存在。如果存储桶不存在，我们将使用 minio_client.make_bucket() 创建它。如果存储桶已经存在，我们将打印一条消息以示表明。我们还使用 try-except 块包含错误处理，以捕获并打印在过程中可能发生的任何 S3Error。

完成基本设置后，我们现在可以继续定义 MinIO 工具功能，并创建代理执行器，我们将在接下来的步骤中进行介绍。

使用 LangChain 的函数装饰器用于代理工具

Langchain 和 Langserve 都提供了类似的方法来封装逻辑和功能，使其能够无缝地集成到代理和链逻辑中。这是通过在定义的函数内使用 @tool 装饰器以及详细的文档字符串来实现的，它将函数标记为可重用组件，这些组件可以被 AI 代理利用和解释。

让我们仔细看看提供的代码示例

from langchain.agents import tool

@tool
def upload_file_to_minio(bucket_name: str, object_name: str, data_bytes: bytes):
    """
    Uploads a file to MinIO.
    
    Parameters:
        bucket_name (str): The name of the bucket.
        object_name (str): The name of the object to create in the bucket.
        data_bytes (bytes): The raw bytes of the file to upload.
    """
    data_stream = io.BytesIO(data_bytes)
    minio_client.put_object(bucket_name, object_name, data_stream, length=len(data_bytes))
    return f"File {object_name} uploaded successfully to bucket {bucket_name}."

upload_file_to_minio 函数使用 @tool 装饰，表明它是一个可重用组件。它接受上传文件到 MinIO 存储桶所需的必要参数，例如存储桶名称、对象名称和文件的原始字节。该函数利用 minio_client 执行文件上传操作，并在完成后返回一条成功消息。

@tool
def download_file_from_minio(file_info):
    """
    Custom function to download a file from MinIO.
    
    Expects file_info dict with 'bucket_name', 'object_name', and 'save_path' keys.
    'save_path' should be the local path where the file will be saved.
    """
    bucket_name = file_info['bucket_name']
    object_name = file_info['object_name']
    save_path = file_info['save_path']
    minio_client.get_object(bucket_name, object_name, save_path)

类似地，download_file_from_minio 函数也使用 @tool 标记。它期望一个 file_info 字典，其中包含从 MinIO 存储桶下载文件所需的必要信息，例如存储桶名称、对象名称和应保存文件的本地路径。该函数使用 minio_client 从指定的存储桶中检索对象，并将其保存到指定的本地路径。

@tool
def list_objects_in_minio_bucket(file_info):
    """
    Custom function to list objects in a MinIO bucket.
    
    Expects file_info dict with 'bucket_name' key.
    Returns a list of dictionaries containing 'ObjectKey' and 'Size' keys.
    """
    bucket_name = file_info['bucket_name']
    response = minio_client.list_objects(bucket_name)
    return [{'ObjectKey': obj.object_name, 'Size': obj.size} for obj in response.items]

list_objects_in_minio_bucket 函数也使用 @tool 装饰，负责列出 MinIO 存储桶中存在的对象。它期望一个 file_info 字典，其中包含 bucket_name 键。该函数使用 minio_client 检索指定存储桶中的对象列表，并返回一个字典列表，每个字典包含每个对象的键和大小。

通过将这些功能封装为工具，Langchain 和 Langserve 使 AI 代理能够将它们无缝地纳入其逻辑和决策过程。代理可以根据手头的任务智能地选择和执行适当的工具，增强其功能，并允许与 MinIO 存储系统进行更复杂和动态的交互。

了解 LangChain 的可运行方法

LangChain 提供了无数种方法来构建自定义逻辑，其中一种方法是使用“可运行”。对于上述演示逻辑，RunnableLambda 是 LangChain 提供的一种结构，它允许将函数视为 AI 代理逻辑中的可执行单元。 

from langchain_core.runnables import RunnableLambda


upload_file_runnable = RunnableLambda(upload_file_to_minio)
download_file_runnable = RunnableLambda(download_file_from_minio)
list_objects_runnable = RunnableLambda(list_objects_in_minio_bucket)

通过使用 RunnableLambda 包装工具函数，我们创建了可运行实例 (upload_file_runnable、download_file_runnable 和 list_objects_runnable)，代理可以在执行过程中调用这些实例。这些可运行实例封装了相应的工具函数，并为代理提供了一个统一的接口来与它们进行交互。

tools = [upload_file_to_minio, download_file_from_minio, list_objects_in_minio_bucket]


llm_with_tools = llm.bind_tools(tools)

工具列表包含原始工具函数 (upload_file_to_minio、download_file_from_minio 和 list_objects_in_minio_bucket)，它们是代理功能的基石。llm.bind_tools(tools) 行将工具绑定到语言模型 (llm)，在模型的推理能力与工具提供的特定功能之间建立联系。生成的 llm_with_tools 表示增强了知识和使用绑定工具的能力的语言模型。

使用 RunnableLambda 和将工具绑定到语言模型，展示了 LangChain 和 LangServe 在创建功能强大且可定制的 AI 代理方面的灵活性和可扩展性。通过将语言模型的功能与工具中封装的特定功能相结合，AI 代理获得了执行复杂任务的能力，例如将文件上传到 MinIO、从 MinIO 下载文件以及列出 MinIO 存储桶中的对象。 

编写一个提示模板来指导我们的代理

接下来，我们将重点放在引导 AI 代理理解和响应用户输入的提示模板上。它使用 `ChatPromptTemplate.from_messages()` 方法定义，该方法接受一个消息列表，消息表示为包含角色和消息内容的元组。

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.agents.format_scratchpad.openai_tools import format_to_openai_tool_messages
from langchain.agents.output_parsers.openai_tools import OpenAIToolsAgentOutputParser
from langchain_core.messages import AIMessage, HumanMessage


prompt_template = ChatPromptTemplate.from_messages([
    ("system", "You are a powerful assistant equipped with file management capabilities."),
    ("user", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

提示包含三条消息

1. 一条“系统”消息，将 AI 代理的上下文设置为具有文件管理功能的强大助手。
2. 一条“用户”消息，表示用户的输入，使用 `{input}` 占位符。
3. 一个名为“agent_scratchpad” 的 `MessagesPlaceholder`，用于存储代理的中间步骤和思考过程。

`format_to_openai_tool_messages` 函数将代理的 scratchpad 格式化为与 OpenAI 工具兼容的格式，而 OpenAIToolsAgentOutputParser 类将模型的响应解析为代理可以解释的结构化格式。

`AIMessage` 和 `HumanMessage` 类表示代理和用户之间交换的消息，为处理代理逻辑中的通信提供了一种标准化方式。

通过定义提示模板，我们为 AI 代理提供了清晰的结构和上下文，以便理解和响应用户输入，利用“agent_scratchpad” 占位符跟踪其在解决任务过程中的中间步骤和思考过程。

使用工具定义代理

最后，为了完成我们的 `agent.py`，我们定义了我们的代理，并创建了一个 AgentExecutor，它可以使用 LangServe 库的 `add_route` 函数从 `server.py` 脚本导入和调用。

我们实例化了必要的组件并将它们链接在一起以创建一个单个代理变量。

agent = (
    {
        "input": lambda x: x["input"],
        "agent_scratchpad": lambda x: format_to_openai_tool_messages(x["intermediate_steps"]),
    }
    | prompt_template
    | llm_with_tools
    | OpenAIToolsAgentOutputParser()
)

代理是使用字典和链式操作的组合定义的。输入键从传入数据中提取用户输入，而 `agent_scratchpad` 键使用 `format_to_openai_tool_messages` 函数格式化代理思考过程的中间步骤。代理还包含提示模板 (`prompt_template`)、带有工具的语言模型 (`llm_with_tools`) 和输出解析器 (`OpenAIToolsAgentOutputParser()`)。

定义 AgentExecutor 来执行代理

为了创建 `AgentExecutor`，我们为其提供定义的代理、可用的工具，并将 `verbose=True` 设置为详细输出。

from langchain.agents import tool, AgentExecutor


agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

`AgentExecutor` 使用提供的代理和工具来理解任务并根据用户的输入选择合适的工具。代理不是为每个工具使用单独的提示，而是使用一个单一的提示模板来指导它如何根据给定的输入使用工具。代理在执行过程中动态地选择合适的工具。

使用 AgentExecutor 定义 LangServe 路由

设置我们的应用程序并将其与 LangServe 集成，为将我们的 LangChain 应用程序部署和管理为 API 提供了一条简化的路径。FastAPI 因其性能和易用性而被选为，它支持异步操作并自动生成 API 文档。

由 FastAPI 构建的 LangServe 库 通过将 LangChain 对象的部署简化为 REST API 来丰富这一点，它提供了内置的 CORS 中间件，以确保我们的 API 可以安全地从不同域调用。

有关更深入的用例演示，可以通过访问 langchain-ai/langserve GitHub 存储库下的 examples 目录 来探索。

from fastapi import FastAPI


app = FastAPI(
    title="MinIO Agent API",
    version="1.0",
    description="A conversational agent facilitating data storage and retrieval with MinIO",
)

为了设置 CORS 标头，我们可以添加以下几行来增强我们的安全性

from fastapi.middleware.cors import CORSMiddleware


# Set all CORS enabled origins
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=["*"],
)

使用 LangServe 实现代理

现在我们已经完成了 `packages/agent.py`，我们可以将其导入并在我们的 `app/server.py` 脚本中使用 LangServe 库的 `add_route` 函数。

from packages.agent import agent_executor
from langserve import add_routes


add_routes(
   app,
   agent_executor.with_types(input_type=Input, output_type=Output).with_config(
       {"run_name": "agent"}
   ), path=”/invoke”
)

通过调用 `add_route(app, agent_executor(…), path="/invoke")`，我们在我们的服务器应用程序 (app) 中添加了一个路由，该路由将 `/invoke` 路径映射到 `agent_executor()` 函数。这允许在向 `/invoke` 端点发出请求时调用代理执行器。

通过此设置，服务器可以处理传入请求，将其传递给代理执行器，并将代理的响应返回给客户端。代理执行器使用定义的代理（包含提示模板、带有工具的语言模型和输出解析器）来处理用户输入并根据可用的工具生成相应的响应。

通过 Uvicorn 启动 LangServe 应用程序

为了启动 LangServe 应用程序，我们使用 Uvicorn 作为 ASGI 服务器，为我们的应用程序运行做好准备。这段代码片段至关重要，因为它激活了服务器，指定了应用程序访问点的通用主机和指定端口。

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

通过将此代码块嵌入到应用程序的主要入口点，我们确保在直接执行脚本时，Uvicorn 会接管，从而在预定义的主机和端口上启动我们的 FastAPI 应用程序。这种方法不仅简化了部署过程，而且还为在开发或生产环境中运行应用程序标记了一个清晰的入口点。

启动服务器应用程序

上面的代码展示了一种模块化方法，其中包括使用“langchain-cli”库、创建一个新的 langchain 应用程序，并将链逻辑保存到 `agent.py`，而 FastAPI 和 LangServe 实现保存到 `server.py`。

这是我们的最后一步，为了演示构建应用程序的目的，我们将我们的应用程序代码保存到 `server.py` 中。

运行我们的服务的简便方法是：

python server.py

此命令将运行应用程序，同时返回需要调试的任何日志或错误消息。

LangServe Playground

在 python 输出中，LangServe 日志将 `/invoke/playground` 识别为应用程序端点。我们现在可以访问 playground WebUI 以及我们的 API 的自动文档，这些文档可以通过访问我们 API 的 `/docs` 路径获得；这为我们提供了一种简化的测试和配置方法，包括每个应用程序功能的“尝试一下”按钮，以及我们可以从 WebUI 执行的预定义 cURL 请求。

因此，我们集成了 MinIO 的 LangChain 代理现在已熟练地转换为可部署的 API，可以为用户进行开发和扩展，其功能范围从批处理到实时交互。

LangServe API 的进一步使用

LangServe 应用程序启动并运行后，我们可以通过定位我们的端点并将其包装在 LangServe 的 `RemoteRunnable` 模块中，从 `server.py` 之外使用它

from langserve import RemoteRunnable


remote_runnable = RemoteRunnable("http://localhost:8000/<path>/")

添加

LangChain 在其库中拥有大量模块，展示了一个旨在赋能开发人员构建复杂 AI 驱动的应用程序的强大工具包。从复杂的链构建到与各种 AI 模型的无缝集成，LangChain 的模块化架构促进了广泛的功能，使在 AI 和机器学习领域创建高度可定制和高级解决方案成为可能。

使用 LangServe 开发 AI 管道

LangServe 不仅消除了部署 LangChain 应用程序的复杂性，而且还大大简化了这一过程。通过弥合开发和部署之间的差距，它确保了利用 MinIO 和 LangChain 的创新应用程序可以迅速从概念转化为现实，准备集成到更广泛的生态系统中，并提升用户体验。

通过我们在探索中涵盖的开发，我们已经看到 MinIO 与 LangChain 的无缝集成是完全可能的，以及 LangServe 如何在部署这些高级解决方案中发挥关键作用。随着我们继续探索不断发展的 AI 和 ML 领域，LangServe 这样的工具将在将尖端技术推向应用开发的最前沿方面发挥重要作用。

在 MinIO，我们对在这个技术丰富的时代，开发人员社区的创造力和潜力感到振奋。现在是协作和知识交流的最佳时机。我们渴望与您联系！加入我们的 MinIO Slack 频道，继续对话，共同攀登新的高峰。