Kartais generatyvinių modelių rezultato pateikimas užtrunka, todėl įdomu pasinaudoti žetonų srautiniu perdavimu, kad rezultatas naudotojo sąsajoje būtų rodomas iš karto. Štai kaip, naudodami "Go", "FastAPI" ir "Javascript", galite sukurti tokį teksto srautinio perdavimo frontendą savo LLM.
Primename, kad simbolis yra unikalus vienetas, kuris gali būti žodis, žodžio dalis arba skyrybos ženklas. Vidutiniškai 1 simbolį sudaro 4 ženklai, o 100 ženklų maždaug atitinka 75 žodžius. Natūralios kalbos apdorojimo modeliai turi paversti jūsų tekstą žetonais, kad galėtų jį apdoroti.
Naudojant tekstą generuojantį dirbtinio intelekto modelį (dar vadinamą generatyviuoju modeliu), atsako laikas gali būti gana ilgas, priklausomai nuo aparatinės įrangos ir modelio dydžio. Pavyzdžiui, naudojant tokį didelį kalbos modelį (alsko vadinamą "LLM") kaip LLaMA 30B, įdiegtą NVIDIA A100 GPU fp16, modelis 100 ženklų sugeneruoja maždaug per 3 sekundes. Taigi, jei tikitės, kad jūsų generatyvinis modelis sugeneruos didelį šimtų ar tūkstančių žodžių tekstą, uždelsimas bus didelis ir reikės laukti galbūt daugiau nei 10 sekundžių, kad gautumėte visą atsakymą.
Ilgas atsakymo laukimas gali būti problema naudotojo patirties požiūriu. Tokiu atveju sprendimas - žetonų srautinė transliacija!
Žetonų srautinis perdavimas - tai kiekvieno naujo žetono generavimas skrydžio metu, užuot laukus, kol bus paruoštas visas atsakymas. Tai yra tai, ką jūs galima pamatyti "ChatGPT" programoje arba, pavyzdžiui, "NLP Cloud ChatDolphin" asistente. Žodžiai rodomi iš karto, kai tik juos sugeneruoja modelis. Išbandykite dirbtinio intelekto asistentą "ChatDolphin" čia.
Žetonų srautinė transliacija su "ChatDolphin" asistentu NLP debesyje. Išbandykite čia.
Pirmasis žingsnis - pasinaudoti išvedimo varikliu, palaikančiu žetonų srautinį duomenų perdavimą.
Štai kelios galimybės, kurias galbūt norėsite apsvarstyti:
Pateikiame pavyzdį, kuriame naudojamas "HuggingFace generate()" metodas:
from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer
from threading import Thread
tokenizer = AutoTokenizer.from_pretrained("gpt2")
model = AutoModelForCausalLM.from_pretrained("gpt2")
inputs = tokenizer(["An increasing sequence: one,"], return_tensors="pt")
streamer = TextIteratorStreamer(tokenizer)
# Run the generation in a separate thread, so that we can fetch the generated text in a non-blocking way.
generation_kwargs = dict(inputs, streamer=streamer, max_new_tokens=20)
thread = Thread(target=model.generate, kwargs=generation_kwargs)
thread.start()
for new_text in streamer:
print(new_text)Šiame pavyzdyje generuojame išvestį, naudodami GPT-2 modelį, ir kiekvieną simbolį išrašome į konsolę, kai tik jis gaunamas.
Dabar, kai pasirinkote išvedimo variklį, reikia pateikti savo modelį ir grąžinti srautiniu būdu perduodamus žetonus.
Jūsų modelis greičiausiai bus paleistas "Python" aplinkoje, todėl jums reikės "Python" serverio, kad galėtumėte grąžinti žetonus. ir padaryti juos prieinamus per HTTP API. FastAPI tapo de facto pasirinkimu tokiais atvejais.
Čia naudojame "Uvicorn" ir "FastAPI" funkciją StreamingResponse, kad galėtume pateikti kiekvieną žetoną iš karto, kai tik jis sugeneruojamas. Štai pavyzdys:
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer
from threading import Thread
model = AutoModelForCausalLM.from_pretrained("gpt2")
tokenizer = AutoTokenizer.from_pretrained("gpt2")
app = FastAPI()
async def generate():
inputs = tokenizer(["An increasing sequence: one,"], return_tensors="pt")
streamer = TextIteratorStreamer(tokenizer)
generation_kwargs = dict(inputs, streamer=streamer, max_new_tokens=20)
thread = Thread(target=model.generate, kwargs=generation_kwargs)
thread.start()
for new_text in streamer:
yield new_text
@app.get("/")
async def main():
return StreamingResponse(generate())
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)Savo srautinio perdavimo serverį galite išbandyti naudodami šią cURL komandą:
curl -N localhost:8000
Dabar turime veikiantį dirbtinio intelekto modelį, kuris tinkamai grąžina srautinius žetonus.
Šiuos srautiniu būdu perduodamus žetonus galime tiesiogiai skaityti iš kliento programos naršyklėje. Tačiau to nedarysime dėl dviejų priežasčių.
Pirma, svarbu atskirti dirbtinio intelekto modelį nuo likusio steko, nes nenorime iš naujo paleisti modelio kiekvieną kartą, kai ketiname atlikti nedidelį API pakeitimą. Turėkite omenyje, kad šiuolaikiniai generatyviniai dirbtinio intelekto modeliai yra labai sunkūs ir jiems iš naujo paleisti dažnai prireikia kelių minučių.
Antroji priežastis yra ta, kad "Python" nebūtinai yra geriausias pasirinkimas kai reikia kurti didelio pralaidumo lygiagrečiąją programą, kokią mes ketiname kurti. Šis pasirinkimas žinoma, galima diskutuoti ir tai gali būti skonio reikalas!
Kaip jau minėta, svarbu pridėti vartus tarp modelio ir galutinio kliento, Go yra gera programavimo kalba tokiai programai. Gamybos metu taip pat galite norėti pridėti atvirkštinį tarp "Go" vartų ir galutinio kliento ir apkrovos balansavimo įrenginį tarp "Go" vartų ir dirbtinio intelekto modelio, kad paskirstyti apkrovą kelioms modelio kopijoms. Tačiau tai nėra mūsų straipsnio sritis!
Mūsų "Go" programa taip pat bus atsakinga už galutinio HTML puslapio atvaizdavimą.
Ši programa pateikia užklausą "FastAPI" programėlei, gauna srautinius žetonus iš "FastAPI" ir perduoda kiekvieną žetoną, naudodamas serverio siunčiamus įvykius (SSE). SSE yra paprastesnis už žiniatinklio lizdus, nes yra vienakryptis. Jis yra geras pasirinkimas, kai norite sukurti programą, kuri siunčia informaciją klientui, neklausydama galimo kliento atsakymo.
Čia pateikiamas Go kodas (HTML/JS/CSS šablonas bus parodytas kitame skyriuje):
package main
import (
"bufio"
"fmt"
"html/template"
"io"
"log"
"net/http"
"strings"
)
var (
templates *template.Template
streamedTextCh chan string
)
func init() {
// Parse all templates in the templates folder.
templates = template.Must(template.ParseGlob("templates/*.html"))
streamedTextCh = make(chan string)
}
// generateText calls FastAPI and returns every token received on the fly through
// a dedicated channel (streamedTextCh).
// If the EOF character is received from FastAPI, it means that text generation is over.
func generateText(streamedTextCh chan<- string) {
var buf io.Reader = nil
req, err := http.NewRequest("GET", "http://127.0.0.1:8000", buf)
if err != nil {
log.Fatal(err)
}
resp, err := http.DefaultClient.Do(req)
if err != nil {
log.Fatal(err)
}
defer resp.Body.Close()
reader := bufio.NewReader(resp.Body)
outerloop:
for {
chunk, err := reader.ReadBytes('\x00')
if err != nil {
if err == io.EOF {
break outerloop
}
log.Println(err)
break outerloop
}
output := string(chunk)
streamedTextCh <- output
}
}
// formatServerSentEvent creates a proper SSE compatible body.
// Server sent events need to follow a specific formatting that
// uses "event:" and "data:" prefixes.
func formatServerSentEvent(event, data string) (string, error) {
sb := strings.Builder{}
_, err := sb.WriteString(fmt.Sprintf("event: %s\n", event))
if err != nil {
return "", err
}
_, err = sb.WriteString(fmt.Sprintf("data: %v\n\n", data))
if err != nil {
return "", err
}
return sb.String(), nil
}
// generate is an infinite loop that waits for new tokens received
// from the streamedTextCh. Once a new token is received,
// it is automatically pushed to the frontend as a server sent event.
func generate(w http.ResponseWriter, r *http.Request) {
flusher, ok := w.(http.Flusher)
if !ok {
http.Error(w, "SSE not supported", http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "text/event-stream")
for text := range streamedTextCh {
event, err := formatServerSentEvent("streamed-text", text)
if err != nil {
http.Error(w, "Cannot format SSE message", http.StatusInternalServerError)
return
}
_, err = fmt.Fprint(w, event)
if err != nil {
http.Error(w, "Cannot format SSE message", http.StatusInternalServerError)
return
}
flusher.Flush()
}
}
// start starts an asynchronous request to the AI engine.
func start(w http.ResponseWriter, r *http.Request) {
go generateText(streamedTextCh)
}
func home(w http.ResponseWriter, r *http.Request) {
if err := templates.ExecuteTemplate(w, "home.html", nil); err != nil {
log.Println(err.Error())
http.Error(w, "", http.StatusInternalServerError)
return
}
}
func main() {
http.HandleFunc("/generate", generate)
http.HandleFunc("/start", start).Methods("POST")
http.HandleFunc("/", home).Methods("GET")
log.Fatal(http.ListenAndServe(":8000", r))
}
Mūsų "/home" puslapyje atvaizduojamas HTML/CSS/JS puslapis (parodytas vėliau). Puslapis "/start" gauna POST užklausą iš JS programos, kuri sukelia užklausą mūsų AI modeliui. O mūsų puslapis "/generate" grąžina rezultatą JS programėlei per serverio siunčiamus įvykius.
Kai tik funkcija start() gauna POST užklausą iš frontendo, ji automatiškai sukuria goroutine, kuri atliks užklausą į mūsų FastAPI programą.
Funkcija generateText() iškviečia FastAPI ir grąžina kiekvieną simbolį, gautą per specialų kanalą (streamedTextCh). Jei iš FastAPI gaunamas EOF simbolis, tai reiškia, kad teksto generavimas baigtas.
Funkcija generate() yra begalinis ciklas, kuris laukia naujų ženklų, gautų iš srautinioTextCh kanalo. Gavus naują simbolį, jis automatiškai išsiunčiamas į frontendą kaip serverio siunčiamas įvykis. Serverio siunčiami įvykiai turi atitikti tam tikrą formatą, kuriame naudojami "event:" ir "data:". priešdėlius, todėl ir sukurta funkcija formatServerSentEvent().
Kad SSE būtų užbaigtas, mums reikia "Javascript" kliento, kuris galėtų klausytis serverio siunčiamų įvykių, prisijungdamas prie puslapio "generuoti". Norėdami suprasti, kaip tai pasiekti, žr. kitą skyrių.
Dabar reikia sukurti katalogą "templates" ir jame pridėti failą "home.html".
Čia pateikiamas "home.html" turinys:
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Our Streamed Tokens App</title>
</head>
<body>
<div id="response-section"></div>
<form method="POST">
<button onclick="start()">Start</button>
</form>
</body>
<script>
// Disable the default behavior of the HTML form.
document.querySelector('form').addEventListener('submit', function(e) {
e.preventDefault()
})
// Make a request to the /start to trigger the request to the AI model.
async function start() {
try {
const response = await fetch("/start", {
method: "POST",
})
} catch (error) {
console.error("Error when starting process:", error)
}
}
// Listen to SSE by subscribing to the /generate page, and
// put the result in the #response-section div.
const evtSource = new EventSource("generate")
evtSource.addEventListener("streamed-text", (event) => {
document.getElementById('response-section').innerHTML = event.data
})
</script>
</html>
Kaip matote, klausytis SSE naršyklėje gana paprasta.
Pirmiausia turite užsiprenumeruoti mūsų SSE galinį tašką (puslapis "/generate"). tada reikia pridėti įvykių klausytoją, kuris nuskaitys srautinius žetonus, kai tik jie bus gauti.
Šiuolaikinės naršyklės automatiškai bando atkurti ryšį įvykio šaltinį, jei kyla ryšio problemų.
Dabar žinote, kaip sukurti modernią generatyvinę dirbtinio intelekto programą, kuri dinamiškai tekstą naršyklėje, à la ChatGPT!
Kaip pastebėjote, tokia paraiška nebūtinai yra paprasta, nes keletas sluoksnių. Ir, žinoma, pirmiau pateiktas kodas yra pernelyg supaprastintas dėl pavyzdį.
Pagrindinė problema, susijusi su simboliniu srautiniu duomenų perdavimu, yra susijusi su tinklo gedimų tvarkymu. Dauguma tokių gedimų įvyks tarp "Go" galinės dalies ir "Javascript" priekinės dalies. Jūs reikės ištirti pažangesnes pakartotinio prisijungimo strategijas ir užtikrinti, kad klaidos būtų būtų tinkamai pranešama apie klaidas vartotojo sąsajoje.
Tikiuosi, kad ši pamoka jums buvo naudinga!
Vincent
NLP Cloud programuotojo advokatas