यह पोस्ट जावास्क्रिप्ट स्ट्रीम एपीआई के साथ काम करने पर विचार करेगी जो HTTP कॉल लाने और टुकड़ों में स्ट्रीमिंग प्रतिक्रिया प्राप्त करने की अनुमति देता है, जो क्लाइंट को सर्वर प्रतिक्रिया पर अधिक प्रतिक्रिया देना शुरू करने की अनुमति देता है। शीघ्रता से चैटजीपीटी जैसे यूआई बनाएं।

एक प्रेरक उदाहरण के रूप में, हम ओपनएआई (या उसी http स्ट्रीमिंग एपीआई का उपयोग करने वाले किसी भी सर्वर) से स्ट्रीमिंग एलएलएम प्रतिक्रिया को संभालने के लिए एक फ़ंक्शन लागू करेंगे, जिसमें कोई एनपीएम निर्भरता नहीं होगी - केवल अंतर्निहित फ़ेच। पूरा कोड यहां है जिसमें घातीय बैकऑफ़, एम्बेडिंग, गैर-स्ट्रीमिंग चैट और चैट पूर्णता और एम्बेडिंग के साथ इंटरैक्ट करने के लिए एक सरल एपीआई शामिल है।

यदि आप यह देखने में रुचि रखते हैं कि ग्राहकों को HTTP स्ट्रीम कैसे लौटाएं, तो इस पोस्ट को देखें।

पूर्ण उदाहरण कोड

यहां पूरा उदाहरण है। हम नीचे प्रत्येक टुकड़े को देखेंगे:

async function createChatCompletion(body: ChatCompletionCreateParams) {
  // Making the request
  const baseUrl = process.env.LLM_BASE_URL || "https://api.openai.com";
  const response = await fetch(baseUrl   "/v1/chat/completions", {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "Authorization": "Bearer "   process.env.LLM_API_KEY,
    },
    body: JSON.stringify(body),
  });
  // Handling errors
  if (!response.ok) {
    const error = await response.text();
    throw new Error(`Failed (${response.status}): ${error}`,
  }
  if (!body.stream) { // the non-streaming case
    return response.json();
  }
  const stream = response.body;
  if (!stream) throw new Error("No body in response");
  // Returning an async iterator
  return {
    [Symbol.asyncIterator]: async function* () {
      for await (const data of splitStream(stream)) {
        // Handling the OpenAI HTTP streaming protocol
        if (data.startsWith("data:")) {
          const json = data.substring("data:".length).trimStart();
          if (json.startsWith("[DONE]")) {
            return;
          }
          yield JSON.parse(json);
        }
      }
    },
  };
}

