راهنمای استریم SSE در API پیکو هوش؛ از curl تا مرورگر

پایتون: پارس دستی با httpx

import json, os, httpx

url = "https://api.picoai.ir/v1/chat/completions"
headers = {"Authorization": f"Bearer {os.environ['PICO_API_KEY']}"}
body = {"model": "gpt-6-luna", "stream": True,
        "messages": [{"role": "user", "content": "یک جمله درباره‌ی SSE بگو"}]}

with httpx.stream("POST", url, headers=headers, json=body, timeout=None) as r:
    r.raise_for_status()
    for line in r.iter_lines():
        if not line.startswith("data:"):
            continue
        payload = line[5:].strip()
        if payload == "[DONE]":
            break
        evt = json.loads(payload)
        for ch in evt.get("choices", []):
            piece = ch.get("delta", {}).get("content")
            if piece:
                print(piece, end="", flush=True)
        if evt.get("usage"):
            usage = evt["usage"]
print()

اگر raise_for_status خطا داد، یعنی خطا قبل از شروع استریم رخ داده (مثل 401 یا 402) و بدنه‌ی JSON معمولی دارد. در عمل SDK رسمی کار را ساده‌تر می‌کند؛ راهنمای Python.

مرورگر: fetch و ReadableStream

EventSource فقط GET می‌پذیرد، پس برای POST از fetch استفاده کنید. هرگز کلید را در مرورگر نگذارید؛ این کد باید به سرور خودتان وصل شود که استریم پیکو هوش را عبور می‌دهد.

const res = await fetch('/api/chat', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({ messages: [{ role: 'user', content: 'سلام' }] }),
});
const reader = res.body.getReader();
const decoder = new TextDecoder();
let buf = '';
while (true) {
  const { value, done } = await reader.read();
  if (done) break;
  buf += decoder.decode(value, { stream: true });
  const lines = buf.split('\n');
  buf = lines.pop();              // خط ناقص را نگه دار
  for (const line of lines) {
    if (!line.startsWith('data:')) continue;
    const p = line.slice(5).trim();
    if (p === '[DONE]') continue;
    const evt = JSON.parse(p);
    const piece = evt.choices?.[0]?.delta?.content;
    if (piece) output.textContent += piece;
  }
}

نکته‌ی کلیدی buf.pop() است: یک رویداد ممکن است بین دو chunk شبکه شکسته شود و پارس مستقیم هر chunk خطا می‌دهد. TextDecoder با stream: true از شکستن نویسه‌های چندبایتی فارسی جلوگیری می‌کند.

سرور میانی

سرور شما باید هدرهای زیر را بفرستد تا پروکسی‌ها بافر نکنند:

Content-Type: text/event-stream; charset=utf-8
Cache-Control: no-cache
X-Accel-Buffering: no

در Nginx برای مسیر استریم:

location /api/chat {
    proxy_pass http://app:3000;
    proxy_http_version 1.1;
    proxy_buffering off;
    proxy_read_timeout 300s;
}

در Cloudflare و برخی CDNها فشرده‌سازی یا بافر ممکن است استریم را مختل کند؛ تنظیمات مربوط به آن مسیر را غیرفعال کنید. نمونه‌ی کامل Express در راهنمای Node.

قطع اتصال و هزینه

اگر کلاینت وسط پاسخ اتصال را ببندد، سرور شما باید درخواست بالادستی را هم قطع کند (AbortController). هزینه بر اساس توکن‌های واقعاً تولیدشده/گزارش‌شده است. اگر مدل پیش از قطع، گزارش usage ندهد، تخمین مصرف انجام می‌شود (تقریبی). خطاهایی که پیش از شروع پاسخ رخ دهند بدون هزینه‌اند. فرمول: تخمین هزینه و توکن.

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

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

استریم Anthropic

مسیر /v1/messages با stream: true رویدادهای نام‌دار Anthropic را می‌فرستد (message_start، content_block_delta، message_stop و ...) که ساختارشان با OpenAI فرق دارد. SDK رسمی Anthropic آن‌ها را مدیریت می‌کند؛ آموزش Claude Code.

عیب‌یابی

سوالات متداول

استریم هزینه‌ی بیشتری دارد؟

خیر؛ هزینه فقط به توکن‌ها بستگی دارد.

آیا می‌توانم تعداد توکن را در حین استریم ببینم؟

تعداد دقیق در رویداد usage نزدیک پایان می‌آید؛ در حین استریم می‌توانید تخمین بزنید.

آیا همه‌ی مدل‌ها استریم دارند؟

مدل‌های چت کاتالوگ استریم را پشتیبانی می‌کنند؛ مشخصات هر مدل در کاتالوگ.

مرجع؟

مستندات، curl.