Ģeneratīvajiem modeļiem dažkārt ir nepieciešams laiks, lai iegūtu rezultātu, tāpēc ir interesanti izmantot žetonu straumēšanu, lai redzētu, kā rezultāts parādās lietotāja saskarnē. Lūk, kā jūs varat izveidot šādu teksta straumēšanas frontendu savam LLM, izmantojot Go, FastAPI un Javascript.
Atgādinām, ka marķieris ir unikāla vienība, kas var būt mazs vārds, vārda daļa vai interpunkcija. Vidēji 1 žetonu veido 4 zīmes, un 100 žetoni aptuveni atbilst 75 vārdiem. Dabiskās valodas apstrādes modeļiem, lai apstrādātu tekstu, tas jāpārvērš žetonos.
Izmantojot teksta ģenerēšanas AI modeli (pazīstams arī kā "ģeneratīvais" modelis), atbildes laiks var būt diezgan ilgs atkarībā no jūsu aparatūras un modeļa lieluma. Piemēram, liela valodas modeļa (alsko pazīstams kā "LLM") gadījumā, piemēram, LLaMA 30B, kas izvietots NVIDIA A100 GPU fp16, modelis ģenerē 100 žetonus aptuveni 3 sekundēs. Tātad, ja jūs sagaidāt, ka jūsu ģeneratīvais modelis ģenerēs lielu teksta gabalu ar simtiem vai tūkstošiem vārdu, kavēšanās būs liela, un jums būs jāgaida, iespējams. vairāk nekā 10 sekundes, lai saņemtu pilnu atbildi.
Ilgi gaidot, lai saņemtu atbildi, var rasties problēmas no lietotāja pieredzes viedokļa. Šādā gadījumā risinājums ir žetonu straumēšana!
Žetonu straumēšana ir saistīta ar katra jauna žetona ģenerēšanu uzreiz, nevis gaidīšanu, līdz visa atbilde ir gatava. Tas ir tas, ko jūs var redzēt lietotnē ChatGPT vai, piemēram, NLP Cloud ChatDolphin palīgierīcē. Vārdi parādās, tiklīdz tos ģenerē modelis. Izmēģiniet ChatDolphin mākslīgā intelekta asistentu šeit.
Žetonu straumēšana ar ChatDolphin palīgu NLP mākoņa vidē. Izmēģiniet šeit.
Pirmais solis būs izmantot secināšanas dzinēju, kas atbalsta žetonu straumēšanu.
Šeit ir dažas iespējas, ko varētu apsvērt:
Šeit ir piemērs, kurā izmantota metode HuggingFace generate():
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)Šajā piemērā mēs ģenerējam izvadi ar GPT-2 modeli un izdrukājam katru žetonu konsoles logā, tiklīdz tas pienāk.
Tagad, kad esat izvēlējies secinājumu izdarīšanas dzinēju, jums būs jāapkalpo modelis un jāatgriež straumētie žetoni.
Jūsu modelis, visticamāk, darbosies Python vidē, tāpēc jums būs nepieciešams Python serveris, lai atgrieztu žetonus. un padarītu tos pieejamus, izmantojot HTTP API. FastAPI ir kļuvusi par de facto izvēli šādās situācijās.
Šeit mēs izmantojam Uvicorn un FastAPI StreamingResponse, lai apkalpotu katru žetonu, tiklīdz tas tiek ģenerēts. Šeit ir piemērs:
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)Varat pārbaudīt savu straumēšanas serveri, izmantojot šādu cURL komandu:
curl -N localhost:8000
Tagad mums ir izveidots darbojošs mākslīgā intelekta modelis, kas pareizi atgriež pārraidītos žetonus.
Šos straumētos marķierus varētu tieši nolasīt no klienta lietojumprogrammas pārlūkprogrammā. Taču mēs to nedarīsim divu iemeslu dēļ.
Pirmkārt, ir svarīgi nodalīt mākslīgā intelekta modeli no pārējās kaudzes, jo mēs nevēlamies restartēt modeli katru reizi, kad mēs veiksim nelielas izmaiņas API. Paturiet prātā, ka mūsdienu ģeneratīvie mākslīgā intelekta modeļi ir ļoti smagi. un to restartēšana bieži vien aizņem vairākas minūtes.
Otrs iemesls ir tas, ka Python ne vienmēr ir labākā izvēle. ja runa ir par augstas caurlaidības vienlaicīgas lietojumprogrammas izveidi, kā mēs gatavojamies darīt. Šī izvēle protams, var apspriest, un tas varētu būt arī gaumes jautājums!
Kā minēts iepriekš, ir svarīgi pievienot vārteju starp modeli un galīgo klientu, un Go ir laba programmēšanas valoda šādai lietojumprogrammai. Ražošanā, iespējams, vēlēsieties pievienot arī reverso starp Go vārteju un galīgo klientu, kā arī slodzes balansētāju starp Go vārteju un AI modeli, lai lai sadalītu slodzi uz vairākām modeļa replikām. Bet tas ir ārpus mūsu raksta darbības jomas!
Mūsu Go lietojumprogramma būs atbildīga arī par galīgās HTML lapas atveidošanu.
Šī lietojumprogramma veic pieprasījumu FastAPI lietojumprogrammai, saņem straumētos marķierus no FastAPI un pārsūta katru no tiem tālāk. žetonu uz frontend lietojumprogrammu, izmantojot servera nosūtītos notikumus (SSE). SSE ir vienkāršāks nekā websockets, jo ir vienvirziena. Tas ir laba izvēle, ja vēlaties izveidot lietojumprogrammu, kas sūta informāciju klientam, neklausoties potenciālo klientu, bet klienta atbildi.
Šeit ir Go kods (HTML/JS/CSS veidne tiks parādīta nākamajā sadaļā):
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ūsu "/home" lapa atveido HTML/CSS/JS lapu (parādīts vēlāk). Lapa "/start" saņem POST pieprasījumu no vietnes JS lietojumprogrammu, kas izraisa pieprasījumu mūsu AI modelim. Un mūsu "/generate" lapa atdod rezultātu JS lietojumprogrammai, izmantojot servera nosūtītos notikumus.
Kad funkcija start() saņem POST pieprasījumu no frontend, tā automātiski izveido goroutine, kas veic pieprasījumu. uz mūsu FastAPI lietojumprogrammu.
Funkcija generateText() izsauc FastAPI un atgriež katru tokenu, kas saņemts lidojuma laikā, izmantojot īpašu kanālu (streamedTextCh). Ja no FastAPI tiek saņemta EOF rakstzīme, tas nozīmē, ka teksta ģenerēšana ir pabeigta.
Funkcija generate() ir bezgalīgs cikls, kas gaida jaunus žetonus, kas saņemti no straumētāTextCh kanāla. Tiklīdz ir saņemts jauns žetons, tas tiek automātiski nosūtīts uz frontend kā servera sūtīts notikums. Servera nosūtītajiem notikumiem ir jāievēro īpašs formāts, kurā izmanto "event:" un "data:". prefiksi, tāpēc ir izveidota funkcija formatServerSentEvent().
Lai SSE būtu pilnvērtīgs, mums ir nepieciešams Javascript klients, kas var klausīties servera sūtītos notikumus, abonējot lapu "ģenerēt". Lai saprastu, kā to panākt, skatiet nākamo sadaļu.
Tagad ir jāizveido direktorijs "templates" un tajā jāpievieno fails "home.html".
Šeit ir "home.html" saturs:
<!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>
Kā redzat, SSE klausīšanās pārlūkprogrammā ir diezgan vienkārša.
Vispirms ir jāpiesakās mūsu SSE galapunktam (lapā "/generate"). pēc tam jums jāpievieno notikumu klausītājs, kas nolasīs straumētos žetonus, tiklīdz tie tiek saņemti.
Mūsdienu pārlūkprogrammas automātiski mēģina atjaunot savienojumu notikumu avotu, ja rodas savienojuma problēmas.
Tagad jūs zināt, kā izveidot mūsdienīgu ģeneratīvā mākslīgā intelekta lietojumprogrammu, kas dinamiski straumē tekstu pārlūkprogrammā, à la ChatGPT!
Kā jūs pamanījāt, šāds pieteikums ne vienmēr ir vienkāršs, jo vairāki ir iesaistīti vairāki slāņi. Un, protams, iepriekš minētais kods ir pārāk vienkāršots, lai piemērs.
Galvenā problēma, kas saistīta ar žetonu straumēšanu, ir saistīta ar tīkla kļūmju novēršanu. Lielākā daļa kļūmju notiks starp Go backend un Javascript frontend. Jums būs būs jāizpēta dažas progresīvākas atkārtotas savienošanās stratēģijas un jāpārliecinās, ka kļūdas tiek novērstas. par kļūdām tiktu pareizi ziņots lietotāja saskarnei.
Es ceru, ka šī pamācība jums ir noderīga!
Vincent
NLP Cloud izstrādātāju advokāts