Skip to main content

🚰 Pipe 函数：创建自定义“代理/模型”

欢迎阅读 Open WebUI Pipes 创建指南！你可以将 Pipes 想象成向 Open WebUI 添加新模型的一种方式。在本文档中，我们将详细介绍什么是 Pipe、它是如何工作的，以及如何创建自己的 Pipe 来为 Open WebUI 模型添加自定义逻辑和处理。我们将使用清晰的比喻并深入每一个细节，以确保你有一个全面的理解。

Pipe 简介​

想象 Open WebUI 是一个供水系统，数据在管道和阀门中流动。在这个类比中：

  • Pipes (管道) 就像插件，让你引入新的数据流动路径，允许你注入自定义逻辑和处理。
  • Valves (配置阀门) 是管道中可配置的部分，用于控制数据如何流过它。

通过创建一个 Pipe，你实际上是在 Open WebUI 框架内构建一个具有特定行为的自定义模型。

理解 Pipe 结构​

让我们从一个基本的 Pipe 骨架开始，以了解其结构：

from pydantic import BaseModel, Field

class Pipe:
    class Valves(BaseModel):
        MODEL_ID: str = Field(default="")

    def __init__(self):
        self.valves = self.Valves()

    def pipe(self, body: dict):
        # 逻辑写在这里
        print(self.valves, body)  # 这将打印配置选项和输入的 body
        return "Hello, World!"

Pipe 类​

  • 定义：Pipe 类是你定义自定义逻辑的地方。
  • 用途：作为插件的蓝图，决定它在 Open WebUI 中的行为方式。

Valves：配置你的 Pipe​

  • 定义：Valves 是 Pipe 内部的一个嵌套类，继承自 BaseModel。
  • 用途：它包含在 Pipe 使用过程中持久化的配置选项（参数）。
  • 示例：在上面的代码中，MODEL_ID 是一个默认值为空字符串的配置选项。

比喻：将 Valves 想象成现实世界管道系统上的旋钮，用于控制水流。在你的 Pipe 中，Valves 允许用户调整影响数据流动和处理方式的设置。

__init__ 方法​

  • 定义：Pipe 类的构造方法。
  • 用途：初始化 Pipe 的状态并设置任何必要的组件。
  • 最佳实践：保持简单；主要在这里初始化 self.valves。
def __init__(self):
    self.valves = self.Valves()

pipe 函数​

  • 定义：核心函数，你的自定义逻辑所在之处。
  • 参数：
    • body：包含输入数据的字典。
  • 用途：使用自定义逻辑处理输入数据并返回结果。
def pipe(self, body: dict):
    # 逻辑写在这里
    print(self.valves, body)  # 这将打印配置选项和输入的 body
    return "Hello, World!"

注意：始终将 Valves 放在 Pipe 类的顶部，然后是 __init__，最后是 pipe 函数。这种结构确保了清晰度和一致性。

使用 Pipes 创建多个模型​

如果你想让你的 Pipe 在 Open WebUI 中创建多个模型怎么办？你可以通过在 Pipe 类中定义一个 pipes 函数或变量来实现。这种设置非正式地被称为 manifold (歧管)，允许你的 Pipe 代表多个模型。

以下是实现方法：

from pydantic import BaseModel, Field

class Pipe:
    class Valves(BaseModel):
        MODEL_ID: str = Field(default="")

    def __init__(self):
        self.valves = self.Valves()

    def pipes(self):
        return [
            {"id": "model_id_1", "name": "model_1"},
            {"id": "model_id_2", "name": "model_2"},
            {"id": "model_id_3", "name": "model_3"},
        ]

    def pipe(self, body: dict):
        # 逻辑写在这里
        print(self.valves, body)  # 打印配置选项和输入的 body
        model = body.get("model", "")
        return f"{model}: Hello, World!"

解释​

  • pipes 函数：

    • 返回一个字典列表。
    • 每个字典代表一个具有唯一 id 和 name 键的模型。
    • 这些模型将分别显示在 Open WebUI 模型选择器中。

  • 更新后的 pipe 函数：

    • 根据选择的模型处理输入。
    • 在本例中，它在返回的字符串中包含了模型名称。

示例：OpenAI 代理 Pipe (Proxy Pipe)​

让我们深入研究一个实际示例，我们将创建一个将请求代理到 OpenAI API 的 Pipe。此 Pipe 将从 OpenAI 获取可用模型，并允许用户通过 Open WebUI 与它们进行交互。

from pydantic import BaseModel, Field
import requests

