Use external APIs

LitServe provides async based concurrency for i/o bound tasks such as external API call, disk operation, etc.

There are several situations where we need to call external APIs, such as ChatGPT, other LLM providers, or services that fetch data from different APIs. These are I/O-bound operations, where the CPU is not actively engaged and remains idle, resulting in wasted resources. Python provides an asynchronous programming mechanism that allows us to run concurrent tasks efficiently. This ensures that the CPU can continue performing meaningful work while other tasks are waiting for I/O operations to complete.

Async external API

Let's say you need to use the OpenAI Chat Completion API service in a LitServe API. Since this is a network call, we can use async to enable concurrency.

The following example shows how to use the Async OpenAI Python client and can be modified for use with any other async-compatible network call.

import litserve as ls
import asyncio
import os
from openai import AsyncOpenAI

class OpenAICompletionAPI(ls.LitAPI):
    def setup(self, device):
        self.client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))

    async def decode_request(self, request: dict):
        return request["prompt"]

    async def predict(self, prompt: str):
        try:
            chat_completion = await self.client.chat.completions.create(
                messages=[
                    {"role": "system", "content": "You are a helpful assistant."},
                    {"role": "user", "content": prompt},
                ],
                model="gpt-4",
                max_tokens=100,
                temperature=0.7,
            )
            generated_text = chat_completion.choices[0].message.content
            return generated_text
        except Exception as e:
            raise RuntimeError(f"OpenAI API error: {e}")

    async def encode_response(self, completion_text: str):
        return {"generated_text": completion_text, "source": "openai"}

if __name__ == "__main__":
    api = OpenAICompletionAPI(enable_async=True)
    server = ls.LitServer(api)
    server.run(port=8000)

Was this page helpful?