diff --git a/neo/command/request.go b/neo/command/request.go index 7f4bf2cc..24e751f7 100644 --- a/neo/command/request.go +++ b/neo/command/request.go @@ -3,44 +3,45 @@ package command import ( "fmt" "strings" - "time" "github.com/google/uuid" v8 "github.com/yaoapp/gou/runtime/v8" + "github.com/yaoapp/kun/log" "github.com/yaoapp/kun/maps" - "github.com/yaoapp/kun/utils" + "github.com/yaoapp/yao/neo/conversation" "github.com/yaoapp/yao/neo/message" "rogchap.com/v8go" ) // Run the command -func (req *Request) Run(messages []map[string]interface{}, cb func(msg *message.JSON) int) (interface{}, error) { +func (req *Request) Run(conversation conversation.Conversation, messages []map[string]interface{}, cb func(msg *message.JSON) int) (interface{}, error) { - prompts, err := req.prepare(messages, cb) + content, err := req.prepare(conversation, messages, cb) if err != nil { - cb(req.msg().Text(fmt.Sprintf("Prepare Before Error: %s\n", err.Error()))) - cb(message.New().Done()) - req.Done() + req.error(err, cb) return nil, err } - utils.Dump(prompts) - - cb(req.msg().Text(fmt.Sprintf("- Command: %s\n", req.Command.Name))) - time.Sleep(200 * time.Millisecond) - - cb(req.msg().Text(fmt.Sprintf("- Session: %s\n", req.sid))) - time.Sleep(200 * time.Millisecond) - - cb(req.msg().Text(fmt.Sprintf("- Request: %s\n", req.sid))) - time.Sleep(200 * time.Millisecond) + fmt.Println(string(content)) + // DONE cb(req.msg().Done()) + + // cb(req.msg().Text(fmt.Sprintf("- Command: %s\n", req.Command.Name))) + // time.Sleep(200 * time.Millisecond) + + // cb(req.msg().Text(fmt.Sprintf("- Session: %s\n", req.sid))) + // time.Sleep(200 * time.Millisecond) + + // cb(req.msg().Text(fmt.Sprintf("- Request: %s\n", req.sid))) + // time.Sleep(200 * time.Millisecond) + + // cb(req.msg().Done()) return nil, nil } // RunPrepare the command -func (req *Request) prepare(messages []map[string]interface{}, cb func(msg *message.JSON) int) ([]Prompt, error) { +func (req *Request) prepare(conversation conversation.Conversation, messages []map[string]interface{}, cb func(msg *message.JSON) int) ([]byte, error) { data, err := req.prepareBefore(messages, cb) if err != nil { @@ -60,7 +61,39 @@ func (req *Request) prepare(messages []map[string]interface{}, cb func(msg *mess } } - return prompts, nil + question, err := req.question(messages) + if err != nil { + req.error(err, cb) + return nil, err + } + + chatMessages, err := req.messages(conversation, prompts, question) + if err != nil { + req.error(err, cb) + return nil, err + } + + // chat with AI + content := []byte{} + _, ex := req.AI.ChatCompletionsWith(req.ctx, chatMessages, req.Prepare.Option, func(data []byte) int { + msg := message.NewOpenAI(data) + if msg != nil { + if msg.IsDone() { + return 0 + } + content = msg.Append(content) + cb(req.msg().Text(msg.String())) + } + return 1 + }) + + if ex != nil { + req.error(fmt.Errorf(ex.Message), cb) + return nil, fmt.Errorf("Chat error: %s", ex.Message) + } + defer req.saveHistory(conversation, content, chatMessages) + + return content, nil } // prepareBefore hook @@ -79,6 +112,64 @@ func (req *Request) prepareBefore(messages []map[string]interface{}, cb func(msg return req.runScript(req.Prepare.Before, args, cb) } +// saveHistory save the history +func (req *Request) saveHistory(conversation conversation.Conversation, content []byte, messages []map[string]interface{}) { + + if len(content) > 0 && req.sid != "" && len(messages) > 0 { + err := conversation.SaveRequest( + req.sid, + req.id, + req.Command.ID, + []map[string]interface{}{ + {"role": "user", "content": messages[len(messages)-1]["content"], "name": req.sid}, + {"role": "assistant", "content": string(content), "name": req.sid}, + }, + ) + + if err != nil { + log.Error("Save request error: %s", err.Error()) + } + } +} + +func (req *Request) error(err error, cb func(msg *message.JSON) int) { + cb(req.msg().Text(err.Error())) + cb(message.New().Done()) + req.Done() +} + +func (req *Request) question(messages []map[string]interface{}) (string, error) { + if len(messages) < 1 { + return "", fmt.Errorf("No messages") + } + + question, ok := messages[len(messages)-1]["content"].(string) + if !ok { + return "", fmt.Errorf("messages content is not string") + } + + return question, nil +} + +func (req *Request) messages(conversation conversation.Conversation, prompts []Prompt, question string) ([]map[string]interface{}, error) { + messages := []map[string]interface{}{} + for _, prompt := range prompts { + message := map[string]interface{}{"role": prompt.Role, "content": prompt.Content} + if prompt.Name != "" { + message["name"] = prompt.Name + } + messages = append(messages, message) + } + + history, err := conversation.GetRequest(req.sid, req.id) + if err != nil { + return nil, err + } + messages = append(messages, history...) + messages = append(messages, map[string]interface{}{"role": "user", "content": question, "name": req.sid}) + return messages, nil +} + func (req *Request) runScript(id string, args []interface{}, cb func(msg *message.JSON) int) (map[string]interface{}, error) { namer := strings.Split(id, ".") @@ -116,7 +207,7 @@ func (req *Request) runScript(id string, args []interface{}, cb func(msg *messag cb(req.msg().Text(text)) } - cb(message.New().Done()) + cb(req.msg().Done()) req.Done() return v8go.Null(v8ctx.Isolate()) }) diff --git a/neo/conversation/mongo.go b/neo/conversation/mongo.go index 415dc3a5..72f263f7 100644 --- a/neo/conversation/mongo.go +++ b/neo/conversation/mongo.go @@ -17,3 +17,13 @@ func (conv *Mongo) GetHistory(sid string) ([]map[string]interface{}, error) { func (conv *Mongo) SaveHistory(sid string, messages []map[string]interface{}) error { return nil } + +// GetRequest get the request +func (conv *Mongo) GetRequest(sid string, rid string) ([]map[string]interface{}, error) { + return nil, nil +} + +// SaveRequest save the request +func (conv *Mongo) SaveRequest(sid string, rid string, cid string, messages []map[string]interface{}) error { + return nil +} diff --git a/neo/conversation/redis.go b/neo/conversation/redis.go index 6481c21f..9c478fc5 100644 --- a/neo/conversation/redis.go +++ b/neo/conversation/redis.go @@ -17,3 +17,13 @@ func (conv *Redis) GetHistory(sid string) ([]map[string]interface{}, error) { func (conv *Redis) SaveHistory(sid string, messages []map[string]interface{}) error { return nil } + +// GetRequest get the request +func (conv *Redis) GetRequest(sid string, rid string) ([]map[string]interface{}, error) { + return nil, nil +} + +// SaveRequest save the request +func (conv *Redis) SaveRequest(sid string, rid string, cid string, messages []map[string]interface{}) error { + return nil +} diff --git a/neo/conversation/types.go b/neo/conversation/types.go index a6a5ca70..5e3575ff 100644 --- a/neo/conversation/types.go +++ b/neo/conversation/types.go @@ -7,3 +7,11 @@ type Setting struct { MaxSize int `json:"max_size,omitempty" yaml:"max_size,omitempty"` TTL int `json:"ttl,omitempty" yaml:"ttl,omitempty"` } + +// Conversation the store interface +type Conversation interface { + GetHistory(sid string) ([]map[string]interface{}, error) + SaveHistory(sid string, messages []map[string]interface{}) error + GetRequest(sid string, rid string) ([]map[string]interface{}, error) + SaveRequest(sid string, rid string, cid string, messages []map[string]interface{}) error +} diff --git a/neo/conversation/weaviate.go b/neo/conversation/weaviate.go index 88542d54..55d3163f 100644 --- a/neo/conversation/weaviate.go +++ b/neo/conversation/weaviate.go @@ -17,3 +17,13 @@ func (conv *Weaviate) GetHistory(sid string) ([]map[string]interface{}, error) { func (conv *Weaviate) SaveHistory(sid string, messages []map[string]interface{}) error { return nil } + +// GetRequest get the request +func (conv *Weaviate) GetRequest(sid string, rid string) ([]map[string]interface{}, error) { + return nil, nil +} + +// SaveRequest save the request +func (conv *Weaviate) SaveRequest(sid string, rid string, cid string, messages []map[string]interface{}) error { + return nil +} diff --git a/neo/conversation/xun.go b/neo/conversation/xun.go index cc15bce3..6844cc69 100644 --- a/neo/conversation/xun.go +++ b/neo/conversation/xun.go @@ -23,6 +23,8 @@ type row struct { Name string `json:"name"` Content string `json:"content"` Sid string `json:"sid"` + Rid string `json:"rid"` + Cid string `json:"cid"` ExpiredAt interface{} `json:"expired_at"` } @@ -66,6 +68,7 @@ func (conv *Xun) GetHistory(sid string) ([]map[string]interface{}, error) { qb := conv.query.Table(conv.setting.Table). Select("role", "name", "content"). Where("sid", sid). + Where("cid", ""). OrderBy("id", "desc") if conv.setting.TTL > 0 { @@ -122,6 +125,71 @@ func (conv *Xun) SaveHistory(sid string, messages []map[string]interface{}) erro return conv.query.Table(conv.setting.Table).Insert(values) } +// GetRequest get the request history +func (conv *Xun) GetRequest(sid string, rid string) ([]map[string]interface{}, error) { + + qb := conv.query.Table(conv.setting.Table). + Select("role", "name", "content", "sid"). + Where("rid", rid). + Where("sid", sid). + OrderBy("id", "desc") + + if conv.setting.TTL > 0 { + qb.Where("expired_at", ">", time.Now()) + } + + limit := 20 + if conv.setting.MaxSize > 0 { + limit = conv.setting.MaxSize + } + + rows, err := qb.Limit(limit).Get() + if err != nil { + return nil, err + } + + res := []map[string]interface{}{} + for _, row := range rows { + res = append([]map[string]interface{}{{ + "role": row.Get("role"), + "name": row.Get("name"), + "content": row.Get("content"), + }}, res...) + } + + return res, nil +} + +// SaveRequest save the request history +func (conv *Xun) SaveRequest(sid string, rid string, cid string, messages []map[string]interface{}) error { + + defer conv.clean() + var expiredAt interface{} = nil + values := []row{} + if conv.setting.TTL > 0 { + expiredAt = time.Now().Add(time.Duration(conv.setting.TTL) * time.Second) + } + + for _, message := range messages { + value := row{ + Role: message["role"].(string), + Name: "", + Content: message["content"].(string), + Sid: sid, + Cid: cid, + Rid: rid, + ExpiredAt: expiredAt, + } + + if message["name"] != nil { + value.Name = message["name"].(string) + } + values = append(values, value) + } + + return conv.query.Table(conv.setting.Table).Insert(values) +} + func (conv *Xun) clean() { nums, err := conv.query.Table(conv.setting.Table).Where("expired_at", "<=", time.Now()).Delete() if err != nil { @@ -148,6 +216,8 @@ func (conv *Xun) Init() error { table.ID("id") // The ID field table.String("sid", 255).Index() + table.String("rid", 255).Null().Index() // The request ID + table.String("cid", 200).Null().Index() // The Command ID table.String("role", 200).Null().Index() table.String("name", 200).Null().Index() table.Text("content").Null() @@ -169,7 +239,7 @@ func (conv *Xun) Init() error { return err } - fields := []string{"id", "sid", "role", "name", "content", "created_at", "updated_at", "expired_at"} + fields := []string{"id", "sid", "rid", "cid", "role", "name", "content", "created_at", "updated_at", "expired_at"} for _, field := range fields { if !tab.HasColumn(field) { return fmt.Errorf("%s is required", field) diff --git a/neo/conversation/xun_test.go b/neo/conversation/xun_test.go index 1ed02bf1..a944607b 100644 --- a/neo/conversation/xun_test.go +++ b/neo/conversation/xun_test.go @@ -43,7 +43,7 @@ func TestNewXunDefault(t *testing.T) { t.Fatal(err) } - fields := []string{"id", "sid", "role", "name", "content", "created_at", "updated_at", "expired_at"} + fields := []string{"id", "sid", "cid", "rid", "role", "name", "content", "created_at", "updated_at", "expired_at"} for _, field := range fields { assert.Equal(t, true, tab.HasColumn(field)) } @@ -101,7 +101,7 @@ func TestNewXunConnector(t *testing.T) { t.Fatal(err) } - fields := []string{"id", "sid", "role", "name", "content", "created_at", "updated_at", "expired_at"} + fields := []string{"id", "sid", "cid", "rid", "role", "name", "content", "created_at", "updated_at", "expired_at"} for _, field := range fields { assert.Equal(t, true, tab.HasColumn(field)) } @@ -150,3 +150,35 @@ func TestXunSaveAndGetHistory(t *testing.T) { } assert.Equal(t, 2, len(data)) } + +func TestXunSaveAndGetRequest(t *testing.T) { + + test.Prepare(t, config.Conf) + defer test.Clean() + defer capsule.Schema().DropTableIfExists("__unit_test_conversation") + + err := capsule.Schema().DropTableIfExists("__unit_test_conversation") + if err != nil { + t.Fatal(err) + } + + conv, err := NewXun(Setting{ + Connector: "default", + Table: "__unit_test_conversation", + TTL: 3600, + }) + + // save the history + err = conv.SaveRequest("123456", "912836", "test.command", []map[string]interface{}{ + {"role": "user", "name": "user1", "content": "hello"}, + {"role": "assistant", "name": "user1", "content": "Hello there, how"}, + }) + assert.Nil(t, err) + + // get the history + data, err := conv.GetRequest("123456", "912836") + if err != nil { + t.Fatal(err) + } + assert.Equal(t, 2, len(data)) +} diff --git a/neo/message/json.go b/neo/message/json.go index cc084830..d7dd1518 100644 --- a/neo/message/json.go +++ b/neo/message/json.go @@ -52,6 +52,13 @@ func NewOpenAI(data []byte) *JSON { return &JSON{msg} } +func (json *JSON) String() string { + if json.Message == nil { + return "" + } + return json.Message.Text +} + // Text set the text func (json *JSON) Text(text string) *JSON { json.Message.Text = text diff --git a/neo/neo.go b/neo/neo.go index 90fdd120..86cf6c70 100644 --- a/neo/neo.go +++ b/neo/neo.go @@ -23,15 +23,6 @@ import ( // API is a method on the Neo type func (neo *DSL) API(router *gin.Engine, path string) error { - prompts := []map[string]interface{}{} - for _, prompt := range neo.Prompts { - message := map[string]interface{}{"role": prompt.Role, "content": prompt.Content} - if prompt.Name != "" { - message["name"] = prompt.Name - } - prompts = append(prompts, message) - } - // set the guard err := neo.setGuard(router) if err != nil { @@ -55,22 +46,11 @@ func (neo *DSL) API(router *gin.Engine, path string) error { return } - messages := append([]map[string]interface{}{}, prompts...) - history, err := neo.Conversation.GetHistory(sid) - if err != nil { - c.JSON(500, gin.H{"message": err.Error(), "code": 500}) - c.Done() - } - - messages = append(messages, history...) - messages = append(messages, map[string]interface{}{"role": "user", "content": content, "name": sid}) - // utils.Dump(messages) - // set the context ctx, cancel := command.NewContextWithCancel(sid, c.Query("context")) defer cancel() - err = neo.Answer(ctx, c, messages) + err = neo.Answer(ctx, content, c) if err != nil { c.JSON(500, gin.H{"message": err.Error(), "code": 500}) c.Done() @@ -78,7 +58,7 @@ func (neo *DSL) API(router *gin.Engine, path string) error { }) - // api router chat histor + // api router chat history router.GET(path+"/history", func(c *gin.Context) { sid := c.GetString("__sid") if sid == "" { @@ -142,12 +122,18 @@ func (neo *DSL) API(router *gin.Engine, path string) error { } // Answer the message -func (neo *DSL) Answer(ctx command.Context, answer Answer, messages []map[string]interface{}) error { +func (neo *DSL) Answer(ctx command.Context, question string, answer Answer) error { chanStream := make(chan *message.JSON, 1) chanError := make(chan error, 1) content := []byte{} + // get the chat messages + messages, err := neo.chatMessages(ctx.Sid, question) + if err != nil { + return err + } + // check the command cmd, isCommand := neo.matchCommand(ctx, messages) go func() { @@ -165,7 +151,7 @@ func (neo *DSL) Answer(ctx command.Context, answer Answer, messages []map[string return } - _, err = req.Run(messages, func(msg *message.JSON) int { + _, err = req.Run(neo.Conversation, messages, func(msg *message.JSON) int { chanStream <- msg return 1 }) @@ -228,6 +214,32 @@ func (neo *DSL) Answer(ctx command.Context, answer Answer, messages []map[string return nil } +// prompts get the prompts +func (neo *DSL) prompts() []map[string]interface{} { + prompts := []map[string]interface{}{} + for _, prompt := range neo.Prompts { + message := map[string]interface{}{"role": prompt.Role, "content": prompt.Content} + if prompt.Name != "" { + message["name"] = prompt.Name + } + prompts = append(prompts, message) + } + + return prompts +} + +// chatMessages get the chat messages +func (neo *DSL) chatMessages(sid, content string) ([]map[string]interface{}, error) { + messages := append([]map[string]interface{}{}, neo.prompts()...) + history, err := neo.Conversation.GetHistory(sid) + if err != nil { + return nil, err + } + messages = append(messages, history...) + messages = append(messages, map[string]interface{}{"role": "user", "content": content, "name": sid}) + return messages, nil +} + // matchCommand match the command func (neo *DSL) matchCommand(ctx command.Context, messages []map[string]interface{}) (*command.Command, bool) { if len(messages) < 1 { diff --git a/neo/types.go b/neo/types.go index 9494c8f0..084030f7 100644 --- a/neo/types.go +++ b/neo/types.go @@ -9,22 +9,16 @@ import ( // DSL AI assistant type DSL struct { - ID string `json:"-" yaml:"-"` - Name string `json:"name,omitempty"` - Guard string `json:"guard,omitempty"` - Connector string `json:"connector"` - ConversationSetting conversation.Setting `json:"conversation" yaml:"conversation"` - Option map[string]interface{} `json:"option"` - Prompts []aigc.Prompt `json:"prompts,omitempty"` - Allows []string `json:"allows,omitempty"` - AI aigc.AI `json:"-" yaml:"-"` - Conversation Conversation `json:"-" yaml:"-"` -} - -// Conversation the store interface -type Conversation interface { - GetHistory(sid string) ([]map[string]interface{}, error) - SaveHistory(sid string, messages []map[string]interface{}) error + ID string `json:"-" yaml:"-"` + Name string `json:"name,omitempty"` + Guard string `json:"guard,omitempty"` + Connector string `json:"connector"` + ConversationSetting conversation.Setting `json:"conversation" yaml:"conversation"` + Option map[string]interface{} `json:"option"` + Prompts []aigc.Prompt `json:"prompts,omitempty"` + Allows []string `json:"allows,omitempty"` + AI aigc.AI `json:"-" yaml:"-"` + Conversation conversation.Conversation `json:"-" yaml:"-"` } // Answer the answer interface