git.iko.soy/crush
on main help sign in
fix(agent): queued messages issues
6e94603Kujtim Hoxha
internal/agent/agent.go+18−16
compare with main
unifiedsplit
@@ -183,21 +183,6 @@ func (a *sessionAgent) Run(ctx context.Context, call SessionAgentCall) (*fantasy
183 FrequencyPenalty: call.FrequencyPenalty,
183 FrequencyPenalty: call.FrequencyPenalty,
184 // Before each step create the new assistant message
184 // Before each step create the new assistant message
185 PrepareStep: func(callContext context.Context, options fantasy.PrepareStepFunctionOptions) (_ context.Context, prepared fantasy.PrepareStepResult, err error) {
185 PrepareStep: func(callContext context.Context, options fantasy.PrepareStepFunctionOptions) (_ context.Context, prepared fantasy.PrepareStepResult, err error) {
186− var assistantMsg message.Message
187− assistantMsg, err = a.messages.Create(callContext, call.SessionID, message.CreateMessageParams{
188− Role: message.Assistant,
189− Parts: []message.ContentPart{},
190− Model: a.largeModel.ModelCfg.Model,
191− Provider: a.largeModel.ModelCfg.Provider,
192− })
193− if err != nil {
194− return callContext, prepared, err
195− }
196−
197− callContext = context.WithValue(callContext, tools.MessageIDContextKey, assistantMsg.ID)
198−
199− currentAssistant = &assistantMsg
200−
201 prepared.Messages = options.Messages
186 prepared.Messages = options.Messages
202 // reset all cached items
187 // reset all cached items
203 for i := range prepared.Messages {
188 for i := range prepared.Messages {
@@ -229,6 +214,19 @@ func (a *sessionAgent) Run(ctx context.Context, call SessionAgentCall) (*fantasy
229 prepared.Messages[i].ProviderOptions = a.getCacheControlOptions()
214 prepared.Messages[i].ProviderOptions = a.getCacheControlOptions()
230 }
215 }
231 }
216 }
217+
218+ var assistantMsg message.Message
219+ assistantMsg, err = a.messages.Create(callContext, call.SessionID, message.CreateMessageParams{
220+ Role: message.Assistant,
221+ Parts: []message.ContentPart{},
222+ Model: a.largeModel.ModelCfg.Model,
223+ Provider: a.largeModel.ModelCfg.Provider,
224+ })
225+ if err != nil {
226+ return callContext, prepared, err
227+ }
228+ callContext = context.WithValue(callContext, tools.MessageIDContextKey, assistantMsg.ID)
229+ currentAssistant = &assistantMsg
232 return callContext, prepared, err
230 return callContext, prepared, err
233 },
231 },
234 OnReasoningDelta: func(id string, text string) error {
232 OnReasoningDelta: func(id string, text string) error {
@@ -432,6 +430,10 @@ func (a *sessionAgent) Run(ctx context.Context, call SessionAgentCall) (*fantasy
432 }
430 }
433 }
431 }
434
432
433+ // release active request before processing queued messages
434+ a.activeRequests.Del(call.SessionID)
435+ cancel()
436+
435 queuedMessages, ok := a.messageQueue.Get(call.SessionID)
437 queuedMessages, ok := a.messageQueue.Get(call.SessionID)
436 if !ok || len(queuedMessages) == 0 {
438 if !ok || len(queuedMessages) == 0 {
437 return result, err
439 return result, err
@@ -439,7 +441,7 @@ func (a *sessionAgent) Run(ctx context.Context, call SessionAgentCall) (*fantasy
439 // there are queued messages restart the loop
441 // there are queued messages restart the loop
440 firstQueuedMessage := queuedMessages[0]
442 firstQueuedMessage := queuedMessages[0]
441 a.messageQueue.Set(call.SessionID, queuedMessages[1:])
443 a.messageQueue.Set(call.SessionID, queuedMessages[1:])
442− return a.Run(genCtx, firstQueuedMessage)
444+ return a.Run(ctx, firstQueuedMessage)
443}
445}
444
446
445func (a *sessionAgent) Summarize(ctx context.Context, sessionID string, opts fantasy.ProviderOptions) error {
447func (a *sessionAgent) Summarize(ctx context.Context, sessionID string, opts fantasy.ProviderOptions) error {