插件框架:pipe 自定义模型与 filter 拦截函数

30 秒导读: Open WebUI 允许用户在后台粘贴一段纯 Python 代码,就地扩展这个聊天智能体——不用改源码、不用重启。两种插件:Pipe 类把自己伪装成一个"模型"接入模型下拉框(自定义后端);Filter 类在请求进出的三个时刻拦截并改写数据。本章讲清"插件是什么、怎么被加载进来、怎么扩展",拦截的精确调用时机只做交叉引用,细节在 02 与 04。

本章属于 Open WebUI 系列。其它章见 index:01 请求生命周期、02 入口装配、03 工具系统、04 智能体循环与出口。

1. 这是什么(零基础也能懂)

一句话定义: 插件框架 = 让用户用一段 Python 代码给这个智能体加一个"新模型"或"一道拦截关卡",而无需碰主程序。

解决什么问题 / 给谁用。 假设你想:

接一个官方没内置的模型服务(比如某个私有网关、某个新出的 API);
或者在每条消息发给模型之前偷偷加一句系统提示、脱敏用户输入;
或者在模型答完之后给回复加个免责声明、翻译一遍。

改主程序太重。Open WebUI 的答案是:把这些扩展点开放成用户可编辑的 Python 源码,存进数据库,运行时动态加载执行。

它能做什么(两类插件):

插件类型	你要写的类	扮演什么角色	白话
Pipe	`Pipe`	一个"自定义模型"后端	模型下拉框里多出一个你的模型;选它,聊天请求就走进你的 `pipe()` 函数
Filter	`Filter`	进出请求的拦截器	在请求送模型前(`inlet`)、流式分片经过时(`stream`)、答完之后(`outlet`)改写数据
Action	`Action`	消息下方的按钮动作	(本章不展开,加载机制同上,见 `load_function_module_by_id`)

用起来什么样。 用户在管理后台粘贴的,就是一段带类的普通 Python。一个最小 Pipe:

# 示意,非源码 —— 这就是用户粘进后台的全部内容
"""
title: Echo Model
requirements: 
"""
from pydantic import BaseModel

class Pipe:
    class Valves(BaseModel):     # 可配置项(见 §6)
        prefix: str = "you said: "

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

    def pipe(self, body: dict):  # 核心:收到聊天请求 body,返回回复
        last = body["messages"][-1]["content"]
        return self.valves.prefix + last   # 直接返回字符串 = 一条完整回复

保存后,模型列表里就会出现 "Echo Model";选它发消息,回复永远是 you said: <你说的>。没有注册表、没有插件清单——一个 Pipe 类就是一个模型。

一句话直觉/类比。 把插件想成给智能体的热插拔配件:Pipe 是"换一个大脑"(模型后端),Filter 是"在耳朵和嘴巴上各装一个滤网"(输入输出拦截)。配件本身是一张纸(源码字符串),用的时候才被"读活"(exec 成模块)。

2. 顶层全景(它大概怎么转)

插件的生命有三段:存 → 加载 → 接线。

用户在后台粘贴 Python 源码
         │  存进 function 表的 content 列(纯文本)
         ▼
┌──────────────────────────────────────────────┐
│  models/functions.py                          │
│  Function 表: id / type / content / valves /   │
│               is_active / is_global           │
└──────────────────────────────────────────────┘
         │  运行时按需取出 content 字符串
         ▼
