这张图是作者用OpenAI DALL-E生成的

目前是2024年初，生成式AI市场领域主要由OpenAI主导。原因有很多——他们是第一个提供易于使用的大型语言模型API的公司，同时还提供的大型语言模型GPT-4是迄今为止最强大的。各种类型的工具开发者（如代理、个人助手、编码扩展）都选择了OpenAI来获取他们的大型语言模型需求。

虽然有多种理由让你用 OpenAI 的 GPT 来激发你的生成 AI 创作，但也有很多理由选择其他替代方案。有时可能不太经济，有时则因为你的数据隐私政策可能禁止你使用 OpenAI，或者你可能托管了一个开源的 LLM（或你自己开发的）。

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

所以，作为快速周末项目的部分，我决定实现一个兼容OpenAI API规范的Python FastAPI服务器，这样你可以将你喜爱的任何LLM（无论是像Anthropic的Claude那样的托管型，还是自托管型）包装起来，以像OpenAI API那样工作。好在，OpenAI API规范中有一个base_url参数，你可以将其设置为指向你自己的服务器，从而使客户端访问你的服务器而非OpenAI的服务器，并且大多数这些工具的开发者都允许你自定义这个参数。

为了做到这一点，我参照了OpenAI公开的聊天API文档（这里），并在vLLM的代码帮助下，vLLM是一个遵照Apache-2.0许可的大型语言模型推理服务器，同时也支持与OpenAI API的兼容性。

我们将构建一个模拟的API，模仿OpenAI的/v1/chat/completions接口的工作方式。虽然这个实现是用Python和FastAPI编写的，但我也尽量保持其简洁性，以便可以轻松地移植到其他现代编程语言，如TypeScript或Go。我们将用Python官方的OpenAI客户端库来进行测试——想法是如果能让库把我们的服务器当作OpenAI，那么任何使用该库的程序都会认为我们的服务器就是OpenAI。

我们将从实现非流式部分开始，先从建模请求开始。

从 typing 导入 List, Optional  

导入 pydantic 的 BaseModel  

类 ChatMessage(BaseModel):  
    role: str  
    content: str  

类 ChatCompletionRequest(BaseModel):  
    model: str = 'mock-gpt-model'  
    messages: List[ChatMessage]  
    max_tokens: Optional[int] = 512 # 最大 token 数量限制  
    temperature: Optional[float] = 0.1 # 温度参数  
    stream: Optional[bool] = False # 是否开启流式输出

PyDantic 模型用于表示客户端的请求，目的是模仿 API 参考。为了简洁起见，该模型并未实现所有规范，而是实现了最基本的必要部分。如果您缺少了例如 top_p 这样的参数，可以参考 API 规范，然后直接将其添加到模型中。

ChatCompletionRequest 描述了 OpenAI 在其请求中使用的参数。聊天 API 的规范要求指定一系列 ChatMessage（类似于聊天记录，客户端通常需要维护并将其反馈到每个请求中）。每个聊天消息具有一个 role 属性（通常是 system、assistant 或 user）和一个包含实际文本的 content 属性。

接下来，我们将编写 FastAPI 的聊天完成接口。

    import time  

    from fastapi import FastAPI  

    app = FastAPI(title="兼容OpenAI的API")  

    @app.post("/chat/completions")  
    async def chat_completions(request: ChatCompletionRequest):  

        if request.messages and request.messages[0].role == 'user':  
          resp_content = "作为一个模拟的AI助手，我只能回复您的最后一条消息：" + request.messages[-1].content  
        else:  
          resp_content = "作为一个模拟的AI助手，我只能回复您的最后一条消息，但没有收到任何消息！"  

        return {  
            "id": "1337",  
            "object": "chat.completion",  
            "created": time.time(),  
            "model": request.model,  
            "choices": [{  
                "message": {"role": "assistant", "content": resp_content}  
            }]  
        }

就这么简单。

假设这些代码都在一个名为 main.py 的文件里，我们将在选定的环境中安装两个 Python 库（最好是新建一个环境）：pip install fastapi openai，接着在终端里启动服务器。

    uvicorn main:app

或者在后台启动服务器，我们可以在另一个终端打开一个Python控制台环境，并复制并粘贴以下代码，这些代码直接取自OpenAI的Python客户端参考：

    from openai import OpenAI  

    # 初始化客户端并连接到本地服务器  
    client = OpenAI(  
        api_key="fake-api-key",  
        base_url="http://localhost:8000"  # 如果需要更改默认端口，可以在这里修改  
    )  

    # 调用API接口  
    chat_completion = client.chat.completions.create(  
        messages=[  
            {  
                "role": "user",  
                "content": "Say this is a test",  
            }  
        ],  
        model="gpt-1337-turbo-pro-max",  
    )  

    # 打印第一个'choice'  
    print(chat_completion.choices[0].message.content)

如果你的操作都正确，服务器的响应应该会正确显示。也值得检查一下 chat_completion 里的属性，确认所有相关属性是否与我们服务器上发送的一致。你应该会看到类似的内容：

作者写的代码段，用Carbon（https://carbon.now.sh/）美化了

升级：支持直播

由于LLM生成通常比较慢（计算资源消耗大），所以值得将生成的内容逐步传回客户端，这样用户可以在内容生成过程中逐步看到响应，而无需等待生成完成。还记得吗？我们在ChatCompletionRequest中添加了一个布尔值stream属性——这允许客户端请求将数据流式返回，而不是一次性发送。

这会让事情稍微复杂一点。我们将创建一个生成器（一种特殊的函数）来包装我们的模拟回应（在实际应用中，我们希望有一个能与我们的LLM生成对接的生成器）。

    import asyncio  
    import json  

    async def _resp_async_generator(text_resp: str):  
        # 假设每个单词都是一个 token，随着时间推移一个个返回它们。
        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"  
            # 等待1秒
            await asyncio.sleep(1)  
        yield "data: [DONE]\n\n"

现在，我们将调整我们原来的端点，在 stream==True 时返回一个 StreamingResponse。

    import time

    from starlette.responses import StreamingResponse

    app = FastAPI(title="兼容 OpenAI 的 API")

    async def chat_completions(request: ChatCompletionRequest):

        if request.messages:
            resp_content = "作为模拟的 AI 助手，我只能回应你的最后一条消息：" + request.messages[-1].content
        else:
            resp_content = "作为模拟的 AI 助手，我没有收到最后一条消息。"
        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 的库文档所示）。

    从openai导入OpenAI客户端  

    # 初始化客户端，并连接到本地服务器（http://localhost:8000）。如果需要更改默认端口，请修改此处  
    client = OpenAI(  
        api_key="fake-api-key",  
        base_url="http://localhost:8000"  # 如果需要更改默认端口，请修改此处  
    )  

    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 '', end='')

你应该看到服务器响应中的每个单词都像被逐个打印出来一样，慢慢显现出来，模仿逐字生成的效果。我们可以通过查看最后一个 chunk 对象来了解类似的内容。

作者写的代码，用Carbon美化

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

• 这里还有很多其他有趣的事情我们可以做，比如支持其他请求参数和其他OpenAI的抽象概念，比如函数调用和助手API。
• 大型语言模型API标准不统一使得公司和开发封装包的开发者更换供应商变得困难。在没有标准的情况下，我的做法是将大型语言模型抽象化，参照最大和最成熟的API规范。