🚰 管道函数：创建自定义“模型 / Agent”

⚠️ 关键安全警告

Pipe Functions 会在你的服务器上执行任意 Python 代码。 Function 创建权限仅限管理员。只应从可信来源安装，导入前必须审阅代码。恶意 Function 可能访问文件系统、外传数据，甚至接管整个系统。完整背景见 Plugin Security Warning。

Pipe 可以理解成：把一个新的“模型入口”接入到 OPL 数据空间。它不一定真的是某个 LLM，也可以是代理、路由器、数据库接口、搜索器或任意自定义逻辑。

为未来兼容性优先使用 async

Pipe 函数通常都应该定义为 async。后端正逐步向全异步执行迁移；同步函数未来可能阻塞执行甚至出现兼容性问题。

Pipe 的基本结构

from pydantic import BaseModel, Field

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

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

    async def pipe(self, body: dict):
        print(self.valves, body)
        return "Hello, World!"

Pipe 类

这里定义你的核心逻辑，是整个自定义模型的入口。

Valves

Valves 是嵌套在 Pipe 中的配置类，通常继承自 BaseModel。它用来保存会持续生效的配置项，例如 API 地址、模型 ID、前缀、密钥等。

可以把它理解成一组“阀门旋钮”，决定数据如何流过这条 Pipe。

__init__

通常只需要做简单初始化，最常见的是：

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

pipe()

pipe() 是最核心的方法，负责接收请求体、执行你的逻辑并返回结果。

一个 Pipe 暴露多个模型

如果你想让一个 Pipe 在 OPL 数据空间里呈现成多个模型，可以再定义 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"},
        ]

    async def pipe(self, body: dict):
        model = body.get("model", "")
        return f"{model}: Hello, World!"

pipes() 返回的每个 {id, name} 都会出现在模型选择器里。这种做法常被称为 manifold。

示例：OpenAI Proxy Pipe

下面这个例子演示如何把 OpenAI API 包装成一个 Pipe：

from pydantic import BaseModel, Field
import httpx


class Pipe:
    class Valves(BaseModel):
        NAME_PREFIX: str = Field(
            default="OPENAI/",
            description="Prefix to be added before model names.",
        )
        OPENAI_API_BASE_URL: str = Field(
            default="https://api.openai.com/v1",
            description="Base URL for accessing OpenAI API endpoints.",
        )
        OPENAI_API_KEY: str = Field(
            default="",
            description="API key for authenticating requests to the OpenAI API.",
        )

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

    async def pipes(self):
        if not self.valves.OPENAI_API_KEY:
            return [{"id": "error", "name": "API Key not provided."}]

        headers = {
            "Authorization": f"Bearer {self.valves.OPENAI_API_KEY}",
            "Content-Type": "application/json",
        }

        try:
            async with httpx.AsyncClient() as client:
                r = await client.get(
                    f"{self.valves.OPENAI_API_BASE_URL}/models", headers=headers
                )
                r.raise_for_status()
                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:
            return [
                {
                    "id": "error",
                    "name": "Error fetching models. Please check your API Key.",
                }
            ]

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

        model_id = body["model"][body["model"].find(".") + 1 :]
        payload = {**body, "model": model_id}
        url = f"{self.valves.OPENAI_API_BASE_URL}/chat/completions"

        try:
            if body.get("stream", False):
                async def event_stream():
                    async with httpx.AsyncClient(timeout=None) as client:
                        async with client.stream(
                            "POST", url, json=payload, headers=headers
                        ) as r:
                            r.raise_for_status()
                            async for line in r.aiter_lines():
                                yield line

                return event_stream()

            async with httpx.AsyncClient(timeout=None) as client:
                r = await client.post(url, json=payload, headers=headers)
                r.raise_for_status()
                return r.json()
        except Exception as e:
            return f"Error: {e}"

优先使用异步 HTTP 客户端

这里使用的是 httpx.AsyncClient，而不是 requests。原因很简单：pipes() 和 pipe() 都运行在 OPL 数据空间的异步事件循环里。若你在 async def 中调用同步 HTTP 库，会阻塞整个事件循环，影响实例上的其他请求。

这个 Pipe 的工作流程：

1. 在 pipes() 里读取 OpenAI 可用模型
2. 把满足条件的模型列表暴露给 OPL 数据空间
3. 在 pipe() 里把用户请求转发到 OpenAI Chat Completions API
4. 根据 stream 决定是返回流式生成器，还是一次性 JSON 结果

自包含 Agent 的注意事项

自包含 agent 不要发出 delta.tool_calls

如果你的 Pipe 自己内部就实现了 agent 规划与工具执行，而不是让 OPL 数据空间再替你执行工具，那么不要在流式输出里发 delta.tool_calls。否则 OPL 数据空间会把它理解为“请再次替我执行这个工具”，从而触发重复执行循环。

这类场景应改为把内部工具调用渲染成 <details type="tool_calls"> 内容块。详见 Pipes → Self-contained agents and delta.tool_calls。

在 Pipe 中调用 OPL 数据空间内部函数

有时你会希望直接复用 OPL 数据空间内部能力，例如统一聊天补全逻辑：

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 = await Users.get_user_by_id(__user__["id"])
        body["model"] = "llama3.2:latest"
        return await generate_chat_completion(__request__, body, user)

这种做法很有用，但也意味着你更依赖内部 API 细节。升级 OPL 数据空间时，要特别注意签名是否变化。

常见问题

为什么要用 Pipe？

因为它让你能以“模型”的形式接入任何自定义逻辑，而不用改 OPL 数据空间核心代码。

Valves 有什么用？

它是 Pipe 的持久化配置层。API Key、基础 URL、行为开关，都很适合放在这里。

pipe() 和 pipes() 的区别是什么？

• pipe()：处理一次实际请求
• pipes()：声明这个 Pipe 对外暴露哪些模型入口

Pipe 可以不用 Valves 吗？

可以。如果没有需要持久化的配置项，可以省略；但保留 Valves 通常更利于后续扩展。

如何保证安全？

不要硬编码密钥，不要记录敏感信息日志，并且只允许可信管理员创建或导入 Pipe。