┌──────────────────────────────────────────────┐
│  utils/plugin.py  —— 把"字符串"读成"活模块"    │
│  replace_imports → exec 进临时模块 →           │
│  找到 Pipe / Filter / Action 类并实例化         │
│  (结果缓存在 request.app.state.FUNCTIONS)      │
└──────────────────────────────────────────────┘
         │
         ├─ 是 Pipe ─────────────────────────────────┐
         │                                           ▼
         │                          ┌────────────────────────────────┐
         │                          │ functions.py                   │
         │                          │ get_function_models() 把每个    │
         │                          │ pipe 变成一个"模型"塞进模型列表 │
         │                          │ generate_function_chat_         │
         │                          │ completion() 执行 pipe()        │
         │                          └────────────────────────────────┘
         │                                           ▲
         │        utils/chat.py: model.get('pipe') 分支路由到这里
         │
         └─ 是 Filter ───────────────────────────────┐
                                                     ▼
                                    ┌────────────────────────────────┐
                                    │ utils/filter.py                │
                                    │ get_sorted_filter_ids() 排序    │
                                    │ process_filter_functions() 执行 │
                                    │ 被 middleware.py 在 inlet /     │
                                    │ outlet / stream 三处接线调用     │
                                    └────────────────────────────────┘

部件一句话职责:

部件	干什么	在哪个文件
`Function` 表	存插件源码、类型、valves、激活状态	`models/functions.py:18`
加载器	把源码字符串 `exec` 成模块并缓存	`utils/plugin.py:250` `load_function_module_by_id`
pipe→模型	把每个 pipe 注册成一个可选"模型"	`functions.py:71` `get_function_models`
pipe 执行	路由进来后调用 `pipe()`,处理流式/非流式	`functions.py:147` `generate_function_chat_completion`
pipe 路由	判断选中的模型是不是 pipe,是就转过去	`utils/chat.py:274` `model.get('pipe')` 分支
filter 排序	收集+按优先级排出该跑哪些 filter	`utils/filter.py:21` `get_sorted_filter_ids`
filter 执行	逐个调用 filter 的 inlet/outlet/stream	`utils/filter.py:66` `process_filter_functions`

主线走一遍(高层)。 用户选了模型 A 发消息 → 请求进入 generate_chat_completion(utils/chat.py:152)→ 如果 A 是 pipe,走 model.get('pipe') 分支交给 generate_function_chat_completion;如果 A 是普通模型,则在送模型前后由 middleware 调 process_filter_functions 跑一圈 filter。pipe 决定"谁来答",filter 决定"进出时改什么"。

3. Pipe 函数:自定义"模型"后端

3.1 它要解决的小问题

官方内置的模型来源就那几种(OpenAI 兼容、Ollama…)。用户想接一个新后端,又不想等官方支持,也不想 fork 主程序。Pipe 让"一个模型 = 一个 Python 类",把接入成本压到"写一个函数"。

3.2 思路:让 pipe"假装"成模型混进模型列表

模型列表不是硬编码的。启动/刷新时,get_function_models 把每个激活的 pipe 翻译成一条模型记录塞进列表——带上一个 pipe 标记,后续路由就靠这个标记认出它。

看它怎么造一条模型记录(单 pipe 情况):

# functions.py:122-139  get_function_models 内部
pipe_flag = {'type': 'pipe'}
pipe_models.append(
    {
        'id': pipe.id,
        'name': pipe.name,
        'object': 'model',
        'created': pipe.created_at,
        'owned_by': 'openai',
        'pipe': pipe_flag,          # ← 关键:带 'pipe' 键 = 这是个 pipe 模型
        'has_user_valves': has_user_valves,
    }
)

get_function_models(functions.py:71)先 Functions.get_functions_by_type('pipe', active_only=True) 取出所有激活的 pipe,再逐个转成模型记录。它被 utils/models.py:65 在装配全量模型表时调用,于是 pipe 就和真模型并排出现在下拉框里。

3.3 Manifold:一个 pipe 暴露多个模型

有时一个后端网关背后有好几个模型(如一个 API key 能访问 gpt-4、gpt-4o…)。Open WebUI 支持 manifold:如果你的模块里有 pipes(注意复数,可以是列表/同步函数/异步函数),框架会把它展开成多条模型记录。

# functions.py:84-121(节选)manifold 展开
if hasattr(function_module, 'pipes'):
    ...
    if callable(function_module.pipes):
        if asyncio.iscoroutinefunction(function_module.pipes):
            sub_pipes = await function_module.pipes()   # 异步函数
        else:
            sub_pipes = function_module.pipes()          # 同步函数
    else:
        sub_pipes = function_module.pipes                # 直接是列表
    for p in sub_pipes:
        sub_pipe_id = f'{pipe.id}.{p["id"]}'             # id 拼成 "父.子"

