diff --git a/common/util/streaming.go b/common/util/streaming.go index 7cb3707..0ede6dd 100644 --- a/common/util/streaming.go +++ b/common/util/streaming.go @@ -1,40 +1,128 @@ package util import ( + "context" "encoding/base64" "encoding/json" "fmt" + "model-gateway/service/gateway" "sort" "strings" "github.com/gogf/gf/v2/encoding/gjson" ) -// ================================================================ - -// ParseStreamResponse 流式响应解析（通用入口） -func ParseStreamResponse(rawBytes []byte, streamConfig map[string]any) (map[string]any, error) { +// ParseStreamResponse 流式响应解析 +func ParseStreamResponse(ctx context.Context, rawBytes []byte, streamConfig map[string]any) (map[string]any, error) { enabled, _ := streamConfig["enabled"].(bool) if !enabled { return gjson.New(string(rawBytes)).Map(), nil } - parser, _ := streamConfig["parser"].(string) - if parser == "base64_concat" { - return parseBase64Stream(rawBytes) + outputType, _ := streamConfig["output_type"].(string) + streamToClient, _ := streamConfig["stream_to_client"].(bool) + events, _ := streamConfig["events"].([]any) + + if len(events) == 0 { + return gjson.New(string(rawBytes)).Map(), nil } - return parseSSEStream(rawBytes, streamConfig) + // 如果业务需要流式返回，直接透传原始数据 + if streamToClient { + return map[string]any{"stream_data": rawBytes}, nil + } + + result := make(map[string]any) + lines := strings.Split(string(rawBytes), "\n") + + for _, evt := range events { + e, _ := evt.(map[string]any) + evtType, _ := e["type"].(string) + + switch evtType { + case "concat": + processConcat(lines, e, result) + case "base64_concat": + processBase64Concat(lines, e, result) + case "collect": + processCollect(lines, e, result) + case "final": + processFinal(lines, e, result) + } + } + + switch outputType { + case "audio": + if audioBytes, ok := result["audio"].([]byte); ok { + oss, err := gateway.UploadByTask(ctx, audioBytes, "mp3") + if err != nil { + return nil, err + } + result["content"] = oss.FileAddressPrefix + oss.FileURL + } + case "text": + if v, ok := result["content"]; ok { + return map[string]any{"content": v, "usage": result["usage"]}, nil + } + case "image": + if v, ok := result["urls"]; ok { + return map[string]any{"content": v, "usage": result["usage"]}, nil + } + } + + return result, nil } -// parseBase64Stream 拼接流式 base64 并解码为二进制（TTS 等音频模型） -func parseBase64Stream(rawBytes []byte) (map[string]any, error) { - lines := strings.Split(string(rawBytes), "\n") - var audioBase64 strings.Builder +// processConcat 文本拼接 +func processConcat(lines []string, event map[string]any, result map[string]any) { + match, _ := event["match"].(string) + aggregateTo, _ := event["aggregate_to"].(string) + fields, _ := event["fields"].(map[string]any) + var parts []string for _, line := range lines { line = strings.TrimSpace(line) - if line == "" { + if line == "" || line == "[DONE]" { + continue + } + if strings.HasPrefix(line, "event:") { + continue + } + if strings.HasPrefix(line, "data:") { + line = strings.TrimPrefix(line, "data:") + line = strings.TrimSpace(line) + } + + var chunk map[string]any + if err := json.Unmarshal([]byte(line), &chunk); err != nil { + continue + } + + chunkType, _ := chunk["type"].(string) + if match != "" && !strings.Contains(chunkType, match) { + continue + } + + for _, chunkPath := range fields { + val := gjson.New(chunk).Get(chunkPath.(string)).String() + if val != "" { + parts = append(parts, val) + } + } + } + + result[aggregateTo] = strings.Join(parts, "") +} + +// processBase64Concat base64 拼接 +func processBase64Concat(lines []string, event map[string]any, result map[string]any) { + aggregateTo, _ := event["aggregate_to"].(string) + fields, _ := event["fields"].(map[string]any) + + var builder strings.Builder + for _, line := range lines { + line = strings.TrimSpace(line) + if line == "" || line == "[DONE]" { continue } @@ -43,8 +131,10 @@ func parseBase64Stream(rawBytes []byte) (map[string]any, error) { continue } - if data, ok := chunk["data"].(string); ok && data != "" { - audioBase64.WriteString(data) + for _, chunkPath := range fields { + if data := gjson.New(chunk).Get(chunkPath.(string)).String(); data != "" { + builder.WriteString(data) + } } } @@ -53,29 +143,73 @@ func parseBase64Stream(rawBytes []byte) (map[string]any, error) { return -1 } return r - }, audioBase64.String()) + }, builder.String()) audioBytes, err := base64.StdEncoding.DecodeString(cleanBase64) if err != nil { - audioBytes, err = base64.RawStdEncoding.DecodeString(cleanBase64) - if err != nil { - return nil, fmt.Errorf("base64 解码失败: %w", err) - } + audioBytes, _ = base64.RawStdEncoding.DecodeString(cleanBase64) } - - return map[string]any{"audio": audioBytes}, nil + result[aggregateTo] = audioBytes } -// parseSSEStream SSE 流式解析（图片模型等） -func parseSSEStream(rawBytes []byte, streamConfig map[string]any) (map[string]any, error) { - events, _ := streamConfig["events"].([]any) - if len(events) == 0 { - return gjson.New(string(rawBytes)).Map(), nil +// processCollect 数组收集 +func processCollect(lines []string, event map[string]any, result map[string]any) { + match, _ := event["match"].(string) + aggregateTo, _ := event["aggregate_to"].(string) + orderBy, _ := event["order_by"].(string) + fields, _ := event["fields"].(map[string]any) + + var items []map[string]any + for _, line := range lines { + line = strings.TrimSpace(line) + if line == "" || line == "[DONE]" { + continue + } + + var chunk map[string]any + if err := json.Unmarshal([]byte(line), &chunk); err != nil { + continue + } + + chunkType, _ := chunk["type"].(string) + if match != "" && !strings.Contains(chunkType, match) { + continue + } + + item := make(map[string]any) + for localKey, chunkPath := range fields { + item[localKey] = gjson.New(chunk).Get(chunkPath.(string)).Val() + } + items = append(items, item) } - lines := strings.Split(string(rawBytes), "\n") - result := make(map[string]any) - var partials []map[string]any + if orderBy != "" { + sort.Slice(items, func(i, j int) bool { + return fmt.Sprint(items[i][orderBy]) < fmt.Sprint(items[j][orderBy]) + }) + } + + // 如果只有一个字段，直接存值数组 + if len(fields) == 1 { + var vals []any + for _, item := range items { + for _, v := range item { + vals = append(vals, v) + } + } + result[aggregateTo] = vals + } else { + result[aggregateTo] = items + } +} + +// processFinal 取最后一条匹配的数据 +func processFinal(lines []string, event map[string]any, result map[string]any) { + match, _ := event["match"].(string) + aggregateTo, _ := event["aggregate_to"].(string) + fields, _ := event["fields"].(map[string]any) + + var lastMatch map[string]any for _, line := range lines { line = strings.TrimSpace(line) @@ -96,55 +230,26 @@ func parseSSEStream(rawBytes []byte, streamConfig map[string]any) (map[string]an } chunkType, _ := chunk["type"].(string) + if match != "" && !strings.Contains(chunkType, match) { + continue + } - for _, evt := range events { - e, _ := evt.(map[string]any) - match, _ := e["match"].(string) - if !strings.Contains(chunkType, match) { - continue - } + lastMatch = chunk + } - fields, _ := e["fields"].(map[string]any) - aggregateTo, _ := e["aggregate_to"].(string) - evtType, _ := e["type"].(string) + if lastMatch == nil { + return + } - switch evtType { - case "partial": - item := make(map[string]any) - for localKey, chunkKey := range fields { - item[localKey] = chunk[chunkKey.(string)] - } - partials = append(partials, item) - - case "final": - for localKey, chunkKey := range fields { - val := gjson.New(chunk).Get(chunkKey.(string)) - if !val.IsNil() { - if _, exists := result[aggregateTo]; !exists { - result[aggregateTo] = make(map[string]any) - } - result[aggregateTo].(map[string]any)[localKey] = val.Val() - } - } - } + data := make(map[string]any) + for localKey, chunkPath := range fields { + val := gjson.New(lastMatch).Get(chunkPath.(string)).Val() + if val != nil { + data[localKey] = val } } - if len(partials) > 0 { - for _, evt := range events { - e, _ := evt.(map[string]any) - if e["type"] == "partial" { - if orderBy, ok := e["order_by"].(string); ok { - sort.Slice(partials, func(i, j int) bool { - return fmt.Sprint(partials[i][orderBy]) < fmt.Sprint(partials[j][orderBy]) - }) - } - result[e["aggregate_to"].(string)] = partials - break - } - } + if len(data) > 0 { + result[aggregateTo] = data } - - mergedBytes, _ := json.Marshal(result) - return gjson.New(mergedBytes).Map(), nil } diff --git a/controller/model_gateway_models_controller.go b/controller/model_gateway_models_controller.go index d7a0810..17ecd15 100644 --- a/controller/model_gateway_models_controller.go +++ b/controller/model_gateway_models_controller.go @@ -17,6 +17,11 @@ func (c *model) CreateModel(ctx context.Context, req *dto.CreateModelReq) (res * return modelService.ModelGatewayModels.Create(ctx, req) } +// GetModel 获取配置详情 +func (c *model) GetModel(ctx context.Context, req *dto.GetModelReq) (res *dto.GetModelRes, err error) { + return modelService.ModelGatewayModels.Get(ctx, req) +} + // UpdateModel 更改配置 func (c *model) UpdateModel(ctx context.Context, req *dto.UpdateModelReq) (res *dto.UpdateModelRes, err error) { err = modelService.ModelGatewayModels.Update(ctx, req) @@ -29,11 +34,6 @@ func (c *model) DeleteModel(ctx context.Context, req *dto.DeleteModelReq) (res * return } -// GetModel 获取配置详情 -func (c *model) GetModel(ctx context.Context, req *dto.GetModelReq) (res *dto.GetModelRes, err error) { - return modelService.ModelGatewayModels.Get(ctx, req) -} - // ListModel 配置列表 func (c *model) ListModel(ctx context.Context, req *dto.ListModelReq) (res *dto.ListModelRes, err error) { return modelService.ModelGatewayModels.List(ctx, req) diff --git a/model/dto/model_gateway_models_dto.go b/model/dto/model_gateway_models_dto.go index 267178e..883281f 100644 --- a/model/dto/model_gateway_models_dto.go +++ b/model/dto/model_gateway_models_dto.go @@ -11,6 +11,7 @@ import ( type CreateModelReq struct { g.Meta `path:"/createModel" method:"post" tags:"模型管理" summary:"创建模型配置" dc:"添加新的模型配置"` ModelName string `p:"modelName" json:"modelName" v:"required#模型名称不能为空" dc:"模型名称（唯一标识）"` + OperatorName string `p:"operatorName" json:"operatorName" dc:"运营商名称"` ModelType int `p:"modelType" json:"modelType" v:"required#模型类型不能为空" dc:"模型类型"` BaseURL string `p:"baseUrl" json:"baseUrl" v:"required#模型地址不能为空" dc:"模型服务地址"` HttpMethod string `p:"httpMethod" json:"httpMethod" dc:"请求方式：GET/POST（默认POST）"` @@ -19,17 +20,15 @@ type CreateModelReq struct { Enabled *int `p:"enabled" json:"enabled" dc:"是否启用：0-停用 1-启用"` IsChatModel *int `p:"isChatModel" json:"isChatModel" dc:"是否为对话模型：0-否 1-是"` CallMode *int `p:"callMode" json:"callMode" dc:"调用模式：0-同步 1-异步 2-流式"` - RequiredFields []string `p:"requiredFields" json:"requiredFields" dc:"必填字段"` ApiKey string `p:"apiKey" json:"apiKey" dc:"调用凭证/密钥"` Form []entity.Form `p:"form" json:"form" dc:"动态表单配置"` RequestMapping map[string]any `p:"requestMapping" json:"requestMapping" dc:"请求映射"` ResponseMapping map[string]any `p:"responseMapping" json:"responseMapping" dc:"返回映射"` - OperatorName string `p:"operatorName" json:"operatorName" dc:"运营商名称"` - TokenConfig map[string]any `p:"tokenConfig" json:"tokenConfig" dc:"token计算配置"` ExtendMapping map[string]any `p:"extendMapping" json:"extendMapping" dc:"附加映射"` QueryConfig map[string]any `p:"queryConfig" json:"queryConfig" dc:"查询/回调配置"` StreamConfig map[string]any `p:"streamConfig" json:"streamConfig" dc:"流式输出配置"` SpecialParams map[string]any `p:"specialParams" json:"specialParams" dc:"请求特殊参数（首尾帧等）"` + RequiredFields []string `p:"requiredFields" json:"requiredFields" dc:"必填字段"` MaxConcurrency int `p:"maxConcurrency" json:"maxConcurrency" dc:"最大并发数（默认10）"` TimeoutSeconds int `p:"timeoutSeconds" json:"timeoutSeconds" dc:"请求超时时间（秒，默认600）"` RetryTimes int `p:"retryTimes" json:"retryTimes" dc:"失败重试次数（默认3）"` @@ -43,6 +42,7 @@ type UpdateModelReq struct { g.Meta `path:"/updateModel" method:"put" tags:"模型管理" summary:"更新模型配置" dc:"更新指定ID的模型配置"` ID int64 `p:"id" json:"id" v:"required#id不能为空" dc:"配置ID"` ModelName string `p:"modelName" json:"modelName" v:"required#模型名称不能为空" dc:"模型名称（唯一标识）"` + OperatorName string `p:"operatorName" json:"operatorName" dc:"运营商名称"` ModelType int `p:"modelType" json:"modelType" v:"required#模型类型不能为空" dc:"模型类型"` BaseURL string `p:"baseUrl" json:"baseUrl" v:"required#模型地址不能为空" dc:"模型服务地址"` HttpMethod string `p:"httpMethod" json:"httpMethod" dc:"请求方式：GET/POST（默认POST）"` @@ -51,17 +51,15 @@ type UpdateModelReq struct { Enabled *int `p:"enabled" json:"enabled" dc:"是否启用：0-停用 1-启用"` IsChatModel *int `p:"isChatModel" json:"isChatModel" dc:"是否为对话模型：0-否 1-是"` CallMode *int `p:"callMode" json:"callMode" dc:"调用模式：0-同步 1-异步 2-流式"` - RequiredFields []string `p:"requiredFields" json:"requiredFields" dc:"必填字段"` ApiKey string `p:"apiKey" json:"apiKey" dc:"调用凭证/密钥"` Form []entity.Form `p:"form" json:"form" dc:"动态表单配置"` RequestMapping map[string]any `p:"requestMapping" json:"requestMapping" dc:"请求映射"` ResponseMapping map[string]any `p:"responseMapping" json:"responseMapping" dc:"返回映射"` - OperatorName string `p:"operatorName" json:"operatorName" dc:"运营商名称"` - TokenConfig map[string]any `p:"tokenConfig" json:"tokenConfig" dc:"token计算配置"` ExtendMapping map[string]any `p:"extendMapping" json:"extendMapping" dc:"附加映射"` QueryConfig map[string]any `p:"queryConfig" json:"queryConfig" dc:"查询/回调配置"` StreamConfig map[string]any `p:"streamConfig" json:"streamConfig" dc:"流式输出配置"` SpecialParams map[string]any `p:"specialParams" json:"specialParams" dc:"请求特殊参数（首尾帧等）"` + RequiredFields []string `p:"requiredFields" json:"requiredFields" dc:"必填字段"` MaxConcurrency int `p:"maxConcurrency" json:"maxConcurrency" dc:"最大并发数（默认10）"` TimeoutSeconds int `p:"timeoutSeconds" json:"timeoutSeconds" dc:"请求超时时间（秒，默认600）"` RetryTimes int `p:"retryTimes" json:"retryTimes" dc:"失败重试次数（默认3）"` @@ -83,11 +81,8 @@ type DeleteModelRes struct { // GetModelReq 获取模型配置详情 type GetModelReq struct { - g.Meta `path:"/getModel" method:"get" tags:"模型管理" summary:"获取模型配置" dc:"根据模型ID获取配置详情"` - ID int64 `p:"id" json:"id,string" dc:"配置ID"` - Creator string `p:"creator" json:"creator" dc:"创建人"` - IsChatModel *int `p:"isChatModel" json:"isChatModel" dc:"是否为聊天模型"` - ModelName string `p:"modelName" json:"modelName" dc:"模型名称（唯一标识）"` + g.Meta `path:"/getModel" method:"get" tags:"模型管理" summary:"获取模型配置" dc:"根据模型ID获取配置详情"` + ID int64 `p:"id" json:"id,string" dc:"配置ID"` } type GetModelRes struct { diff --git a/model/entity/model_gateway_model.go b/model/entity/model_gateway_model.go index df9f23b..b5d8983 100644 --- a/model/entity/model_gateway_model.go +++ b/model/entity/model_gateway_model.go @@ -22,7 +22,6 @@ type modelGatewayModelCol struct { TimeoutSeconds string RetryTimes string OperatorName string - TokenConfig string ExtendMapping string QueryConfig string StreamConfig string @@ -50,7 +49,6 @@ var ModelGatewayModelCol = modelGatewayModelCol{ TimeoutSeconds: "timeout_seconds", RetryTimes: "retry_times", OperatorName: "operator_name", - TokenConfig: "token_config", ExtendMapping: "extend_mapping", QueryConfig: "query_config", StreamConfig: "stream_config", @@ -61,29 +59,28 @@ var ModelGatewayModelCol = modelGatewayModelCol{ type ModelGatewayModel struct { beans.SQLBaseDO `orm:",inline"` ModelName string `orm:"model_name" json:"modelName"` + OperatorName string `orm:"operator_name" json:"operatorName"` ModelType int `orm:"model_type" json:"modelType"` BaseURL string `orm:"base_url" json:"baseUrl"` HttpMethod string `orm:"http_method" json:"httpMethod"` HeadMsg map[string]any `orm:"head_msg" json:"headMsg"` - Form []Form `orm:"form_json" json:"form"` - RequestMapping map[string]any `orm:"request_mapping" json:"requestMapping"` - ResponseMapping map[string]any `orm:"response_mapping" json:"responseMapping"` - RequiredFields []string `orm:"required_fields" json:"requiredFields"` IsPrivate *int `orm:"is_private" json:"isPrivate"` IsChatModel *int `orm:"is_chat_model" json:"isChatModel"` CallMode *int `orm:"call_mode" json:"callMode"` ApiKey string `orm:"api_key" json:"apiKey"` Enabled *int `orm:"enabled" json:"enabled"` - MaxConcurrency int `orm:"max_concurrency" json:"maxConcurrency"` - TimeoutSeconds int `orm:"timeout_seconds" json:"timeoutSeconds"` - RetryTimes int `orm:"retry_times" json:"retryTimes"` - OperatorName string `orm:"operator_name" json:"operatorName"` - TokenConfig map[string]any `orm:"token_config" json:"tokenConfig"` + Form []Form `orm:"form_json" json:"form"` + RequestMapping map[string]any `orm:"request_mapping" json:"requestMapping"` + ResponseMapping map[string]any `orm:"response_mapping" json:"responseMapping"` ExtendMapping map[string]any `orm:"extend_mapping" json:"extendMapping"` QueryConfig map[string]any `orm:"query_config" json:"queryConfig"` StreamConfig map[string]any `orm:"stream_config" json:"streamConfig"` SpecialParams map[string]any `orm:"special_params" json:"specialParams"` BillingConfig map[string]any `orm:"billing_config" json:"billingConfig"` + RequiredFields []string `orm:"required_fields" json:"requiredFields"` + MaxConcurrency int `orm:"max_concurrency" json:"maxConcurrency"` + TimeoutSeconds int `orm:"timeout_seconds" json:"timeoutSeconds"` + RetryTimes int `orm:"retry_times" json:"retryTimes"` } type Form struct { diff --git a/service/model/model_service.go b/service/model/model_service.go index f2f301f..488c667 100644 --- a/service/model/model_service.go +++ b/service/model/model_service.go @@ -81,16 +81,11 @@ func (s *modelService) Get(ctx context.Context, req *dto.GetModelReq) (*dto.GetM if err != nil { return nil, err } - if g.IsEmpty(req.ID) { - req.Creator = user.UserName - } model, err := dao.ModelGatewayModels.Get(ctx, &entity.ModelGatewayModel{ SQLBaseDO: beans.SQLBaseDO{ Id: req.ID, Creator: user.UserName, }, - ModelName: req.ModelName, - IsChatModel: req.IsChatModel, }) if err != nil || model == nil { return nil, err diff --git a/service/task/worker.go b/service/task/worker.go index 6bdadca..979a349 100644 --- a/service/task/worker.go +++ b/service/task/worker.go @@ -63,7 +63,7 @@ func (w *asyncWorker) handleOne(ctx context.Context, task *entity.ModelGatewayTa switch { case model.CallMode != nil && *model.CallMode == public.CallModeStream: // 流式 if err == nil { - result, err = util.ParseStreamResponse(rawData, model.StreamConfig) + result, err = util.ParseStreamResponse(ctx, rawData, model.StreamConfig) } case model.CallMode != nil && *model.CallMode == public.CallModeAsync: // 异步 if err == nil {