วิธีสร้าง Voice Agent ตอบสนองไว ไร้ช่องว่างด้วย ADK

19 ต.ค. 2026

คู่มือสำหรับ streaming tools*, non-blocking function calls, และ response scheduling ใน* Agent Development Kit (ADK) Gemini Live API Toolkit*.*

ในการแชทด้วยข้อความ การรอคิวรีฐานข้อมูลหรือ RAG pipeline นาน 13 วินาที อาจแสดงแค่สัญลักษณ์โหลดข้อมูล แต่สำหรับการสนทนาด้วยเสียง หากเครื่องมือหยุดการทำงาน (blocking tool) นานขนาดนั้น จะเกิดความเงียบสนิทที่ทำให้ผู้ใช้เข้าใจผิดว่าสายหลุดและวางสายไปได้

ในเมื่อเราไม่สามารถทำให้การประมวลผลหลังบ้านเร็วขึ้นได้เสมอไป ทางแก้จึงอยู่ที่สถาปัตยกรรม โดยการทำให้เอเจนท์ยังคงสนทนาต่อได้ในขณะที่เครื่องมือทำงานอยู่เบื้องหลัง ADK แบบโอเพนซอร์สและ gemini-3.8-live ช่วยให้คุณใช้เครื่องมือเสียงแบบ non-blocking ได้ในตัวด้วย Python จากการทดสอบพบว่าการเปลี่ยนจาก blocking tool มาเป็น streaming tool ช่วยลดเวลาการรอจนถึงคำพูดแรกของเอเจนท์จาก 14.4 วินาที เหลือเพียง 1.2 วินาทีเท่านั้น

คู่มือนี้ครอบคลุมวิธีใช้ streaming tools, การกำหนด FunctionTool แบบ non-blocking และ response_scheduling เพื่อสร้าง voice agent ที่ตอบสนองได้ดีและเป็นธรรมชาติ

ชมวิดีโอสาธิต ADK Live AI Concierge ความยาว 1 นาที เพื่อฟังการทำงานของ scheduled responses:

วิดีโอสาธิต ADK Live AI Concierge อธิบายการทำงานของ scheduled responses

การใช้ non-blocking tool calls ใน ADK มอบประโยชน์ด้านประสบการณ์ผู้ใช้ทันทีหลายประการ:

• Zero Dead Air: เอเจนท์สามารถตอบรับคำขอได้ทันที เช่น "กำลังตรวจสอบตัวเลขให้เดี๋ยวนี้ครับ..." แทนการเงียบหายไป
• Audible Progress Updates: สำหรับงานหลายขั้นตอน เอเจนท์สามารถเล่าลำดับเหตุการณ์ระหว่างทางเพื่อให้ผู้ใช้รับรู้ถึงความคืบหน้า
• Multitasking and Parallel Dialogue: ผู้ใช้สามารถถามคำถามอื่นแทรกได้ในขณะที่ระบบกำลังประมวลผลงานชิ้นเดิมในเบื้องหลัง
• Granular Delivery Control: กำหนดได้ว่าผลลัพธ์สุดท้ายจะแจ้งตอนไหน เช่น รอจังหวะหยุดพูดที่เป็นธรรมชาติ หรือแทรกแทรงทันทีเมื่อเป็นข้อมูลสำคัญ

การเปลี่ยนจาก Blocking เป็น Streaming

ในการพัฒนาเอเจนท์มาตรฐาน เครื่องมือมักถูกกำหนดเป็นฟังก์ชัน async def ที่ทำงานจนเสร็จแล้วส่งกลับค่าเดียว ซึ่งจะบล็อกการทำงานของโมเดลจนกว่าจะได้คำตอบ

