A. 最終效果

B. 環境配置

配置并啟動Open-WebUI

• 隨后瀏覽器訪問：http://localhost:8080

pip install open-webui
open-webui serve

配置并啟動Pipelines

• Pipelines默認占用80端口
• 相比于Docker的啟動方式，可以在相同的命令行中，查看pipelines 的日志

git clone https://github.com/open-webui/pipelines.git
cd pipelines
pip install -r requirements.txt
sh ./start.sh

配置并啟動Dify

• dify/docker/docker-compose.yaml文件會包含很多image資源，默認只啟動其中的幾個

cd dify
cd docker
cp .env.example .env
docker compose up -d

• 3min配置一個ChatFlow應用
• 點擊LLM大模型模塊自定義System中的提示詞
• 點擊預覽測試對話
• 首次對外使用要點擊發布，再次應用更新要點擊發布

• API調用的測試代碼：

import requests
import jsonAPI_KEY = "app-w1pVOdGHpJ81OmqsZ2YIXyT8"  # 你的真實 API Keyurl = "http://localhost/v1/chat-messages"
headers = {"Authorization": f"Bearer {API_KEY}","Content-Type": "application/json"
}payload = {"inputs": {},"query": "你的名字是什么？","response_mode": "streaming","conversation_id": "","user": "abc-123","files": [{"type": "image","transfer_method": "remote_url","url": "https://cloud.dify.ai/logo/logo-site.png"}]
}with requests.post(url, headers=headers, data=json.dumps(payload), stream=True) as r:for raw in r.iter_lines():if not raw:continueif not raw.startswith(b"data:"):continuedata_str = raw[len(b"data:"):].strip().decode("utf-8", errors="ignore")if data_str in ("[DONE]", ""):continuetry:event = json.loads(data_str)except json.JSONDecodeError:continueetype = event.get("event")if etype == "message":chunk = event.get("answer", "")if chunk:print(chunk, end="", flush=True)if etype in ("message_end", "workflow_finished"):print()break

C. 測試案例

• 案例1： pipelines官方代碼

from typing import List, Union, Generator, Iterator
from schemas import OpenAIChatMessage
import subprocessclass Pipeline:def __init__(self):# Optionally, you can set the id and name of the pipeline.# Best practice is to not specify the id so that it can be automatically inferred from the filename, so that users can install multiple versions of the same pipeline.# The identifier must be unique across all pipelines.# The identifier must be an alphanumeric string that can include underscores or hyphens. It cannot contain spaces, special characters, slashes, or backslashes.# self.id = "python_code_pipeline"self.name = "Python Code Pipeline"passasync def on_startup(self):# This function is called when the server is started.print(">>>" * 80)print(f"on_startup:{__name__}")passasync def on_shutdown(self):# This function is called when the server is stopped.print("<<<" * 80)print(f"on_shutdown:{__name__}")passdef execute_python_code(self, code):try:result = subprocess.run(["python", "-c", code], capture_output=True, text=True, check=True)stdout = result.stdout.strip()return stdout, result.returncodeexcept subprocess.CalledProcessError as e:return e.output.strip(), e.returncodedef pipe(self, user_message: str, model_id: str, messages: List[dict], body: dict) -> Union[str, Generator, Iterator]:# This is where you can add your custom pipelines like RAG.print(f"pipe:{__name__}")print(messages)print(user_message)if body.get("title", False):print("Title Generation")return "Python Code Pipeline"else:# stdout, return_code = self.execute_python_code(user_message)stdout = "This is a test"return stdout

• 案例2： dify自定義chatflow

