استریمینگ — Streaming

API آشا به شما اجازه می‌دهد پاسخ را از هر مدلی به‌صورت استریمی دریافت کنید. این کار برای ساخت چت‌های آنی یا اپلیکیشن‌هایی مفید است که UI باید هم‌زمان با تولید پاسخ به‌روزرسانی شود.

برای فعال‌سازی استریمینگ، پارامتر stream را در درخواست روی true قرار دهید. مدل به‌جای بازگرداندن کل پاسخ در یک بار، پاسخ را به‌صورت تکه‌تکه به سمت کلاینت استریم می‌کند.

در اینجا نمونه‌ای از استریم پاسخ و پردازش آن آمده است:

import OpenAI from 'openai';

const openai = new OpenAI({
  baseURL: 'https://app.asha-ai.ir/v1',
  apiKey: '<ASHA_API_KEY>',
});

const question = 'How would you build the tallest building ever?';

const stream = await openai.chat.completions.create({
  model: 'openai/gpt-4o',
  messages: [{ role: 'user', content: question }],
  stream: true,
});

for await (const chunk of stream) {
  const content = chunk.choices?.[0]?.delta?.content;
  if (content) {
    process.stdout.write(content);
  }

  // Final chunk includes usage stats
  if (chunk.usage) {
    console.log('Usage:', chunk.usage);
  }
}
import requests
import json
import os

question = "How would you build the tallest building ever?"

url = "https://app.asha-ai.ir/v1/chat/completions"
headers = {
  "Authorization": f"Bearer {os.environ['ASHA_API_KEY']}",
  "Content-Type": "application/json"
}

payload = {
  "model": "openai/gpt-4o",
  "messages": [{"role": "user", "content": question}],
  "stream": True
}

buffer = ""
with requests.post(url, headers=headers, json=payload, stream=True) as r:
  for chunk in r.iter_content(chunk_size=1024, decode_unicode=True):
    buffer += chunk
    while True:
      try:
        # Find the next complete SSE line
        line_end = buffer.find('n')
        if line_end == -1:
          break

        line = buffer[:line_end].strip()
        buffer = buffer[line_end + 1:]

        # Skip SSE comment lines (starting with ":")
        if line.startswith(':'):
          continue

        if line.startswith('data: '):
          data = line[6:]
          if data == '[DONE]':
            break

          try:
            data_obj = json.loads(data)
            content = data_obj["choices"][0]["delta"].get("content")
            if content:
              print(content, end="", flush=True)
          except json.JSONDecodeError:
            pass
      except Exception:
        break
const question = 'How would you build the tallest building ever?';
const response = await fetch('https://app.asha-ai.ir/v1/chat/completions', {
  method: 'POST',
  headers: {
    Authorization: `Bearer ${process.env.ASHA_API_KEY}`,
    'Content-Type': 'application/json',
  },
  body: JSON.stringify({
    model: 'openai/gpt-4o',
    messages: [{ role: 'user', content: question }],
    stream: true,
  }),
});

const reader = response.body?.getReader();
if (!reader) {
  throw new Error('Response body is not readable');
}

const decoder = new TextDecoder();
let buffer = '';

try {
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    // Append new chunk to buffer
    buffer += decoder.decode(value, { stream: true });

    // Process complete lines from buffer
    while (true) {
      const lineEnd = buffer.indexOf('n');
      if (lineEnd === -1) break;

      const line = buffer.slice(0, lineEnd).trim();
      buffer = buffer.slice(lineEnd + 1);

      // Skip SSE comment lines (starting with ":")
      if (line.startsWith(':')) continue;

      if (line.startsWith('data: ')) {
        const data = line.slice(6);
        if (data === '[DONE]') break;

        try {
          const parsed = JSON.parse(data);
          const content = parsed.choices[0].delta.content;
          if (content) {
            console.log(content);
          }
        } catch (e) {
          // Ignore invalid JSON
        }
      }
    }
  }
} finally {
  reader.cancel();
}

اطلاعات تکمیلی

در جریان‌های SSE ممکن است خط‌های کامنت (خط‌هایی که با : شروع می‌شوند) برای نگه‌داشتن اتصال ارسال شوند. payload این کامنت‌ها طبق قوانین SSE قابل نادیده‌گرفتن است.