from typing import Any
async def query_sales_database(region: str) -> dict[str, Any]:
"""Runs the quarterly sales report query for a region.
Args:
region: Sales region to report on, e.g. "EMEA" or "APAC".
Returns:
A dict containing report metrics upon completion.
"""
report = await run_expensive_query(region) # e.g., 13 seconds of work
return {"status": "success", "report": report} # Single return at the end
``` เพื่อให้การสนทนาลื่นไหล คุณสามารถใช้ **streaming tool** โดยเปลี่ยนจาก `return` เป็น `yield` เท่านั้น ซึ่ง ADK จะทำเครื่องหมายเป็น `NON_BLOCKING` โดยอัตโนมัติ และส่งการตอบรับไปยังโมเดลทันที
 
```python
import asyncio
from typing import AsyncGenerator
from google.adk.agents import Agent
async def query_sales_database(region: str) -> AsyncGenerator[str, None]:
"""Runs the quarterly sales report. Call this ONCE; it pushes updates automatically.
Args:
region: Sales region to report on, e.g. "EMEA" or "APAC".
Yields:
Progress updates while the query runs, followed by the final summary.
"""
stages = [
("Connecting to the sales warehouse...", 1.5),
("Scanning order records...", 3.5),
("Aggregating revenue by product line...", 3.5),
("Formatting the report...", 1.5),
]
for message, delay in stages:
yield f"PROGRESS: {message}"
await asyncio.sleep(delay)
yield f"DONE: {region} Q2 revenue reached $4.81M, up 12.4% year over year."
agent = Agent(
name="sales_analyst",
model="gemini-3.8-live",
tools=[query_sales_database],
instruction=(
"You are a helpful sales assistant. When the user asks about sales data, "
"call query_sales_database once. Acknowledge the request immediately, "
"briefly paraphrase each progress update, and present the final figures "
"when completed. Keep answers conversational and concise."
),
)

เทคนิคการรันงานเบื้องหลังอย่างเงียบๆ

หากการรอไม่มีเรื่องราวที่ต้องเล่า การรันงานเงียบๆ ในเบื้องหลังอาจเหมาะสมกว่า คุณสามารถใช้ async generator ที่ yield เพียงครั้งเดียวเมื่อเสร็จสิ้น หรือใช้ FunctionTool ร่วมกับ response_scheduling เพื่อกำหนดจังหวะการตอบกลับ

from google.adk.tools import FunctionTool
from google.genai import types
sales_tool = FunctionTool(query_sales_database)
sales_tool.response_scheduling = types.FunctionResponseScheduling.WHEN_IDLE

แอตทริบิวต์ response_scheduling มี 3 รูปแบบหลัก:

• WHEN_IDLE: รอจนกว่าโมเดลหรือผู้ใช้จะพูดจบ แล้วจึงตอบในจังหวะหยุดพักธรรมชาติ
• INTERRUPT: ตัดจังหวะการพูดปัจจุบันทันทีเพื่อแจ้งผลลัพธ์
• SILENT: เพิ่มข้อมูลลงในประวัติการสนทนาโดยไม่ประกาศออกมา

นอกจากนี้ ใน ADK 2.8 ขึ้นไป คุณยังสามารถกำหนดการตั้งเวลาแบบไดนามิกต่อการอัปเดตแต่ละครั้งได้ด้วยการ yield types.FunctionResponse เพื่อสลับโหมดระหว่างทำงานได้ตามต้องการ

สรุปคือ หากการรอมีประเด็นให้เล่าควรใช้ stream แต่หากเป็นเพียงการรอผลลัพธ์ให้เลือกใช้การรันเบื้องหลัง (idle) แม้ความเร็วหลังบ้านจะเท่าเดิม แต่ความรู้สึกของผู้ใช้จะเปลี่ยนไป เอเจนท์จะดูใส่ใจและเป็นธรรมชาติมากขึ้นตลอดการทำงานที่ซับซ้อน

ลองด้วยตัวคุณเอง: การสาธิตแบบโต้ตอบ ทดสอบความแตกต่างของแต่ละรูปแบบบน gemini-3.8-live

Framework: google-adk >= 2.10.0
Model: gemini-3.8-live บน Gemini Enterprise
API Environment: ต้องใช้เซสชันแบบสดสองทิศทาง (run_live)