OpenAI API 格式规范与实现指南

1
{
2
  "id": "chatcmpl-123456789",
3
  "object": "chat.completion",
4
  "created": 1677652288,
5
  "model": "gpt-3.5-turbo",
6
  "choices": [
7
    {
8
      "index": 0,
9
      "message": {
10
        "role": "assistant",
11
        "content": "Hello! I'm doing well, thank you for asking.",
12
        "reasoning": "用户问候，我应该礼貌回应"
13
      },
14
      "finish_reason": "stop"
15
    }
16
  ],
17
  "usage": {
18
    "prompt_tokens": 20,
19
    "completion_tokens": 15,
20
    "total_tokens": 35
21
  }
22
}

1
{
2
  "id": "chatcmpl-123456789",
3
  "object": "chat.completion.chunk",
4
  "created": 1677652288,
5
  "model": "gpt-3.5-turbo",
6
  "choices": [
7
    {
8
      "index": 0,
9
      "delta": {
10
        "role": "assistant",
11
        "content": "Hello",
12
        "reasoning": "推理内容（可选）"
13
      },
14
      "finish_reason": null
15
    }
16
  ]
17
}

1
from typing import List, Optional, Union, Dict, Any
2
from pydantic import BaseModel, Field
3
from enum import Enum
4

5
class Role(str, Enum):
6
    """消息角色枚举"""
7
    SYSTEM = "system"
8
    USER = "user"
9
    ASSISTANT = "assistant"
10
    FUNCTION = "function"
11

12
class Message(BaseModel):
13
    """聊天消息模型"""
14
    role: Role
15
    content: str
16
    name: Optional[str] = None
17
    reasoning: Optional[str] = None
18

19
class ChatCompletionRequest(BaseModel):
20
    """聊天完成请求模型"""
21
    model: str
22
    messages: List[Message]
23
    temperature: Optional[float] = Field(default=1.0, ge=0.0, le=2.0)
24
    top_p: Optional[float] = Field(default=1.0, ge=0.0, le=1.0)
25
    n: Optional[int] = Field(default=1, ge=1)
26
    stream: Optional[bool] = False
27
    stop: Optional[Union[str, List[str]]] = None
28
    max_tokens: Optional[int] = Field(default=None, ge=1)
29
    presence_penalty: Optional[float] = Field(default=0.0, ge=-2.0, le=2.0)
30
    frequency_penalty: Optional[float] = Field(default=0.0, ge=-2.0, le=2.0)
31
    user: Optional[str] = None
32

33
class Usage(BaseModel):
34
    """使用情况统计"""
35
    prompt_tokens: int
36
    completion_tokens: int
37
    total_tokens: int
38

39
class Choice(BaseModel):
40
    """选择项模型"""
41
    index: int
42
    message: Message
43
    finish_reason: Optional[str] = None
44

45
class ChatCompletionResponse(BaseModel):
46
    """聊天完成响应模型"""
47
    id: str
48
    object: str = "chat.completion"
49
    created: int
50
    model: str
51
    choices: List[Choice]
52
    usage: Usage
53

54
class Delta(BaseModel):
55
    """流式响应增量"""
56
    role: Optional[str] = None
57
    content: Optional[str] = None
58
    reasoning: Optional[str] = None
59

60
class StreamChoice(BaseModel):
61
    """流式选择项"""
62
    index: int
63
    delta: Delta
64
    finish_reason: Optional[str] = None
65

66
class ChatCompletionStreamResponse(BaseModel):
67
    """流式聊天完成响应"""
68
    id: str
69
    object: str = "chat.completion.chunk"
70
    created: int
71
    model: str
72
    choices: List[StreamChoice]

1
from fastapi import FastAPI, HTTPException, Request
2
from fastapi.middleware.cors import CORSMiddleware
3
from fastapi.responses import StreamingResponse
4
import logging
5
import time
6
from typing import AsyncGenerator
7

8
from models import (
9
    ChatCompletionRequest,
10
    ChatCompletionResponse,
11
    ChatCompletionStreamResponse,
12
    StreamChoice,
13
    Delta
14
)
15

16
app = FastAPI(
17
    title="OpenAI Compatible API",
18
    description="OpenAI 格式兼容的 API 服务",
19
    version="1.0.0"
20
)
21

22
# CORS 中间件
23
app.add_middleware(
24
    CORSMiddleware,
25
    allow_origins=["*"],
26
    allow_credentials=True,
27
    allow_methods=["*"],
28
    allow_headers=["*"],
29
)
30

