larry-zy commented on code in PR #1540:
URL: https://github.com/apache/dubbo-admin/pull/1540#discussion_r3912371407
##########
ai/component/agent/react/react.go:
##########
@@ -77,18 +85,30 @@ func NewReActAgent(g *genkit.Genkit, promptBasePath string,
defaultModel string,
// Interact runs one interaction asynchronously and returns immediately with
the
// Channels the caller streams from. The loop, final answer emission, and
channel
// close all happen on a background goroutine; the caller owns draining
Channels.
-func (ra *ReActAgent) Interact(input *schema.UserInput, sessionID string)
*agent.Channels {
+func (ra *ReActAgent) Interact(parent context.Context, input
*schema.UserInput, sessionID string) *agent.Channels {
chans := agent.NewChannels(ra.bufferSize)
go func() {
- ctx, s, history, err := ra.newInteraction(input, sessionID)
+ if parent == nil {
+ parent = context.Background()
+ }
+ ctx, s, err := ra.newInteraction(parent, input, sessionID)
if err != nil {
chans.ErrorChan <- err
chans.Close()
return
}
+ defer s.cancelPersistence()
if err := runLoop(ctx, s, ra.maxIterations,
ra.buildSteps(chans)...); err != nil {
chans.ErrorChan <- err
+ chans.Close()
+ return
+ }
+
+ if err :=
ra.messageStore.NextTurnForTurn(s.persistenceContext(ctx), sessionID,
s.TurnID); err != nil {
Review Comment:
is only reached after succeeds. It is skipped by the cancellation/error
return above, leaving the persisted Turn active.
##########
ai/store/gorm/store.go:
##########
@@ -0,0 +1,659 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package gormstore
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "strings"
+ "time"
+
+ conversationstore "dubbo-admin-ai/store"
+
+ "github.com/firebase/genkit/go/ai"
+ "gorm.io/gorm"
+ "gorm.io/gorm/clause"
+)
+
+const sessionExpiration = 24 * time.Hour
+const expirationCleanupBatchSize = 100
+
+// DefaultMaxTurns matches MemorySpec's default conversation limit.
+const DefaultMaxTurns = 100
+
+// GormStore persists sessions, turns, and messages in a relational database.
+// It deliberately does not use database foreign keys: relationship checks and
+// deletion ordering are handled explicitly by the store transaction.
+type GormStore struct {
+ db *gorm.DB
+ limit int
+}
+
+var _ conversationstore.Store = (*GormStore)(nil)
+
+// NewGormStore creates a store around an already opened Gorm database. The
+// optional limit exists for runtime configuration and tests; the default
+// matches MemorySpec.
+func NewGormStore(db *gorm.DB, limits ...int) (*GormStore, error) {
+ if db == nil {
+ return nil, fmt.Errorf("gorm database is nil")
+ }
+ // The Store owns relationship validation and deletion ordering. Keep
Gorm
+ // from creating database foreign-key constraints if models gain fields
in
+ // the future.
+ if db.Config == nil {
+ db.Config = &gorm.Config{}
+ }
+ db.Config.DisableForeignKeyConstraintWhenMigrating = true
+ limit := DefaultMaxTurns
+ if len(limits) > 0 && limits[0] > 0 {
+ limit = limits[0]
+ }
+ return &GormStore{db: db, limit: limit}, nil
+}
+
+// Migrate creates or updates the Store tables. Gorm's model associations are
+// not declared, so this migration does not create foreign-key constraints.
+func (s *GormStore) Migrate(ctx context.Context) error {
+ if err := s.checkContext(ctx); err != nil {
+ return err
+ }
+ return s.db.WithContext(normalizeContext(ctx)).AutoMigrate(
+ &SessionModel{}, &TurnModel{}, &MessageModel{},
+ )
+}
+
+// DB returns the underlying database for connection-pool configuration and
+// test inspection. Callers must not replace the database instance.
+func (s *GormStore) DB() *gorm.DB { return s.db }
+
+// Close closes the underlying SQL database connection.
+func (s *GormStore) Close() error {
+ if s == nil || s.db == nil {
+ return nil
+ }
+ sqlDB, err := s.db.DB()
+ if err != nil {
+ return err
+ }
+ return sqlDB.Close()
+}
+
+func (s *GormStore) Create(ctx context.Context, session
*conversationstore.Session) error {
+ if err := s.checkContext(ctx); err != nil {
+ return err
+ }
+ if err := validateSession(session); err != nil {
+ return err
+ }
+ model := sessionModelFromDomain(session)
+ return s.db.WithContext(normalizeContext(ctx)).Create(&model).Error
+}
+
+func (s *GormStore) Get(ctx context.Context, sessionID string)
(*conversationstore.Session, error) {
+ if err := s.checkContext(ctx); err != nil {
+ return nil, err
+ }
+ var model SessionModel
+ err := s.db.WithContext(normalizeContext(ctx)).Where("id = ?",
sessionID).First(&model).Error
+ if errors.Is(err, gorm.ErrRecordNotFound) {
+ return nil, conversationstore.ErrSessionNotFound
+ }
+ if err != nil {
+ return nil, err
+ }
+ if isExpired(model.UpdatedAt, time.Now()) {
+ return nil, conversationstore.ErrSessionExpired
+ }
+ return sessionDomainFromModel(&model), nil
+}
+
+func (s *GormStore) List(ctx context.Context) ([]*conversationstore.Session,
error) {
+ if err := s.checkContext(ctx); err != nil {
+ return nil, err
+ }
+ var models []SessionModel
+ cutoff := time.Now().Add(-sessionExpiration)
+ err := s.db.WithContext(normalizeContext(ctx)).
+ Where("status = ? AND updated_at >= ?", "active", cutoff).
+ Find(&models).Error
+ if err != nil {
+ return nil, err
+ }
+ result := make([]*conversationstore.Session, 0, len(models))
+ for i := range models {
+ result = append(result, sessionDomainFromModel(&models[i]))
+ }
+ return result, nil
+}
+
+func (s *GormStore) Touch(ctx context.Context, sessionID string, updatedAt
time.Time) error {
+ if err := s.checkContext(ctx); err != nil {
+ return err
+ }
+ return s.db.WithContext(normalizeContext(ctx)).Transaction(func(tx
*gorm.DB) error {
+ model, err := s.findSession(tx, sessionID, true)
+ if err != nil {
+ return err
+ }
+ if model.Status != "active" {
+ return fmt.Errorf("session %q is not active", sessionID)
+ }
+ if isExpired(model.UpdatedAt, time.Now()) {
+ return conversationstore.ErrSessionExpired
+ }
+ return tx.Model(&SessionModel{}).Where("id = ?",
sessionID).Update("updated_at", updatedAt).Error
+ })
+}
+
+func (s *GormStore) Delete(ctx context.Context, sessionID string) error {
+ if err := s.checkContext(ctx); err != nil {
+ return err
+ }
+ return s.db.WithContext(normalizeContext(ctx)).Transaction(func(tx
*gorm.DB) error {
+ if _, err := s.findSession(tx, sessionID, true); err != nil {
+ return err
+ }
+ return deleteSessionData(tx, sessionID)
+ })
+}
+
+func (s *GormStore) DeleteExpired(ctx context.Context, now time.Time) (int,
error) {
+ if err := s.checkContext(ctx); err != nil {
+ return 0, err
+ }
+ cutoff := now.Add(-sessionExpiration)
+ deleted := 0
+ for {
+ batchDeleted := 0
+ batchSize := 0
+ err :=
s.db.WithContext(normalizeContext(ctx)).Transaction(func(tx *gorm.DB) error {
+ query := tx.Where("updated_at < ?",
cutoff).Order("updated_at ASC").Limit(expirationCleanupBatchSize)
+ if supportsRowLock(tx) {
+ query = query.Clauses(clause.Locking{Strength:
"UPDATE"})
+ }
+ var sessions []SessionModel
+ if err := query.Find(&sessions).Error; err != nil {
+ return err
+ }
+ batchSize = len(sessions)
+ for i := range sessions {
+ // Re-read the row after acquiring the
transaction lock. This prevents
+ // cleanup from deleting a session refreshed by
another instance.
+ current, err := s.findSession(tx,
sessions[i].ID, true)
+ if err != nil {
+ if errors.Is(err,
conversationstore.ErrSessionNotFound) {
+ continue
+ }
+ return err
+ }
+ if !isExpired(current.UpdatedAt, now) {
+ continue
+ }
+ if err := deleteSessionData(tx,
sessions[i].ID); err != nil {
+ return err
+ }
+ batchDeleted++
+ }
+ return nil
+ })
+ if err != nil {
+ return deleted, err
+ }
+ deleted += batchDeleted
+ if batchSize < expirationCleanupBatchSize {
+ return deleted, nil
+ }
+ }
+}
+
+func (s *GormStore) BeginTurn(ctx context.Context, sessionID string) (uint64,
error) {
+ if err := s.checkContext(ctx); err != nil {
+ return 0, err
+ }
+ var turnID uint64
+ err := s.db.WithContext(normalizeContext(ctx)).Transaction(func(tx
*gorm.DB) error {
+ model, err := s.findSession(tx, sessionID, true)
+ if err != nil {
+ return err
+ }
+ if model.Status != "active" {
+ return fmt.Errorf("session %q is not active", sessionID)
+ }
+ if isExpired(model.UpdatedAt, time.Now()) {
+ return conversationstore.ErrSessionExpired
+ }
+
+ var turnCount int64
+ if err := tx.Model(&TurnModel{}).
+ Where("session_id = ?", sessionID).
Review Comment:
This count includes both completed and active Turns. Consequently, active
Turns left by canceled interactions consume ; repeated client disconnects can
exhaust a persisted session.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
