diff --git a/config.yml b/config.yml index 587080c..f30876e 100644 --- a/config.yml +++ b/config.yml @@ -1,5 +1,5 @@ server: - address: ":8002" + address: ":3005" name: "prompts-core" workerId: 1 # 雪花算法 worker ID（用于 common/db/gfdb） @@ -39,7 +39,6 @@ jaeger: addr: 116.204.74.41:4318 task: - modelKey: "" # 动态请求头；会透传给 model-gateway，用于覆盖/补充模型配置中的 head_msg，没有可留空 waitTimeoutSeconds: 300 # /composeMessages 同步等待最终结果的最长时间（秒） pollIntervalMillis: 500 # 同步等待期间，轮询本地任务表 / 网关状态的时间间隔（毫秒） @@ -47,6 +46,19 @@ session: maxRounds: 10 # 最大轮数 expireTime: 1800 # 过期时间（秒），30分钟 +# 文件处理配置 +userFiles: + zipMaxSizeMB: 10 # zip 下载最大大小（MB） + zipEntryMaxSizeKB: 500 # zip 内单文件最大读取大小（KB） + textFileMaxSizeKB: 500 # 普通文本文件最大读取大小（KB） + httpTimeoutSec: 8 # HTTP 请求超时（秒） +skillFiles: + httpTimeoutSec: 500 # zip 下载超时（秒） + zipMaxSizeMB: 10 # zip 最大下载大小（MB） + mdMaxSizeKB: 5000 # 单个 md 文件最大读取大小（KB） +promptsRetry: + maxRetryTimes: 3 + modelPrompts: types: 1: | @@ -63,7 +75,6 @@ modelPrompts: 当用户提供具体音频需求时，需要结合音色、语速、语言类型、情绪风格、背景环境、输出格式等参数完成对应处理；若涉及语音识别或音频分析，则需要尽可能保留原始语义与声音特征，并明确标注不确定内容。 4: | 你是一个智能向量化处理助手，专注于文本向量化、语义检索、知识索引、相似度计算与语义聚类任务，能够将文本内容转换为高维语义向量，并基于向量相似度完成语义搜索、知识召回、内容聚类、文档匹配与知识库构建等处理流程。 - 在执行向量化任务时，你需要以语义检索工程师、知识库架构师、AI检索系统专家的身份进行处理，重点保证语义表达准确性、向量一致性、检索稳定性与召回有效性，同时确保不同文本之间的语义关系能够被正确表达与计算。 当用户提供文本集合、知识内容或检索需求时，需要结合文本上下文、主题方向、检索目标、相似度要求与业务场景生成最终结果；若涉及聚类或知识库构建，则必须明确类别关系、索引结构与召回逻辑。 5: | @@ -71,53 +82,55 @@ modelPrompts: 在执行多模态任务时，你需要以全链路AI内容架构师、多模态交互专家、综合内容生成系统的身份完成处理，重点保证不同模态之间的语义一致性、风格统一性、信息完整性与交互连贯性，避免出现跨模态语义断裂或输出不一致的问题。 当用户提供混合输入内容时，需要结合文本、图片、音频、视频等多种信息共同分析用户真实目标，并根据任务场景自动决定最终输出形式；若涉及跨模态生成，则必须保证生成结果能够准确映射原始语义与核心信息。 -projectPrompts: | - 你是【语义理解 + 结构对齐】的JSON生成专家，必须严格遵守以下所有规则。 - 【强制阅读规则 · 必须100%%遵守】 - 1. 必须完整通读全部文本、上下文、规则、表单内容，严禁跳读、略读； - 2. 未读完全部信息前，禁止输出任何内容； - 3. 必须全覆盖所有约束、所有细节、所有字段后再推理； - 4. 禁止断章取义，禁止遗漏任何参数； - 5. 必须严格区分系统表单、用户表单。 - - 【绝对优先级规则 · 最高优先级】 - 1. 用户自然语言描述 > 用户表单(UserForm) > 系统表单(Form) - 2. 高优先级内容 100%% 覆盖低优先级内容，禁止冲突 - 3. 所有参数最终以【优先级最高】的描述为准 - - 【核心语义规则】 - 1. Form = 系统提示词、系统参数、默认配置 - 2. UserForm = 用户真实输入全文，所有字段都必须作为用户提示词来源 - 3. 若 UserForm 字段与 Form 含义相同 → UserForm 严格覆盖 Form - 4. 必须完整使用 UserForm 所有内容，不得遗漏任何一个字段 - - 【数量计算规则 · 必须严格遵守 · 带示例】 - 1. 文案与图片为【一一对应关系】：1条文案对应1张图 - 2. 示例1：2条文案，各配1张图 → 最终数量number=2 - 3. 示例2：3条文案，各配1张图 → 最终数量number=3 - 4. 禁止错误计算：2条文案×2张图=4（这种逻辑严禁出现） - 5. 最终number = 文案总条数 = 图片总张数（一一对应） - 6. 自动补全默认值：size=1024*1024（为空时必须填充） - - 【任务】 - 根据双表单内容，智能填充JSON结构： - 1. 理解意图：图片/文案 - 2. 按【优先级规则】确定最终参数 - 3. 按【数量一一对应规则】计算number - 4. 严格按结构输出，不修改字段 - - 【输出结构】 - %s - 【字段映射关系】 - %s - 【完整输入信息】 - %s - - 【输出铁律】 - 1. 只输出单行JSON，无任何多余字符 - 2. 禁止换行、禁止转义、禁止解释 - 3. 内容准确、无废话、不编造 - 4. 必须完整读取 UserForm 全部内容 - 5. 严格遵守【优先级】和【数量计算规则】 - - 请输出最终JSON： \ No newline at end of file +buildProject: + types: + 1: | + 你是专业的JSON结构生成专家，必须严格遵守以下全部规则。 + 【强制规则】 + 必须根据【输出结构】里面返回的JSON结构进行生成，不得任何更改，最终内容与输出结构返回一致； + 完整阅读所有文本、规则、表单内容，禁止跳读、漏读； + 完整读取UserForm所有字段，不得忽略任何字段； + 如果有skill相关内容必须完整的将内容拼接到system角色描述中； + 理解全部语义后再输出，禁止断章取义； + UserForm所有字段内容必须完整拼接赋值到user角色描述中，不得有任何遗漏。 + 【优先级】 + 用户自然语言 > UserForm > Form； + UserForm与Form同名字段时，仅保留UserForm值； + Form仅用于组装system角色内容。 + 【表单处理】 + Form：系统提示词、默认参数、基础配置 → 专属填充system角色； + UserForm：用户业务输入、文案、配图数量、比例、prompt等 → 全部解析后拼接进user角色content； + 自动提取UserForm中每条文案的配图数量，总图片数 = 各文案配图数累加求和（示例：10条文案各配5张图 → 总50张，parameters.n=50）,用户没有相关数量必须默认1； + 图片尺寸为空时自动填充size=1024*1024。 + 【结构铁律】 + 严格沿用固定输出结构，不增删字段或修改层级； + messages元素必须按结构返回； + 禁止将role对象转为字符串、禁止嵌套错乱； + 输出纯净JSON：无多余转义符、无换行符、无额外字符； + 所有括号、引号必须成对闭合，保证JSON合法。 + 【参数赋值】 + model固定沿用传入值； + 返回结构里面的参数，需要根据语意进行赋值，缺失补默认值； + history历史信息必须结合UserForm里的内容对用户描述部分进行补充； + 从UserForm提取信息整合进user描述，确保数量、尺寸、文案语义无遗漏。 + 【输出要求】 + 仅输出单行纯净JSON，无任何解释、备注、Markdown或多余符号； + 完整合UserForm全部字段语义到user描述； + 生成后自检JSON语法、结构、数量；错误则自动重新生成。 + 【输出结构】 + %s + 【字段映射】 + %s + 【完整输入信息】 + %s + 直接输出最终JSON： + 2: | + 你是流程路由助手，你的任务是根据上下文，选择一个正确的节点ID返回。 + 规则： + 1. 只允许从下面的可选节点ID列表中选择一个返回 + 2. 不要返回任何多余文字、标点、解释、标题 + 3. 只返回纯节点ID + 可选节点ID（ID: 节点描述）： + %s + 上下文内容： + %s \ No newline at end of file diff --git a/dao/compose_session_dao.go b/dao/compose_session_dao.go index f3b8737..8977603 100644 --- a/dao/compose_session_dao.go +++ b/dao/compose_session_dao.go @@ -21,41 +21,11 @@ func (d *composeSessionDao) Insert(ctx context.Context, m *entity.ComposeSession return r.LastInsertId() } -func (d *composeSessionDao) GetById(ctx context.Context, Id int64) (m *entity.ComposeSession, err error) { +func (d *composeSessionDao) Update(ctx context.Context, m *entity.ComposeSession) (rows int64, err error) { r, err := gfdb.DB(ctx).Model(ctx, public.TableNameComposeSession). - Where("deleted_at IS NULL"). - Where(entity.ComposeSessionCol.Id, Id). - One() - if err != nil { - return nil, err - } - if r.IsEmpty() { - return nil, nil - } - err = r.Struct(&m) - return -} - -func (d *composeSessionDao) GetBySessionId(ctx context.Context, sessionId string) (m *entity.ComposeSession, err error) { - r, err := gfdb.DB(ctx).Model(ctx, public.TableNameComposeSession). - Where("deleted_at IS NULL"). - Where(entity.ComposeSessionCol.SessionId, sessionId). - One() - if err != nil { - return nil, err - } - if r.IsEmpty() { - return nil, nil - } - err = r.Struct(&m) - return -} - -func (d *composeSessionDao) UpdateById(ctx context.Context, id int64, data map[string]any) (rows int64, err error) { - data[entity.ComposeSessionCol.Updater] = "" - r, err := gfdb.DB(ctx).Model(ctx, public.TableNameComposeSession). - Where(entity.ComposeSessionCol.Id, id). - Data(data). + Where(entity.ComposeSessionCol.Id, m.Id). + Data(m). + OmitEmpty(). Update() if err != nil { return 0, err @@ -86,6 +56,52 @@ func (d *composeSessionDao) List(ctx context.Context, page, size int, where map[ return } +func (d *composeSessionDao) GetListBySessionId(ctx context.Context, sessionId string, limit int) ([]*entity.ComposeSession, error) { + var sessions []*entity.ComposeSession + err := gfdb.DB(ctx).Model(ctx, public.TableNameComposeSession). + Where(entity.ComposeSessionCol.SessionId, sessionId). + WhereNull(entity.ComposeSessionCol.DeletedAt). + OrderDesc(entity.ComposeSessionCol.Id). + Limit(limit). + Scan(&sessions) + if err != nil { + return nil, err + } + // 反转成时间正序 + for i, j := 0, len(sessions)-1; i < j; i, j = i+1, j-1 { + sessions[i], sessions[j] = sessions[j], sessions[i] + } + return sessions, nil +} + +func (d *composeSessionDao) GetById(ctx context.Context, Id int64) (m *entity.ComposeSession, err error) { + r, err := gfdb.DB(ctx).Model(ctx, public.TableNameComposeSession). + Where(entity.ComposeSessionCol.Id, Id). + One() + if err != nil { + return nil, err + } + if r.IsEmpty() { + return nil, nil + } + err = r.Struct(&m) + return +} + +func (d *composeSessionDao) GetBySessionId(ctx context.Context, sessionId string) (m *entity.ComposeSession, err error) { + r, err := gfdb.DB(ctx).Model(ctx, public.TableNameComposeSession). + Where(entity.ComposeSessionCol.SessionId, sessionId). + One() + if err != nil { + return nil, err + } + if r.IsEmpty() { + return nil, nil + } + err = r.Struct(&m) + return +} + func (d *composeSessionDao) DeleteBySessionId(ctx context.Context, sessionId string) (rows int64, err error) { r, err := gfdb.DB(ctx).Model(ctx, public.TableNameComposeSession). Where(entity.ComposeSessionCol.SessionId, sessionId). diff --git a/dao/compose_task_dao.go b/dao/compose_task_dao.go index 79c6347..a52101f 100644 --- a/dao/compose_task_dao.go +++ b/dao/compose_task_dao.go @@ -23,7 +23,6 @@ func (d *composeTaskDao) Insert(ctx context.Context, m *entity.ComposeTask) (id func (d *composeTaskDao) GetByTaskId(ctx context.Context, taskId string) (m *entity.ComposeTask, err error) { r, err := gfdb.DB(ctx).Model(ctx, public.TableNameComposeTask). - Where("deleted_at IS NULL"). Where(entity.ComposeTaskCol.TaskId, taskId). One() if err != nil { diff --git a/dao/model_dao.go b/dao/model_dao.go index de10541..6c8da8c 100644 --- a/dao/model_dao.go +++ b/dao/model_dao.go @@ -7,6 +7,7 @@ import ( "prompts-core/model/entity" "gitea.com/red-future/common/db/gfdb" + "gitea.com/red-future/common/utils" ) var Model = &modelDao{} @@ -28,8 +29,13 @@ func (d *modelDao) GetByModelName(ctx context.Context, modelName string) (m *ent } func (d *modelDao) GetByIsChatModel(ctx context.Context) (m *entity.AsynchModel, err error) { + userInfo, err := utils.GetUserInfo(ctx) + if err != nil { + return nil, err + } r, err := gfdb.DB(ctx).Model(ctx, public.TableNameModel). Where(entity.AsynchModelCol.IsChatModel, 1). + Where(entity.AsynchModelCol.Creator, userInfo.UserName). One() if err != nil { return nil, err diff --git a/model/dto/compose_messages_dto.go b/model/dto/compose_messages_dto.go index a1cd11a..2288b43 100644 --- a/model/dto/compose_messages_dto.go +++ b/model/dto/compose_messages_dto.go @@ -10,8 +10,8 @@ type Message struct { 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" v:"required#sessionId不能为空" dc:"会话ID"` - IsBuilder bool `p:"isBuild" json:"isBuild" v:"required#isBuild不能为空" dc:"是否构建"` Cause string `p:"cause" json:"cause" v:"required-if:IsBuilder,false#原因不能为空" dc:"原因"` Form map[string]any `p:"form" json:"form" dc:"系统表单：form 下所有字段都作为系统提示词来源"` UserForm map[string]any `p:"userForm" json:"userForm" dc:"用户表单：userForm 下所有字段都作为用户提示词来源；若与 form 含义接近则严格覆盖系统字段"` diff --git a/model/entity/asynch_model.go b/model/entity/asynch_model.go index fdca743..9714cb9 100644 --- a/model/entity/asynch_model.go +++ b/model/entity/asynch_model.go @@ -5,7 +5,7 @@ import "gitea.com/red-future/common/beans" type asynchModelCol struct { beans.SQLBaseCol ModelName string - ModelsType string + ModelType string BaseURL string HttpMethod string HeadMsg string @@ -32,7 +32,7 @@ type asynchModelCol struct { var AsynchModelCol = asynchModelCol{ SQLBaseCol: beans.DefSQLBaseCol, ModelName: "model_name", - ModelsType: "models_type", + ModelType: "model_type", BaseURL: "base_url", HttpMethod: "http_method", HeadMsg: "head_msg", @@ -60,7 +60,7 @@ var AsynchModelCol = asynchModelCol{ type AsynchModel struct { beans.SQLBaseDO `orm:",inline"` ModelName string `orm:"model_name" json:"modelName"` - ModelsType int `orm:"models_type" json:"modelsType"` + ModelType int `orm:"model_type" json:"modelType"` BaseURL string `orm:"base_url" json:"baseUrl"` HttpMethod string `orm:"http_method" json:"httpMethod"` HeadMsg string `orm:"head_msg" json:"headMsg"` diff --git a/service/build_service.go b/service/build_service.go index 9bd7647..87ac326 100644 --- a/service/build_service.go +++ b/service/build_service.go @@ -3,6 +3,7 @@ package service import ( "context" "encoding/json" + "errors" "fmt" "prompts-core/model/dto" "prompts-core/model/entity" @@ -12,45 +13,61 @@ import ( "github.com/gogf/gf/v2/util/gconv" ) -// 获取系统模型提示词 -func getConfPrompt(ctx context.Context, modelType int) string { - return g.Cfg().MustGet(ctx, "modelPrompts.types."+gconv.String(modelType), "").String() +// 获取请求模型的提示词 +func GetModelPrompt(ctx context.Context, Type int) string { + return g.Cfg().MustGet(ctx, "modelPrompts.types."+gconv.String(Type), "").String() } -func buildInferenceRequest(ctx context.Context, req *dto.ComposeMessagesReq, sessionModel *entity.AsynchModel, model *entity.AsynchModel, historyMessages []Message) (map[string]any, error) { - // 读取 task 相关的配置 - // 构建消息数组 - // 1. 系统提示词（不动） - - fmt.Println("打印sessionModel结果", sessionModel) - fmt.Println("打印model结果", model) +// 获取构建提示词 +func GetBuildPrompt(ctx context.Context, Type int) string { + return g.Cfg().MustGet(ctx, "buildProject.types."+gconv.String(Type), "").String() +} +// buildInferenceRequest 构建返回请求 +func buildInferenceRequest(ctx context.Context, req *dto.ComposeMessagesReq, chatModel *entity.AsynchModel, model *entity.AsynchModel, history []map[string]any) (map[string]any, error) { messages := []map[string]any{} - messages = append(messages, map[string]any{ - "role": "system", - "content": GetSystemPrompt(ctx, req, model), - }) - - // 2. 历史对话 - 动态添加（新增部分） - for _, msg := range historyMessages { + switch req.BuildType { + //构建提示词请求 + case 1: + //1. 构建系统提示词 messages = append(messages, map[string]any{ - "role": msg.Role, - "content": msg.GetContentString(), + "role": "system", + "content": promptBuild(ctx, req, model), }) - } - // 3. 当前用户问题（原来的最后一条） - messages = append(messages, map[string]any{ - "role": "user", - "content": buildCombinedInput(req, getConfPrompt(ctx, model.ModelsType)), - }) + // 2. 构建历史会话提示词 + for _, msg := range history { + role := gconv.String(msg["role"]) + content := gconv.String(msg["content"]) + if role != "user" && role != "assistant" { + continue + } + messages = append(messages, map[string]any{ + "role": role, + "content": content, + }) + } + // 3. 当前用户问题（原来的最后一条） + messages = append(messages, map[string]any{ + "role": "user", + "content": buildUserPrompt(ctx, req, GetModelPrompt(ctx, model.ModelType)), + }) + //构建节点请求 + case 2: + messages = append(messages, map[string]any{ + "role": "user", + "content": NodeBuid(ctx, req), + }) + default: + return nil, errors.New("不支持的构建类型") + } // 构建请求体 return map[string]any{ - "modelName": sessionModel.ModelName, + "modelName": chatModel.ModelName, "bizName": "prompts-core", "callbackUrl": "/prompt/callback", "requestPayload": map[string]any{ - "model": sessionModel.ModelName, + "model": chatModel.ModelName, "messages": messages, "stream": false, }, @@ -58,10 +75,11 @@ func buildInferenceRequest(ctx context.Context, req *dto.ComposeMessagesReq, ses } // ============================================ -// 输入构建 +// 构建用户提示词 // ============================================ -func buildCombinedInput(req *dto.ComposeMessagesReq, prompt string) string { +func buildUserPrompt(ctx context.Context, req *dto.ComposeMessagesReq, prompt string) string { payload := map[string]any{ + "model": req.ModelName, //数据库提示信息 "promptInfo": prompt, // 系统表单 @@ -71,20 +89,20 @@ func buildCombinedInput(req *dto.ComposeMessagesReq, prompt string) string { //文件url "userFiles": req.UserFiles, //解读文件（只支持可读类型 如：xml，json,yaml） - "userFilesText": fetchFileTexts(context.Background(), req.UserFiles), + "userFilesText": FetchFileTexts(ctx, req.UserFiles), + //skill 相关（根据传入的 skillName 获取 zip 内所有 md 文件拼接内容） + "skills": SkillMdContent(ctx, req.SkillName), } return mustMarshal(payload) } -// GetSystemPrompt 定义获取系统提示词的函数 -// GetSystemPrompt 从配置文件读取提示词 + 格式化变量 -func GetSystemPrompt(ctx context.Context, req *dto.ComposeMessagesReq, model *entity.AsynchModel) string { +// promptBuild 提示词构建 +func promptBuild(ctx context.Context, req *dto.ComposeMessagesReq, model *entity.AsynchModel) string { // 1. 从配置文件读取提示词模板 - promptTpl := g.Cfg().MustGet(ctx, "projectPrompts", "").String() + promptTpl := GetBuildPrompt(ctx, req.BuildType) if promptTpl == "" { return "" } - // 2. 构建字段映射说明 mappingBytes, _ := json.Marshal(model.RequestMapping) mappingStr := string(mappingBytes) @@ -115,11 +133,13 @@ func GetSystemPrompt(ctx context.Context, req *dto.ComposeMessagesReq, model *en return fmt.Sprintf(promptTpl, mappingStr, fieldDesc.String(), formInfo) } -// formToJSON 工具函数不变 -func formToJSON(form map[string]any) string { - if form == nil { - return "{}" +// NodeBuid 节点构建 +func NodeBuid(ctx context.Context, req *dto.ComposeMessagesReq) string { + promptTpl := GetBuildPrompt(ctx, req.BuildType) + if promptTpl == "" { + return "" } - b, _ := json.Marshal(form) - return string(b) + formStr := formToJSON(req.Form) + userFormStr := formToJSON(req.UserForm) + return fmt.Sprintf(promptTpl, formStr, userFormStr) } diff --git a/service/compose_parser.go b/service/compose_parser.go deleted file mode 100644 index c5fec25..0000000 --- a/service/compose_parser.go +++ /dev/null @@ -1,362 +0,0 @@ -package service - -import ( - "encoding/json" - "errors" - "fmt" - "strings" - - "prompts-core/model/dto" -) - -// ============================================ -// 类型定义 -// ============================================ - -// modelOutput 推理模型的标准输出格式 -type modelOutput struct { - Messages []dto.Message `json:"messages"` - System any `json:"system"` - User any `json:"user"` -} - -// gatewayResponse 模型网关的标准响应格式 -type gatewayResponse struct { - Choices []choice `json:"choices"` -} - -type choice struct { - Message message `json:"message"` -} - -type message struct { - Content string `json:"content"` -} - -// ============================================ -// 核心解析函数 -// ============================================ - -// ParseModelResponse 解析推理模型的文本响应，返回消息列表 -// 支持三种格式： -// 1. 标准 messages 格式: {"messages": [...]} -// 2. 简化 system/user 格式: {"system": "...", "user": "..."} -// 3. 网关包装格式: {"choices": [{"message": {"content": "..."}}]} -func ParseModelResponse(text string) ([]dto.Message, error) { - text = strings.TrimSpace(text) - if text == "" { - return nil, errors.New("模型响应为空") - } - - // 1. 尝试解包网关响应 - if content := unwrapGatewayResponse(text); content != "" { - text = content - } - - // 2. 解析为标准格式 - output, err := parseAsModelOutput(text) - if err != nil { - return nil, fmt.Errorf("解析模型输出失败: %w", err) - } - - // 3. 优先使用 messages 字段 - if len(output.Messages) > 0 { - messages := normalizeMessageContents(output.Messages) - if err := validateMessageList(messages); err != nil { - return nil, err - } - return messages, nil - } - - // 4. 兼容 system/user 格式 - return buildMessagesFromSystemUser(output) -} - -// ParseStoredMessages 从数据库存储的数据中解析消息列表 -func ParseStoredMessages(data any) []dto.Message { - if data == nil { - return nil - } - - // 统一序列化 - jsonBytes, err := json.Marshal(data) - if err != nil { - return nil - } - - // 尝试直接解析 - var messages []dto.Message - if err := json.Unmarshal(jsonBytes, &messages); err == nil { - return messages - } - - // 尝试解析为 JSON 字符串再解析 - var jsonStr string - if err := json.Unmarshal(jsonBytes, &jsonStr); err != nil { - return nil - } - - if err := json.Unmarshal([]byte(jsonStr), &messages); err != nil { - return nil - } - - return messages -} - -// ============================================ -// 内部解析函数 -// ============================================ - -// parseAsModelOutput 将文本解析为 modelOutput 结构 -func parseAsModelOutput(text string) (*modelOutput, error) { - // 清理可能的 Markdown 代码块标记 - text = cleanMarkdownCodeBlock(text) - - var output modelOutput - if err := json.Unmarshal([]byte(text), &output); err != nil { - return nil, err - } - - return &output, nil -} - -// unwrapGatewayResponse 解包网关的标准响应格式 -func unwrapGatewayResponse(text string) string { - // 快速检查是否可能是网关响应 - if !strings.Contains(text, `"choices"`) { - return "" - } - - var resp gatewayResponse - if err := json.Unmarshal([]byte(text), &resp); err != nil { - return "" - } - - if len(resp.Choices) == 0 { - return "" - } - - content := strings.TrimSpace(resp.Choices[0].Message.Content) - return content -} - -// buildMessagesFromSystemUser 从 system/user 字段构建消息列表 -func buildMessagesFromSystemUser(output *modelOutput) ([]dto.Message, error) { - messages := make([]dto.Message, 0, 2) - - // 添加 user 消息 - if !isEmptyValue(output.User) { - messages = append(messages, dto.Message{ - Role: "user", - Content: normalizeContent(output.User), - }) - } - - // 添加 system 消息 - if !isEmptyValue(output.System) { - messages = append(messages, dto.Message{ - Role: "system", - Content: normalizeContent(output.System), - }) - } - - if len(messages) == 0 { - return nil, errors.New("未解析到有效的 system 或 user 内容") - } - - if err := validateMessageList(messages); err != nil { - return nil, err - } - - return messages, nil -} - -// ============================================ -// 内容规范化 -// ============================================ - -// normalizeMessageContents 规范化消息列表中的所有内容 -func normalizeMessageContents(messages []dto.Message) []dto.Message { - for i := range messages { - messages[i].Content = normalizeContent(messages[i].Content) - } - return messages -} - -// normalizeContent 规范化单个消息内容 -// - 如果是 JSON 字符串，尝试解析为对象/数组 -// - 否则保持原样 -func normalizeContent(content any) any { - switch v := content.(type) { - case string: - return tryUnmarshalJSON(v) - default: - return content - } -} - -// tryUnmarshalJSON 尝试将 JSON 字符串解析为结构化对象 -func tryUnmarshalJSON(s string) any { - s = strings.TrimSpace(s) - if s == "" { - return s - } - - // 只处理看起来像 JSON 的内容 - if !looksLikeJSON(s) { - return s - } - - var result any - if err := json.Unmarshal([]byte(s), &result); err != nil || result == nil { - return s - } - - return result -} - -// looksLikeJSON 判断字符串是否可能是 JSON -func looksLikeJSON(s string) bool { - s = strings.TrimSpace(s) - return strings.HasPrefix(s, "{") || strings.HasPrefix(s, "[") -} - -// cleanMarkdownCodeBlock 清理 Markdown 代码块标记 -func cleanMarkdownCodeBlock(text string) string { - // 去除可能的 ```json 和 ``` 标记 - text = strings.TrimPrefix(text, "```json") - text = strings.TrimPrefix(text, "```JSON") - text = strings.TrimPrefix(text, "```") - text = strings.TrimSuffix(text, "```") - return strings.TrimSpace(text) -} - -// ============================================ -// 验证 -// ============================================ - -// validateMessageList 验证消息列表的合法性 -func validateMessageList(messages []dto.Message) error { - if len(messages) == 0 { - return errors.New("消息列表不能为空") - } - - hasUser := false - for i, msg := range messages { - if err := validateMessage(msg); err != nil { - return fmt.Errorf("消息[%d]验证失败: %w", i, err) - } - if msg.Role == "user" { - hasUser = true - } - } - - // 至少需要一条 user 消息 - if !hasUser { - return errors.New("消息列表必须包含至少一条 user 角色消息") - } - - return nil -} - -// validateMessage 验证单条消息的合法性 -func validateMessage(msg dto.Message) error { - role := strings.TrimSpace(msg.Role) - if role == "" { - return errors.New("role 不能为空") - } - - if !isValidRole(role) { - return fmt.Errorf("role 值非法: %s (仅允许 system/user/assistant)", role) - } - - // user 角色的 content 不能为空 - if role == "user" && isEmptyValue(msg.Content) { - return errors.New("user 角色的 content 不能为空") - } - - return nil -} - -// isValidRole 判断角色是否合法 -func isValidRole(role string) bool { - switch role { - case "system", "user", "assistant": - return true - default: - return false - } -} - -// HasUserMessage 判断消息列表中是否包含非空的 user 消息 -func HasUserMessage(messages []dto.Message) bool { - for _, msg := range messages { - if msg.Role == "user" && !isEmptyValue(msg.Content) { - return true - } - } - return false -} - -// HasSystemMessage 判断消息列表中是否包含非空的 system 消息 -func HasSystemMessage(messages []dto.Message) bool { - for _, msg := range messages { - if msg.Role == "system" && !isEmptyValue(msg.Content) { - return true - } - } - return false -} - -// ExtractUserContent 提取消息列表中第一个 user 角色的内容 -func ExtractUserContent(messages []dto.Message) any { - for _, msg := range messages { - if msg.Role == "user" { - return msg.Content - } - } - return nil -} - -// ExtractSystemContent 提取消息列表中第一个 system 角色的内容 -func ExtractSystemContent(messages []dto.Message) any { - for _, msg := range messages { - if msg.Role == "system" { - return msg.Content - } - } - return nil -} - -// ============================================ -// 测试辅助函数 (可选) -// ============================================ - -// MockModelResponse 创建模拟的模型响应用于测试 -func MockModelResponse(systemContent, userContent string) string { - output := modelOutput{ - Messages: []dto.Message{ - {Role: "system", Content: systemContent}, - {Role: "user", Content: userContent}, - }, - } - - bytes, _ := json.Marshal(output) - return string(bytes) -} - -// MockGatewayResponse 创建模拟的网关响应用于测试 -func MockGatewayResponse(innerJSON string) string { - resp := gatewayResponse{ - Choices: []choice{ - { - Message: message{ - Content: innerJSON, - }, - }, - }, - } - - bytes, _ := json.Marshal(resp) - return string(bytes) -} diff --git a/service/compose_service.go b/service/compose_service.go index b9e943d..ba300cf 100644 --- a/service/compose_service.go +++ b/service/compose_service.go @@ -24,12 +24,98 @@ import ( // ComposeMessages 拼接提示词主流程 func (s *promptService) ComposeMessages(ctx context.Context, req *dto.ComposeMessagesReq) (*dto.ComposeMessagesRes, error) { var ( - epicycleId int64 - err error - historyMessages []Message // 用来存放历史会话 + epicycleId int64 + taskID string + history []map[string]any + message map[string]any + err error + taskRecord *entity.ComposeTask ) - // 1. 如果不需要构建返回记录id - if req.IsBuilder == false { + // 获取模型信息 + chatModel, model, err := s.GetModelMessage(ctx, req) + if err != nil { + return nil, err + } + // 根据构建类型进行判断处理 + switch req.BuildType { + //提示词构建 + case 1: + maxRetryTimes := g.Cfg().MustGet(ctx, "promptsRetry.maxRetryTimes", 3).Int() + //1. 获取历史会话 + history, err = Session.GetHistoryMessages(ctx, req.SessionId) + if err != nil { + g.Log().Errorf(ctx, "获取历史会话失败: %v，将不使用历史会话", err) + history = nil // 出错就用空的，不影响主流程 + } + // 重试循环 + for attempt := 0; attempt <= maxRetryTimes; attempt++ { + if attempt > 0 { + g.Log().Warningf(ctx, "[重试]第 %d/%d 次调用推理模型", attempt, maxRetryTimes) + } + // 2. 调用推理模型 + taskID, err = s.callInferenceModel(ctx, req, chatModel, model, history) + if err != nil { + g.Log().Errorf(ctx, "调用推理模型失败(第%d次): %v", attempt+1, err) + continue + } + + // 3. 保存记录 + _, err = dao.ComposeTask.Insert(ctx, &entity.ComposeTask{ + TaskId: taskID, + ModelName: req.ModelName, + SkillName: req.SkillName, + RequestPayload: mustMarshal(req), + Status: public.ComposeStatusPending, + }) + if err != nil { + g.Log().Errorf(ctx, "保存任务记录失败(第%d次): %v", attempt+1, err) + continue + } + + // 4. 等待结果 + taskRecord, err = s.waitForResult(ctx, taskID) + if err != nil { + g.Log().Errorf(ctx, "等待结果失败(第%d次): %v", attempt+1, err) + continue + } + // 校验结果 + message = s.parsePromptBuild(taskRecord, chatModel) + if message != nil && isMessageValid(message) { + break + } + g.Log().Warningf(ctx, "[重试] 推理结果不合法(第%d次)，准备重新请求", attempt+1) + message = nil + } + if message == nil { + return nil, errors.New("推理模型调用失败，请稍后再试") + } + //5.创建会话记录 + epicycleId, err = dao.ComposeSession.Insert(ctx, &entity.ComposeSession{ + SessionId: req.SessionId, + RequestContent: message, + }) + //节点构建 + case 2: + //1. 调用推理模型 + taskID, err = s.callInferenceModel(ctx, req, chatModel, model, nil) + if err != nil { + return nil, err + } + //2. 保存相关记录 + _, err = dao.ComposeTask.Insert(ctx, &entity.ComposeTask{ + TaskId: taskID, + ModelName: req.ModelName, + SkillName: req.SkillName, + RequestPayload: mustMarshal(req), + Status: public.ComposeStatusPending, + }) + //5. 等待结果 + taskRecord, err := s.waitForResult(ctx, taskID) + if err != nil { + return nil, err + } + message = s.parseNodeBuild(taskRecord) + default: epicycleId, err = dao.ComposeSession.Insert(ctx, &entity.ComposeSession{ SessionId: req.SessionId, Remark: req.Cause, @@ -38,80 +124,8 @@ func (s *promptService) ComposeMessages(ctx context.Context, req *dto.ComposeMes EpicycleId: epicycleId, }, nil } - - // 2. 获取当前用户模型信息 - sessionModel, err := dao.Model.GetByIsChatModel(ctx) //获取会话模型 - if err != nil { - return nil, err - } - if sessionModel == nil { - return nil, errors.New("当前没有对话模型，请添加") - } - model, err := dao.Model.GetByModelName(ctx, req.ModelName) //获取模型信息 - if err != nil { - return nil, err - } - if model == nil { - return nil, fmt.Errorf("模型 %s 不存在", sessionModel.ModelName) - } - - // 3 获取历史会话 - historyMessages, err = Session.GetSessionHistoryForInference(ctx, req.SessionId) - if err != nil { - g.Log().Errorf(ctx, "获取历史会话失败: %v，将不使用历史会话", err) - historyMessages = nil // 出错就用空的，不影响主流程 - } - - // 4. 调用推理模型 - taskID, err := s.callInferenceModel(ctx, req, sessionModel, model, historyMessages) - if err != nil { - return nil, err - } - - // 5. 保存相关记录 - _, err = dao.ComposeTask.Insert(ctx, &entity.ComposeTask{ - TaskId: taskID, - ModelName: req.ModelName, - SkillName: req.SkillName, - RequestPayload: mustMarshal(req), - Status: public.ComposeStatusPending, - }) - if err != nil { - return nil, err - } - - // 6. 等待结果 - taskRecord, err := s.waitForResult(ctx, taskID) - if err != nil { - return nil, err - } - - // 7. 处理返回结果 - messages := s.processResult(taskRecord) - - //8.1 数据库查询当前会话是否存在 - session, err := dao.ComposeSession.GetBySessionId(ctx, req.SessionId) - if err != nil { - return nil, err - } - if session == nil { - //8.2 不存在则创建新会话记录 - epicycleId, err = dao.ComposeSession.Insert(ctx, &entity.ComposeSession{ - SessionId: req.SessionId, - RequestContent: messages, - }) - if err != nil { - return nil, err - } - } - - // 9. 更新历史会话 - _, err = dao.ComposeSession.UpdateById(ctx, epicycleId, map[string]any{ - entity.ComposeSessionCol.RequestContent: messages, - }) - return &dto.ComposeMessagesRes{ - Messages: messages, + Messages: message, EpicycleId: epicycleId, }, nil } @@ -139,7 +153,7 @@ func (s *promptService) Callback(ctx context.Context, req *dto.CallbackReq) erro } // ====================================== // 成功：解析模型输出 - result, err := parseModelOutput(req.Text) + result, err := parseOutput(req.Text) if err != nil { _, updateErr := dao.ComposeTask.UpdateByTaskId(ctx, req.TaskId, map[string]any{ entity.ComposeTaskCol.Status: public.ComposeStatusFailed, @@ -195,13 +209,31 @@ func (s *promptService) GetComposeTask(ctx context.Context, taskID string) (*dto }, nil } -// ============================================ -// 步骤4：调用推理模型 -// ============================================ +// GetModelMessage 获取模型信息 +func (s *promptService) GetModelMessage(ctx context.Context, req *dto.ComposeMessagesReq) (*entity.AsynchModel, *entity.AsynchModel, error) { + // 1. 获取当前用户的会话模型 + chatModel, err := dao.Model.GetByIsChatModel(ctx) + if err != nil { + return nil, nil, err + } + if chatModel == nil { + return nil, nil, errors.New("当前没有对话模型，请添加") + } + // 2. 获取要构建的模型信息 + model, err := dao.Model.GetByModelName(ctx, req.ModelName) + if err != nil { + return nil, nil, err + } + if model == nil { + return nil, nil, fmt.Errorf("需要构建的模型 %s 不存在", req.ModelName) + } + return chatModel, model, nil +} -func (s *promptService) callInferenceModel(ctx context.Context, req *dto.ComposeMessagesReq, sessionModel *entity.AsynchModel, model *entity.AsynchModel, historyMessages []Message) (string, error) { +// callInferenceModel 调用推理模型 +func (s *promptService) callInferenceModel(ctx context.Context, req *dto.ComposeMessagesReq, chatModel *entity.AsynchModel, model *entity.AsynchModel, history []map[string]any) (string, error) { // 构建推理模型请求 - taskReq, err := buildInferenceRequest(ctx, req, sessionModel, model, historyMessages) + taskReq, err := buildInferenceRequest(ctx, req, chatModel, model, history) if err != nil { return "", fmt.Errorf("构建推理请求失败: %w", err) } @@ -222,16 +254,27 @@ func (s *promptService) callInferenceModel(ctx context.Context, req *dto.Compose // ============================================ // 步骤6：等待结果 // ============================================ - func (s *promptService) waitForResult(ctx context.Context, taskID string) (*entity.ComposeTask, error) { - timeout := time.Duration(getIntConfig(ctx, "task.waitTimeoutSeconds", 30)) * time.Second - pollInterval := time.Duration(getIntConfig(ctx, "task.pollIntervalMillis", 500)) * time.Millisecond + timeout := time.Duration(g.Cfg().MustGet(ctx, "task.waitTimeoutSeconds", 300).Int()) * time.Second + pollInterval := time.Duration(g.Cfg().MustGet(ctx, "task.pollIntervalMillis", 500).Int()) * time.Millisecond deadline := time.Now().Add(timeout) for { + // ===================== 修复点 1：检查上下文是否取消 ===================== + select { + case <-ctx.Done(): + // 请求已被取消，直接返回，不继续查库 + return nil, ctx.Err() + default: + } + // 1. 查数据库 record, err := dao.ComposeTask.GetByTaskId(ctx, taskID) if err != nil { + // ===================== 修复点 2：如果是上下文取消，直接返回 ===================== + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return nil, err + } return nil, err } if record != nil { @@ -239,19 +282,22 @@ func (s *promptService) waitForResult(ctx context.Context, taskID string) (*enti case public.ComposeStatusSuccess: return record, nil case public.ComposeStatusFailed: - return nil, formatTaskError(taskID, record.ErrorMessage) + if strings.TrimSpace(record.ErrorMessage) == "" { + return nil, fmt.Errorf("任务失败(taskId=%s)", taskID) + } + return nil, fmt.Errorf("任务失败(taskId=%s): %s", taskID, record.ErrorMessage) } } // 2. 查网关状态 state, err := queryGatewayTaskState(ctx, taskID) if err != nil { - // ============ 网关不可达不终止，继续轮询 ============ + // 网关不可达不终止，继续轮询 g.Log().Warningf(ctx, "[waitForResult] 查询网关失败 taskId=%s err=%v", taskID, err) } else { switch state { case 2: // 网关成功 - // ============ 网关已成功，主动更新数据库 ============ + // 网关已成功，主动更新数据库 if record != nil { dao.ComposeTask.UpdateByTaskId(ctx, taskID, map[string]any{ entity.ComposeTaskCol.Status: public.ComposeStatusSuccess, @@ -272,239 +318,97 @@ func (s *promptService) waitForResult(ctx context.Context, taskID string) (*enti if time.Now().After(deadline) { return nil, fmt.Errorf("等待任务回调超时(taskId=%s)", taskID) } - time.Sleep(pollInterval) + + // ===================== 修复点3：sleep 也要监听 ctx 取消 ===================== + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(pollInterval): + } } } -// ============================================ -// 步骤6：处理结果 -// ============================================ -func (s *promptService) processResult(taskRecord *entity.ComposeTask) map[string]any { +// parsePromptBuild 解析提示词构建结果（BuildType == 1） +func (s *promptService) parsePromptBuild(taskRecord *entity.ComposeTask, model *entity.AsynchModel) map[string]any { if taskRecord == nil { return nil } - // 1. 解析 Messages 获取 content - var contentStr string + // 1. 解析 Messages + var mapped map[string]any switch v := taskRecord.Messages.(type) { case *gvar.Var: if v != nil { - var mapped map[string]any json.Unmarshal([]byte(v.String()), &mapped) - if c, ok := mapped["content"].(string); ok { - contentStr = c - } } case string: - var mapped map[string]any json.Unmarshal([]byte(v), &mapped) - if c, ok := mapped["content"].(string); ok { - contentStr = c - } case map[string]any: - if c, ok := v["content"].(string); ok { - contentStr = c + mapped = v + default: + b, _ := json.Marshal(v) + json.Unmarshal(b, &mapped) + } + + // 2. 解析模型 ResponseMapping 获取 content 字段名 + contentField := "content" // 默认值 + if model != nil { + var respMapping map[string]string + switch v := model.ResponseMapping.(type) { + case *gvar.Var: + if v != nil { + json.Unmarshal([]byte(v.String()), &respMapping) + } + case string: + json.Unmarshal([]byte(v), &respMapping) + case map[string]interface{}: + respMapping = make(map[string]string) + for k, val := range v { + if s, ok := val.(string); ok { + respMapping[k] = s + } + } + } + // 从映射中找到 content 对应的字段名 + for k, v := range respMapping { + if strings.Contains(v, "content") { + contentField = k + break + } } } - // 2. 清理并解析 - contentStr = cleanJSONString(contentStr) + // 3. 提取 content 的值 + contentStr, ok := mapped[contentField].(string) + if !ok || contentStr == "" { + return mapped + } + + // 4. 解析 content 内的 JSON var innerData map[string]any json.Unmarshal([]byte(contentStr), &innerData) return innerData } -// ============================================ -// 消息处理管道 -// ============================================ - -// parseStoredMessages 从数据库存储的数据中解析消息列表 -// 处理多层 JSON 嵌套的情况 -func parseStoredMessages(data any) []dto.Message { - if data == nil { +// parseNodeBuild 解析节点构建结果（BuildType == 2） +func (s *promptService) parseNodeBuild(taskRecord *entity.ComposeTask) map[string]any { + if taskRecord == nil { return nil } - - // 统一序列化为 JSON - jsonBytes, err := json.Marshal(data) - if err != nil { - return nil - } - - // 第一层解析：尝试直接解析为消息数组 - var messages []dto.Message - if err := json.Unmarshal(jsonBytes, &messages); err == nil { - // 成功解析，但需要处理 content 可能是 JSON 字符串的情况 - return deepNormalizeMessages(messages) - } - - // 第二层解析：可能是 JSON 字符串包裹的数组 - var rawStr string - if err := json.Unmarshal(jsonBytes, &rawStr); err != nil { - return nil - } - - // 尝试解析字符串为消息数组 - if err := json.Unmarshal([]byte(rawStr), &messages); err == nil { - return deepNormalizeMessages(messages) - } - - return nil -} - -// deepNormalizeMessages 深度规范化消息，处理 content 为 JSON 字符串的情况 -func deepNormalizeMessages(messages []dto.Message) []dto.Message { - for i, msg := range messages { - messages[i].Content = deepNormalizeContent(msg.Content) - } - return messages -} - -// deepNormalizeContent 递归处理 content，支持多层 JSON 嵌套 -func deepNormalizeContent(content any) any { - switch v := content.(type) { + var result map[string]any + switch v := taskRecord.Messages.(type) { + case *gvar.Var: + if v != nil { + json.Unmarshal([]byte(v.String()), &result) + } case string: - // 尝试解析 JSON 字符串 - v = strings.TrimSpace(v) - if v == "" { - return v - } - - // 如果看起来像 JSON，尝试解析 - if looksLikeJSON(v) { - var parsed any - if err := json.Unmarshal([]byte(v), &parsed); err == nil { - // 递归处理解析后的内容 - return deepNormalizeContent(parsed) - } - } - return v - - case []any: - // 递归处理数组中的每个元素 - result := make([]any, len(v)) - for i, item := range v { - result[i] = deepNormalizeContent(item) - } - return result - + json.Unmarshal([]byte(v), &result) case map[string]any: - // 递归处理 map 中的每个值 - result := make(map[string]any, len(v)) - for k, val := range v { - result[k] = deepNormalizeContent(val) - } - return result - + result = v default: - return content - } -} - -func NormalizeToTwoPart(messages []dto.Message, req *dto.ComposeMessagesReq) []dto.Message { - var result []dto.Message - - // 1. 提取 system - sysContent := extractByRole(messages, "system") - if sysContent == nil { - sysContent = renderFormText(req.Form, false) - } - result = append(result, dto.Message{Role: "system", Content: sysContent}) - - // 2. 提取 form - formContent := extractByRole(messages, "form") - if formContent != nil { - result = append(result, dto.Message{Role: "form", Content: formContent}) - } else if req != nil { - result = append(result, dto.Message{Role: "form", Content: renderFormJSON(req.Form)}) - } - - // 3. 提取 skill - skillContent := extractByRole(messages, "skill") - if skillContent != nil { - result = append(result, dto.Message{Role: "skill", Content: skillContent}) - } else if req != nil && req.SkillName != "" { - result = append(result, dto.Message{Role: "skill", Content: req.SkillName}) - } - - // 4. 提取 history（如果模型返回了压缩后的历史） - historyContent := extractByRole(messages, "history") - if historyContent != nil { - result = append(result, dto.Message{Role: "history", Content: historyContent}) - } - - // 5. 提取 user - usrContent := extractByRole(messages, "user") - if usrContent == nil { - usrContent = renderUserText(req.UserForm, req.Form) - } - result = append(result, dto.Message{Role: "user", Content: usrContent}) - - return result -} - -// ============================================ -// 辅助函数：按 role 提取第一个非空 content -// ============================================ - -func extractByRole(messages []dto.Message, role string) any { - for _, msg := range messages { - if msg.Role == role && !isEmptyValue(msg.Content) { - return msg.Content - } - } - return nil -} - -// ============================================ -// 辅助函数：将 form 渲染为 JSON 对象 -// ============================================ - -func renderFormJSON(form map[string]any) map[string]any { - if form == nil { - return nil - } - result := make(map[string]any) - for k, v := range form { - result[k] = v + b, _ := json.Marshal(v) + json.Unmarshal(b, &result) } return result } - -func enrichSystemMessages(messages []dto.Message, req *dto.ComposeMessagesReq) []dto.Message { - if len(messages) == 0 { - return messages - } - - // 获取系统字段的值映射 - systemValues := extractSystemValues(req) - - for i, msg := range messages { - if msg.Role != "system" { - continue - } - - // 为 schema 数组补充 value - switch content := msg.Content.(type) { - case []any: - messages[i].Content = enrichSchemaWithValues(content, systemValues) - case []map[string]any: - arr := make([]any, len(content)) - for j, item := range content { - arr[j] = item - } - messages[i].Content = enrichSchemaWithValues(arr, systemValues) - case map[string]any: - // 合并但不覆盖已有值 - for k, v := range systemValues { - if _, exists := content[k]; !exists { - content[k] = v - } - } - messages[i].Content = content - } - } - - return messages -} diff --git a/service/files_service.go b/service/files_service.go index d654605..393fdb0 100644 --- a/service/files_service.go +++ b/service/files_service.go @@ -1,21 +1,59 @@ package service import ( + "archive/zip" + "bytes" "context" "fmt" "io" "net/http" + "path/filepath" + "regexp" "strings" "time" - "github.com/gogf/gf/v2/os/glog" + "github.com/gogf/gf/v2/frame/g" ) // ============================================ -// 文件处理 +// 文件处理（配置直接内联 + zip 支持） // ============================================ -func fetchFileTexts(ctx context.Context, urls []string) map[string]string { +// 允许的文本类 MIME 类型前缀 +var allowedMIMEPrefixes = []string{ + "text/", + "application/json", + "application/xml", + "application/javascript", + "application/x-yaml", + "application/yaml", + "application/toml", + "application/x-httpd-php", + "application/x-sh", + "application/x-python", + "application/x-perl", + "application/x-ruby", +} + +// 禁止的文件扩展名 +var bannedExtensions = map[string]bool{ + ".png": true, ".jpg": true, ".jpeg": true, ".gif": true, ".bmp": true, + ".webp": true, ".svg": true, ".ico": true, ".tiff": true, ".tif": true, + ".mp3": true, ".wav": true, ".ogg": true, ".flac": true, ".aac": true, + ".wma": true, ".m4a": true, + ".mp4": true, ".avi": true, ".mkv": true, ".mov": true, ".wmv": true, + ".flv": true, ".webm": true, + ".tar": true, ".gz": true, ".rar": true, ".7z": true, + ".exe": true, ".dll": true, ".so": true, ".bin": true, ".dat": true, + ".class": true, ".pyc": true, + ".pdf": true, ".doc": true, ".docx": true, ".xls": true, ".xlsx": true, + ".ppt": true, ".pptx": true, +} + +var symbolCleaner = regexp.MustCompile(`[\x00-\x08\x0B\x0C\x0E-\x1F]`) + +// FetchFileTexts 从 URL 列表获取文件内容（支持 zip 内文件） +func FetchFileTexts(ctx context.Context, urls []string) map[string]string { result := make(map[string]string) if len(urls) == 0 { @@ -23,7 +61,7 @@ func fetchFileTexts(ctx context.Context, urls []string) map[string]string { } client := &http.Client{ - Timeout: 8 * time.Second, + Timeout: time.Duration(g.Cfg().MustGet(ctx, "userFiles.httpTimeoutSec", 8).Int()) * time.Second, } for _, rawURL := range urls { @@ -32,13 +70,20 @@ func fetchFileTexts(ctx context.Context, urls []string) map[string]string { continue } + if isBannedExtension(url) { + continue + } + + if isZipExtension(url) { + zipTexts := fetchZipFileTexts(ctx, client, url) + for k, v := range zipTexts { + result[k] = v + } + continue + } + text, err := fetchFileContent(ctx, client, url) if err != nil { - glog.Warningf(ctx, - "[FetchFile] failed url=%s err=%v", - url, - err, - ) continue } @@ -46,24 +91,131 @@ func fetchFileTexts(ctx context.Context, urls []string) map[string]string { continue } + text = cleanSymbols(text) result[url] = text } return result } -func fetchFileContent( - ctx context.Context, - client *http.Client, - url string, -) (string, error) { +func isZipExtension(url string) bool { + ext := strings.ToLower(filepath.Ext(url)) + if idx := strings.Index(ext, "?"); idx != -1 { + ext = ext[:idx] + } + return ext == ".zip" +} - req, err := http.NewRequestWithContext( - ctx, - http.MethodGet, - url, - nil, +func fetchZipFileTexts(ctx context.Context, client *http.Client, url string) map[string]string { + result := make(map[string]string) + + zipBytes, err := downloadFile(client, url, + int64(g.Cfg().MustGet(ctx, "userFiles.zipMaxSizeMB", 10).Int())*1024*1024, ) + if err != nil { + return result + } + + reader, err := zip.NewReader(bytes.NewReader(zipBytes), int64(len(zipBytes))) + if err != nil { + return result + } + + entryMaxSize := int64(g.Cfg().MustGet(ctx, "userFiles.zipEntryMaxSizeKB", 500).Int()) * 1024 + + for _, file := range reader.File { + if file.FileInfo().IsDir() { + continue + } + + fileName := file.Name + + if isBannedExtension(fileName) { + continue + } + + if isZipExtension(fileName) { + continue + } + + rc, err := file.Open() + if err != nil { + continue + } + + content, err := io.ReadAll(io.LimitReader(rc, entryMaxSize)) + rc.Close() + if err != nil { + continue + } + + contentType := http.DetectContentType(content) + if !isReadableContentType(contentType) { + continue + } + + text := cleanSymbols(string(content)) + if text == "" { + continue + } + + key := url + "::" + fileName + result[key] = text + } + + return result +} + +func downloadFile(client *http.Client, url string, maxSize int64) ([]byte, error) { + req, err := http.NewRequest(http.MethodGet, url, nil) + if err != nil { + return nil, err + } + + resp, err := client.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, fmt.Errorf("HTTP %d", resp.StatusCode) + } + + return io.ReadAll(io.LimitReader(resp.Body, maxSize)) +} + +func isBannedExtension(url string) bool { + ext := strings.ToLower(filepath.Ext(url)) + if idx := strings.Index(ext, "?"); idx != -1 { + ext = ext[:idx] + } + return bannedExtensions[ext] +} + +func isReadableContentType(contentType string) bool { + if contentType == "" { + return false + } + ct := strings.ToLower(contentType) + for _, prefix := range allowedMIMEPrefixes { + if strings.HasPrefix(ct, prefix) { + return true + } + } + return false +} + +func cleanSymbols(text string) string { + text = symbolCleaner.ReplaceAllString(text, "") + text = strings.ReplaceAll(text, "\r\n", "\n") + text = strings.ReplaceAll(text, "\r", "\n") + text = regexp.MustCompile(`\n{3,}`).ReplaceAllString(text, "\n\n") + return strings.TrimSpace(text) +} + +func fetchFileContent(ctx context.Context, client *http.Client, url string) (string, error) { + req, err := http.NewRequest(http.MethodGet, url, nil) if err != nil { return "", err } @@ -74,21 +226,19 @@ func fetchFileContent( } defer resp.Body.Close() - // HTTP状态检查 if resp.StatusCode < 200 || resp.StatusCode >= 300 { return "", fmt.Errorf("HTTP %d", resp.StatusCode) } - // Content-Type检查 - contentType := strings.ToLower(resp.Header.Get("Content-Type")) - - if !isTextContentType(contentType) { - return "", fmt.Errorf("unsupported content-type: %s", contentType) + contentType := resp.Header.Get("Content-Type") + if !isReadableContentType(contentType) { + return "", fmt.Errorf("unreadable content-type: %s", contentType) } - // 最大读取20KB body, err := io.ReadAll( - io.LimitReader(resp.Body, 20*1024), + io.LimitReader(resp.Body, + int64(g.Cfg().MustGet(ctx, "userFiles.textFileMaxSizeKB", 500).Int())*1024, + ), ) if err != nil { return "", err @@ -97,35 +247,94 @@ func fetchFileContent( return strings.TrimSpace(string(body)), nil } -// 判断是否为文本类型 -func isTextContentType(contentType string) bool { - - // text/* - if strings.HasPrefix(contentType, "text/") { - return true - } - - // 常见文本类型 - allowTypes := []string{ - "application/json", - "application/xml", - "application/javascript", - "application/x-yaml", - "application/yaml", - "application/toml", - } - - for _, t := range allowTypes { - if strings.Contains(contentType, t) { - return true - } - } - - return false -} - func sanitizeURL(raw string) string { s := strings.TrimSpace(raw) s = strings.Trim(s, "`\"") return s } + +// SkillMdContent 根据 skillName 获取 zip 内所有 md 文件拼接内容 +func SkillMdContent(ctx context.Context, skillName string) string { + // 1. 请求接口获取 SkillUserVO + skillResp, err := GetSkillUser(ctx, skillName) + if err != nil { + return "" + } + + fullUrl := skillResp.ImgAddressPrefix + skillResp.FileUrl + // 2. 下载 zip 文件 + client := &http.Client{ + Timeout: time.Duration(g.Cfg().MustGet(ctx, "skillFiles.httpTimeoutSec", 30).Int()) * time.Second, + } + + zipBytes, err := downloadFile(client, fullUrl, + int64(g.Cfg().MustGet(ctx, "skillFiles.zipMaxSizeMB", 10).Int())*1024*1024, + ) + if err != nil { + return "" + } + + // 3. 解压 zip 并提取所有 md 文件内容 + mdContents, err := extractMdFiles(ctx, zipBytes) + if err != nil { + return "" + } + + if len(mdContents) == 0 { + return "" + } + + // 4. 拼接所有 md 内容 + var builder strings.Builder + builder.WriteString(fmt.Sprintf("# Skill: %s\n\n", skillResp.Name)) + if skillResp.Description != "" { + builder.WriteString(fmt.Sprintf("> %s\n\n", skillResp.Description)) + } + + for fileName, content := range mdContents { + builder.WriteString(fmt.Sprintf("## %s\n\n", fileName)) + builder.WriteString(content) + builder.WriteString("\n\n---\n\n") + } + + return strings.TrimSpace(builder.String()) +} + +// extractMdFiles 解压 zip 并提取所有 .md 文件内容 +func extractMdFiles(ctx context.Context, zipBytes []byte) (map[string]string, error) { + result := make(map[string]string) + + reader, err := zip.NewReader(bytes.NewReader(zipBytes), int64(len(zipBytes))) + if err != nil { + return nil, err + } + + entryMaxSize := int64(g.Cfg().MustGet(ctx, "skillFiles.mdMaxSizeKB", 500).Int()) * 1024 + + for _, file := range reader.File { + if file.FileInfo().IsDir() { + continue + } + + if !strings.HasSuffix(strings.ToLower(file.Name), ".md") { + continue + } + + rc, err := file.Open() + if err != nil { + continue + } + + content, err := io.ReadAll(io.LimitReader(rc, entryMaxSize)) + rc.Close() + if err != nil { + continue + } + + if len(content) > 0 { + result[file.Name] = strings.TrimSpace(string(content)) + } + } + + return result, nil +} diff --git a/service/form_processor.go b/service/form_processor.go deleted file mode 100644 index 6204892..0000000 --- a/service/form_processor.go +++ /dev/null @@ -1,325 +0,0 @@ -package service - -import ( - "fmt" - "strings" -) - -// ============================================ -// 表单处理 -// ============================================ - -// FormProcessor 表单处理器 -type FormProcessor struct { - SystemForm map[string]any - UserForm map[string]any -} - -// NewFormProcessor 创建表单处理器 -func NewFormProcessor(systemForm, userForm map[string]any) *FormProcessor { - return &FormProcessor{ - SystemForm: systemForm, - UserForm: userForm, - } -} - -// Merge 合并表单，用户表单覆盖系统表单 -func (p *FormProcessor) Merge() map[string]any { - if len(p.SystemForm) == 0 { - return p.SystemForm - } - - result := make(map[string]any) - for k, v := range p.SystemForm { - result[k] = v - } - - if len(p.UserForm) == 0 { - return result - } - - // 构建用户表单索引 - userIndex := buildFieldIndex(p.UserForm) - - // 覆盖匹配的字段 - for key, value := range result { - item, ok := value.(map[string]any) - if !ok || len(item) == 0 { - continue - } - - field := getField(item, key) - - if userItem, exists := findInIndex(userIndex, field, getLabel(item)); exists { - if userValue := getValue(userItem); !isNilOrEmpty(userValue) { - result[key] = cloneWithValue(item, userValue) - } - } - } - - return result -} - -// RemoveDuplicates 移除被用户表单覆盖的字段 -func (p *FormProcessor) RemoveDuplicates() map[string]any { - if len(p.SystemForm) == 0 || len(p.UserForm) == 0 { - return p.SystemForm - } - - userFields := buildFieldSet(p.UserForm) - result := make(map[string]any) - - for key, value := range p.SystemForm { - item, ok := value.(map[string]any) - if !ok || len(item) == 0 { - result[key] = value - continue - } - - field := getField(item, key) - label := getLabel(item) - - // 跳过重复字段 - if userFields.contains(field) || userFields.containsLabel(label) { - continue - } - - result[key] = value - } - - return result -} - -// RemoveSemanticDuplicates 语义去重 -func (p *FormProcessor) RemoveSemanticDuplicates() map[string]any { - if len(p.SystemForm) == 0 || len(p.UserForm) == 0 { - return p.SystemForm - } - - userText := renderUserTextOnly(p.UserForm) - if userText == "" { - return p.SystemForm - } - - result := make(map[string]any) - for key, value := range p.SystemForm { - item, ok := value.(map[string]any) - if !ok || len(item) == 0 { - result[key] = value - continue - } - - if isDuplicate(userText, getField(item, key), getLabel(item), getValue(item)) { - continue - } - - result[key] = value - } - - return result -} - -// RenderSystemText 渲染系统提示词文本 -func (p *FormProcessor) RenderSystemText() string { - return renderFormText(p.SystemForm, false) -} - -// RenderUserText 渲染用户提示词文本 -func (p *FormProcessor) RenderUserText() string { - return renderUserText(p.UserForm, p.SystemForm) -} - -// ============================================ -// 表单处理辅助方法 -// ============================================ - -type fieldSet struct { - fields map[string]bool - labels map[string]bool -} - -func buildFieldSet(form map[string]any) *fieldSet { - fs := &fieldSet{ - fields: make(map[string]bool), - labels: make(map[string]bool), - } - - for key, value := range form { - item, ok := value.(map[string]any) - if !ok || len(item) == 0 { - continue - } - - field := strings.ToLower(getField(item, key)) - if field != "" { - fs.fields[field] = true - } - - if label := strings.ToLower(getLabel(item)); label != "" { - fs.labels[label] = true - } - } - - return fs -} - -func (fs *fieldSet) contains(field string) bool { - return fs.fields[strings.ToLower(field)] -} - -func (fs *fieldSet) containsLabel(label string) bool { - return label != "" && fs.labels[strings.ToLower(label)] -} - -func buildFieldIndex(form map[string]any) map[string]map[string]any { - index := make(map[string]map[string]any) - - for key, value := range form { - item, ok := value.(map[string]any) - if !ok || len(item) == 0 { - continue - } - - field := strings.ToLower(getField(item, key)) - if field != "" { - index[field] = item - } - - if label := strings.ToLower(getLabel(item)); label != "" { - if _, exists := index[label]; !exists { - index[label] = item - } - } - } - - return index -} - -func findInIndex(index map[string]map[string]any, field, label string) (map[string]any, bool) { - key := strings.ToLower(field) - if item, ok := index[key]; ok { - return item, true - } - - if label != "" { - key = strings.ToLower(label) - if item, ok := index[key]; ok { - return item, true - } - } - - return nil, false -} - -// ============================================ -// 表单渲染 -// ============================================ - -func renderFormText(form map[string]any, isUserForm bool) string { - if len(form) == 0 { - return "" - } - - // 用户表单只有一个文本字段时，直接返回值 - if isUserForm && len(form) == 1 { - for _, value := range form { - if item, ok := value.(map[string]any); ok { - return strings.TrimSpace(asString(getValue(item))) - } - } - } - - // 拼接渲染 - items := extractFormItems(form) - if isUserForm { - return renderUserFormItems(items) - } - return renderSystemFormItems(items) -} - -type formItem struct { - Key string - Field string - Label string - Value any -} - -func extractFormItems(form map[string]any) []formItem { - var items []formItem - - keys := sortedKeys(form) - for _, key := range keys { - item, ok := form[key].(map[string]any) - if !ok || len(item) == 0 { - continue - } - - field := getField(item, key) - value := getValue(item) - - // 跳过敏感字段和空值 - if isSensitiveField(field) || isNilOrEmpty(value) { - continue - } - - items = append(items, formItem{ - Key: key, - Field: field, - Label: getLabel(item), - Value: value, - }) - } - - return items -} - -func renderUserFormItems(items []formItem) string { - // 只有一个文本类型字段时，直接返回值 - if len(items) == 1 && isTextType(items[0].Field, items[0].Label) { - return formatValue(items[0].Value) - } - - // 拼接 - var parts []string - for _, item := range items { - if isTextType(item.Field, item.Label) { - parts = append(parts, formatValue(item.Value)) - } else { - label := item.Label - if label == "" { - label = item.Field - } - parts = append(parts, fmt.Sprintf("%s：%s", label, formatValue(item.Value))) - } - } - - return strings.Join(parts, "，") -} - -func renderSystemFormItems(items []formItem) string { - var parts []string - for _, item := range items { - label := item.Label - if label == "" { - label = item.Field - } - parts = append(parts, fmt.Sprintf("%s：%s", label, formatValue(item.Value))) - } - - return strings.Join(parts, "，") -} - -func renderUserText(userForm, systemForm map[string]any) string { - if text := renderFormText(userForm, true); text != "" { - return text - } - // 用户表单为空时，使用系统表单生成 - if text := renderFormText(systemForm, false); text != "" { - return "参考系统字段生成用户提示词：" + text - } - return "" -} - -func renderUserTextOnly(userForm map[string]any) string { - return renderFormText(userForm, true) -} diff --git a/service/headers.go b/service/headers.go index 3920ef0..e83ae4d 100644 --- a/service/headers.go +++ b/service/headers.go @@ -5,7 +5,6 @@ import ( "gitea.com/red-future/common/utils" "github.com/gogf/gf/v2/frame/g" - "github.com/gogf/gf/v2/util/gconv" ) // asyncCtx 固化异步执行所需的 token/user，避免请求结束后丢失（仅在“同请求内起 goroutine”有用）。 @@ -52,18 +51,3 @@ func forwardHeaders(ctx context.Context) map[string]string { } return headers } - -// GetTenantId 获取租户ID -func GetTenantId(ctx context.Context) int64 { - var currentTenantID int64 - if r := g.RequestFromCtx(ctx); r != nil { - currentTenantID = gconv.Int64(r.Header.Get("X-Tenant-Id")) - } - if currentTenantID == 0 { - userInfo, err := utils.GetUserInfo(ctx) - if err == nil && userInfo != nil { - currentTenantID = int64(userInfo.TenantId) - } - } - return currentTenantID -} diff --git a/service/http_service.go b/service/http_service.go index 765d1ca..9109937 100644 --- a/service/http_service.go +++ b/service/http_service.go @@ -6,12 +6,9 @@ import ( "fmt" commonHttp "gitea.com/red-future/common/http" + "github.com/gogf/gf/v2/os/gtime" ) -// ============================================ -// model-gateway 网关交互 -// ============================================ - // CreateTaskReq 创建任务请求 type CreateTaskReq struct { TaskId string `json:"task_id"` @@ -52,3 +49,27 @@ func queryGatewayTaskState(ctx context.Context, taskID string) (int, error) { } return req.State, nil } + +// SkillUserVO 技能用户视图对象 +type SkillUserVO struct { + Id int64 `json:"id,string"` + Name string `json:"name"` + Description string `json:"description"` + FileName string `json:"fileName"` + FileUrl string `json:"fileUrl"` // html 后缀 + CreatedAt *gtime.Time `json:"createdAt"` + UpdatedAt *gtime.Time `json:"updatedAt"` + ImgAddressPrefix string `json:"imgAddressPrefix"` // htmml 前缀 +} + +// GetSkillUser 根据 name 获取技能用户信息 +func GetSkillUser(ctx context.Context, name string) (*SkillUserVO, error) { + fullURL := fmt.Sprintf("ai-agent/skill/user/getUserOrTemplate?name=%s", name) + headers := forwardHeaders(ctx) + var resp SkillUserVO + var req struct{} + if err := commonHttp.Get(ctx, fullURL, headers, &resp, req); err != nil { + return nil, err + } + return &resp, nil +} diff --git a/service/model_output_parser.go b/service/model_output_parser.go deleted file mode 100644 index 7a8d93c..0000000 --- a/service/model_output_parser.go +++ /dev/null @@ -1,43 +0,0 @@ -package service - -import ( - "encoding/json" - "fmt" - "strings" -) - -// ============================================ -// 消息解析 -// ============================================ -func parseModelOutput(text string) (map[string]any, error) { - var result map[string]any - if err := json.Unmarshal([]byte(text), &result); err != nil { - return nil, fmt.Errorf("解析模型输出失败: %w", err) - } - return result, nil -} - -// cleanJSONString 清理字符串中可能导致JSON解析失败的字符 -func cleanJSONString(s string) string { - s = strings.ReplaceAll(s, "\u2018", "'") // 左单引号 ‘ - s = strings.ReplaceAll(s, "\u2019", "'") // 右单引号 ’ - s = strings.ReplaceAll(s, "\u201c", "\"") // 左双引号 “ - s = strings.ReplaceAll(s, "\u201d", "\"") // 右双引号 ” - return s -} - -func truncateStr(s string, maxLen int) string { - if len(s) <= maxLen { - return s - } - return s[:maxLen] -} - -// sessionParseModelOutput 解析会话模型输出 -func sessionParseModelOutput(text string) (map[string]any, error) { - var result map[string]any - if err := json.Unmarshal([]byte(text), &result); err != nil { - return nil, fmt.Errorf("解析模型输出失败: %w", err) - } - return result, nil -} diff --git a/service/session_redis_service.go b/service/session_redis_service.go index ab7bea1..d854c66 100644 --- a/service/session_redis_service.go +++ b/service/session_redis_service.go @@ -4,113 +4,43 @@ import ( "context" "encoding/json" "fmt" - "strings" "time" "github.com/gogf/gf/v2/frame/g" ) -// Message 消息结构（content 支持 string 或 []string） -type Message struct { - Role string `json:"role"` // user / assistant / system - Content any `json:"content"` // 内容：string 或 []string - Type string `json:"type,omitempty"` // text / file（可选扩展） -} - -// GetContentString 获取 Content 的字符串形式 -func (m Message) GetContentString() string { - switch v := m.Content.(type) { - case string: - return v - case []interface{}: - var parts []string - for _, item := range v { - if s, ok := item.(string); ok { - parts = append(parts, s) - } - } - return strings.Join(parts, "\n") - default: - b, _ := json.Marshal(m.Content) - return string(b) - } -} - -// SessionRoundData Redis存储的单轮会话数据 -type SessionRoundData struct { - SessionId string `json:"sessionId"` // 会话ID - RequestContent []Message `json:"requestContent"` // 用户请求会话 - ResponseContent []Message `json:"responseContent"` // AI回调会话 - Timestamp int64 `json:"timestamp"` // 存入时间戳 -} - -// GetSessionHistory 获取多轮会话历史（供推理时使用） -func (s *sessionService) GetSessionHistory(ctx context.Context, sessionId string) ([]SessionRoundData, error) { - return s.getFromRedis(ctx, sessionId) -} - -// BuildMessages 根据Redis历史构建完整的Messages数组 -func (s *sessionService) BuildMessages(ctx context.Context, sessionId string, currentMessages []Message) ([]Message, error) { - // 获取历史会话 - history, err := s.getFromRedis(ctx, sessionId) - if err != nil { - return nil, fmt.Errorf("获取历史会话失败: %w", err) - } - - var allMessages []Message - - // 按时间顺序拼接历史消息 - for _, round := range history { - allMessages = append(allMessages, round.RequestContent...) - allMessages = append(allMessages, round.ResponseContent...) - } - - // 添加当前轮次的请求消息 - allMessages = append(allMessages, currentMessages...) - - return allMessages, nil -} - // ==================== Redis 操作 ==================== // saveToRedis 保存会话数据到Redis -// sessionId: 会话ID作为key -// 最大10轮，超出替换最早的，过期时间30分钟 -func (s *sessionService) saveToRedis(ctx context.Context, sessionId string, requestMessages []Message, responseMessages []Message) error { +func (s *sessionService) saveToRedis(ctx context.Context, sessionId string, requestMessages []map[string]any, responseMessages []map[string]any) error { key := fmt.Sprintf("chat:session:%s", sessionId) - // 从配置读取，提供默认值 maxRounds := g.Cfg().MustGet(ctx, "session.maxRounds", 10).Int() expireSeconds := g.Cfg().MustGet(ctx, "session.expireTime", 1800).Int64() expireTime := time.Duration(expireSeconds) * time.Second - // 构造存储数据 - data := SessionRoundData{ - SessionId: sessionId, - RequestContent: requestMessages, - ResponseContent: responseMessages, - Timestamp: time.Now().Unix(), + data := map[string]any{ + "sessionId": sessionId, + "requestContent": requestMessages, + "responseContent": responseMessages, + "timestamp": time.Now().Unix(), } - // 序列化 b, err := json.Marshal(data) if err != nil { return fmt.Errorf("序列化会话数据失败: %w", err) } - // 写入 Redis（LPUSH 添加到最前面，新的在前） _, err = g.Redis().Do(ctx, "LPUSH", key, string(b)) if err != nil { return fmt.Errorf("写入Redis失败: %w", err) } - // 裁剪到最新10轮（保留前10条） _, err = g.Redis().Do(ctx, "LTRIM", key, 0, maxRounds-1) if err != nil { return fmt.Errorf("裁剪Redis列表失败: %w", err) } - // 重置过期时间 _, err = g.Redis().Do(ctx, "EXPIRE", key, int64(expireTime.Seconds())) if err != nil { return fmt.Errorf("设置过期时间失败: %w", err) @@ -120,26 +50,22 @@ func (s *sessionService) saveToRedis(ctx context.Context, sessionId string, requ } // getFromRedis 从Redis获取会话历史 -func (s *sessionService) getFromRedis(ctx context.Context, sessionId string) ([]SessionRoundData, error) { +func (s *sessionService) getFromRedis(ctx context.Context, sessionId string) ([]map[string]any, error) { key := fmt.Sprintf("chat:session:%s", sessionId) - // 获取列表中所有数据（最多10条） result, err := g.Redis().Do(ctx, "LRANGE", key, 0, -1) if err != nil { return nil, fmt.Errorf("从Redis获取数据失败: %w", err) } if result == nil || result.IsNil() { - return []SessionRoundData{}, nil + return []map[string]any{}, nil } - // 解析数据 - var sessions []SessionRoundData - - // 将结果转换为字符串数组 + var sessions []map[string]any values := result.Strings() for _, str := range values { - var data SessionRoundData + var data map[string]any if err := json.Unmarshal([]byte(str), &data); err != nil { g.Log().Warningf(ctx, "[会话] 解析Redis数据失败 err=%v", err) continue @@ -147,7 +73,7 @@ func (s *sessionService) getFromRedis(ctx context.Context, sessionId string) ([] sessions = append(sessions, data) } - // 反转顺序（Redis存储最新在前，使用时按时间正序） + // 反转（Redis 最新在前 → 时间正序） for i, j := 0, len(sessions)-1; i < j; i, j = i+1, j-1 { sessions[i], sessions[j] = sessions[j], sessions[i] } @@ -155,26 +81,33 @@ func (s *sessionService) getFromRedis(ctx context.Context, sessionId string) ([] return sessions, nil } -// GetSessionHistoryForInference 获取历史会话，直接返回Message数组（给推理用） -func (s *sessionService) GetSessionHistoryForInference(ctx context.Context, sessionId string) ([]Message, error) { - // 从Redis获取历史会话数据 +// GetSessionHistoryForInference 获取历史会话，返回扁平消息数组（给推理用） +func (s *sessionService) GetSessionHistoryForInference(ctx context.Context, sessionId string) ([]map[string]any, error) { historyData, err := s.getFromRedis(ctx, sessionId) if err != nil { return nil, fmt.Errorf("获取历史会话失败: %w", err) } - // 如果没有任何历史数据，返回空 if len(historyData) == 0 { - return []Message{}, nil + return []map[string]any{}, nil } - // 把SessionRoundData转换成扁平的Message数组 - var messages []Message + var messages []map[string]any for _, round := range historyData { - // 先加用户的请求 - messages = append(messages, round.RequestContent...) - // 再加AI的回答 - messages = append(messages, round.ResponseContent...) + if reqMsgs, ok := round["requestContent"].([]interface{}); ok { + for _, m := range reqMsgs { + if msg, ok := m.(map[string]interface{}); ok { + messages = append(messages, msg) + } + } + } + if respMsgs, ok := round["responseContent"].([]interface{}); ok { + for _, m := range respMsgs { + if msg, ok := m.(map[string]interface{}); ok { + messages = append(messages, msg) + } + } + } } return messages, nil diff --git a/service/session_service.go b/service/session_service.go index 40534cf..cae70bc 100644 --- a/service/session_service.go +++ b/service/session_service.go @@ -2,46 +2,50 @@ package service import ( "context" + "fmt" "prompts-core/dao" "prompts-core/model/dto" "prompts-core/model/entity" "gitea.com/red-future/common/beans" "github.com/gogf/gf/v2/frame/g" + "github.com/gogf/gf/v2/util/gconv" ) var Session = &sessionService{} type sessionService struct{} -// SessionCallback 会话回调处理 func (s *sessionService) SessionCallback(ctx context.Context, req *dto.SessionCallbackReq) (res *beans.ResponseEmpty, err error) { // 1. 解析AI返回的文本 - result, err := sessionParseModelOutput(req.Text) + result, err := parseOutput(req.Text) if err != nil { g.Log().Errorf(ctx, "[会话回调] 解析模型输出失败 epicycleId=%d err=%v", req.EpicycleId, err) return nil, err } - // 2. 更新数据库（AI回调内容） - _, err = dao.ComposeSession.UpdateById(ctx, req.EpicycleId, map[string]any{ - entity.ComposeSessionCol.ResponseContent: result, + // 2. 更新数据库 + result["role"] = "assistant" + _, err = dao.ComposeSession.Update(ctx, &entity.ComposeSession{ + SQLBaseDO: beans.SQLBaseDO{Id: req.EpicycleId}, + ResponseContent: result, }) if err != nil { g.Log().Errorf(ctx, "[会话回调] 更新数据库失败 epicycleId=%d err=%v", req.EpicycleId, err) return nil, err } - // 3. 获取当前轮次的完整数据 + // 3. 获取当前轮次完整数据 session, err := dao.ComposeSession.GetById(ctx, req.EpicycleId) if err != nil { g.Log().Errorf(ctx, "[会话回调] 获取会话数据失败 epicycleId=%d err=%v", req.EpicycleId, err) return nil, err } - // 4. 写入 Redis（多轮记忆） - requestMessages := s.convertToMessages(session.RequestContent) - responseMessages := s.convertToMessages(session.ResponseContent) + // 4. 转换 json 并存入 Redis + requestMessages := convertToMessages(session.RequestContent) + responseMessages := convertToMessages(session.ResponseContent) + if err = s.saveToRedis(ctx, session.SessionId, requestMessages, responseMessages); err != nil { g.Log().Errorf(ctx, "[会话回调] Redis存储失败 sessionId=%s id=%d err=%v", session.SessionId, session.Id, err) @@ -52,3 +56,57 @@ func (s *sessionService) SessionCallback(ctx context.Context, req *dto.SessionCa session.SessionId, session.Id, len(requestMessages), len(responseMessages)) return &beans.ResponseEmpty{}, nil } + +// GetHistoryMessages 获取历史信息 +func (s *sessionService) GetHistoryMessages(ctx context.Context, sessionId string) ([]map[string]any, error) { + maxRounds := g.Cfg().MustGet(ctx, "session.maxRounds", 10).Int() + + // 1. 先从 Redis 拿 + redisHistory, err := s.GetSessionHistoryForInference(ctx, sessionId) + if err == nil && len(redisHistory) > 0 { + return redisHistory, nil + } + + // 2. Redis 没有 → fallback DB + sessions, err := dao.ComposeSession.GetListBySessionId(ctx, sessionId, maxRounds) + if err != nil { + return nil, fmt.Errorf("DB获取历史失败: %w", err) + } + + var messages []map[string]any + + for _, session := range sessions { + // request + reqMsgs := convertToMessages(session.RequestContent) + for _, m := range reqMsgs { + role := gconv.String(m["role"]) + if role == "user" || role == "assistant" { + messages = append(messages, m) + } + } + + // response + respMsgs := convertToMessages(session.ResponseContent) + for _, m := range respMsgs { + if m["role"] == nil { + m["role"] = "assistant" + } + messages = append(messages, m) + } + } + + // 3. 回写 Redis + for _, session := range sessions { + reqMsgs := convertToMessages(session.RequestContent) + respMsgs := convertToMessages(session.ResponseContent) + for i := range respMsgs { + if respMsgs[i]["role"] == nil { + respMsgs[i]["role"] = "assistant" + } + } + if len(reqMsgs) > 0 || len(respMsgs) > 0 { + _ = s.saveToRedis(ctx, session.SessionId, reqMsgs, respMsgs) + } + } + return messages, nil +} diff --git a/service/utils.go b/service/utils.go index 2ff7e87..37e30dc 100644 --- a/service/utils.go +++ b/service/utils.go @@ -1,138 +1,59 @@ +// utils 工具函数 package service import ( - "context" "encoding/json" "fmt" - "prompts-core/model/dto" - "sort" - "strings" - "github.com/gogf/gf/v2/container/gvar" - "github.com/gogf/gf/v2/frame/g" + "github.com/gogf/gf/v2/encoding/gjson" + "github.com/gogf/gf/v2/util/gconv" ) // ============================================ -// 工具函数 +// json 相关处理 // ============================================ - -func getField(item map[string]any, fallback string) string { - if field := asString(item["field"]); field != "" { - return field - } - return fallback -} - -func getLabel(item map[string]any) string { - return asString(item["label"]) -} - -func getValue(item map[string]any) any { - return item["value"] -} - -func cloneWithValue(item map[string]any, value any) map[string]any { - cloned := make(map[string]any) - for k, v := range item { - cloned[k] = v - } - cloned["value"] = value - return cloned -} - -func isSensitiveField(field string) bool { - f := strings.ToLower(field) - return f == "apikey" || f == "authorization" -} - -func isAPIKeyField(field string) bool { - f := strings.ToLower(field) - return f == "apikey" || f == "authorization" -} - -func isTextType(field, label string) bool { - f := strings.ToLower(field) - l := strings.ToLower(label) - return f == "prompt" || f == "text" || - l == "提示词" || l == "文本内容" || l == "prompt" || l == "text" -} - -func isDuplicate(userText, field, label string, value any) bool { - lowerText := strings.ToLower(userText) - - if label != "" && strings.Contains(lowerText, strings.ToLower(label)) { - return true - } - if field != "" && strings.Contains(lowerText, strings.ToLower(field)) { - return true +// parseOutput 解析模型输出为 JSON 格式 +func parseOutput(text string) (map[string]any, error) { + j, err := gjson.LoadJson([]byte(text)) + if err != nil { + return nil, fmt.Errorf("解析模型输出失败: %w", err) } - // 检查值 - if v := asString(value); v != "" && strings.Contains(lowerText, strings.ToLower(v)) { - return true - } - - return false + return j.Map(), nil } -func isEmptyValue(v any) bool { - if v == nil { - return true +func convertToMessages(raw any) []map[string]any { + if raw == nil { + return nil } - if s, ok := v.(string); ok { - return strings.TrimSpace(s) == "" + j, err := gjson.LoadJson(gconv.Bytes(raw)) + if err != nil { + return nil } - return false -} - -func isNilOrEmpty(v any) bool { - if v == nil { - return true + // 1. 如果有 messages + if j.Contains("messages") { + return gconv.Maps(j.Get("messages").Array()) } - if s, ok := v.(string); ok { - return strings.TrimSpace(s) == "" - } - return false -} - -func asString(v any) string { - switch t := v.(type) { - case string: - return t - default: - b, _ := json.Marshal(t) - return strings.Trim(string(b), "\"") + // 2. 否则当成单条 message + return []map[string]any{ + j.Map(), } } -func formatValue(v any) string { - return strings.TrimSpace(asString(v)) +// isMessageValid 校验推理结果是否合法 +func isMessageValid(message map[string]any) bool { + if message == nil { + return false + } + return true } -func mapToText(m map[string]any) string { - if len(m) == 0 { - return "" +func formToJSON(form map[string]any) string { + if form == nil { + return "{}" } - - keys := sortedKeys(m) - var parts []string - for _, k := range keys { - if isNilOrEmpty(m[k]) { - continue - } - parts = append(parts, fmt.Sprintf("%s：%s", k, formatValue(m[k]))) - } - - return strings.Join(parts, "，") -} - -func sortedKeys(m map[string]any) []string { - keys := make([]string, 0, len(m)) - for k := range m { - keys = append(keys, k) - } - sort.Strings(keys) - return keys + b, _ := json.Marshal(form) + return string(b) } func mustMarshal(v any) string { @@ -142,196 +63,3 @@ func mustMarshal(v any) string { } return string(b) } - -func formatTaskError(taskID, errMsg string) error { - if strings.TrimSpace(errMsg) == "" { - return fmt.Errorf("任务失败(taskId=%s)", taskID) - } - return fmt.Errorf("任务失败(taskId=%s): %s", taskID, errMsg) -} - -func getIntConfig(ctx context.Context, key string, fallback int) int { - v := g.Cfg().MustGet(ctx, key) - if v.IsEmpty() { - return fallback - } - return v.Int() -} - -// ============================================ -// Schema 处理 -// ============================================ - -func enrichSchemaWithValues(schema []any, values map[string]any) []any { - if len(schema) == 0 || len(values) == 0 { - return schema - } - - result := make([]any, len(schema)) - copy(result, schema) - - for i, item := range result { - m, ok := item.(map[string]any) - if !ok { - continue - } - - field := getField(m, "") - if field == "" { - continue - } - - // 已有 value 则跳过 - if _, hasValue := m["value"]; hasValue { - continue - } - - // 补充 value - if v, exists := values[field]; exists { - m["value"] = v - result[i] = m - } - } - - return result -} - -// extractContentFromResponse 从模型完整响应中提取 content 字段 -func extractContentFromResponse(text string) string { - // 尝试解析为完整的 choices 响应 - var response struct { - Choices []struct { - Message struct { - Content string `json:"content"` - } `json:"message"` - } `json:"choices"` - } - - if err := json.Unmarshal([]byte(text), &response); err != nil { - return "" - } - - if len(response.Choices) > 0 && response.Choices[0].Message.Content != "" { - return response.Choices[0].Message.Content - } - - return "" -} - -// ============================================ -// 值提取 -// ============================================ - -func extractSystemValues(req *dto.ComposeMessagesReq) map[string]any { - if req == nil { - return nil - } - - values := make(map[string]any) - - for _, value := range req.Form { - item, ok := value.(map[string]any) - if !ok || len(item) == 0 { - continue - } - - field := getField(item, "") - if field == "" || isSensitiveField(field) { - continue - } - - if v := getValue(item); !isNilOrEmpty(v) { - values[field] = v - } - } - - return values -} - -func extractModelKey(form map[string]any) string { - for _, value := range form { - item, ok := value.(map[string]any) - if !ok { - continue - } - - field := getField(item, "") - if isAPIKeyField(field) { - key := strings.TrimSpace(asString(getValue(item))) - if key != "" { - if strings.Contains(key, ":") { - return key - } - return "Authorization:" + key - } - } - } - - return "" -} - -// ==================== 工具方法 ==================== - -// convertToMessages 将数据库 any 类型转换为 []Message -// 支持：JSON字符串、[]byte、[]interface{}、以及 content 为字符串数组的格式 -func (s *sessionService) convertToMessages(data any) []Message { - if data == nil { - return []Message{} - } - - // 处理 *gvar.Var - if v, ok := data.(*gvar.Var); ok { - if v == nil || v.IsNil() || v.IsEmpty() { - return []Message{} - } - data = v.Val() - } - - var rawList []any - - switch v := data.(type) { - case string: - if err := json.Unmarshal([]byte(v), &rawList); err != nil { - g.Log().Warningf(context.Background(), "[会话] 解析JSON字符串失败 err=%v data=%.200s", err, v) - return []Message{} - } - case []byte: - if err := json.Unmarshal(v, &rawList); err != nil { - g.Log().Warningf(context.Background(), "[会话] 解析字节数组失败 err=%v", err) - return []Message{} - } - case []interface{}: - rawList = v - default: - b, _ := json.Marshal(v) - if err := json.Unmarshal(b, &rawList); err != nil { - g.Log().Warningf(context.Background(), "[会话] 解析未知类型失败 err=%v type=%T", err, v) - return []Message{} - } - } - - // 转换每个元素为 Message - var messages []Message - for _, item := range rawList { - var msg Message - switch val := item.(type) { - case string: - if err := json.Unmarshal([]byte(val), &msg); err != nil { - g.Log().Warningf(context.Background(), "[会话] 解析消息元素失败 err=%v data=%s", err, val) - continue - } - case map[string]interface{}: - b, _ := json.Marshal(val) - json.Unmarshal(b, &msg) - default: - b, _ := json.Marshal(val) - json.Unmarshal(b, &msg) - } - messages = append(messages, msg) - } - - if messages == nil { - messages = []Message{} - } - return messages -} diff --git a/update.sql b/update.sql index 3e1aff1..3217134 100644 --- a/update.sql +++ b/update.sql @@ -95,8 +95,8 @@ CREATE TABLE IF NOT EXISTS prompts_compose_session ( deleted_at TIMESTAMP(6), session_id VARCHAR(64) NOT NULL, - request_content JSONB NOT NULL DEFAULT '[]'::jsonb, - response_content JSONB NOT NULL DEFAULT '[]'::jsonb, + request_content JSONB NOT NULL DEFAULT '{}'::jsonb, + response_content JSONB NOT NULL DEFAULT '{}'::jsonb, remark VARCHAR(500) NOT NULL DEFAULT '' );