子模型 id 长成 父pipe.子id 的形式(如 my_gateway.gpt-4)。这个"点号拼接"很关键——路由时要靠它还原出真正该加载哪个 pipe 模块。

3.4 路由:选中 pipe 后,请求怎么走进 pipe()

普通聊天请求都经过 generate_chat_completion(utils/chat.py:152)。它检查选中的模型有没有 pipe 标记:

# utils/chat.py:274-276
if model.get('pipe'):
    # 这是唯一走这个函数的路径,已经在 bypass filter
    return await generate_function_chat_completion(request, form_data, user=user, models=models)

命中就把整个请求交给 generate_function_chat_completion(functions.py:147)。这个函数干三件事:

还原 pipe id。 manifold 的模型 id 是 父.子,得先切回父 id 才能找到模块:

# functions.py:180-184  get_pipe_id
def get_pipe_id(form_data: dict) -> str:
    pipe_id = form_data['model']
    if '.' in pipe_id:
        pipe_id, _ = pipe_id.split('.', 1)   # "my_gateway.gpt-4" → "my_gateway"
    return pipe_id

按签名注入参数。 pipe 函数想要什么,框架就给什么——用 inspect.signature 读出 pipe() 声明了哪些形参,只注入它要的那些(约定俗成的 __user__、__event_emitter__、__request__ 等双下划线参数):
```
# functions.py:193-194  get_function_params
sig = inspect.signature(function_module.pipe)
params = {'body': form_data} | {k: v for k, v in extra_params.items() if k in sig.parameters}
```
可注入的"上下文"在 extra_params(functions.py:252-266)里备齐:__event_emitter__ / __event_call__(向前端推事件)、__files__、__user__、__oauth_token__、__request__、__tools__ 等。你的 pipe() 只声明需要的,别的自动忽略——这让插件签名可以很简单。
执行并归一化返回值。 pipe 可以是同步或异步(execute_pipe,functions.py:148);返回值支持多种形态,框架统一包装成 OpenAI 风格的响应。

返回值支持四种形态(这是 pipe 写起来自由的关键):

pipe 返回	框架怎么处理	引用
`str` 字符串	当成一条完整回复,包成 chat completion	`functions.py:309-311`(流式)/ `340-341`(非流式)
`dict`	当成已成型的响应/分片直接透传	`functions.py:300-301`
`Iterator` / `Generator`(同步生成器)	逐行 `process_line` 转成 SSE 分片	`functions.py:313-315`
`AsyncGenerator`(异步生成器)	逐行异步 `process_line` 转 SSE	`functions.py:317-319`
`StreamingResponse`	body 直接原样转发	`functions.py:296-299`

流式时,每行都过 process_line(functions.py:162):已经是 data: 开头的 SSE 就原样加换行,否则用 openai_chat_chunk_message_template 包成标准分片。收尾再补一个 finish_reason: stop 和 data: [DONE](functions.py:321-324)。

一句话记住 pipe: 你写一个 pipe(body, ...),返回字符串/字典/生成器随你;框架负责把它伪装成模型、路由进来、把返回值翻译成标准聊天流。

4. Filter 函数:inlet / outlet / stream 三段拦截

4.1 它要解决的小问题

Pipe 是"换后端";Filter 是"不换后端,但想在请求进出时插一脚"。典型需求:发给模型前改 payload(加系统提示、脱敏)、流式分片经过时改文字、模型答完后改最终回复。

4.2 思路:一个 Filter 类,最多三个钩子

你在 Filter 类里可选地实现三个方法,框架在三个时刻分别调它们:

