diff --git a/main.go b/main.go index 6cef41f..1b40ff3 100644 --- a/main.go +++ b/main.go @@ -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 {} } diff --git a/workflow/controller/session/session_controller.go b/workflow/controller/session/session_controller.go new file mode 100644 index 0000000..a799218 --- /dev/null +++ b/workflow/controller/session/session_controller.go @@ -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) +} \ No newline at end of file diff --git a/workflow/service/session/session_service.go b/workflow/service/session/session_service.go new file mode 100644 index 0000000..fe6b405 --- /dev/null +++ b/workflow/service/session/session_service.go @@ -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 +} \ No newline at end of file