OpenWebUI 零源码改造接入自研 RAG 知识库|完整落地教程(私有化部署)

"""

title: Langgraph stream integration

author: bartonzzx

author_url: https://github.com/bartonzzx

git_url: 

description: Integrate langgraph with open webui pipeline

required_open_webui_version: 0.4.3

requirements: none

version: 0.4.3

licence: MIT

"""

import os

import sys

import json

from dotenv import load_dotenv

from typing import Annotated, Literal

from typing_extensions import TypedDict

from fastapi import FastAPI

from fastapi.responses import StreamingResponse

from langgraph.graph import StateGraph, START, END

from langgraph.graph.message import add_messages

from langchain_openai import ChatOpenAI

from langgraph.config import get_stream_writer

# ========== 修复1：移除getpass，加载.env环境变量，兼容后台nohup ==========

# 加载同目录.env文件，无交互式输入

load_dotenv()

def _set_env(var: str):

# 后台运行禁用交互式输入，缺失环境变量直接抛出明确异常

if not os.environ.get(var):

raise RuntimeError(f"环境变量 {var} 未配置，请在当前目录.env文件中填写")

# 读取密钥，不会触发终端输入

_set_env("OPENAI_API_KEY")

'''

Define Langgraph

'''

# 新增：全局缓冲阈值，30字符批量输出

BUFFER_THRESHOLD = 30

def generate_custom_stream(type: Literal["think","normal"], buffer_obj, content: str):

# 写入缓冲区，不再直接推送

buffer_obj["buf"] += content

# 达到阈值才输出

if len(buffer_obj["buf"]) >= BUFFER_THRESHOLD:

custom_stream_writer = get_stream_writer()

custom_stream_writer({type: buffer_obj["buf"]})

buffer_obj["buf"] = ""

class State(TypedDict):

messages: Annotated[list, add_messages]

llm = ChatOpenAI(

model="qwen3.6-27b",

temperature=0.1,

base_url='http://10.199.5.10:8003/v1',

api_key=os.getenv("OPENAI_API_KEY", "dummy_key") # 从环境变量读取，兜底默认

)

# ========== 修复2：同步llm.invoke阻塞问题，改为异步astream实时输出 + 字符缓冲 ==========

async def chatbot(state: State):

# 初始化缓冲区

think_buf = {"buf": ""}

normal_buf = {"buf": ""}

# 异步分步输出思考内容，不阻塞主流

async for chunk in llm.astream(["Please reasoning:"] + state["messages"]):

generate_custom_stream("think", think_buf, chunk.content)

# 思考流剩余缓存强制输出

if think_buf["buf"]:

custom_stream_writer = get_stream_writer()

custom_stream_writer({"think": think_buf["buf"]})

async for chunk in llm.astream(state["messages"]):

generate_custom_stream("normal", normal_buf, chunk.content)

# 回答流剩余缓存强制输出

if normal_buf["buf"]:

custom_stream_writer = get_stream_writer()

custom_stream_writer({"normal": normal_buf["buf"]})

return {"messages": [normal_buf["buf"]]}

# Define graph

graph_builder = StateGraph(State)

# Define nodes

graph_builder.add_node("chatbot", chatbot)

graph_builder.add_edge("chatbot", END)

# Define edges

graph_builder.add_edge(START, "chatbot")

# Compile graph

graph = graph_builder.compile()

'''

Define api processing 

'''

app = FastAPI(

title="Langgraph API",

description="Langgraph API",

)

@app.get("/test")

async def test():

return {"message": "Hello World"}

@app.post("/stream")

async def stream(inputs: State):

async def event_stream():

try:

stream_start_msg = {

'choices': 

[

{

'delta': {}, 

'finish_reason': None

}

]

}

yield f"data: {json.dumps(stream_start_msg)}\n\n"

#接口层二次缓冲兜底，防止极端细碎分片

think_buffer = ""

normal_buffer = ""

# Processing langgraph stream response withblock support

async for event in graph.astream(input=inputs, stream_mode="custom"):

print(f"stream event: {event}")

think_content = event.get("think", None)

normal_content = event.get("normal", None)

# 分段推送思考流

if think_content:

think_buffer += think_content

# 攒够字符一次性输出

if len(think_buffer) >= BUFFER_THRESHOLD:

think_msg = {

'choices': 

[

{

'delta':

{

'reasoning_content': think_buffer, 

},

'finish_reason': None

}

]

}

yield f"data: {json.dumps(think_msg)}\n\n"

think_buffer = ""

# 分段推送回答流

if normal_content:

normal_buffer += normal_content

if len(normal_buffer) >= BUFFER_THRESHOLD:

normal_msg = {

'choices': 

[

{

'delta':

{

'content': normal_buffer, 

},

'finish_reason': None

}

]

}

yield f"data: {json.dumps(normal_msg)}\n\n"

normal_buffer = ""

#循环结束，输出缓冲区剩余所有文字

if think_buffer:

think_msg = {

'choices': 

[

{

'delta':

{

'reasoning_content': think_buffer, 

},

'finish_reason': None

}

]

}

yield f"data: {json.dumps(think_msg)}\n\n"

if normal_buffer:

normal_msg = {

'choices': 

[

{

'delta':

{

'content': normal_buffer, 

},

'finish_reason': None

}

]

}

yield f"data: {json.dumps(normal_msg)}\n\n"

# End of the stream

stream_end_msg = {

'choices': [ 

{

'delta': {}, 

'finish_reason': 'stop'

}

]

}

yield f"data: {json.dumps(stream_end_msg)}\n\n"

yield f"data: [DONE]\n\n" # 兼容OpenAI标准SSE结束标记

except Exception as e:

print(f"Stream error: {e}", file=sys.stderr)

error_msg = {

"error": {"message": str(e), "code": 500}

}

yield f"data: {json.dumps(error_msg)}\n\n"

return StreamingResponse(

event_stream(), 

media_type="text/event-stream",

headers={

"Cache-Control": "no-cache",

"Connection": "keep-alive",

"X-Accel-Buffering": "no" # 关闭nginx缓冲，流式实时输出

}

)

if __name__ == "__main__":

import uvicorn

# 固定监听0.0.0.0:8863，移除--reload避免多进程IO崩溃