วิธีใช้ OpenAI WebSocket Mode API

TL;DR

OpenAI มีโหมด WebSocket API สองโหมดสำหรับกรณีการใช้งานที่แตกต่างกัน: โหมด Responses API WebSocket สำหรับเวิร์กโฟลว์แบบ agentic ที่มีการเรียกใช้เครื่องมือจำนวนมาก (เร็วขึ้นสูงสุด 40% สำหรับการเรียกใช้เครื่องมือ 20+ ครั้ง) และ Realtime API สำหรับแอปพลิเคชันเสียงและวิดีโอที่มีความหน่วงต่ำ ทั้งสองโหมดใช้การเชื่อมต่อ WebSocket แบบถาวรแทนการร้องขอ HTTP แบบไร้สถานะ ซึ่งช่วยลดความหน่วงโดยการกำจัดค่าใช้จ่ายในการเชื่อมต่อซ้ำๆ และเปิดใช้งานการโต้ตอบแบบมีสถานะที่ขับเคลื่อนด้วยเหตุการณ์

บทนำ

API ของ OpenAI ได้พัฒนาไปไกลกว่ารูปแบบการร้องขอ-ตอบกลับแบบง่ายๆ สำหรับแอปพลิเคชันที่ต้องการการเรียกใช้เครื่องมืออย่างรวดเร็วหรือการสตรีมเสียงแบบเรียลไทม์ โมเดล HTTP แบบดั้งเดิม จะสร้างค่าใช้จ่ายที่ไม่จำเป็น การร้องขอใหม่ทุกครั้งต้องมีการตั้งค่าการเชื่อมต่อ การยืนยันตัวตน และการส่งสถานะ แม้ว่าคุณจะกำลังสนทนาต่อเนื่องในบทสนทนาเดียวกันก็ตาม

WebSocket API ของ OpenAI แก้ปัญหานี้โดยการรักษาการเชื่อมต่อแบบสองทิศทางที่ถาวร สำหรับเวิร์กโฟลว์แบบ agentic ที่มีการเรียกใช้เครื่องมือต่อเนื่อง 20+ ครั้ง สิ่งนี้หมายถึงการดำเนินการแบบ end-to-end ที่เร็วขึ้นประมาณ 40% สำหรับแอปพลิเคชันเสียง ช่วยให้สามารถสนทนาได้อย่างเป็นธรรมชาติ สามารถขัดจังหวะได้ และมีความหน่วงต่ำกว่า 500ms

คู่มือนี้ครอบคลุมโหมด WebSocket ทั้งสองของ OpenAI: Responses API สำหรับเวิร์กโฟลว์ตัวแทนที่เน้นเครื่องมือ และ Realtime API สำหรับการสตรีมเสียง คุณจะได้เรียนรู้ว่าจะใช้แต่ละโหมดเมื่อใด วิธีนำไปใช้งาน และวิธีทดสอบอย่างมีประสิทธิภาพ

OpenAI WebSocket API คืออะไร?

OpenAI WebSocket API มีกลไกการขนส่งทางเลือกแทน HTTP สำหรับการโต้ตอบกับโมเดลภาษาของ OpenAI แทนที่จะสร้างการเชื่อมต่อใหม่สำหรับการเรียก API แต่ละครั้ง WebSocket จะสร้างการเชื่อมต่อเดียวที่ใช้งานได้ยาวนาน ซึ่งยังคงเปิดอยู่ตลอดระยะเวลาเซสชันของคุณ

คุณสมบัติหลัก

การเชื่อมต่อแบบถาวร: เมื่อสร้างแล้ว การเชื่อมต่อ WebSocket จะยังคงเปิดอยู่จนกว่าจะปิดอย่างชัดเจน ซึ่งช่วยลดค่าใช้จ่ายในการเชื่อมต่อต่อคำขอ

การสื่อสารแบบสองทิศทาง: ทั้งไคลเอนต์และเซิร์ฟเวอร์สามารถส่งข้อความได้ตลอดเวลา ทำให้สถาปัตยกรรมที่ขับเคลื่อนด้วยเหตุการณ์เป็นจริงได้

เซสชันที่มีสถานะ: เซิร์ฟเวอร์จะรักษาสถานะบริบทสำหรับการเชื่อมต่อปัจจุบัน ทำให้คุณสามารถอ้างอิงการตอบกลับก่อนหน้าได้โดยไม่ต้องส่งประวัติการสนทนาทั้งหมดซ้ำ

โมเดลที่ขับเคลื่อนด้วยเหตุการณ์: การสื่อสารเกิดขึ้นผ่านเหตุการณ์ที่แยกจากกัน (ข้อความ JSON) แทนที่จะเป็นคู่การร้องขอ-ตอบกลับ

พื้นฐานโปรโตคอล WebSocket

การเชื่อมต่อ WebSocket เริ่มต้นด้วย คำขออัปเกรด HTTP จากนั้นจึงเปลี่ยนไปใช้โปรโตคอล WebSocket สำหรับ OpenAI คุณจะเชื่อมต่อกับปลายทางต่างๆ เช่น:

• Responses API: wss://api.openai.com/v1/responses
• Realtime API: wss://api.openai.com/v1/realtime?model=gpt-realtime

รูปแบบ wss:// แสดงถึงการเชื่อมต่อ WebSocket ที่ปลอดภัย (คล้ายกับ HTTPS สำหรับ HTTP)

