现在是2024年初，Gen AI市场正被OpenAI所主导。这并非没有原因的——他们拥有先行者优势，是第一个为大型语言模型（LLM）提供易于使用的API的公司，而且他们也提供了目前可能是最强大的LLM，即GPT 4。鉴于这种情况，各种工具（如代理、个人助手、编程扩展）的开发者都转向OpenAI来满足他们对LLM的需求。

尽管有很多理由支持使用OpenAI的GPT来推动你的Gen AI创作，但同样有很多理由选择其他替代方案。有时，使用OpenAI可能成本效益不高；有时，你的数据隐私政策可能禁止使用OpenAI；或者你可能正在托管一个开源的LLM（或你自己的LLM）。

OpenAI的市场主导地位意味着，你可能想要使用的许多工具都只支持OpenAI的API。像OpenAI、Anthropic和Google这样的Gen AI和LLM提供商似乎都在创建不同的API架构（可能是故意的），这给想要支持所有这些架构的开发者增加了大量额外工作。

因此，作为一个周末快速项目，我决定实现一个与OpenAI API规范兼容的Python FastAPI服务器，以便你可以包装任何你喜欢的LLM（无论是像Anthropic的Claude这样的托管LLM，还是自托管的LLM），来模拟OpenAI的API。幸运的是，OpenAI的API规范中有一个参数，你可以设置它，以便客户端指向你的服务器，而不是OpenAI的服务器，而且大多数上述工具的开发者都允许你根据自己的喜好设置这个参数。

为了完成这个任务，我遵循了OpenAI公开提供的Chat API参考，并在vLLM（一个为LLM提供OpenAPI兼容性的Apache-2.0许可的推理服务器）的代码帮助下进行了操作。

游戏计划

我们将构建一个模拟API，模仿OpenAI的Chat Completion API（聊天完成API）的工作方式。虽然这个实现是在Python中使用FastAPI完成的，但我尽量保持其简单，以便可以轻松地转移到其他现代编程语言，如TypeScript或Go。我们将使用Python官方的OpenAI客户端库进行测试——如果我们可以让库认为我们的服务器是OpenAI，那么任何使用它的程序也会这么认为。/v1/chat/completions

第一步——聊天完成API，无流处理

我们将从实现非流处理部分开始。首先，我们来建模我们的请求：

from typing import List, Optional

from 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

PyDantic模型代表客户端的请求，旨在复制API参考。为了简洁起见，此模型没有实现整个规范，而是实现了使其工作所需的基本框架。如果你发现缺少API规范中的某个参数（如top_p），你可以简单地将其添加到模型中。

模型定义了OpenAI在其请求中使用的参数。聊天API规范要求指定一个消息列表（如聊天历史记录，客户端通常负责维护并在每次请求时反馈）。每个聊天消息都有一个role属性（通常是user、system、assistant）和一个content属性，包含实际的消息文本。

接下来，我们将编写FastAPI的聊天完成端点：

import time

from 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)
        }]
    }

就这么简单。

测试我们的实现

假设两个代码块都在一个文件中，我们将在所选环境中安装两个Python库（最好创建一个新的环境）：和，然后从终端启动服务器：main.pypip install fastapi openai

uvicorn main:app

在另一个终端（或通过将服务器在后台启动）中，我们将打开一个Python控制台，并复制粘贴以下直接从OpenAI的Python客户端参考中取得的代码：

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)

如果你已经正确完成了所有步骤，服务器的响应应该会正确打印出来。同时，也值得检查返回的对象，以确保所有相关属性都是从我们的服务器发送的。你应该会看到类似这样的内容：chat_completion

升级——支持流处理

由于LLM（大型语言模型）的生成往往较慢（计算成本高），因此将生成的内容流回客户端是有价值的，这样用户就可以在内容生成时看到响应，而无需等待其完成。如果你还记得的话，我们给了一个布尔属性——这允许客户端请求数据被流回，而不是一次性发送。

这会让事情变得稍微复杂一些。我们将创建一个生成器函数来包装我们的模拟响应（在真实场景中，我们将需要一个与我们的LLM生成相连接的生成器）

import asyncio
import json

async 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"

现在，我们将修改原始端点，在以下情况下返回流式响应：stream==True

import time

from 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)        }]
    }

测试流处理实现

在重新启动uvicorn服务器后，我们将打开一个Python控制台并输入以下代码（同样来自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 "")

你应该会看到服务器响应中的每个单词慢慢打印出来，模仿令牌生成。我们可以检查最后一个对象，它看起来应该像这样：

整合所有内容

最后，在下面的要点中，你可以看到服务器的完整代码。

最后说明

这里还有许多其他有趣的事情可以做，比如支持其他请求参数，以及OpenAI的其他抽象概念，如函数调用和助理API。

LLM（大型语言模型）API缺乏标准化，使得无论是公司还是LLM包装包的开发者，都很难切换提供商。在没有任何标准的情况下，我在这里采用的方法是，将LLM抽象化，隐藏在最大、最成熟的API规范之后。