🧩 什么是 Pipelines？

Pipelines 是 Open WebUI 的插件框架，允许你通过 Python 代码扩展和自定义 AI 功能。它作为一个中间层，可以在用户请求和 AI 响应之间插入自定义逻辑。

🌟 核心优势

  • 无需修改 Open WebUI 源码
  • 支持任意 Python 库
  • 可热插拔，随时启用/禁用
  • 完全控制请求和响应流程

🚀 安装 Pipelines

bash 
# 克隆 Pipelines 仓库
git clone https://github.com/open-webui/pipelines.git
cd pipelines

# 安装依赖
pip install -r requirements.txt

# 启动 Pipelines 服务
python server.py --host 0.0.0.0 --port 9099

然后在 Open WebUI 中配置：

环境变量 
# Open WebUI 连接 Pipelines
OPENAI_API_BASE_URL=http://localhost:9099/v1

📦 内置插件

🔧

Function Calling

让 AI 调用外部函数和 API

⏱️

速率限制

控制用户请求频率

🌐

实时翻译

集成 LibreTranslate

🛡️

内容过滤

过滤敏感和有害内容

📊

使用监控

集成 Langfuse 追踪

🎭

角色扮演

自定义 AI 人格

💻 开发自定义插件

插件模板

Python 
from pydantic import BaseModel
from typing import Optional

class Pipeline:
    class Valves(BaseModel):
        # 插件配置项
        pass

    def __init__(self):
        self.name = "My Plugin"
        self.valves = self.Valves()

    async def on_startup(self):
        # 服务启动时执行
        print(f"Plugin {self.name} started")

    async def on_shutdown(self):
        # 服务关闭时执行
        print(f"Plugin {self.name} stopped")

    async def inlet(self, body: dict, user: dict) -> dict:
        # 处理用户输入
        return body

    async def outlet(self, body: dict, user: dict) -> dict:
        # 处理 AI 响应
        return body

🔗 插件示例

速率限制插件

Python 
from datetime import datetime, timedelta
from collections import defaultdict

class Pipeline:
    def __init__(self):
        self.name = "Rate Limiter"
        self.requests = defaultdict(list)
        self.limit = 10  # 每分钟 10 次

    async def inlet(self, body: dict, user: dict) -> dict:
        user_id = user.get("id")
        now = datetime.now()
        
        # 清理过期记录
        self.requests[user_id] = [
            t for t in self.requests[user_id]
            if now - t < timedelta(minutes=1)
        ]
        
        # 检查限制
        if len(self.requests[user_id]) >= self.limit:
            raise Exception("请求过于频繁，请稍后再试")
        
        self.requests[user_id].append(now)
        return body