larry-zy commented on code in PR #1540:
URL: https://github.com/apache/dubbo-admin/pull/1540#discussion_r3912369300
##########
ai/component/agent/react/react.go:
##########
@@ -101,36 +121,41 @@ func (ra *ReActAgent) Interact(input *schema.UserInput,
sessionID string) *agent
chans.Send(schema.StreamFinal(final))
chans.Close()
- history.NextTurn(sessionID)
}()
return chans
}
// newInteraction records the user input into history and returns a
session-scoped
// context plus a fresh state.
-func (ra *ReActAgent) newInteraction(input *schema.UserInput, sessionID
string) (context.Context, *state, *memory.HistoryMemory, error) {
- history, err := memory.GetHistoryMemory(ra.memoryCtx,
memory.ChatHistoryKey)
- if err != nil {
- return nil, nil, nil, fmt.Errorf("failed to get history from
context: %w", err)
+func (ra *ReActAgent) newInteraction(parent context.Context, input
*schema.UserInput, sessionID string) (context.Context, *state, error) {
+ if ra.messageStore == nil {
+ return nil, nil, fmt.Errorf("message store is not configured")
+ }
+ if input == nil {
+ return nil, nil, fmt.Errorf("user input is nil")
}
// Record the user's message as plain text. The session id travels via
- // context (memory.SessionIDKey), so there is no need to wrap the input
in a
+ // context, so there is no need to wrap the input in a
// JSON envelope the model would otherwise have to read through.
- history.AddHistory(sessionID,
ai.NewUserMessage(ai.NewTextPart(input.Content)))
+ turnID, err := ra.messageStore.BeginTurn(parent, sessionID)
+ if err != nil {
+ return nil, nil, fmt.Errorf("failed to begin turn: %w", err)
+ }
+ if err := ra.messageStore.AddHistoryToTurn(parent, sessionID, turnID,
ai.NewUserMessage(ai.NewTextPart(input.Content))); err != nil {
Review Comment:
The user message is persisted immediately after . Therefore, a client
disconnect after this point leaves durable conversation state even if the
remaining interaction is canceled.
##########
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 {
Review Comment:
is derived from the HTTP request context in , so remains
cancellation-sensitive even though is detached.
##########
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
Review Comment:
When a stage returns after a client disconnect, this immediate return has
no cleanup/abort operation for the Turn created above.
--
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]
