LangGraph 接入 Open WebUI：OpenAI 兼容接口与流式输出

在做官网生成 Agent的时候，用 LangGraph 串联了几个节点：先判断需求类型，再做需求分析或内容检查，中间可能调用图片、视频搜索工具，最后由 code_agent 生成 HTML。

最初调试时，我一直看控制台。路由结果是一段 JSON，需求分析又是一段 JSON，最后还有几百行 HTML，想要找到生成结果还比较费力。

后来想到接一个现成的聊天界面，就找到了 Open WebUI。Agent 继续负责编码，FastAPI 在外面提供 /v1/models 和 /v1/chat/completions，让聊天界面能发现它、调用它，再把生成的内容逐段显示出来。

这篇记录这一层接口的写法。正文保留最终 HTML，中间节点用简短的执行进度展示，调试时可以展开查看。

下面按项目里的节点名整理代码。网关片段放在同一个 fast_api.py 中；init_embedding()、create_assistant() 沿用项目中的实现，导入路径按实际目录填写。示例使用 Python 3.11+、Pydantic 2，以及 LangGraph 的 astream 多模式接口。

一、先把图接成一个可选择的模型

对 Open WebUI 来说，后面可以是一个模型，也可以是一张会调用多个模型和工具的图。客户端需要知道模型 ID，并能按照约定提交消息、接收回答。

这里对外使用 website-agent。它代表“生成网站”这项能力，不是底层模型的名称。/v1/models 返回的 ID、请求里的 model 和响应里的 model 保持一致。

这次接入只处理文字聊天，工具调用仍在图内部完成。图片输入、客户端传入的 tools、JSON 输出格式等参数暂不接收；各节点的温度和输出长度由图自己的模型配置管理。客户端传入未实现的参数时直接返回 422，避免界面上改了参数，实际执行却没有变化。

先定义请求和消息转换：

import asyncio
import json
import logging
import os
import secrets
import time
import uuid
from contextlib import aclosing, asynccontextmanager
from typing import Literal

from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse, StreamingResponse
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from langchain_core.messages import (
    AIMessage, AIMessageChunk, HumanMessage, SystemMessage,
)
from pydantic import BaseModel, ConfigDict, Field


class TextPart(BaseModel):
    model_config = ConfigDict(extra="forbid")
    type: Literal["text"]
    text: str


class ChatMessage(BaseModel):
    model_config = ConfigDict(extra="forbid")
    role: Literal["system", "user", "assistant"]
    content: str | list[TextPart]


class StreamOptions(BaseModel):
    model_config = ConfigDict(extra="forbid")
    include_usage: Literal[False] = False


class ChatRequest(BaseModel):
    model_config = ConfigDict(extra="forbid")
    model: str
    messages: list[ChatMessage] = Field(min_length=1)
    stream: bool = False
    stream_options: StreamOptions | None = None
    user: str | None = None


def convert_messages(messages):
    message_types = {
        "system": SystemMessage,
        "user": HumanMessage,
        "assistant": AIMessage,
    }
    result = []
    for item in messages:
        content = item.content
        if isinstance(content, list):
            content = "".join(part.text for part in content)
        result.append(message_types[item.role](content=content))
    return result

content 支持字符串，也支持纯文字数组。如果数组中夹了一张图片，请求校验就会失败，不会只留下文字继续生成。tool 消息同样不会被误当成用户消息。

user 可以作为调用方提供的标识保留，但不拿它选择图的会话。上下文怎么传，在下一节单独处理。

模型与向量库在启动阶段初始化，API Key 从环境变量读取。这里的 Key 用于访问自己的网关，上游模型使用的 Key 仍放在上游配置中。

MODEL_ID = "website-agent"
logger = logging.getLogger(__name__)
bearer = HTTPBearer(auto_error=False)