payload کامنت را می‌توان با خیال راحت نادیده گرفت (طبق قوانین SSE). اما می‌توانید از آن برای بهبود تجربهٔ کاربری هم استفاده کنید؛ مثلاً برای نمایش یک نشانگر «در حال پردازش».

هشدار: اگر استریم را دستی پارس می‌کنید، قبل از صدا زدن JSON.parse خط‌هایی که با : شروع می‌شوند را رد کنید. دادن خط کامنت به JSON.parse خطا می‌دهد و در صورت مدیریت‌نشدن، حلقهٔ استریم شما را از کار می‌اندازد. کدهای بالا این مورد را مدیریت می‌کنند.

یک parser منطبق با استاندارد مثل eventsource-parser کار کامنت‌ها، فیلدهای data: چندخطی و بافر شدن را برای شما انجام می‌دهد:

TypeScript eventsource-parser
import { createParser } from 'eventsource-parser';

const response = await fetch('https://app.asha-ai.ir/v1/chat/completions', {
  method: 'POST',
  headers: {
    Authorization: `Bearer ${process.env.ASHA_API_KEY}`,
    'Content-Type': 'application/json',
  },
  body: JSON.stringify({
    model: 'openai/gpt-4o',
    messages: [{ role: 'user', content: 'Hello' }],
    stream: true,
  }),
});

// Errors that occur before streaming starts are plain JSON, not SSE
if (!response.ok) {
  const error = await response.json();
  throw new Error(error.error.message);
}

const parser = createParser({
  onEvent(event) {
    if (event.data === '[DONE]') return;
    try {
      const chunk = JSON.parse(event.data);
      const content = chunk.choices?.[0]?.delta?.content;
      if (content) {
        console.log(content);
      }
    } catch {
      // Ignore invalid JSON
    }
  },
});

const reader = response.body!.getReader();
const decoder = new TextDecoder();
while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  parser.feed(decoder.decode(value, { stream: true }));
}

parser فقط قاب‌بندی SSE را مدیریت می‌کند. خطاهایی که حین تولید رخ می‌دهند همچنان به‌صورت رویدادهای عادی data: با فیلد error می‌رسند. نگاه کنید به مدیریت خطاها هنگام استریمینگ در پایین.

بعضی پیاده‌سازی‌های کلاینت SSE ممکن است payload را طبق استاندارد پارس نکنند و هنگام JSON.stringify کردن payload های غیر-JSON، خطای مدیریت‌نشده بدهند. کلاینت‌های زیر را توصیه می‌کنیم:

لغو استریم

درخواست‌های استریمی را می‌توان با قطع کردن اتصال لغو کرد. برای ارائه‌دهنده‌هایی که پشتیبانی می‌کنند، این کار بلافاصله پردازش مدل و محاسبهٔ هزینه را متوقف می‌کند.

نکته: آیا لغو استریم بلافاصله پردازش را متوقف می‌کند به ارائه‌دهندهٔ بالا‌دستی که مدل را سرویس می‌دهد بستگی دارد. برای برخی ارائه‌دهندگان، قطع اتصال ممکن است فقط کلاینت را متوقف کند؛ جزئیات خاص ارائه‌دهنده را بررسی کنید.

برای پیاده‌سازی لغو استریم به این صورت عمل کنید:

const controller = new AbortController();

try {
  const response = await fetch(
    'https://app.asha-ai.ir/v1/chat/completions',
    {
      method: 'POST',
      headers: {
        Authorization: `Bearer ${process.env.ASHA_API_KEY}`,
        'Content-Type': 'application/json',
      },
      body: JSON.stringify({
        model: 'openai/gpt-4o',
        messages: [{ role: 'user', content: 'Write a story' }],
        stream: true,
      }),
      signal: controller.signal,
    },
  );

  // Process the stream...
} catch (error) {
  if (error.name === 'AbortError') {
    console.log('Stream cancelled');
  } else {
    throw error;
  }
}

// To cancel the stream:
controller.abort();
import requests
from threading import Event, Thread

