BCAI/easyai-ai-gateway
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

docs(api): 同步 /api/v1/chat/completions 的 OpenAPI 与同步响应

Browse Source 
补充 Chat Completions 的兼容响应模型与路由注释，确保 /api/v1/chat/completions 按同步兼容格式返回并更新对应测试与 Swagger 文档。
This commit is contained in:
chensipeng
2026-05-16 00:19:39 +08:00
parent 34c3251c6d
commit ae197a742f
7 changed files with 491 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/internal/httpapi/handlers.go
+117 -64
View File
@@ -13,6 +13,7 @@ import (
"github.com/easyai/easyai-ai-gateway/apps/api/internal/auth"
"github.com/easyai/easyai-ai-gateway/apps/api/internal/clients"
"github.com/easyai/easyai-ai-gateway/apps/api/internal/netproxy"
"github.com/easyai/easyai-ai-gateway/apps/api/internal/runner"
"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
)
@@ -858,7 +859,7 @@ func (s *Server) listModelRateLimitStatuses(w http.ResponseWriter, r *http.Reque
// createTask godoc
// @Summary 创建或执行 AI 任务
// @Description 网关任务接口按 model 选择平台模型；/api/v1 路径返回任务受理结果，OpenAI-compatible 路径同步返回兼容响应或 SSE 流。
// @Description 网关任务接口按 model 选择平台模型；除 /api/v1/chat/completions 以外的 /api/v1 任务路径返回任务受理结果，OpenAI-compatible 路径同步返回兼容响应或 SSE 流。
// @Tags tasks
// @Accept json
// @Produce json
@@ -874,7 +875,6 @@ func (s *Server) listModelRateLimitStatuses(w http.ResponseWriter, r *http.Reque
// @Failure 404 {object} ErrorEnvelope
// @Failure 429 {object} ErrorEnvelope
// @Failure 502 {object} ErrorEnvelope
// @Router /api/v1/chat/completions [post]
// @Router /api/v1/responses [post]
// @Router /api/v1/images/generations [post]
// @Router /api/v1/images/edits [post]
@@ -909,13 +909,13 @@ func (s *Server) createTask(kind string, compatible bool) http.Handler {
writeError(w, http.StatusForbidden, "api key scope does not allow this capability")
return
}
asyncMode := asyncRequest(r)
responsePlan := planTaskResponse(kind, compatible, body, r)
task, err := s.store.CreateTask(r.Context(), store.CreateTaskInput{
Kind: kind,
Model: model,
RunMode: runModeFromRequest(body),
Async: asyncMode,
Async: responsePlan.asyncMode,
Request: body,
}, user)
if err != nil {
@@ -923,7 +923,7 @@ func (s *Server) createTask(kind string, compatible bool) http.Handler {
writeError(w, http.StatusInternalServerError, "create task failed")
return
}
if asyncMode {
if responsePlan.asyncMode {
if err := s.runner.EnqueueAsyncTask(r.Context(), task); err != nil {
writeError(w, http.StatusInternalServerError, err.Error(), "enqueue_failed")
return
@@ -933,65 +933,8 @@ func (s *Server) createTask(kind string, compatible bool) http.Handler {
}
runCtx, cancelRun := s.requestExecutionContext(r)
defer cancelRun()
if compatible {
if boolValue(body, "stream") {
flusher := prepareCompatibleStream(w)
result, runErr := s.runner.ExecuteStream(runCtx, task, user, func(delta string) error {
if !requestStillConnected(r) {
return nil
}
writeCompatibleDelta(w, kind, model, delta)
if flusher != nil {
flusher.Flush()
}
return nil
})
if runErr != nil {
if !requestStillConnected(r) {
return
}
status := statusFromRunError(runErr)
errorPayload := map[string]any{
"code": runErrorCode(runErr),
"message": runErrorMessage(runErr),
"status": status,
}
if result.Task.ID != "" {
errorPayload["taskId"] = result.Task.ID
}
if result.Task.RequestID != "" {
errorPayload["requestId"] = result.Task.RequestID
}
for key, value := range runErrorDetails(runErr) {
errorPayload[key] = value
}
sendSSE(w, "error", map[string]any{"error": errorPayload})
if flusher != nil {
flusher.Flush()
}
return
}
if !requestStillConnected(r) {
return
}
writeCompatibleDone(w, kind, model, result.Output)
if flusher != nil {
flusher.Flush()
}
return
}
result, runErr := s.runner.Execute(runCtx, task, user)
if runErr != nil {
if !requestStillConnected(r) {
return
}
writeErrorWithDetails(w, statusFromRunError(runErr), runErrorMessage(runErr), runErrorDetails(runErr), runErrorCode(runErr))
return
}
if !requestStillConnected(r) {
return
}
writeJSON(w, http.StatusOK, result.Output)
if responsePlan.compatibleMode {
writeCompatibleTaskResponse(runCtx, w, r, s.runner, kind, model, task, user, responsePlan.streamMode)
return
}
result, runErr := s.runner.Execute(runCtx, task, user)
@@ -1006,6 +949,29 @@ func (s *Server) createTask(kind string, compatible bool) http.Handler {
})
}
// createAPIV1ChatCompletions godoc
// @Summary 创建 Chat Completions
// @Description /api/v1/chat/completions 同步执行：stream=true 返回 text/event-stream SSE；stream=false 或未传返回兼容 JSON；该接口忽略 X-Async。
// @Tags tasks
// @Accept json
// @Produce json
// @Produce text/event-stream
// @Security BearerAuth
// @Param X-Async header bool false "该接口忽略此参数"
// @Param input body TaskRequest true "Chat Completions 请求"
// @Success 200 {object} ChatCompletionCompatibleResponse
// @Failure 400 {object} ErrorEnvelope
// @Failure 401 {object} ErrorEnvelope
// @Failure 402 {object} ErrorEnvelope
// @Failure 403 {object} ErrorEnvelope
// @Failure 404 {object} ErrorEnvelope
// @Failure 429 {object} ErrorEnvelope
// @Failure 502 {object} ErrorEnvelope
// @Router /api/v1/chat/completions [post]
func (s *Server) createAPIV1ChatCompletions() http.Handler {
return s.createTask("chat.completions", false)
}
func (s *Server) requestExecutionContext(r *http.Request) (context.Context, context.CancelFunc) {
base := context.WithoutCancel(r.Context())
if s.ctx == nil {
@@ -1031,11 +997,98 @@ func requestStillConnected(r *http.Request) bool {
}
}
type taskExecutor interface {
Execute(context.Context, store.GatewayTask, *auth.User) (runner.Result, error)
ExecuteStream(context.Context, store.GatewayTask, *auth.User, clients.StreamDelta) (runner.Result, error)
}
func writeCompatibleTaskResponse(runCtx context.Context, w http.ResponseWriter, r *http.Request, executor taskExecutor, kind string, model string, task store.GatewayTask, user *auth.User, streamMode bool) {
if streamMode {
flusher := prepareCompatibleStream(w)
result, runErr := executor.ExecuteStream(runCtx, task, user, func(delta string) error {
if !requestStillConnected(r) {
return nil
}
writeCompatibleDelta(w, kind, model, delta)
if flusher != nil {
flusher.Flush()
}
return nil
})
if runErr != nil {
if !requestStillConnected(r) {
return
}
status := statusFromRunError(runErr)
errorPayload := map[string]any{
"code": runErrorCode(runErr),
"message": runErrorMessage(runErr),
"status": status,
}
if result.Task.ID != "" {
errorPayload["taskId"] = result.Task.ID
}
if result.Task.RequestID != "" {
errorPayload["requestId"] = result.Task.RequestID
}
for key, value := range runErrorDetails(runErr) {
errorPayload[key] = value
}
sendSSE(w, "error", map[string]any{"error": errorPayload})
if flusher != nil {
flusher.Flush()
}
return
}
if !requestStillConnected(r) {
return
}
writeCompatibleDone(w, kind, model, result.Output)
if flusher != nil {
flusher.Flush()
}
return
}
result, runErr := executor.Execute(runCtx, task, user)
if runErr != nil {
if !requestStillConnected(r) {
return
}
writeErrorWithDetails(w, statusFromRunError(runErr), runErrorMessage(runErr), runErrorDetails(runErr), runErrorCode(runErr))
return
}
if !requestStillConnected(r) {
return
}
writeJSON(w, http.StatusOK, result.Output)
}
func asyncRequest(r *http.Request) bool {
value := strings.TrimSpace(strings.ToLower(r.Header.Get("x-async")))
return value == "1" || value == "true" || value == "yes" || value == "on"
}
type taskResponsePlan struct {
asyncMode bool
compatibleMode bool
streamMode bool
}
func planTaskResponse(kind string, compatible bool, body map[string]any, r *http.Request) taskResponsePlan {
asyncMode := asyncRequest(r)
compatibleMode := compatible
if kind == "chat.completions" && !compatible {
asyncMode = false
compatibleMode = true
}
return taskResponsePlan{
asyncMode: asyncMode,
compatibleMode: compatibleMode,
streamMode: boolValue(body, "stream"),
}
}
func writeTaskAccepted(w http.ResponseWriter, task store.GatewayTask) {
writeJSON(w, http.StatusAccepted, map[string]any{
"taskId": task.ID,