31
@app.get("/")
32
async def root():
33
    """根路径"""
34
    return {
35
        "message": "OpenAI Compatible API Server",
36
        "version": "1.0.0",
37
        "endpoints": {
38
            "chat": "/v1/chat/completions",
39
            "models": "/v1/models",
40
            "health": "/health"
41
        }
42
    }
43

44
@app.get("/health")
45
async def health_check():
46
    """健康检查"""
47
    return {"status": "healthy", "timestamp": int(time.time())}
48

49
@app.post("/v1/chat/completions")
50
async def create_chat_completion(request: ChatCompletionRequest, http_request: Request):
51
    """创建聊天完成"""
52
    # 获取 API Key
53
    auth_header = http_request.headers.get("Authorization")
54
    if not auth_header or not auth_header.startswith("Bearer "):
55
        raise HTTPException(
56
            status_code=401,
57
            detail={
58
                "error": {
59
                    "message": "Invalid API key provided",
60
                    "type": "invalid_request_error",
61
                    "code": "invalid_api_key"
62
                }
63
            }
64
        )
65

66
    api_key = auth_header[7:]  # 移除 "Bearer " 前缀
67

68
    try:
69
        if request.stream:
70
            # 流式响应
71
            return StreamingResponse(
72
                generate_stream_response(request, api_key),
73
                media_type="text/event-stream",
74
                headers={
75
                    "Cache-Control": "no-cache",
76
                    "Connection": "keep-alive",
77
                    "Content-Type": "text/event-stream"
78
                }
79
            )
80
        else:
81
            # 非流式响应
82
            response = await generate_completion(request, api_key)
83
            return response
84

85
    except Exception as e:
86
        raise HTTPException(
87
            status_code=500,
88
            detail={
89
                "error": {
90
                    "message": str(e),
91
                    "type": "server_error",
92
                    "code": "internal_error"
93
                }
94
            }
95
        )
96

97
async def generate_completion(request: ChatCompletionRequest, api_key: str) -> ChatCompletionResponse:
98
    """生成非流式完成响应"""
99
    # 这里实现你的模型调用逻辑
100
    # 示例实现
101
    response_id = f"chatcmpl-{int(time.time())}"
102

103
    # 模拟响应
104
    return ChatCompletionResponse(
105
        id=response_id,
106
        created=int(time.time()),
107
        model=request.model,
108
        choices=[
109
            Choice(
110
                index=0,
111
                message=Message(
112
                    role="assistant",
113
                    content="这是一个示例响应"
114
                ),
115
                finish_reason="stop"
116
            )
117
        ],
118
        usage=Usage(
119
            prompt_tokens=10,
120
            completion_tokens=5,
121
            total_tokens=15
122
        )
123
    )
124

125
async def generate_stream_response(request: ChatCompletionRequest, api_key: str) -> AsyncGenerator[str, None]:
126
    """生成流式响应"""
127
    response_id = f"chatcmpl-{int(time.time())}"
128
    created = int(time.time())
129

130
    try:
131
        # 模拟流式输出
132
        content_chunks = ["这是", "一个", "流式", "响应", "示例"]
133

134
        for chunk in content_chunks:
135
            stream_response = ChatCompletionStreamResponse(
136
                id=response_id,
137
                created=created,
138
                model=request.model,
139
                choices=[
140
                    StreamChoice(
141
                        index=0,
142
                        delta=Delta(content=chunk),
143
                        finish_reason=None
144
                    )
145
                ]
146
            )
147
            yield f"data: {stream_response.model_dump_json()}\n\n"
148

149
            # 模拟延迟
150
            import asyncio
151
            await asyncio.sleep(0.1)
152

153
        # 发送结束标记
154
        final_response = ChatCompletionStreamResponse(
155
            id=response_id,
156
            created=created,
157
            model=request.model,
158
            choices=[
159
                StreamChoice(
160
                    index=0,
161
                    delta=Delta(),
162
                    finish_reason="stop"
163
                )
164
            ]
165
        )
166
        yield f"data: {final_response.model_dump_json()}\n\n"
167
        yield "data: [DONE]\n\n"
168

169
    except Exception as e:
170
        # 错误处理
171
        error_response = {
172
            "error": {
173
                "message": str(e),
174
                "type": "server_error",
175
                "code": "internal_error"
176
            }
177
        }
178
        yield f"data: {json.dumps(error_response)}\n\n"