钩子方法	触发时机	拿到/返回什么	精确接线点(交叉引用)
`inlet`	请求送模型之前	收/返 `body`(整个请求)	见 02 入口装配;`middleware.py:2539`
`stream`	流式分片逐个经过时	收/返 `event`(单个分片)	`middleware.py:5229`
`outlet`	模型答完之后	收/返 `body`(含最终 messages)	见 04 智能体循环与出口;`outlet_filter_handler` @ `middleware.py:3274`,执行在 `3378`

三个钩子都是可选的——只实现你需要的那个;没实现的钩子会被跳过(handler = getattr(function_module, filter_type, None),utils/filter.py:77,取不到就 continue)。

4.3 谁来跑、按什么顺序:get_sorted_filter_ids

一次请求可能命中多个 filter,得先决定跑哪些、什么顺序。get_sorted_filter_ids(utils/filter.py:21)负责这件事:

先收集。 两个来源合并——全局 filter(is_global=True,对每次聊天都生效)加上该模型绑定的 filter(模型 info.meta.filterIds):

# utils/filter.py:33-36
filter_ids = [function.id for function in await Functions.get_global_filter_functions()]
if 'info' in model and 'meta' in model['info']:
    filter_ids.extend(model['info']['meta'].get('filterIds', []))
    filter_ids = list(set(filter_ids))   # 去重

再筛激活/开关。 只保留 type='filter' 且激活的;如果某 filter 声明了 toggle(用户可在 UI 手动开关),还要看它是否在本次 enabled_filter_ids 里(get_active_status,utils/filter.py:39-45)。

最后排序。 按每个 filter 的 Valves.priority 排,priority 相同再按 id 排,保证顺序稳定:

# utils/filter.py:56-59
priorities = {}
for fid in filter_ids:
    priorities[fid] = await get_priority(fid)   # 读 valves.priority,默认 0
filter_ids.sort(key=lambda fid: (priorities.get(fid, 0), fid))

注意:这里为什么用 for 循环预先算 priority、再排序,而不是直接把 async 函数塞进 sort(key=...)?因为排序键必须同步,而读 priority 要 await 数据库。代码里两处注释点破了这个坑(utils/filter.py:47、55:"async functions can't be used in sort keys")。

4.4 怎么跑:process_filter_functions

get_sorted_filter_ids 给出有序 id,Functions.get_functions_by_ids 取出函数对象,再交给 process_filter_functions(utils/filter.py:66)逐个执行。它对每个 filter 做的事,和 pipe 的参数注入是同一套思路:

取对应钩子。 filter_type 是 'inlet'/'outlet'/'stream' 之一,用它从模块上取同名方法:

# utils/filter.py:75-79
function_module = await get_function_module(request, filter_id, load_from_db=(filter_type != 'stream'))
handler = getattr(function_module, filter_type, None)   # inlet/outlet/stream
if not handler:
    continue                                             # 没实现这个钩子就跳过

细节:load_from_db=(filter_type != 'stream')——inlet/outlet 每次从数据库重读源码(用户可能刚改过),而 stream 走缓存(性能敏感,分片逐个来,不能每片都读库)。这个"读库 vs 读缓存"的开关在加载器里落实,见 §5.4。

按签名注入参数(和 pipe 同理)。 inlet/outlet 传 body,stream 传 event;再把 extra_params 里 filter 声明了的注入进去:

# utils/filter.py:92-105
sig = inspect.signature(handler)
params = {'body': form_data}
if filter_type == 'stream':
    params = {'event': form_data}     # stream 钩子拿的是单个分片,叫 event
params = params | {k: v for k, v in {**extra_params, '__id__': filter_id}.items() if k in sig.parameters}

执行并把返回值当作新的 form_data。 关键:filter 是串成链的——上一个的输出是下一个的输入:

# utils/filter.py:118-121
if inspect.iscoroutinefunction(handler):
    form_data = await handler(**params)   # 返回值覆盖 form_data,传给下一个 filter
else:
    form_data = handler(**params)

