Cloudflare WorkersによるServerless Edge AI: D1、Vectorize、Streaming RAGガイド (2026)

npm install -g wrangler
wrangler login

wrangler d1 create my-rag-metadata
# Output will include a binding name, e.g., D1_BINDING_NAME

# wrangler.toml
name = "edge-rag-worker"
main = "src/index.ts"
compatibility_date = "2024-10-27"
compatibility_flags = ["nodejs_compat"]

[[d1_databases]]
binding = "DB" # This is the binding name you'll use in your Worker
database_name = "my-rag-metadata"
database_id = "<YOUR_D1_DATABASE_ID>" # From `wrangler d1 create` output

-- migrations/0001_initial_schema.sql
CREATE TABLE IF NOT EXISTS document_chunks (
    id TEXT PRIMARY KEY,
    document_id TEXT NOT NULL,
    chunk_text TEXT NOT NULL,
    metadata JSON,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX IF NOT EXISTS idx_document_id ON document_chunks (document_id);

wrangler d1 migrations apply my-rag-metadata --local --env production

wrangler vectorize create my-rag-vectors --dimensions=768 --metric=cosine
# Output will include a binding name, e.g., VECTORIZE_BINDING_NAME

# wrangler.toml (continued)
[[vectorize]]
binding = "VECTORIZE_INDEX" # This is the binding name you'll use in your Worker
index_name = "my-rag-vectors"

// src/ingest.ts
import { Hono } from 'hono';
import { env } from 'hono/adapter';
import { Bindings } from './types'; // Define your Bindings interface

const ingestApp = new Hono();

// Helper to chunk text (simplified for brevity)
function chunkText(text: string, chunkSize: number = 512, overlap: number = 50): string[] {
  const chunks: string[] = [];
  let i = 0;
  while (i < text.length) {
    const end = Math.min(i + chunkSize, text.length);
    chunks.push(text.substring(i, end));
    i += chunkSize - overlap;
    if (i < 0) i = 0; // Prevent negative index if overlap > chunkSize
  }
  return chunks;
}

ingestApp.post('/ingest', async (c) => {
  const { DB, VECTORIZE_INDEX, AI } = env<Bindings>(c);
  const { documentId, text, metadata } = await c.req.json();

  if (!documentId || !text) {
    return c.json({ error: 'documentId and text are required' }, 400);
  }

  const chunks = chunkText(text);
  const vectorsToInsert: { id: string; values: number[]; metadata: Record<string, any> }[] = [];
  const dbInserts: { id: string; document_id: string; chunk_text: string; metadata: string }[] = [];

  for (const chunk of chunks) {
    // Generate embedding using Workers AI
    const { data } = await AI.run('@cf/baai/bge-small-en-v1.5', { text: chunk });
    const embedding = data[0];

    const chunkId = crypto.randomUUID(); // Unique ID for each chunk
    vectorsToInsert.push({
      id: chunkId,
      values: embedding,
      metadata: { documentId, ...metadata }, // Store relevant metadata with vector
    });
    dbInserts.push({
      id: chunkId,
      document_id: documentId,
      chunk_text: chunk,
      metadata: JSON.stringify({ documentId, ...metadata }),
    });
  }

  // Insert into Vectorize
  const vectorizeResult = await VECTORIZE_INDEX.upsert(vectorsToInsert);

  // Insert into D1
  const stmt = DB.prepare(
    'INSERT INTO document_chunks (id, document_id, chunk_text, metadata) VALUES (?, ?, ?, ?)'
  );
  const dbResult = await DB.batch(dbInserts.map((d) => stmt.bind(d.id, d.document_id, d.chunk_text, d.metadata)));

  return c.json({
    message: `Ingested ${chunks.length} chunks for document ${documentId}`,
    vectorizeResult,
    dbResult,
  });
});

export default ingestApp;

// src/index.ts
import { Hono } from 'hono';
import { streamText, streamSSE } from 'hono/streaming';
import { env } from 'hono/adapter';
import { Bindings } from './types'; // Define your Bindings interface

const app = new Hono();

// Define the Bindings interface for better type safety
export interface Bindings {
  DB: D1Database;
  VECTORIZE_INDEX: VectorizeIndex;
  AI: Ai;
  GOOGLE_GEMINI_API_KEY?: string; // Optional for Gemini direct access
}

app.get('/', (c) => c.text('Edge RAG Worker is running!'));

app.post('/ask', async (c) => {
  const { DB, VECTORIZE_INDEX, AI, GOOGLE_GEMINI_API_KEY } = env<Bindings>(c);
  const { query } = await c.req.json();

  if (!query) {
    return c.json({ error: 'Query is required' }, 400);
  }

  return streamSSE(c, async (stream) => {
    try {
      await stream.writeSSE({ event: 'status', data: 'Generating query embedding...' });
      // 1. Generate embedding for the query
      const { data: queryEmbedding } = await AI.run('@cf/baai/bge-small-en-v1.5', { text: query });
      const embedding = queryEmbedding[0];

      await stream.writeSSE({ event: 'status', data: 'Searching Vectorize...' });
      // 2. Search Vectorize for similar document chunks
      const searchResults = await VECTORIZE_INDEX.query(embedding, { topK: 5, returnMetadata: true });

      if (!searchResults.matches || searchResults.matches.length === 0) {
        await stream.writeSSE({ event: 'error', data: 'No relevant documents found.' });
        await stream.close();
        return;
      }

      await stream.writeSSE({ event: 'status', data: 'Retrieving context from D1...' });
      // 3. Retrieve full chunk text from D1 using the IDs
      const chunkIds = searchResults.matches.map((match) => match.id);
      const { results: dbResults } = await DB.prepare(
        `SELECT chunk_text FROM document_chunks WHERE id IN (${chunkIds.map(() => '?').join(',')})`
      )
        .bind(...chunkIds)
        .all<{ chunk_text: string }>();

      const context = dbResults.map((row) => row.chunk_text).join('\n\n');

      await stream.writeSSE({ event: 'status', data: 'Generating answer with LLM...' });
      // 4. Assemble prompt for LLM
      const prompt = `You are a helpful AI assistant. Use the following context to answer the question. If you don't know the answer, state that you don't know.

      Context:
      ${context}

      Question: ${query}

      Answer:`;

      // 5. Stream LLM response
      // Option A: Workers AI (recommended for simplicity and edge performance)
      const response = await AI.run('@cf/mistral/mistral-7b-instruct-v0.1', {
        prompt,
        stream: true,
      });

      // Option B: Google Gemini (requires API key and direct fetch)
      // if (GOOGLE_GEMINI_API_KEY) {
      //   const geminiResponse = await fetch('https://generativelanguage.googleapis.com/v1beta/models/gemini-pro:streamGenerateContent', {
      //     method: 'POST',
      //     headers: {
      //       'Content-Type': 'application/json',
      //       'x-goog-api-key': GOOGLE_GEMINI_API_KEY,
      //     },
      //     body: JSON.stringify({
      //       contents: [{ parts: [{ text: prompt }] }],
      //       generationConfig: {
      //         stream: true,
      //       },
      //     }),
      //   });

      //   if (!geminiResponse.ok) {
      //     throw new Error(`Gemini API error: ${geminiResponse.statusText}`);
      //   }

      //   // Parse and stream Gemini's SSE-like response
      //   const reader = geminiResponse.body?.getReader();
      //   if (!reader) throw new Error('Failed to get reader from Gemini response');

      //   const decoder = new TextDecoder();
      //   while (true) {
      //     const { done, value } = await reader.read();
      //     if (done) break;
      //     const chunk = decoder.decode(value, { stream: true });
      //     // Gemini's streaming format is complex, often JSON objects per line.
      //     // This is a simplified parse. Production code needs robust parsing.
      //     const lines = chunk.split('\n').filter(Boolean);
      //     for (const line of lines) {
      //       try {
      //         const json = JSON.parse(line);
      //         if (json.candidates && json.candidates[0] && json.candidates[0].content && json.candidates[0].content.parts) {
      //           const textPart = json.candidates[0].content.parts[0].text;
      //           await stream.writeSSE({ event: 'data', data: textPart });
      //         }
      //       } catch (e) {
      //         console.error('Failed to parse Gemini stream chunk:', e, chunk);
      //       }
      //     }
      //   }
      //   await stream.writeSSE({ event: 'status', data: 'Stream complete.' });
      //   await stream.close();
      //   return;
      // }

      // Workers AI streaming
      const reader = response.body?.getReader();
      if (!reader) throw new Error('Failed to get reader from AI response');

      const decoder = new TextDecoder();
      while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        const chunk = decoder.decode(value, { stream: true });
        await stream.writeSSE({ event: 'data', data: chunk });
      }

      await stream.writeSSE({ event: 'status', data: 'Stream complete.' });
      await stream.close();
    } catch (error: any) {
      console.error('RAG stream error:', error);
      await stream.writeSSE({ event: 'error', data: `An error occurred: ${error.message}` });
      await stream.close();
    }
  });
});

