wlw/ai-agent
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(runtime): streamline request handling and improve tool preparation methods

Browse Source
This commit is contained in:
mlogclub
2026-04-19 11:19:51 +08:00
parent 92712d882a
commit 680793f527
9 changed files with 184 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/ai/application/runtime/prepare_service.go
+10 -22
View File
@@ -1,5 +1,7 @@
package runtime
import "cs-agent/internal/ai/runtime/registry"
func newPrepareService(catalog *toolCatalog) *prepareService {
return &prepareService{catalog: catalog}
}
@@ -8,30 +10,16 @@ type prepareService struct {
catalog *toolCatalog
}
func (s *prepareService) prepareToolsForRun(req *Request) error {
if req == nil || req.ToolSet != nil || s.catalog == nil {
return nil
func (s *prepareService) prepareToolsForRun(req Request) (*registry.ToolSet, error) {
if req.ToolSet != nil {
return req.ToolSet, nil
}
toolSet, err := s.catalog.resolveForRun(req)
if err != nil {
return err
}
if toolSet != nil {
req.ToolSet = toolSet
}
return nil
return s.catalog.resolveForRun(req)
}
func (s *prepareService) prepareToolsForResume(req *ResumeRequest) error {
if req == nil || req.ToolSet != nil || s.catalog == nil {
return nil
func (s *prepareService) prepareToolsForResume(req ResumeRequest) (*registry.ToolSet, error) {
if req.ToolSet != nil {
return req.ToolSet, nil
}
toolSet, err := s.catalog.resolveForResume(req)
if err != nil {
return err
}
if toolSet != nil {
req.ToolSet = toolSet
}
return nil
return s.catalog.resolveForResume(req)
}
internal/ai/application/runtime/service.go
+6 -8
View File
@@ -23,13 +23,12 @@ func NewService() *Service {
}
func (s *Service) Run(ctx context.Context, req Request) (*Summary, error) {
if s == nil || s.runtime == nil || s.prepare == nil {
return nil, nil
}
req.UserMessage.Content = utils.BuildRuntimeMessageText(req.UserMessage.MessageType, req.UserMessage.Content)
if err := s.prepare.prepareToolsForRun(&req); err != nil {
toolSet, err := s.prepare.prepareToolsForRun(req)
if err != nil {
return nil, err
}
req.ToolSet = toolSet
summary, err := s.runtime.ExecuteRun(ctx, executor.RunInput{
Conversation: req.Conversation,
UserMessage: req.UserMessage,
@@ -45,12 +44,11 @@ func (s *Service) Run(ctx context.Context, req Request) (*Summary, error) {
}
func (s *Service) Resume(ctx context.Context, req ResumeRequest) (*Summary, error) {
if s == nil || s.runtime == nil || s.prepare == nil {
return nil, nil
}
if err := s.prepare.prepareToolsForResume(&req); err != nil {
toolSet, err := s.prepare.prepareToolsForResume(req)
if err != nil {
return nil, err
}
req.ToolSet = toolSet
summary, err := s.runtime.ExecuteResume(ctx, executor.ResumeInput{
Conversation: req.Conversation,
AIAgent: req.AIAgent,
internal/ai/application/runtime/tool_catalog.go
+2 -8
View File
@@ -32,10 +32,7 @@ func buildRuntimeStaticTools() []registry.Tool {
return ret
}
func (c *toolCatalog) resolveForRun(req *Request) (*registry.ToolSet, error) {
if req == nil || req.ToolSet != nil || c == nil || c.registry == nil {
return nil, nil
}
func (c *toolCatalog) resolveForRun(req Request) (*registry.ToolSet, error) {
return c.registry.Resolve(registry.Context{
Conversation: req.Conversation,
AIAgent: req.AIAgent,
@@ -45,10 +42,7 @@ func (c *toolCatalog) resolveForRun(req *Request) (*registry.ToolSet, error) {
})
}
func (c *toolCatalog) resolveForResume(req *ResumeRequest) (*registry.ToolSet, error) {
if req == nil || req.ToolSet != nil || c == nil || c.registry == nil {
return nil, nil
}
func (c *toolCatalog) resolveForResume(req ResumeRequest) (*registry.ToolSet, error) {
return c.registry.Resolve(registry.Context{
Conversation: req.Conversation,
AIAgent: req.AIAgent,
internal/ai/runtime/reply_commit_service.go
+29 -21
View File
@@ -16,46 +16,54 @@ import (
type replyCommitService struct{}
type replyCommitInput struct {
Conversation models.Conversation
Message models.Message
AIAgent models.AIAgent
ReplyText string
Trace *aiReplyTraceData
ClientPrefix string
IncrementRound bool
}
func newReplyCommitService() *replyCommitService {
return &replyCommitService{}
}
func (s *replyCommitService) SendAIReply(conversation models.Conversation, message models.Message, aiAgent models.AIAgent,
replyText string, trace *aiReplyTraceData, clientPrefix string) (*models.Message, error) {
replyText = strings.TrimSpace(replyText)
func (s *replyCommitService) SendAIReply(input replyCommitInput) (*models.Message, error) {
replyText := strings.TrimSpace(input.ReplyText)
if replyText == "" {
return nil, nil
}
commitStartedAt := time.Now()
replyMessage, err := svc.MessageService.SendAIMessage(
conversation.ID,
aiAgent.ID,
fmt.Sprintf("%s_%d", strings.TrimSpace(clientPrefix), message.ID),
input.Conversation.ID,
input.AIAgent.ID,
fmt.Sprintf("%s_%d", strings.TrimSpace(input.ClientPrefix), input.Message.ID),
enums.IMMessageTypeText,
replyText,
"",
s.buildAIPrincipal(aiAgent),
s.buildAIPrincipal(input.AIAgent),
)
if trace != nil {
trace.CommitMs = time.Since(commitStartedAt).Milliseconds()
trace.ReplySent = err == nil && replyMessage != nil
if input.Trace != nil {
input.Trace.CommitMs = time.Since(commitStartedAt).Milliseconds()
input.Trace.ReplySent = err == nil && replyMessage != nil
if replyMessage != nil {
trace.ReplyMessageID = replyMessage.ID
input.Trace.ReplyMessageID = replyMessage.ID
}
}
if err != nil || !input.IncrementRound {
return replyMessage, err
}
if err := s.IncrementAIReplyRounds(input.Conversation.ID, input.Conversation.AIReplyRounds+1, input.AIAgent.Name); err != nil {
return nil, err
}
return replyMessage, err
}
func (s *replyCommitService) CommitAIReply(conversation models.Conversation, message models.Message, aiAgent models.AIAgent,
replyText string, trace *aiReplyTraceData, clientPrefix string) (*models.Message, error) {
replyMessage, err := s.SendAIReply(conversation, message, aiAgent, replyText, trace, clientPrefix)
if err != nil {
return nil, err
}
if err := s.IncrementAIReplyRounds(conversation.ID, conversation.AIReplyRounds+1, aiAgent.Name); err != nil {
return nil, err
}
return replyMessage, nil
func (s *replyCommitService) CommitAIReply(input replyCommitInput) (*models.Message, error) {
input.IncrementRound = true
return s.SendAIReply(input)
}
func (s *replyCommitService) IncrementAIReplyRounds(conversationID int64, nextRounds int, aiAgentName string) error {
internal/ai/runtime/reply_interrupt_service.go
+40 -6
View File
@@ -18,10 +18,16 @@ func newReplyInterruptService() *replyInterruptService {
func (s *replyInterruptService) ResumePendingInterrupt(ctx context.Context, owner *aiReplyService, conversation models.Conversation, message models.Message, aiAgent models.AIAgent,
pendingInterrupt *models.ConversationInterrupt, trace *aiReplyTraceData, summaryRef **applicationruntime.Summary) error {
if pendingInterrupt == nil || owner == nil || owner.executor == nil {
if pendingInterrupt == nil || owner == nil {
return nil
}
summary, err := owner.executor.ResumePendingInterrupt(ctx, conversation, message, aiAgent, pendingInterrupt, trace)
summary, err := owner.executor.ResumePendingInterrupt(ctx, runtimeReplyResumeInput{
Conversation: conversation,
Message: message,
AIAgent: aiAgent,
PendingInterrupt: pendingInterrupt,
Trace: trace,
})
*summaryRef = summary
if err != nil {
if isCheckpointMissingError(err) {
@@ -29,7 +35,14 @@ func (s *replyInterruptService) ResumePendingInterrupt(ctx context.Context, owne
*summaryRef = summary
trace.Status = "interrupt_expired"
trace.FinalAction = "expired"
replyMessage, expireErr := owner.commit.CommitAIReply(conversation, message, aiAgent, summary.ReplyText, trace, "ai_interrupt_expired")
replyMessage, expireErr := owner.commit.CommitAIReply(replyCommitInput{
Conversation: conversation,
Message: message,
AIAgent: aiAgent,
ReplyText: summary.ReplyText,
Trace: trace,
ClientPrefix: "ai_interrupt_expired",
})
if expireErr != nil {
return expireErr
}
@@ -48,7 +61,14 @@ func (s *replyInterruptService) ResumePendingInterrupt(ctx context.Context, owne
return s.HandleInterruptedResume(owner, conversation, message, aiAgent, pendingInterrupt, summary, trace)
}
if summary != nil && strings.TrimSpace(summary.ReplyText) != "" {
replyMessage, err := owner.commit.CommitAIReply(conversation, message, aiAgent, summary.ReplyText, trace, "ai_resume")
replyMessage, err := owner.commit.CommitAIReply(replyCommitInput{
Conversation: conversation,
Message: message,
AIAgent: aiAgent,
ReplyText: summary.ReplyText,
Trace: trace,
ClientPrefix: "ai_resume",
})
if err != nil {
return err
}
@@ -75,7 +95,14 @@ func (s *replyInterruptService) HandleInterruptedSummary(owner *aiReplyService,
}
pending = svc.ConversationInterruptService.GetByCheckPointID(summary.CheckPointID)
replyText := resolveInterruptPrompt(summary)
replyMessage, err := owner.commit.CommitAIReply(conversation, message, aiAgent, replyText, trace, "ai_interrupt")
replyMessage, err := owner.commit.CommitAIReply(replyCommitInput{
Conversation: conversation,
Message: message,
AIAgent: aiAgent,
ReplyText: replyText,
Trace: trace,
ClientPrefix: "ai_interrupt",
})
if err != nil {
return err
}
@@ -91,7 +118,14 @@ func (s *replyInterruptService) HandleInterruptedResume(owner *aiReplyService, c
return nil
}
replyText := resolveInterruptPrompt(summary)
replyMessage, err := owner.commit.CommitAIReply(conversation, message, aiAgent, replyText, trace, "ai_interrupt_resume")
replyMessage, err := owner.commit.CommitAIReply(replyCommitInput{
Conversation: conversation,
Message: message,
AIAgent: aiAgent,
ReplyText: replyText,
Trace: trace,
ClientPrefix: "ai_interrupt_resume",
})
if err != nil {
return err
}
internal/ai/runtime/reply_runlog_service.go
+38 -28
View File
@@ -18,47 +18,57 @@ func newReplyRunLogService() *replyRunLogService {
type replyRunLogService struct{}
func (s *replyRunLogService) Write(startedAt time.Time, message models.Message, conversation models.Conversation, aiAgent models.AIAgent,
question string, runErr error, trace *aiReplyTraceData, summary *applicationruntime.Summary) {
type replyRunLogInput struct {
StartedAt time.Time
Message models.Message
Conversation models.Conversation
AIAgent models.AIAgent
Question string
RunErr error
Trace *aiReplyTraceData
Summary *applicationruntime.Summary
}
func (s *replyRunLogService) Write(input replyRunLogInput) {
errorMessage := ""
if runErr != nil {
errorMessage = runErr.Error()
} else if summary != nil && strings.TrimSpace(summary.ErrorMessage) != "" {
errorMessage = strings.TrimSpace(summary.ErrorMessage)
if input.RunErr != nil {
errorMessage = input.RunErr.Error()
} else if input.Summary != nil && strings.TrimSpace(input.Summary.ErrorMessage) != "" {
errorMessage = strings.TrimSpace(input.Summary.ErrorMessage)
}
traceData := buildAIReplyTraceData(trace)
plannedAction, plannedToolCode, planReason := buildRunLogPlan(summary)
traceData := buildAIReplyTraceData(input.Trace)
plannedAction, plannedToolCode, planReason := buildRunLogPlan(input.Summary)
logItem := &models.AgentRunLog{
ConversationID: conversation.ID,
MessageID: message.ID,
AIAgentID: aiAgent.ID,
AIConfigID: aiAgent.AIConfigID,
UserMessage: strings.TrimSpace(question),
ConversationID: input.Conversation.ID,
MessageID: input.Message.ID,
AIAgentID: input.AIAgent.ID,
AIConfigID: input.AIAgent.AIConfigID,
UserMessage: strings.TrimSpace(input.Question),
PlannedAction: plannedAction,
PlannedSkillCode: strings.TrimSpace(summaryPlannedSkillCode(summary)),
PlannedSkillName: strings.TrimSpace(summaryPlannedSkillName(summary)),
SkillRouteTrace: strings.TrimSpace(summarySkillRouteTrace(summary)),
ToolSearchTrace: extractToolSearchTrace(summary),
GraphToolTrace: extractGraphToolTrace(summary),
GraphToolCode: firstGraphToolCode(summary),
HandoffReason: extractHandoffReason(summary),
PlannedSkillCode: strings.TrimSpace(summaryPlannedSkillCode(input.Summary)),
PlannedSkillName: strings.TrimSpace(summaryPlannedSkillName(input.Summary)),
SkillRouteTrace: strings.TrimSpace(summarySkillRouteTrace(input.Summary)),
ToolSearchTrace: extractToolSearchTrace(input.Summary),
GraphToolTrace: extractGraphToolTrace(input.Summary),
GraphToolCode: firstGraphToolCode(input.Summary),
HandoffReason: extractHandoffReason(input.Summary),
PlannedToolCode: plannedToolCode,
PlanReason: planReason,
InterruptType: firstInterruptType(summary),
ResumeSource: runLogResumeSource(trace),
FinalAction: toRunLogFinalAction(summary),
FinalStatus: runLogFinalStatus(summary),
ReplyText: buildRunLogReplyText(summary),
InterruptType: firstInterruptType(input.Summary),
ResumeSource: runLogResumeSource(input.Trace),
FinalAction: toRunLogFinalAction(input.Summary),
FinalStatus: runLogFinalStatus(input.Summary),
ReplyText: buildRunLogReplyText(input.Summary),
ErrorMessage: errorMessage,
LatencyMs: time.Since(startedAt).Milliseconds(),
LatencyMs: time.Since(input.StartedAt).Milliseconds(),
TraceData: traceData,
CreatedAt: time.Now(),
}
if err := svc.AgentRunLogService.Create(logItem); err != nil {
slog.Warn("create agent run log failed",
"message_id", message.ID,
"message_id", input.Message.ID,
"conversation_id", logItem.ConversationID,
"ai_agent_id", aiAgent.ID,
"ai_agent_id", input.AIAgent.ID,
"error", err)
}
}
internal/ai/runtime/reply_trigger_service.go
+24 -9
View File
@@ -53,7 +53,16 @@ func (s *aiReplyService) TriggerReply(ctx context.Context, conversation models.C
return nil
}
defer func() {
s.runlog.Write(startedAt, message, conversation, aiAgent, message.Content, retErr, trace, summary)
s.runlog.Write(replyRunLogInput{
StartedAt: startedAt,
Message: message,
Conversation: conversation,
AIAgent: aiAgent,
Question: message.Content,
RunErr: retErr,
Trace: trace,
Summary: summary,
})
}()
if pendingInterrupt := svc.ConversationInterruptService.FindLatestPendingByConversationID(conversation.ID); pendingInterrupt != nil {
return s.resumePendingInterrupt(ctx, conversation, message, aiAgent, pendingInterrupt, trace, &summary)
@@ -63,18 +72,17 @@ func (s *aiReplyService) TriggerReply(ctx context.Context, conversation models.C
func (s *aiReplyService) resumePendingInterrupt(ctx context.Context, conversation models.Conversation, message models.Message, aiAgent models.AIAgent,
pendingInterrupt *models.ConversationInterrupt, trace *aiReplyTraceData, summaryRef **applicationruntime.Summary) error {
if s == nil || s.interrupts == nil {
return nil
}
return s.interrupts.ResumePendingInterrupt(ctx, s, conversation, message, aiAgent, pendingInterrupt, trace, summaryRef)
}
func (s *aiReplyService) executeReply(ctx context.Context, conversation models.Conversation, message models.Message, aiAgent models.AIAgent,
trace *aiReplyTraceData, summaryRef **applicationruntime.Summary) error {
if s == nil || s.executor == nil {
return nil
}
summary, err := s.executor.Run(ctx, conversation, message, aiAgent, trace)
summary, err := s.executor.Run(ctx, runtimeReplyRunInput{
Conversation: conversation,
Message: message,
AIAgent: aiAgent,
Trace: trace,
})
if summaryRef != nil {
*summaryRef = summary
}
@@ -85,7 +93,14 @@ func (s *aiReplyService) executeReply(ctx context.Context, conversation models.C
return s.interrupts.HandleInterruptedSummary(s, conversation, message, aiAgent, summary, trace)
}
if summary != nil && strings.TrimSpace(summary.ReplyText) != "" {
replyMessage, err := s.commit.CommitAIReply(conversation, message, aiAgent, summary.ReplyText, trace, "ai_reply")
replyMessage, err := s.commit.CommitAIReply(replyCommitInput{
Conversation: conversation,
Message: message,
AIAgent: aiAgent,
ReplyText: summary.ReplyText,
Trace: trace,
ClientPrefix: "ai_reply",
})
if err != nil {
return err
}
internal/ai/runtime/runtime_reply_executor.go
+35 -20
View File
@@ -15,53 +15,68 @@ import (
type runtimeReplyExecutor struct{}
type runtimeReplyRunInput struct {
Conversation models.Conversation
Message models.Message
AIAgent models.AIAgent
Trace *aiReplyTraceData
}
type runtimeReplyResumeInput struct {
Conversation models.Conversation
Message models.Message
AIAgent models.AIAgent
PendingInterrupt *models.ConversationInterrupt
Trace *aiReplyTraceData
}
func newRuntimeReplyExecutor() *runtimeReplyExecutor {
return &runtimeReplyExecutor{}
}
func (e *runtimeReplyExecutor) Run(ctx context.Context, conversation models.Conversation, message models.Message, aiAgent models.AIAgent, trace *aiReplyTraceData) (*applicationruntime.Summary, error) {
aiConfig := svc.AIConfigService.Get(aiAgent.AIConfigID)
func (e *runtimeReplyExecutor) Run(ctx context.Context, input runtimeReplyRunInput) (*applicationruntime.Summary, error) {
aiConfig := svc.AIConfigService.Get(input.AIAgent.AIConfigID)
if aiConfig == nil {
return nil, fmt.Errorf("ai config is nil")
}
runtimeStartedAt := time.Now()
summary, err := Service.Run(ctx, applicationruntime.Request{
Conversation: conversation,
UserMessage: message,
AIAgent: aiAgent,
Conversation: input.Conversation,
UserMessage: input.Message,
AIAgent: input.AIAgent,
AIConfig: *aiConfig,
})
if trace != nil {
trace.RuntimeLatencyMs = time.Since(runtimeStartedAt).Milliseconds()
e.fillTraceFromSummary(trace, summary, err)
if input.Trace != nil {
input.Trace.RuntimeLatencyMs = time.Since(runtimeStartedAt).Milliseconds()
e.fillTraceFromSummary(input.Trace, summary, err)
}
return summary, err
}
func (e *runtimeReplyExecutor) ResumePendingInterrupt(ctx context.Context, conversation models.Conversation, message models.Message, aiAgent models.AIAgent, pendingInterrupt *models.ConversationInterrupt, trace *aiReplyTraceData) (*applicationruntime.Summary, error) {
if pendingInterrupt == nil {
func (e *runtimeReplyExecutor) ResumePendingInterrupt(ctx context.Context, input runtimeReplyResumeInput) (*applicationruntime.Summary, error) {
if input.PendingInterrupt == nil {
return nil, nil
}
aiConfig := svc.AIConfigService.Get(aiAgent.AIConfigID)
aiConfig := svc.AIConfigService.Get(input.AIAgent.AIConfigID)
if aiConfig == nil {
return nil, fmt.Errorf("ai config is nil")
}
runtimeStartedAt := time.Now()
if trace != nil {
trace.ResumeSource = "pending_interrupt"
if input.Trace != nil {
input.Trace.ResumeSource = "pending_interrupt"
}
summary, err := Service.Resume(ctx, applicationruntime.ResumeRequest{
Conversation: conversation,
AIAgent: aiAgent,
Conversation: input.Conversation,
AIAgent: input.AIAgent,
AIConfig: *aiConfig,
CheckPointID: strings.TrimSpace(pendingInterrupt.CheckPointID),
CheckPointID: strings.TrimSpace(input.PendingInterrupt.CheckPointID),
ResumeData: map[string]string{
strings.TrimSpace(pendingInterrupt.InterruptID): strings.TrimSpace(message.Content),
strings.TrimSpace(input.PendingInterrupt.InterruptID): strings.TrimSpace(input.Message.Content),
},
})
if trace != nil {
trace.RuntimeLatencyMs = time.Since(runtimeStartedAt).Milliseconds()
e.fillTraceFromSummary(trace, summary, err)
if input.Trace != nil {
input.Trace.RuntimeLatencyMs = time.Since(runtimeStartedAt).Milliseconds()
e.fillTraceFromSummary(input.Trace, summary, err)
}
return summary, err
}
internal/ai/runtime/service.go
-6
View File
@@ -19,15 +19,9 @@ type service struct {
}
func (s *service) Run(ctx context.Context, req applicationruntime.Request) (*applicationruntime.Summary, error) {
if s == nil || s.app == nil {
return nil, nil
}
return s.app.Run(ctx, req)
}
func (s *service) Resume(ctx context.Context, req applicationruntime.ResumeRequest) (*applicationruntime.Summary, error) {
if s == nil || s.app == nil {
return nil, nil
}
return s.app.Resume(ctx, req)
}