วิธีใช้ Streaming Response กับ Gemini API

Streaming Response คือการรับคำตอบจาก Gemini API ทีละส่วนขณะที่โมเดลกำลังสร้างคำตอบ แทนการรอให้ Gemini สร้างข้อความทั้งหมดเสร็จแล้วจึงส่ง Response กลับมาครั้งเดียว

เหมาะอย่างมากกับ Chatbot, AI Assistant, Coding Assistant และ Application ที่คำตอบค่อนข้างยาว เพราะผู้ใช้สามารถเริ่มอ่านข้อความได้ทันที ทำให้ระบบรู้สึกตอบสนองเร็วขึ้น

ตัวอย่างแบบไม่ Streaming

User ส่งคำถาม
↓
Gemini ประมวลผล
↓
สร้างคำตอบทั้งหมด
↓
รอ...
↓
รอ...
↓
ส่งคำตอบทั้งหมดกลับ

แบบ Streaming

User ส่งคำถาม
↓
Gemini เริ่มสร้างคำตอบ
↓
ข้อความส่วนที่ 1
↓
ข้อความส่วนที่ 2
↓
ข้อความส่วนที่ 3
↓
...
↓
คำตอบสมบูรณ์

สำหรับ Interactions API รุ่นปัจจุบัน การเปิด Streaming ทำได้ด้วย

stream=True

ใน Python หรือ

stream: true

ใน JavaScript

และ Response จะถูกส่งมาเป็น Event ตามลำดับ เช่น

interaction.created
step.start
step.delta
step.delta
step.stop
interaction.completed

สำหรับข้อความทั่วไป สิ่งที่ Developer ใช้บ่อยที่สุดคือ

step.delta

เพราะแต่ละ Event จะมีข้อความส่วนใหม่ที่สามารถส่งไปแสดงให้ผู้ใช้ได้ทันที

❶ 🌊 Streaming Response คืออะไร

สมมติ Gemini ต้องตอบข้อความประมาณ 1,000 คำ

แบบปกติ User อาจต้องรอจนโมเดลสร้างครบ

1,000 คำ

แล้วจึงเห็นคำตอบ

แต่ Streaming จะทยอยส่ง

คำที่ 1–20
↓
คำที่ 21–40
↓
คำที่ 41–60
↓
...

ไปยัง Application

จึงลด Perceived Latency หรือความรู้สึกว่าระบบกำลังค้าง

❷ ⚡ Streaming ช่วยให้ Gemini เร็วขึ้นจริงไหม

ต้องแยกเป็นสองเรื่อง

Time to First Output

ผู้ใช้เริ่มเห็นคำตอบเร็วขึ้น

Total Generation Time

โมเดลยังต้องสร้างคำตอบทั้งหมดอยู่

ดังนั้น Streaming ไม่ได้หมายความว่า

Model สร้างคำตอบทั้งหมดเร็วขึ้น 10 เท่า

แต่หมายถึง

User ไม่ต้องรอคำตอบทั้งหมดก่อนเริ่มอ่าน

นี่เป็นเหตุผลที่ Chat Application ส่วนใหญ่เหมาะกับ Streaming

❸ 🎯 Streaming เหมาะกับงานอะไร

เหมาะมากกับ

AI Chat

User
↔
Gemini

Coding Assistant

สร้าง Code ยาว ๆ ทีละส่วน

Article Generator

เริ่มแสดงบทความก่อน Generation จบ

Research Assistant

แสดง Progress และคำตอบระหว่างทำงาน

Agent

แสดง Steps หรือ Tool Activity

Long Explanation

เช่น Tutorial หรือ Report

หาก Response สั้นมาก เช่น

{
  "status": "ok"
}

Streaming อาจไม่ได้สร้างประโยชน์มากนัก

❹ 🐍 วิธีใช้ Streaming กับ Gemini API ด้วย Python

SDK ปัจจุบันคือ

google-genai

ติดตั้ง

pip install -U google-genai

จากนั้น

from google import genai

client = genai.Client()

stream = client.interactions.create(
    model="gemini-3.7-flash",
    input="อธิบายว่า AI ทำงานอย่างไร",
    stream=True,
)

for event in stream:
    if event.event_type == "step.delta":
        if event.delta.type == "text":
            print(
                event.delta.text,
                end="",
                flush=True,
            )

แทนที่จะได้ String ใหญ่ก้อนเดียว เราจะได้รับ Stream ของ Events

❺ 🔍 อธิบาย Python Code

ส่วนสำคัญคือ

stream=True

ทำให้ Interactions API ส่ง Response แบบ Stream

จากนั้น

for event in stream:

จะอ่าน Event ที่เข้ามาทีละตัว

ตรวจ

event.event_type == "step.delta"

เพื่อหา Incremental Update

และ

event.delta.type == "text"

เพื่อเลือกเฉพาะข้อความ

สุดท้าย

print(
    event.delta.text,
    end="",
    flush=True,
)

จะแสดงข้อความทันทีโดยไม่ขึ้นบรรทัดใหม่ทุก Chunk

❻ 🟨 Streaming ด้วย JavaScript

ติดตั้ง SDK

npm install @google/genai

ตัวอย่าง

import { GoogleGenAI } from "@google/genai";

const ai = new GoogleGenAI({});

const stream =
  await ai.interactions.create({
    model: "gemini-3.7-flash",
    input: "อธิบายว่า AI ทำงานอย่างไร",
    stream: true,
  });

for await (const event of stream) {
  if (
    event.event_type === "step.delta" &&
    event.delta.type === "text"
  ) {
    process.stdout.write(
      event.delta.text
    );
  }
}

จุดสำคัญคือ

for await (const event of stream)

เพราะ Stream เป็น Async Iterable

❼ 🆚 Streaming กับ Non-streaming Code ต่างกันตรงไหน

Non-streaming

interaction = client.interactions.create(
    model="gemini-3.7-flash",
    input="อธิบาย AI",
)

print(interaction.output_text)

รอจน Interaction เสร็จก่อน

Streaming

stream = client.interactions.create(
    model="gemini-3.7-flash",
    input="อธิบาย AI",
    stream=True,
)

for event in stream:
    ...

รับข้อมูลระหว่าง Generation

❽ 📡 Interactions API Streaming ใช้ SSE

Gemini Interactions API ใช้

Server-Sent Events หรือ SSE