export default app;

# wrangler.toml (continued)
[ai]
binding = "AI" # This is the binding name you'll use in your Worker

# wrangler.toml (continued)
[vars]
GOOGLE_GEMINI_API_KEY = "<YOUR_GEMINI_API_KEY>" # Set this in production via `wrangler secret put`

<!-- public/index.html -->
<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Edge RAG Chat</title>
    <style>
        body { font-family: sans-serif; margin: 20px; }
        #chat-container { max-width: 800px; margin: auto; border: 1px solid #ccc; padding: 15px; border-radius: 8px; }
        #messages { height: 400px; overflow-y: scroll; border: 1px solid #eee; padding: 10px; margin-bottom: 10px; background-color: #f9f9f9; }
        .message { margin-bottom: 8px; }
        .user-message { text-align: right; color: blue; }
        .ai-message { text-align: left; color: green; }
        #query-input { width: calc(100% - 80px); padding: 8px; border: 1px solid #ccc; border-radius: 4px; }
        #send-button { width: 70px; padding: 8px; margin-left: 5px; background-color: #007bff; color: white; border: none; border-radius: 4px; cursor: pointer; }
        #send-button:disabled { background-color: #cccccc; cursor: not-allowed; }
    </style>
</head>
<body>
    <div id="chat-container">
        <h1>Edge RAG Chat</h1>
        <div id="messages"></div>
        <input type="text" id="query-input" placeholder="Ask me anything...">
        <button id="send-button">Send</button>
    </div>

    <script>
        const queryInput = document.getElementById('query-input');
        const sendButton = document.getElementById('send-button');
        const messagesDiv = document.getElementById('messages');

        async function sendMessage() {
            const query = queryInput.value.trim();
            if (!query) return;

            appendMessage('user', query);
            queryInput.value = '';
            sendButton.disabled = true;

            const aiMessageDiv = appendMessage('ai', 'Thinking...');
            let fullResponse = '';

            try {
                const eventSource = new EventSource('/ask'); // Adjust endpoint if needed
                eventSource.onopen = () => {
                    console.log('SSE connection opened.');
                    // Send the query immediately after connection opens
                    fetch('/ask', {
                        method: 'POST',
                        headers: { 'Content-Type': 'application/json' },
                        body: JSON.stringify({ query }),
                    }).catch(error => {
                        console.error('Error sending query:', error);
                        eventSource.close();
                        aiMessageDiv.textContent = `Error: ${error.message}`;
                        sendButton.disabled = false;
                    });
                };

                eventSource.onmessage = (event) => {
                    const data = JSON.parse(event.data);
                    if (data.event === 'data') {
                        fullResponse += data.data;
                        aiMessageDiv.textContent = fullResponse;
                        messagesDiv.scrollTop = messagesDiv.scrollHeight; // Scroll to bottom
                    } else if (data.event === 'status') {
                        console.log('Status:', data.data);
                        // Optionally update a status indicator
                    } else if (data.event === 'error') {
                        console.error('Worker Error:', data.data);
                        aiMessageDiv.textContent = `Error: ${data.data}`;
                        eventSource.close();
                    } else if (data.event === 'stream_complete') {
                        console.log('Stream complete.');
                        eventSource.close();
                    }
                };

                eventSource.onerror = (error) => {
                    console.error('SSE Error:', error);
                    eventSource.close();
                    if (aiMessageDiv.textContent === 'Thinking...') {
                        aiMessageDiv.textContent = 'Error: Failed to connect or stream.';
                    }
                    sendButton.disabled = false;
                };

                // Note: EventSource doesn't support POST directly.
                // The above `fetch` call is a workaround to send the initial POST data.
                // For a true SSE POST, you'd need a custom fetch-based streaming solution
                // or pass the query in the URL for GET (less secure/flexible).
                // Cloudflare Workers can handle the POST and then initiate SSE.
                // The current setup assumes the Worker will process the POST and then
                // use the established SSE connection for streaming. This is a common pattern.

            } catch (error) {
                console.error('Fetch error:', error);
                aiMessageDiv.textContent = `Error: ${error.message}`;
            } finally {
                sendButton.disabled = false;
            }
        }

        function appendMessage(sender, text) {
            const messageDiv = document.createElement('div');
            messageDiv.classList.add('message', `${sender}-message`);
            messageDiv.textContent = text;
            messagesDiv.appendChild(messageDiv);
            messagesDiv.scrollTop = messagesDiv.scrollHeight;
            return messageDiv; // Return the div to update it later
        }

        sendButton.addEventListener('click', sendMessage);
        queryInput.addEventListener('keypress', (e) => {
            if (e.key === 'Enter') {
                sendMessage();
            }
        });
    </script>
</body>
</html>