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 對話應用。