كيفية استخدام واجهة برمجة تطبيقات OpenAI WebSocket Mode؟

ملخص سريع

تقدم OpenAI وضعين لواجهة برمجة تطبيقات WebSocket لاستخدامات مختلفة: وضع WebSocket لواجهة برمجة تطبيقات الاستجابات (Responses API) لسير العمل المعتمد على الوكلاء مع استدعاءات الأدوات الكثيفة (أسرع بنسبة تصل إلى 40% لأكثر من 20 استدعاءً للأدوات)، وواجهة برمجة تطبيقات الوقت الفعلي (Realtime API) لتطبيقات الصوت والكلام ذات زمن الاستجابة المنخفض. يستخدم كلاهما اتصالات WebSocket المستمرة بدلاً من طلبات HTTP عديمة الحالة، مما يقلل زمن الاستجابة عن طريق إلغاء النفقات العامة المتكررة للاتصال وتمكين التفاعلات القائمة على الأحداث وذات الحالة.

مقدمة

تطورت واجهة برمجة تطبيقات OpenAI لتتجاوز أنماط الطلب والاستجابة البسيطة. بالنسبة للتطبيقات التي تتطلب استدعاءات أدوات سريعة أو تدفقًا صوتيًا في الوقت الفعلي، فإن نموذج HTTP التقليدي يخلق نفقات غير ضرورية. يتطلب كل طلب جديد إعداد الاتصال، والمصادقة، ونقل الحالة—حتى عندما تستمر في نفس المحادثة.

واجهة برمجة تطبيقات WebSocket من OpenAI تحل هذه المشكلة من خلال الحفاظ على اتصال ثنائي الاتجاه ومستمر. بالنسبة لسير عمل الوكلاء مع أكثر من 20 استدعاءً تسلسليًا للأدوات، يترجم هذا إلى تنفيذ أسرع بنسبة 40% تقريبًا من البداية إلى النهاية. وبالنسبة لتطبيقات الصوت، فإنه يتيح محادثات طبيعية قابلة للمقاطعة بزمن استجابة أقل من 500 مللي ثانية.

💡

كان اختبار اتصالات WebSocket يتطلب تقليديًا أدوات تصحيح معقدة. تتيح لك واجهة اختبار WebSocket من Apidog إنشاء اتصالات وإرسال الأحداث ومراقبة الاستجابات في الوقت الفعلي—وهو أمر ضروري للتحقق من تكاملات OpenAI WebSocket قبل النشر في بيئة الإنتاج.

زر

يغطي هذا الدليل وضعين لـ WebSocket من OpenAI: واجهة برمجة تطبيقات الاستجابات لسير عمل الوكلاء الغني بالأدوات، وواجهة برمجة تطبيقات الوقت الفعلي لتدفق الصوت. ستتعرف على متى تستخدم كل منهما، وكيفية تنفيذهما، وكيفية اختبارهما بفعالية.

ما هي واجهة برمجة تطبيقات WebSocket من OpenAI؟

واجهة برمجة تطبيقات WebSocket من OpenAI توفر آلية نقل بديلة لـ 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، كل منهما مُحسّن لحالات استخدام مختلفة.

وضع WebSocket لواجهة برمجة تطبيقات الاستجابات (Responses API)

تدعم واجهة برمجة تطبيقات الاستجابات اتصالات WebSocket لسير عمل الوكلاء حيث تحتاج إلى إجراء العديد من استدعاءات الأدوات المتسلسلة. تم تصميم هذا الوضع لمساعدي الترميز، وأنظمة التنسيق، والوكلاء المستقلين الذين يستدعون الأدوات بشكل متكرر لإنجاز مهام معقدة.

كيف يعمل:

على اتصال WebSocket نشط، تحافظ الخدمة على حالة استجابة سابقة واحدة في ذاكرة التخزين المؤقت المحلية للاتصال (الاستجابة الأحدث). عندما تستمر في دورة، ترسل فقط:

previous_response_id (مرجع للاستجابة الأخيرة)
عناصر إدخال جديدة (رسائل المستخدم، نتائج الأدوات، إلخ)