คำอธิบายสองโหมด WebSocket

OpenAI มี โหมด WebSocket สองแบบที่แตกต่างกัน ซึ่งแต่ละโหมดได้รับการปรับให้เหมาะสมกับกรณีการใช้งานที่แตกต่างกัน

โหมด Responses API WebSocket

Responses API รองรับการเชื่อมต่อ WebSocket สำหรับ เวิร์กโฟลว์แบบ agentic ที่คุณต้องทำการเรียกใช้เครื่องมือตามลำดับจำนวนมาก โหมดนี้ออกแบบมาสำหรับผู้ช่วยเขียนโค้ด ระบบจัดการ และตัวแทนอิสระที่เรียกใช้เครื่องมือซ้ำๆ เพื่อทำงานที่ซับซ้อนให้สำเร็จ

วิธีการทำงาน:

ในการเชื่อมต่อ WebSocket ที่ทำงานอยู่ บริการจะรักษาสถานะการตอบกลับก่อนหน้าหนึ่งรายการไว้ในแคชในหน่วยความจำเฉพาะการเชื่อมต่อ (การตอบกลับล่าสุด) เมื่อคุณดำเนินการต่อ คุณจะส่งเพียง:

• previous_response_id (การอ้างอิงถึงการตอบกลับล่าสุด)
• รายการอินพุตใหม่ (ข้อความผู้ใช้ ผลลัพธ์เครื่องมือ ฯลฯ)

เซิร์ฟเวอร์จะนำสถานะที่แคชไว้กลับมาใช้ใหม่ แทนที่จะประมวลผลประวัติการสนทนาทั้งหมดซ้ำ

ประโยชน์ด้านประสิทธิภาพ:

สำหรับเวิร์กโฟลว์ที่มีการเรียกใช้เครื่องมือ 20+ ครั้ง OpenAI รายงานว่าการดำเนินการแบบ end-to-end เร็วขึ้นสูงสุด 40% เมื่อเทียบกับ HTTP ซึ่งมาจากการ:

• ไม่มีการตั้งค่าการเชื่อมต่อต่อคำขอ
• ไม่มีค่าใช้จ่ายในการยืนยันตัวตนซ้ำๆ
• สถานะที่แคชไว้ช่วยลดเวลาในการประมวลผล
• ความหน่วงของเครือข่ายที่ต่ำลงสำหรับข้อความต่อเนื่องขนาดเล็ก

ความเข้ากันได้:

โหมด WebSocket ใช้งานได้ทั้งกับ Zero Data Retention (ZDR) และตัวเลือก store=false ทำให้เหมาะสำหรับแอปพลิเคชันที่คำนึงถึงความเป็นส่วนตัว

โหมด Realtime API WebSocket

Realtime API ให้ความสามารถในการสตรีมเสียงที่มีความหน่วงต่ำสำหรับ แอปพลิเคชันที่ขับเคลื่อนด้วยเสียง ช่วยให้สามารถโต้ตอบแบบพูด-ต่อ-พูดที่โมเดลสามารถตอบกลับอินพุตเสียงด้วยเอาต์พุตเสียง จัดการการขัดจังหวะได้อย่างเป็นธรรมชาติ

วิธีการทำงาน:

Realtime API ใช้ WebSocket เพื่อสร้างเซสชันที่มีสถานะและขับเคลื่อนด้วยเหตุการณ์ คุณสตรีมส่วนข้อมูลเสียงไปยัง API และ API จะสตรีมกลับมาทั้งการถอดความและเสียงที่สร้างขึ้น การเชื่อมต่อรองรับ:

• การสตรีมอินพุตเสียง (ส่งส่วนข้อมูลเสียงเมื่อถูกบันทึก)
• การสตรีมเอาต์พุตเสียง (รับเสียงที่สร้างขึ้นแบบเรียลไทม์)
• อินพุต/เอาต์พุตข้อความ (สำหรับการโต้ตอบแบบข้อความ+เสียงผสม)
• การจัดการการขัดจังหวะอัตโนมัติ (หยุดการสร้างเมื่อผู้ใช้พูด)

คุณสมบัติหลัก:

การตรวจจับกิจกรรมเสียง (VAD): API มี VAD เชิงความหมายที่เข้าใจว่าผู้ใช้พูดจบแล้วกับการหยุดชั่วคราว สิ่งนี้สร้างการสนทนาที่เป็นธรรมชาติมากขึ้น

ความสามารถแบบ Multimodal: เข้าถึงความสามารถแบบ multimodal ดั้งเดิมของ GPT-4o ได้โดยตรง ประมวลผลทั้งเสียงและข้อความในโมเดลแบบรวม

ความหน่วงต่ำ: ออกแบบมาสำหรับความหน่วงต่ำกว่า 500ms สำหรับการโต้ตอบด้วยเสียง เหมาะสำหรับการสนทนาแบบเรียลไทม์

WebSocket เทียบกับ HTTP: การเปรียบเทียบประสิทธิภาพ

การเลือกระหว่าง WebSocket และ HTTP ขึ้นอยู่กับลักษณะของแอปพลิเคชันของคุณ นี่คือเมื่อแต่ละโปรโตคอลทำงานได้ดีที่สุด

เมื่อ WebSocket ทำงานได้ดีกว่า HTTP