class Pipe:
    class Valves(BaseModel):
        NAME_PREFIX: str = Field(
            default="OPENAI/",
            description="添加到模型名称前的前缀。",
        )
        OPENAI_API_BASE_URL: str = Field(
            default="https://api.openai.com/v1",
            description="访问 OpenAI API 端点的基础 URL。",
        )
        OPENAI_API_KEY: str = Field(
            default="",
            description="用于验证 OpenAI API 请求的 API 密钥。",
        )

    def __init__(self):
        self.valves = self.Valves()

    def pipes(self):
        if self.valves.OPENAI_API_KEY:
            try:
                headers = {
                    "Authorization": f"Bearer {self.valves.OPENAI_API_KEY}",
                    "Content-Type": "application/json",
                }

                r = requests.get(
                    f"{self.valves.OPENAI_API_BASE_URL}/models", headers=headers
                )
                models = r.json()
                return [
                    {
                        "id": model["id"],
                        "name": f'{self.valves.NAME_PREFIX}{model.get("name", model["id"])}',
                    }
                    for model in models["data"]
                    if "gpt" in model["id"]
                ]

            except Exception as e:
                return [
                    {
                        "id": "error",
                        "name": "获取模型出错。请检查您的 API 密钥。",
                    },
                ]
        else:
            return [
                {
                    "id": "error",
                    "name": "未提供 API 密钥。",
                },
            ]

    def pipe(self, body: dict, __user__: dict):
        print(f"pipe:{__name__}")
        headers = {
            "Authorization": f"Bearer {self.valves.OPENAI_API_KEY}",
            "Content-Type": "application/json",
        }

        # 从模型名称中提取模型 id
        model_id = body["model"][body["model"].find(".") + 1 :]

        # 更新 body 中的模型 id
        payload = {**body, "model": model_id}
        try:
            r = requests.post(
                url=f"{self.valves.OPENAI_API_BASE_URL}/chat/completions",
                json=payload,
                headers=headers,
                stream=True,
            )

            r.raise_for_status()

            if body.get("stream", False):
                return r.iter_lines()
            else:
                return r.json()
        except Exception as e:
            return f"Error: {e}"

详细解析​

Valves 配置​

  • NAME_PREFIX：
    • 为 Open WebUI 中显示的模型名称添加前缀。
    • 默认值："OPENAI/"。
  • OPENAI_API_BASE_URL：
    • 指定 OpenAI API 的基础 URL。
    • 默认值："https://api.openai.com/v1"。
  • OPENAI_API_KEY：
    • 你的 OpenAI API 密钥，用于身份验证。
    • 默认值：""（空字符串；必须提供）。

pipes 函数​

  • 用途：获取可用的 OpenAI 模型并使其在 Open WebUI 中可访问。

  • 过程：

    1. 检查 API 密钥：确保提供了 API 密钥。
    2. 获取模型：向 OpenAI API 发送 GET 请求以检索可用模型。
    3. 过滤模型：返回 id 中包含 "gpt" 的模型。
    4. 错误处理：如果出现问题，返回错误消息。

  • 返回格式：包含每个模型 id 和 name 的字典列表。

pipe 函数​

  • 用途：处理对所选 OpenAI 模型的请求并返回响应。

  • 参数：

    • body：包含请求数据。
    • __user__：包含用户信息（在本例中未使用，但对于身份验证或日志记录很有用）。

  • 过程：

    1. 准备请求头：设置包含 API 密钥和内容类型的请求头。
    2. 提取模型 ID：从所选模型名称中提取实际的模型 ID。
    3. 准备 Payload：使用正确的模型 ID 更新 body。
    4. 发起 API 请求：向 OpenAI API 的聊天补全端点发送 POST 请求。
    5. 处理流式传输：如果 stream 为 True，则返回行的可迭代对象。
    6. 错误处理：捕获异常并返回错误消息。

扩展代理 Pipe​

你可以修改此代理 Pipe 以支持其他服务提供商（如 Anthropic、Perplexity 等），只需调整 pipes 和 pipe 函数中的 API 端点、请求头和逻辑即可。

使用 Open WebUI 内部函数​

有时，你可能希望在 Pipe 中利用 Open WebUI 的内部函数。你可以直接从 open_webui 包导入这些函数。请记住，虽然可能性较小，但内部函数可能会为了优化目的而发生变化，因此请始终参考最新文档。

以下是如何使用 Open WebUI 内部函数的方法：

from pydantic import BaseModel, Field
from fastapi import Request

from open_webui.models.users import Users
from open_webui.utils.chat import generate_chat_completion

