WordPress Headless 架构下基于 GraphQL 实现 AI 流式输出

前言

在构建现代 Web 应用时，Headless WordPress + Nuxt.js 的架构组合越来越受到开发者的青睐。WordPress 提供强大的内容管理后端，Nuxt.js 负责高性能的前端渲染，而 GraphQL 则作为两者之间的数据桥梁。但当我们需要在前端展示 AI 生成的内容时——尤其是需要流式输出（Streaming）的场景，传统 REST API 方案就显得力不从心了。本文将详细介绍如何在这一架构中通过 GraphQL 实现 AI 内容的流式输出。

为什么选择 GraphQL + SSE？

传统的 REST API 在处理 AI 流式输出时面临几个核心问题：

请求-响应模型限制：REST 是一次请求一次响应，无法持续推送数据。要实现流式输出，通常需要借助 WebSocket，但这会增加架构的复杂度和维护成本。

数据过载：REST 接口返回固定结构的数据，前端可能只需要部分字段，却不得不同时接收大量冗余信息。在 Headless 架构中，页面通常需要请求多个 REST 端点才能组装完整数据，这无疑增加了网络开销和页面加载时间。

扩展性不足：当 AI 服务升级或切换模型时，REST 接口往往需要同步修改，前后端耦合度高。

GraphQL 的优势在于按需查询——前端精确指定所需字段，减少数据传输量；所有数据通过一个 GraphQL 端点获取，免去多次请求的烦恼；通过自定义 Mutation 和 Subscription 可以灵活扩展 AI 能力。结合 SSE（Server-Sent Events），我们可以在 GraphQL 框架内实现高效的实时数据推送。

架构设计

整体架构分为三层，清晰且解耦：

Nuxt.js 前端 (SSE Consumer)
    ↕ SSE (text/event-stream)
WordPress + WPGraphQL 插件
    ↕ GraphQL Mutation + REST SSE Endpoint
AI 服务 (DeepSeek / OpenAI / Claude)

在 WordPress 端，我们通过 WPGraphQL 插件注册自定义 Mutation，该 Mutation 接收用户输入的 prompt，调用 AI 服务并返回一个流 ID。前端拿到流 ID 后，通过独立的 SSE 端点消费流式数据。这种设计将 GraphQL 的查询能力与 SSE 的实时推送能力有机结合，各司其职。

后端实现：WordPress + WPGraphQL

第一步：注册 GraphQL Mutation

首先在主题或插件的 functions.php 中注册一个自定义 Mutation，用于接收 prompt 并生成流 ID：

add_action('graphql_register_types', function () {
    register_graphql_mutation('generateAiContent', [
        'inputFields' => [
            'prompt' => ['type' => 'String'],
            'model'  => ['type' => 'String'],
        ],
        'outputFields' => [
            'streamId' => ['type' => 'String'],
        ],
        'mutateAndGetPayload' => function ($input) {
            $prompt = sanitize_text_field($input['prompt']);
            $model  = $input['model'] ?? 'deepseek-chat';
            $stream_id = wp_generate_uuid4();
            set_transient("ai_stream_{$stream_id}", [
                'prompt' => $prompt,
                'model'  => $model,
            ], 300);
            return ['streamId' => $stream_id];
        },
    ]);
});

这里使用 WordPress Transients API 临时存储请求信息，设置 5 分钟过期时间，避免内存泄漏。

第二步：创建 SSE 端点

注册一个自定义 REST API 端点来流式输出 AI 生成内容。关键点：设置正确的响应头、使用 cURL 的 CURLOPT_WRITEFUNCTION 逐块传输数据、禁用 Nginx 缓冲：

add_action('rest_api_init', function () {
    register_rest_route('longxiao/v1', '/ai/stream/(?P<id>[a-zA-Z0-9-]+)', [
        'methods'  => 'GET',
        'callback' => function ($request) {
            $stream_id = $request->get_param('id');
            $data = get_transient("ai_stream_{$stream_id}");
            if (!$data) {
                return new WP_Error('not_found', 'Stream not found', ['status' => 404]);
            }

            header('Content-Type: text/event-stream');
            header('Cache-Control: no-cache');
            header('Connection: keep-alive');
            header('X-Accel-Buffering: no');

            $ch = curl_init();
            curl_setopt_array($ch, [
                CURLOPT_URL => 'https://api.deepseek.com/v1/chat/completions',
                CURLOPT_POST => true,
                CURLOPT_HTTPHEADER => [
                    'Authorization: Bearer ' . DEEPSEEK_API_KEY,
                    'Content-Type: application/json',
                ],
                CURLOPT_POSTFIELDS => json_encode([
                    'model'    => $data['model'],
                    'messages' => [['role' => 'user', 'content' => $data['prompt']]],
                    'stream'   => true,
                ]),
                CURLOPT_WRITEFUNCTION => function ($ch, $chunk) {
                    $lines = explode("n", $chunk);
                    foreach ($lines as $line) {
                        $line = trim($line);
                        if (str_starts_with($line, 'data: ')) {
                            $json = substr($line, 6);
                            if ($json === '[DONE]') {
                                echo "data: [DONE]nn";
                            } else {
                                $decoded = json_decode($json, true);
                                $content = $decoded['choices'][0]['delta']['content'] ?? '';
                                if ($content) {
                                    echo "data: " . json_encode(['content' => $content]) . "nn";
                                }
                            }
                            ob_flush();
                            flush();
                        }
                    }
                    return strlen($chunk);
                },
            ]);
            curl_exec($ch);
            curl_close($ch);
            delete_transient("ai_stream_{$stream_id}");
            exit;
        },
    ]);
});

