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.
This commit is contained in:
mlogclub
@@ -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,
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
@@ -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
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user