def stream_with_cancellation(prompt: str, cancel_event: Event):
    with requests.Session() as session:
        response = session.post(
            "https://app.asha-ai.ir/v1/chat/completions",
            headers={"Authorization": f"Bearer {os.environ['ASHA_API_KEY']}"},
            json={"model": "openai/gpt-4o", "messages": [{"role": "user", "content": prompt}], "stream": True},
            stream=True
        )

        try:
            for line in response.iter_lines():
                if cancel_event.is_set():
                    response.close()
                    return
                if line:
                    print(line.decode(), end="", flush=True)
        finally:
            response.close()

# Example usage:
cancel_event = Event()
stream_thread = Thread(target=lambda: stream_with_cancellation("Write a story", cancel_event))
stream_thread.start()

# To cancel the stream:
cancel_event.set()
هشدار: لغو فقط برای درخواست‌های استریمیِ ارائه‌دهنده‌هایی که پشتیبانی می‌کنند کار می‌کند. برای درخواست‌های غیر-استریمی یا ارائه‌دهنده‌های پشتیبانی‌نشده، مدل به پردازش ادامه می‌دهد و هزینهٔ کل پاسخ از شما حساب می‌شود.

مدیریت خطاها هنگام استریمینگ

آشا خطاها را بسته به زمانی که در فرآیند استریمینگ رخ می‌دهند به شکل متفاوتی مدیریت می‌کند:

خطا قبل از ارسال هر token

اگر خطا قبل از اینکه حتی یک token به کلاینت استریم شود رخ دهد، آشا یک پاسخ خطای استاندارد JSON را با کد وضعیت HTTP مناسب برمی‌گرداند. این پاسخ از قالب استاندارد خطا پیروی می‌کند:

JSON error response
{
  "error": {
    "code": 400,
    "message": "Invalid model specified"
  }
}

کدهای وضعیت HTTP رایج عبارت‌اند از:

  • 400: Bad Request (پارامتر نامعتبر)
  • 401: Unauthorized (کلید API نامعتبر)
  • 402: Payment Required (اعتبار کافی نیست)
  • 429: Too Many Requests (rate limit)
  • 502: Bad Gateway (خطای ارائه‌دهنده)
  • 503: Service Unavailable (ارائه‌دهندهٔ در دسترس نیست)

خطا بعد از ارسال token ها (حین استریم)

اگر خطا بعد از اینکه چند token به کلاینت استریم شد رخ دهد، آشا دیگر نمی‌تواند کد وضعیت HTTP را تغییر دهد (که اکنون 200 OK است). در عوض، خطا به‌صورت یک رویداد Server-Sent Event (SSE) با ساختار یکپارچه ارسال می‌شود:

SSE mid-stream error
data: {"id":"cmpl-abc123","object":"chat.completion.chunk","created":1234567890,"model":"openai/gpt-4o","error":{"code":"server_error","message":"Provider disconnected unexpectedly"},"choices":[{"index":0,"delta":{"content":""},"finish_reason":"error"}]}

ویژگی‌های کلیدی خطاهای حین استریم:

  • خطا در سطح بالا و در کنار فیلدهای استاندارد پاسخ (id، object، created و…) ظاهر می‌شود.
  • یک آرایهٔ choices با finish_reason: "error" برای پایان دادن درست به استریم اضافه می‌شود.
  • کد وضعیت HTTP همان 200 OK می‌ماند، چون هدرها قبلاً ارسال شده‌اند.
  • استریم بعد از این رویداد خطای یکپارچه خاتمه می‌یابد.

نمونه‌های کد

در اینجا نحوهٔ مدیریت درست هر دو نوع خطا در پیاده‌سازی استریمی‌تان آمده است:

import OpenAI from 'openai';

const openai = new OpenAI({
  baseURL: 'https://app.asha-ai.ir/v1',
  apiKey: '<ASHA_API_KEY>',
});