ปริมาณการเรียกใช้เครื่องมือสูง:
หากเวิร์กโฟลว์ของคุณมีการเรียกใช้เครื่องมือตามลำดับ 10+ ครั้งขึ้นไป การเชื่อมต่อแบบถาวรของ WebSocket จะช่วยขจัดค่าใช้จ่ายในการตั้งค่าซ้ำๆ การร้องขอ HTTP แต่ละครั้งต้องมี:

• การค้นหา DNS (หากไม่ได้แคชไว้)
• การจับมือ TCP (3-way)
• การจับมือ TLS (2 รอบสำหรับการส่งข้อมูลสำหรับ TLS 1.3)
• ส่วนหัวการร้องขอ/การตอบกลับ HTTP

WebSocket ทำสิ่งนี้เพียงครั้งเดียว จากนั้นจึงนำการเชื่อมต่อกลับมาใช้ใหม่

แอปพลิเคชันที่คำนึงถึงความหน่วง:
สำหรับแอปพลิเคชันเสียงหรือแชทแบบเรียลไทม์ที่ทุกมิลลิวินาทีมีความสำคัญ การเชื่อมต่อแบบถาวรของ WebSocket และความสามารถในการสตรีมจะช่วยลดความหน่วงที่รับรู้ได้อย่างมาก

การอัปเดตที่เริ่มต้นโดยเซิร์ฟเวอร์:
WebSocket ช่วยให้เซิร์ฟเวอร์สามารถส่งข้อมูลไปยังไคลเอนต์ได้โดยไม่ต้องมีการพอลลิง สำหรับงานตัวแทนที่ใช้เวลานาน เซิร์ฟเวอร์สามารถส่งการอัปเดตความคืบหน้าเมื่อเกิดเหตุการณ์ขึ้น

เมื่อ HTTP เพียงพอ

การร้องขอ-ตอบกลับแบบง่าย:
สำหรับความสามารถในการเรียก API แบบครั้งเดียว หรือเวิร์กโฟลว์ที่มีการเรียกใช้เครื่องมือ 1-2 ครั้ง HTTP นั้นง่ายต่อการนำไปใช้งานและดีบัก นักพัฒนาส่วนใหญ่คุ้นเคยกับไคลเอนต์ HTTP และโครงสร้างพื้นฐาน (โหลดบาลานเซอร์, พร็อกซี) ก็จัดการ HTTP ได้ดี

การดำเนินการแบบไร้สถานะ:
หากคุณไม่จำเป็นต้องรักษาสถานะเซสชันระหว่างการร้องขอ ลักษณะไร้สถานะของ HTTP ก็เป็นข้อดี — ไม่จำเป็นต้องมีการจัดการการเชื่อมต่อ

ข้อจำกัดของโครงสร้างพื้นฐาน:
สภาพแวดล้อมการปรับใช้บางอย่าง (ฟังก์ชันไร้เซิร์ฟเวอร์, พร็อกซีบางประเภท) ไม่รองรับการเชื่อมต่อ WebSocket ที่ใช้งานได้ยาวนาน HTTP ใช้งานได้ทั่วไป

เมตริกประสิทธิภาพ

จากเอกสารประกอบของ OpenAI และการทดสอบของชุมชน:

วิธีใช้โหมด Responses API WebSocket

มานำการเชื่อมต่อ WebSocket ไปใช้กับ Responses API สำหรับเวิร์กโฟลว์แบบ agentic กัน

ข้อกำหนดเบื้องต้น

• คีย์ OpenAI API ที่สามารถเข้าถึง Responses API
• ไลบรารีไคลเอนต์ WebSocket (ws สำหรับ Node.js หรือ websocket-client สำหรับ Python)
• ความเข้าใจเกี่ยวกับการเรียกใช้เครื่องมือใน OpenAI API

การตั้งค่าการเชื่อมต่อ

ตัวอย่าง Node.js:

const WebSocket = require('ws');

// Connect to Responses API WebSocket endpoint
const ws = new WebSocket('wss://api.openai.com/v1/responses', {
  headers: {
    'Authorization': `Bearer ${process.env.OPENAI_API_KEY}`,
    'OpenAI-Beta': 'responses-api=v1'
  }
});

ws.on('open', () => {
  console.log('Connected to OpenAI Responses API');

  // Send initial request
  const initialMessage = {
    model: 'gpt-4o',
    messages: [
      { role: 'user', content: 'Help me analyze this codebase and suggest improvements.' }
    ],
    tools: [
      {
        type: 'function',
        function: {
          name: 'read_file',
          description: 'Read contents of a file',
          parameters: {
            type: 'object',
            properties: {
              path: { type: 'string', description: 'File path to read' }
            },
            required: ['path']
          }
        }
      },
      {
        type: 'function',
        function: {
          name: 'search_code',
          description: 'Search for code patterns',
          parameters: {
            type: 'object',
            properties: {
              pattern: { type: 'string', description: 'Regex pattern to search' }
            },
            required: ['pattern']
          }
        }
      }
    ]
  };

  ws.send(JSON.stringify(initialMessage));
});

