From 164121836df8dc843a82308dfa2fc842a2309d99 Mon Sep 17 00:00:00 2001 From: mlogclub Date: Tue, 14 Apr 2026 09:33:49 +0800 Subject: [PATCH] Refactor runtime executor integration and introduce eino package - Replaced the existing runtime executor with a new eino package for better modularity. - Updated Service struct to use the new RuntimeExecutor from eino. - Adjusted Run and Resume methods to accommodate changes in input types and execution logic. - Introduced new types in the eino package to align with the previous executor's functionality. - Refactored tooling preparation logic to streamline tool definitions and improve clarity. - Added new context builders and event consumers to enhance runtime event handling. - Removed legacy executor code and ensured all references are updated to the new eino package. --- internal/ai/application/runtime/service.go | 10 +- .../ai/application/runtime/summary_builder.go | 4 +- internal/ai/infra/eino/service.go | 31 +++ internal/ai/infra/eino/types.go | 8 + .../executor/context_builders.go | 0 .../{internal => }/executor/event_consumer.go | 0 .../{internal => }/executor/helpers.go | 0 .../{internal => }/executor/run_options.go | 0 internal/ai/runtime/executor/service.go | 220 +++++++++++++++++- .../{internal => }/executor/tooling.go | 0 internal/ai/runtime/executor/types.go | 60 ++++- .../ai/runtime/internal/engine/service.go | 10 +- internal/ai/runtime/internal/engine/types.go | 10 +- .../ai/runtime/internal/executor/service.go | 220 +----------------- .../ai/runtime/internal/executor/types.go | 59 +---- 15 files changed, 335 insertions(+), 297 deletions(-) create mode 100644 internal/ai/infra/eino/service.go create mode 100644 internal/ai/infra/eino/types.go rename internal/ai/runtime/{internal => }/executor/context_builders.go (100%) rename internal/ai/runtime/{internal => }/executor/event_consumer.go (100%) rename internal/ai/runtime/{internal => }/executor/helpers.go (100%) rename internal/ai/runtime/{internal => }/executor/run_options.go (100%) rename internal/ai/runtime/{internal => }/executor/tooling.go (100%) diff --git a/internal/ai/application/runtime/service.go b/internal/ai/application/runtime/service.go index dad46e9..35e058b 100644 --- a/internal/ai/application/runtime/service.go +++ b/internal/ai/application/runtime/service.go @@ -3,11 +3,11 @@ package runtime import ( "context" - runtimeexecutor "cs-agent/internal/ai/runtime/executor" + runtimeeino "cs-agent/internal/ai/infra/eino" ) type Service struct { - runtime *runtimeexecutor.Service + runtime *runtimeeino.RuntimeExecutor catalog *toolCatalog prepare *prepareService } @@ -15,7 +15,7 @@ type Service struct { func NewService() *Service { catalog := newToolCatalog() return &Service{ - runtime: runtimeexecutor.NewService(), + runtime: runtimeeino.NewRuntimeExecutor(), catalog: catalog, prepare: newPrepareService(catalog), } @@ -35,7 +35,7 @@ func (s *Service) Run(ctx context.Context, req Request) (*Summary, error) { if err := s.prepare.prepareToolsForRun(&req); err != nil { return nil, err } - summary, err := s.runtime.ExecuteRun(ctx, runtimeexecutor.RunInput{ + summary, err := s.runtime.ExecuteRun(ctx, runtimeeino.RunInput{ Conversation: req.Conversation, UserMessage: req.UserMessage, AIAgent: req.AIAgent, @@ -67,7 +67,7 @@ func (s *Service) Resume(ctx context.Context, req ResumeRequest) (*Summary, erro if err := s.prepare.prepareToolsForResume(&req); err != nil { return nil, err } - summary, err := s.runtime.ExecuteResume(ctx, runtimeexecutor.ResumeInput{ + summary, err := s.runtime.ExecuteResume(ctx, runtimeeino.ResumeInput{ Conversation: req.Conversation, AIAgent: req.AIAgent, AIConfig: req.AIConfig, diff --git a/internal/ai/application/runtime/summary_builder.go b/internal/ai/application/runtime/summary_builder.go index 8ce9003..14ee6dd 100644 --- a/internal/ai/application/runtime/summary_builder.go +++ b/internal/ai/application/runtime/summary_builder.go @@ -3,10 +3,10 @@ package runtime import ( "strings" - runtimeexecutor "cs-agent/internal/ai/runtime/executor" + runtimeeino "cs-agent/internal/ai/infra/eino" ) -func toSummary(summary *runtimeexecutor.RunResult) *Summary { +func toSummary(summary *runtimeeino.RunResult) *Summary { if summary == nil { return nil } diff --git a/internal/ai/infra/eino/service.go b/internal/ai/infra/eino/service.go new file mode 100644 index 0000000..46d58fa --- /dev/null +++ b/internal/ai/infra/eino/service.go @@ -0,0 +1,31 @@ +package eino + +import ( + "context" + + runtimeexecutor "cs-agent/internal/ai/runtime/executor" +) + +type RuntimeExecutor struct { + inner *runtimeexecutor.Service +} + +func NewRuntimeExecutor() *RuntimeExecutor { + return &RuntimeExecutor{ + inner: runtimeexecutor.NewService(), + } +} + +func (s *RuntimeExecutor) ExecuteRun(ctx context.Context, req RunInput) (*RunResult, error) { + if s == nil || s.inner == nil { + return nil, nil + } + return s.inner.ExecuteRun(ctx, runtimeexecutor.RunInput(req)) +} + +func (s *RuntimeExecutor) ExecuteResume(ctx context.Context, req ResumeInput) (*RunResult, error) { + if s == nil || s.inner == nil { + return nil, nil + } + return s.inner.ExecuteResume(ctx, runtimeexecutor.ResumeInput(req)) +} diff --git a/internal/ai/infra/eino/types.go b/internal/ai/infra/eino/types.go new file mode 100644 index 0000000..2d517e4 --- /dev/null +++ b/internal/ai/infra/eino/types.go @@ -0,0 +1,8 @@ +package eino + +import runtimeexecutor "cs-agent/internal/ai/runtime/executor" + +type RunInput = runtimeexecutor.RunInput +type ResumeInput = runtimeexecutor.ResumeInput +type InterruptContextSummary = runtimeexecutor.InterruptContextSummary +type RunResult = runtimeexecutor.RunResult diff --git a/internal/ai/runtime/internal/executor/context_builders.go b/internal/ai/runtime/executor/context_builders.go similarity index 100% rename from internal/ai/runtime/internal/executor/context_builders.go rename to internal/ai/runtime/executor/context_builders.go diff --git a/internal/ai/runtime/internal/executor/event_consumer.go b/internal/ai/runtime/executor/event_consumer.go similarity index 100% rename from internal/ai/runtime/internal/executor/event_consumer.go rename to internal/ai/runtime/executor/event_consumer.go diff --git a/internal/ai/runtime/internal/executor/helpers.go b/internal/ai/runtime/executor/helpers.go similarity index 100% rename from internal/ai/runtime/internal/executor/helpers.go rename to internal/ai/runtime/executor/helpers.go diff --git a/internal/ai/runtime/internal/executor/run_options.go b/internal/ai/runtime/executor/run_options.go similarity index 100% rename from internal/ai/runtime/internal/executor/run_options.go rename to internal/ai/runtime/executor/run_options.go diff --git a/internal/ai/runtime/executor/service.go b/internal/ai/runtime/executor/service.go index d168387..73c689f 100644 --- a/internal/ai/runtime/executor/service.go +++ b/internal/ai/runtime/executor/service.go @@ -2,30 +2,232 @@ package executor import ( "context" + "fmt" + "strings" - internalexecutor "cs-agent/internal/ai/runtime/internal/executor" + "cs-agent/internal/ai/runtime/internal/impl/callbacks" + "cs-agent/internal/ai/runtime/internal/impl/factory" + + "github.com/google/uuid" ) type Service struct { - inner *internalexecutor.Service + agentFactory *factory.AgentFactory + runnerFactory *factory.RunnerFactory } func NewService() *Service { return &Service{ - inner: internalexecutor.NewService(), + agentFactory: factory.NewAgentFactory(), + runnerFactory: factory.NewRunnerFactory(), } } func (s *Service) ExecuteRun(ctx context.Context, req RunInput) (*RunResult, error) { - if s == nil || s.inner == nil { - return nil, nil + summary := &RunResult{ + RunID: uuid.NewString(), + Status: "started", + ToolCodes: make([]string, 0), + InvokedToolCodes: make([]string, 0), } - return s.inner.ExecuteRun(ctx, req) + collector := callbacks.NewRuntimeTraceCollector() + collector.Data.RunID = summary.RunID + if req.AIAgent == nil || req.Conversation == nil || req.UserMessage == nil { + summary.Status = "error" + summary.ErrorMessage = "invalid runtime request" + collector.Data.Status = summary.Status + collector.Data.Error.Message = summary.ErrorMessage + collector.Data.Error.Stage = "prepare" + summary.TraceData = collector.Marshal() + return summary, fmt.Errorf("%s", summary.ErrorMessage) + } + if req.AIConfig == nil { + summary.Status = "error" + summary.ErrorMessage = "ai config is nil" + collector.Data.Status = summary.Status + collector.Data.Error.Message = summary.ErrorMessage + collector.Data.Error.Stage = "prepare" + summary.TraceData = collector.Marshal() + return summary, fmt.Errorf("%s", summary.ErrorMessage) + } + + toolDefs, err := factory.NewToolFactory().BuildMCPTools(req.AIAgent) + if err != nil { + summary.Status = "error" + summary.ErrorMessage = err.Error() + collector.Data.Status = summary.Status + collector.Data.Error.Message = err.Error() + collector.Data.Error.Stage = "prepare" + summary.TraceData = collector.Marshal() + return summary, err + } + tooling := prepareTooling(toolDefs, req.SelectedSkill, req.ToolSet, req.SelectedSkill != nil) + summary.ToolCodes = append(summary.ToolCodes, tooling.toolCodes...) + collector.Data.Input.ToolCodes = append(collector.Data.Input.ToolCodes, summary.ToolCodes...) + collector.SetTooling(tooling.staticToolCodes, definitionToolCodes(tooling.definitions), len(tooling.definitions) > 0) + + collector.Data.Model.Provider = string(req.AIConfig.Provider) + collector.Data.Model.Name = req.AIConfig.ModelName + summary.SelectedSkillCode = "" + summary.SelectedSkillName = "" + summary.SkillRouteReason = strings.TrimSpace(req.SkillRouteReason) + summary.SkillRouteTrace = strings.TrimSpace(req.SkillRouteTrace) + if req.SelectedSkill != nil { + summary.SelectedSkillCode = strings.TrimSpace(req.SelectedSkill.Code) + summary.SelectedSkillName = strings.TrimSpace(req.SelectedSkill.Name) + summary.SkillAllowedToolCodes = parseJSONArrayList(req.SelectedSkill.ToolWhitelist) + collector.Data.Skill.Code = summary.SelectedSkillCode + collector.Data.Skill.Name = summary.SelectedSkillName + collector.Data.Skill.AllowedToolCodes = append([]string(nil), summary.SkillAllowedToolCodes...) + } + collector.Data.Skill.RouteReason = summary.SkillRouteReason + collector.Data.Skill.RouteTrace = summary.SkillRouteTrace + + agent, err := s.agentFactory.BuildCustomerServiceAgent(ctx, factory.BuildCustomerServiceAgentInput{ + AIAgent: req.AIAgent, + AIConfig: req.AIConfig, + SelectedSkill: req.SelectedSkill, + InstructionToolDefinitions: tooling.definitions, + DynamicMCPToolDefinitions: tooling.definitions, + StaticTools: tooling.staticTools, + StaticToolCodes: tooling.staticToolCodeMap, + StaticToolMetadata: tooling.staticToolMetadata, + Collector: collector, + }) + if err != nil { + summary.Status = "error" + summary.ErrorMessage = err.Error() + collector.Data.Status = summary.Status + collector.Data.Error.Message = err.Error() + collector.Data.Error.Stage = "prepare" + summary.TraceData = collector.Marshal() + return summary, err + } + + checkPointID := resolveCheckPointID(req.CheckPointID, summary.RunID) + summary.CheckPointID = checkPointID + runner := s.runnerFactory.Build(ctx, agent, false, true) + if runner == nil { + summary.Status = "error" + summary.ErrorMessage = "failed to build runner" + collector.Data.Status = summary.Status + collector.Data.Error.Message = summary.ErrorMessage + collector.Data.Error.Stage = "prepare" + summary.TraceData = collector.Marshal() + return summary, fmt.Errorf("%s", summary.ErrorMessage) + } + messages := buildRunMessages(ctx, req, summary, collector) + collector.Data.Interrupt.CheckPointID = checkPointID + consumeAgentEvents(runner.Run(ctx, messages, buildRunOptions(checkPointID)...), summary, collector, tooling.toolDefsByModelName) + summary.ModelName = req.AIConfig.ModelName + collector.Data.Status = summary.Status + collector.Data.Output.ReplyText = summary.ReplyText + collector.Data.Output.FinishReason = summary.Status + summary.TraceData = collector.Marshal() + return summary, nil } func (s *Service) ExecuteResume(ctx context.Context, req ResumeInput) (*RunResult, error) { - if s == nil || s.inner == nil { - return nil, nil + summary := &RunResult{ + RunID: uuid.NewString(), + Status: "started", + CheckPointID: strings.TrimSpace(req.CheckPointID), + ToolCodes: make([]string, 0), + InvokedToolCodes: make([]string, 0), + Interrupts: make([]InterruptContextSummary, 0), } - return s.inner.ExecuteResume(ctx, req) + collector := callbacks.NewRuntimeTraceCollector() + collector.Data.RunID = summary.RunID + collector.Data.Interrupt.CheckPointID = summary.CheckPointID + if req.AIAgent == nil { + summary.Status = "error" + summary.ErrorMessage = "ai agent is nil" + collector.Data.Status = summary.Status + collector.Data.Error.Message = summary.ErrorMessage + collector.Data.Error.Stage = "resume_prepare" + summary.TraceData = collector.Marshal() + return summary, fmt.Errorf("%s", summary.ErrorMessage) + } + if req.AIConfig == nil { + summary.Status = "error" + summary.ErrorMessage = "ai config is nil" + collector.Data.Status = summary.Status + collector.Data.Error.Message = summary.ErrorMessage + collector.Data.Error.Stage = "resume_prepare" + summary.TraceData = collector.Marshal() + return summary, fmt.Errorf("%s", summary.ErrorMessage) + } + if summary.CheckPointID == "" { + summary.Status = "error" + summary.ErrorMessage = "checkpoint id is required" + collector.Data.Status = summary.Status + collector.Data.Error.Message = summary.ErrorMessage + collector.Data.Error.Stage = "resume_prepare" + summary.TraceData = collector.Marshal() + return summary, fmt.Errorf("%s", summary.ErrorMessage) + } + toolDefs, err := factory.NewToolFactory().BuildMCPTools(req.AIAgent) + if err != nil { + summary.Status = "error" + summary.ErrorMessage = err.Error() + collector.Data.Status = summary.Status + collector.Data.Error.Message = err.Error() + collector.Data.Error.Stage = "resume_prepare" + summary.TraceData = collector.Marshal() + return summary, err + } + tooling := prepareTooling(toolDefs, nil, req.ToolSet, false) + summary.ToolCodes = append(summary.ToolCodes, tooling.toolCodes...) + collector.Data.Input.ToolCodes = append(collector.Data.Input.ToolCodes, summary.ToolCodes...) + collector.SetTooling(tooling.staticToolCodes, definitionToolCodes(tooling.definitions), len(tooling.definitions) > 0) + collector.Data.Model.Provider = string(req.AIConfig.Provider) + collector.Data.Model.Name = req.AIConfig.ModelName + + agent, err := s.agentFactory.BuildCustomerServiceAgent(ctx, factory.BuildCustomerServiceAgentInput{ + AIAgent: req.AIAgent, + AIConfig: req.AIConfig, + InstructionToolDefinitions: tooling.definitions, + DynamicMCPToolDefinitions: tooling.definitions, + StaticTools: tooling.staticTools, + StaticToolCodes: tooling.staticToolCodeMap, + StaticToolMetadata: tooling.staticToolMetadata, + Collector: collector, + }) + if err != nil { + summary.Status = "error" + summary.ErrorMessage = err.Error() + collector.Data.Status = summary.Status + collector.Data.Error.Message = err.Error() + collector.Data.Error.Stage = "resume_prepare" + summary.TraceData = collector.Marshal() + return summary, err + } + runner := s.runnerFactory.Build(ctx, agent, false, true) + if runner == nil { + summary.Status = "error" + summary.ErrorMessage = "failed to build runner" + collector.Data.Status = summary.Status + collector.Data.Error.Message = summary.ErrorMessage + collector.Data.Error.Stage = "resume_prepare" + summary.TraceData = collector.Marshal() + return summary, fmt.Errorf("%s", summary.ErrorMessage) + } + resumeData := buildResumeDataMessage(req.ResumeData) + iter, err := runner.Resume(ctx, summary.CheckPointID, buildResumeOptions(summary.CheckPointID, resumeData)...) + if err != nil { + summary.Status = "error" + summary.ErrorMessage = err.Error() + collector.Data.Status = summary.Status + collector.Data.Error.Message = err.Error() + collector.Data.Error.Stage = "resume_execute" + summary.TraceData = collector.Marshal() + return summary, err + } + consumeAgentEvents(iter, summary, collector, tooling.toolDefsByModelName) + summary.ModelName = req.AIConfig.ModelName + collector.Data.Status = summary.Status + collector.Data.Output.ReplyText = summary.ReplyText + collector.Data.Output.FinishReason = summary.Status + summary.TraceData = collector.Marshal() + return summary, nil } diff --git a/internal/ai/runtime/internal/executor/tooling.go b/internal/ai/runtime/executor/tooling.go similarity index 100% rename from internal/ai/runtime/internal/executor/tooling.go rename to internal/ai/runtime/executor/tooling.go diff --git a/internal/ai/runtime/executor/types.go b/internal/ai/runtime/executor/types.go index 2321543..9b35a38 100644 --- a/internal/ai/runtime/executor/types.go +++ b/internal/ai/runtime/executor/types.go @@ -1,9 +1,57 @@ package executor -import internalexecutor "cs-agent/internal/ai/runtime/internal/executor" +import ( + "cs-agent/internal/ai/runtime/registry" + "cs-agent/internal/models" +) -// TODO 为什么要定义类型别名？ -type RunInput = internalexecutor.RunInput -type ResumeInput = internalexecutor.ResumeInput -type InterruptContextSummary = internalexecutor.InterruptContextSummary -type RunResult = internalexecutor.RunResult +type RunInput struct { + Conversation *models.Conversation + UserMessage *models.Message + AIAgent *models.AIAgent + AIConfig *models.AIConfig + SelectedSkill *models.SkillDefinition + SkillRouteReason string + SkillRouteTrace string + CheckPointID string + ToolSet *registry.ToolSet +} + +type ResumeInput struct { + Conversation *models.Conversation + AIAgent *models.AIAgent + AIConfig *models.AIConfig + CheckPointID string + ResumeData map[string]any + ToolSet *registry.ToolSet +} + +type InterruptContextSummary struct { + Type string `json:"type,omitempty"` + ID string `json:"id"` + InfoPreview string `json:"infoPreview,omitempty"` +} + +type RunResult struct { + RunID string + Status string + ReplyText string + SelectedSkillCode string + SelectedSkillName string + SkillRouteReason string + SkillRouteTrace string + SkillAllowedToolCodes []string + ModelName string + PromptTokens int + CompletionTokens int + HistoryMessageCount int + RetrieverCount int + ToolCallCount int + ToolCodes []string + InvokedToolCodes []string + CheckPointID string + Interrupted bool + Interrupts []InterruptContextSummary + TraceData string + ErrorMessage string +} diff --git a/internal/ai/runtime/internal/engine/service.go b/internal/ai/runtime/internal/engine/service.go index c76ab1b..a77f936 100644 --- a/internal/ai/runtime/internal/engine/service.go +++ b/internal/ai/runtime/internal/engine/service.go @@ -3,16 +3,16 @@ package engine import ( "context" - "cs-agent/internal/ai/runtime/internal/executor" + runtimeeino "cs-agent/internal/ai/infra/eino" ) type Service struct { - executor *executor.Service + executor *runtimeeino.RuntimeExecutor } func NewService() *Service { return &Service{ - executor: executor.NewService(), + executor: runtimeeino.NewRuntimeExecutor(), } } @@ -21,7 +21,7 @@ func (s *Service) Run(ctx context.Context, req Request) (*Summary, error) { } func (s *Service) ExecuteRun(ctx context.Context, req RunInput) (*RunResult, error) { - return s.executor.ExecuteRun(ctx, executor.RunInput(req)) + return s.executor.ExecuteRun(ctx, runtimeeino.RunInput(req)) } func (s *Service) Resume(ctx context.Context, req ResumeRequest) (*Summary, error) { @@ -29,5 +29,5 @@ func (s *Service) Resume(ctx context.Context, req ResumeRequest) (*Summary, erro } func (s *Service) ExecuteResume(ctx context.Context, req ResumeInput) (*RunResult, error) { - return s.executor.ExecuteResume(ctx, executor.ResumeInput(req)) + return s.executor.ExecuteResume(ctx, runtimeeino.ResumeInput(req)) } diff --git a/internal/ai/runtime/internal/engine/types.go b/internal/ai/runtime/internal/engine/types.go index e239405..16783f9 100644 --- a/internal/ai/runtime/internal/engine/types.go +++ b/internal/ai/runtime/internal/engine/types.go @@ -1,12 +1,12 @@ package engine -import "cs-agent/internal/ai/runtime/internal/executor" +import runtimeeino "cs-agent/internal/ai/infra/eino" // TODO 这个地方为什么要定义类型别名，不能直接用吗？ -type RunInput = executor.RunInput -type ResumeInput = executor.ResumeInput -type InterruptContextSummary = executor.InterruptContextSummary -type RunResult = executor.RunResult +type RunInput = runtimeeino.RunInput +type ResumeInput = runtimeeino.ResumeInput +type InterruptContextSummary = runtimeeino.InterruptContextSummary +type RunResult = runtimeeino.RunResult type Request = RunInput type ResumeRequest = ResumeInput diff --git a/internal/ai/runtime/internal/executor/service.go b/internal/ai/runtime/internal/executor/service.go index 73c689f..55bb34a 100644 --- a/internal/ai/runtime/internal/executor/service.go +++ b/internal/ai/runtime/internal/executor/service.go @@ -2,232 +2,30 @@ package executor import ( "context" - "fmt" - "strings" - "cs-agent/internal/ai/runtime/internal/impl/callbacks" - "cs-agent/internal/ai/runtime/internal/impl/factory" - - "github.com/google/uuid" + publicexecutor "cs-agent/internal/ai/runtime/executor" ) type Service struct { - agentFactory *factory.AgentFactory - runnerFactory *factory.RunnerFactory + inner *publicexecutor.Service } func NewService() *Service { return &Service{ - agentFactory: factory.NewAgentFactory(), - runnerFactory: factory.NewRunnerFactory(), + inner: publicexecutor.NewService(), } } func (s *Service) ExecuteRun(ctx context.Context, req RunInput) (*RunResult, error) { - summary := &RunResult{ - RunID: uuid.NewString(), - Status: "started", - ToolCodes: make([]string, 0), - InvokedToolCodes: make([]string, 0), + if s == nil || s.inner == nil { + return nil, nil } - collector := callbacks.NewRuntimeTraceCollector() - collector.Data.RunID = summary.RunID - if req.AIAgent == nil || req.Conversation == nil || req.UserMessage == nil { - summary.Status = "error" - summary.ErrorMessage = "invalid runtime request" - collector.Data.Status = summary.Status - collector.Data.Error.Message = summary.ErrorMessage - collector.Data.Error.Stage = "prepare" - summary.TraceData = collector.Marshal() - return summary, fmt.Errorf("%s", summary.ErrorMessage) - } - if req.AIConfig == nil { - summary.Status = "error" - summary.ErrorMessage = "ai config is nil" - collector.Data.Status = summary.Status - collector.Data.Error.Message = summary.ErrorMessage - collector.Data.Error.Stage = "prepare" - summary.TraceData = collector.Marshal() - return summary, fmt.Errorf("%s", summary.ErrorMessage) - } - - toolDefs, err := factory.NewToolFactory().BuildMCPTools(req.AIAgent) - if err != nil { - summary.Status = "error" - summary.ErrorMessage = err.Error() - collector.Data.Status = summary.Status - collector.Data.Error.Message = err.Error() - collector.Data.Error.Stage = "prepare" - summary.TraceData = collector.Marshal() - return summary, err - } - tooling := prepareTooling(toolDefs, req.SelectedSkill, req.ToolSet, req.SelectedSkill != nil) - summary.ToolCodes = append(summary.ToolCodes, tooling.toolCodes...) - collector.Data.Input.ToolCodes = append(collector.Data.Input.ToolCodes, summary.ToolCodes...) - collector.SetTooling(tooling.staticToolCodes, definitionToolCodes(tooling.definitions), len(tooling.definitions) > 0) - - collector.Data.Model.Provider = string(req.AIConfig.Provider) - collector.Data.Model.Name = req.AIConfig.ModelName - summary.SelectedSkillCode = "" - summary.SelectedSkillName = "" - summary.SkillRouteReason = strings.TrimSpace(req.SkillRouteReason) - summary.SkillRouteTrace = strings.TrimSpace(req.SkillRouteTrace) - if req.SelectedSkill != nil { - summary.SelectedSkillCode = strings.TrimSpace(req.SelectedSkill.Code) - summary.SelectedSkillName = strings.TrimSpace(req.SelectedSkill.Name) - summary.SkillAllowedToolCodes = parseJSONArrayList(req.SelectedSkill.ToolWhitelist) - collector.Data.Skill.Code = summary.SelectedSkillCode - collector.Data.Skill.Name = summary.SelectedSkillName - collector.Data.Skill.AllowedToolCodes = append([]string(nil), summary.SkillAllowedToolCodes...) - } - collector.Data.Skill.RouteReason = summary.SkillRouteReason - collector.Data.Skill.RouteTrace = summary.SkillRouteTrace - - agent, err := s.agentFactory.BuildCustomerServiceAgent(ctx, factory.BuildCustomerServiceAgentInput{ - AIAgent: req.AIAgent, - AIConfig: req.AIConfig, - SelectedSkill: req.SelectedSkill, - InstructionToolDefinitions: tooling.definitions, - DynamicMCPToolDefinitions: tooling.definitions, - StaticTools: tooling.staticTools, - StaticToolCodes: tooling.staticToolCodeMap, - StaticToolMetadata: tooling.staticToolMetadata, - Collector: collector, - }) - if err != nil { - summary.Status = "error" - summary.ErrorMessage = err.Error() - collector.Data.Status = summary.Status - collector.Data.Error.Message = err.Error() - collector.Data.Error.Stage = "prepare" - summary.TraceData = collector.Marshal() - return summary, err - } - - checkPointID := resolveCheckPointID(req.CheckPointID, summary.RunID) - summary.CheckPointID = checkPointID - runner := s.runnerFactory.Build(ctx, agent, false, true) - if runner == nil { - summary.Status = "error" - summary.ErrorMessage = "failed to build runner" - collector.Data.Status = summary.Status - collector.Data.Error.Message = summary.ErrorMessage - collector.Data.Error.Stage = "prepare" - summary.TraceData = collector.Marshal() - return summary, fmt.Errorf("%s", summary.ErrorMessage) - } - messages := buildRunMessages(ctx, req, summary, collector) - collector.Data.Interrupt.CheckPointID = checkPointID - consumeAgentEvents(runner.Run(ctx, messages, buildRunOptions(checkPointID)...), summary, collector, tooling.toolDefsByModelName) - summary.ModelName = req.AIConfig.ModelName - collector.Data.Status = summary.Status - collector.Data.Output.ReplyText = summary.ReplyText - collector.Data.Output.FinishReason = summary.Status - summary.TraceData = collector.Marshal() - return summary, nil + return s.inner.ExecuteRun(ctx, publicexecutor.RunInput(req)) } func (s *Service) ExecuteResume(ctx context.Context, req ResumeInput) (*RunResult, error) { - summary := &RunResult{ - RunID: uuid.NewString(), - Status: "started", - CheckPointID: strings.TrimSpace(req.CheckPointID), - ToolCodes: make([]string, 0), - InvokedToolCodes: make([]string, 0), - Interrupts: make([]InterruptContextSummary, 0), + if s == nil || s.inner == nil { + return nil, nil } - collector := callbacks.NewRuntimeTraceCollector() - collector.Data.RunID = summary.RunID - collector.Data.Interrupt.CheckPointID = summary.CheckPointID - if req.AIAgent == nil { - summary.Status = "error" - summary.ErrorMessage = "ai agent is nil" - collector.Data.Status = summary.Status - collector.Data.Error.Message = summary.ErrorMessage - collector.Data.Error.Stage = "resume_prepare" - summary.TraceData = collector.Marshal() - return summary, fmt.Errorf("%s", summary.ErrorMessage) - } - if req.AIConfig == nil { - summary.Status = "error" - summary.ErrorMessage = "ai config is nil" - collector.Data.Status = summary.Status - collector.Data.Error.Message = summary.ErrorMessage - collector.Data.Error.Stage = "resume_prepare" - summary.TraceData = collector.Marshal() - return summary, fmt.Errorf("%s", summary.ErrorMessage) - } - if summary.CheckPointID == "" { - summary.Status = "error" - summary.ErrorMessage = "checkpoint id is required" - collector.Data.Status = summary.Status - collector.Data.Error.Message = summary.ErrorMessage - collector.Data.Error.Stage = "resume_prepare" - summary.TraceData = collector.Marshal() - return summary, fmt.Errorf("%s", summary.ErrorMessage) - } - toolDefs, err := factory.NewToolFactory().BuildMCPTools(req.AIAgent) - if err != nil { - summary.Status = "error" - summary.ErrorMessage = err.Error() - collector.Data.Status = summary.Status - collector.Data.Error.Message = err.Error() - collector.Data.Error.Stage = "resume_prepare" - summary.TraceData = collector.Marshal() - return summary, err - } - tooling := prepareTooling(toolDefs, nil, req.ToolSet, false) - summary.ToolCodes = append(summary.ToolCodes, tooling.toolCodes...) - collector.Data.Input.ToolCodes = append(collector.Data.Input.ToolCodes, summary.ToolCodes...) - collector.SetTooling(tooling.staticToolCodes, definitionToolCodes(tooling.definitions), len(tooling.definitions) > 0) - collector.Data.Model.Provider = string(req.AIConfig.Provider) - collector.Data.Model.Name = req.AIConfig.ModelName - - agent, err := s.agentFactory.BuildCustomerServiceAgent(ctx, factory.BuildCustomerServiceAgentInput{ - AIAgent: req.AIAgent, - AIConfig: req.AIConfig, - InstructionToolDefinitions: tooling.definitions, - DynamicMCPToolDefinitions: tooling.definitions, - StaticTools: tooling.staticTools, - StaticToolCodes: tooling.staticToolCodeMap, - StaticToolMetadata: tooling.staticToolMetadata, - Collector: collector, - }) - if err != nil { - summary.Status = "error" - summary.ErrorMessage = err.Error() - collector.Data.Status = summary.Status - collector.Data.Error.Message = err.Error() - collector.Data.Error.Stage = "resume_prepare" - summary.TraceData = collector.Marshal() - return summary, err - } - runner := s.runnerFactory.Build(ctx, agent, false, true) - if runner == nil { - summary.Status = "error" - summary.ErrorMessage = "failed to build runner" - collector.Data.Status = summary.Status - collector.Data.Error.Message = summary.ErrorMessage - collector.Data.Error.Stage = "resume_prepare" - summary.TraceData = collector.Marshal() - return summary, fmt.Errorf("%s", summary.ErrorMessage) - } - resumeData := buildResumeDataMessage(req.ResumeData) - iter, err := runner.Resume(ctx, summary.CheckPointID, buildResumeOptions(summary.CheckPointID, resumeData)...) - if err != nil { - summary.Status = "error" - summary.ErrorMessage = err.Error() - collector.Data.Status = summary.Status - collector.Data.Error.Message = err.Error() - collector.Data.Error.Stage = "resume_execute" - summary.TraceData = collector.Marshal() - return summary, err - } - consumeAgentEvents(iter, summary, collector, tooling.toolDefsByModelName) - summary.ModelName = req.AIConfig.ModelName - collector.Data.Status = summary.Status - collector.Data.Output.ReplyText = summary.ReplyText - collector.Data.Output.FinishReason = summary.Status - summary.TraceData = collector.Marshal() - return summary, nil + return s.inner.ExecuteResume(ctx, publicexecutor.ResumeInput(req)) } diff --git a/internal/ai/runtime/internal/executor/types.go b/internal/ai/runtime/internal/executor/types.go index 9b35a38..3a4065e 100644 --- a/internal/ai/runtime/internal/executor/types.go +++ b/internal/ai/runtime/internal/executor/types.go @@ -1,57 +1,8 @@ package executor -import ( - "cs-agent/internal/ai/runtime/registry" - "cs-agent/internal/models" -) +import publicexecutor "cs-agent/internal/ai/runtime/executor" -type RunInput struct { - Conversation *models.Conversation - UserMessage *models.Message - AIAgent *models.AIAgent - AIConfig *models.AIConfig - SelectedSkill *models.SkillDefinition - SkillRouteReason string - SkillRouteTrace string - CheckPointID string - ToolSet *registry.ToolSet -} - -type ResumeInput struct { - Conversation *models.Conversation - AIAgent *models.AIAgent - AIConfig *models.AIConfig - CheckPointID string - ResumeData map[string]any - ToolSet *registry.ToolSet -} - -type InterruptContextSummary struct { - Type string `json:"type,omitempty"` - ID string `json:"id"` - InfoPreview string `json:"infoPreview,omitempty"` -} - -type RunResult struct { - RunID string - Status string - ReplyText string - SelectedSkillCode string - SelectedSkillName string - SkillRouteReason string - SkillRouteTrace string - SkillAllowedToolCodes []string - ModelName string - PromptTokens int - CompletionTokens int - HistoryMessageCount int - RetrieverCount int - ToolCallCount int - ToolCodes []string - InvokedToolCodes []string - CheckPointID string - Interrupted bool - Interrupts []InterruptContextSummary - TraceData string - ErrorMessage string -} +type RunInput = publicexecutor.RunInput +type ResumeInput = publicexecutor.ResumeInput +type InterruptContextSummary = publicexecutor.InterruptContextSummary +type RunResult = publicexecutor.RunResult