ai_scheduler/internal/biz/advice_strategy.go

542 lines
17 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package biz
import (
"ai_scheduler/internal/biz/llm_service/third_party"
"ai_scheduler/internal/config"
"ai_scheduler/internal/data/constants"
"ai_scheduler/internal/data/impl"
dbmodel "ai_scheduler/internal/data/model"
"ai_scheduler/internal/data/mongo_model"
"ai_scheduler/internal/entitys"
"ai_scheduler/internal/pkg"
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"sync"
"time"
"github.com/sashabaranov/go-openai"
"go.mongodb.org/mongo-driver/bson"
"go.mongodb.org/mongo-driver/mongo/options"
"xorm.io/builder"
)
// 策略动作(AdvicerStrategyTestRes.Action)
const (
StrategyActionReply = "reply" // 已生成回复(意向客户)
StrategyActionActivityPush = "activity_push" // 活动推送(沉睡客户)
StrategyActionRegularWait = "regular_wait" // 熟客:AI 暂不介入,等待人工/超时兜底
StrategyActionSkip = "skip" // 跳过(无素材/开关关闭等)
StrategyActionNotBound = "not_bound" // 未绑定客户
StrategyActionAiOff = "ai_off" // 客户 AI 参与开关关闭
StrategyActionNonClient = "non_client" // 非客户:AI 不参与回复
)
// AdviceStrategyBiz 四级客户回复策略引擎
// 熟客:AI 不主动介入(人工接管),仅超时未回复时礼貌兜底(RegularTimeoutScan);
// 意向客户:及时回复、跟进邀约、活动推荐;沉睡客户:只推活动并尝试邀约;非客户:不参与。
type AdviceStrategyBiz struct {
cfg *config.Config
openai *third_party.OpenAi
mongo *pkg.Mongo
clientBiz *AdviceClientBiz
wxMsgMongo *mongo_model.AdvicerWxMsgMongo
versionMongo *mongo_model.AdvicerVersionMongo
projectImpl *impl.AdviceProjectImpl
modelSupImpl *impl.AiAdviceModelSupImpl
activityBiz *AdviceActivityBiz
sendBiz *AdviceWxSendBiz
locks sync.Map // 客户 wxid 维度互斥锁(防同一客户并发回复)
}
func NewAdviceStrategyBiz(
cfg *config.Config,
openai *third_party.OpenAi,
mongo *pkg.Mongo,
clientBiz *AdviceClientBiz,
wxMsgMongo *mongo_model.AdvicerWxMsgMongo,
versionMongo *mongo_model.AdvicerVersionMongo,
projectImpl *impl.AdviceProjectImpl,
modelSupImpl *impl.AiAdviceModelSupImpl,
activityBiz *AdviceActivityBiz,
sendBiz *AdviceWxSendBiz,
) *AdviceStrategyBiz {
return &AdviceStrategyBiz{
cfg: cfg,
openai: openai,
mongo: mongo,
clientBiz: clientBiz,
wxMsgMongo: wxMsgMongo,
versionMongo: versionMongo,
projectImpl: projectImpl,
modelSupImpl: modelSupImpl,
activityBiz: activityBiz,
sendBiz: sendBiz,
}
}
// ==================== 回调链入口 ====================
// OnCustomerMessage 客户消息回调触发(异步调用):按客户等级执行回复策略。
// 非客户/AI 关闭不回复;熟客不介入(由超时兜底任务处理);沉睡客户仅推活动;意向客户(含未评估)生成回复并发送。
func (a *AdviceStrategyBiz) OnCustomerMessage(ctx context.Context, wxid, content string) {
if a == nil || !a.cfg.Advicer.AutoReply {
return
}
// 同一客户正在处理中则跳过(后续消息会在下一轮处理),避免重叠回复
if ok, unlock := a.lockWxid(wxid); !ok {
return
} else {
defer unlock()
}
client, found, err := a.clientBiz.FindByWxid(ctx, wxid)
if err != nil || !found {
return
}
c := &client
res, err := a.Decide(ctx, c, content)
if err != nil {
return
}
if res.Action != StrategyActionReply && res.Action != StrategyActionActivityPush {
return
}
if len(res.Replies) == 0 {
return
}
if _, err = a.sendBiz.SendMulti(ctx, c.AppId, c.Wxid, res.Replies, mongo_model.WxMsgSourceWeb); err == nil {
res.Sent = true
}
}
// ==================== 策略决策 ====================
// Decide 策略决策 + 生成回复内容(只生成不发送)
func (a *AdviceStrategyBiz) Decide(ctx context.Context, c *mongo_model.AdvicerClientItem, customerMsg string) (*entitys.AdvicerStrategyTestRes, error) {
res := &entitys.AdvicerStrategyTestRes{
Level: c.ClientLevel,
LevelDesc: mongo_model.ClientLevelDesc[c.ClientLevel],
Replies: []string{},
}
if !c.AiEnabled {
res.Action = StrategyActionAiOff
res.Reason = "客户 AI 参与开关已关闭,不回复"
return res, nil
}
switch c.ClientLevel {
case mongo_model.ClientLevelNon:
res.Action = StrategyActionNonClient
res.Reason = "非客户:AI 不参与回复"
return res, nil
case mongo_model.ClientLevelRegular:
res.Action = StrategyActionRegularWait
res.Reason = fmt.Sprintf("熟客:AI 不主动介入,超 %d 分钟未回复时执行礼貌兜底", a.delayMinutes())
return res, nil
case mongo_model.ClientLevelSleeping:
// 沉睡客户:只推活动,并尝试邀约
acts, _ := a.activityBiz.ActiveList(ctx, c.ProjectId, mongo_model.ClientLevelSleeping)
if len(acts) == 0 {
res.Action = StrategyActionSkip
res.Reason = "沉睡客户:当前无生效活动,暂不回复"
return res, nil
}
replies, err := a.GenerateReply(ctx, c, constants.StrategySleepingActivityMission, buildActivityCtx(acts[0].AdvicerActivityMongo), customerMsg)
if err != nil {
return res, err
}
res.Action = StrategyActionActivityPush
res.Reason = "沉睡客户:推送活动并尝试邀约"
res.Replies = replies
return res, nil
}
// 意向客户(未评估客户按意向处理):及时回复
replies, err := a.GenerateReply(ctx, c, constants.StrategyReplyMission, "", customerMsg)
if err != nil {
return res, err
}
res.Action = StrategyActionReply
res.Reason = "意向客户:及时回复并跟进"
res.Replies = replies
return res, nil
}
// ==================== 熟客超时兜底 ====================
// RegularTimeoutScan 熟客超时兜底扫描:客户最后发言后超过配置时长(默认 10 分钟)仍未回复时,
// 生成礼貌性回复并简单解答(人工已回复/已由 AI 回复的不会重复触发)。
// force=true 时忽略 AutoReply 总开关(手动触发)。
func (a *AdviceStrategyBiz) RegularTimeoutScan(ctx context.Context, force bool) (sent int, err error) {
if !force && !a.cfg.Advicer.AutoReply {
return 0, nil
}
delay := time.Duration(a.delayMinutes()) * time.Minute
now := time.Now()
cursor, err := a.mongo.Co(a.clientBiz.AdvicerClientMongo).Find(ctx, bson.M{
"clientLevel": mongo_model.ClientLevelRegular,
"aiEnabled": true,
"wxid": bson.M{"$ne": ""},
})
if err != nil {
return 0, err
}
var clients []mongo_model.AdvicerClientItem
for cursor.Next(ctx) {
var item mongo_model.AdvicerClientItem
if e := cursor.Decode(&item); e != nil {
return sent, e
}
clients = append(clients, item)
}
if err = cursor.Err(); err != nil {
return sent, err
}
for i := range clients {
item := clients[i]
lastMsg := item.LastClientMsgAt
if lastMsg.IsZero() {
continue
}
// 客户最后消息之后已有回复(人工或 AI),无需兜底
if !lastMsg.After(item.LastReplyAt) {
continue
}
du := now.Sub(lastMsg)
if du < delay || du > 24*time.Hour {
// 未到兜底时间;超过 24 小时的陈年消息不再兜底
continue
}
// 该客户处理中则跳过,下一轮再试
ok, unlock := a.lockWxid(item.Wxid)
if !ok {
continue
}
replies, e := a.GenerateReply(ctx, &item, constants.StrategyRegularTimeoutMission, "", "")
if e != nil || len(replies) == 0 {
unlock()
continue
}
if _, e = a.sendBiz.SendMulti(ctx, item.AppId, item.Wxid, replies, mongo_model.WxMsgSourceWeb); e == nil {
sent++
}
unlock()
}
return sent, nil
}
// ==================== 内容生成 ====================
// GenerateReply 生成拟人化发送内容(销售人设 + 客户画像 + 聊天记录 + 场景任务)。
// mission 场景任务说明;extraCtx 场景素材(如活动信息,可空);trigger 作为 user 输入(空则使用默认指令)。
func (a *AdviceStrategyBiz) GenerateReply(ctx context.Context, c *mongo_model.AdvicerClientItem, mission, extraCtx, trigger string) ([]string, error) {
if a == nil || a.openai == nil {
return nil, errors.New("LLM 未初始化")
}
rule, model := projectModelOf(ctx, a.projectImpl, a.modelSupImpl, c.ProjectId)
if model == nil {
return nil, errors.New("项目未配置可用模型(请检查 modelSupId)")
}
// 系统提示:基础人设 + 场景任务 + 风控红线 + 输出约束
var sys strings.Builder
sys.WriteString(constants.BasePrompt)
sys.WriteString(mission)
sys.WriteString("\n")
if rule = strings.TrimSpace(rule); rule != "" {
sys.WriteString(constants.RulePromptPrefix)
sys.WriteString(rule)
sys.WriteString("\n")
}
sys.WriteString(constants.StrategyOutputHint)
messages := []openai.ResponseInputMessage{
{Role: openai.ChatMessageRoleSystem, Content: sys.String()},
}
// 销售人设风格(与 chat 会话共用编译逻辑,保证各链路语气一致)
if persona := formatPersonaPrompt(a.versionOf(ctx, c.AdvicerId)); persona != "" {
messages = append(messages, openai.ResponseInputMessage{
Role: openai.ChatMessageRoleSystem,
Content: persona,
})
}
messages = append(messages, openai.ResponseInputMessage{
Role: openai.ChatMessageRoleSystem,
Content: "[客户画像]\n" + pkg.JsonStringIgonErr(c.Entity()),
})
messages = append(messages, openai.ResponseInputMessage{
Role: openai.ChatMessageRoleSystem,
Content: "[最近聊天记录(按时间先后)]\n" + formatWxMsgHis(a.recentWxMsgs(ctx, c.Wxid, 20)),
})
if strings.TrimSpace(extraCtx) != "" {
messages = append(messages, openai.ResponseInputMessage{
Role: openai.ChatMessageRoleSystem,
Content: extraCtx,
})
}
if strings.TrimSpace(trigger) == "" {
trigger = "请根据以上信息,生成你要发送给客户的内容。"
}
messages = append(messages, openai.ResponseInputMessage{
Role: openai.ChatMessageRoleUser,
Content: trigger,
})
resp, err := a.openai.CreateResponseMessages(ctx, model.Key, model.URL, model.ChatModel, messages, "")
if err != nil {
return nil, fmt.Errorf("生成回复失败: %w", err)
}
return splitReplies(resp.GetOutputText()), nil
}
// ==================== 手动触发 ====================
// Execute 对客户最新消息执行完整策略(决策→生成→发送),供后台手动触发
func (a *AdviceStrategyBiz) Execute(ctx context.Context, wxid string, force bool) (*entitys.AdvicerStrategyTestRes, error) {
res := &entitys.AdvicerStrategyTestRes{Replies: []string{}}
if !force && !a.cfg.Advicer.AutoReply {
res.Action = StrategyActionSkip
res.Reason = "AI 自动回复总开关未开启(advicer.auto_reply)"
return res, nil
}
client, found, err := a.clientBiz.FindByWxid(ctx, wxid)
if err != nil {
return res, err
}
if !found {
res.Action = StrategyActionNotBound
res.Reason = "该 wxid 未绑定客户"
return res, nil
}
ok, unlock := a.lockWxid(wxid)
if !ok {
res.Action = StrategyActionSkip
res.Reason = "该客户正在处理中,请稍后再试"
return res, nil
}
defer unlock()
c := &client
res, err = a.Decide(ctx, c, a.latestCustomerMsg(ctx, wxid))
if err != nil {
return res, err
}
if len(res.Replies) > 0 {
sent, e := a.sendBiz.SendMulti(ctx, c.AppId, c.Wxid, res.Replies, mongo_model.WxMsgSourceWeb)
if e != nil {
return res, e
}
res.Sent = len(sent) > 0
}
return res, nil
}
// Test 策略干跑:模拟客户消息返回决策与生成内容;Send=true 时实际发送
func (a *AdviceStrategyBiz) Test(ctx context.Context, param *entitys.AdvicerStrategyTestReq) (*entitys.AdvicerStrategyTestRes, error) {
res := &entitys.AdvicerStrategyTestRes{Replies: []string{}}
client, found, err := a.clientBiz.FindByWxid(ctx, param.Wxid)
if err != nil {
return res, err
}
if !found {
res.Action = StrategyActionNotBound
res.Reason = "该 wxid 未绑定客户"
return res, nil
}
c := &client
content := strings.TrimSpace(param.Content)
if content == "" {
content = a.latestCustomerMsg(ctx, param.Wxid)
}
res, err = a.Decide(ctx, c, content)
if err != nil {
return res, err
}
if param.Send && len(res.Replies) > 0 {
ok, unlock := a.lockWxid(param.Wxid)
if !ok {
res.Reason += ";该客户正在处理中,未执行发送"
return res, nil
}
defer unlock()
sent, e := a.sendBiz.SendMulti(ctx, c.AppId, c.Wxid, res.Replies, mongo_model.WxMsgSourceWeb)
if e != nil {
return res, e
}
res.Sent = len(sent) > 0
}
return res, nil
}
// ==================== 内部工具 ====================
// delayMinutes 熟客超时兜底延迟(分钟,默认 10)
func (a *AdviceStrategyBiz) delayMinutes() int {
d := a.cfg.Advicer.RegularReplyDelayMinutes
if d <= 0 {
d = 10
}
return d
}
// lockWxid 尝试获取客户维度互斥锁(忙则 ok=false,表示该客户正在处理中)
func (a *AdviceStrategyBiz) lockWxid(wxid string) (ok bool, unlock func()) {
if len(wxid) == 0 {
return true, func() {}
}
v, _ := a.locks.LoadOrStore(wxid, &sync.Mutex{})
mu := v.(*sync.Mutex)
if !mu.TryLock() {
return false, nil
}
return true, mu.Unlock
}
// versionOf 取销售最新版本人设(无则返回 nil,跳过人设注入)
func (a *AdviceStrategyBiz) versionOf(ctx context.Context, advicerId int32) *mongo_model.AdvicerVersionMongoEntity {
if advicerId == 0 {
return nil
}
res := a.mongo.Co(a.versionMongo).FindOne(ctx, bson.M{"advicerId": advicerId},
options.FindOne().SetSort(bson.D{{Key: "lastUpdateTime", Value: -1}}))
if res.Err() != nil {
return nil
}
var v mongo_model.AdvicerVersionMongo
if err := res.Decode(&v); err != nil {
return nil
}
return v.Entity()
}
// recentWxMsgs 拉取客户最近聊天记录(时间正序返回)
func (a *AdviceStrategyBiz) recentWxMsgs(ctx context.Context, wxid string, limit int64) (list []mongo_model.AdvicerWxMsgMongo) {
if len(wxid) == 0 {
return nil
}
cursor, err := a.mongo.Co(a.wxMsgMongo).Find(ctx, bson.M{"wxid": wxid},
options.Find().SetSort(bson.D{{Key: "createAt", Value: -1}}).SetLimit(limit))
if err != nil {
return nil
}
for cursor.Next(ctx) {
var m mongo_model.AdvicerWxMsgMongo
if err := cursor.Decode(&m); err != nil {
return nil
}
list = append(list, m)
}
// 倒序查询后反转回时间正序
for i, j := 0, len(list)-1; i < j; i, j = i+1, j-1 {
list[i], list[j] = list[j], list[i]
}
return list
}
// latestCustomerMsg 取客户最近一条消息内容(策略触发输入)
func (a *AdviceStrategyBiz) latestCustomerMsg(ctx context.Context, wxid string) string {
res := a.mongo.Co(a.wxMsgMongo).FindOne(ctx,
bson.M{"wxid": wxid, "direction": mongo_model.WxMsgDirectionCustomer},
options.FindOne().SetSort(bson.D{{Key: "createAt", Value: -1}}))
if res.Err() != nil {
return ""
}
var m mongo_model.AdvicerWxMsgMongo
if err := res.Decode(&m); err != nil {
return ""
}
return m.Content
}
// ==================== 包级公共工具 ====================
// projectModelOf 查询项目对应的模型配置与风控红线(策略/主动触达/迭代共用;调用频率低不做缓存)
func projectModelOf(ctx context.Context, projectImpl *impl.AdviceProjectImpl, modelSupImpl *impl.AiAdviceModelSupImpl, projectId int32) (rule string, model *dbmodel.AiAdviceModelSup) {
if projectId == 0 {
return "", nil
}
var p dbmodel.AiAdviceProject
cond := builder.NewCond().And(builder.Eq{"project_id": projectId})
if err := projectImpl.GetOneBySearchToStrut(&cond, &p); err != nil || p.ProjectID == 0 {
return "", nil
}
rule = p.RuleDimension
if p.ModelSupID == 0 {
return rule, nil
}
var m dbmodel.AiAdviceModelSup
cond2 := builder.NewCond().And(builder.Eq{"sup_id": p.ModelSupID})
if err := modelSupImpl.GetOneBySearchToStrut(&cond2, &m); err != nil || m.SupID == 0 {
return rule, nil
}
return rule, &m
}
// formatWxMsgHis 将微信消息流水格式化为「客户/我」对话文本
func formatWxMsgHis(list []mongo_model.AdvicerWxMsgMongo) string {
if len(list) == 0 {
return "(暂无聊天记录)"
}
var b strings.Builder
for _, m := range list {
who := "客户"
if m.Direction != mongo_model.WxMsgDirectionCustomer {
who = "我"
}
content := m.Content
if m.MsgType != mongo_model.WxMsgTypeText {
content = "[" + m.MsgType + "消息] " + content
}
b.WriteString(who + ":" + content + "\n")
}
return b.String()
}
// buildActivityCtx 活动素材文本(注入生成提示)
func buildActivityCtx(act mongo_model.AdvicerActivityMongo) string {
var b strings.Builder
b.WriteString("[活动信息]\n")
b.WriteString("活动名称:" + act.Name + "\n")
b.WriteString("活动内容:" + act.Content + "\n")
if !act.StartAt.IsZero() {
b.WriteString("活动时间:" + act.StartAt.Format("2006-01-02") + " 至 ")
if act.EndAt.IsZero() {
b.WriteString("长期有效")
} else {
b.WriteString(act.EndAt.Format("2006-01-02"))
}
b.WriteString("\n")
}
return b.String()
}
// splitReplies 将模型输出拆分为多条待发送消息(兼容偶发 json/代码围栏包裹)
func splitReplies(raw string) []string {
raw = strings.TrimSpace(raw)
// 模型偶发输出 {"result": "..."} 包装:取出 result 字段
if strings.HasPrefix(raw, "{") && strings.Contains(raw, "\"result\"") {
var out struct {
Result string `json:"result"`
}
if err := json.Unmarshal([]byte(extractJsonObject(raw)), &out); err == nil && strings.TrimSpace(out.Result) != "" {
raw = out.Result
}
}
raw = strings.TrimPrefix(raw, "```json")
raw = strings.TrimPrefix(raw, "```")
raw = strings.TrimSuffix(raw, "```")
var replies []string
for _, line := range strings.Split(raw, "\n") {
line = strings.TrimSpace(line)
line = strings.Trim(line, "\"")
line = strings.TrimSpace(line)
if len(line) == 0 {
continue
}
replies = append(replies, line)
}
return replies
}