Thank you, thank you for this great introduction. Very hard to live up to the expectations. Great to be here. Today I'm going to talk about a few things. By the way, when I give advice to my students, I tell them, 'Hey, you should talk about one thing and do it well.' I'm not going to do that, so I'm not going to follow my own advice. I'm going to talk about three projects: Ray, vLLM, and Chatbot Arena. The way I'm going to link them is that they actually create a kind of stack. vLLM is using Ray, and Chatbot Arena is using vLLM, but they solve quite different problems. So from that perspective, there are three different pieces. Let me start with Ray. This is a project I started a long time ago, almost 8 years ago. To provide some historical perspective, before Ray we worked on other projects including Spark. Spark was about scaling data processing. At that time, the workloads we were looking to scale were Big Data, if you remember Hadoop of the day, and also classic machine learning like random forest, logistic regression, things like that before LLMs. The target was homogeneous clusters, every node was more or less the same with a CPU and some storage. Ray started 7 years later and targeted deep learning and reinforcement learning, a new workload at that time. The target cluster became more complex, not only distributed but heterogeneous clusters including both CPUs and GPUs. To be more precise, when we started Ray, these were the trends: AI demands were growing quicker than ever, clusters were becoming heterogeneous, and AI workloads themselves were becoming more complex. Everyone sees this plot showing the demands in terms of compute for training the largest models. On the y-axis is log scale, on the x-axis is time, and it shows the size of these models increasing over time. If you do some approximate fitting, you get that the demands in terms of compute for training the largest models were growing like 10x every 18 months. Even if you look at the cost of the latest models, GPT-2 was about $40,000 to train, GPT-3 was between 4 and 5 million, GPT-4 was between 60 and 100 million, and it's expected that GPT-5 will be between 1 and 3 billion. So it's kind of 10x even in terms of dollars to train, and of course you need to factor in improvements in compute, so in terms of FLOPs it's going to be more than 10x. At the same time, we knew that Moore's Law was the cornerstone of progress in computer science, improving performance 2x every 18 months. But Moore's Law started to fizzle at the beginning of the century, and people started to come up with new things like GPUs and DPUs to bridge the gap between the capabilities of a single core. GPUs and DPUs are far more performant, but still the growth is around 2x every 18 months or two years, so the slope is very different. There's still a huge gap, so you need to deal with much more complex infrastructure. It's not enough only to scale horizontally to have more machines, but you also have to have heterogeneous infrastructure composed of CPUs and GPUs. The workloads are becoming more complex. Here's the simplest example, a typical pipeline for machine learning from 5-6 years ago. You get some data like user logs, preprocess these logs, do feature engineering, compute embeddings, train a model or pre-train in today's terminology, then fine-tune or post-train, and then serve the model. You get more data and iterate. Before you serve, you do batch prediction to figure out how the new model performs on past data before deploying it in production. The main point is that each of these stages eventually needs to be scaled. When we started Ray, and even until very recently, there were systems you could use to scale each of these stages in isolation, but to build a pipeline you need to stitch together these distributed systems. The problems are: first, it's hard to develop because each system has its own API. Second, it's hard to deploy and manage different distributed systems, each with its own semantics not only about processing data but about failures and recovery. It's very complicated to have reasonable end-to-end semantics for failures and recovery. Third, it's slow because to move data between stages you need to serialize, deserialize, and store it on a blob store. Ray solves this. Ray is a general framework for distributed computing. You can use Ray to scale all these workloads, and to make it easier we have libraries on top of it, but fundamentally you need only one system instead of many. That's the TL;DR. Ray has a core part and a bunch of libraries. I'm going to talk about Ray Core, which is a distributed framework with a very minimalist API. Ray implements what we call the FAST compute model. FAST stands for Actors and Tasks. The reason Ray is so general is simple: it takes the main compute abstractions you have in almost every language, which are functions and classes. You can call a function remotely, a bit like RPC, transparently. You can instantiate classes remotely, but classes have state encapsulated with them, and they become actors. You take these constructs and you can remotely and transparently run them. In addition, you have a future abstraction. When you instantiate a task remotely, you get back immediately a reference to the future result, without waiting for it to run. This allows parallelism because you can instantiate another task. Finally, there is a shared in-memory distributed object store that allows you to pass arguments and results by reference. In distributed systems, typically arguments are passed by value like RPC, but when you have large data, passing by reference has advantages. Here's a simple Python example. If you have a trivial function that takes one second and you call it twice with different arguments, it will take two seconds. With Ray, you add the @ray.remote decorator to make it a task, and call it with f.remote. When you call f.remote, the program runs on a driver, submits a task for f to the system, and Ray decides where to schedule it. But even before scheduling, it returns a future, a reference to the result. This allows you to call the next function f.remote immediately, so you submit two tasks. The system figures out where to schedule them, and because there is no data dependency, they run in parallel. To get the results, you use ray.get on these futures. f.remote is asynchronous and returns immediately, but ray.get is blocking and waits for the results. When the tasks finish, the results are sent to the driver. For classes, you add the @ray.remote decorator and instantiate the class, which becomes an actor. When you invoke a method on that class, it's also asynchronous. Ray was designed to handle heterogeneous clusters, so you can specify resource requirements like number of CPUs or GPUs for your class or tasks. About shared memory, Ray has a distributed object store. Here's an example where a function G takes as argument the result returned by a function F, so it's G of F of X. First, we call f.remote, which returns a reference to the result. That reference is passed to G.remote as a task. Now there is a dependency between G and F, so G will wait for F to finish. When F finishes, it pushes the result into the shared object store. G has a reference to the result, and when the result X is ready, it is transferred to the node where G runs, and G consumes the input and creates the result. The main point is that X is transferred only once between node 2 and node 3, which is trivially optimal because F runs on node 2 and G runs on node 3. If you implemented this using RPC, you would pass the argument by value, so X would be moved twice: from node 2 to node 1, and then from node 1 to node 3. That's less efficient. This is why you pass by reference to avoid copies when results are very large. The ecosystem is like this: Ray Core is a unified computing framework implementing the FAST compute model with tasks, actors, a shared object store, and futures. You can build virtually any distributed application, but to make it easy to implement some workloads, we provide libraries for data processing, training, hyperparameter tuning, serving, and reinforcement learning. Ray has been quite successful. Many companies use it to implement their AI and ML infrastructures and platforms. Initially, they implemented these platforms by stitching together systems, but now in the second generation of platforms, many use Ray. One of the most impressive use cases is OpenAI training GPT-4 and other large language models on top of Ray. It's a quite successful open source project. Now something else is changing. In the past several years, we have large language models, and cluster heterogeneity has increased. Not only do we have separate GPUs, but we used to train with data parallelism where you have a model on one GPU and shuffle data to build batches. Now we have model parallelism, and you have TPU pods or GPU pods with a bunch of GPUs that are very well connected. One of the biggest trends today is the rapid transition from CPU-centric to accelerator-centric workloads. The valuation of Nvidia is a proof point. This is a challenge, and we need to recognize this strength and probably have changes in curriculum. Ray is very dynamic and flexible like Python, but it has high overhead for small tasks. Like Python, which is very dynamic, it has overhead. You need to pay the cost of RPC and dynamic memory allocation. It's difficult to efficiently support peer-to-peer protocols like RDMA and NVLink for direct communication between GPUs, and it requires careful resource allocation to avoid deadlocks. Let me show you the challenges with a simple example. You have two tasks or actors. You have a value, pass it to the first task A, which increments it, passes it to B, which increments it, and returns the result to the driver. With RPC, you pass 1 to A, it returns 2, then the driver passes 2 to B, and it returns 3. Here's what Ray does to implement this. First, the driver puts the initial value 1 in the shared object store. This gives a reference to the value, which is passed to the task or actor. A passes the future of its own result back to the driver. When A runs, it reads the value 1 and creates the result 2. The driver receives this reference, the future of A's result, and passes it to B. B passes the reference to the final result 3. B reads 2 and creates 3, then sends the result back to the driver. From the driver's point of view, there are two data copies and six RPCs. There are also dynamic memory allocations. In the past, there were four data copies between the driver and the rest of the system. Here, there are only two memory copies to the driver, which is good if the data is huge, but if the data is small, you pay a price for the control overhead and memory allocation. This is the problem we tried to solve with what we call compiled graphs or accelerated DAGs. We wanted to run tasks between 1 and 10 milliseconds with less than 1% overhead, and have GPU-to-GPU communication set up in less than microseconds. If you have repeated computation like training, you compile the graph. You set up the graph in advance and avoid the dynamic overhead for each invocation. You allocate resources in advance and declare the peer-to-peer communication between GPUs. Ideally, with a shared memory object store, you would have little or no control overhead, get rid of the RPC overhead, and avoid dynamic memory allocation. You construct a DAG where A gets the input, sends the output to B, and B sends the result back to the driver. You declare this graph in advance and compile it. Compiling means you start these actors, which run in a loop to get data and produce results. You pre-allocate static buffers and reuse them across multiple executions. Once compiled, you execute the entire graph. Because memory allocation is static, it doesn't change over multiple iterations. This is very fast. For GPU-to-GPU communication, with Ray today, if you want to send a tensor from GPU 0 to GPU 1, you copy it to shared memory in RAM, move it over a slow network to the other node, and copy it back to GPU memory. With compiled DAGs, you create a direct graph between GPU 0 and GPU 1 to transfer the tensor directly, specifying the transport, like NVLink. So basically, if you have repeated execution, you make it static by compiling a static execution DAG and pre-allocating memory, exactly what a compiler does. There are many application examples. For online or batch inference, you read data, do model inference, and write. You gain a lot because you can mix different resources. Read, pre-processing, and post-processing can be done on CPUs, while inference can be done on GPUs. By mixing and matching, you can reduce cost by 10x compared to systems that cannot mix and do everything on GPU. You can do fine-tuning, multi-model search, and embedding computation similarly. Heterogeneity is increasing even for inference. You may want to use different GPUs or CPUs and GPUs. For example, for prefilling to compute attention from the prompt, you can use CPUs, and then for generating one token at a time, you can use GPUs. For multimodal encoding and decoding, you can use weaker, cheaper GPUs, and for running the Transformer inference, you can use expensive, high-performance GPUs. So a lot of things are increasing quite a bit. For pre-training or post-training, it's very complex. That's Ray. Next is vLLM. Ray was a distributed framework for distributed compute on heterogeneous clusters targeting scaling AI workloads. vLLM is one particular workload going up the stack, about serving inference. Inference becomes more and more important because large language models are more general purpose, so they can do more things at inference time. Previously, for every use case you needed to train a model, but now because you have very general models, you don't need to train them all the time, so inference becomes more important. You've heard about test-time compute, which is doing even more at inference time. The problem with this, when we started this project last year, is that it's very expensive. You get only a few requests per second even on an A100 or H100 for Llama 13B, a small model with 13 billion parameters. Between when they were launched last year and this year, model inference decreased by an order of magnitude. One technique used to get these improvements is what I'll describe. Everyone knows how large language models work, but let me remind you briefly because it's important to understand the problem. You predict the next token based on the previous tokens using an attention mechanism. The main point is that there is a dependence between the next token and the previous ones. The problem is that this is sequential, and GPUs get their FLOPs from parallelism. You cannot exploit GPU parallelism, so utilization is low, leading to low throughput and high cost. The solution is to have more queries at the same time, processing them in parallel across queries. But then you run out of memory because for each query you need to keep all previous tokens in memory as embeddings. For a 13 billion parameter model with 2 bytes per parameter, you have 26 gigabytes for parameters. Then you have the KV cache, which keeps the embeddings for the tokens. This is about 30% of memory, plus other activations. Many things have changed since then, like quantization reducing memory for parameters. But the main problem is that for each query, you need to put this prefix in the cache, and each token embedding is about 1 megabyte. People also want longer contexts, like Google Gemini with 1.5 million tokens, so one request can have several gigabytes. At that time, systems handled memory with contiguous allocations. This makes sense because you allocate a contiguous memory region for the entire request. But this leads to fragmentation. You have internal fragmentation because you don't know how long the output will be. You have to reserve space, even if you know exactly how long the output will be, you allocate that space, but you fill it token by token, so there is an opportunity to use memory for other requests. You also have external fragmentation because not all requests are the same size. This was pretty bad. The main takeaway from this plot is that the green stuff is how much you actually use from the entire memory, and the rest is wasted due to inefficiencies. This was state of the art back then. The last one shows that if you were an oracle and knew how much to allocate for each request, it would be 38% utilization. The solution is something we know from operating systems memory management: paging and virtual memory to handle internal and external fragmentation. That's what we do here, an application-level virtual memory for inference. We have KV blocks instead of pages. The smallest unit is a block containing the embedding associated with a token. On the left is the logical KV space, on the right is the physical KV block space. Applications see a contiguous region of memory in the logical space, and we have a table mapping tokens from logical space to physical space. We optimize for this application. For each block, we allocate only when needed. The second column tells how many tokens are filled in the block. For example, in block one, you have 'computer science', so only two tokens are filled. The block size is four. When you generate the next token, you add it to block one, and when the block is exhausted, you allocate another block. Blocks are fixed size, allocated anywhere, so you don't have external fragmentation, only a little internal fragmentation. You can also share blocks, like in operating systems you can share pages. For example, in parallel sampling, you generate different outputs for the same input, so you share the prefix. You can also share the prefix in a multi-turn dialogue, where you feed the model with all previous conversation. There is a lot of prefix sharing. With no external fragmentation, the improvements are massive, 2x to 5x, translating to higher throughput and lower cost. Here are some numbers. At that time, Hugging Face's Text Generation Inference was the most popular inference engine, and vLLM had higher throughput, better performance, a few times improvement. Higher throughput, lower cost, more requests on the same hardware.
Fascinating, you can use these techniques which are before, but when the problem setting changes, the solutions start to change a little bit as well. This is cool. So what are the analogies? What are the similarities? Like I told you, OS pages are the KV blocks. Reduce memory fragmentation, the same reason you can share a page, you can share the blocks like you share pages. This is what are the differences? The eviction right now. Eviction here, in virtual memory you do eviction typically at a page level. Here when you do the eviction, if I am going to miss one of the tokens, before you know it the others are not very useful. So if you evict, you can evict the entire request, the entire prefix. Also the other question is about when you bring it back. When you need it, with an operating system you put it in a swap file and you bring pages back when you need them. It turns out that here, actually in some cases it's faster to recompute. In a swap space, the storage space will be on the CPU RAM right because it's on the same machine, but here it's actually in many cases because the GPU has so much parallelism, it's faster to recompute all these tokens. And there are many other things, we've done a lot of other work here, it's much more sophisticated now, but I'll be happy to answer questions. It's very successful right now. It's probably the most popular open source inference system, it's used by many many companies, cloud providers, and the paged attention as a technique is also used by basically every other inference engine, not only vLLM. By the way, the technique I mentioned to you is called paged attention. So this is a second one. vLLM works on top of Ray, where Ray helps when things get distributed, like for pipeline parallelism when you have to divide a model on multiple GPUs and then you need to feed the outputs. You divide it by layers, there are many ways. You do tensor parallelism, horizontally divide these kinds of models, and then you need to handle the resource allocations, the communication between these GPUs, and this is where Ray is the default engine to do that. Finally, the last one, Chatbot Arena. This is about LLM evaluation. This was an interesting project, very different from what I've done before. I'll tell you a little bit later what was the progression to get to this project. Obviously we know all benchmarks are very important. Patterson used to say, 'For better or worse, benchmarks shape a field.' We have ImageNet and all these very celebrated benchmarks in AI, which are used to put deep learning on the map. Remember that the fact that deep neural networks achieved performance on ImageNet exceeding the other classic solutions was the first time people really started to pay attention to deep neural networks, and then they matched human accuracy. The problem with language models is that they are extremely hard to evaluate, and the evaluation is expensive and unreliable. Let me give you an example. Prompt: 'Develop a Python program that reads text files under a directory and returns the top five words with the most number of occurrences in these files.' You ask the LLM this, and you get these answers. Which one is good, which one is better? It takes some time. You say, 'Let me try this one: Photosynthesis is a vital process for life on Earth. Could you outline the two main stages of photosynthesis, including where they take place within the chloroplast and the primary inputs and outputs for the stage?' Biology 101. And I'm not asking you to give the answer, I'm asking you to say which one of these two answers is better. Not easy. The other problem is they are unreliable, and that is because of data contamination. These large language models train on everything. One of the biggest problems today is running out of data, talking about synthetic data, and that's why each of these foundation labs pays a lot of money to get new datasets, proprietary, to improve their performance. This tweet from early on: 'I suspect GPT-4 performance is influenced by data contamination, at least on Codeforces. Codeforces's easiest problems: solve 10 out of 10 pre-2021 problems and zero out of 10 recent problems.' So solve everything before the cutoff date of the data used for training that model, and nothing after. Then there was this paper from MIT claiming GPT-4 can score 100% on MIT CS curriculum with the right prompting. That's pretty good. It turns out that is not the case. Later they had this disclaimer about zero contamination, and of course GPT-4 was trained on all of these kinds of exams. If you take that into account, the accuracy is still good but decreases to 58% from 100%. We need to be very careful. Let me tell you the story of how we got into this problem. In February 2023, Facebook released Llama open source. Then we did some fine-tuning of Llama using data called ShareGPT data. ShareGPT data was a site where you could share the inputs and outputs from your GPT conversation with GPT-4. It was supposed to be open. In the meantime, they changed the license, but at that time it was open, and it was only 70,000 conversations, not much, but very high quality because people only share the interesting conversations. We did that and called this model Vicuna. For a while it was the best open source model. But when we released it, how were we going to evaluate it? The typical evaluation you do with humans: you use a bunch of graduate students, buy them pizza, do some evaluation. However, one of the students on the team realized: why not use GPT-4 to evaluate? GPT-4 had been released only two weeks earlier. This was the genesis of using LLM as a judge. So we used that. For the Python program, which one is better? GPT-4 answers: 'Assistant A and B have similar answers, but A does not handle case sensitivity or punctuation.' Makes sense. For the photosynthesis one: 'Assistant A makes a mistake confusing the two stages as the input and output. He said the inputs from one stage are from the other stage.' Makes sense. So that was our use of LLM as a judge. But we still wanted to double-check with human evaluation as ground truth. After all, these answers are for humans, so let's see how they, the ultimate arbiters, do. But how do you do that? This is the result: Chatbot Arena. Ideally, if you want to evaluate, you have a question, get an answer from each LLM, and rank them. That would be ideal, but it turns out ranking n choices is very hard. Actually it's easier to pick best of five, but even that is hard. There's an interesting book: The Paradox of Choice. Having multiple choices is a lot of stress for humans. The easiest is to give only two choices to pick one. You cannot do less than that. So that is what we do. Even then, how do you organize this kind of comparison between two at a time? The complete and most accurate way is a tournament, like in sports, where every team plays every other team. The problem is this is n squared, and if a new model comes in, it's hard to add it. Fortunately, we have ratings from sports: Elo rating in chess, ATP ranking in tennis. Even when comparing different teams internationally, not all of them play each other, you have ratings. So we built that. We use an Elo-like rating. The system: you ask a question, you are presented with two answers from two anonymized and randomized language models, and you say A is better, B is better, or tie. The voting computes these ratings. The Arena score is a version of Elo score. The best is GPT-4-0125-preview, a bit lower, and the top open source is Llama 3.1. We have accumulated over 80 million prompts and over 2 million votes. The top one recently: all the rumors that OpenAI is testing GPT-4 o1, called Strawberry back then. People found out and came to test it. You don't need to vote for the main Chatbot Arena, only the blue things are votes, between 5 and 10%, used to generate the funnel for voting. People come because they can use state-of-the-art models for free. We have a lot of diversity in categories: coding, reasoning, problem solving, and it is large scale with different countries: English, Chinese, Russian. The results from humans are high quality because people try to solve their own problems, and they evaluate their own answers to their own questions, which provides higher quality labeling because they have the context. We also reconciled LLM as a judge results with human results. We did a systematic study and found that the limitations of LLM as a judge are similar to humans: position bias, verbosity bias, self-enhancement bias, and limited reasoning. However, the agreement between GPT-4 and human experts was 85%, and human to human was 81%, so it's in the same ballpark. One of the most interesting and satisfying things is that frontier labs use this to evaluate their models before releasing. We release 20% of the votes to avoid overfitting. We also have Arena Hard, which uses harder prompts to increase distance between strong and weak models. We have new evaluations like Vision Arena, and one for co-pilot with an integration in VS Code where you get two suggestions and pick which is better. There's also red teaming to fool the LLM, and RepoChat where you give your GitHub repository link and ask questions about your code. In summary, everything I showed is open source. For Chatbot Arena, we release 20% of the votes, but all the stack we used is open source. When serving models, we use vLLM on top of Ray for large models. I am still optimistic about open source despite the narrative. These three projects cover different layers of the stack. We are still at the very beginning and very excited about things to come. Thank you.
