Running a language model
Serve chat completions from your own weights, with a request shape you define, and see where it stops scaling.
Introduction
Chat completions from your own weights, with a request shape you decide.
Qwen3 4B Instruct is Apache 2.0 and ungated, so you deploy it with no token.
The app
"""A language model behind an endpoint whose shape you decide."""
from __future__ import annotations
from typing import Literal
from pydantic import BaseModel, Field
from runware_serverless import serve
MODEL = "Qwen/Qwen3-4B-Instruct-2507"
"""Apache 2.0 and ungated, so this deploys with no token of any kind."""
CACHE = "/weights"
"""Mount a volume here, the same as any other app with weights."""
DEFAULT_TOKENS = 512
DEFAULT_TEMPERATURE = 0.7
class Message(BaseModel):
role: Literal["system", "user", "assistant"]
content: str
class Request(BaseModel):
messages: list[Message] = Field(description="The conversation so far, oldest first.")
max_tokens: int = Field(default=DEFAULT_TOKENS, ge=1, le=4096)
temperature: float = Field(default=DEFAULT_TEMPERATURE, ge=0, le=2)
class Reply(BaseModel):
content: str
tokens: int = Field(description="Tokens generated, not counting the prompt.")
@serve
class ChatEndpoint:
requirements = [
"torch==2.14.1",
"transformers==5.21.0",
"accelerate==1.15.0",
]
def load(self) -> None:
if hasattr(self, "model"):
return
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
self.tokenizer = AutoTokenizer.from_pretrained(MODEL, cache_dir=CACHE)
self.model = AutoModelForCausalLM.from_pretrained(
MODEL, torch_dtype=torch.bfloat16, cache_dir=CACHE
).to("cuda")
self.model.eval()
def predict(self, request: Request) -> Reply:
import torch
# The template is the model's own, shipped with its tokenizer. The turn
# markers a model was trained on are not yours to guess at, and a prompt
# assembled by hand is how an instruct model ends up answering nothing.
prompt = self.tokenizer.apply_chat_template(
[message.model_dump() for message in request.messages],
tokenize=False,
add_generation_prompt=True,
)
inputs = self.tokenizer([prompt], return_tensors="pt").to(self.model.device)
with torch.inference_mode():
produced = self.model.generate(
**inputs,
max_new_tokens=request.max_tokens,
temperature=request.temperature,
do_sample=request.temperature > 0,
)
# generate returns the prompt back with the answer appended, so the answer
# is what sits past the input length.
answer = produced[0][inputs.input_ids.shape[-1] :]
return Reply(
content=self.tokenizer.decode(answer, skip_special_tokens=True),
tokens=len(answer),
)Loading the model
Nothing about it is special to the platform. You load it in load, you keep it on a volume, and you answer requests from it.
The request shape is yours
messages, max_tokens and temperature look familiar because this file wrote them that way, and the platform publishes whatever your signature declares. Want an OpenAI-shaped body, write one. Want three fields of your own, write those. Nothing translates between you and your caller.
The roles publish as an enum, so a fourth role is refused at the boundary and never reaches your worker.
The prompt template
apply_chat_template reads the turn markers the tokenizer ships with, and you should not guess at those. Assemble the prompt by hand and an instruct model will answer you with nothing.
Where this stops scaling
generate serves one request at a time. You get no continuous batching and no paged attention, so two callers queue behind each other for the whole length of a generation. That is fine for a single consumer or an internal tool, and wrong for real traffic.
A server built for that work brings its own HTTP server, which makes it a container deploy instead of this one. Start here, and move when you can see the queue.
Deploying it
runware serverless deploy model.py \
--id chat-endpoint \
--gpu-type rtx-pro-6000 \
--volume /weightsCalling it
import { createClient } from '@runware/sdk'
const client = await createClient({ apiKey: process.env.RUNWARE_API_KEY })
const task = await client.invoke({
appId: 'chat-endpoint',
endpointPath: 'predict',
payload: {
request: {
messages: [
{
role: 'user',
content: 'Name three uses for a serverless GPU.'
}
]
}
}
})import asyncio
import os
from runware import Runware
async def main():
client = Runware(api_key=os.environ["RUNWARE_API_KEY"])
try:
task = await client.invoke(
"chat-endpoint",
"predict",
{
"request": {
"messages": [
{
"role": "user",
"content": "Name three uses for a serverless GPU."
}
]
}
},
)
finally:
await client.close()
asyncio.run(main())curl https://api.serverless.runware.ai/v1/apps/chat-endpoint/invoke-async/predict \
-H "Authorization: Bearer $RUNWARE_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"taskId": "7c9e6679-7425-40de-944b-e07fc1f90ae7",
"payload": {
"request": {
"messages": [
{
"role": "user",
"content": "Name three uses for a serverless GPU."
}
]
}
}
}'cat <<'EOF' | runware serverless apps invoke chat-endpoint predict -f -
{
"request": {
"messages": [
{
"role": "user",
"content": "Name three uses for a serverless GPU."
}
]
}
}
EOF{
"taskId": "7c9e6679-7425-40de-944b-e07fc1f90ae7",
"payload": {
"request": {
"messages": [
{
"role": "user",
"content": "Name three uses for a serverless GPU."
}
]
}
}
}The reply comes back at output.predictions[0], with content and the number of tokens generated.
Send the whole conversation each time. The worker keeps nothing between requests, and a handler that did would be holding state a second worker does not have.
Measured on an RTX PRO 6000: 5m48s for the first call with an empty volume, 56 seconds for a restart that finds the weights already there, and 13 seconds on a worker already serving. The volume does not keep a worker warm, it makes a restart cheap.