fuzhongyun
/
ai_scheduler
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/v3' into v3

This commit is contained in:
renzhiyuan 2025-12-27 10:19:59 +08:00
parent 642a77b662 43edfc54c7
commit de7d09e7a7
8 changed files with 391 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
config/config.yaml
View File

@ -65,6 +65,13 @@ tools:
enabled: true
base_url: "https://revcl.1688sup.com/api/admin/afterSales/reseller_pre_ai"
dingtalk:
api_key: "dingsbbntrkeiyazcfdg"
api_secret: "ObqxwyR20r9rVNhju0sCPQyQA98_FZSc32W4vgxnGFH_b02HZr1BPCJsOAF816nu"
table_demand:
url: "https://alidocs.dingtalk.com/i/nodes/YQBnd5ExVE6qAbnOiANQg2KKJyeZqMmz"
base_id: "2Amq4vjg89RnYx9DTp66m2orW3kdP0wQ"
sheet_id_or_name: "数据表"
default_prompt:
img_recognize:

7
config/config_env.yaml
View File

@ -105,6 +105,13 @@ eino_tools:
hytGoodsBrandSearch:
base_url: "https://gateway.dev.cdlsxd.cn/goods-admin/api/v1/goods/brand/list"
dingtalk:
api_key: "dingsbbntrkeiyazcfdg"
api_secret: "ObqxwyR20r9rVNhju0sCPQyQA98_FZSc32W4vgxnGFH_b02HZr1BPCJsOAF816nu"
table_demand:
url: "https://alidocs.dingtalk.com/i/nodes/YQBnd5ExVE6qAbnOiANQg2KKJyeZqMmz"
base_id: "YQBnd5ExVE6qAbnOiANQg2KKJyeZqMmz"
sheet_id_or_name: "数据表"
default_prompt:

7
config/config_test.yaml
View File

@ -119,6 +119,13 @@ eino_tools:
hytGoodsBrandSearch:
base_url: "https://gateway.dev.cdlsxd.cn/goods-admin/api/v1/goods/brand/list"
dingtalk:
api_key: "dingsbbntrkeiyazcfdg"
api_secret: "ObqxwyR20r9rVNhju0sCPQyQA98_FZSc32W4vgxnGFH_b02HZr1BPCJsOAF816nu"
table_demand:
url: "https://alidocs.dingtalk.com/i/nodes/YQBnd5ExVE6qAbnOiANQg2KKJyeZqMmz"
base_id: "YQBnd5ExVE6qAbnOiANQg2KKJyeZqMmz"
sheet_id_or_name: "数据表"
default_prompt:
img_recognize:

102
internal/biz/do/handle.go
View File