一个特殊能力:file_handler。 如果某 inlet filter 声明了 file_handler = True,表示"文件我自己处理了,框架别再把文件塞进上下文",于是执行完后会把 form_data 里的 files 删掉(utils/filter.py:82-83 记录标记,128-133 清理):

# utils/filter.py:127-133
if skip_files:
    if 'files' in form_data.get('metadata', {}):
        del form_data['metadata']['files']
    if 'files' in form_data:
        del form_data['files']

4.5 三处接线在哪(只交叉引用,不展开)

process_filter_functions 在 middleware 里被调用三次,分别对应三个钩子——具体上下文与时机在别章讲:

一条聊天请求
   │
   ├─ inlet   filter  ← middleware.py:2539   详见 [02 入口装配]
   │
  (送模型 / pipe 出流)
   │
   ├─ stream  filter  ← middleware.py:5229   (逐分片改写)
   │
   └─ outlet  filter  ← outlet_filter_handler @ middleware.py:3274,执行 3378   详见 [04 出口处理]

本章到此为止:你只需知道 filter 是"三段可选钩子 + 优先级串链",调用时机的完整故事在 02 和 04。

5. 插件加载与沙箱:字符串怎么变成活代码

这是整个框架最"魔法"的一环:数据库里存的只是一段文本,怎么变成能调 pipe()/inlet() 的对象?答案在 utils/plugin.py。

5.1 核心:load_function_module_by_id —— exec 进一个临时模块

load_function_module_by_id(utils/plugin.py:250)把源码字符串执行进一个凭空造出来的模块,再看里面有没有约定的类:

# utils/plugin.py:263-289(节选)
module_name = f'function_{function_id}'
module = types.ModuleType(module_name)       # 造一个空模块对象
sys.modules[module_name] = module            # 注册进 sys.modules
...
module.__dict__['__file__'] = temp_file.name # 让模块以为自己有个真文件
exec(content, module.__dict__)               # ← 关键:把源码执行进模块命名空间
...
if hasattr(module, 'Pipe'):
    return module.Pipe(), 'pipe', frontmatter      # 找到 Pipe 类 → 实例化
elif hasattr(module, 'Filter'):
    return module.Filter(), 'filter', frontmatter  # 找到 Filter 类
elif hasattr(module, 'Action'):
    return module.Action(), 'action', frontmatter
else:
    raise Exception('No Function class found in the module')

"约定优于配置": 加载器不问你这是什么插件,而是按类名认——有 Pipe 类就是 pipe,有 Filter 类就是 filter。返回值第二项('pipe'/'filter'/'action')就是插件类型,用户写的时候完全不用声明 type。

关于"沙箱":要诚实——这里并没有真正的隔离沙箱。exec(content, module.__dict__) 是在主进程里直接执行用户代码,能 import 任意东西、访问文件系统。所以插件是受信管理员级能力,不是不可信用户的任意上传。真临时文件只用来给模块一个 __file__(utils/plugin.py:269),执行完就 os.unlink 删掉(utils/plugin.py:247/298)。load_tool_module_by_id(utils/plugin.py:204)是工具版的同一套机制,只是找的是 Tools 类。

加载失败会自动停用。 如果 exec 抛错(语法错、缺依赖…),加载器会把这个 function 标记为 is_active=False 并清掉 sys.modules 里的残留,避免坏插件反复拖垮请求:

# utils/plugin.py:290-296
except Exception as e:
    log.error(f'Error loading module: {function_id}: {e}')
    del sys.modules[module_name]                                  # 清理残留
    await Functions.update_function_by_id(function_id, {'is_active': False})  # 自动停用
    raise e

5.2 replace_imports:把老写法的 import 改对

社区插件常按老目录结构写 from utils... / from config...。加载前先做字符串替换,补上 open_webui. 前缀,让老插件也能跑:

# utils/plugin.py:185-199  replace_imports
replacements = {
    'from utils': 'from open_webui.utils',
    'from apps': 'from open_webui.apps',
    'from main': 'from open_webui.main',
    'from config': 'from open_webui.config',
}
for old, new in replacements.items():
    content = content.replace(old, new)