يعيد الخادم استخدام الحالة المخزنة مؤقتًا بدلاً من إعادة معالجة سجل المحادثة بالكامل.

فوائد الأداء:

بالنسبة لسير العمل الذي يتضمن أكثر من 20 استدعاءً للأدوات، تُفيد OpenAI عن تنفيذ أسرع بنسبة تصل إلى 40% من البداية إلى النهاية مقارنةً بـ HTTP. يأتي هذا من:

لا يوجد إعداد اتصال لكل طلب
لا يوجد حمل مصادقة متكرر
الحالة المخزنة مؤقتًا تقلل وقت المعالجة
زمن استجابة أقل للشبكة لرسائل الاستمرارية الصغيرة

التوافق:

يعمل وضع WebSocket مع كل من خيارات الاحتفاظ بالبيانات الصفرية (ZDR) و store=false، مما يجعله مناسبًا للتطبيقات الحساسة للخصوصية.

وضع WebSocket لواجهة برمجة تطبيقات الوقت الفعلي (Realtime API)

توفر واجهة برمجة تطبيقات الوقت الفعلي إمكانيات تدفق صوتي بزمن استجابة منخفض لتطبيقات التحكم الصوتي. تتيح تفاعلات الكلام-إلى-كلام حيث يمكن للنموذج الاستجابة للإدخال الصوتي بإخراج صوتي، مع التعامل مع المقاطعات بشكل طبيعي.

كيف يعمل:

تستخدم واجهة برمجة تطبيقات الوقت الفعلي WebSocket لإنشاء جلسة قائمة على الأحداث وذات حالة. تقوم ببث مقاطع صوتية إلى واجهة برمجة التطبيقات، وتقوم بدورها ببث كل من النصوص المكتوبة واستجابات الصوت المُنشأة. يدعم الاتصال:

تدفق الإدخال الصوتي (إرسال مقاطع صوتية فور التقاطها)
تدفق الإخراج الصوتي (استقبال الصوت المُنشأ في الوقت الفعلي)
إدخال/إخراج النص (للتفاعلات الهجينة نص + صوت)
المعالجة التلقائية للمقاطعات (إيقاف الإنشاء عندما يتحدث المستخدم)

الميزات الرئيسية:

اكتشاف نشاط الصوت (VAD): تتضمن واجهة برمجة التطبيقات اكتشاف نشاط الصوت الدلالي (semantic VAD) الذي يفهم متى انتهى المستخدم من الكلام مقابل التوقف المؤقت. هذا يخلق تدفق محادثة أكثر طبيعية.

قدرات متعددة الوسائط (Multimodal Capabilities): وصول مباشر إلى قدرات GPT-4o الأصلية متعددة الوسائط، ومعالجة الصوت والنص في نموذج موحد.

زمن استجابة منخفض: مصمم لزمن استجابة أقل من 500 مللي ثانية للتفاعلات الصوتية، ومناسب للمحادثات في الوقت الفعلي.

مقارنة الأداء: WebSocket مقابل HTTP

يعتمد الاختيار بين WebSocket و HTTP على خصائص تطبيقك. إليك متى يتفوق كل بروتوكول.

متى يتفوق WebSocket على HTTP

حجم استدعاء الأدوات العالي:
إذا كان سير عملك يجري أكثر من 10 استدعاءات أدوات متسلسلة، فإن الاتصال المستمر لـ WebSocket يلغي النفقات العامة للإعداد المتكرر. يتطلب كل طلب HTTP ما يلي:

بحث DNS (إذا لم يكن مخزّنًا مؤقتًا)
مصافحة TCP (ثلاثية الاتجاه)
مصافحة TLS (جولتان لـ TLS 1.3)
رؤوس طلب/استجابة HTTP

يقوم WebSocket بذلك مرة واحدة، ثم يعيد استخدام الاتصال.

التطبيقات الحساسة لزمن الاستجابة:
بالنسبة لتطبيقات الصوت أو الدردشة في الوقت الفعلي حيث كل مللي ثانية مهمة، فإن الاتصال المستمر لـ WebSocket وقدرات التدفق تقلل بشكل كبير من زمن الاستجابة الملحوظ.

