feat: 新增会话管理 service/controller 并注册路由

2026-08-13 13:04:00 +08:00
parent ab7b8889ed
commit 1dce62921f
3 changed files with 239 additions and 1 deletions
@@ -2,23 +2,40 @@ package main

 import (
 	digitalhumanController "ai-agent/digital-human/controller"
+	// 空导入：加载内置模型工具与工作流处理器（组合根统一装载，业务侧不感知实现位置）
+	_ "ai-agent/tools/builtin/current_time"
+	"ai-agent/workflow/consts/public"
 	workController "ai-agent/workflow/controller"
 	workflowController "ai-agent/workflow/controller/flow"
 	workflowNodeController "ai-agent/workflow/controller/node"
 	workflowSkillController "ai-agent/workflow/controller/skill"
+	toolController "ai-agent/workflow/controller/tool"
+	sessionController "ai-agent/workflow/controller/session"
+	flowService "ai-agent/workflow/service/flow"
+	_ "ai-agent/workflow/service/flow/processor/builtin/split_batch"
+	_ "ai-agent/workflow/service/flow/processor/builtin/split_shots_pipeline"
+	toolWsService "ai-agent/workflow/service/tool"
 	"context"
+	"os"
+	"os/signal"
+	"syscall"

 	"gitea.redpowerfuture.com/red-future/common/http"
 	"gitea.redpowerfuture.com/red-future/common/jaeger"
+	gmq "github.com/bjang03/gmq/core/gmq"
+	"github.com/bjang03/gmq/mq"
 	_ "github.com/gogf/gf/contrib/drivers/pgsql/v2"
 	_ "github.com/gogf/gf/contrib/nosql/redis/v2"
+	"github.com/gogf/gf/v2/frame/g"
 )

 func main() {
 	ctx := context.Background()
 	defer jaeger.ShutDown(ctx)
-	// 注册路由
+
+	// 注册HTTP路由
 	http.Httpserver.BindHandler("/httpNodeCallback", workflowController.FlowCallBack.HttpNodeCallback)
+
 	http.RouteRegister([]interface{}{
 		//digitalhuman相关接口
 		digitalhumanController.Audio,        // 语音相关接口
@@ -26,6 +43,7 @@ func main() {
 		digitalhumanController.DigitalHuman, // 数字人相关接口
 		digitalhumanController.Video,        // 视频相关接口
 		digitalhumanController.AsyncTask,    // 异步任务相关接口
+		digitalhumanController.ModelTest,    // 模型流式测试接口
 		workController.CreationInfo,
 		workflowController.FlowExecution,
 		workflowController.FlowUser,
@@ -34,12 +52,33 @@ func main() {
 		workflowNodeController.NodePrompt,
 		workflowSkillController.SkillTemplate,
 		workflowSkillController.SkillUser,
+		toolController.Tool,
+		sessionController.Session,
 	})
 	//workflow.ExternalInterruptDemo()
 	//err := activePullService.ActivePullService.AllList(ctx)
 	//if err != nil {
 	//	g.Log().Error(ctx, "ActivePullService err: %v", err)
 	//}
+
+	gmq.GmqRegister(public.GmqMsgPluginsName, &mq.NatsConn{
+		NatsConfig: mq.NatsConfig{
+			Addr:     g.Config().MustGet(ctx, "nats.addr").String(),
+			Port:     g.Config().MustGet(ctx, "nats.port").String(),
+			Username: g.Config().MustGet(ctx, "nats.username").String(),
+			Password: g.Config().MustGet(ctx, "nats.password").String(),
+		},
+	})
+
+	// 监听退出信号，执行优雅关闭
+	sigCh := make(chan os.Signal, 1)
+	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
+	go func() {
+		<-sigCh
+		flowService.FlowWsService.Close()
+		toolWsService.ToolWsService.Close()
+	}()
+
 	// 保持应用运行
 	select {}
 }
@@ -0,0 +1,43 @@
+package session
+
+import (
+	"context"
+
+	"gitea.redpowerfuture.com/red-future/common/beans"
+	sessionDto "ai-agent/workflow/model/dto/session"
+	sessionService "ai-agent/workflow/service/session"
+)
+
+type session struct{}
+
+var Session = new(session)
+
+func (c *session) Create(ctx context.Context, req *sessionDto.CreateSessionReq) (res *sessionDto.CreateSessionRes, err error) {
+	return sessionService.SessionService.Create(ctx, req)
+}
+
+func (c *session) List(ctx context.Context, req *sessionDto.ListSessionReq) (res *sessionDto.ListSessionRes, err error) {
+	return sessionService.SessionService.List(ctx, req)
+}
+
+func (c *session) Delete(ctx context.Context, req *sessionDto.DeleteSessionReq) (res *beans.ResponseEmpty, err error) {
+	if err = sessionService.SessionService.Delete(ctx, req); err != nil {
+		return
+	}
+	return &beans.ResponseEmpty{}, nil
+}
+
+func (c *session) Results(ctx context.Context, req *sessionDto.ListSessionResultsReq) (res *sessionDto.ListSessionResultsRes, err error) {
+	return sessionService.SessionService.ListSessionResults(ctx, req)
+}
+
+func (c *session) ResultDelete(ctx context.Context, req *sessionDto.DeleteSessionResultReq) (res *beans.ResponseEmpty, err error) {
+	if err = sessionService.SessionService.DeleteResult(ctx, req); err != nil {
+		return
+	}
+	return &beans.ResponseEmpty{}, nil
+}
+
+func (c *session) WorkflowResults(ctx context.Context, req *sessionDto.ListWorkflowResultsReq) (res *sessionDto.ListWorkflowResultsRes, err error) {
+	return sessionService.SessionService.ListWorkflowResults(ctx, req)
+}
@@ -0,0 +1,156 @@
+package session
+
+import (
+	sessionDao "ai-agent/workflow/dao/session"
+	sessionDto "ai-agent/workflow/model/dto/session"
+	"ai-agent/workflow/model/entity"
+	"context"
+	"fmt"
+	"sort"
+
+	"gitea.redpowerfuture.com/red-future/common/utils"
+	"github.com/gogf/gf/v2/os/gtime"
+)
+
+var SessionService = &sessionService{}
+
+type sessionService struct{}
+
+func (s *sessionService) Create(ctx context.Context, req *sessionDto.CreateSessionReq) (res *sessionDto.CreateSessionRes, err error) {
+	id, err := sessionDao.SessionDao.Insert(ctx, req)
+	if err != nil {
+		return
+	}
+	return &sessionDto.CreateSessionRes{Id: id}, nil
+}
+
+func (s *sessionService) List(ctx context.Context, req *sessionDto.ListSessionReq) (res *sessionDto.ListSessionRes, err error) {
+	user, err := utils.GetUserInfo(ctx)
+	if err != nil {
+		return
+	}
+	list, total, err := sessionDao.SessionDao.List(ctx, user.UserName, req.Page)
+	if err != nil {
+		return
+	}
+	res = &sessionDto.ListSessionRes{Total: total}
+	for _, item := range list {
+		res.List = append(res.List, &sessionDto.VOSession{
+			Id:          item.Id,
+			SessionName: item.SessionName,
+			CreatedAt:   item.CreatedAt,
+		})
+	}
+	return
+}
+
+func (s *sessionService) Delete(ctx context.Context, req *sessionDto.DeleteSessionReq) (err error) {
+	return sessionDao.SessionDao.DeleteCascade(ctx, req.Id)
+}
+
+// ListSessionResults 会话内全部结果：工作流 + 普通对话混排，按创建时间倒序
+func (s *sessionService) ListSessionResults(ctx context.Context, req *sessionDto.ListSessionResultsReq) (res *sessionDto.ListSessionResultsRes, err error) {
+	wfList, err := sessionDao.WorkflowSessionResultDao.ListBySession(ctx, req.SessionId)
+	if err != nil {
+		return
+	}
+	chatList, err := sessionDao.ChatSessionResultDao.ListBySession(ctx, req.SessionId)
+	if err != nil {
+		return
+	}
+
+	type mixed struct {
+		createdAt *gtime.Time
+		vo        *sessionDto.VOSessionResult
+	}
+	var items []mixed
+	for _, w := range wfList {
+		items = append(items, mixed{createdAt: w.CreatedAt, vo: wfResultVO(w)})
+	}
+	for _, c := range chatList {
+		items = append(items, mixed{createdAt: c.CreatedAt, vo: chatResultVO(c)})
+	}
+	sort.Slice(items, func(i, j int) bool {
+		return items[i].createdAt.After(items[j].createdAt)
+	})
+
+	res = new(sessionDto.ListSessionResultsRes)
+	for _, it := range items {
+		res.List = append(res.List, it.vo)
+	}
+	return
+}
+
+func wfResultVO(w *entity.WorkflowSessionResult) *sessionDto.VOSessionResult {
+	return &sessionDto.VOSessionResult{
+		ResultId:      w.Id,
+		Type:          "workflow",
+		Status:        w.Status,
+		FlowId:        w.FlowId,
+		FlowName:      w.FlowName,
+		RequestParams: w.RequestParams,
+		ResultParams:  w.ResultParams,
+		TotalTokens:   w.TotalTokens,
+		TotalFee:      w.TotalFee,
+		ErrorMsg:      w.ErrorMessage,
+		CreatedAt:     w.CreatedAt,
+	}
+}
+
+func chatResultVO(c *entity.ChatSessionResult) *sessionDto.VOSessionResult {
+	status := sessionDto.ResultStatusSuccess
+	if c.ErrorMessage != "" {
+		status = sessionDto.ResultStatusFailed
+	}
+	return &sessionDto.VOSessionResult{
+		ResultId:    c.Id,
+		Type:        "chat",
+		Status:      status,
+		Question:    c.Question,
+		Answer:      c.Answer,
+		TotalTokens: c.TotalTokens,
+		TotalFee:    c.TotalFee,
+		ErrorMsg:    c.ErrorMessage,
+		CreatedAt:   c.CreatedAt,
+	}
+}
+
+func (s *sessionService) DeleteResult(ctx context.Context, req *sessionDto.DeleteSessionResultReq) (err error) {
+	switch req.Type {
+	case "workflow":
+		return sessionDao.WorkflowSessionResultDao.SoftDelete(ctx, req.Id)
+	case "chat":
+		return sessionDao.ChatSessionResultDao.SoftDelete(ctx, req.Id)
+	default:
+		return fmt.Errorf("未知结果类型: %s", req.Type)
+	}
+}
+
+// ListWorkflowResults 工作流维度结果平铺列表（含已删除会话下的结果）
+func (s *sessionService) ListWorkflowResults(ctx context.Context, req *sessionDto.ListWorkflowResultsReq) (res *sessionDto.ListWorkflowResultsRes, err error) {
+	user, err := utils.GetUserInfo(ctx)
+	if err != nil {
+		return
+	}
+	list, total, err := sessionDao.WorkflowSessionResultDao.ListWorkflowResults(ctx, user.UserName, req.FlowId, req.Page)
+	if err != nil {
+		return
+	}
+	res = &sessionDto.ListWorkflowResultsRes{Total: total}
+	for _, w := range list {
+		res.List = append(res.List, &sessionDto.VOWorkflowResult{
+			ResultId:      w.Id,
+			SessionId:     w.SessionId,
+			FlowId:        w.FlowId,
+			FlowName:      w.FlowName,
+			RequestParams: w.RequestParams,
+			ResultParams:  w.ResultParams,
+			Status:        w.Status,
+			TotalTokens:   w.TotalTokens,
+			TotalFee:      w.TotalFee,
+			ErrorMsg:      w.ErrorMessage,
+			CreatedAt:     w.CreatedAt,
+		})
+	}
+	return
+}