替换后如果内容变了,还会把改好的版本写回数据库(load_function_module_by_id 的 content is None 分支,utils/plugin.py:257-258),下次就不用再改。

5.3 frontmatter 与依赖安装

插件文件顶部可以放一段 """...""" 的 frontmatter(注意:是 Python 三引号,不是 YAML 的 ---)。extract_frontmatter(utils/plugin.py:149)解析它——必须第一行就是 """,否则返回空字典:

# utils/plugin.py:160-176(节选)
if len(lines) < 1 or lines[0].strip() != '"""':
    return {}                    # 不是三引号开头 → 没有 frontmatter
...
frontmatter_pattern = re.compile(r'^\s*([a-z_]+):\s*(.*)\s*$', re.IGNORECASE)
# 逐行 key: value 解析,直到遇到闭合的 """

frontmatter 里最有用的是 requirements(逗号分隔的 pip 包)。install_frontmatter_requirements(utils/plugin.py:391)在加载时按需 pip install:

# utils/plugin.py:401-413(节选)
req_list = [req.strip() for req in requirements.split(',')]
new_reqs = [req for req in req_list if req and req not in _installed_requirements]
if not new_reqs:
    return                       # 装过的不再装(进程级去重)
subprocess.check_call([sys.executable, '-m', 'pip', 'install'] + PIP_OPTIONS + new_reqs + ...)
_installed_requirements.update(new_reqs)

三道闸门:(1)全局开关 ENABLE_PIP_INSTALL_FRONTMATTER_REQUIREMENTS 关了就直接跳过(utils/plugin.py:393);(2)OFFLINE_MODE 下也跳过(397);(3)进程级 _installed_requirements 集合去重(388),同一个包不重复装。这意味着插件能在运行时给主进程装 pip 包——又一处印证它是受信能力,不是沙箱。

5.4 缓存:每次重读 vs 用缓存

动态加载不便宜(要 exec、可能 pip install)。get_function_module_from_cache(utils/plugin.py:340)在"新鲜度"和"性能"之间做权衡,靠 load_from_db 参数切换两种策略:

`load_from_db`	行为	用在哪	为什么
`True`(默认)	从数据库重读 content,比对缓存内容,变了才重新 `exec`	inlet / outlet 钩子	用户可能刚改过源码,要用最新版
`False`	直接用 `request.app.state.FUNCTIONS` 里的缓存,没缓存才加载	stream 钩子	分片逐个来,不能每片都读库

# utils/plugin.py:360-374(节选)关键的缓存命中判断
if (... 'FUNCTION_CONTENTS' ... and ... 'FUNCTIONS' ...):
    if request.app.state.FUNCTION_CONTENTS[function_id] == content:
        return request.app.state.FUNCTIONS[function_id], None, None   # 内容没变 → 直接用缓存对象
...
# 否则重新加载,并回写缓存
request.app.state.FUNCTIONS[function_id] = function_module
request.app.state.FUNCTION_CONTENTS[function_id] = content

缓存的键是内容本身:数据库里的 content 和缓存里记的 content 一字不差,才复用旧模块对象;否则重新 exec。这既保证"改了就生效",又避免"没改也重编译"。§4.4 提到的 load_from_db=(filter_type != 'stream') 就是在选这两条路。

6. Valves:插件的配置面板

插件常需要配置项(API key、开关、优先级)。Open WebUI 的约定是在类里定义一个 Valves(Pydantic 模型),框架就自动为它生成配置表单、存取值。

两层 valves:

名字	谁定义	谁能改	存哪
`Valves`	插件作者	管理员(全局一份)	`function.valves` 列
`UserValves`	插件作者	每个用户(各自一份)	用户 settings 里 `functions.valves`

加载时注入 valves。 pipe 的 valves 在 get_function_module_by_id(functions.py:52)里从数据库读出、实例化并挂到模块上;filter 的在 process_filter_functions(utils/filter.py:86-88)里同样处理:

# functions.py:55-66  给 pipe 模块挂上 valves
if hasattr(function_module, 'valves') and hasattr(function_module, 'Valves'):
    Valves = function_module.Valves
    valves = await Functions.get_function_valves_by_id(pipe_id)   # 从 DB 读
    if valves:
        function_module.valves = Valves(**{k: v for k, v in valves.items() if v is not None})
    else:
        function_module.valves = Valves()                        # 用默认值

UserValves 按用户注入。 若 pipe/filter 声明了 UserValves 且签名里要 __user__,框架会读出该用户自己的一份填进去(functions.py:196-202;filter 版 utils/filter.py:108-115)。这样同一个插件对不同用户能有不同配置。

动态下拉选项:resolve_valves_schema_options。 valves 字段可以要一个下拉框,选项既可以是静态列表,也可以是方法名(运行时调用得到选项,比如"列出当前可用模型")。resolve_valves_schema_options(utils/plugin.py:25)负责在生成 schema 时把方法名解析成真实选项:

# utils/plugin.py:100-124(节选)两种 options
if isinstance(options, list):
    resolved_options = options                    # Case 1: 直接是列表
elif isinstance(options, str) and options:
    method = getattr(valves_class, options, None) # Case 2: 是方法名 → 调它拿列表
    ...
    resolved_options = method(**kwargs) if kwargs else method()

调用时按方法签名注入 __user__/user(utils/plugin.py:119-122),和 pipe/filter 参数注入是同一套"按需给"的哲学。

7. 巧妙之处(可借鉴的技术)

"约定即注册":一个类名就是一次注册。 没有插件清单、没有 register() 调用——加载器直接 hasattr(module, 'Pipe') 认类型(utils/plugin.py:282-289)。用户写插件的心智负担被压到最低。
按签名注入(dependency injection by signature)。 pipe 和 filter 都用 inspect.signature 读出函数要哪些参数,只给它要的(functions.py:193-194、utils/filter.py:98-105)。插件想要 __user__ 就声明 __user__,不想要就不写——签名即依赖声明。
返回值多态归一化。 pipe 可以返回字符串/字典/同步生成器/异步生成器/StreamingResponse,框架用一串 isinstance 分支统一翻译成 OpenAI SSE(functions.py:296-324)。作者怎么方便怎么写,协议一致性由框架兜底。
manifold 用点号做命名空间。 子模型 id = 父.子,路由时 split('.', 1) 还原父 id(functions.py:182-184)。一个约定同时解决了"一 pipe 多模型"的注册和路由。
内容哈希式缓存 + 按钩子选新鲜度。 缓存键是 content 本身(utils/plugin.py:363),改了才重编译;并用 load_from_db 让高频的 stream 钩子走缓存、低频的 inlet/outlet 走最新源码(utils/filter.py:75)。
坏插件自我隔离。 加载失败自动 is_active=False(utils/plugin.py:295),避免一个语法错误的插件反复炸请求。

8. 边界与局限(诚实)

