# 龙霄 - llms-full.txt ## 文章 # Implementing AI Streaming Output with GraphQL in Headless WordPress **URL:** https://lx.yfdxs.com/7328.html ## Introduction When building modern web applications, the Headless WordPress + Nuxt.js architecture combination is increasingly favored by developers. WordPress provides powerful content management on the backend, Nuxt.js handles high-performance frontend rendering, and GraphQL serves as the data bridge between them. However, when we need to display AI-generated content on the frontend — especially in streaming scenarios — traditional REST API approaches fall short. This article provides a detailed guide on implementing AI content streaming through GraphQL in this architecture. ## Why GraphQL + SSE? Traditional REST APIs face several core challenges when handling AI streaming output: **Request-Response Model Limitations**: REST follows a one-request-one-response pattern and cannot continuously push data. Streaming typically requires WebSocket, which adds architectural complexity and maintenance overhead. **Data Over-fetching**: REST endpoints return fixed data structures. The frontend may only need a few fields but must receive large amounts of redundant information. In a Headless architecture, pages often require multiple REST endpoints to assemble complete data, increasing network overhead and page load times. **Limited Extensibility**: When upgrading AI services or switching models, REST interfaces often require synchronized modifications, creating tight coupling between frontend and backend. GraphQL’s advantages lie in on-demand querying — the frontend precisely specifies needed fields, reducing data transfer; all data is fetched through a single GraphQL endpoint, eliminating multiple requests; and custom Mutations and Subscriptions enable flexible AI capability extension. Combined with SSE (Server-Sent Events), we can achieve efficient real-time data pushing within the GraphQL framework. ## Architecture Design The overall architecture consists of three clearly decoupled layers: ``` Nuxt.js Frontend (SSE Consumer) ↕ SSE (text/event-stream) WordPress + WPGraphQL Plugin ↕ GraphQL Mutation + REST SSE Endpoint AI Service (DeepSeek / OpenAI / Claude) ``` On the WordPress side, we register a custom Mutation through the WPGraphQL plugin. This Mutation receives the user’s prompt, calls the AI service, and returns a stream ID. The frontend then consumes the streaming data through a dedicated SSE endpoint. This design organically combines GraphQL’s query capabilities with SSE’s real-time pushing, with each component playing to its strengths. ## Backend Implementation: WordPress + WPGraphQL ### Step 1: Register a GraphQL Mutation First, register a custom Mutation in your theme or plugin’s functions.php to receive prompts and generate stream IDs: ``` 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]; }, ]); }); ``` We use the WordPress Transients API to temporarily store request information with a 5-minute expiration to prevent memory leaks. ### Step 2: Create the SSE Endpoint Register a custom REST API endpoint to stream AI-generated content. Key points: set correct response headers, use cURL’s CURLOPT\_WRITEFUNCTION for chunked data transfer, and disable Nginx buffering: ``` add_action('rest_api_init', function () { register_rest_route('longxiao/v1', '/ai/stream/(?P[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; }, ]); }); ``` Important: the CURLOPT\_WRITEFUNCTION callback must use ob\_flush() and flush() to ensure real-time data output to the client. Always delete the transient after processing to free resources. ## Frontend Implementation: Nuxt.js Composable On the Nuxt.js side, we create a Composable to handle GraphQL requests and SSE stream consumption. The core approach has two steps: first, get the streamId through a GraphQL Mutation; second, consume the SSE stream via the fetch API: ``` // composables/useAiStream.ts export function useAiStream() { const content = ref('') const isStreaming = ref(false) const error = ref(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('Failed to get stream 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('Failed to read response stream') 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 || 'Streaming error' } } finally { isStreaming.value = false } } return { content, isStreaming, error, streamGenerate } } ``` Key design decisions: using AbortController for 2-minute timeout protection; reactive binding via ref for automatic Vue component UI updates; distinguishing AbortError from other errors to avoid showing error messages on timeout. ### Vue Component Integration ```