179
        yield "data: [DONE]\n\n"
180

181
@app.get("/v1/models")
182
async def list_models():
183
    """获取模型列表"""
184
    return {
185
        "object": "list",
186
        "data": [
187
            {
188
                "id": "gpt-3.5-turbo",
189
                "object": "model",
190
                "created": 1677610602,
191
                "owned_by": "openai"
192
            },
193
            {
194
                "id": "gpt-4",
195
                "object": "model",
196
                "created": 1687882411,
197
                "owned_by": "openai"
198
            }
199
        ]
200
    }

1
import os
2
from typing import List
3
from dotenv import load_dotenv
4

5
load_dotenv()
6

7
class Config:
8
    """应用配置类"""
9

10
    def __init__(self):
11
        # 服务器配置
12
        self.host: str = os.getenv("HOST", "0.0.0.0")
13
        self.port: int = int(os.getenv("PORT", "8000"))
14
        self.debug: bool = os.getenv("DEBUG", "false").lower() == "true"
15

16
        # API 配置
17
        self.api_prefix: str = os.getenv("API_PREFIX", "/v1")
18
        self.cors_origins: List[str] = os.getenv("CORS_ORIGINS", "*").split(",")
19

20
        # 模型配置
21
        self.default_model: str = os.getenv("DEFAULT_MODEL", "gpt-3.5-turbo")
22
        self.max_tokens: int = int(os.getenv("MAX_TOKENS", "2048"))
23

24
        # 安全配置
25
        self.api_keys: List[str] = os.getenv("API_KEYS", "").split(",")
26

27
    def validate_api_key(self, api_key: str) -> bool:
28
        """验证 API Key"""
29
        if not self.api_keys or not self.api_keys[0]:
30
            return True  # 如果没有配置 API Key，则不验证
31
        return api_key in self.api_keys
32

33
config = Config()

1
from fastapi import HTTPException, Depends
2
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
3

4
security = HTTPBearer()
5

6
async def verify_api_key(credentials: HTTPAuthorizationCredentials = Depends(security)):
7
    """验证 API Key"""
8
    if not config.validate_api_key(credentials.credentials):
9
        raise HTTPException(
10
            status_code=401,
11
            detail={
12
                "error": {
13
                    "message": "Invalid API key provided",
14
                    "type": "invalid_request_error",
15
                    "code": "invalid_api_key"
16
                }
17
            }
18
        )
19
    return credentials.credentials
20

21
# 在路由中使用
22
@app.post("/v1/chat/completions")
23
async def create_chat_completion(
24
    request: ChatCompletionRequest,
25
    api_key: str = Depends(verify_api_key)
26
):
27
    # 处理请求
28
    pass

1
# 统一错误处理中间件
2
@app.middleware("http")
3
async def error_handling_middleware(request: Request, call_next):
4
    try:
5
        response = await call_next(request)
6
        return response
7
    except Exception as e:
8
        return JSONResponse(
9
            status_code=500,
10
            content={
11
                "error": {
12
                    "message": "Internal server error",
13
                    "type": "server_error",
14
                    "code": "internal_error"
15
                }
16
            }
17
        )

1
import logging
2
from datetime import datetime
3

4
# 配置日志
5
logging.basicConfig(
6
    level=logging.INFO,
7
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
8
)
9
logger = logging.getLogger(__name__)
10

11
# 请求日志中间件
12
@app.middleware("http")
13
async def log_requests(request: Request, call_next):
14
    start_time = datetime.now()
15

16
    # 记录请求
17
    logger.info(f"Request: {request.method} {request.url}")
18

19
    response = await call_next(request)
20

21
    # 记录响应
22
    process_time = (datetime.now() - start_time).total_seconds()
23
    logger.info(f"Response: {response.status_code} - {process_time:.3f}s")
24

25
    return response

1
from pydantic import validator
2

3
class ChatCompletionRequest(BaseModel):
4
    model: str
5
    messages: List[Message]
6
    temperature: Optional[float] = 1.0
7

8
    @validator('temperature')
9
    def validate_temperature(cls, v):
10
        if v < 0.0 or v > 2.0:
11
            raise ValueError('temperature must be between 0.0 and 2.0')
12
        return v
13

14
    @validator('messages')
15
    def validate_messages(cls, v):
16
        if not v:
17
            raise ValueError('messages cannot be empty')
18
        return v

1
# 异步处理
2
import asyncio
3
from concurrent.futures import ThreadPoolExecutor
4

