wechat-api/internal/pkg/customer_of_im/channel/channel.go

package channel

import (
	"context"
	"encoding/json"
	"github.com/openai/openai-go"
	"github.com/zeromicro/go-zero/core/logx"
	"math/rand"
	"time"
	"wechat-api/internal/pkg/customer_of_im/proto"
	"wechat-api/internal/pkg/wechat_ws"
	"wechat-api/internal/svc"
)

type IChannel interface {
	OnMessage(msg *wechat_ws.MsgJsonObject) error                       // channel的入口，接收消息
	FriendTalkHandle(msgStr string) error                               // 其他人发送过来的消息处理
	TextToChatMessage(msgStr string) (*ChatMessage, error)              // 文本转ChatMessage
	GenerateRelevantContext(msg *ChatMessage) (*RelevantContext, error) // 创建相关上下文
	Send(message *ReplyMessage) error                                   // 发送消息
}

type Channel struct {
	subChannel IChannel
}

func (c *Channel) InitChannel(subChannel IChannel) {
	c.subChannel = subChannel
}

func (c *Channel) After(chatCtx *RelevantContext, message *ChatMessage, reply *ReplyMessage) bool {
	return true
}
func (c *Channel) Before(chatCtx *RelevantContext, message *ChatMessage, reply *ReplyMessage) bool {
	return true
}

func (c *Channel) randomPause() {
	// 创建一个新的随机数源
	source := rand.NewSource(time.Now().UnixNano())
	// 创建一个新的随机数生成器
	rng := rand.New(source)

	// 生成 2 到 5 之间的随机整数
	pauseDuration := 2 + rng.Intn(4) // 2 + (0, 1, 2, 3) => 2, 3, 4, 5

	// 暂停程序执行相应的时间
	time.Sleep(time.Duration(pauseDuration) * time.Second)
}

func (c *Channel) beforeSendReply(reply *ReplyMessage) error {
	jsonReplay := make([]*ReplyJson, 0)
	err := json.Unmarshal([]byte(reply.Content), &jsonReplay)
	if err == nil && len(jsonReplay) > 0 {
		for i, item := range jsonReplay {
			if item.Content == "" {
				continue
			}
			switch item.Type {
			case "TEXT":
				reply.Content = item.Content
				_ = c.beforeSendReply(reply)
			default:
				logx.Error("未知类型", item.Type)
			}
			// 随机暂停
			if i < len(jsonReplay)-1 {
				c.randomPause()
			}
		}
		return nil
	}

	err = c.subChannel.Send(reply)

	return err
}

func (c *Channel) handle(chatCtx *RelevantContext, message *ChatMessage, reply *ReplyMessage) error {
	chat := NewChatEngine(chatCtx.BaseURL, chatCtx.APIKey, chatCtx.IsFastGPT)

	replyText, err := chat.ChatCompletions(context.Background(), openai.ChatModelGPT4, message.Content, message.UserId)
	if err != nil {
		logx.Error(err)
		return err
	}

	reply.Content = replyText

	err = c.beforeSendReply(reply)

	return err
}

func (c *Channel) Produce(svcCtx *svc.ServiceContext, chatCtx *RelevantContext, message *ChatMessage) {
	reply := &ReplyMessage{
		ChatMessage: &ChatMessage{
			UserId:           message.UserId,
			MsgId:            svcCtx.NodeID.Generate().String(),
			CreateTime:       time.Time{},
			Type:             proto.EnumContentType_Text,
			Content:          "",
			FromUserId:       message.ToUserId,
			FromUserNickname: message.ToUserNickname,
			ToUserId:         message.FromUserId,
			ToUserNickname:   message.FromUserNickname,
			GroupId:          message.GroupId,
			GroupName:        message.GroupName,
			IsGroup:          message.IsGroup,
			IsAt:             message.IsAt,
		},
	}

	// 进行前置处理
	ret := c.Before(chatCtx, message, reply)
	if !ret {
		return
	}

	// 可以使用 https://github.com/tmc/langchaingo 来进行会话记忆
	err := c.handle(chatCtx, message, reply)
	if err != nil {
		logx.Error(err)
		return
	}

	_ = c.After(chatCtx, message, reply)

}