ws.on('message', (data) => {
  const response = JSON.parse(data);
  console.log('Received:', response);

  // Check if model wants to call tools
  if (response.choices[0].finish_reason === 'tool_calls') {
    const toolCalls = response.choices[0].message.tool_calls;

    // Execute tools (simplified)
    const toolResults = toolCalls.map(call => ({
      tool_call_id: call.id,
      output: executeToolLocally(call.function.name, call.function.arguments)
    }));

    // Continue the conversation with tool results
    const continuation = {
      previous_response_id: response.id, // Reference previous response
      input: toolResults
    };

    ws.send(JSON.stringify(continuation));
  }
});

ws.on('error', (error) => {
  console.error('WebSocket error:', error);
});

ws.on('close', () => {
  console.log('Connection closed');
});

function executeToolLocally(name, args) {
  // Your tool execution logic
  if (name === 'read_file') {
    const { path } = JSON.parse(args);
    return fs.readFileSync(path, 'utf-8');
  }
  // ... other tools
}

ตัวอย่าง Python:

import websocket
import json
import os

def on_message(ws, message):
    response = json.loads(message)
    print(f"Received: {response}")

    # Handle tool calls
    if response['choices'][0]['finish_reason'] == 'tool_calls':
        tool_calls = response['choices'][0]['message']['tool_calls']

        # Execute tools
        tool_results = []
        for call in tool_calls:
            result = execute_tool(call['function']['name'],
                                 json.loads(call['function']['arguments']))
            tool_results.append({
                'tool_call_id': call['id'],
                'output': result
            })

        # Send continuation with only new input + previous_response_id
        continuation = {
            'previous_response_id': response['id'],
            'input': tool_results
        }
        ws.send(json.dumps(continuation))

def on_error(ws, error):
    print(f"Error: {error}")

def on_close(ws, close_status_code, close_msg):
    print("Connection closed")

def on_open(ws):
    print("Connected to OpenAI Responses API")

    # Send initial request
    initial_message = {
        'model': 'gpt-4o',
        'messages': [
            {'role': 'user', 'content': 'Analyze this codebase and suggest improvements.'}
        ],
        'tools': [
            {
                'type': 'function',
                'function': {
                    'name': 'read_file',
                    'description': 'Read file contents',
                    'parameters': {
                        'type': 'object',
                        'properties': {
                            'path': {'type': 'string'}
                        },
                        'required': ['path']
                    }
                }
            }
        ]
    }
    ws.send(json.dumps(initial_message))

def execute_tool(name, args):
    if name == 'read_file':
        with open(args['path'], 'r') as f:
            return f.read()
    # ... other tools

# Create WebSocket connection
ws = websocket.WebSocketApp(
    "wss://api.openai.com/v1/responses",
    header={
        "Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}",
        "OpenAI-Beta": "responses-api=v1"
    },
    on_open=on_open,
    on_message=on_message,
    on_error=on_error,
    on_close=on_close
)

ws.run_forever()

รายละเอียดการนำไปใช้งานที่สำคัญ

การจัดการสถานะ:
ความแตกต่างที่สำคัญจาก HTTP คือการใช้ previous_response_id ในการดำเนินการต่อ สิ่งนี้จะบอก API ให้ใช้สถานะที่แคชไว้จากการตอบกลับล่าสุดซ้ำ

การดำเนินการต่อเฉพาะอินพุต:
เมื่อดำเนินการต่อในเทิร์น ให้ส่งเพียง:

• previous_response_id: อ้างอิงถึงการตอบกลับที่แคชไว้
• input: ข้อมูลใหม่ (ผลลัพธ์เครื่องมือ ข้อความผู้ใช้ ฯลฯ)

อย่าส่งอาร์เรย์ messages ทั้งหมดซ้ำ — เซิร์ฟเวอร์มีบริบทนั้นอยู่แล้ว

การเก็บรักษาข้อมูลเป็นศูนย์:
หากต้องการใช้ ZDR กับโหมด WebSocket ให้ใส่ store: false ในคำขอเริ่มต้นของคุณ

วิธีใช้โหมด Realtime API WebSocket

Realtime API เปิดใช้งานการโต้ตอบด้วยเสียงที่มีความหน่วงต่ำ นี่คือวิธีนำไปใช้งาน

ข้อกำหนดเบื้องต้น

• คีย์ OpenAI API ที่สามารถเข้าถึง Realtime API
• ความสามารถในการบันทึก/เล่นเสียง
• ไลบรารีไคลเอนต์ WebSocket
• การเข้ารหัสเสียง (24kHz, 16-bit, mono PCM เพื่อผลลัพธ์ที่ดีที่สุด)

การตั้งค่าการเชื่อมต่อ

ตัวอย่าง JavaScript (เบราว์เซอร์):

// เชื่อมต่อกับ Realtime API
const ws = new WebSocket(
  `wss://api.openai.com/v1/realtime?model=gpt-realtime`,
  ['realtime', 'openai-insecure-api-key.' + process.env.OPENAI_API_KEY]
);

ws.addEventListener('open', () => {
  console.log('Connected to Realtime API');

  // กำหนดค่าเซสชัน
  ws.send(JSON.stringify({
    type: 'session.update',
    session: {
      modalities: ['text', 'audio'],
      voice: 'alloy',
      input_audio_format: 'pcm16',
      output_audio_format: 'pcm16',
      turn_detection: {
        type: 'server_vad', // or 'semantic_vad' for smarter detection
        threshold: 0.5,
        prefix_padding_ms: 300,
        silence_duration_ms: 500
      }
    }
  }));
});