async function streamWithErrorHandling(prompt: string) {
  try {
    const stream = await openai.chat.completions.create({
      model: 'openai/gpt-4o',
      messages: [{ role: 'user', content: prompt }],
      stream: true,
    });

    for await (const chunk of stream) {
      // Check for errors in chunk
      if ((chunk as any).error) {
        console.error(`Stream error: ${(chunk as any).error.message}`);
        if (chunk.choices?.[0]?.finish_reason === 'error') {
          console.log('Stream terminated due to error');
        }
        return;
      }

      // Process normal content
      const content = chunk.choices?.[0]?.delta?.content;
      if (content) {
        process.stdout.write(content);
      }
    }
  } catch (error: any) {
    // Handle pre-stream errors
    console.error(`Error: ${error.message}`);
  }
}
import requests
import json
import os

async def stream_with_error_handling(prompt):
    response = requests.post(
        'https://app.asha-ai.ir/v1/chat/completions',
        headers={'Authorization': f'Bearer {os.environ["ASHA_API_KEY"]}'},
        json={
            'model': 'openai/gpt-4o',
            'messages': [{'role': 'user', 'content': prompt}],
            'stream': True
        },
        stream=True
    )

    # Check initial HTTP status for pre-stream errors
    if response.status_code != 200:
        error_data = response.json()
        print(f"Error: {error_data['error']['message']}")
        return

    # Process stream and handle mid-stream errors
    for line in response.iter_lines():
        if line:
            line_text = line.decode('utf-8')
            if line_text.startswith('data: '):
                data = line_text[6:]
                if data == '[DONE]':
                    break

                try:
                    parsed = json.loads(data)

                    # Check for mid-stream error
                    if 'error' in parsed:
                        print(f"Stream error: {parsed['error']['message']}")
                        # Check finish_reason if needed
                        if parsed.get('choices', [{}])[0].get('finish_reason') == 'error':
                            print("Stream terminated due to error")
                        break

                    # Process normal content
                    content = parsed['choices'][0]['delta'].get('content')
                    if content:
                        print(content, end='', flush=True)

                except json.JSONDecodeError:
                    pass
async function streamWithErrorHandling(prompt: string) {
  const response = await fetch(
    'https://app.asha-ai.ir/v1/chat/completions',
    {
      method: 'POST',
      headers: {
        'Authorization': `Bearer ${process.env.ASHA_API_KEY}`,
        'Content-Type': 'application/json',
      },
      body: JSON.stringify({
        model: 'openai/gpt-4o',
        messages: [{ role: 'user', content: prompt }],
        stream: true,
      }),
    }
  );

  // Check initial HTTP status for pre-stream errors
  if (!response.ok) {
    const error = await response.json();
    console.error(`Error: ${error.error.message}`);
    return;
  }

  const reader = response.body?.getReader();
  if (!reader) throw new Error('No response body');

  const decoder = new TextDecoder();
  let buffer = '';

  try {
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;

      buffer += decoder.decode(value, { stream: true });

      while (true) {
        const lineEnd = buffer.indexOf('n');
        if (lineEnd === -1) break;

        const line = buffer.slice(0, lineEnd).trim();
        buffer = buffer.slice(lineEnd + 1);

        if (line.startsWith('data: ')) {
          const data = line.slice(6);
          if (data === '[DONE]') return;

          try {
            const parsed = JSON.parse(data);

            // Check for mid-stream error
            if (parsed.error) {
              console.error(`Stream error: ${parsed.error.message}`);
              // Check finish_reason if needed
              if (parsed.choices?.[0]?.finish_reason === 'error') {
                console.log('Stream terminated due to error');
              }
              return;
            }

            // Process normal content
            const content = parsed.choices[0].delta.content;
            if (content) {
              console.log(content);
            }
          } catch (e) {
            // Ignore parsing errors
          }
        }
      }
    }
  } finally {
    reader.cancel();
  }
}

رفتار خاص هر API

endpoint های مختلف API ممکن است خطاهای استریمینگ را کمی متفاوت مدیریت کنند:

  • OpenAI Chat Completions API: اگر هیچ chunk ای پردازش نشده باشد مستقیم ErrorResponse برمی‌گرداند، یا اگر چند chunk پردازش شده باشد اطلاعات خطا را در پاسخ قرار می‌دهد.
  • OpenAI Responses API: بعضی کدهای خطا (مثل context_length_exceeded) را به‌جای خطا، به یک پاسخ موفق با finish_reason: "length" تبدیل می‌کند.