@asynccontextmanager
async def lifespan(app: FastAPI):
    api_key = os.environ.get("GATEWAY_API_KEY", "")
    if not api_key:
        raise RuntimeError("请配置 GATEWAY_API_KEY")

    # 在文件顶部导入项目现有的这两个函数。
    init_embedding()
    app.state.graph = create_assistant()
    app.state.api_key = api_key
    app.state.show_progress = (
        os.environ.get("GATEWAY_SHOW_PROGRESS", "0") == "1"
    )
    yield


api = FastAPI(lifespan=lifespan)


async def check_auth(
    request: Request,
    credentials: HTTPAuthorizationCredentials | None = Depends(bearer),
):
    if credentials is None or not secrets.compare_digest(
        credentials.credentials.encode("utf-8"),
        request.app.state.api_key.encode("utf-8"),
    ):
        raise HTTPException(
            status_code=401,
            detail="API Key 无效",
            headers={"WWW-Authenticate": "Bearer"},
        )


@api.get("/health/ready")
async def ready():
    return {"status": "ready"}


@api.get("/v1/models", dependencies=[Depends(check_auth)])
async def list_models():
    return {
        "object": "list",
        "data": [{
            "id": MODEL_ID,
            "object": "model",
            "created": 0,
            "owned_by": "local",
        }],
    }

初始化失败时服务直接启动失败，不等第一个聊天请求进来才加载模型。/health/ready 只表示当前进程已经完成初始化，不代表每个上游服务此刻都可用。若初始化函数创建了需要关闭的连接池，其释放也放在 lifespan 的退出阶段。

Open WebUI 可以通过兼容接口的模型列表发现这个 ID，再向聊天接口提交请求。Open WebUI 连接说明

二、历史消息只交给一处管理

接聊天界面时，要先确定上一轮消息由谁保存。

这里使用客户端传入的 messages 作为本次请求的完整历史。用户在同一个聊天窗口继续提问时，客户端把前面的对话一起提交；网关收到后，新建一次图执行。

inputs = {"messages": convert_messages(body.messages)}
config = {
    "configurable": {"thread_id": uuid.uuid4().hex},
}

这样不同聊天窗口不会因为共用一个 user 值而读到同一份图状态，也不会把客户端提交的完整历史再追加到上一轮的 checkpoint 中。

这里的 thread_id 只区分本次执行。若图本身不需要 checkpoint，可以不配置持久化；若需要保留运行记录，要另外设置清理规则，避免每次请求留下的记录不断累积。

网站生成还有一个实际问题：下一轮只传聊天记录，未必能恢复上一轮的 task_result、check_result 或工作目录。采用本文方式时，每次执行都需要从消息重新整理需求。若要在原任务上继续编辑文件，就要使用独立的任务 ID，校验任务所属用户，并改为提交新增消息；不能直接把这套“每次提交完整历史”的接口换成复用旧线程。

图内部的 messages 也只追加本节点的新消息。状态可以继承 MessagesState，保留项目中单独使用的需求、检查字段：

from langgraph.graph import MessagesState


class TaskState(MessagesState):
    task_result: str
    check_result: str

以工具调用节点为例，tool_model 使用项目中已经绑定工具的模型实例。节点取得响应后，直接返回这条新消息：

async def llm_tool_call(state: TaskState, config):
    response = await tool_model.ainvoke(state["messages"], config=config)
    return {"messages": [response]}

MessagesState 的消息字段使用 add_messages 合并更新。节点函数不需要再拼上 state["messages"]。保留原来的 response 对象，也能保留其中的消息 ID、工具调用和响应元数据。LangGraph 消息状态说明

三、正文和执行进度分开输出

这张图里，code_agent 最后生成 HTML，随后进入 END。其他节点负责准备数据：

llm_tool_call 完成只说明调用决策已经生成，不表示图片或视频搜索已经执行成功。工具执行结果由实际工具节点处理。

LangGraph 的 messages 模式给出模型消息片段和元数据，可以用 langgraph_node 判断来源；updates 模式给出节点完成后的状态更新。这里同时订阅两种事件：前者用于输出 HTML，后者用于显示完成进度。LangGraph 流式输出说明

