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

我们在LangChain的创新世界中的旅程揭示了其在转变数据管理和应用程序功能方面的强大能力。

通过之前的讨论，我们深入探讨了几个主题，同时探索了LangChain的复杂功能。在本文中，我们将以“使用 MinIO 赋能 Langchain 代理”中涵盖的概念为基础，扩展 MinIO 代理的功能，以封装其他功能并通过 LangServe 部署自定义代理。

**使用 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

开发 LangChain MinIO 代理以使用 LangServe 进行部署

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

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

**注意：**阅读之前的出版物“MinIO Langchain Tool”将为使用 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-exception块进行错误处理，以捕获和打印在此过程中可能发生的任何 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的Runnable方法

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 . MessagesPlaceholder 一个名为“agent_scratchpad”，用于存储代理的中间步骤和思维过程。

该 format_to_openai_tool_messages 函数将代理的暂存器格式化为 OpenAI 工具的兼容格式，而 OpenAIToolsAgentOutputParser 类将模型的响应解析为可由代理解释的结构化格式。

AIMessage and 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 可以从不同的域安全地调用。

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


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 游乐场 LangServe

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

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

LangServe API 的进一步使用

随着LangServe应用程序的启动和运行，我们可以从外部 server.py 使用它，方法是以我们的端点为目标，并将其封装在Langserve的 RemoteRunnable 模块中：


from langserve import RemoteRunnable


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

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

使用 LangServe 开发 AI 管道

LangServe 不仅揭开了 LangChain 应用程序的神秘面纱，而且大大简化了部署 LangChain 应用程序的过程。通过弥合开发和部署之间的差距，它确保利用MinIO和LangChain的创新应用程序能够迅速从概念变为现实，准备好集成到更广泛的生态系统中并增强用户体验。

通过我们探索中涵盖的开发，我们已经看到MinIO与LangChain的无缝集成是绝对可能的，以及LangServe如何在部署这些高级解决方案方面发挥关键作用。随着我们继续驾驭不断发展的人工智能和机器学习领域，像LangServe这样的工具将继续在将尖端技术带到应用程序开发的最前沿方面发挥重要作用。

在MinIO，在这个技术丰富的时代，开发者社区的创造力和潜力让我们充满活力。