通过 LangServe 和 MinIO 集成轻松启动 LangChain API

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

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

• 使用 LangChain 创新 S3 存储桶检索：利用 LangChain 的 S3 Loaders 和 OpenAI API 来制作自定义对话式 AI 代理的演练，建立一种简化的数据管理方法。

  
  

• 使用 MinIO 为 LangChain 代理提供支持：深入研究如何利用 MinIO，展示 LangChain 如何与 OpenAI 的 GPT 相结合，开拓 AI 和 ML 数据处理的新领域。

  
  

基于这些见解，我们现在将重点转向语言服务，这是将 LangChain 应用程序从开发转变为部署的关键工具，简化了启动生产就绪 API 的过程。

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

语言服务作为开发人员的基石，消除了传统上与 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 应用程序模板或创建您自己的。

要创建一个新的 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中。

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

首先，使用文本编辑器打开 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()方法检查它是否已存在。如果 bucket 不存在，我们使用minio_client.make_bucket()创建它。如果 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 bucket 所需的参数，例如 bucket 名称、对象名称和文件的原始字节。该函数利用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 bucket 下载文件所需的信息，例如 bucket 名称、对象名称以及应保存文件的本地路径。该函数使用minio_client从指定的 bucket 中检索对象并将其保存到指定的本地路径。

 @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 bucket 中存在的对象。它需要一个带有bucket_name键的file_info字典。该函数使用minio_client检索指定 bucket 中的对象列表，并返回包含每个对象的对象键和大小的字典列表。

通过将这些功能封装为工具，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. 一个名为“agent_scratchpad”的MessagesPlaceholder ，用于存储代理的中间步骤和思考过程。

   
   

format_to_openai_tool_messages函数将代理的暂存器格式化为与 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 集成的应用程序提供了一条简化的途径，以 API 形式部署和管理我们的 LangChain 应用程序。选择 FastAPI 是因为它的性能和易用性，支持异步操作并自动生成 API 文档。

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

如需更深入/用例演示，请访问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 游乐场

在 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 驱动应用程序的多样化工具包。从复杂的链结构到与各种 AI 模型的无缝集成，LangChain 的模块化架构促进了广泛的功能，从而能够在 AI 和机器学习领域创建高度可定制和先进的解决方案。

使用 LangServe 开发 AI 管道

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

通过我们探索的开发，我们看到了 MinIO 与朗链绝对有可能，以及如何语言服务在部署这些先进解决方案方面发挥着关键作用。随着我们继续探索不断发展的 AI 和 ML 领域，像 LangServe 这样的工具将继续发挥重要作用，将尖端技术推向应用程序开发的前沿。

在 MinIO，我们为这个技术丰富的时代开发者社区的创造力和潜力而感到振奋。现在是合作和知识交流的最佳时机。我们渴望与您建立联系！加入我们MinIO Slack渠道继续对话并共同达到新的高度。