SSE คือวิธีที่ Server สามารถส่ง Event หลายชุดผ่าน HTTP Connection เดียว

แนวคิด

Client
↓
HTTP Request
↓
Server เปิด Connection ไว้
↓
event 1
↓
event 2
↓
event 3
↓
event 4
↓
Connection จบ

เหมาะกับการส่งข้อมูลทางเดียวจาก Server มายัง Client อย่างต่อเนื่อง

❾ 🌐 REST API ใช้ Streaming อย่างไร

สำหรับ REST Interactions API ใช้ Endpoint เดิมและเพิ่ม

/v1beta/interactions?alt=sse

พร้อม Request Body

{
  "model": "gemini-3.7-flash",
  "input": "อธิบายว่า AI ทำงานอย่างไร",
  "stream": true
}

แนวคิดสำคัญคือ

Interactions API
=
Endpoint เดิม
+
stream: true

ไม่จำเป็นต้องใช้ Endpoint Generation แยกสำหรับ Streaming แบบโครงสร้าง API รุ่นเก่าบางแบบ

❿ 🆕 Interactions API ต่างจาก streamGenerateContent

ใน generateContent API แบบเดิม Developer อาจคุ้นกับ

streamGenerateContent

หรือ SDK

generate_content_stream()

แต่เมื่อใช้ Interactions API รุ่นปัจจุบัน Google เปลี่ยนแนวทางเป็น

interactions.create()
+
stream=True

ดังนั้น Tutorial เก่าที่ใช้ GenerateContent Streaming ไม่ได้หมายความว่าผิด แต่เป็น API คนละ Interface

สำหรับ Project ใหม่ควรรู้ว่ากำลังใช้ API แบบใดก่อน Copy Code

⓫ 🧩 Event ใน Streaming มีอะไรบ้าง

Response ไม่ได้มีแค่ Text Chunk

Interactions API ใช้ Event เพื่ออธิบาย Lifecycle ของงาน

Event สำคัญประกอบด้วย

interaction.created
step.start
step.delta
step.stop
interaction.completed

และใน SSE Stream จะจบด้วย Done Event ตาม Protocol

แต่ละ Event มีหน้าที่ต่างกัน

⓬ 🟢 interaction.created

Event แรก ๆ จะบอกว่า Interaction ถูกสร้างแล้ว

เช่น

interaction.created

สามารถใช้เก็บ

• Interaction ID
• Status
• Model

ได้

Interaction ID มีประโยชน์มากกับ

• Conversation
• Function Calling
• Logging
• Debugging
• Stateful Follow-up

⓭ ▶️ step.start

หมายความว่า Gemini กำลังเริ่ม Step ใหม่

ตัวอย่าง Step อาจเป็น

thought

model_output

function_call

หรือ Tool Step ตาม Workflow ที่ใช้

จึงสามารถใช้สร้าง UI เช่น

กำลังวิเคราะห์...
กำลังค้นข้อมูล...
กำลังสร้างคำตอบ...

ตามข้อมูลที่เหมาะสมและ Feature ที่เปิดใช้งาน

⓮ 📝 step.delta

นี่คือ Event ที่สำคัญที่สุดสำหรับ Text Streaming

ตัวอย่าง

step.delta
delta.type = text
delta.text = "Artificial intelligence..."

Event ถัดไปอาจเป็น

delta.text = " can analyze..."

เมื่อเอามาต่อกันจะได้ Final Text

Artificial intelligence can analyze...

⓯ ⏹️ step.stop

หมายความว่า Step นั้นเสร็จแล้ว

ไม่ได้หมายความว่า Interaction ทั้งหมดจบเสมอไป

เพราะ Agent หรือ Tool Workflow สามารถมีหลาย Steps เช่น

Thought
↓
Step Stop
↓
Search
↓
Step Stop
↓
Model Output
↓
Step Stop

ดังนั้นถ้าต้องการรู้ว่างานทั้งหมดเสร็จ ควรดู

interaction.completed

⓰ ✅ interaction.completed

Event นี้หมายความว่า Interaction เสร็จสมบูรณ์

เหมาะกับการทำงาน เช่น

• ปิด Loading
• Save Final Response
• เก็บ Usage
• Update Conversation
• Record Metrics

ตัวอย่าง Python

for event in stream:
    if event.event_type == "step.delta":
        if event.delta.type == "text":
            print(
                event.delta.text,
                end="",
                flush=True,
            )

    elif (
        event.event_type
        == "interaction.completed"
    ):
        print("\n\nเสร็จแล้ว")

⓱ 📊 ดู Token Usage ตอน Streaming ได้ไหม

ได้

เมื่อ Interaction เสร็จ Event

interaction.completed

สามารถมี Usage ของ Interaction

เช่น

total_tokens
total_input_tokens
total_output_tokens

และ Metric เพิ่มตาม Feature/Model ที่ใช้

จึงสามารถเก็บ Token Usage หลัง Stream จบได้

⓲ 🐍 ตัวอย่างเก็บ Usage

from google import genai

client = genai.Client()

stream = client.interactions.create(
    model="gemini-3.7-flash",
    input="อธิบาย Gemini API แบบละเอียด",
    stream=True,
)

for event in stream:
    if event.event_type == "step.delta":
        if event.delta.type == "text":
            print(
                event.delta.text,
                end="",
                flush=True,
            )

    elif (
        event.event_type
        == "interaction.completed"
    ):
        usage = event.interaction.usage

        print(
            "\n\nTotal tokens:",
            usage.total_tokens,
        )

เหมาะกับ Production Cost Monitoring

⓳ 💰 Streaming ถูกกว่า Non-streaming ไหม

ไม่ควรคิดว่า

Streaming
=
ลด Token Cost

โดยอัตโนมัติ

หาก Model สร้าง Input/Output เท่าเดิม Token Usage โดยหลักก็ยังเกี่ยวข้องกับข้อมูลที่ประมวลผลและสร้าง

Streaming เป็น Optimization ด้าน

User Experience
+
Perceived Latency

ไม่ใช่ Discount Pricing Mode แบบ Batch API

ถ้าต้องการลด Cost ต้องดู

• Model
• Tokens
• Context Caching
• Batch
• Prompt
• Output Length

เพิ่มเติม

⓴ ⚡ Time to First Token สำคัญ

Metric ที่ควรเก็บสำหรับ Streaming คือ

TTFT
=
Time To First Token