# pipelines/python_code_pipeline/__init__.py
from typing import List, Union, Generator, Iterator
import os
import json
import requests
from requests.exceptions import RequestException, Timeout# 可選：Open-WebUI 的消息類型（不是必須）
# from schemas import OpenAIChatMessageclass Pipeline:def __init__(self):self.name = "Dify_ChatFlow"# 優先走環境變量，便于 Docker 與本地雙環境切換# 本機直接跑 ./start.sh 時，Dify 通常可用 http://localhost# 若 pipelines 用 Docker 跑，而 Dify 也在 Docker，則一般用 http://host.docker.internalself.DIFY_BASE_URL = "http://localhost"self.DIFY_API_KEY  = "app-w1pVOdGHpJ81OmqsZ2YIXyT8"   # !!! 改成你的 app key# Dify “對話機器人”的接口路徑（保持默認即可）self.DIFY_CHAT_MESSAGES_PATH = os.getenv("DIFY_CHAT_MESSAGES_PATH", "/v1/chat-messages")# 超時（秒）self.CONNECT_TIMEOUT = float(os.getenv("DIFY_CONNECT_TIMEOUT", "10"))self.READ_TIMEOUT    = float(os.getenv("DIFY_READ_TIMEOUT", "120"))async def on_startup(self):print(">>> Dify bridge ready:", self.DIFY_BASE_URL)async def on_shutdown(self):print("Dify bridge shutdown")# ------ 內部：把請求打到 Dify 并以生成器流式返回 ------def _dify_stream(self, query: str, conversation_id: str | None, user_id: str = "openwebui-user") -> Iterator[str]:url = f"{self.DIFY_BASE_URL.rstrip('/')}{self.DIFY_CHAT_MESSAGES_PATH}"headers = {"Authorization": f"Bearer {self.DIFY_API_KEY}","Content-Type": "application/json"}payload = {"inputs": {},"query": query,"response_mode": "streaming","conversation_id": conversation_id or "","user": user_id,# 如需帶圖：按需傳 files# "files": [{"type": "image","transfer_method": "remote_url","url": "https://cloud.dify.ai/logo/logo-site.png"}]}try:with requests.post(url,headers=headers,data=json.dumps(payload),stream=True,timeout=(self.CONNECT_TIMEOUT, self.READ_TIMEOUT),) as r:r.raise_for_status()for raw in r.iter_lines():if not raw:continueif not raw.startswith(b"data:"):continuedata_str = raw[len(b"data:"):].strip().decode("utf-8", errors="ignore")if data_str in ("[DONE]", ""):continuetry:event = json.loads(data_str)except json.JSONDecodeError:continueetype = event.get("event")if etype == "message":chunk = event.get("answer", "")if chunk:yield chunkelif etype in ("message_end", "workflow_finished"):breakexcept Timeout:yield "\n[Pipeline] Dify request timed out."except RequestException as e:yield f"\n[Pipeline] Dify request error: {e}"# ------ Open-WebUI 調用的主入口 ------def pipe(self, user_message: str, model_id: str, messages: List[dict], body: dict) -> Union[str, Generator, Iterator]:print("pipe:python_code_pipeline")# 忽略 Open-WebUI 的標題/標簽探測任務（它們會帶一大段“### Task:”提示詞）if body.get("title", False) or user_message.strip().startswith("### Task:"):# 返回一個簡單標題或空串即可，避免把系統提示詞發給 Difyreturn "" if not body.get("title") else "Python Code Pipeline"# 你也可以從 body 中取會話 id（如果前端傳了）conversation_id = body.get("conversation_id") or ""# 把用戶問題轉發到 Dify，并以流式返回return self._dify_stream(query=user_message, conversation_id=conversation_id)

D. 界面設置

如下操作即可

E. 在主界面中使用

• 看到左邊的對話記錄，就知道我測試多少次，不太懂，不太熟，但是坑又太多
• 點贊、留言、轉發：謝謝

References

• open-webui/pipelines: Pipelines: Versatile, UI-Agnostic OpenAI-Compatible Plugin Framework
• 哎…要會員，要收費、有點難：將Dify平臺開發的工作流集成到Open WebUI中_dify接入openwebui-CSDN博客
• 雖熱官網實例代碼沒有跑通，但是還是要給開放的博主點贊：OpenWebUI通過pipeline對接dify的workflow-CSDN博客
• The link between the workflow and open-webUI · langgenius/dify · Discussion #20982
• The link between the workflow and open-webUI · langgenius/dify · Discussion #20982
• [Feature]: Integrate with AI Workflow platforms such as Flowise & dify · open-webui/open-webui · Discussion #2023
• langgenius/dify: Production-ready platform for agentic workflow development.