التحديثات التي يبدأها الخادم:
يسمح WebSocket للخادم بدفع البيانات إلى العملاء دون استقصاء. بالنسبة لمهام الوكلاء طويلة الأمد، يمكن للخادم إرسال تحديثات التقدم مع حدوث الأحداث.

متى يكون HTTP كافياً

طلب-استجابة بسيط:
بالنسبة لاستدعاءات API لمرة واحدة أو لسير العمل الذي يتضمن 1-2 استدعاءً للأدوات، فإن HTTP أبسط في التنفيذ والتصحيح. معظم المطورين على دراية بعملاء HTTP، والبنية التحتية (موازنات التحميل، الوكلاء) تتعامل مع HTTP جيدًا.

العمليات عديمة الحالة:
إذا لم تكن بحاجة إلى الحفاظ على حالة الجلسة بين الطلبات، فإن طبيعة HTTP عديمة الحالة تعتبر ميزة—لا تتطلب إدارة الاتصال.

قيود البنية التحتية:
لا تدعم بعض بيئات النشر (وظائف بلا خادم، وكلاء معينون) اتصالات WebSocket طويلة الأمد. يعمل HTTP بشكل عالمي.

مقاييس الأداء

بناءً على وثائق OpenAI واختبارات المجتمع:

المقياس	HTTP	WebSocket (واجهة برمجة تطبيقات الاستجابات)	WebSocket (واجهة برمجة تطبيقات الوقت الفعلي)
إعداد الاتصال	كل طلب (حوالي 100-300 مللي ثانية)	مرة واحدة (حوالي 100-300 مللي ثانية)	مرة واحدة (حوالي 100-300 مللي ثانية)
سير عمل يتضمن 20+ استدعاءً للأدوات	خط الأساس	أسرع بنسبة 40% تقريبًا	غير متوفر
زمن استجابة الصوت ذهابًا وإيابًا	غير متوفر (لم يصمم لهذا)	غير متوفر	< 500 مللي ثانية
نفقات الذاكرة	منخفضة (عديمة الحالة)	متوسطة (حالة مخزنة مؤقتًا)	متوسطة-عالية (حالة الجلسة)
تعقيد التنفيذ	منخفض	متوسط	متوسط-عالي

كيفية استخدام وضع WebSocket لواجهة برمجة تطبيقات الاستجابات (Responses API)

دعنا نُنفّذ اتصال WebSocket بواجهة برمجة تطبيقات الاستجابات لسير عمل وكيل.

المتطلبات الأساسية

مفتاح OpenAI 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 في الاستمراريات. هذا يخبر واجهة برمجة التطبيقات بإعادة استخدام الحالة المخزنة مؤقتًا من الاستجابة الأخيرة.

الاستمراريات ذات الإدخال فقط:
عند الاستمرار في دورة، أرسل فقط:

previous_response_id: يشير إلى الاستجابة المخزنة مؤقتًا
input: بيانات جديدة (نتائج الأدوات، رسائل المستخدم، إلخ.)

لا تعيد إرسال مصفوفة messages بالكامل — فالخادم يمتلك هذا السياق بالفعل.

الاحتفاظ بالبيانات الصفرية (Zero Data Retention):
لاستخدام ZDR مع وضع WebSocket، قم بتضمين store: false في طلبك الأولي.

كيفية استخدام وضع WebSocket لواجهة برمجة تطبيقات الوقت الفعلي (Realtime API)

تُتيح واجهة برمجة تطبيقات الوقت الفعلي تفاعلات صوتية بزمن استجابة منخفض. إليك كيفية تنفيذها.

المتطلبات الأساسية

مفتاح OpenAI API مع الوصول إلى واجهة برمجة تطبيقات الوقت الفعلي
إمكانيات التقاط/تشغيل الصوت
مكتبة عميل WebSocket
ترميز الصوت (24 كيلو هرتز، 16 بت، أحادي PCM للحصول على أفضل النتائج)

إعداد الاتصال

مثال JavaScript (المتصفح):