ในทาง Product สามารถตีความเป็นเวลาตั้งแต่ User กด Submit จนเริ่มเห็นข้อความแรก

ตัวอย่าง

Request sent
10:00:00.000

First text delta
10:00:00.850

TTFT
≈ 850 ms

ส่วน Total Generation Time อาจเป็นอีกหลายวินาที

Streaming ช่วยให้ User เริ่มอ่านตั้งแต่ประมาณ 850 ms ในตัวอย่างนี้

⏱️ วัด TTFT ด้วย Python

import time

from google import genai

client = genai.Client()

start = time.perf_counter()
first_token_time = None

stream = client.interactions.create(
    model="gemini-3.7-flash",
    input="อธิบาย AI อย่างละเอียด",
    stream=True,
)

for event in stream:
    if (
        event.event_type == "step.delta"
        and event.delta.type == "text"
        and event.delta.text
    ):
        if first_token_time is None:
            first_token_time = (
                time.perf_counter()
            )

            print(
                "TTFT:",
                first_token_time - start,
            )

        print(
            event.delta.text,
            end="",
            flush=True,
        )

Production ควรดู

p50 TTFT
p95 TTFT
p99 TTFT

ไม่ใช่เพียงครั้งเดียว

🧠 Streaming กับ Thinking

Gemini รุ่นที่รองรับ Thinking สามารถ Stream Thinking Summary ตาม Configuration ที่รองรับ

เมื่อเปิด

thinking_summaries = auto

Stream สามารถมี Delta ประเภท

thought_summary

และ

thought_signature

แยกจาก Final Text

จึงต้องตรวจ delta.type ให้ถูก ไม่ควรสมมติว่าทุก step.delta เป็นข้อความ Final Answer

🐍 ตัวอย่างแยก Text กับ Thought Summary

from google import genai

client = genai.Client()

stream = client.interactions.create(
    model="gemini-3.7-flash",
    input="แก้โจทย์นี้อย่างเป็นระบบ...",
    generation_config={
        "thinking_summaries": "auto",
    },
    stream=True,
)

for event in stream:
    if event.event_type != "step.delta":
        continue

    if event.delta.type == "text":
        print(
            event.delta.text,
            end="",
            flush=True,
        )

    elif (
        event.delta.type
        == "thought_summary"
    ):
        if event.delta.content.type == "text":
            print(
                event.delta.content.text,
                end="",
                flush=True,
            )

Application จริงควรออกแบบ UI แยก Thought Summary กับ Final Answer หากต้องการแสดงทั้งสอง

⚠️ Thought Summary ไม่ใช่ Final Answer

ไม่ควรนำ

thought_summary

ไปบันทึกเป็นคำตอบ User โดยไม่แยกประเภท

และไม่ควรตีความว่าเป็น Hidden Chain-of-Thought ทั้งหมดของโมเดล

มันเป็น Summary Content ที่ API รองรับสำหรับ Feature ดังกล่าว

🛠️ Streaming กับ Function Calling

Streaming สามารถใช้ร่วมกับ Function Calling ได้

ตัวอย่าง

User
↓
Stream
↓
Gemini สร้าง Function Call
↓
Application รับ Arguments
↓
Execute Function
↓
ส่ง function_result
↓
เปิด Stream ต่อ
↓
Final Answer

แต่ Function Arguments สามารถมาทีละส่วน

ดังนั้น ห้าม Execute Function ก่อน Arguments สมบูรณ์

🧩 Function Arguments อาจมาเป็น Partial Delta

ตัวอย่าง Concept