没有真沙箱。 exec 在主进程直接跑用户代码,能读写文件、能 pip install(§5.1、§5.3)。插件是受信管理员级扩展点,绝不能开放给不可信用户随意上传——这不是安全隔离机制。
依赖安装是全局副作用。 install_frontmatter_requirements 往主进程环境装 pip 包(utils/plugin.py:410),多个插件的依赖版本可能互相打架;去重只在进程内存里(_installed_requirements),重启后重来。
filter 顺序依赖 priority 约定。 顺序由各 filter 的 Valves.priority 决定(utils/filter.py:59),没有全局编排;两个插件都想"最先跑"时,得靠人工调 priority。
frontmatter 格式严格。 必须第一行就是 """,且 key 只匹配 [a-z_]+(utils/plugin.py:156、160);写成 YAML 的 --- 或首行有空行,会被当成"没有 frontmatter"静默忽略。
pipe 走的是 bypass_filter 路径。 model.get('pipe') 分支注释明说它"已经在 bypass filter"(utils/chat.py:275)——也就是说 pipe 自定义模型的调用不会再叠加 §4 的 filter 链,行为上和普通模型不同,需要留意。

9. 代码地图(导航索引)

主题	文件路径	符号名
pipe 执行入口(注入参数、归一化返回)	`backend/open_webui/functions.py:147`	`generate_function_chat_completion`
pipe → 模型列表(含 manifold 展开)	`backend/open_webui/functions.py:71`	`get_function_models`
加载 pipe 模块 + 挂 valves	`backend/open_webui/functions.py:52`	`get_function_module_by_id`
manifold 子 id 还原	`backend/open_webui/functions.py:180`	`get_pipe_id`
流式分片格式化	`backend/open_webui/functions.py:162`	`process_line`
pipe 路由分支	`backend/open_webui/utils/chat.py:274`	`model.get('pipe')`(在 `generate_chat_completion`)
pipe 模型注册进全量模型表	`backend/open_webui/utils/models.py:65`	`get_function_models` 调用点
filter 收集+优先级排序	`backend/open_webui/utils/filter.py:21`	`get_sorted_filter_ids`
filter 逐个执行(inlet/outlet/stream)	`backend/open_webui/utils/filter.py:66`	`process_filter_functions`
inlet 接线(时机见 [02])	`backend/open_webui/utils/middleware.py:2539`	`process_filter_functions('inlet')`
outlet 接线(时机见 [04])	`backend/open_webui/utils/middleware.py:3274`	`outlet_filter_handler` / `:3378`
stream 接线	`backend/open_webui/utils/middleware.py:5229`	`process_filter_functions('stream')`
动态加载:exec 成模块	`backend/open_webui/utils/plugin.py:250`	`load_function_module_by_id`
工具版动态加载	`backend/open_webui/utils/plugin.py:204`	`load_tool_module_by_id`
frontmatter 解析	`backend/open_webui/utils/plugin.py:149`	`extract_frontmatter`
import 路径修正	`backend/open_webui/utils/plugin.py:185`	`replace_imports`
依赖 pip 安装	`backend/open_webui/utils/plugin.py:391`	`install_frontmatter_requirements`
模块缓存 + 新鲜度策略	`backend/open_webui/utils/plugin.py:340`	`get_function_module_from_cache`
valves 动态下拉选项	`backend/open_webui/utils/plugin.py:25`	`resolve_valves_schema_options`
Function 表结构 / valves 存取	`backend/open_webui/models/functions.py:18`	`Function` / `FunctionsTable`
取某类型激活函数	`backend/open_webui/models/functions.py:262`	`get_functions_by_type`
取全局 filter	`backend/open_webui/models/functions.py:272`	`get_global_filter_functions`

1. 这是什么(零基础也能懂)​

2. 顶层全景(它大概怎么转)​

3. Pipe 函数:自定义"模型"后端​

3.1 它要解决的小问题​

3.2 思路:让 pipe"假装"成模型混进模型列表​

3.3 Manifold:一个 pipe 暴露多个模型​

3.4 路由:选中 pipe 后,请求怎么走进 pipe()​

4. Filter 函数:inlet / outlet / stream 三段拦截​

4.1 它要解决的小问题​

4.2 思路:一个 Filter 类,最多三个钩子​

4.3 谁来跑、按什么顺序:get_sorted_filter_ids​

4.4 怎么跑:process_filter_functions​

4.5 三处接线在哪(只交叉引用,不展开)​

5. 插件加载与沙箱:字符串怎么变成活代码​

5.1 核心:load_function_module_by_id —— exec 进一个临时模块​

5.2 replace_imports:把老写法的 import 改对​

5.3 frontmatter 与依赖安装​

5.4 缓存:每次重读 vs 用缓存​

6. Valves:插件的配置面板​

7. 巧妙之处(可借鉴的技术)​

8. 边界与局限(诚实)​

9. 代码地图(导航索引)​