prompts-core

This commit is contained in:
2026-05-15 09:45:51 +08:00
parent c4b8382dbb
commit 6497629fd0
18 changed files with 804 additions and 1643 deletions

View File

@@ -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)
// ===================== 修复点3sleep 也要监听 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
}