PROGRESS_LABELS = {
    "my_router": "路由处理完成",
    "task_agent": "需求整理完成",
    "check_agent": "内容检查完成",
    "llm_tool_call": "工具调用决策完成",
}
FINAL_NODE = "code_agent"


def output_text(content):
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        return "".join(
            part["text"]
            for part in content
            if isinstance(part, dict)
            and part.get("type") == "text"
            and isinstance(part.get("text"), str)
        )
    return ""


async def graph_events(graph, inputs, config):
    emitted_text = False
    final_message = None
    source = graph.astream(
        inputs,
        config=config,
        stream_mode=["messages", "updates"],
    )

    async with aclosing(source):
        async for mode, data in source:
            if mode == "messages":
                message, metadata = data
                if metadata.get("langgraph_node") != FINAL_NODE:
                    continue
                if not isinstance(message, AIMessageChunk):
                    continue
                text = output_text(message.content)
                if text:
                    emitted_text = True
                    yield "content", text

            elif mode == "updates":
                if "__interrupt__" in data:
                    raise RuntimeError("图等待人工输入，本接口不支持恢复执行")
                for node, update in data.items():
                    if node in PROGRESS_LABELS:
                        yield "progress", PROGRESS_LABELS[node] + "\n"
                    if node == FINAL_NODE:
                        if final_message is not None:
                            raise RuntimeError("最终生成节点执行了多次")
                        messages = (update or {}).get("messages", [])
                        if len(messages) != 1 or not isinstance(messages[0], AIMessage):
                            raise RuntimeError("最终节点必须返回一条 AIMessage")
                        final_message = messages[0]

    if final_message is None:
        raise RuntimeError("图未返回最终回答")
    if final_message.tool_calls or final_message.invalid_tool_calls:
        raise RuntimeError("最终节点仍包含未处理的工具调用")

    reason = final_message.response_metadata.get("finish_reason")
    if reason not in {"stop", "length", "content_filter"}:
        raise RuntimeError("上游未返回可识别的结束原因")

    # 某些调用只返回完整消息，没有产生文字片段。
    if not emitted_text:
        text = output_text(final_message.content)
        if text:
            emitted_text = True
            yield "content", text
    if not emitted_text and reason != "content_filter":
        raise RuntimeError("最终回答为空")

    yield "finish_reason", reason

这段代码使用 LangGraph 默认的 v1 流格式，多模式事件按 (mode, data) 解包；不要与显式指定 version="v2" 后的字典格式混用。依赖版本仍以项目锁文件为准。

code_agent 在这里有一个明确约定：每次执行只调用一次最终文本生成模型，原样返回 {"messages": [response]}，随后结束图。如果节点还会审稿、重写，或一次调用多个模型，就需要给真正用于回答的调用单独加标记，不能仅靠节点名把所有文字片段都发出去。

节点拿到的完整消息用于读取结束原因，不会在已经输出 token 后再把全文发一次。length 也会照实返回，方便识别 HTML 被长度限制截断的情况。若更换上游后元数据字段不同，应在节点中完成转换，再交给网关。

新增节点默认不向客户端输出内容。需要展示时，再加入进度表或指定最终输出来源。路由 JSON、工具参数和中间模型输出仍留在服务端，不混入 HTML。

四、把事件转换成 SSE

前面拿到的是图事件，客户端需要的是 Chat Completions 响应。

SSE 每条事件用空行隔开。下面统一生成 data: JSON\n\n，同一次回答的 id、created 和 model 保持不变：

def sse_data(payload):
    return "data: " + json.dumps(payload, ensure_ascii=False) + "\n\n"


def completion_chunk(request_id, created, delta, finish_reason=None):
    return {
        "id": request_id,
        "object": "chat.completion.chunk",
        "created": created,
        "model": MODEL_ID,
        "choices": [{
            "index": 0,
            "delta": delta,
            "finish_reason": finish_reason,
        }],
    }