ws.addEventListener('message', (event) => {
  const message = JSON.parse(event.data);

  switch (message.type) {
    case 'session.created':
      console.log('Session created:', message.session);
      break;

    case 'conversation.item.created':
      console.log('New item:', message.item);
      break;

    case 'response.audio.delta':
      // ได้รับส่วนข้อมูลเสียง - เล่น
      const audioChunk = base64ToArrayBuffer(message.delta);
      playAudioChunk(audioChunk);
      break;

    case 'response.audio_transcript.delta':
      // ได้รับส่วนข้อมูลถอดความ
      console.log('Transcript:', message.delta);
      break;

    case 'input_audio_buffer.speech_started':
      console.log('User started speaking');
      break;

    case 'input_audio_buffer.speech_stopped':
      console.log('User stopped speaking');
      break;

    case 'error':
      console.error('API error:', message.error);
      break;
  }
});

// Send audio from microphone
async function streamMicrophoneToAPI() {
  const stream = await navigator.mediaDevices.getUserMedia({ audio: true });
  const audioContext = new AudioContext({ sampleRate: 24000 });
  const source = audioContext.createMediaStreamSource(stream);
  const processor = audioContext.createScriptProcessor(4096, 1, 1);

  processor.onaudioprocess = (e) => {
    const inputData = e.inputBuffer.getChannelData(0);

    // แปลง Float32 เป็น Int16 PCM
    const pcmData = new Int16Array(inputData.length);
    for (let i = 0; i < inputData.length; i++) {
      pcmData[i] = Math.max(-32768, Math.min(32767, inputData[i] * 32768));
    }

    // ส่งไปยัง API
    ws.send(JSON.stringify({
      type: 'input_audio_buffer.append',
      audio: arrayBufferToBase64(pcmData.buffer)
    }));
  };

  source.connect(processor);
  processor.connect(audioContext.destination);
}

// ส่งอินพุตข้อความ
function sendTextMessage(text) {
  ws.send(JSON.stringify({
    type: 'conversation.item.create',
    item: {
      type: 'message',
      role: 'user',
      content: [
        { type: 'input_text', text: text }
      ]
    }
  }));

  // ร้องขอการสร้างการตอบกลับ
  ws.send(JSON.stringify({
    type: 'response.create'
  }));
}

function playAudioChunk(arrayBuffer) {
  const audioContext = new AudioContext({ sampleRate: 24000 });
  audioContext.decodeAudioData(arrayBuffer, (buffer) => {
    const source = audioContext.createBufferSource();
    source.buffer = buffer;
    source.connect(audioContext.destination);
    source.start();
  });
}

ตัวอย่าง Python:

import websocket
import json
import base64
import pyaudio

# การกำหนดค่าเสียง
RATE = 24000
CHUNK = 4096
FORMAT = pyaudio.paInt16
CHANNELS = 1

audio = pyaudio.PyAudio()

def on_open(ws):
    print("Connected to Realtime API")

    # กำหนดค่าเซสชัน
    ws.send(json.dumps({
        'type': 'session.update',
        'session': {
            'modalities': ['text', 'audio'],
            'voice': 'alloy',
            'input_audio_format': 'pcm16',
            'output_audio_format': 'pcm16',
            'turn_detection': {
                'type': 'server_vad',
                'threshold': 0.5,
                'silence_duration_ms': 500
            }
        }
    }))

    # เริ่มสตรีมไมโครโฟน
    stream_microphone(ws)

def on_message(ws, message):
    data = json.loads(message)

    if data['type'] == 'response.audio.delta':
        # ถอดรหัสและเล่นเสียง
        audio_chunk = base64.b64decode(data['delta'])
        play_audio(audio_chunk)

    elif data['type'] == 'response.audio_transcript.delta':
        print(f"Transcript: {data['delta']}", end='', flush=True)

    elif data['type'] == 'input_audio_buffer.speech_started':
        print("\n[User speaking...]")

    elif data['type'] == 'error':
        print(f"Error: {data['error']}")

def stream_microphone(ws):
    stream = audio.open(
        format=FORMAT,
        channels=CHANNELS,
        rate=RATE,
        input=True,
        frames_per_buffer=CHUNK
    )

    def audio_thread():
        while True:
            audio_data = stream.read(CHUNK)
            ws.send(json.dumps({
                'type': 'input_audio_buffer.append',
                'audio': base64.b64encode(audio_data).decode('utf-8')
            }))

    import threading
    threading.Thread(target=audio_thread, daemon=True).start()

def play_audio(audio_chunk):
    stream = audio.open(
        format=FORMAT,
        channels=CHANNELS,
        rate=RATE,
        output=True
    )
    stream.write(audio_chunk)
    stream.stop_stream()
    stream.close()

# สร้างการเชื่อมต่อ WebSocket
ws = websocket.WebSocketApp(
    f"wss://api.openai.com/v1/realtime?model=gpt-realtime",
    header={
        "Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}"
    },
    on_open=on_open,
    on_message=on_message
)

ws.run_forever()

รายละเอียดการนำไปใช้งานที่สำคัญ

ประเภทเหตุการณ์:

Realtime API ใช้การสื่อสารแบบขับเคลื่อนด้วยเหตุการณ์ เหตุการณ์ทั่วไป:

ไคลเอนต์ → เซิร์ฟเวอร์:

• session.update - กำหนดค่าพารามิเตอร์เซสชัน
• input_audio_buffer.append - ส่งส่วนข้อมูลเสียง
• conversation.item.create - เพิ่มข้อความ
• response.create - ร้องขอการตอบกลับจาก AI

เซิร์ฟเวอร์ → ไคลเอนต์:

• session.created - ยืนยันการตั้งค่าเซสชัน
• response.audio.delta - ส่วนข้อมูลเสียงจาก AI
• response.audio_transcript.delta - การถอดความเสียง AI
• input_audio_buffer.speech_started/stopped - เหตุการณ์ VAD
• error - การแจ้งเตือนข้อผิดพลาด

การตรวจจับกิจกรรมเสียง:

เลือกระหว่างโหมด VAD สองโหมด:

server_vad: การตรวจจับกิจกรรมเสียงพื้นฐานโดยอิงจากระดับเสียงและระยะเวลาความเงียบ

semantic_vad: การตรวจจับที่ชาญฉลาดกว่าที่เข้าใจการหยุดชั่วคราวตามธรรมชาติกับการสิ้นสุดเทิร์น ใช้สิ่งนี้สำหรับการสนทนาที่เป็นธรรมชาติมากขึ้นที่ผู้ใช้อาจหยุดชั่วคราวระหว่างความคิด

การทดสอบการเชื่อมต่อ WebSocket ด้วย Apidog

การทดสอบ WebSocket API แตกต่างจากการทดสอบ HTTP — คุณต้องรักษาการเชื่อมต่อ ส่งเหตุการณ์ และตรวจสอบการไหลของข้อความแบบสองทิศทาง Apidog มีความสามารถในการทดสอบ WebSocket ที่เชี่ยวชาญ

การตั้งค่าการทดสอบ WebSocket ใน Apidog

ขั้นตอนที่ 1: สร้างคำขอ WebSocket

ใน Apidog สร้างคำขอใหม่และเลือก "WebSocket" เป็นโปรโตคอล ป้อน URL การเชื่อมต่อของคุณ:

wss://api.openai.com/v1/responses

ขั้นตอนที่ 2: กำหนดค่าส่วนหัว

เพิ่มส่วนหัวการยืนยันตัวตน:

Authorization: Bearer YOUR_OPENAI_API_KEY
OpenAI-Beta: responses-api=v1

สำหรับ Realtime API คุณยังสามารถใช้การยืนยันตัวตนแบบ URL ได้:

wss://api.openai.com/v1/realtime?model=gpt-realtime

พร้อมคีย์ API ในส่วนหัว subprotocol

ขั้นตอนที่ 3: สร้างการเชื่อมต่อ

คลิก "Connect" เพื่อสร้างการเชื่อมต่อ WebSocket Apidog จะแสดง:

• สถานะการเชื่อมต่อ (เชื่อมต่อแล้ว/ยกเลิกการเชื่อมต่อ)
• เมตริกความหน่วง
• ระยะเวลาการเชื่อมต่อ

ขั้นตอนที่ 4: ส่งเหตุการณ์

ใช้ตัวสร้างข้อความของ Apidog เพื่อส่งเหตุการณ์ JSON สำหรับ Responses API:

{
  "model": "gpt-4o",
  "messages": [
    {
      "role": "user",
      "content": "What's the weather in San Francisco?"
    }
  ],
  "tools": [
    {
      "type": "function",
      "function": {
        "name": "get_weather",
        "description": "Get current weather",
        "parameters": {
          "type": "object",
          "properties": {
            "location": { "type": "string" }
          }
        }
      }
    }
  ]
}

ขั้นตอนที่ 5: ตรวจสอบการตอบกลับ

Apidog จะแสดง:

• ข้อความที่ได้รับทั้งหมดเรียงตามลำดับเวลา
• การประทับเวลาและขนาดข้อความ
• การจัดรูปแบบ JSON และการไฮไลต์ไวยากรณ์
• ความสามารถในการคัดลอก/ส่งออกสำหรับการดีบัก

การทดสอบการดำเนินการต่อ

หากต้องการทดสอบรูปแบบการดำเนินการต่อด้วย previous_response_id:

1. ส่งข้อความเริ่มต้น สังเกต response.id ในการตอบกลับ
2. ส่งการดำเนินการต่อพร้อมอินพุตใหม่เท่านั้น:

{
  "previous_response_id": "resp_abc123",
  "input": [
    {
      "tool_call_id": "call_xyz789",
      "output": "{\"temperature\": 72, \"conditions\": \"sunny\"}"
    }
  ]
}

1. สังเกตความหน่วงที่ลดลงเมื่อเทียบกับการส่งบริบททั้งหมดซ้ำ

การทดสอบ Realtime API

สำหรับ Realtime API, Apidog ช่วยให้คุณสามารถ:

• ส่งส่วนข้อมูลเสียงที่เข้ารหัส base64
• ตรวจสอบเหตุการณ์ session.update
• ติดตามเหตุการณ์ VAD (เริ่ม/หยุดพูด)
• ดูการเปลี่ยนแปลงของการถอดความแบบเรียลไทม์