5
executor = ThreadPoolExecutor(max_workers=4)
6

7
async def process_model_request(request_data):
8
    """异步处理模型请求"""
9
    loop = asyncio.get_event_loop()
10
    result = await loop.run_in_executor(
11
        executor,
12
        sync_model_call,
13
        request_data
14
    )
15
    return result
16

17
# 连接池管理
18
from httpx import AsyncClient
19

20
class ModelClient:
21
    def __init__(self):
22
        self.client = AsyncClient(
23
            timeout=30.0,
24
            limits=httpx.Limits(max_connections=100, max_keepalive_connections=20)
25
        )
26

27
    async def call_model(self, data):
28
        response = await self.client.post(
29
            "https://api.model-provider.com/v1/completions",
30
            json=data
31
        )
32
        return response.json()

1
import requests
2
import json
3

4
# 非流式请求
5
def test_completion():
6
    url = "http://localhost:8000/v1/chat/completions"
7
    headers = {
8
        "Content-Type": "application/json",
9
        "Authorization": "Bearer your-api-key"
10
    }
11
    data = {
12
        "model": "gpt-3.5-turbo",
13
        "messages": [
14
            {"role": "user", "content": "Hello, how are you?"}
15
        ],
16
        "temperature": 0.7,
17
        "max_tokens": 100
18
    }
19

20
    response = requests.post(url, headers=headers, json=data)
21
    print(response.json())
22

23
# 流式请求
24
def test_stream():
25
    url = "http://localhost:8000/v1/chat/completions"
26
    headers = {
27
        "Content-Type": "application/json",
28
        "Authorization": "Bearer your-api-key"
29
    }
30
    data = {
31
        "model": "gpt-3.5-turbo",
32
        "messages": [
33
            {"role": "user", "content": "Tell me a story"}
34
        ],
35
        "stream": True
36
    }
37

38
    response = requests.post(url, headers=headers, json=data, stream=True)
39

40
    for line in response.iter_lines():
41
        if line:
42
            line = line.decode('utf-8')
43
            if line.startswith('data: '):
44
                data_content = line[6:]
45
                if data_content == '[DONE]':
46
                    break
47
                try:
48
                    chunk = json.loads(data_content)
49
                    content = chunk['choices'][0]['delta'].get('content', '')
50
                    if content:
51
                        print(content, end='', flush=True)
52
                except json.JSONDecodeError:
53
                    pass
54
    print()  # 换行
55

56
if __name__ == "__main__":
57
    test_completion()
58
    test_stream()

1
// 非流式请求
2
async function testCompletion() {
3
    const response = await fetch('http://localhost:8000/v1/chat/completions', {
4
        method: 'POST',
5
        headers: {
6
            'Content-Type': 'application/json',
7
            'Authorization': 'Bearer your-api-key'
8
        },
9
        body: JSON.stringify({
10
            model: 'gpt-3.5-turbo',
11
            messages: [
12
                { role: 'user', content: 'Hello, how are you?' }
13
            ],
14
            temperature: 0.7,
15
            max_tokens: 100
16
        })
17
    });
18

19
    const data = await response.json();
20
    console.log(data);
21
}
22

23
// 流式请求
24
async function testStream() {
25
    const response = await fetch('http://localhost:8000/v1/chat/completions', {
26
        method: 'POST',
27
        headers: {
28
            'Content-Type': 'application/json',
29
            'Authorization': 'Bearer your-api-key'
30
        },
31
        body: JSON.stringify({
32
            model: 'gpt-3.5-turbo',
33
            messages: [
34
                { role: 'user', content: 'Tell me a story' }
35
            ],
36
            stream: true
37
        })
38
    });
39

40
    const reader = response.body.getReader();
41
    const decoder = new TextDecoder();
42

43
    while (true) {
44
        const { done, value } = await reader.read();
45
        if (done) break;
46

47
        const chunk = decoder.decode(value);
48
        const lines = chunk.split('\n');
49

50
        for (const line of lines) {
51
            if (line.startsWith('data: ')) {
52
                const data = line.slice(6);
53
                if (data === '[DONE]') {
54
                    return;
55
                }
56

57
                try {
58
                    const parsed = JSON.parse(data);
59
                    const content = parsed.choices[0]?.delta?.content;
60
                    if (content) {
61
                        process.stdout.write(content);
62
                    }
63
                } catch (e) {
64
                    // 忽略解析错误
65
                }
66
            }
67
        }
68
    }
69
}