🚰 管道函数：创建自定义"代理/模型"

欢迎来到在Open WebUI中创建管道的指南！将管道视为向Open WebUI添加新模型的方式。在本文档中，我们将分解什么是管道、它如何工作，以及如何创建自己的管道来为Open WebUI模型添加自定义逻辑和处理。我们将使用清晰的比喻并详细介绍每个细节，确保您有全面的理解。

管道简介

将Open WebUI想象为一个管道系统，数据通过管道和阀门流动。在这个类比中：

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

通过创建管道，您本质上是在Open WebUI框架内制作具有您想要的特定行为的自定义模型。

---

理解管道结构

让我们从管道的基本框架版本开始理解其结构：

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类是您定义自定义逻辑的地方。
• 目的：作为您插件的蓝图，确定它在Open WebUI中的行为方式。

阀门：配置您的管道

• 定义：Valves是Pipe内的嵌套类，继承自BaseModel。
• 目的：它包含在使用管道时持续存在的配置选项（参数）。
• 示例：在上面的代码中，MODEL_ID是一个配置选项，默认为空字符串。

比喻：将阀门视为真实管道系统上控制水流的旋钮。在您的管道中，阀门允许用户调整影响数据流动和处理方式的设置。

__init__方法

• 定义：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函数。这种结构确保清晰和一致性。

---

使用管道创建多个模型

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

以下是您可以做到的方法：

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代理管道

让我们深入一个实际示例，我们将创建一个将请求代理到OpenAI API的管道。这个管道将从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"错误：{e}"

详细分解

阀门配置

• 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. 准备负载：使用正确的模型ID更新body。
  4. 发出API请求：向OpenAI API的聊天完成端点发送POST请求。
  5. 处理流：如果stream为True，返回行的可迭代对象。
  6. 错误处理：捕获异常并返回错误消息。

扩展代理管道

您可以通过调整pipes和pipe函数中的API端点、头部和逻辑来修改此代理管道以支持其他服务提供商，如Anthropic、Perplexity等。

---

使用Open WebUI内部函数

有时，您可能希望在管道中利用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代码库或文档以获取最准确的函数签名和参数。
• 最佳实践：始终优雅地处理异常和错误，以确保流畅的用户体验。

---

常见问题

问题1：为什么要在Open WebUI中使用管道？

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

---

问题2：什么是阀门，为什么它们很重要？

答：阀门是管道的可配置参数。它们的功能类似于设置或控件，确定管道的操作方式。通过调整阀门，您可以更改管道的行为而无需修改底层代码。

---

问题3：我可以创建没有阀门的管道吗？

答：是的，如果您的管道不需要任何持久配置选项，您可以创建不定义Valves类的简单管道。但是，包含阀门是灵活性和未来可扩展性的良好实践。

---

问题4：使用API密钥时如何确保管道安全？

答：永远不要将API密钥等敏感信息硬编码到管道中。相反，使用阀门安全地输入和存储API密钥。确保您的代码适当地处理这些密钥，避免记录或暴露它们。

---

问题5：pipe和pipes函数有什么区别？

答：

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

• pipes函数：通过返回模型定义列表，允许您的管道代表多个模型。每个模型将在Open WebUI中单独出现。

---

问题6：如何在管道中处理错误？

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

---

问题7：我可以在管道中使用外部库吗？

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

---

问题8：如何测试我的管道？

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

---

问题9：组织管道代码有什么最佳实践吗？

答：是的，遵循这些指导原则：

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

---

问题10：我在哪里可以找到最新的Open WebUI文档？

答：访问官方Open WebUI存储库或文档站点以获取最新信息，包括函数签名、示例和迁移指南（如果发生任何更改）。

---

结论

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

记住：

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

编程愉快，享受使用管道扩展您的Open WebUI！