ensure_ascii=False 只是让抓包时的中文更容易读，保留默认的 Unicode 转义也能正常解析。事件末尾的空行则用于分隔事件，不能漏掉。

正常响应先发送 role，再发送正文片段，最后发送结束原因和 [DONE]。这里的 finish_reason 来自最终生成节点，不能不管执行结果都写成 stop。Chat Completions 流式响应字段

进度展示借用 Open WebUI 能识别的 reasoning_content。这是界面扩展，不是 OpenAI Chat Completions 的标准字段。 这里面放的是我们自己定义的节点状态，并不代表模型的内部思考。Open WebUI 的展示和历史回传方式还受版本、连接类型影响，因此用环境变量单独开启。Open WebUI 推理内容显示说明

主入口把这些处理接起来：

@api.post("/v1/chat/completions", dependencies=[Depends(check_auth)])
async def chat_completions(body: ChatRequest, request: Request):
    if body.model != MODEL_ID:
        raise HTTPException(status_code=404, detail="模型不存在")
    if body.messages[-1].role != "user":
        raise HTTPException(status_code=422, detail="最后一条消息需要是用户消息")

    inputs = {"messages": convert_messages(body.messages)}
    config = {"configurable": {"thread_id": uuid.uuid4().hex}}
    request_id = "chatcmpl-" + uuid.uuid4().hex
    created = int(time.time())
    graph = request.app.state.graph
    show_progress = request.app.state.show_progress

    async def events():
        # 流式和非流式共用同一条事件处理路径。
        async with asyncio.timeout(180):
            async with aclosing(graph_events(graph, inputs, config)) as source:
                async for event in source:
                    yield event

    def log_failure(exc):
        logger.error(
            "graph_failed request_id=%s exception=%s",
            request_id, type(exc).__name__,
        )

    if body.stream:
        async def stream_body():
            yield sse_data(completion_chunk(
                request_id, created, {"role": "assistant", "content": ""}
            ))
            reason = None
            try:
                async with aclosing(events()) as source:
                    async for kind, value in source:
                        if kind == "finish_reason":
                            reason = value
                        elif kind == "content":
                            yield sse_data(completion_chunk(
                                request_id, created, {"content": value}
                            ))
                        elif show_progress and kind == "progress":
                            yield sse_data(completion_chunk(
                                request_id, created, {"reasoning_content": value}
                            ))
            except asyncio.CancelledError:
                raise
            except Exception as exc:
                log_failure(exc)
                yield sse_data({"error": {
                    "message": "生成未完成，请使用请求 ID 查询服务端记录。",
                    "type": "server_error",
                    "code": "graph_execution_failed",
                    "request_id": request_id,
                }})
                return

            yield sse_data(completion_chunk(request_id, created, {}, reason))
            yield "data: [DONE]\n\n"

        return StreamingResponse(
            stream_body(),
            media_type="text/event-stream",
            headers={
                "Cache-Control": "no-cache",
                "X-Accel-Buffering": "no",
                "X-Request-ID": request_id,
            },
        )

    parts = []
    reason = None
    try:
        async with aclosing(events()) as source:
            async for kind, value in source:
                if kind == "content":
                    parts.append(value)
                elif kind == "finish_reason":
                    reason = value
    except asyncio.CancelledError:
        raise
    except Exception as exc:
        log_failure(exc)
        raise HTTPException(
            status_code=502,
            detail={"message": "生成未完成", "request_id": request_id},
        ) from None

    return JSONResponse(
        content={
            "id": request_id,
            "object": "chat.completion",
            "created": created,
            "model": MODEL_ID,
            "choices": [{
                "index": 0,
                "message": {"role": "assistant", "content": "".join(parts)},
                "finish_reason": reason,
            }],
        },
        headers={"X-Request-ID": request_id},
    )