สิ่งนี้มีประโยชน์อย่างยิ่งสำหรับการดีบักว่าทำไมผู้ช่วยเสียงของคุณถึงตัดคำพูดของผู้ใช้ หรือไม่ตรวจจับคำพูดอย่างถูกต้อง

ตัวแปรสภาพแวดล้อม

จัดเก็บคีย์ API อย่างปลอดภัยโดยใช้ตัวแปรสภาพแวดล้อมของ Apidog:

{{OPENAI_API_KEY}}

สิ่งนี้ช่วยให้คุณสามารถสลับไปมาระหว่างคีย์สำหรับพัฒนาและคีย์สำหรับใช้งานจริงได้โดยไม่ต้องแก้ไขคำขอ

กรณีศึกษาในโลกแห่งความเป็นจริง

มาสำรวจสถานการณ์จริงที่โหมด WebSocket ของ OpenAI มีความเป็นเลิศกัน

กรณีศึกษาที่ 1: ตัวแทนเขียนโค้ดอัตโนมัติ

สถานการณ์: ผู้ช่วยเขียนโค้ดที่วิเคราะห์โค้ดเบส ระบุปัญหา และทำการปรับปรุงโดยอัตโนมัติ

ทำไมต้อง Responses API WebSocket:

• เวิร์กโฟลว์ทั่วไป: อ่านไฟล์ → วิเคราะห์ → ค้นหารูปแบบที่คล้ายกัน → อ่านไฟล์เพิ่มเติม → แนะนำการเปลี่ยนแปลง
• สิ่งนี้สร้างการเรียกใช้เครื่องมือ 15-30 ครั้งต่องาน
• โหมด WebSocket ช่วยลดเวลาดำเนินการทั้งหมดลงประมาณ 40%
• การเชื่อมต่อแบบถาวรรักษาสถานะบริบทในการเรียกใช้เครื่องมือทั้งหมด

รูปแบบการนำไปใช้งาน:

// งานเริ่มต้น
ws.send({ messages: [{ role: 'user', content: 'Audit security vulnerabilities' }], tools: [...] })

// การตอบกลับครั้งแรก: โมเดลเรียก read_file
ws.on('message', (resp1) => {
  ws.send({ previous_response_id: resp1.id, input: [tool_result_1] })
})

// การตอบกลับครั้งที่สอง: โมเดลเรียก search_code
ws.on('message', (resp2) => {
  ws.send({ previous_response_id: resp2.id, input: [tool_result_2] })
})

// ดำเนินการต่ออีก 20+ ครั้ง...

กรณีศึกษาที่ 2: บอทบริการลูกค้าด้วยเสียง

สถานการณ์: บอทสนับสนุนทางโทรศัพท์ที่จัดการคำถามของลูกค้าด้วยการสนทนาที่เป็นธรรมชาติ

ทำไมต้อง Realtime API WebSocket:

• ความหน่วงต่ำเป็นสิ่งสำคัญ (<500ms สำหรับการสนทนาที่เป็นธรรมชาติ)
• ต้องจัดการการขัดจังหวะ (ลูกค้าพูดแทรกบอท)
• ประมวลผลอินพุตเสียงโดยตรงโดยไม่ต้องมีการถอดความเป็นข้อความแยกต่างหาก
• สตรีมการตอบกลับแบบเรียลไทม์ (ไม่ต้องรอประโยคที่สมบูรณ์)

รูปแบบการนำไปใช้งาน:

// สตรีมเสียงโทรศัพท์ไปยัง API
phoneSystem.on('audio', (chunk) => {
  ws.send({
    type: 'input_audio_buffer.append',
    audio: base64Encode(chunk)
  })
})

// เล่นการตอบกลับของ AI ทันที
ws.on('message', (event) => {
  if (event.type === 'response.audio.delta') {
    phoneSystem.playAudio(base64Decode(event.delta))
  }
})

การแก้ไขปัญหาทั่วไป

การเชื่อมต่อล้มเหลว

อาการ: การเชื่อมต่อ WebSocket ไม่เปิดขึ้นมาเลย, เกิดเหตุการณ์ปิดทันที

สาเหตุทั่วไป:

1. คีย์ API ไม่ถูกต้อง - ตรวจสอบส่วนหัว Authorization ของคุณอีกครั้ง
2. ส่วนหัวเบต้าหายไป - Responses API ต้องการ OpenAI-Beta: responses-api=v1
3. ข้อจำกัดของเครือข่าย - เครือข่ายองค์กรบางแห่งบล็อก WebSocket
4. URL ไม่ถูกต้อง - ตรวจสอบ wss:// (ไม่ใช่ ws://) และเส้นทางปลายทาง

วิธีแก้ปัญหา:
ใช้ Apidog เพื่อทดสอบการเชื่อมต่อพร้อมข้อความแสดงข้อผิดพลาดโดยละเอียด ตัวตรวจสอบคำขอจะแสดงส่วนหัวที่ถูกส่งไปอย่างถูกต้อง ทำให้ง่ายต่อการระบุคีย์ API ที่ขาดหายไปหรือไม่ถูกต้อง

โค้ดดีบัก:

ws.on('error', (error) => {
  console.error('Connection error:', error);
});

ws.on('close', (code, reason) => {
  console.log(`Closed with code ${code}: ${reason}`);
  // รหัสทั่วไป:
  // 1006: การปิดที่ผิดปกติ (มักเกิดจากปัญหาการตรวจสอบสิทธิ์)
  // 1008: การละเมิดนโยบาย (ส่วนหัวไม่ถูกต้อง)
});