需要注意：CURLOPT_WRITEFUNCTION 回调中必须使用 ob_flush() 和 flush() 确保数据实时输出到客户端；处理完请求后及时删除 transient 释放资源。

前端实现：Nuxt.js Composable

在 Nuxt.js 端，封装一个 Composable 来处理 GraphQL 请求和 SSE 流消费。核心思路分两步：先通过 GraphQL Mutation 获取 streamId，再通过 fetch API 消费 SSE 流：

// composables/useAiStream.ts
export function useAiStream() {
  const content = ref('')
  const isStreaming = ref(false)
  const error = ref<string | null>(null)

  async function streamGenerate(prompt: string, model = 'deepseek-chat') {
    content.value = ''
    isStreaming.value = true
    error.value = null

    try {
      const mutation = `
        mutation GenerateAiContent($prompt: String!, $model: String!) {
          generateAiContent(input: { prompt: $prompt, model: $model }) {
            streamId
          }
        }
      `
      const { data } = await useGql({ query: mutation, variables: { prompt, model } })
      const streamId = data.value?.generateAiContent?.streamId
      if (!streamId) throw new Error('无法获取流 ID')

      const streamUrl = `https://api.your-site.com/wp-json/longxiao/v1/ai/stream/${streamId}`
      const controller = new AbortController()
      const timeout = setTimeout(() => controller.abort(), 120000)

      const response = await fetch(streamUrl, { signal: controller.signal })
      const reader = response.body?.getReader()
      const decoder = new TextDecoder()
      if (!reader) throw new Error('无法读取响应流')

      while (true) {
        const { done, value } = await reader.read()
        if (done) break
        const chunk = decoder.decode(value, { stream: true })
        const lines = chunk.split('n')
        for (const line of lines) {
          if (line.startsWith('data: ')) {
            const data = line.slice(6)
            if (data === '[DONE]') break
            try {
              const parsed = JSON.parse(data)
              content.value += parsed.content || ''
            } catch {}
          }
        }
      }
      clearTimeout(timeout)
    } catch (e: any) {
      if (e.name !== 'AbortError') {
        error.value = e.message || '流式输出异常'
      }
    } finally {
      isStreaming.value = false
    }
  }

  return { content, isStreaming, error, streamGenerate }
}

关键设计：使用 AbortController 实现 2 分钟超时保护；通过 ref 响应式绑定让 Vue 组件自动更新 UI；区分 AbortError 和其他错误，避免超时时显示错误提示。

Vue 组件集成

<template>
  <div class="ai-chat">
    <textarea v-model="prompt" placeholder="请输入你的问题..." />
    <button @click="handleGenerate" :disabled="isStreaming">
      {{ isStreaming ? '生成中...' : '发送' }}
    </button>
    <div class="ai-output" v-html="renderedContent" />
    <p v-if="error" class="error">{{ error }}</p>
  </div>
</template>

<script setup>
const prompt = ref('')
const { content, isStreaming, error, streamGenerate } = useAiStream()
const renderedContent = computed(() => content.value.replace(/n/g, '<br>'))
function handleGenerate() {
  if (!prompt.value.trim() || isStreaming.value) return
  streamGenerate(prompt.value)
}
</script>

性能优化与生产实践

Nginx 缓冲配置

在 Nginx 反向代理中默认会缓冲后端响应，导致 SSE 流无法实时推送。必须对 SSE 端点禁用缓冲：

location /wp-json/longxiao/v1/ai/stream/ {
    proxy_buffering off;
    proxy_cache off;
    proxy_set_header X-Accel-Buffering no;
    proxy_read_timeout 300s;
    chunked_transfer_encoding on;
}

连接池与并发控制

在高并发场景下，建议使用 Redis 管理 AI 请求队列，避免大量请求同时打到 AI 服务导致限流或超时。可以在 WordPress 端实现一个简单的令牌桶算法来控制请求速率。

错误处理与重连

SSE 连接可能因网络波动而中断，建议在 Composable 中实现自动重连逻辑。当检测到连接异常断开时，可以携带已接收的内容长度发起续传请求，避免用户看到内容中断。

总结

通过 WPGraphQL + SSE 的组合方案，我们成功在 WordPress Headless 架构中实现了 AI 内容的流式输出。这套方案的优势在于架构简洁——无需额外引入 WebSocket 服务器，降低了运维成本；渐进式渲染——用户无需等待完整响应即可看到内容逐字出现，交互体验极佳；灵活扩展——可以轻松切换 DeepSeek、OpenAI、Claude 等不同 AI 模型；与现有技术栈无缝集成——充分利用了 WordPress 的插件生态和 Nuxt.js 的全栈能力。在实际项目中，你还可以进一步扩展这套架构，比如集成 Markdown 渲染、代码高亮、历史对话管理等功能，构建完整的 AI 对话应用。