请求格式、鉴权和模型 ID 都在流开始前校验，可以正常返回 4xx。开始发送后若图执行失败，HTTP 状态码已经无法修改，便发送独立的 error 数据并结束连接，不把报错拼进生成的 HTML，也不再发送成功结束的 chunk。不同客户端对流中错误的展示可能不同，接入时需要确认，不能只凭 HTTP 200 判断生成成功。

180 秒是示例中的执行超时，可以按网站生成任务的耗时调整。异步取消会继续向图内传播；图里的模型客户端和搜索工具也要配置各自的超时，避免同步阻塞调用拖住整个请求。

这里没有返回 usage，也不接受 include_usage=true。一次网站生成可能调用多个模型，只拿 code_agent 的 token 数并不能代表整次任务消耗。需要统计时，应在每次上游调用完成后收集实际用量，再明确是统计最终回答，还是统计整个任务。

五、接入 Open WebUI

依赖继续使用项目已有的 uv 配置。先设置网关 Key，再启动服务。启动对象是 fast_api.py 中的 api：

uv sync --frozen
uv run --frozen uvicorn fast_api:api --host 127.0.0.1 --port 8000

在 Open WebUI 管理员设置的连接页面，添加 OpenAI 兼容连接：

先关闭该模型上的额外请求参数和客户端工具功能。本文请求模型没有接收 temperature、max_tokens、tools 等字段；如果 Open WebUI 的模型预设或过滤器自动添加了这些参数，需要移除，否则会得到 422。

自动标题、标签、后续问题也可能触发额外模型调用。网站 Agent 不适合处理这些后台小任务，可以关闭对应功能，或把它们分配给另一个普通聊天模型，避免每次生成标题都运行一遍网站生成流程。

如果 Open WebUI 在 Docker 中，容器里的 127.0.0.1 指向容器自身。应改成容器能够访问的服务地址；使用 Docker Desktop 时可以考虑 host.docker.internal，但网关也必须监听容器能够到达的地址。上面的回环监听适用于同机调用或同机 Nginx 代理，不能直接套到所有容器部署中。

接好后，可以发送“生成一个二次元手办商城官网”。经过的节点会逐个产生完成提示，最终 HTML 进入正文。部分节点是条件分支，不会每次都执行，界面也不应提前把整张表里的步骤全部显示为完成。

换用其他客户端时，可以关闭进度扩展，只返回正文。Open WebUI 若把进度内容重新带入下一轮历史，需要按其实际版本和连接配置调整历史回传；不要靠删除任意 <think> 文本来清理消息，用户提交的代码或文章中也可能包含这样的标签。

六、经过 Nginx 时关闭响应缓冲

本地直连时，文字能逐段出现；放到 Nginx 后面，却可能变成等很久后一次性显示。这里要检查代理是否把上游响应缓冲了。

网关已经设置 X-Accel-Buffering: no，Nginx 的接口路径也明确关闭缓冲：

location /v1/ {
    proxy_pass http://127.0.0.1:8000;
    proxy_http_version 1.1;
    proxy_set_header Host $host;
    proxy_set_header Authorization $http_authorization;
    proxy_set_header Connection "";
    proxy_buffering off;
    proxy_cache off;
    gzip off;
    proxy_read_timeout 240s;
}

这里 proxy_pass 后面没有额外的 URI，原请求的 /v1/ 路径会保留。外层站点继续使用 HTTPS。

proxy_read_timeout 是两次读取上游数据之间允许的等待时间，不是整个任务的总时长。图在检索或分析阶段可能暂时没有正文，所以它要与服务端执行超时一起设置；如果前面还有 CDN 或其他代理，也要确认那一层的空闲超时和缓冲设置。

调试时先看直连网关，再看经过代理的响应。如果直连可以逐段输出、代理后只能整段显示，就先检查代理路径；如果两边都要等到最后才输出，再检查最终生成节点及上游模型是否实际产生了流式片段。