// Reading the stream  
async function* splitStream(stream: ReadableStream) {
  const reader = stream.getReader();
  let lastFragment = "";
  try {
    while (true) {
      const { value, done } = await reader.read();
      if (done) {
        // Flush the last fragment now that we're done
        if (lastFragment !== "") {
          yield lastFragment;
        }
        break;
      }
      const data = new TextDecoder().decode(value);
      lastFragment  = data;
      const parts = lastFragment.split("\n\n");
      // Yield all except for the last part
      for (let i = 0; i 



ऐसे संस्करण के लिए यहां कोड देखें जिसमें पुनर्प्रयास और अन्य सुधारों के साथ-साथ स्ट्रीमिंग और गैर-स्ट्रीमिंग पैरामीटर वेरिएंट के लिए अच्छे टाइप किए गए ओवरलोड हैं।

बाकी पोस्ट यह समझने के बारे में है कि यह कोड क्या करता है।


  
  
  अनुरोध कर रहा हूँ


यह हिस्सा वास्तव में बहुत आसान है। स्ट्रीमिंग HTTP प्रतिक्रिया सामान्य HTTP अनुरोध से आती है:


const baseUrl = process.env.LLM_BASE_URL || "https://api.openai.com";
const response = await fetch(baseUrl   "/v1/chat/completions", {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
    "Authorization": "Bearer "   process.env.LLM_API_KEY,
  },
  body: JSON.stringify(body),
});




HTTP हेडर सामान्य रूप से भेजे जाते हैं, और स्ट्रीमिंग को सक्षम करने के लिए विशेष रूप से कुछ भी सेट करने की आवश्यकता नहीं होती है। और आप अभी भी HTTP स्ट्रीमिंग के लिए नियमित कैशिंग हेडर का लाभ उठा सकते हैं।


  
  
  त्रुटियों को संभालना


क्लाइंट पक्ष पर त्रुटियों से जुड़ी कहानी HTTP स्ट्रीमिंग के लिए थोड़ी दुर्भाग्यपूर्ण है। अच्छी बात यह है कि HTTP स्ट्रीमिंग के लिए, क्लाइंट को शुरुआती प्रतिक्रिया में तुरंत स्टेटस कोड मिलते हैं और वह वहां विफलता का पता लगा सकता है। HTTP प्रोटोकॉल का नकारात्मक पक्ष यह है कि यदि सर्वर सफलता लौटाता है लेकिन फिर मध्य-स्ट्रीम को तोड़ देता है, तो प्रोटोकॉल स्तर पर ऐसा कुछ भी नहीं है जो क्लाइंट को बताएगा कि स्ट्रीम बाधित हो गई थी। हम नीचे देखेंगे कि ओपनएआई इसके आसपास काम करने के लिए अंत में एक "सभी पूर्ण" प्रहरी को कैसे एनकोड करता है।


if (!response.ok) {
  const error = await response.text();
  throw new Error(`Failed (${response.status}): ${error}`,
}





  
  
  स्ट्रीम पढ़ना


HTTP स्ट्रीमिंग प्रतिक्रिया को पढ़ने के लिए, क्लाइंट प्रतिक्रिया.बॉडी प्रॉपर्टी का उपयोग कर सकता है जो एक ReadableStream है जो आपको .getReader() विधि का उपयोग करके सर्वर से आने वाले टुकड़ों पर पुनरावृति करने की अनुमति देता है। 1



const reader = request.body.getReader();
try {
    while (true) {
      const { value, done } = await reader.read();
      if (done) break;
      const text = TextDecoder().decode(value);
      //... do something with the chunk
    }
} finally {
  reader.releaseLock();
}




यह हमें वापस मिलने वाले हर बिट डेटा को संभालता है, लेकिन ओपनएआई HTTP प्रोटोकॉल के लिए हम उम्मीद कर रहे हैं कि डेटा JSON को न्यूलाइन्स द्वारा अलग किया जाएगा, इसलिए इसके बजाय हम रिस्पॉन्स बॉडी को विभाजित करेंगे और प्रत्येक लाइन को उनके अनुसार "यील्ड" करेंगे। पुनः पूरा किया गया. हम इन-प्रोग्रेस लाइन को लास्टफ्रैगमेंट में बफर करते हैं और केवल पूरी लाइनें लौटाते हैं जिन्हें दो नई लाइनों द्वारा अलग किया गया है:


// stream here is request.body
async function* splitStream(stream: ReadableStream) {
  const reader = stream.getReader();
  let lastFragment = "";
  try {
    while (true) {
      const { value, done } = await reader.read();
      if (done) {
        // Flush the last fragment now that we're done
        if (lastFragment !== "") {
          yield lastFragment;
        }
        break;
      }
      const data = new TextDecoder().decode(value);
      lastFragment  = data;
      const parts = lastFragment.split("\n\n");
      // Yield all except for the last part
      for (let i = 0; i 



यदि यह फ़ंक्शन* और यील्ड सिंटैक्स आपके लिए अपरिचित है, तो फ़ंक्शन* को एक ऐसे फ़ंक्शन के रूप में मानें जो एक लूप में कई चीजें लौटा सकता है, और यील्ड को एक फ़ंक्शन से कई बार कुछ वापस करने के तरीके के रूप में मान सकता है।

फिर आप इस स्प्लिटस्ट्रीम फ़ंक्शन पर लूप कर सकते हैं जैसे:


for await (const data of splitStream(response.body)) {
  // data here is a full line of text. For OpenAI, it might look like
  // "data: {...some json object...}" or "data: [DONE]" at the end
}




यदि यह "प्रतीक्षा के लिए" सिंटैक्स आपको परेशान करता है, तो यह "एसिंक इटरेटर" का उपयोग कर रहा है - एक नियमित इटरेटर की तरह जिसे आप लूप के साथ उपयोग करेंगे, लेकिन हर बार जब इसे अगला मान मिलता है, तो यह प्रतीक्षित होता है। 

 हमारे उदाहरण के लिए, जब हमें ओपनएआई से कुछ टेक्स्ट मिला है और हम और अधिक की प्रतीक्षा कर रहे हैं, तो लूप तब तक इंतजार करेगा जब तक कि स्प्लिटस्ट्रीम एक और मूल्य उत्पन्न न कर दे, जो तब होगा जब प्रतीक्षा रीडर.रीड () एक मान लौटाता है जो समाप्त होता है पाठ की एक या अधिक पंक्तियाँ।

आगे हम एक एसिंक इटरेटर को वापस करने का दूसरा तरीका देखेंगे जो स्प्लिटस्ट्रीम जैसा फ़ंक्शन नहीं है, इसलिए एक कॉलर इस डेटा पर पुनरावृति करने के लिए "प्रतीक्षा के लिए" लूप का उपयोग कर सकता है।


  
  
  एक एसिंक इटरेटर लौटा रहा है


अब हमारे पास टेक्स्ट की पूरी लाइनें लौटाने वाला एक एसिंक इटरेटर है, हम केवल स्प्लिटस्ट्रीम (रिस्पॉन्स.बॉडी) लौटा सकते हैं, लेकिन हम प्रत्येक लाइन को इंटरसेप्ट करना और उन्हें बदलना चाहते हैं, जबकि अभी भी हमारे फ़ंक्शन के कॉलर को पुनरावृत्त करने की अनुमति देते हैं . 

यह दृष्टिकोण उपरोक्त async फ़ंक्शन* सिंटैक्स के समान है। यहां हम एक एसिंक फ़ंक्शन के बजाय सीधे एक एसिंक इटरेटर लौटाएंगे, जो कॉल करने पर एक लौटाता है। अंतर यह है कि प्रकार AsyncGenerator के बजाय AsyncIterator है जिसे पहले कॉल करने की आवश्यकता है। एक AsyncIterator को एक निश्चित नामित फ़ंक्शन द्वारा परिभाषित किया जा सकता है: Symbol.asyncIterator.2



      return {
        [Symbol.asyncIterator]: async function* () {
          for await (const data of splitStream(stream)) {
            //handle the data
            yield data;
          }
        },
      };




यह तब उपयोगी होता है जब आप स्प्लिटस्ट्रीम से आने वाले डेटा से कुछ अलग लौटाना चाहते हैं। हर बार जब स्ट्रीमिंग HTTP अनुरोध से एक नई लाइन आती है, तो स्प्लिटस्ट्रीम इसे प्राप्त करेगा, यह फ़ंक्शन इसे डेटा में प्राप्त करेगा और इसे अपने कॉलर को देने से पहले कुछ कर सकता है। 

आगे हम देखेंगे कि ओपनएआई की स्ट्रीमिंग चैट पूर्णता एपीआई के मामले में विशेष रूप से इस डेटा की व्याख्या कैसे की जाए।


  
  
  OpenAI HTTP स्ट्रीमिंग प्रोटोकॉल को संभालना


ओपनएआई प्रतिक्रिया प्रोटोकॉल लाइनों की एक श्रृंखला है जो डेटा से शुरू होती है: या घटना:, लेकिन हम केवल डेटा प्रतिक्रियाओं को संभालेंगे, क्योंकि चैट पूर्णता के लिए यह उपयोगी हिस्सा है। यदि स्ट्रीम हो गई है तो [DONE] का एक प्रहरी है, अन्यथा यह सिर्फ JSON है।


for await (const data of splitStream(stream)) {
  if (data.startsWith("data:")) {
    const json = data.substring("data:".length).trimStart();
    if (json.startsWith("[DONE]")) {
      return;
    }
    yield JSON.parse(json);
  } else {
    console.debug("Unexpected data:", data);
  }
}





  
  
  यह सब एक साथ लाना


अब जब आप HTTP स्ट्रीमिंग को समझ गए हैं, तो आप एसडीके या लाइब्रेरी पर भरोसा किए बिना स्ट्रीमिंग एपीआई के साथ सीधे काम करने में आत्मविश्वास महसूस कर सकते हैं। यह आपको विलंबता को छिपाने की अनुमति देता है, क्योंकि आपका यूआई कई अनुरोधों के साथ अधिक बैंडविड्थ का उपभोग किए बिना तुरंत अपडेट करना शुरू कर सकता है। आप उपरोक्त फ़ंक्शन का उपयोग वैसे ही कर सकते हैं जैसे आप आधिकारिक ओपनई एनपीएम पैकेज के साथ करेंगे:


  const response = await createChatCompletion({
    model: "llama3",
    messages: [...your messages...],
    stream: true,
  });
  for await (const chunk of response) {
    if (chunk.choices[0].delta?.content) {
      console.log(chunk.choices[0].delta.content);
    }
  }




यहां कोड देखें जो आपको मॉडल को पूर्व-कॉन्फ़िगर करके और .choices[0].delta.content:
 निकालकर इसे और भी आसान बनाने के लिए कुछ उपयोगिता फ़ंक्शन बनाने की सुविधा देता है।


const response = await chatStream(messages);
for await (const content of response) {
  console.log(content);
}




कोड को कॉपी करने से पहले, इसे एसिंक फ़ंक्शंस में एक अभ्यास के रूप में स्वयं लागू करने का प्रयास करें।


  
  
  और अधिक संसाधनों


अपने स्वयं के सर्वर एंडपॉइंट से HTTP स्ट्रीमिंग डेटा वापस करने के बारे में जानकारी के लिए, HTTP स्ट्रीमिंग के साथ AI चैट पर इस पोस्ट को देखें, जो OpenAI (या समान) से डेटा को आपके सर्वर पर स्ट्रीम करता है और साथ ही इसे क्लाइंट तक स्ट्रीम करता है। कस्टम तर्क जैसा कि यह होता है (जैसे डेटाबेस में टुकड़ों को सहेजना)।
एमडीएन दस्तावेज़, हमेशा की तरह, बहुत अच्छे हैं। ऊपर दिए गए लिंक के अलावा, यहां पठनीय स्ट्रीम एपीआई पर एक गाइड है जो दिखाता है कि एक छवि अनुरोध में स्ट्रीम करने के लिए एक पठनीय स्ट्रीम को  टैग से कैसे जोड़ा जाए। ध्यान दें: यह मार्गदर्शिका एक एसिंक इटरेटर के रूप में प्रतिक्रिया.बॉडी का उपयोग करती है, लेकिन वर्तमान में इसे व्यापक रूप से लागू नहीं किया गया है और टाइपस्क्रिप्ट प्रकारों में नहीं।

नोट: आपके पास एक समय में स्ट्रीम का केवल एक रीडर हो सकता है, इसलिए आप आम तौर पर .getReader() को कई बार कॉल नहीं करते हैं - आप संभवतः उस स्थिति में .tee() चाहते हैं, और यदि आप उपयोग करना चाहते हैं। getReader() किसी कारण से कई बार, सुनिश्चित करें कि पहले .releaseLock() हो। ↩



या वैकल्पिक रूप से यदि आप प्रतीक से परिचित नहीं हैं, तो इसका उपयोग किसी ऑब्जेक्ट में कुंजी रखने के लिए किया जाता है जो स्ट्रिंग या संख्या नहीं हैं। यदि आपने asyncIterator नामक कुंजी जोड़ी है तो इस तरह उनमें विरोध नहीं होता है। आप फ़ंक्शन को myIterator[Symbol.asyncIterator]() से एक्सेस कर सकते हैं। ↩