前言
在構建現代 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 對話應用。