Chunk 1:
{"order

Chunk 2:
_id":"12

Chunk 3:
345"}

ต้องนำมาต่อก่อน

{
  "order_id": "12345"
}

จากนั้นจึง

1. Parse
2. Validate
3. Authorization
4. Execute

🚫 อย่า Execute Tool จาก Chunk แรก

Code Logic ที่อันตรายคือ

ได้รับ Function Delta
↓
Execute ทันที

เพราะ Arguments อาจยังไม่ครบ

ต้องรอ Tool Call Step สมบูรณ์ตาม Event Lifecycle

🔄 Streaming Function Calling มีสองช่วง

Turn 1

User ส่งคำถาม

Gemini Stream จนต้องการ Function Result

Application

Execute Tool

Turn 2

ส่ง

function_result
+
previous_interaction_id

จากนั้นเปิด

stream=True

อีกรอบเพื่อรับ Final Answer

เป็น Workflow แบบ Stateful ที่เหมาะกับ Agent

🔎 Streaming กับ Google Search

สามารถ Stream Interaction ที่ใช้

google_search

ได้เช่นกัน

Workflow อาจมี Steps

interaction.created
↓
thought
↓
google_search_call
↓
google_search_result
↓
model_output
↓
interaction.completed

Application สามารถแสดง Final Text ทีละส่วน ขณะเดียวกันเก็บ Search/Citation Metadata จาก Steps ที่เกี่ยวข้อง

🧰 Streaming หลาย Tools

ใน Agent Workflow ที่ใช้

Google Search
+
Custom Function

Stream อาจประกอบด้วยหลาย Step Types

ดังนั้น Code Production ไม่ควรเขียนว่า

ทุก Event
=
Text

แต่ควรสร้าง Event Router

🧱 Event Router คืออะไร

ตัวอย่างแนวคิด

def handle_event(event):
    if (
        event.event_type
        == "interaction.created"
    ):
        handle_created(event)

    elif (
        event.event_type
        == "step.start"
    ):
        handle_step_start(event)

    elif (
        event.event_type
        == "step.delta"
    ):
        handle_delta(event)

    elif (
        event.event_type
        == "step.stop"
    ):
        handle_step_stop(event)

    elif (
        event.event_type
        == "interaction.completed"
    ):
        handle_completed(event)

ช่วยให้เพิ่ม Tool และ Agent ภายหลังง่ายกว่า

📦 Structured Output ใช้ Streaming ได้ไหม

ได้

Gemini API รองรับ Structured Output พร้อม Streaming

ตัวอย่าง

from typing import Literal

from google import genai
from pydantic import BaseModel


class Feedback(BaseModel):
    sentiment: Literal[
        "positive",
        "neutral",
        "negative",
    ]
    summary: str


client = genai.Client()

stream = client.interactions.create(
    model="gemini-3.7-flash",
    input=(
        "สินค้าใช้งานง่าย "
        "แต่จัดส่งช้ามาก"
    ),
    response_format={
        "type": "text",
        "mime_type": "application/json",
        "schema":
            Feedback.model_json_schema(),
    },
    stream=True,
)

json_text = ""

for event in stream:
    if (
        event.event_type == "step.delta"
        and event.delta.type == "text"
        and event.delta.text
    ):
        json_text += event.delta.text

print(json_text)

⚠️ Streaming JSON ยัง Parse ไม่ได้ทุก Chunk

Chunk อาจเป็น

{"sent

ถัดไป

iment":"negative",

ถัดไป

"summary":"จัดส่งช้า"}

แต่ละ Chunk เป็นเพียง Partial JSON String

ควรนำมาต่อ

Chunk 1
+
Chunk 2
+
Chunk 3
=
Final JSON

แล้วจึง Parse/Validate เมื่อข้อมูลครบ

🐍 Validate Structured Output หลัง Stream จบ

result = Feedback.model_validate_json(
    json_text
)

print(result.sentiment)
print(result.summary)

นี่ปลอดภัยกว่าพยายาม Parse JSON ทุก Chunk

🎨 Streaming Image ได้ไหม

Interactions Streaming รองรับ Output Type อื่นตาม Model และ Feature ที่ใช้งาน เช่น Image Generation ในโมเดลที่รองรับ

Event Delta สามารถมี Type เช่น

image

ดังนั้น Application ที่รองรับ Multimodal Output ต้อง Handle Data Type มากกว่า Text

ไม่ควรสมมติ

delta.text

มีอยู่ทุก Event

🤖 Streaming กับ Managed Agents

Managed Agent สามารถ Stream Steps เพื่อให้ Application เห็นความคืบหน้าของงานที่ใช้เวลานานได้

เช่น

Agent เริ่มงาน
↓
ค้นข้อมูล
↓
ประมวลผล
↓
สร้างไฟล์
↓
เสร็จ

มีประโยชน์มากกับ Research หรือ Multi-step Agent เพราะ User ไม่ต้องเห็นหน้าจอ Loading แบบไม่มีข้อมูลตลอดระยะเวลาทำงาน

🔄 Streaming กับ Background Execution ต่างกันอย่างไร

Streaming

Connection เปิดอยู่

Request
↔
Stream Events

เหมาะกับ User ที่กำลังรอ

Background Execution

สร้าง Interaction แล้วปล่อยให้ Server ทำงานต่อ

Create Job
↓
รับ Interaction ID
↓
ปิด Request
↓
เช็กผลภายหลัง

เหมาะกับงานยาวที่ไม่ควรเปิด Connection ค้าง

🆚 ควรเลือก Streaming หรือ Background

ใช้หลัก

User กำลังรอและอยากเห็นคำตอบทันที?
→ Streaming

งานอาจใช้เวลานานมากและ User ไม่ต้องรอหน้าเดิม?
→ Background

บาง Agent Workflow สามารถใช้ Background และ Streaming ตาม Feature ที่รองรับได้ แต่ Architecture ต้องออกแบบให้เหมาะกับ Product

🌐 Browser ควรเรียก Gemini Streaming โดยตรงไหม

โดยทั่วไป Web Application ที่ใช้ Secret Gemini API Key ไม่ควรให้ Browser ถือ Key

Architecture ที่เหมาะคือ

Browser
↓
Your Backend
↓
Gemini API Stream
↓
Your Backend
↓
Browser

Backend ทำหน้าที่

• Authentication
• API Key Security
• Rate Limit
• Logging
• Streaming Proxy

API Key ต้องอยู่ฝั่ง Server

🔐 อย่าใส่ Gemini API Key ใน Frontend

ไม่ควร

const key = "REAL_GEMINI_API_KEY";

แล้วเรียก Gemini จาก Browser โดยตรงสำหรับ Architecture ที่ต้องเก็บ Credential เป็น Secret

แม้ใช้ Streaming ก็ไม่ได้เปลี่ยน Security Rule

Key ยังต้องเก็บ Server-side

🔁 Backend ต้อง Forward Stream ไป Frontend

Architecture

Gemini SSE
↓
Backend
↓
Frontend Stream

Backend สามารถเลือก Protocol ระหว่างตนเองกับ Browser เช่น

• SSE
• Fetch Streaming
• WebSocket

ตาม Product Architecture

ไม่จำเป็นต้องใช้ Protocol เดียวกับ Gemini ทุกชั้นเสมอไป

📡 SSE เหมาะกับ Frontend เมื่อไร

เหมาะเมื่อ Data ไหลหลัก ๆ จาก Server ไป Client

เช่น

User ส่ง Prompt หนึ่งครั้ง
↓
Server
↓
ส่ง Text Chunks กลับเรื่อย ๆ

หากต้องสื่อสารสองทิศทางอย่างต่อเนื่องมาก เช่น Real-time Voice อาจต้องใช้ Protocol อื่นหรือ Live API ที่เหมาะกว่า

🆚 SSE กับ WebSocket

SSE

เหมาะกับ

Server
→
Client

เป็นหลัก

ข้อดี

• ใช้ HTTP
• Simple
• เหมาะกับ Text Streaming

WebSocket

เหมาะกับ

Client
↔
Server

แบบต่อเนื่อง

เหมาะกับ Real-time Interactive System มากกว่าในบางกรณี

สำหรับ Text Chat ธรรมดา SSE หรือ Fetch Streaming มักเพียงพอ

⚠️ Proxy สามารถ Buffer Stream ได้

ปัญหาที่พบได้คือ Gemini ส่ง Chunk มาแล้ว แต่ Reverse Proxy หรือ Web Server เก็บข้อมูลไว้ก่อน

ผลคือ User เห็น

ไม่มีอะไร
↓
รอ
↓
คำตอบทั้งหมดโผล่พร้อมกัน

ทั้งที่ Backend ใช้ Streaming แล้ว

ต้องตรวจ

• Reverse Proxy Buffering
• Compression
• CDN
• Server Framework
• Response Headers

เมื่อ Debug Streaming

🔍 วิธีรู้ว่า Buffering อยู่ตรงไหน

ทดสอบเป็นชั้น

Test 1

Python CLI เรียก Gemini โดยตรง

ถ้า Stream ได้ แสดงว่า Gemini/SDK ทำงาน

Test 2

Backend Endpoint

ใช้ Command-line Client ดู Chunk

Test 3

Browser

ถ้า Backend Stream แต่ Browser ไม่ Stream ปัญหาอาจอยู่ Frontend/CDN/Proxy

Debug ทีละ Layer จะเร็วกว่าเปลี่ยนทุกอย่างพร้อมกัน

📦 อย่าสะสมคำตอบทั้งหมดก่อนส่งไป Browser

Code ที่ผิดแนวคิดคือ

Gemini Stream
↓
Backend สะสมทั้งหมด
↓
รอจนจบ
↓
ส่ง Browser

แบบนี้ใช้ Streaming จาก Gemini แต่ User ไม่ได้ประโยชน์

ควร

Gemini Chunk
↓
Backend
↓
Browser ทันที

เมื่อ Architecture และ Validation อนุญาต

📝 แต่ควรเก็บ Final Text ด้วย

แม้ Forward Chunk ทันที Backend อาจสะสมอีก Copy หนึ่งสำหรับ

• Conversation History
• Logging
• Database
• Analytics

ตัวอย่าง

full_text = ""

for event in stream:
    if (
        event.event_type == "step.delta"
        and event.delta.type == "text"
    ):
        chunk = event.delta.text

        full_text += chunk

        send_to_client(chunk)

หลัง Stream จบจึง Save full_text

⚠️ อย่าบันทึก Final Answer ก่อน Interaction Completed

Stream อาจ

• ขาด
• Error
• Client Disconnect
• Model Failure

จึงควรมีสถานะ เช่น

STREAMING
COMPLETED
FAILED
CANCELLED

ใน Database

ไม่ใช่ Save ข้อความครึ่งหนึ่งแล้วทำเหมือน Response สมบูรณ์

🔌 Client Disconnect ต้องจัดการ

User อาจ

• ปิด Tab
• กด Stop
• เปลี่ยนหน้า
• Internet หลุด

Backend ต้องกำหนด Policy ว่า

Client หาย
↓
ยกเลิกงาน?
หรือ
ให้ Gemini ทำต่อ?

ขึ้นกับ Product

ถ้ายังคง Generation ต่อทั้งที่ไม่มีผู้ใช้อ่าน อาจสร้าง Token Cost โดยไม่จำเป็น

🛑 ปุ่ม Stop Generation

Chat UI ควรมี

Stop

สำหรับคำตอบยาว

เมื่อ User กด Stop

Frontend ส่ง Cancel Signal ไป Backend

Backend ควรหยุดอ่าน/ประมวลผล Stream ตาม Capability และ Architecture ที่ใช้

ช่วยลด

• Output ที่ไม่ต้องการ
• User Frustration
• Cost

🔁 Regenerate ต้องเป็น Request ใหม่

หลัง User กด

Regenerate

ควรบันทึกว่า Response เดิม

cancelled
หรือ
superseded

ตาม Product

แล้วเริ่ม Request ใหม่

ไม่ควรเอา Chunk จากสอง Streams มาต่อกัน

🚦 Rate Limit ยังมีผลกับ Streaming

Streaming ไม่ได้หลบ

RPM
TPM
RPD

หรือ Limit อื่น

Request Streaming หนึ่งครั้งยังเป็น Gemini API Usage

หากระบบเปิดหลาย Streams พร้อมกันจำนวนมากต้องควบคุม

• Concurrent Users
• Rate Limit
• Token Usage
• Cost

💸 User เปิด Stream แล้วทิ้งไว้มี Cost หรือไม่

หาก Model ยัง Generation อยู่ก็สามารถมี Token Usage ต่อ

ดังนั้น Public Application ควรมี

• Maximum Output
• Timeout
• Stop
• User Quota

ไม่ควรเปิด Generation แบบไร้ขอบเขต

⏲️ ตั้ง Timeout

ตัวอย่าง

Maximum Request Time
=
120 seconds

เป็นเพียงตัวอย่าง

ค่าจริงขึ้นกับ Use Case

เมื่อเกิน

• Cancel
• Show Error
• Allow Retry

ดีกว่าเปิด Connection ค้างตลอดไปหากระบบผิดปกติ

🔁 Retry Streaming ต้องระวัง Duplicate Text

สมมติ User ได้

ข้อความ A
ข้อความ B
ข้อความ C

แล้ว Connection หลุด

ถ้า Backend Retry ทั้ง Request ใหม่ อาจได้

ข้อความ A
ข้อความ B
ข้อความ C
ข้อความ D

ถ้านำไปต่อทันทีจะกลายเป็น

A B C A B C D

จึงต้องออกแบบ Retry ให้ดี

🧠 วิธี Handle Stream Failure

ทางเลือกที่ปลอดภัย เช่น

Option 1

แสดงว่า Response ถูกตัด

การเชื่อมต่อขาด กรุณาลองใหม่

Option 2

เริ่ม Generation ใหม่เป็น Response ใหม่

Option 3

Resume ผ่าน Stateful Workflow เฉพาะเมื่อ API/Use Case รองรับอย่างถูกต้อง

ไม่ควรสร้างระบบต่อ String จาก Retry แบบเดาเอง

📊 เก็บ Metrics อะไรสำหรับ Streaming

อย่างน้อย

request_count
ttft
total_latency
output_tokens
stream_errors
client_disconnects

และ

completion_rate

เช่น

Started Streams
=
10,000

Completed
=
9,700

Completion Rate
=
97%

ช่วยค้นหาปัญหาที่ไม่เห็นจาก HTTP 200 อย่างเดียว

📈 Latency ที่ควรวัด

Time to First Token

ผู้ใช้เริ่มเห็นข้อความเมื่อไร

Time to Last Token

คำตอบจบเมื่อไร

Generation Duration

ใช้เวลาสร้างเท่าไร

Tool Latency

หากใช้ Search/Function

Network Flush Delay

Backend ส่ง Chunk ไป Browserช้าหรือไม่

Metrics เหล่านี้ช่วยแยก Model Latency กับ Infrastructure Latency

📊 Streaming ไม่ควร Render ทุกตัวอักษรทันทีเสมอไป

หาก Backend ส่ง Chunk เล็กมากและ Frontend Render DOM ทุก Chunk

อาจทำให้ Browser ทำงานหนัก

Frontend สามารถ Buffer ระยะสั้น เช่น

รับหลาย Chunks
↓
รวมเล็กน้อย
↓
Update UI

เช่นทุก

20–50 ms

เป็นแนวคิดหนึ่งที่ช่วยให้ UI ลื่น

ค่าจริงต้อง Benchmark

🖥️ Frontend ควรมี Cursor

User Experience ที่ดีอาจแสดง

Gemini กำลังตอบ... ▌

แล้วเลื่อน Cursor ตามข้อความ

เมื่อ

interaction.completed

จึงเอา Cursor ออก

ช่วยให้ User รู้ว่าคำตอบยังไม่จบ

🧾 Markdown Streaming ต้องระวัง

Gemini อาจส่ง

**

ก่อน

แล้ว Chunk ถัดไปจึงเป็น

ข้อความตัวหนา**

หาก Render Markdown ทุก Chunk อาจเกิด Layout กระตุก

แนวทางมี เช่น

• Render Plain Text ระหว่าง Stream
• Incremental Markdown Parser
• Re-render อย่างมี Throttle

ตาม Framework

💻 Code Block Streaming ก็มีปัญหาเดียวกัน

อาจได้รับ

```py

แล้ว Code มาใน Chunks ต่อไป

Frontend ต้องรองรับ Code Block ที่ยังไม่ปิด

ไม่ควรสมมติทุก Chunk เป็น Valid Markdown Document สมบูรณ์

🔐 อย่า Render HTML จาก AI ตรง ๆ

ถ้า Application แปลง Markdown/HTML ให้ User

ควร Sanitization ตาม Security Requirement

Streaming ไม่ได้เปลี่ยนหลัก

AI Output
=
Untrusted Content

โดยเฉพาะ Web UI ที่เปิดให้ Render HTML

🛡️ XSS ยังเป็นเรื่องสำคัญ

ไม่ควร

element.innerHTML = aiChunk;

กับ Content ที่ไม่ได้ Sanitized

ควรใช้ Rendering Library ที่จัดการ Security เหมาะสม

AI Generated Text ไม่ควรได้รับ Trust สูงกว่าข้อมูลจาก User

📄 Streaming กับ Long Article

เหมาะกับบทความยาวเพราะ User เห็น Progress

แต่ถ้า Application จะบันทึกบทความลง CMS

ควรรอ

interaction.completed

ก่อน Validate Final Content

จากนั้นจึง

Final Text
↓
Validation
↓
CMS Draft

ไม่ควร Publish Chunk แรกทันที

🧪 Development ควร Print Event ทั้งหมดก่อน

ตอนเริ่มสร้าง Streaming Integration สามารถทำ

for event in stream:
    print(event)

เพื่อดู Event Structure จริง

เมื่อเข้าใจแล้วจึงเขียน Handler เฉพาะ

step.delta
interaction.completed

หรือ Tool Events ที่ต้องใช้

ช่วยลดการเดา Schema

🚫 อย่าใช้ content.delta จาก Tutorial เก่า

Interactions API รุ่นก่อนเคยใช้ Event Type

content.delta

แต่ Migration Guide ปัจจุบันระบุว่า Streaming ใช้

step.delta

แทน

ดังนั้น Code แบบ

if event.event_type == "content.delta":

อาจเป็น Tutorial รุ่นเก่า

Code ปัจจุบันควรใช้

if event.event_type == "step.delta":

ตาม API Schema ล่าสุด

🧠 ทำไมเปลี่ยนเป็น Step-based Streaming

เพราะ Interactions API ไม่ได้สร้างเพียงข้อความ

หนึ่ง Interaction สามารถมี

Thought
Search
Function
Model Output
Image
Agent Step

จึงใช้โครงสร้าง

Step

เพื่อ Track Timeline ของ Interaction ได้ละเอียดขึ้น

นี่ช่วยให้ Streaming รองรับ Agentic Workflow ได้ดีกว่า Text Chunks อย่างเดียว

🧩 step.start + step.delta + step.stop

รูปแบบนี้คิดได้เป็น

Step Start
↓
Delta
↓
Delta
↓
Delta
↓
Step Stop

จากนั้นอาจเริ่ม Step ใหม่

Step Start
↓
...

จน Interaction Completed

Application ที่สร้าง Agent UI สามารถใช้ Pattern นี้แสดง Progress ได้อย่างเป็นระบบ

🔄 Streaming Stateful Conversation

หลัง Interaction แรกจบ สามารถถามต่อโดยใช้

previous_interaction_id

และเปิด Stream อีกครั้ง

Python

first = client.interactions.create(
    model="gemini-3.7-flash",
    input="อธิบาย Context Caching",
)

stream = client.interactions.create(
    model="gemini-3.7-flash",
    previous_interaction_id=first.id,
    input="ยกตัวอย่างเพิ่มเติม",
    stream=True,
)

for event in stream:
    if (
        event.event_type == "step.delta"
        and event.delta.type == "text"
    ):
        print(
            event.delta.text,
            end="",
            flush=True,
        )

เหมาะกับ Chat ที่ User ถามต่อเนื่อง

🧠 Stateful ช่วยไม่ต้องส่ง History ทั้งหมดเอง

ใช้

previous_interaction_id

ทำให้ Server เชื่อม Context กับ Interaction ก่อนหน้า

Application จึงไม่จำเป็นต้องส่ง Transcript ทั้งหมดทุก Turn ใน Workflow แบบ Stateful

แต่ต้องออกแบบ Retention/Privacy ตาม Requirement ของระบบ

📡 Streaming Stateless ก็ทำได้

สามารถใช้

store=false

ตาม Workflow ที่รองรับ แล้ว Application จัดการ History เอง

เหมาะกับระบบที่ต้องควบคุม State เอง

แต่จะซับซ้อนกว่า เพราะต้องรักษา Steps ที่จำเป็นของ Interaction History ให้ครบ

⚠️ Stateless Agent Streaming ต้องเก็บ Steps ให้ถูก

โดยเฉพาะ

• Thought-related information ที่ API ต้องการรักษา
• Function Calls
• Function Results
• Tool Steps

หากส่ง History ไม่ครบ อาจทำให้ Workflow ต่อเนื่องผิด

สำหรับมือใหม่ Stateful Interactions มักง่ายกว่า

💰 ควรจำกัด Output Length ไหม

ควรตาม Use Case

Streaming ทำให้คำตอบยาวดูน่าอ่านขึ้น แต่ไม่ได้หมายความว่าควรสร้าง Output ไม่จำกัด

ตัวอย่าง Chatbot อาจกำหนด Prompt

ตอบไม่เกิน 5 ย่อหน้า

ถ้า Requirement ไม่ต้องการ Report ยาว

ช่วยลด

• Token
• Latency
• Cost

🚦 Streaming หลาย User พร้อมกัน

สมมติมี

10,000 Users

ออนไลน์พร้อมกัน

Server อาจต้องรักษา Connections จำนวนมาก

ดังนั้นนอกจาก Gemini Rate Limit ต้องวาง Infrastructure สำหรับ

• Open Connections
• Memory
• Load Balancer
• Proxy Timeout
• Connection Limits

ด้วย

Streaming Architecture Scale ต่างจาก REST Response สั้น ๆ

🏗️ Serverless ใช้ Streaming ได้ไหม

ขึ้นกับ Platform และ Runtime

ต้องตรวจว่า Hosting รองรับ

• Streaming HTTP Response
• Long-lived Connection
• Timeout ที่เพียงพอ
• No forced buffering

ถ้า Platform รอจน Function จบก่อนส่ง Response Streaming จะไม่ทำงานตามต้องการ

🌐 CDN ต้องตรวจ Streaming Support

CDN หรือ Proxy บาง Configuration อาจ

• Buffer
• Compress
• Delay
• Terminate long connection

เมื่อ Production แล้ว Stream ช้ากว่า Local ให้ตรวจ Network Layer ก่อนโทษ Gemini API

⏲️ Keep-alive และ Idle Timeout

Agent ที่ไม่มี Text Output หลายนาทีระหว่าง Tool Work อาจถูก Proxy มองว่า Connection Idle

ระบบ Long-running Streaming ต้องออกแบบ Timeout ให้สอดคล้องกับ

• Gemini Behavior
• Backend
• Load Balancer
• Browser

หากงานยาวมาก Background Execution อาจเหมาะกว่า

🔄 Streaming กับ Batch API ต่างกันอย่างไร

Streaming

ตอบทันที
ทีละส่วน
User กำลังรอ

Batch

งานจำนวนมาก
รอได้
ต้นทุนต่ำกว่า

ไม่ใช่ Feature ที่ใช้แทนกัน

ตัวอย่าง

Chat
→ Streaming

100,000 Product Classification
→ Batch

🆚 Streaming กับ Live API

Streaming Text API คือ

Request
↓
Model Generates
↓
Text Chunks

Live API ถูกออกแบบสำหรับ Real-time Interaction ที่ต่อเนื่องมากกว่า เช่น

Audio
↔
Gemini

พร้อม Low-latency Session

หากกำลังสร้าง Voice Assistant แบบพูดโต้ตอบทันที หัวข้อถัดไปเกี่ยวกับ Live API จะเหมาะกว่า Text Streaming ธรรมดา

🧪 Checklist Debug Streaming

ถ้า Streaming ไม่ทำงาน ให้ตรวจ

stream=True

ตั้งแล้วหรือไม่

Event Type

ใช้ step.delta หรือยัง

Delta Type

เป็น text หรือไม่

CLI

Stream ได้หรือไม่

Backend

Forward Chunk ทันทีหรือไม่

Proxy

Buffer หรือไม่

CDN

รองรับ Streaming หรือไม่

Frontend

อ่าน Stream Incrementally หรือไม่

Markdown Renderer

Buffer Text อยู่หรือไม่

Timeout

Connection ถูกตัดหรือไม่

การตรวจตาม Layer จะหา Root Cause ได้เร็วกว่า

🚫 15 ข้อผิดพลาดเมื่อใช้ Streaming Response

❶ ลืม stream=True

ก็จะได้ Response ปกติ

❷ ใช้ content.delta

เป็น Event รุ่นเก่า

❸ คิดว่าทุก Delta เป็น Text

อาจมี Thought, Image หรือ Tool Data

❹ สะสม Stream จนจบก่อนส่ง Frontend

เสียประโยชน์ Streaming

❺ Parse JSON ทุก Chunk

Structured JSON อาจยังไม่ครบ

❻ Execute Function จาก Partial Arguments

อันตราย

❼ ไม่มี Client Disconnect Handling

เสีย Resource/Cost

❽ ไม่มี Timeout

Connection ค้าง

❾ Retry แล้วต่อ Text เดิมทันที

เกิดข้อความซ้ำ

❿ Render HTML จาก Chunk ตรง ๆ

เสี่ยง Security

⓫ Save Response เป็น Completed ก่อนจบ

ข้อมูลครึ่งเดียวถูกมองว่าสมบูรณ์

⓬ ไม่เก็บ Usage ตอน Completed

วัด Cost ไม่ได้

⓭ ไม่วัด TTFT

ไม่รู้ Streaming UX จริง

⓮ Proxy Buffering

Local Stream แต่ Production ไม่ Stream

⓯ ใช้ Streaming กับงาน Batch จำนวนมาก

เลือก API ผิด Use Case

🪜 วิธีใช้ Streaming Response แบบ Step by Step

❶ ติดตั้ง SDK ล่าสุด

Python

google-genai

JavaScript

@google/genai

❷ สร้าง Client

เก็บ API Key Server-side

❸ เลือก Model

เช่น

gemini-3.7-flash

❹ เปิด Streaming

stream=True

❺ Iterate Events

อ่านทีละ Event

❻ ตรวจ step.delta

หา Incremental Update

❼ ตรวจ delta.type

Text, Tool, Thought หรือข้อมูลอื่น

❽ Forward Text ไป Frontend

ทันทีตาม Architecture

❾ สะสม Final Text

สำหรับ Save ภายหลัง

❿ Handle Tool Calls

หากเปิด Function/Search

⓫ Handle Disconnect

User ปิดหน้า

⓬ Handle Error

อย่าปล่อย Connection ค้าง

⓭ รอ interaction.completed

ก่อน Mark Completed

⓮ เก็บ Usage

Token/Cost Metrics

⓯ วัด TTFT

ปรับ UX

⓰ ตรวจ Proxy/CDN

ก่อนเปิด Production

สำหรับระบบ AI ของ comsiam การแยก Stream Handler ออกจาก Business Logic จะช่วยให้ภายหลังเพิ่ม Search, Function Calling หรือ Agent Steps ได้โดยไม่ต้องเขียน Streaming Code ใหม่ทั้งหมด

✅ Checklist ก่อนเปิด Streaming ใน Production

SDK เป็นรุ่นปัจจุบัน

ใช้ step.delta

API Key อยู่ Backend

Frontend อ่าน Stream ได้

Proxy ไม่ Buffer

Timeout ถูกตั้ง

Disconnect Handler มี

Stop Generation มีเมื่อเหมาะ

Full Response ถูกสะสม

Completion State ถูกเก็บ

Error State ถูกเก็บ

Usage ถูก Monitor

TTFT ถูก Monitor

Retry ไม่สร้างข้อความซ้ำ

Markdown ถูก Handle

HTML ถูก Sanitize

Function Arguments รอครบ

Structured JSON รอครบ

Rate Limit มี

User Quota มี

📊 Metrics ที่ควรเก็บ

streams_started
streams_completed
streams_failed
streams_cancelled

ด้านเวลา

time_to_first_token
time_to_last_token
total_latency

ด้าน Usage

input_tokens
output_tokens
total_tokens

ด้าน Product

stop_generation_rate
regenerate_rate
client_disconnect_rate

ช่วยให้รู้ว่า Streaming ดีต่อ User จริงหรือไม่

💡 ตัวอย่าง Streaming Service แบบง่าย

Python

from google import genai

client = genai.Client()


def stream_answer(question: str):
    stream = client.interactions.create(
        model="gemini-3.7-flash",
        input=question,
        stream=True,
    )

    full_text = ""

    for event in stream:
        if (
            event.event_type
            == "step.delta"
            and event.delta.type
            == "text"
            and event.delta.text
        ):
            full_text += event.delta.text

            yield event.delta.text

        elif (
            event.event_type
            == "interaction.completed"
        ):
            # บันทึก Usage หรือสถานะ
            # ตามระบบของ Application
            pass

Web Framework สามารถนำ Generator นี้ไป Forward ให้ Client ตาม Streaming Capability ของ Framework นั้น

💡 ตัวอย่างสถานะ Frontend

เริ่มต้น

status = generating

รับ Chunk

append text

Interaction Completed

status = completed

เกิด Error

status = failed

User กด Stop

status = cancelled

State ที่ชัดช่วยป้องกัน UI สับสน

📌 Streaming ไม่ใช่แค่ Effect ตัวหนังสือวิ่ง

การทำ Streaming ที่ดีต้องครอบคลุม

Gemini Event Stream
+
Backend
+
Proxy
+
Frontend
+
Error Handling
+
Cancellation
+
Metrics

หากทำเพียง

เปิด stream=True

แต่ Backend Buffer Response ไว้ทั้งหมด User ก็ยังไม่ได้ประโยชน์

❓ คำถามที่พบบ่อย

Gemini API Streaming Response คืออะไร

คือการให้ Gemini ส่งผลลัพธ์กลับทีละส่วนระหว่างที่โมเดลกำลังสร้างคำตอบ ทำให้ User เริ่มเห็นข้อความก่อน Generation ทั้งหมดเสร็จ

Interactions API เปิด Streaming อย่างไร

Python ใช้ stream=True และ JavaScript ใช้ stream: true ใน interactions.create() จากนั้นอ่าน Event ที่ส่งกลับแบบต่อเนื่อง

Text Chunk อยู่ใน Event ไหน

รูปแบบปัจจุบันใช้ step.delta และสำหรับข้อความทั่วไป event.delta.type จะเป็น text โดยอ่านข้อความใหม่จาก event.delta.text

content.delta ยังใช้ได้ไหม

เป็นรูปแบบ Legacy ของ Interactions API ก่อน Breaking Changes เดือนพฤษภาคม 2026 รูปแบบปัจจุบันเปลี่ยนเป็น step.delta

REST Streaming ใช้อะไร

Interactions API ใช้ SSE โดยเรียก /v1beta/interactions?alt=sse พร้อม "stream": true ใน Request Body

Streaming ลดค่า Gemini API หรือไม่

ไม่โดยอัตโนมัติ Streaming ช่วยด้าน User Experience และ Time to First Output แต่ Token Usage ยังคงขึ้นกับ Input, Output, Model และ Tools ที่ใช้งาน

🎯 สรุป

Streaming Response ช่วยให้ Gemini API ส่งคำตอบกลับ ทีละส่วนระหว่าง Generation ทำให้ Chatbot และ AI Assistant รู้สึกตอบสนองเร็วขึ้น เพราะ User ไม่ต้องรอ Final Answer ทั้งหมดก่อนเริ่มอ่านข้อความ

สำหรับ Interactions API รุ่นปัจจุบัน Python เปิด Streaming ด้วย stream=True ส่วน JavaScript ใช้ stream: true และ REST ใช้ SSE ผ่าน Interactions Endpoint พร้อม alt=sse

Event Structure ปัจจุบันเป็นแบบ Step-based โดยมี Event สำคัญ เช่น interaction.created, step.start, step.delta, step.stop และ interaction.completed สำหรับ Text Streaming ควรอ่าน step.delta ที่มี delta.type == "text" และนำ delta.text ไปแสดงทีละส่วน

Tutorial เก่าที่ใช้ content.delta เป็นรูปแบบก่อน Breaking Changes เดือนพฤษภาคม 2026 จึงควรเปลี่ยนเป็น step.delta สำหรับ Code ใหม่

Streaming ยังสามารถทำงานร่วมกับ Thinking Summaries, Function Calling, Google Search, Structured Output, Image Output และ Agent Workflow ได้ แต่ Developer ต้อง Handle Delta Type ให้ถูก และห้าม Execute Function หรือ Parse Structured JSON ก่อนข้อมูลครบ

ใน Production ต้องคิดมากกว่าเพียงเปิด stream=True เพราะ Backend, Reverse Proxy, CDN และ Frontend ต้อง Forward Data แบบ Incremental จริง รวมถึงต้องรองรับ Client Disconnect, Timeout, Stop Generation, Retry, Error State และ Security

แนวทางของ comsiam คือวัดทั้ง Time to First Token, Total Latency, Completion Rate และ Token Usage หลัง interaction.completed พร้อมแยก Event Handler ตามประเภทตั้งแต่ต้น วิธีนี้ทำให้ Streaming Architecture พร้อมรองรับ Gemini Tool และ Agent ที่ซับซ้อนขึ้นในอนาคตโดยไม่ต้องรื้อระบบทั้งหมด