@ -4,6 +4,7 @@ import (
"ai_scheduler/internal/biz/llm_service"
"ai_scheduler/internal/config"
"ai_scheduler/internal/data/constants"
errorcode "ai_scheduler/internal/data/error"
errors "ai_scheduler/internal/data/error"
"ai_scheduler/internal/data/impl"
"ai_scheduler/internal/data/model"
@ -11,6 +12,7 @@ import (
"ai_scheduler/internal/entitys"
"ai_scheduler/internal/gateway"
"ai_scheduler/internal/pkg"
"ai_scheduler/internal/pkg/dingtalk"
"ai_scheduler/internal/pkg/l_request"
"ai_scheduler/internal/pkg/mapstructure"
"ai_scheduler/internal/pkg/rec_extra"
@ -28,16 +30,19 @@ import (
"strings"
"github.com/coze-dev/coze-go"
"github.com/gofiber/fiber/v2/log"
"gorm.io/gorm/utils"
)
type Handle struct {
Ollama *llm_service.OllamaService
toolManager *tools.Manager
conf *config.Config
sessionImpl *impl.SessionImpl
workflowManager *runtime.Registry
dingtalkOldClient *dingtalk.OldClient
dingtalkContactClient *dingtalk.ContactClient
dingtalkNotableClient *dingtalk.NotableClient
}
func NewHandle(
@ -45,16 +50,20 @@ func NewHandle(
toolManager *tools.Manager,
conf *config.Config,
sessionImpl *impl.SessionImpl,
workflowManager *runtime.Registry,
dingtalkOldClient *dingtalk.OldClient,
dingtalkContactClient *dingtalk.ContactClient,
dingtalkNotableClient *dingtalk.NotableClient,
) *Handle {
return &Handle{
Ollama: Ollama,
toolManager: toolManager,
conf: conf,
sessionImpl: sessionImpl,
workflowManager: workflowManager,
dingtalkOldClient: dingtalkOldClient,
dingtalkContactClient: dingtalkContactClient,
dingtalkNotableClient: dingtalkNotableClient,
}
}
@ -119,10 +128,12 @@ func (r *Handle) HandleMatch(ctx context.Context, client *gateway.Client, rec *e
switch constants.TaskType(pointTask.Type) {
case constants.TaskTypeApi:
return r.handleApiTask(ctx, rec, pointTask)
case constants.TaskTypeFunc:
return r.handleTask(ctx, rec, pointTask)
case constants.TaskTypeKnowle:
return r.handleKnowle(ctx, rec, pointTask)
case constants.TaskTypeFunc:
return r.handleTask(ctx, rec, pointTask)
case constants.TaskTypeBot:
return r.handleBot(ctx, rec, pointTask)
case constants.TaskTypeEinoWorkflow:
return r.handleEinoWorkflow(ctx, rec, pointTask)
case constants.TaskTypeCozeWorkflow:
@ -235,6 +246,87 @@ func (r *Handle) handleKnowle(ctx context.Context, rec *entitys.Recognize, task
return
}
// bot 临时实现，后续转到 eino 工作流
func (r *Handle) handleBot(ctx context.Context, rec *entitys.Recognize, task *model.AiTask) (err error) {
if task.Index == "bug_optimization_submit" {
// Ext 中获取 sessionId
sessionID := rec.GetSession()
// 获取dingtalk accessToken
accessToken, _ := r.dingtalkOldClient.GetAccessToken()
// 获取创建者 dingtalk unionId
unionId := r.getUserDingtalkUnionId(ctx, accessToken, sessionID)
// 附件url
var attachmentUrl string
for _, file := range rec.UserContent.File {
attachmentUrl = file.FileUrl
break
}
recordId, err := r.dingtalkNotableClient.InsertRecord(accessToken, &dingtalk.InsertRecordReq{
BaseId: r.conf.Dingtalk.TableDemand.BaseId,
SheetIdOrName: r.conf.Dingtalk.TableDemand.SheetIdOrName,
// OperatorId: tool_callback.BotBugOptimizationSubmitAdminUnionId,
OperatorId: unionId,
CreatorUnionId: unionId,
Content: rec.UserContent.Text,
AttachmentUrl: attachmentUrl,
})
if err != nil {
errCode := r.dingtalkNotableClient.GetHTTPStatus(err)
// 权限不足
if errCode == 403 {
return errorcode.ForbiddenErr("您当前没有AI需求表编辑权限，请联系管理员添加权限")
}
return err
}
if recordId == "" {
return errors.NewBusinessErr(422, "创建记录失败，请联系管理员")
}
// 构建跳转链接
detailPage := util.BuildJumpLink(r.conf.Dingtalk.TableDemand.Url, "去查看")
entitys.ResText(rec.Ch, "", fmt.Sprintf("问题已记录，正在分配相关人员处理，请您耐心等待处理结果。点击查看工单进度：%s", detailPage))
return nil
}
return errors.NewBusinessErr(422, "bot 任务未实现")
}
// getUserDingtalkUnionId 获取用户的 dingtalk unionId
func (r *Handle) getUserDingtalkUnionId(ctx context.Context, accessToken, sessionID string) (unionId string) {
// 查询用户名
session, has, err := r.sessionImpl.FindOne(r.sessionImpl.WithSessionId(sessionID))
if err != nil || !has {
log.Warnf("session not found: %s", sessionID)
return
}
creatorName := session.UserName
// 获取创建者uid 用户名 -> dingtalk uid
creatorId, err := r.dingtalkContactClient.SearchUserOne(accessToken, creatorName)
if err != nil {
log.Warnf("search dingtalk user one failed: %v", err)
return
}
// 获取用户详情 dingtalk uid -> dingtalk unionId
userDetails, err := r.dingtalkOldClient.QueryUserDetails(ctx, creatorId)
if err != nil {
log.Warnf("query user dingtalk details failed: %v", err)
return
}
if userDetails == nil {
log.Warnf("user details not found: %s", creatorId)
return
}
unionId = userDetails.UnionID
return
}
func (r *Handle) handleApiTask(ctx context.Context, rec *entitys.Recognize, task *model.AiTask) (err error) {
var (
request l_request.Request

15
internal/config/config.go
View File

@ -23,6 +23,7 @@ type Config struct {
PermissionConfig PermissionConfig `mapstructure:"permissionConfig"`
LLM LLM `mapstructure:"llm"`
// DingTalkBots map[string]*DingTalkBot `mapstructure:"ding_talk_bots"`
Dingtalk DingtalkConfig `mapstructure:"dingtalk"`
}
type SysPrompt struct {
@ -61,6 +62,20 @@ type LLMCapabilityConfig struct {
Parameters LLMParameters `mapstructure:"parameters"`
}
// DingtalkConfig 钉钉配置
type DingtalkConfig struct {
ApiKey string `mapstructure:"api_key"`
ApiSecret string `mapstructure:"api_secret"`
TableDemand AITableConfig `mapstructure:"table_demand"`
}
// TableDemandConfig 需求表配置
type AITableConfig struct {
Url string `mapstructure:"url"`
BaseId string `mapstructure:"base_id"`
SheetIdOrName string `mapstructure:"sheet_id_or_name"`
}
// SysConfig 系统配置
type SysConfig struct {
SessionLen int `mapstructure:"session_len"`

5
internal/data/error/error_code.go
View File

@ -5,6 +5,7 @@ import "fmt"
var (
Success = &BusinessErr{code: 200, message: "成功"}
ParamError = &BusinessErr{code: 401, message: "参数错误"}
ForbiddenError = &BusinessErr{code: 403, message: "权限不足"}
NotFoundError = &BusinessErr{code: 404, message: "请求地址未找到"}
SystemError = &BusinessErr{code: 405, message: "系统错误"}
@ -67,3 +68,7 @@ func (e *BusinessErr) Wrap(err error) *BusinessErr {
func WorkflowErr(message string) *BusinessErr {
return NewBusinessErr(WorkflowError.code, message)
}
func ForbiddenErr(message string) *BusinessErr {
return NewBusinessErr(ForbiddenError.code, message)
}

170
internal/domain/workflow/zltx/bug_optimization_submit.bak.go Normal file
View File

@ -0,0 +1,170 @@
package zltx
import (
"context"
"encoding/json"
"errors"
"time"
"ai_scheduler/internal/domain/component/callback"
"ai_scheduler/internal/domain/repo"
"ai_scheduler/internal/domain/workflow/runtime"
"ai_scheduler/internal/entitys"
"ai_scheduler/internal/pkg"
"ai_scheduler/internal/pkg/l_request"
"github.com/cloudwego/eino/compose"
"github.com/google/uuid"
"github.com/redis/go-redis/v9"
)
const WorkflowIDBugOptimizationSubmitBak = "bug_optimization_submit_bak"
func init() {
runtime.Register(WorkflowIDBugOptimizationSubmitBak, func(d *runtime.Deps) (runtime.Workflow, error) {
// 从 Deps.Repos 获取 SessionRepo
return &bugOptimizationSubmitBak{
manager: d.Component.Callback,
sessionRepo: d.Repos.Session,
}, nil
})
}
type bugOptimizationSubmitBak struct {
manager callback.Manager
sessionRepo repo.SessionRepo
redisCli *redis.Client
}
func (w *bugOptimizationSubmitBak) ID() string {
return WorkflowIDBugOptimizationSubmitBak
}
type BugOptimizationSubmitBakInput struct {
Ch chan entitys.Response
RequireData *entitys.Recognize
}
type BugOptimizationSubmitBakOutput struct {
Msg string
}
type contextWithTaskBak struct {
Input *BugOptimizationSubmitBakInput
TaskID string
}
func (w *bugOptimizationSubmitBak) Invoke(ctx context.Context, recognize *entitys.Recognize) (map[string]any, error) {
chain, err := w.buildWorkflow(ctx)
if err != nil {
return nil, err
}
input := &BugOptimizationSubmitBakInput{
Ch: recognize.Ch,
RequireData: recognize,
}
out, err := chain.Invoke(ctx, input)
if err != nil {
return nil, err
}
return map[string]any{"msg": out.Msg}, nil
}
func (w *bugOptimizationSubmitBak) buildWorkflow(ctx context.Context) (compose.Runnable[*BugOptimizationSubmitBakInput, *BugOptimizationSubmitBakOutput], error) {
c := compose.NewChain[*BugOptimizationSubmitBakInput, *BugOptimizationSubmitBakOutput]()
// Node 1: Prepare and Call
c.AppendLambda(compose.InvokableLambda(w.prepareAndCall))
// Node 2: Wait
c.AppendLambda(compose.InvokableLambda(w.waitCallback))
return c.Compile(ctx)
}
func (w *bugOptimizationSubmitBak) prepareAndCall(ctx context.Context, in *BugOptimizationSubmitBakInput) (*contextWithTaskBak, error) {
// 生成 TaskID
taskID := uuid.New().String()
// Ext 中获取 sessionId
sessionID := in.RequireData.GetSession()
// 注册回调映射
if err := w.manager.Register(ctx, taskID, sessionID); err != nil {
return nil, err
}
// 查询用户名
userName := "unknown"
if w.sessionRepo != nil {
name, err := w.sessionRepo.GetUserName(ctx, sessionID)
if err == nil && name != "" {
userName = name
}
}
// 构建请求参数
var fileUrls, fileContent string
if len(in.RequireData.UserContent.File) > 0 {
for _, file := range in.RequireData.UserContent.File {
fileUrls += file.FileUrl + ","
fileContent += file.FileRec + ","
}
fileUrls = fileUrls[:len(fileUrls)-1]
fileContent = fileContent[:len(fileContent)-1]
}
body := map[string]string{
"mark": in.RequireData.Match.Index,
"text": in.RequireData.UserContent.Text,
"img": fileUrls,
"img_content": fileContent,
"creator": userName,
"task_id": taskID,
}
request := l_request.Request{
Url: "https://connector.dingtalk.com/webhook/flow/10352c521dd02104cee9000c",
Method: "POST",
Headers: map[string]string{
"Content-Type": "application/json",
},
JsonByte: pkg.JsonByteIgonErr(body),
}
res, err := request.Send()
if err != nil {
return nil, err
}
var data map[string]any
if err := json.Unmarshal(res.Content, &data); err != nil {
return nil, err
}
if success, ok := data["success"].(bool); !ok || !success {
return nil, errors.New("dingtalk flow failed")
}
entitys.ResLog(in.Ch, in.RequireData.Match.Index, "问题记录中")
entitys.ResLoading(in.Ch, in.RequireData.Match.Index, "问题记录中...")
return &contextWithTaskBak{Input: in, TaskID: taskID}, nil
}
func (w *bugOptimizationSubmitBak) waitCallback(ctx context.Context, in *contextWithTask) (*BugOptimizationSubmitBakOutput, error) {
// 阻塞等待回调信号
// 设置 5 分钟超时
waitCtx, cancel := context.WithTimeout(ctx, 5*time.Minute)
defer cancel()
res, err := w.manager.Wait(waitCtx, in.TaskID, 5*time.Minute)
if err != nil {
return nil, err
}
return &BugOptimizationSubmitBakOutput{Msg: res}, nil
}

69
internal/pkg/dingtalk/notable_client.go
View File

@ -3,6 +3,8 @@ package dingtalk
import (
"ai_scheduler/internal/config"
errorcode "ai_scheduler/internal/data/error"
"encoding/json"
"time"
openapi "github.com/alibabacloud-go/darabonba-openapi/v2/client"
notable "github.com/alibabacloud-go/dingtalk/notable_1_0"
@ -72,3 +74,70 @@ func (c *NotableClient) UpdateRecord(accessToken string, req *UpdateRecordReq) (
return true, nil
}
type InsertRecordReq struct {
BaseId string
SheetIdOrName string
OperatorId string
CreatorUnionId string
Content string
AttachmentUrl string
}
func (c *NotableClient) InsertRecord(accessToken string, req *InsertRecordReq) (string, error) {
// 默认使用“数据表”
if req.SheetIdOrName == "" {
req.SheetIdOrName = "数据表"
}
headers := &notable.InsertRecordsHeaders{}
headers.XAcsDingtalkAccessToken = tea.String(accessToken)
resp, err := c.cli.InsertRecordsWithOptions(
tea.String(req.BaseId),
tea.String(req.SheetIdOrName),
&notable.InsertRecordsRequest{
OperatorId: tea.String(req.OperatorId),
Records: []*notable.InsertRecordsRequestRecords{
{
Fields: map[string]any{
"创建日期": time.Now().Format(time.DateTime),
"需求内容": req.Content,
"提交人": []map[string]any{
{
"unionId": req.CreatorUnionId,
},
},
"附件": map[string]any{
"link": req.AttachmentUrl,
},
},
},
},
}, headers, &util.RuntimeOptions{})
if err != nil {
return "", err
}
if resp.Body == nil || resp.Body.Value == nil || len(resp.Body.Value) == 0 {
return "", errorcode.ParamErrf("empty response body")
}
return *resp.Body.Value[0].Id, nil
}
func (c *NotableClient) GetHTTPStatus(err error) int {
if sdkErr, ok := err.(*tea.SDKError); ok {
if sdkErr.StatusCode != nil {
return *sdkErr.StatusCode
}
if sdkErr.Data != nil {
var m struct {
StatusCode int `json:"statusCode"`
}
if json.Unmarshal([]byte(*sdkErr.Data), &m) == nil {
return m.StatusCode
}
}
}
return 0 // 0 = 非 HTTP 错误
}