class Pipe:
    def __init__(self):
        pass

    async def pipe(
        self,
        body: dict,
        __user__: dict,
        __request__: Request,
    ) -> str:
        # 使用更新签名的统一端点
        user = Users.get_user_by_id(__user__["id"])
        body["model"] = "llama3.2:latest"
        return await generate_chat_completion(__request__, body, user)

解释​

  • 导入：

    • 来自 open_webui.models.users 的 Users：用于获取用户信息。
    • 来自 open_webui.utils.chat 的 generate_chat_completion：使用内部逻辑生成聊天补全。

  • 异步 pipe 函数：

    • 参数：
      • body：模型的输入数据。
      • __user__：包含用户信息的字典。
      • __request__：FastAPI 的请求对象（generate_chat_completion 所需）。
    • 过程：
      1. 获取用户对象：使用用户 ID 检索用户对象。
      2. 设置模型：指定要使用的模型。
      3. 生成补全：调用 generate_chat_completion 来处理输入并产生输出。

重要提示​

  • 函数签名：参考最新的 Open WebUI 代码库或文档，以获取最准确的函数签名和参数。
  • 最佳实践：始终优雅地处理异常和错误，以确保流畅的用户体验。

常见问题解答​

Q1: 为什么我应该在 Open WebUI 中使用 Pipes？​

答：Pipes 允许你向 Open WebUI 添加具有自定义逻辑和处理的新“模型”。这是一个灵活的插件系统，让你能够集成外部 API、自定义模型行为并创建创新功能，而无需更改核心代码库。

Q2: 什么是 Valves，为什么它们很重要？​

答：Valves 是 Pipe 的可配置参数。它们的功能类似于决定 Pipe 如何运行的设置或控制项。通过调整 Valves，你可以在不修改底层代码的情况下更改 Pipe 的行为。

Q3: 我可以创建一个没有 Valves 的 Pipe 吗？​

答：可以，如果你的 Pipe 不需要任何持久的配置选项，你可以创建一个不定义 Valves 类的简单 Pipe。然而，为了灵活性和未来的可扩展性，包含 Valves 是一个很好的做法。

Q4: 在使用 API 密钥时，如何确保我的 Pipe 是安全的？​

答：永远不要将 API 密钥等敏感信息硬编码到你的 Pipe 中。相反，使用 Valves 来安全地输入和存储 API 密钥。确保你的代码妥善处理这些密钥，并避免记录或公开它们。

Q5: pipe 和 pipes 函数有什么区别？​

答：

  • pipe 函数：处理输入数据并生成输出的主要函数。它处理单个模型的逻辑。

  • pipes 函数：允许你的 Pipe 通过返回模型定义列表来代表多个模型。每个模型都将分别显示在 Open WebUI 中。

Q6: 我如何在 Pipe 中处理错误？​

答：在 pipe 和 pipes 函数中使用 try-except 块来捕获异常。返回有意义的错误消息或优雅地处理错误，以确保用户了解出了什么问题。

Q7: 我可以在我的 Pipe 中使用外部库吗？​

答：可以，你可以根据需要导入和使用外部库。确保在你的环境中正确安装和管理了任何依赖项。

Q8: 我如何测试我的 Pipe？​

答：通过在开发环境中运行 Open WebUI 并在界面中选择你的自定义模型来测试你的 Pipe。验证你的 Pipe 在各种输入和配置下是否按预期运行。

Q9: 组织 Pipe 代码有哪些最佳实践？​

答：是的，请遵循以下准则：

  • 将 Valves 放在 Pipe 类的顶部。
  • 在 __init__ 方法中初始化变量，主要是 self.valves。
  • 将 pipe 函数放在 __init__ 方法之后。
  • 使用清晰且描述性的变量名称。
  • 为代码添加注释以提高清晰度。

Q10: 在哪里可以找到最新的 Open WebUI 文档？​

答：访问 Open WebUI 官方仓库或文档网站以获取最新信息，包括函数签名、示例以及任何更改时的迁移指南。

总结​

到目前为止，你应该对如何在 Open WebUI 中创建和使用 Pipes 有了深入的了解。Pipes 提供了一种强大的方式来扩展和自定义 Open WebUI 的功能，以满足你的特定需求。无论你是集成外部 API、添加新模型还是注入复杂的逻辑，Pipes 都能提供实现这些目标的灵活性。

请记住：

  • 在你的 Pipe 类中使用清晰且一致的结构。
  • 利用 Valves 进行可配置选项。
  • 优雅地处理错误以改善用户体验。
  • 查阅最新文档以了解任何更新或更改。

祝你编码愉快，尽情使用 Pipes 扩展你的 Open WebUI 吧！