ความหน่วงสูงแม้ใช้ WebSocket

อาการ: การเชื่อมต่อ WebSocket ใช้งานได้ แต่ไม่เร็วกว่า HTTP

สาเหตุทั่วไป:

1. ไม่ได้ใช้ previous_response_id - คุณกำลังส่งบริบททั้งหมดซ้ำ
2. การเริ่มต้นแบบ Cold start - คำขอแรกในการเชื่อมต่อใหม่จะช้าลง
3. ความหน่วงของเครือข่าย - ระยะทางทางภูมิศาสตร์ไปยังเซิร์ฟเวอร์ API
4. เพย์โหลดขนาดใหญ่ - ส่งข้อมูลที่ไม่จำเป็นในการดำเนินการต่อ

วิธีแก้ปัญหา:
ตรวจสอบว่าคุณกำลังส่งเฉพาะอินพุตใหม่ในการดำเนินการต่อ:

// ผิด - ส่งบริบททั้งหมดทุกครั้ง
ws.send({
  messages: [...allPreviousMessages, newMessage],
  tools: [...]
})

// ถูก - อ้างอิงสถานะที่แคชไว้
ws.send({
  previous_response_id: lastResponse.id,
  input: [newMessage]
})

หน่วยความจำรั่วไหลในการเชื่อมต่อที่ใช้งานได้ยาวนาน

อาการ: หน่วยความจำของแอปพลิเคชันเพิ่มขึ้นเมื่อเวลาผ่านไปกับการเชื่อมต่อแบบถาวร

สาเหตุทั่วไป:

1. ตัวฟังเหตุการณ์ไม่ได้ถูกลบออก - สะสมตัวฟังเมื่อมีการเชื่อมต่อใหม่
2. บัฟเฟอร์เสียงไม่ได้ถูกปล่อย - เก็บการอ้างอิงถึงเสียงที่เล่นไปแล้ว
3. ประวัติข้อความเพิ่มขึ้น - จัดเก็บข้อความที่ได้รับทั้งหมด

วิธีแก้ปัญหา:

// ล้างตัวฟังเหตุการณ์เมื่อเชื่อมต่อใหม่
function cleanupAndReconnect(ws) {
  ws.removeAllListeners();
  ws.close();

  const newWs = createConnection();
  return newWs;
}

// Release audio buffers after playing
function playAndRelease(audioBuffer) {
  const audioContext = new AudioContext({ sampleRate: 24000 });
  audioContext.decodeAudioData(audioBuffer, (buffer) => {
    const source = audioContext.createBufferSource();
    source.buffer = buffer;
    source.connect(audioContext.destination);
    source.start();

    source.onended = () => {
      source.disconnect();
      // บัฟเฟอร์จะถูกเก็บขยะ
    };
  });
}

// จำกัดประวัติข้อความ
const messageHistory = [];
const MAX_HISTORY = 100;

ws.on('message', (data) => {
  messageHistory.push(data);
  if (messageHistory.length > MAX_HISTORY) {
    messageHistory.shift(); // ลบรายการที่เก่าที่สุด
  }
});

สรุป

โหมด WebSocket API ของ OpenAI ปลดล็อกความเป็นไปได้ใหม่ๆ สำหรับแอปพลิเคชัน AI โหมด Responses API WebSocket ให้การดำเนินการที่เร็วขึ้นสูงสุด 40% สำหรับเวิร์กโฟลว์แบบ agentic ที่มีการเรียกใช้เครื่องมือจำนวนมาก ทำให้เหมาะสำหรับผู้ช่วยเขียนโค้ดอัตโนมัติและระบบจัดการ Realtime API ให้ความหน่วงต่ำกว่า 500ms สำหรับแอปพลิเคชันเสียง ทำให้สามารถสนทนาได้อย่างเป็นธรรมชาติและสามารถขัดจังหวะได้

การเลือกโหมดที่เหมาะสมขึ้นอยู่กับกรณีการใช้งานของคุณ:

• Responses API WebSocket: ตัวแทนที่เน้นเครื่องมือ, ผู้ช่วยเขียนโค้ด, เครื่องมือวิจัย (การเรียกใช้เครื่องมือ 10+ ครั้ง)
• Realtime API WebSocket: ผู้ช่วยเสียง, บอทโทรศัพท์, ครูสอนภาษา (การสตรีมเสียง)
• HTTP: การร้องขอแบบง่าย, สภาพแวดล้อมไร้เซิร์ฟเวอร์, การเรียก API 1-5 ครั้ง

ลักษณะการเชื่อมต่อแบบถาวรและขับเคลื่อนด้วยเหตุการณ์ของ WebSocket ต้องการวิธีการทดสอบที่แตกต่างจาก HTTP ทดสอบ OpenAI WebSocket API ของคุณด้วยไคลเอนต์ WebSocket แบบเรียลไทม์ของ Apidog — นำเข้าคีย์ API ของคุณ, สร้างการเชื่อมต่อ, ส่งเหตุการณ์ และตรวจสอบการตอบกลับด้วยการบันทึกรายละเอียด ลองใช้ฟรีเพื่อตรวจสอบการผสานรวมของคุณก่อนการนำไปใช้งานจริง