// Connect to 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');

  // Configure session
  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':
      // Received audio chunk - play it
      const audioChunk = base64ToArrayBuffer(message.delta);
      playAudioChunk(audioChunk);
      break;

    case 'response.audio_transcript.delta':
      // Received transcript chunk
      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);

    // Convert Float32 to 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));
    }

    // Send to API
    ws.send(JSON.stringify({
      type: 'input_audio_buffer.append',
      audio: arrayBufferToBase64(pcmData.buffer)
    }));
  };

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

// Send text input
function sendTextMessage(text) {
  ws.send(JSON.stringify({
    type: 'conversation.item.create',
    item: {
      type: 'message',
      role: 'user',
      content: [
        { type: 'input_text', text: text }
      ]
    }
  }));

  // Request response generation
  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

# Audio configuration
RATE = 24000
CHUNK = 4096
FORMAT = pyaudio.paInt16
CHANNELS = 1

audio = pyaudio.PyAudio()

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

    # Configure session
    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
            }
        }
    }))

    # Start streaming microphone
    stream_microphone(ws)

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

    if data['type'] == 'response.audio.delta':
        # Decode and play audio
        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=CHANNETLS,
        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()

# Create WebSocket connection
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()

تفاصيل التنفيذ الرئيسية

أنواع الأحداث:

تستخدم واجهة برمجة تطبيقات الوقت الفعلي الاتصال المعتمد على الأحداث. الأحداث الشائعة:

العميل ← الخادم:

session.update - تكوين معلمات الجلسة
input_audio_buffer.append - إرسال أجزاء الصوت
conversation.item.create - إضافة رسائل نصية
response.create - طلب استجابة الذكاء الاصطناعي

الخادم ← العميل:

session.created - يؤكد إعداد الجلسة
response.audio.delta - جزء صوتي من الذكاء الاصطناعي
response.audio_transcript.delta - نسخة صوت الذكاء الاصطناعي
input_audio_buffer.speech_started/stopped - أحداث VAD
error - إشعارات الأخطاء

اكتشاف نشاط الصوت (Voice Activity Detection):

اختر بين وضعين لـ VAD:

server_vad: اكتشاف نشاط الصوت الأساسي بناءً على حجم الصوت ومدة الصمت.

semantic_vad: اكتشاف أذكى يفهم التوقفات الطبيعية مقابل اكتمال الدور. استخدم هذا للمحادثات الأكثر طبيعية حيث قد يتوقف المستخدمون مؤقتًا في منتصف التفكير.

اختبار اتصالات WebSocket باستخدام Apidog

يختلف اختبار واجهات برمجة تطبيقات WebSocket عن اختبار HTTP—تحتاج إلى الحفاظ على اتصال، وإرسال الأحداث، ومراقبة تدفق الرسائل ثنائي الاتجاه. يوفر Apidog إمكانيات اختبار WebSocket متخصصة.

إعداد اختبارات WebSocket في Apidog

الخطوة 1: إنشاء طلب WebSocket

في Apidog، قم بإنشاء طلب جديد واختر "WebSocket" كبروتوكول. أدخل عنوان URL للاتصال الخاص بك:

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

الخطوة 2: تكوين الرؤوس (Headers)

أضف رؤوس المصادقة:

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

بالنسبة لواجهة برمجة تطبيقات الوقت الفعلي، يمكنك أيضًا استخدام المصادقة المستندة إلى عنوان URL:

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

مع مفتاح API في رأس البروتوكول الفرعي.

الخطوة 3: إنشاء اتصال

انقر على "اتصال" لإنشاء اتصال WebSocket. يُظهر Apidog:

حالة الاتصال (متصل/غير متصل)
مقاييس زمن الاستجابة
مدة الاتصال

الخطوة 4: إرسال الأحداث

استخدم منشئ الرسائل في Apidog لإرسال أحداث JSON. بالنسبة لواجهة برمجة تطبيقات الاستجابات:

{
  "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:

أرسل الرسالة الأولية، لاحظ response.id في الاستجابة
أرسل الاستمرارية بإدخال جديد فقط:

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

لاحظ زمن الاستجابة المخفض مقارنة بإعادة إرسال السياق الكامل

اختبار واجهة برمجة تطبيقات الوقت الفعلي (Realtime API)

بالنسبة لواجهة برمجة تطبيقات الوقت الفعلي، يتيح لك Apidog ما يلي:

إرسال أجزاء صوتية مشفرة بـ base64
مراقبة أحداث session.update
تتبع أحداث VAD (بدء/إيقاف الكلام)
عرض دلتا النصوص المكتوبة في الوقت الفعلي

هذا مفيد بشكل خاص لتصحيح أخطاء سبب قطع مساعدك الصوتي للمستخدمين أو عدم اكتشاف الكلام بشكل صحيح.

متغيرات البيئة

قم بتخزين مفاتيح API بأمان باستخدام متغيرات بيئة Apidog:

{{OPENAI_API_KEY}}

يتيح لك هذا التبديل بين مفاتيح التطوير والإنتاج دون تعديل الطلبات.

حالات الاستخدام في العالم الحقيقي

دعنا نستكشف السيناريوهات العملية التي تتفوق فيها أوضاع WebSocket من OpenAI.

حالة الاستخدام 1: وكيل ترميز مستقل

السيناريو: مساعد ترميز يحلل قواعد البيانات، ويحدد المشاكل، ويجري تحسينات بشكل مستقل.

لماذا وضع WebSocket لواجهة برمجة تطبيقات الاستجابات (Responses API):

سير عمل نموذجي: قراءة ملف ← تحليل ← البحث عن أنماط مماثلة ← قراءة المزيد من الملفات ← اقتراح تغييرات
يُنشئ هذا ما بين 15 إلى 30 استدعاءً للأدوات لكل مهمة
يُقلل وضع WebSocket وقت التنفيذ الإجمالي بنسبة 40% تقريبًا
يحافظ الاتصال المستمر على السياق عبر جميع استدعاءات الأدوات

نموذج التنفيذ:

// Initial task
ws.send({ messages: [{ role: 'user', content: 'Audit security vulnerabilities' }], tools: [...] })

// First response: model calls read_file
ws.on('message', (resp1) => {
  ws.send({ previous_response_id: resp1.id, input: [tool_result_1] })
})

// Second response: model calls search_code
ws.on('message', (resp2) => {
  ws.send({ previous_response_id: resp2.id, input: [tool_result_2] })
})

// Continue for 20+ iterations...

حالة الاستخدام 2: روبوت خدمة العملاء الصوتي

السيناريو: روبوت دعم هاتفي يتعامل مع استفسارات العملاء بتدفق محادثة طبيعي.

لماذا وضع WebSocket لواجهة برمجة تطبيقات الوقت الفعلي (Realtime API):

زمن الاستجابة المنخفض حاسم (<500 مللي ثانية للمحادثات الطبيعية)
يحتاج إلى التعامل مع المقاطعات (يتحدث العميل أثناء حديث الروبوت)
يعالج الإدخال الصوتي مباشرة دون الحاجة إلى نسخ منفصل
يدفق الاستجابات في الوقت الفعلي (لا ينتظر الجملة الكاملة)

نموذج التنفيذ:

// Stream phone audio to API
phoneSystem.on('audio', (chunk) => {
  ws.send({
    type: 'input_audio_buffer.append',
    audio: base64Encode(chunk)
  })
})

// Play AI responses immediately
ws.on('message', (event) => {
  if (event.type === 'response.audio.delta') {
    phoneSystem.playAudio(base64Decode(event.delta))
  }
})

استكشاف الأخطاء الشائعة وإصلاحها

فشل الاتصال

الأعراض: اتصال WebSocket لا يفتح أبدًا، حدث إغلاق فوري.

الأسباب الشائعة:

مفتاح API غير صالح - تحقق مرة أخرى من رأس Authorization الخاص بك
رأس بيتا مفقود - واجهة برمجة تطبيقات الاستجابات تتطلب OpenAI-Beta: responses-api=v1
قيود الشبكة - بعض شبكات الشركات تحظر WebSocket
عنوان 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}`);
  // Common codes:
  // 1006: Abnormal closure (often auth issues)
  // 1008: Policy violation (invalid headers)
});

زمن استجابة عالٍ على الرغم من استخدام WebSocket

الأعراض: اتصال WebSocket يعمل ولكنه ليس أسرع من HTTP.

الأسباب الشائعة:

عدم استخدام previous_response_id - أنت تعيد إرسال السياق الكامل
بدء بارد (Cold start) - الطلب الأول على اتصال جديد يكون أبطأ
زمن استجابة الشبكة - المسافة الجغرافية إلى خوادم API
حمولات كبيرة - إرسال بيانات غير ضرورية في الاستمراريات

الحل:
تحقق من أنك ترسل إدخالًا جديدًا فقط في الاستمراريات:

// WRONG - sends full context every time
ws.send({
  messages: [...allPreviousMessages, newMessage],
  tools: [...]
})

// RIGHT - references cached state
ws.send({
  previous_response_id: lastResponse.id,
  input: [newMessage]
})

تسرب الذاكرة في الاتصالات طويلة الأمد

الأعراض: تزداد ذاكرة التطبيق بمرور الوقت مع الاتصال المستمر.

الأسباب الشائعة:

عدم إزالة مستمعي الأحداث - تراكم المستمعين عند إعادة الاتصال
عدم تحرير المخازن المؤقتة الصوتية - الاحتفاظ بإشارات إلى الصوت الذي تم تشغيله
تزايد سجل الرسائل - تخزين جميع الرسائل المستلمة

الحل:

// Clean up event listeners on reconnection
function cleanupAndReconnect(ws) {
  ws.removeAllListeners();
  ws.close();

  const newWs = createConnection();
  return newWs;
}

// Release audio buffers after playing
function playAndRelease(audioBuffer) {
  const source = audioContext.createBufferSource();
  source.buffer = audioBuffer;
  source.connect(audioContext.destination);
  source.start();

  source.onended = () => {
    source.disconnect();
    // Buffer will be garbage collected
  };
}

// Limit message history
const messageHistory = [];
const MAX_HISTORY = 100;

ws.on('message', (data) => {
  messageHistory.push(data);
  if (messageHistory.length > MAX_HISTORY) {
    messageHistory.shift(); // Remove oldest
  }
});

الخاتمة

تفتح أوضاع WebSocket API من OpenAI إمكانيات جديدة لتطبيقات الذكاء الاصطناعي. يوفر وضع WebSocket لواجهة برمجة تطبيقات الاستجابات تنفيذًا أسرع بنسبة تصل إلى 40% لسير عمل الوكلاء مع استدعاءات الأدوات الكثيفة، مما يجعله مثاليًا لمساعدي الترميز المستقلين وأنظمة التنسيق. توفر واجهة برمجة تطبيقات الوقت الفعلي زمن استجابة أقل من 500 مللي ثانية لتطبيقات الصوت، مما يتيح محادثات طبيعية قابلة للمقاطعة.

يعتمد اختيار الوضع الصحيح على حالة الاستخدام الخاصة بك:

WebSocket لواجهة برمجة تطبيقات الاستجابات: الوكلاء الغنيون بالأدوات، مساعدو الترميز، أدوات البحث (أكثر من 10 استدعاءات للأدوات)
WebSocket لواجهة برمجة تطبيقات الوقت الفعلي: المساعدون الصوتيون، روبوتات الهاتف، معلمو اللغة (تدفق الصوت)
HTTP: الطلبات البسيطة، بيئات بلا خادم، 1-5 استدعاءات API

تتطلب طبيعة اتصالات WebSocket المستمرة والقائمة على الأحداث طرق اختبار مختلفة عن HTTP. اختبر واجهات برمجة تطبيقات WebSocket من OpenAI باستخدام عميل WebSocket في الوقت الفعلي من Apidog—قم باستيراد مفتاح API الخاص بك، وإنشاء الاتصالات، وإرسال الأحداث، ومراقبة الاستجابات بتسجيل مفصل. جربه مجانًا للتحقق من تكاملاتك قبل النشر في بيئة الإنتاج.

زر