From b2cad4cac2d8c6aac98174452edcd42ac7a700da Mon Sep 17 00:00:00 2001 From: WangLiZhao <1838393649@qq.com> Date: Wed, 3 Jun 2026 18:37:18 +0800 Subject: [PATCH] =?UTF-8?q?refactor(model-gateway):=20=E9=87=8D=E6=9E=84?= =?UTF-8?q?=E4=BB=A3=E7=A0=81=E7=BB=93=E6=9E=84=E5=B9=B6=E4=BC=98=E5=8C=96?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E5=BA=93=E6=9F=A5=E8=AF=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- common/util/network.go | 130 ---------- config.yml | 38 --- consts/public/public.go | 1 + controller/prompt_compose_controller.go | 37 --- model/dto/prompt_compose_dto.go | 11 +- model/entity/prompts_compose_session.go | 3 + service/gateway/gateway_http_service.go | 82 ++----- service/prompt/prompt_build_service.go | 59 +++-- service/prompt/prompt_compose_service.go | 297 ++++++++++------------- service/prompt/prompt_ir_service.go | 2 +- 10 files changed, 190 insertions(+), 470 deletions(-) delete mode 100644 common/util/network.go diff --git a/common/util/network.go b/common/util/network.go deleted file mode 100644 index 62acd7b..0000000 --- a/common/util/network.go +++ /dev/null @@ -1,130 +0,0 @@ -package util - -import ( - "context" - "net" - "strings" - - "github.com/gogf/gf/v2/frame/g" -) - -// GetLocalIP 获取本机有效的局域网 IPv4 地址 -func GetLocalIP() string { - addrs, err := net.InterfaceAddrs() - if err != nil { - return "127.0.0.1" - } - - var validIPs []string - - for _, addr := range addrs { - ipnet, ok := addr.(*net.IPNet) - if !ok { - continue - } - - ip := ipnet.IP - - if isIPValid(ip) { - validIPs = append(validIPs, ip.String()) - } - } - - // 优先返回非 169.254.x.x 的 IP - for _, ip := range validIPs { - if !strings.HasPrefix(ip, "169.254.") { - return ip - } - } - - // 其次返回 169.254.x.x（最后的选择） - if len(validIPs) > 0 { - return validIPs[0] - } - - return "127.0.0.1" -} - -// isIPValid 判断 IP 是否有效 -func isIPValid(ip net.IP) bool { - // 不是 loopback (127.0.0.1) - if ip.IsLoopback() { - return false - } - - // 是 IPv4 - if ip.To4() == nil { - return false - } - - // 不是链路本地地址 (169.254.0.0/16) - if ip[0] == 169 && ip[1] == 254 { - return false - } - - // 不是组播地址 - if ip.IsMulticast() { - return false - } - - // 不是未指定地址 (0.0.0.0) - if ip.IsUnspecified() { - return false - } - - return true -} - -// GetLocalAddress 获取局域网地址（IP:端口） -func GetLocalAddress(ctx context.Context) string { - ip := GetLocalIP() - port := GetServerPort(ctx) - - if port == "80" || port == "443" { - return ip - } - return ip + ":" + port -} - -// GetSchemaFromRequest 从当前请求中获取协议（http/https） -func GetSchemaFromRequest(ctx context.Context) string { - r := g.RequestFromCtx(ctx) - if r == nil { - return "http" - } - - // 1. 代理场景：X-Forwarded-Proto - if proto := r.Header.Get("X-Forwarded-Proto"); proto != "" { - return proto - } - - // 2. 代理场景：X-Forwarded-Scheme - if proto := r.Header.Get("X-Forwarded-Scheme"); proto != "" { - return proto - } - - // 3. TLS 连接（直接 HTTPS） - if r.TLS != nil { - return "https" - } - - // 4. 默认 HTTP（这行很重要！） - return "http" // ← 确保有这行 -} - -// GetLocalBaseURL 获取局域网基础 URL（动态协议 + IP + 端口） -func GetLocalBaseURL(ctx context.Context) string { - schema := GetSchemaFromRequest(ctx) - addr := GetLocalAddress(ctx) - return schema + "://" + addr -} - -// GetCallbackURL 获取回调地址（完整 URL） -func GetCallbackURL(ctx context.Context, path string) string { - baseURL := GetLocalBaseURL(ctx) - // 确保 path 以 / 开头 - if !strings.HasPrefix(path, "/") { - path = "/" + path - } - return baseURL + path -} diff --git a/config.yml b/config.yml index 9f501b9..a00b2ed 100644 --- a/config.yml +++ b/config.yml @@ -112,41 +112,3 @@ nodePrompts: | %s 上下文内容： %s - -#你是专业的JSON结构生成专家，必须严格遵守以下全部规则。 -# 【强制规则】 -# 必须根据【输出结构】里面返回的JSON结构进行生成，不得任何更改，最终内容与输出结构返回一致； -# 完整阅读所有文本、规则、表单内容，禁止跳读、漏读； -# 完整读取UserForm所有字段，不得忽略任何字段； -# 如果有skill相关内容必须完整的将内容拼接到system角色描述中； -# 理解全部语义后再输出，禁止断章取义； -# UserForm所有字段内容必须完整拼接赋值到user角色描述中，不得有任何遗漏。 -# 【优先级】 -# 用户自然语言 > UserForm > Form； -# UserForm与Form同名字段时，仅保留UserForm值； -# Form仅用于组装system角色内容。 -# 【表单处理】 -# Form：系统提示词、默认参数、基础配置 → 专属填充system角色； -# UserForm：用户业务输入、文案、配图数量、比例、prompt等 → 全部解析后拼接进user角色content； -# 自动提取UserForm中每条文案的配图数量，总图片数 = 各文案配图数累加求和，用户没有相关数量必须默认1； -# 图片尺寸为空时自动填充size=1024*1024。 -# 【结构铁律】 -# 严格沿用固定输出结构，不增删字段或修改层级； -# messages元素必须按结构返回； -# 禁止将role对象转为字符串、禁止嵌套错乱； -# 输出纯净JSON：无多余转义符、无换行符、无额外字符； -# 所有括号、引号必须成对闭合，保证JSON合法。 -# 【参数赋值】 -# model固定沿用传入值； -# 返回结构里面的参数，需要根据语意进行赋值，缺失补默认值； -# history历史信息必须结合UserForm里的内容对用户描述部分进行补充； -# 从UserForm提取信息整合进user描述，确保数量、尺寸、文案语义无遗漏。 -# 【输出要求】 -# 仅输出单行纯净JSON，无任何解释、备注、Markdown或多余符号； -# 完整合UserForm全部字段语义到user描述； -# 生成后自检JSON语法、结构、数量；错误则自动重新生成。 -# 【输出结构】 -# %s -# 【完整输入信息】 -# %s -# 直接输出最终JSON： \ No newline at end of file diff --git a/consts/public/public.go b/consts/public/public.go index acf3ee1..36bb062 100644 --- a/consts/public/public.go +++ b/consts/public/public.go @@ -9,4 +9,5 @@ const ( const ( BuildTypePrompt = 1 //提示词构建 BuildTypeNode = 2 //节点构建 + BuildTypeStruct = 3 //结构构建 ) diff --git a/controller/prompt_compose_controller.go b/controller/prompt_compose_controller.go index d23fb98..96be2e4 100644 --- a/controller/prompt_compose_controller.go +++ b/controller/prompt_compose_controller.go @@ -2,17 +2,8 @@ package controller import ( "context" - "prompts-core/common/util" - "prompts-core/dao" "prompts-core/model/dto" - "prompts-core/model/entity" - "prompts-core/service/gateway" - promptService "prompts-core/service/prompt" - - "gitea.com/red-future/common/beans" - "github.com/gogf/gf/v2/encoding/gjson" - "github.com/gogf/gf/v2/frame/g" ) type prompt struct{} @@ -35,31 +26,3 @@ func (c *prompt) Callback(ctx context.Context, req *dto.CallbackReq) (res *dto.C func (c *prompt) GetComposeTask(ctx context.Context, req *dto.GetComposeTaskReq) (res *dto.GetComposeTaskRes, err error) { return promptService.GetComposeTask(ctx, req.TaskId) } - -func (c *prompt) Text(ctx context.Context, req *dto.TextReq) (res *dto.TextRes, err error) { - composeTask, err := dao.ComposeTask.Get(ctx, &entity.ComposeTask{ - TaskId: "c58c9296-994f-4e83-8285-1daebf3c492d", - }) - if err != nil { - return - } - model, err := gateway.GetModelConfig(ctx, &gateway.AsynchModel{ - SQLBaseDO: beans.SQLBaseDO{Creator: composeTask.Creator}, - ModelName: composeTask.ModelName, - }) - if err != nil { - return - } - message := promptService.ParsePromptResult(composeTask.ResultText) - - // 加这两行 - g.Log().Infof(ctx, "[Text] RequestPayload.consult: %v", composeTask.RequestPayload["consult"]) - g.Log().Infof(ctx, "[Text] ExtendMapping: %v", model.ExtendMapping) - messages := util.MergeConsult(composeTask.RequestPayload, message, model.ExtendMapping) - g.Log().Infof(ctx, "[Text] MergeConsult 结果 rounds[0].messages[0].content: %v", - gjson.New(messages).Get("rounds.0.messages.0.content")) - res = &dto.TextRes{ - Messages: messages, - } - return -} diff --git a/model/dto/prompt_compose_dto.go b/model/dto/prompt_compose_dto.go index 972662e..4ffded8 100644 --- a/model/dto/prompt_compose_dto.go +++ b/model/dto/prompt_compose_dto.go @@ -6,7 +6,8 @@ type ComposeMessagesReq struct { g.Meta `path:"/composeMessages" method:"post" tags:"提示词处理" summary:"拼接提示词" dc:"按 modelTypeId 读取 prompts_model_prompt.prompt_info 与 response_json_schema；form 作为系统表单，userForm 作为用户表单，结合 userFiles 调用 model-gateway，并直接返回最终 messages"` ModelName string `p:"modelName" json:"modelName" v:"required#modelName不能为空" dc:"实际请求的网关模型名称"` BuildType int `p:"buildType" json:"buildType" v:"required#buildType不能为空" dc:"构建类型"` //判断节点 - SessionId string `p:"sessionId" json:"sessionId" dc:"会话ID"` //v:"required#sessionId不能为空" + NodeId string `p:"nodeId" json:"nodeId" dc:"节点ID"` + SessionId string `p:"sessionId" json:"sessionId" dc:"会话ID"` //v:"required#sessionId不能为空" Cause string `p:"cause" json:"cause" v:"required-if:IsBuilder,false#原因不能为空" dc:"原因"` CallbackUrl string `p:"callbackUrl" json:"callbackUrl" dc:"回调地址"` Form []map[string]any `p:"form" json:"form" dc:"系统表单：form 下所有字段都作为系统提示词来源"` @@ -59,11 +60,3 @@ type GetComposeTaskRes struct { OssFile string `json:"ossFile" dc:"结果文件地址"` FileType string `json:"fileType" dc:"结果文件类型"` } - -type TextReq struct { - g.Meta `path:"/text" method:"post" tags:"提示词处理" summary:"拼接提示词" dc:"按 modelTypeId 读取 prompts_model_prompt.prompt_info 与 response_json_schema；form 作为系统表单，userForm 作为用户表单，结合 userFiles 调用 model-gateway，并直接返回最终 messages"` -} - -type TextRes struct { - Messages any `json:"messages" dc:"文本结果"` -} diff --git a/model/entity/prompts_compose_session.go b/model/entity/prompts_compose_session.go index bead3e3..ac3f3aa 100644 --- a/model/entity/prompts_compose_session.go +++ b/model/entity/prompts_compose_session.go @@ -5,6 +5,7 @@ import "gitea.com/red-future/common/beans" type ComposeSession struct { beans.SQLBaseDO `orm:",inline"` SessionId string `orm:"session_id" json:"sessionId"` + NodeId string `orm:"node_id" json:"nodeId"` RequestContent map[string]any `orm:"request_content" json:"requestContent"` ResponseContent map[string]any `orm:"response_content" json:"responseContent"` Remark string `orm:"remark" json:"remark"` @@ -13,6 +14,7 @@ type ComposeSession struct { type composeSessionCol struct { beans.SQLBaseCol SessionId string + NodeId string RequestContent string ResponseContent string Remark string @@ -21,6 +23,7 @@ type composeSessionCol struct { var ComposeSessionCol = composeSessionCol{ SQLBaseCol: beans.DefSQLBaseCol, SessionId: "session_id", + NodeId: "node_id", RequestContent: "request_content", ResponseContent: "response_content", Remark: "remark", diff --git a/service/gateway/gateway_http_service.go b/service/gateway/gateway_http_service.go index f8fbc2b..6768d05 100644 --- a/service/gateway/gateway_http_service.go +++ b/service/gateway/gateway_http_service.go @@ -6,9 +6,11 @@ import ( "fmt" "prompts-core/common/util" "prompts-core/model/entity" + "strings" "gitea.com/red-future/common/beans" commonHttp "gitea.com/red-future/common/http" + "github.com/gogf/gf/v2/encoding/gjson" "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/os/gtime" ) @@ -77,16 +79,28 @@ type AsynchModel struct { // GetModelConfig 获取模型配置 func GetModelConfig(ctx context.Context, req *AsynchModel) (model *AsynchModel, err error) { - fmt.Println("req参数", req) - fullURL := fmt.Sprintf("model-gateway/model/getModel?creator=%s&modelName=%s&isChatModel=%d", - req.Creator, req.ModelName, req.IsChatModel) + fullURL := "model-gateway/model/getModel" + // 拼接 query 参数 + var params []string + if req.Creator != "" { + params = append(params, fmt.Sprintf("creator=%s", req.Creator)) + } + if req.ModelName != "" { + params = append(params, fmt.Sprintf("modelName=%s", req.ModelName)) + } + if req.IsChatModel != 0 { + params = append(params, fmt.Sprintf("isChatModel=%d", req.IsChatModel)) + } + if len(params) > 0 { + fullURL += "?" + strings.Join(params, "&") + } headers := util.ForwardHeaders(ctx) var resp GetModelConfigResp if err = commonHttp.Get(ctx, fullURL, headers, &resp, nil); err != nil { return nil, fmt.Errorf("获取模型配置失败: %w", err) } if resp.Model == nil { - return nil, fmt.Errorf("模型不存在: creator=%s modelName=%s isChatModel=%d", req.Creator, req.ModelName, req.IsChatModel) + return nil, fmt.Errorf("模型不存在") } return resp.Model, nil } @@ -134,15 +148,11 @@ func GetSkillUser(ctx context.Context, name string) (*SkillUserVO, error) { // SendCallbackReq 发送回调的请求体 type SendCallbackReq struct { - TaskId string `json:"taskId"` - Status string `json:"status"` - Messages *MultiRoundResult `json:"messages,omitempty"` - EpicycleId int64 `json:"epicycleId"` - ErrorMsg string `json:"errorMsg,omitempty"` -} -type MultiRoundResult struct { - TotalRounds int `json:"total_rounds"` // 总轮数 - Rounds []map[string]any `json:"rounds"` // 每轮详情（动态类型） + TaskId string `json:"taskId"` + Status string `json:"status"` + Messages map[string]any `json:"messages,omitempty"` + EpicycleId int64 `json:"epicycleId"` + ErrorMsg string `json:"errorMsg,omitempty"` } // SendCallback 向业务方发送回调 @@ -151,61 +161,21 @@ func SendCallback(ctx context.Context, composeTask *entity.ComposeTask) error { if composeTask.CallbackUrl == "" { return fmt.Errorf("回调地址为空，taskId=%s", composeTask.TaskId) } - // 2. 构造请求体 req := SendCallbackReq{ TaskId: composeTask.TaskId, Status: composeTask.Status, - Messages: parseMessagesToResult(composeTask.Messages), // 需要将 JSON 字符串转为结构体 + Messages: composeTask.Messages, ErrorMsg: composeTask.ErrorMessage, } // 3. 发送 POST 请求 headers := util.ForwardHeaders(ctx) var resp struct{} g.Log().Infof(ctx, "[回调业务] 开始发送 taskId=%s 回调地址=%s 消息=%v", - composeTask.TaskId, composeTask.CallbackUrl, req.Messages) + composeTask.TaskId, composeTask.CallbackUrl, gjson.New(req.Messages).String()) if err := commonHttp.Post(ctx, composeTask.CallbackUrl, headers, &resp, req); err != nil { return fmt.Errorf("[回调业务] 发送失败 taskId=%s url=%s err=%w", composeTask.TaskId, composeTask.CallbackUrl, err) } - g.Log().Infof(ctx, "[回调业务] 发送成功 taskId=%s 回调地址=%s", composeTask.TaskId, composeTask.CallbackUrl) + g.Log().Infof(ctx, "[回调业务] 发送成功 taskId=%s 回调地址=%s ", composeTask.TaskId, composeTask.CallbackUrl) return nil } - -// parseMessagesToResult 将 any 类型的 Messages 转为 *MultiRoundResult -func parseMessagesToResult(messages any) *MultiRoundResult { - if messages == nil { - return nil - } - - var result MultiRoundResult - - switch v := messages.(type) { - case *MultiRoundResult: - return v - case MultiRoundResult: - return &v - case string: - if err := json.Unmarshal([]byte(v), &result); err != nil { - return nil - } - case []byte: - if err := json.Unmarshal(v, &result); err != nil { - return nil - } - case map[string]any: - // 通过 JSON 序列化再反序列化 - data, _ := json.Marshal(v) - if err := json.Unmarshal(data, &result); err != nil { - return nil - } - default: - data, err := json.Marshal(v) - if err != nil { - return nil - } - if err = json.Unmarshal(data, &result); err != nil { - return nil - } - } - return &result -} diff --git a/service/prompt/prompt_build_service.go b/service/prompt/prompt_build_service.go index a73816d..b08556c 100644 --- a/service/prompt/prompt_build_service.go +++ b/service/prompt/prompt_build_service.go @@ -2,10 +2,9 @@ package prompt import ( "context" - "errors" "fmt" - "prompts-core/consts/public" "prompts-core/service/gateway" + "prompts-core/service/session" "strings" "prompts-core/common/util" @@ -13,6 +12,7 @@ import ( "prompts-core/model/dto" "prompts-core/model/entity" + "gitea.com/red-future/common/utils" "github.com/gogf/gf/v2/encoding/gjson" "github.com/gogf/gf/v2/util/gconv" ) @@ -29,30 +29,13 @@ type UserPromptPayload struct { BuildType int `json:"buildType"` } -// buildInferenceRequest 构建推理请求 -func buildInferenceRequest(ctx context.Context, req *dto.ComposeMessagesReq, chatModel *gateway.AsynchModel, aiModel *gateway.AsynchModel, history []map[string]any) (map[string]any, error) { - //1) 处理表单分批 - processedReq, totalBatches, err := ProcessUserFormBatches(ctx, req, aiModel) - if err != nil { - return nil, fmt.Errorf("处理用户表单分批失败: %w", err) - } - ir := NewPromptIR() - switch req.BuildType { - case public.BuildTypePrompt: - return buildPromptTypeRequest(ctx, processedReq, aiModel, chatModel, history, ir, totalBatches) - case public.BuildTypeNode: - return buildNodeTypeRequest(ctx, req, chatModel, ir) - default: - return nil, errors.New("不支持的构建类型") - } -} - // buildPromptTypeRequest 构建提示词类型请求（BuildType=1） -func buildPromptTypeRequest(ctx context.Context, req *dto.ComposeMessagesReq, aiModel *gateway.AsynchModel, chatModel *gateway.AsynchModel, history []map[string]any, ir *PromptIR, totalBatches int) (map[string]any, error) { +func buildPromptTypeRequest(ctx context.Context, req *dto.ComposeMessagesReq, aiModel *gateway.AsynchModel, chatModel *gateway.AsynchModel, ir *PromptIR, totalBatches int) (map[string]any, error) { //1) 构建系统提示词 systemPrompt := promptBuildWithRounds(ctx, req, chatModel, aiModel, totalBatches) ir.AddSystem(systemPrompt) //2) 构建历史对话 + history, _ := session.GetHistoryMessages(ctx, req.SessionId) for _, msg := range history { role := gconv.String(msg["role"]) if role != "user" && role != "assistant" { @@ -75,24 +58,40 @@ func buildNodeTypeRequest(ctx context.Context, req *dto.ComposeMessagesReq, chat return compileToProviderRequest(ctx, ir, chatModel) } -// compileToProviderRequest 编译为 Provider 请求 -func compileToProviderRequest(ctx context.Context, ir *PromptIR, chatModel *gateway.AsynchModel) (map[string]any, error) { - protocol, err := GetProtocolByProvider(ctx, chatModel.OperatorName) - if err != nil { - return nil, fmt.Errorf("获取协议配置失败: %w", err) +// buildStructTypeRequest 构建结构体类型请求（BuildType=3） +func buildStructTypeRequest(ctx context.Context, req *dto.ComposeMessagesReq, chatModel *gateway.AsynchModel, ir *PromptIR) (map[string]any, error) { + // 提取 userForm 中的 prompt 作为自定义提示词 + var customPrompt string + for _, item := range req.UserForm { + if prompt, ok := item["prompt"]; ok && gconv.String(prompt) != "" { + customPrompt = gconv.String(prompt) + break + } } - if protocol == nil { - return nil, errors.New("协议配置不存在") + // 用户消息 + ir.AddSystem(customPrompt) + ir.AddUser(buildUserPrompt(ctx, req, "")) + return compileToProviderRequest(ctx, ir, chatModel, customPrompt) +} + +// compileToProviderRequest 编译为 Provider 请求 +func compileToProviderRequest(ctx context.Context, ir *PromptIR, chatModel *gateway.AsynchModel, customPrompt ...string) (map[string]any, error) { + protocol, err := GetProtocolByProvider(ctx, chatModel.OperatorName) + if err != nil || protocol == nil { + return nil, fmt.Errorf("协议配置不存在或获取失败: %w", err) + } + // 如果传了自定义提示词，替换掉协议模板 + if len(customPrompt) > 0 && customPrompt[0] != "" { + protocol.SystemPromptTemplate = customPrompt[0] } providerReq, err := Compile(ir, protocol, chatModel) if err != nil { return nil, fmt.Errorf("编译请求失败: %w", err) } - return map[string]any{ "modelName": chatModel.ModelName, "bizName": util.GetServerName(ctx), - "callbackUrl": util.GetCallbackURL(ctx, "/prompt/callback"), + "callbackUrl": utils.GetCallbackURL(ctx, "/prompt/callback"), "requestPayload": providerReq, }, nil } diff --git a/service/prompt/prompt_compose_service.go b/service/prompt/prompt_compose_service.go index 9076471..a6ab82e 100644 --- a/service/prompt/prompt_compose_service.go +++ b/service/prompt/prompt_compose_service.go @@ -5,11 +5,6 @@ import ( "encoding/json" "errors" "fmt" - "prompts-core/service/session" - - "gitea.com/red-future/common/beans" - "gitea.com/red-future/common/utils" - "github.com/gogf/gf/v2/frame/g" "prompts-core/common/util" "prompts-core/consts/public" @@ -17,28 +12,24 @@ import ( "prompts-core/model/dto" "prompts-core/model/entity" "prompts-core/service/gateway" + + "gitea.com/red-future/common/beans" + "gitea.com/red-future/common/utils" + "github.com/gogf/gf/v2/frame/g" ) // ComposeMessages 核心拼接提示词主流程 func ComposeMessages(ctx context.Context, req *dto.ComposeMessagesReq) (*dto.ComposeMessagesRes, error) { - //1) 获取模型信息 + // 1) 获取模型信息 chatModel, aiModel, err := GetModelMessage(ctx, req) if err != nil { return nil, err } - //2) 校验用户表单 + // 2) 校验用户表单 if err = validateUserForm(req, aiModel); err != nil { return nil, err } - //3) 处理不同类型 - switch req.BuildType { - case public.BuildTypePrompt: - return handlePromptBuild(ctx, req, chatModel, aiModel) // 提示词构建 - case public.BuildTypeNode: - return handleNodeBuild(ctx, req, chatModel, aiModel) // 节点构建 - default: - return nil, errors.New("BuildType 不支持") - } + return handleBuild(ctx, req, chatModel, aiModel) } // GetModelMessage 获取模型信息 @@ -51,24 +42,19 @@ func GetModelMessage(ctx context.Context, req *dto.ComposeMessagesReq) (*gateway SQLBaseDO: beans.SQLBaseDO{Creator: userInfo.UserName}, IsChatModel: 1, }) - if err != nil { - return nil, nil, err - } - if chatModel == nil { + if err != nil || chatModel == nil { return nil, nil, errors.New("当前没有对话模型，请添加") } - aiModels, err := gateway.GetModelConfig(ctx, &gateway.AsynchModel{ + + aiModel, err := gateway.GetModelConfig(ctx, &gateway.AsynchModel{ SQLBaseDO: beans.SQLBaseDO{TenantId: userInfo.TenantId, Creator: userInfo.UserName}, ModelName: req.ModelName, }) - if err != nil { - return nil, nil, err - } - - if aiModels == nil { + if err != nil || aiModel == nil { return nil, nil, errors.New("需要构建的模型不存在") } - return chatModel, aiModels, nil + + return chatModel, aiModel, nil } // validateUserForm 校验用户表单 @@ -89,103 +75,96 @@ func validateUserForm(req *dto.ComposeMessagesReq, model *gateway.AsynchModel) e return nil } -// handlePromptBuild 处理提示词构建（BuildType=1） -func handlePromptBuild(ctx context.Context, req *dto.ComposeMessagesReq, chatModel, aiModel *gateway.AsynchModel) (*dto.ComposeMessagesRes, error) { - // 获取历史会话 - history, err := session.GetHistoryMessages(ctx, req.SessionId) +// handleBuild 通用构建处理 +func handleBuild(ctx context.Context, req *dto.ComposeMessagesReq, chatModel, aiModel *gateway.AsynchModel) (*dto.ComposeMessagesRes, error) { + // 1) 处理表单分批 + processedReq, totalBatches, err := ProcessUserFormBatches(ctx, req, aiModel) if err != nil { - g.Log().Errorf(ctx, "获取历史会话失败: %v，将不使用历史会话", err) - history = nil + return nil, fmt.Errorf("处理用户表单分批失败: %w", err) } - // 调用推理模型 - taskID, id, err := callInferenceModel(ctx, req, chatModel, aiModel, history) - if err != nil { - return nil, fmt.Errorf("调用推理模型失败: %w", err) - } - // 保存任务记录 - _, err = dao.ComposeTask.Insert(ctx, &entity.ComposeTask{ - TaskId: taskID, - ModelName: req.ModelName, - SkillName: req.SkillName, - BuildType: req.BuildType, - CallbackUrl: req.CallbackUrl, - RequestPayload: util.MustMarshalToMap(req), - Status: public.ComposeStatusPending, - }) - if err != nil { - return nil, fmt.Errorf("保存任务记录失败: %w", err) + // 2) 构建推理请求 + ir := NewPromptIR() + var taskReq map[string]any + switch req.BuildType { + case public.BuildTypePrompt: + taskReq, err = buildPromptTypeRequest(ctx, processedReq, aiModel, chatModel, ir, totalBatches) + case public.BuildTypeNode: + taskReq, err = buildNodeTypeRequest(ctx, req, chatModel, ir) + case public.BuildTypeStruct: + taskReq, err = buildStructTypeRequest(ctx, req, chatModel, ir) + default: + return nil, errors.New("不支持的构建类型") + } + if err != nil { + return nil, fmt.Errorf("构建推理请求失败: %w", err) } - return &dto.ComposeMessagesRes{ - TaskId: taskID, - EpicycleId: id, - }, nil -} -// handleNodeBuild 处理节点构建（BuildType=2） -func handleNodeBuild(ctx context.Context, req *dto.ComposeMessagesReq, chatModel, aiModel *gateway.AsynchModel) (*dto.ComposeMessagesRes, error) { - taskID, id, err := callInferenceModel(ctx, req, chatModel, aiModel, nil) - if err != nil { - return nil, fmt.Errorf("调用推理模型失败: %w", err) - } - // 保存任务记录 - _, err = dao.ComposeTask.Insert(ctx, &entity.ComposeTask{ - TaskId: taskID, - ModelName: req.ModelName, - SkillName: req.SkillName, - BuildType: req.BuildType, - CallbackUrl: req.CallbackUrl, - RequestPayload: util.MustMarshalToMap(req), - Status: public.ComposeStatusPending, - }) - if err != nil { - return nil, fmt.Errorf("保存任务记录失败: %w", err) - } - return &dto.ComposeMessagesRes{ - TaskId: taskID, - EpicycleId: id, - }, nil -} - -// callInferenceModel 调用推理模型 -func callInferenceModel(ctx context.Context, req *dto.ComposeMessagesReq, chatModel *gateway.AsynchModel, aiModel *gateway.AsynchModel, history []map[string]any) (string, int64, error) { - taskReq, err := buildInferenceRequest(ctx, req, chatModel, aiModel, history) - if err != nil { - return "", 0, fmt.Errorf("构建推理请求失败: %w", err) - } - id := int64(0) - if req.SessionId != "" { - id, err = dao.ComposeSession.Insert(ctx, &entity.ComposeSession{ - SessionId: req.SessionId, - RequestContent: util.GetUserMessage(taskReq), - }) - if err != nil { - return "", 0, fmt.Errorf("保存历史会话失败: %w", err) - } - } + // 3) 调用网关创建任务 taskID, err := gateway.CreateGatewayTask(ctx, taskReq) if err != nil { - return "", 0, fmt.Errorf("创建网关任务失败: %w", err) + return nil, fmt.Errorf("创建网关任务失败: %w", err) } - if taskID == "" { - return "", 0, errors.New("网关未返回taskId") + return nil, errors.New("网关未返回taskId") } - return taskID, id, nil + // 4) 保存任务记录 + if _, err = dao.ComposeTask.Insert(ctx, &entity.ComposeTask{ + TaskId: taskID, + ModelName: req.ModelName, + SkillName: req.SkillName, + BuildType: req.BuildType, + CallbackUrl: req.CallbackUrl, + RequestPayload: util.MustMarshalToMap(req), + Status: public.ComposeStatusPending, + }); err != nil { + return nil, err + } + return &dto.ComposeMessagesRes{TaskId: taskID}, nil } // Callback 回调处理 func Callback(ctx context.Context, req *dto.CallbackReq) error { - g.Log().Infof(ctx, "[Callback][RECV] taskId=%s state=%d ossFile=%s fileType=%s textLen=%d", - req.TaskId, req.State, req.OssFile, req.FileType, len(req.Messages)) - // 查询任务 - composeTask, err := dao.ComposeTask.Get(ctx, &entity.ComposeTask{ - TaskId: req.TaskId, - }) + g.Log().Infof(ctx, "[开始回调处理] taskId=%s state=%d", req.TaskId, req.State) + // 1) 查询任务 + composeTask, err := dao.ComposeTask.Get(ctx, &entity.ComposeTask{TaskId: req.TaskId}) if err != nil { return fmt.Errorf("查询任务失败: %w", err) } + // 2) 处理失败 + if req.State == 3 { + return handleCallbackFailed(ctx, req, composeTask) + } + // 3) 处理成功 + if req.State == 2 { + return handleCallbackSuccess(ctx, req, composeTask) + } + return nil +} + +// handleCallbackFailed 处理回调失败 +func handleCallbackFailed(ctx context.Context, req *dto.CallbackReq, composeTask *entity.ComposeTask) error { + _, err := dao.ComposeTask.Update(ctx, &entity.ComposeTask{ + TaskId: req.TaskId, + Status: public.ComposeStatusFailed, + ErrorMessage: req.ErrorMsg, + GatewayState: req.State, + OssFile: req.OssFile, + FileType: req.FileType, + ResultText: req.Messages, + }) + if composeTask.CallbackUrl != "" { + composeTask.Status = public.ComposeStatusFailed + composeTask.ErrorMessage = req.ErrorMsg + _ = gateway.SendCallback(ctx, composeTask) + } + return err +} + +// handleCallbackSuccess 处理回调成功 +func handleCallbackSuccess(ctx context.Context, req *dto.CallbackReq, composeTask *entity.ComposeTask) error { + // 1) 查模型配置 model, err := gateway.GetModelConfig(ctx, &gateway.AsynchModel{ SQLBaseDO: beans.SQLBaseDO{Creator: composeTask.Creator}, ModelName: composeTask.ModelName, @@ -193,75 +172,55 @@ func Callback(ctx context.Context, req *dto.CallbackReq) error { if err != nil { return fmt.Errorf("查询模型失败: %w", err) } - //处理失败 - if req.State == 3 { - _, err = dao.ComposeTask.Update(ctx, &entity.ComposeTask{ - TaskId: req.TaskId, - Status: public.ComposeStatusFailed, - ErrorMessage: req.ErrorMsg, - GatewayState: req.State, - OssFile: req.OssFile, - FileType: req.FileType, - ResultText: req.Messages, - }) - // 用更新后的值发送回调 - if composeTask.CallbackUrl != "" { - failedTask := &entity.ComposeTask{ - TaskId: req.TaskId, - Status: public.ComposeStatusFailed, - ErrorMessage: req.ErrorMsg, - CallbackUrl: composeTask.CallbackUrl, - Messages: composeTask.Messages, - } - gateway.SendCallback(ctx, failedTask) + + // 2) 解析结果 + var messages map[string]any + switch composeTask.BuildType { + case public.BuildTypePrompt, public.BuildTypeNode: + messages = ParseResult(req.Messages, model.ResponseBody) + case public.BuildTypeStruct: + messages = map[string]any{ + "total_rounds": 1, + "rounds": []map[string]any{req.Messages}, } + default: + messages = req.Messages + } + // 3) 合并附加结构 + messages = util.MergeConsult(composeTask.RequestPayload, messages, model.ExtendMapping) + // 4) 更新数据库 + _, err = dao.ComposeTask.Update(ctx, &entity.ComposeTask{ + TaskId: req.TaskId, + Status: public.ComposeStatusSuccess, + Messages: messages, + GatewayState: req.State, + OssFile: req.OssFile, + FileType: req.FileType, + ResultText: req.Messages, + }) + if err != nil { return err } - //处理成功 - if req.State == 2 { - // 1. 根据 BuildType 解析结果 - var messages map[string]any - switch composeTask.BuildType { - case public.BuildTypePrompt: // 提示词构建解析 - messages = ParsePromptResult(req.Messages) - case public.BuildTypeNode: // 节点构建解析 - messages = ParseNodeResult(req.Messages) - default: - messages = req.Messages - } - // 2. 处理附加字段 - messages = util.MergeConsult(composeTask.RequestPayload, messages, model.ExtendMapping) - // 3. 更新数据库 - _, err = dao.ComposeTask.Update(ctx, &entity.ComposeTask{ - TaskId: req.TaskId, - Status: public.ComposeStatusSuccess, - Messages: messages, - GatewayState: req.State, - OssFile: req.OssFile, - FileType: req.FileType, - ResultText: req.Messages, - }) - if err != nil { - g.Log().Errorf(ctx, "[Callback] 更新成功状态失败 taskId=%s err=%v", req.TaskId, err) - return err - } - // 4. 发送回调给业务方 - if composeTask.CallbackUrl != "" { - successTask := &entity.ComposeTask{ - TaskId: req.TaskId, - Status: public.ComposeStatusSuccess, - Messages: messages, - CallbackUrl: composeTask.CallbackUrl, - } - gateway.SendCallback(ctx, successTask) - } + // 5) 存储历史结果 + + // 6) 回调业务方 + if composeTask.CallbackUrl != "" { + composeTask.Status = public.ComposeStatusSuccess + composeTask.Messages = messages + _ = gateway.SendCallback(ctx, composeTask) } - return err + return nil } -// ParsePromptResult 解析提示词构建结果 -func ParsePromptResult(raw map[string]any) map[string]any { - contentStr, ok := raw["content"].(string) +// ParseResult 解析回调结果 +func ParseResult(raw map[string]any, responseBody string) map[string]any { + // responseBody 为空，直接返回原始数据 + if responseBody == "" { + return raw + } + + // 按 responseBody 路径取值 + contentStr, ok := raw[responseBody].(string) if !ok || contentStr == "" { return raw } diff --git a/service/prompt/prompt_ir_service.go b/service/prompt/prompt_ir_service.go index 7bdac56..a0eaf8a 100644 --- a/service/prompt/prompt_ir_service.go +++ b/service/prompt/prompt_ir_service.go @@ -286,7 +286,7 @@ func renderTemplate(tmpl map[string]any, messages []map[string]any, chatModel *g str = strings.ReplaceAll(str, `"{{messages}}"`, string(msgBytes)) var result map[string]any - json.Unmarshal([]byte(str), &result) + _ = json.Unmarshal([]byte(str), &result) return result }