Inizieremo con l'implementazione del bit non streaming. Iniziamo con la modellazione della nostra richiesta:
from typing import List, Optionalfrom pydantic import BaseModel
class ChatMessage(BaseModel):
role: str
content: str
class ChatCompletionRequest(BaseModel):
model: str = "mock-gpt-model"
messages: List(ChatMessage)
max_tokens: Optional(int) = 512
temperature: Optional(float) = 0.1
stream: Optional(bool) = False
IL PyDantic Il modello rappresenta la richiesta del client, con l'obiettivo di replicare il riferimento API. Per brevità, questo modello non implementa tutte le specifiche, ma piuttosto le ossa essenziali necessarie affinché funzioni. Se ti manca un parametro che fa parte del file Specifiche API (Piace top_p), puoi semplicemente aggiungerlo al modello.
IL ChatCompletionRequest modella i parametri che OpenAI utilizza nelle sue richieste. Le specifiche API della chat richiedono di specificare un elenco di ChatMessage (come la cronologia di una chat, il cliente è solitamente responsabile di conservarla e di fornire feedback ad ogni richiesta). Ogni messaggio di chat ha un file role attributo (normalmente system, assistant O user ) e a content attributo contenente il testo effettivo del messaggio.
Successivamente, scriveremo il nostro endpoint di completamento chat FastAPI:
import timefrom fastapi import FastAPI
app = FastAPI(title="OpenAI-compatible API")
@app.post("/chat/completions")
async def chat_completions(request: ChatCompletionRequest):
if request.messages and request.messages(0).role == 'user':
resp_content = "As a mock AI Assitant, I can only echo your last message:" + request.messages(-1).content
else:
resp_content = "As a mock AI Assitant, I can only echo your last message, but there were no messages!"
return {
"id": "1337",
"object": "chat.completion",
"created": time.time(),
"model": request.model,
"choices": ({
"message": ChatMessage(role="assistant", content=resp_content)
})
}
Così semplice.
Testare la nostra implementazione
Supponendo che entrambi i blocchi di codice si trovino in un file chiamato main.pyinstalleremo due librerie Python nel nostro ambiente preferito (è sempre meglio crearne una nuova): pip install fastapi openai e avvia il server da un terminale:
uvicorn main:app
Utilizzando un altro terminale (o avviando il server in background), apriremo una console Python e faremo copia-incolla del seguente codice, preso direttamente da Riferimento al client Python di OpenAI:
from openai import OpenAI# init client and connect to localhost server
client = OpenAI(
api_key="fake-api-key",
base_url="http://localhost:8000" # change the default port if needed
)
# call API
chat_completion = client.chat.completions.create(
messages=(
{
"role": "user",
"content": "Say this is a test",
}
),
model="gpt-1337-turbo-pro-max",
)
# print the top "choice"
print(chat_completion.choices(0).message.content)
Se hai fatto tutto correttamente, la risposta dal server dovrebbe essere stampata correttamente. Vale anche la pena ispezionare il chat_completion obiettare per vedere che tutti gli attributi rilevanti siano quelli inviati dal nostro server. Dovresti vedere qualcosa del genere:
Poiché la generazione LLM tende a essere lenta (costosa dal punto di vista computazionale), vale la pena trasmettere in streaming il contenuto generato al client, in modo che l'utente possa vedere la risposta mentre viene generata, senza dover attendere che finisca. Se ricordi, abbiamo dato ChatCompletionRequest un booleano stream proprietà: consente al client di richiedere che i dati gli vengano ritrasmessi, anziché inviati immediatamente.
Questo rende le cose un po’ più complesse. Creeremo un funzione del generatore per racchiudere la nostra risposta simulata (in uno scenario reale, vorremo un generatore collegato alla nostra generazione LLM)
import asyncio
import jsonasync def _resp_async_generator(text_resp: str):
# let's pretend every word is a token and return it over time
tokens = text_resp.split(" ")
for i, token in enumerate(tokens):
chunk = {
"id": i,
"object": "chat.completion.chunk",
"created": time.time(),
"model": "blah",
"choices": ({"delta": {"content": token + " "}}),
}
yield f"data: {json.dumps(chunk)}\n\n"
await asyncio.sleep(1)
yield "data: (DONE)\n\n"
E ora modificheremo il nostro endpoint originale per restituire una StreamingResponsewhen stream==True
import timefrom starlette.responses import StreamingResponse
app = FastAPI(title="OpenAI-compatible API")
@app.post("/chat/completions")
async def chat_completions(request: ChatCompletionRequest):
if request.messages:
resp_content = "As a mock AI Assitant, I can only echo your last message:" + request.messages(-1).content
else:
resp_content = "As a mock AI Assitant, I can only echo your last message, but there wasn't one!"
if request.stream:
return StreamingResponse(_resp_async_generator(resp_content), media_type="application/x-ndjson")
return {
"id": "1337",
"object": "chat.completion",
"created": time.time(),
"model": request.model,
"choices": ({
"message": ChatMessage(role="assistant", content=resp_content) })
}
Testare l'implementazione dello streaming
Dopo aver riavviato il server uvicorn, apriremo una console Python e inseriremo questo codice (di nuovo, tratto dai documenti della libreria OpenAI)
from openai import OpenAI# init client and connect to localhost server
client = OpenAI(
api_key="fake-api-key",
base_url="http://localhost:8000" # change the default port if needed
)
stream = client.chat.completions.create(
model="mock-gpt-model",
messages=({"role": "user", "content": "Say this is a test"}),
stream=True,
)
for chunk in stream:
print(chunk.choices(0).delta.content or "")
Dovresti vedere ogni parola nella risposta del server stampata lentamente, imitando la generazione del token. Possiamo ispezionare l'ultimo chunk obiettare di vedere qualcosa del genere:
Mettere tutto insieme
Infine, nell'essenza qui sotto, puoi vedere l'intero codice per il server.
Fonte: towardsdatascience.com
