refactor: remove unused tool filter middleware and helpers

- Deleted tool_filter_middleware_test.go and tool_helpers.go as they are no longer needed.
- Removed associated test cases in tool_helpers_test.go.
- Refactored knowledge retriever logic by moving it to a new file and updating imports.
- Introduced tooling package for tool result reduction logic.
- Updated traces package to include new trace types and structures.
- Adjusted executor and tool search tool to reflect new package structure.
This commit is contained in:
mlogclub
2026-06-26 17:52:18 +08:00
parent 84094f4ad8
commit 023f224d51
47 changed files with 18 additions and 4491 deletions
@@ -1,25 +0,0 @@
package runtime
import "agent-desk/internal/ai/runtime/registry"
func newPrepareService(catalog *toolCatalog) *prepareService {
return &prepareService{catalog: catalog}
}
type prepareService struct {
catalog *toolCatalog
}
func (s *prepareService) prepareToolsForRun(req Request) (*registry.ToolSet, error) {
if req.ToolSet != nil {
return req.ToolSet, nil
}
return s.catalog.resolveForRun(req)
}
func (s *prepareService) prepareToolsForResume(req ResumeRequest) (*registry.ToolSet, error) {
if req.ToolSet != nil {
return req.ToolSet, nil
}
return s.catalog.resolveForResume(req)
}
+2 -27
View File
@@ -6,7 +6,6 @@ import (
"strings"
"time"
"agent-desk/internal/ai/runtime/executor"
workflowexecutor "agent-desk/internal/ai/runtime/workflow"
"agent-desk/internal/models"
"agent-desk/internal/pkg/errorsx"
@@ -17,9 +16,6 @@ import (
)
type Service struct {
runtime *executor.Service
catalog *toolCatalog
prepare *prepareService
}
const (
@@ -29,12 +25,7 @@ const (
)
func NewService() *Service {
catalog := newToolCatalog()
return &Service{
runtime: executor.NewService(),
catalog: catalog,
prepare: newPrepareService(catalog),
}
return &Service{}
}
func (s *Service) Run(ctx context.Context, req Request) (*Summary, error) {
@@ -106,23 +97,7 @@ func (s *Service) Resume(ctx context.Context, req ResumeRequest) (*Summary, erro
return toWorkflowSummary(workflowResult, req.AIConfig.ModelName, workflow, workflowRunID), 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,
AIConfig: req.AIConfig,
CheckPointID: req.CheckPointID,
ResumeData: req.ResumeData,
ToolSet: req.ToolSet,
})
if err != nil {
return toSummary(summary), err
}
return toSummary(summary), nil
return nil, errorsx.InvalidParam("legacy checkpoint is not supported; please start a new workflow reply")
}
func firstWorkflowResumeText(data map[string]string) string {
@@ -1,45 +0,0 @@
package runtime
import (
"agent-desk/internal/ai/runtime/executor"
"strings"
)
func toSummary(summary *executor.RunResult) *Summary {
if summary == nil {
return nil
}
ret := &Summary{
RunID: summary.RunID,
Status: summary.Status,
ReplyText: summary.ReplyText,
PlannedSkillID: summary.SelectedSkillID,
PlannedSkillName: strings.TrimSpace(summary.SelectedSkillName),
PlanReason: strings.TrimSpace(summary.SkillRouteReason),
SkillRouteTrace: strings.TrimSpace(summary.SkillRouteTrace),
SkillAllowedToolCodes: append([]string(nil), summary.SkillAllowedToolCodes...),
ModelName: summary.ModelName,
PromptTokens: summary.PromptTokens,
CompletionTokens: summary.CompletionTokens,
HistoryMessageCount: summary.HistoryMessageCount,
RetrieverCount: summary.RetrieverCount,
ToolCallCount: summary.ToolCallCount,
ToolCodes: append([]string(nil), summary.ToolCodes...),
InvokedToolCodes: append([]string(nil), summary.InvokedToolCodes...),
CheckPointID: summary.CheckPointID,
Interrupted: summary.Interrupted,
TraceData: summary.TraceData,
ErrorMessage: summary.ErrorMessage,
}
if len(summary.Interrupts) > 0 {
ret.Interrupts = make([]InterruptContextSummary, 0, len(summary.Interrupts))
for _, item := range summary.Interrupts {
ret.Interrupts = append(ret.Interrupts, InterruptContextSummary{
Type: item.Type,
ID: item.ID,
InfoPreview: item.InfoPreview,
})
}
}
return ret
}
@@ -1,93 +0,0 @@
package runtime
import (
"encoding/json"
"strings"
"agent-desk/internal/ai/runtime/registry"
"agent-desk/internal/ai/runtime/tools"
"agent-desk/internal/models"
"agent-desk/internal/pkg/toolx"
)
type toolCatalog struct {
registry *registry.Registry
}
func newToolCatalog() *toolCatalog {
return &toolCatalog{
registry: registry.NewRegistry(buildRuntimeStaticTools()...),
}
}
func buildRuntimeStaticTools() []registry.Tool {
ret := make([]registry.Tool, 0, len(toolx.ListRuntimeStaticToolSpecs()))
for _, spec := range toolx.ListRuntimeStaticToolSpecs() {
tool := tools.NewRuntimeStaticTool(spec.Code)
if tool == nil {
continue
}
ret = append(ret, tool)
}
return ret
}
func (c *toolCatalog) resolveForRun(req Request) (*registry.ToolSet, error) {
return c.registry.Resolve(registry.Context{
Conversation: req.Conversation,
AIAgent: req.AIAgent,
AIConfig: req.AIConfig,
UserMessage: req.UserMessage,
AllowedToolCodes: c.parseAgentAllowedToolCodes(req.AIAgent),
})
}
func (c *toolCatalog) resolveForResume(req ResumeRequest) (*registry.ToolSet, error) {
return c.registry.Resolve(registry.Context{
Conversation: req.Conversation,
AIAgent: req.AIAgent,
AIConfig: req.AIConfig,
AllowedToolCodes: c.parseAgentAllowedToolCodes(req.AIAgent),
})
}
func (c *toolCatalog) parseSkillAllowedToolCodes(skill *models.SkillDefinition) []string {
if skill == nil {
return nil
}
raw := strings.TrimSpace(skill.ToolWhitelist)
if raw == "" {
return nil
}
var items []string
if err := json.Unmarshal([]byte(raw), &items); err != nil {
return nil
}
return toolx.NormalizeToolCodes(items)
}
func (c *toolCatalog) parseAgentAllowedToolCodes(aiAgent models.AIAgent) []string {
ret := make([]string, 0)
if raw := strings.TrimSpace(aiAgent.AllowedMCPTools); raw != "" {
items, err := toolx.ParseAgentMCPToolsJSON(raw)
if err == nil {
for _, item := range items {
ret = append(ret, item.ToolCode)
}
}
}
if raw := strings.TrimSpace(aiAgent.AllowedGraphTools); raw != "" {
var graphTools []string
if err := json.Unmarshal([]byte(raw), &graphTools); err == nil {
ret = append(ret, graphTools...)
}
}
if workflow, err := resolveAgentWorkflow(aiAgent); err == nil {
ret = append(ret, workflow.Compiled.ToolCodes...)
}
return toolx.NormalizeToolCodes(ret)
}
func (c *toolCatalog) resolveAllowedToolCodes(aiAgent models.AIAgent, skill *models.SkillDefinition) []string {
return toolx.IntersectToolCodes(c.parseAgentAllowedToolCodes(aiAgent), c.parseSkillAllowedToolCodes(skill))
}
@@ -1,255 +0,0 @@
package runtime
import (
"context"
"encoding/json"
"strings"
"testing"
"agent-desk/internal/ai/workflow/dsl"
workflowregistry "agent-desk/internal/ai/workflow/registry"
"agent-desk/internal/models"
"agent-desk/internal/pkg/enums"
"agent-desk/internal/pkg/toolx"
"github.com/glebarez/sqlite"
"github.com/mlogclub/simple/sqls"
"gorm.io/gorm"
)
func TestNormalizeAllowedToolCodes(t *testing.T) {
ret := toolx.NormalizeToolCodes([]string{
" ",
"graph/create_ticket_with_confirmation",
"builtin/create_ticket_with_confirmation",
"graph/handoff_to_human",
"graph/handoff_to_human",
})
if len(ret) != 2 {
t.Fatalf("expected 2 tool codes, got %d: %#v", len(ret), ret)
}
if ret[0] != "graph/create_ticket_with_confirmation" {
t.Fatalf("unexpected first tool code: %s", ret[0])
}
if ret[1] != "graph/handoff_to_human" {
t.Fatalf("unexpected second tool code: %s", ret[1])
}
}
func TestToolCatalogResolveAllowedToolCodes(t *testing.T) {
catalog := newToolCatalog()
agent := models.AIAgent{
AllowedMCPTools: `[{"toolCode":"graph/create_ticket_with_confirmation"},{"toolCode":"graph/handoff_to_human"}]`,
}
skill := &models.SkillDefinition{
ToolWhitelist: `["builtin/create_ticket_with_confirmation","graph/prepare_ticket_draft"]`,
}
ret := catalog.resolveAllowedToolCodes(agent, skill)
if len(ret) != 1 {
t.Fatalf("expected 1 tool code, got %d: %#v", len(ret), ret)
}
if ret[0] != "graph/create_ticket_with_confirmation" {
t.Fatalf("unexpected tool code: %s", ret[0])
}
}
func TestToolCatalogResolveAllowedToolCodesFallsBackWhenSkillEmpty(t *testing.T) {
catalog := newToolCatalog()
agent := models.AIAgent{
AllowedMCPTools: `[{"toolCode":"graph/create_ticket_with_confirmation"},{"toolCode":"graph/handoff_to_human"}]`,
}
ret := catalog.resolveAllowedToolCodes(agent, nil)
if len(ret) != 2 {
t.Fatalf("expected 2 tool codes, got %d: %#v", len(ret), ret)
}
}
func TestBuildRuntimeStaticTools(t *testing.T) {
ret := buildRuntimeStaticTools()
if len(ret) != len(toolx.ListRuntimeStaticToolSpecs()) {
t.Fatalf("expected %d runtime static tools, got %d", len(toolx.ListRuntimeStaticToolSpecs()), len(ret))
}
}
func TestToolCatalogIncludesPublishedWorkflowGraphTools(t *testing.T) {
setupWorkflowRuntimeTestDB(t)
version := createWorkflowRuntimeTestVersion(t, dsl.Definition{
SchemaVersion: 1,
EntryNodeID: "start",
Nodes: []dsl.Node{
{ID: "start", Type: workflowregistry.NodeTypeStart, Name: "Start"},
{ID: "draft", Type: workflowregistry.NodeTypePrepareTicketDraft, Name: "Draft Ticket"},
{ID: "create", Type: workflowregistry.NodeTypeCreateTicket, Name: "Create Ticket"},
{ID: "handoff", Type: workflowregistry.NodeTypeHandoffToHuman, Name: "Handoff"},
},
})
catalog := newToolCatalog()
ret := catalog.parseAgentAllowedToolCodes(models.AIAgent{
WorkflowVersionID: version.ID,
})
assertContainsToolCode(t, ret, toolx.GraphPrepareTicketDraft.Code)
assertContainsToolCode(t, ret, toolx.GraphCreateTicketConfirm.Code)
assertContainsToolCode(t, ret, toolx.GraphHandoffConversation.Code)
}
func TestPrepareWorkflowAgentAppendsPublishedWorkflow(t *testing.T) {
setupWorkflowRuntimeTestDB(t)
version := createWorkflowRuntimeTestVersion(t, dsl.Definition{
SchemaVersion: 1,
EntryNodeID: "start",
Nodes: []dsl.Node{
{ID: "start", Type: workflowregistry.NodeTypeStart, Name: "Start"},
{ID: "handoff", Type: workflowregistry.NodeTypeHandoffToHuman, Name: "Handoff"},
},
})
agent, _, err := prepareWorkflowAgent(models.AIAgent{
SystemPrompt: "Base prompt.",
WorkflowVersionID: version.ID,
})
if err != nil {
t.Fatalf("prepare workflow agent: %v", err)
}
if agent.SystemPrompt == "Base prompt." {
t.Fatalf("expected workflow appendix to be appended")
}
if !strings.Contains(agent.SystemPrompt, "Published customer-service workflow") {
t.Fatalf("missing workflow appendix: %s", agent.SystemPrompt)
}
}
func TestPrepareWorkflowAgentRejectsMissingPublishedWorkflow(t *testing.T) {
_, _, err := prepareWorkflowAgent(models.AIAgent{})
if err == nil {
t.Fatalf("expected missing workflow version error")
}
if !strings.Contains(err.Error(), "AI Agent workflow is not published") {
t.Fatalf("unexpected error: %v", err)
}
}
func TestPrepareWorkflowAgentRejectsDeletedPublishedWorkflow(t *testing.T) {
setupWorkflowRuntimeTestDB(t)
version := createWorkflowRuntimeTestVersion(t, dsl.Definition{
SchemaVersion: 1,
EntryNodeID: "start",
Nodes: []dsl.Node{
{ID: "start", Type: workflowregistry.NodeTypeStart, Name: "Start"},
{ID: "end", Type: workflowregistry.NodeTypeEnd, Name: "End"},
},
})
if err := sqls.DB().Model(&models.AIWorkflowVersion{}).Where("id = ?", version.ID).Update("status", enums.StatusDeleted).Error; err != nil {
t.Fatalf("delete workflow version: %v", err)
}
_, _, err := prepareWorkflowAgent(models.AIAgent{WorkflowVersionID: version.ID})
if err == nil {
t.Fatalf("expected invalid workflow version error")
}
if !strings.Contains(err.Error(), "workflow version does not exist") {
t.Fatalf("unexpected error: %v", err)
}
}
func TestServiceRunExecutesPublishedWorkflow(t *testing.T) {
setupWorkflowRuntimeTestDB(t)
version := createWorkflowRuntimeTestVersion(t, dsl.Definition{
SchemaVersion: 1,
EntryNodeID: "start_1",
Nodes: []dsl.Node{
{ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"},
{ID: "reply_1", Type: workflowregistry.NodeTypeLLMReply, Name: "Reply", Config: []byte(`{"staticReply":"workflow reply"}`)},
{ID: "send_1", Type: workflowregistry.NodeTypeSendReply, Name: "Send", Inputs: map[string]dsl.VariableSelector{
"replyText": {NodeID: "reply_1", Field: "replyText"},
}},
{ID: "end_1", Type: workflowregistry.NodeTypeEnd, Name: "End"},
},
Edges: []dsl.Edge{
{ID: "edge_start_reply", Source: "start_1", Target: "reply_1"},
{ID: "edge_reply_send", Source: "reply_1", Target: "send_1"},
{ID: "edge_send_end", Source: "send_1", Target: "end_1"},
},
})
summary, err := NewService().Run(context.Background(), Request{
UserMessage: models.Message{Content: "hello"},
AIAgent: models.AIAgent{
WorkflowVersionID: version.ID,
},
})
if err != nil {
t.Fatalf("run workflow: %v", err)
}
if summary.ReplyText != "workflow reply" {
t.Fatalf("unexpected workflow reply: %q", summary.ReplyText)
}
var runCount int64
if err := sqls.DB().Model(&models.AIWorkflowRun{}).Count(&runCount).Error; err != nil {
t.Fatalf("count workflow runs: %v", err)
}
if runCount != 1 {
t.Fatalf("expected one workflow run, got %d", runCount)
}
var nodeRunCount int64
if err := sqls.DB().Model(&models.AIWorkflowNodeRun{}).Count(&nodeRunCount).Error; err != nil {
t.Fatalf("count workflow node runs: %v", err)
}
if nodeRunCount != 4 {
t.Fatalf("expected four workflow node runs, got %d", nodeRunCount)
}
var replyNodeRun models.AIWorkflowNodeRun
if err := sqls.DB().First(&replyNodeRun, "node_id = ?", "reply_1").Error; err != nil {
t.Fatalf("find reply node run: %v", err)
}
if replyNodeRun.InputPreview == "" || replyNodeRun.OutputPreview == "" {
t.Fatalf("expected node input/output previews, got input=%q output=%q", replyNodeRun.InputPreview, replyNodeRun.OutputPreview)
}
if !strings.Contains(replyNodeRun.OutputPreview, "workflow reply") {
t.Fatalf("expected reply output preview, got %q", replyNodeRun.OutputPreview)
}
if replyNodeRun.DurationMS < 0 {
t.Fatalf("unexpected negative duration: %d", replyNodeRun.DurationMS)
}
}
func setupWorkflowRuntimeTestDB(t *testing.T) {
t.Helper()
db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{})
if err != nil {
t.Fatalf("open sqlite db: %v", err)
}
if err := db.AutoMigrate(&models.AIWorkflowVersion{}, &models.AIWorkflowRun{}, &models.AIWorkflowNodeRun{}); err != nil {
t.Fatalf("auto migrate: %v", err)
}
sqls.SetDB(db)
}
func createWorkflowRuntimeTestVersion(t *testing.T, def dsl.Definition) *models.AIWorkflowVersion {
t.Helper()
definition, err := json.Marshal(def)
if err != nil {
t.Fatalf("marshal definition: %v", err)
}
version := &models.AIWorkflowVersion{
WorkflowID: 1,
Version: 1,
Status: enums.StatusOk,
Definition: string(definition),
}
if err := sqls.DB().Create(version).Error; err != nil {
t.Fatalf("create workflow version: %v", err)
}
return version
}
func assertContainsToolCode(t *testing.T, items []string, want string) {
t.Helper()
for _, item := range items {
if item == want {
return
}
}
t.Fatalf("expected tool code %s in %#v", want, items)
}
-3
View File
@@ -1,7 +1,6 @@
package runtime
import (
"agent-desk/internal/ai/runtime/registry"
"agent-desk/internal/models"
)
@@ -11,7 +10,6 @@ type Request struct {
AIAgent models.AIAgent
AIConfig models.AIConfig
CheckPointID string
ToolSet *registry.ToolSet
}
type ResumeRequest struct {
@@ -21,7 +19,6 @@ type ResumeRequest struct {
AIConfig models.AIConfig
CheckPointID string
ResumeData map[string]string
ToolSet *registry.ToolSet
}
type InterruptContextSummary struct {
@@ -1,293 +0,0 @@
package executor
import (
"context"
"strings"
"time"
"agent-desk/internal/ai/runtime/internal/impl/callbacks"
"agent-desk/internal/ai/runtime/internal/impl/retrievers"
"agent-desk/internal/models"
"agent-desk/internal/pkg/utils"
"github.com/cloudwego/eino/compose"
"github.com/cloudwego/eino/schema"
)
const (
answerabilityNodeRetrieve = "retrieve_knowledge"
answerabilityNodeAllow = "allow_agent"
answerabilityNodeFallback = "fallback"
answerabilityStatusSkipped = "skipped"
answerabilityStatusNoContext = "no_context"
answerabilityStatusHasContext = "has_context"
answerabilityStatusUnanswerable = "unanswerable"
)
type knowledgeContextRetriever interface {
KnowledgeBaseIDs() []int64
RetrieveContextByOptions(ctx context.Context, opts retrievers.KnowledgeRetrieveOptions, query string) (*retrievers.KnowledgeRetrieveResult, error)
}
type answerabilityRetrieverFactory func(aiAgent models.AIAgent) knowledgeContextRetriever
type KnowledgeAnswerabilityGate struct {
newRetriever answerabilityRetrieverFactory
}
type answerabilityGateInput struct {
Request RunInput
Summary *RunResult
Collector *callbacks.RuntimeTraceCollector
Messages []*schema.Message
}
type answerabilityGateState struct {
Input answerabilityGateInput
KnowledgeIDs []int64
RetrieveResult *retrievers.KnowledgeRetrieveResult
Decision knowledgeGuardDecision
SkipGate bool
FallbackReply string
ErrorMessage string
}
func NewKnowledgeAnswerabilityGate() *KnowledgeAnswerabilityGate {
return &KnowledgeAnswerabilityGate{
newRetriever: func(aiAgent models.AIAgent) knowledgeContextRetriever {
return retrievers.NewKnowledgeRetriever(aiAgent)
},
}
}
func (g *KnowledgeAnswerabilityGate) withDefaults() *KnowledgeAnswerabilityGate {
if g == nil {
return NewKnowledgeAnswerabilityGate()
}
ret := *g
defaults := NewKnowledgeAnswerabilityGate()
if ret.newRetriever == nil {
ret.newRetriever = defaults.newRetriever
}
return &ret
}
func (g *KnowledgeAnswerabilityGate) Evaluate(ctx context.Context, input answerabilityGateInput) (*answerabilityGateState, error) {
gate := g.withDefaults()
graph := compose.NewGraph[*answerabilityGateState, *answerabilityGateState]()
if err := graph.AddLambdaNode(answerabilityNodeRetrieve, compose.InvokableLambda(gate.retrieveKnowledge)); err != nil {
return nil, err
}
if err := graph.AddLambdaNode(answerabilityNodeAllow, compose.InvokableLambda(allowAnswerabilityPassThrough)); err != nil {
return nil, err
}
if err := graph.AddLambdaNode(answerabilityNodeFallback, compose.InvokableLambda(fallbackAnswerabilityPassThrough)); err != nil {
return nil, err
}
if err := graph.AddEdge(compose.START, answerabilityNodeRetrieve); err != nil {
return nil, err
}
if err := graph.AddBranch(answerabilityNodeRetrieve, compose.NewGraphBranch(routeAnswerabilityGate, map[string]bool{
answerabilityNodeAllow: true,
answerabilityNodeFallback: true,
})); err != nil {
return nil, err
}
if err := graph.AddEdge(answerabilityNodeAllow, compose.END); err != nil {
return nil, err
}
if err := graph.AddEdge(answerabilityNodeFallback, compose.END); err != nil {
return nil, err
}
runnable, err := graph.Compile(ctx)
if err != nil {
return nil, err
}
return runnable.Invoke(ctx, &answerabilityGateState{Input: input})
}
func routeAnswerabilityGate(ctx context.Context, state *answerabilityGateState) (string, error) {
if state == nil {
return answerabilityNodeFallback, nil
}
if state.SkipGate || strings.TrimSpace(state.FallbackReply) == "" {
return answerabilityNodeAllow, nil
}
return answerabilityNodeFallback, nil
}
func allowAnswerabilityPassThrough(ctx context.Context, state *answerabilityGateState) (*answerabilityGateState, error) {
if state == nil {
return &answerabilityGateState{}, nil
}
if len(state.Decision.Instructions) > 0 {
state.Input.Messages = append(state.Input.Messages, state.Decision.Instructions...)
}
if state.RetrieveResult != nil {
if contextText := strings.TrimSpace(state.RetrieveResult.ContextText); contextText != "" {
state.Input.Messages = append(state.Input.Messages, schema.SystemMessage(contextText))
}
}
return state, nil
}
func fallbackAnswerabilityPassThrough(ctx context.Context, state *answerabilityGateState) (*answerabilityGateState, error) {
if state == nil {
return &answerabilityGateState{}, nil
}
return state, nil
}
func (g *KnowledgeAnswerabilityGate) retrieveKnowledge(ctx context.Context, state *answerabilityGateState) (*answerabilityGateState, error) {
if state == nil {
state = &answerabilityGateState{}
}
gate := g.withDefaults()
req := state.Input.Request
if isRuntimeActionIntent(req.UserMessage.Content) {
state.SkipGate = true
state.recordAnswerability(answerabilityStatusSkipped, "runtime action intent", nil)
return state, nil
}
configuredKnowledgeIDs := utils.SplitInt64s(req.AIAgent.KnowledgeIDs)
if len(configuredKnowledgeIDs) == 0 {
state.SkipGate = true
state.recordAnswerability(answerabilityStatusSkipped, "no knowledge configured", nil)
return state, nil
}
retriever := gate.newRetriever(req.AIAgent)
state.KnowledgeIDs = append([]int64(nil), configuredKnowledgeIDs...)
if retriever == nil {
state.FallbackReply = resolveKnowledgeHumanSupportFallback(req.AIAgent)
state.recordAnswerability(answerabilityStatusUnanswerable, "knowledge retriever unavailable", nil)
return state, nil
}
knowledgeIDs := retriever.KnowledgeBaseIDs()
state.KnowledgeIDs = append([]int64(nil), knowledgeIDs...)
if len(knowledgeIDs) == 0 {
state.SkipGate = true
state.recordAnswerability(answerabilityStatusSkipped, "no knowledge configured", nil)
return state, nil
}
query := strings.TrimSpace(req.UserMessage.Content)
if query == "" {
state.Decision = buildKnowledgeNoContextDecision(req.AIAgent, knowledgeIDs)
state.recordAnswerability(answerabilityStatusNoContext, "empty user question", nil)
return state, nil
}
retrieveOptions := retrievers.DefaultKnowledgeRetrieveOptions()
retrieveOptions.QueryPreview = preview(req.UserMessage.Content, 120)
result, err := retriever.RetrieveContextByOptions(ctx, retrieveOptions, query)
if err != nil {
state.Decision = buildKnowledgeRetrievalErrorDecision(req.AIAgent, knowledgeIDs)
state.ErrorMessage = err.Error()
state.recordAnswerability(answerabilityStatusUnanswerable, "knowledge retrieval failed", err)
return state, nil
}
state.RetrieveResult = result
if state.Input.Summary != nil && result != nil {
state.Input.Summary.RetrieverCount = len(result.Hits)
}
if state.Input.Collector != nil && result != nil {
state.Input.Collector.SetRetrieverSummary(result.TraceSummary)
state.Input.Collector.AddRetrieverItems(result.TraceItems)
}
if result == nil || len(result.Hits) == 0 || strings.TrimSpace(result.ContextText) == "" {
state.Decision = buildKnowledgeNoContextDecision(req.AIAgent, knowledgeIDs)
state.recordAnswerability(answerabilityStatusNoContext, "no retrieved context", nil)
return state, nil
}
state.Decision = buildKnowledgeGuardDecision(req.AIAgent, result)
state.recordAnswerability(answerabilityStatusHasContext, "retrieved context injected", nil)
return state, nil
}
func isRuntimeActionIntent(content string) bool {
text := strings.ToLower(strings.TrimSpace(content))
if text == "" {
return false
}
compact := strings.NewReplacer(" ", "", "\t", "", "\n", "", "\r", "").Replace(text)
handoffPhrases := []string{
"我要转人工",
"帮我转人工",
"转人工",
"接人工",
"找人工",
"真人客服",
"humanagent",
"liveagent",
}
for _, phrase := range handoffPhrases {
if strings.Contains(compact, phrase) {
return true
}
}
if containsAny(compact, []string{"人工客服", "人工服务", "人工处理"}) &&
!containsAny(compact, []string{"是什么", "怎么", "如何", "多少", "几", "吗", "?"}) &&
(isShortActionPhrase(compact) || containsAny(compact, []string{"我要", "帮我", "请", "联系", "需要"})) {
return true
}
ticketPhrases := []string{
"创建工单",
"新建工单",
"提交工单",
"发起工单",
"建工单",
"开工单",
"我要建单",
"帮我建单",
"创建ticket",
"createticket",
}
for _, phrase := range ticketPhrases {
if strings.Contains(compact, phrase) {
return true
}
}
if strings.Contains(compact, "工单") {
for _, action := range []string{"创建", "新建", "提交", "发起", "建", "开", "帮我", "我要", "请"} {
if strings.Contains(compact, action) {
return true
}
}
}
return false
}
func containsAny(text string, values []string) bool {
for _, value := range values {
if strings.Contains(text, value) {
return true
}
}
return false
}
func isShortActionPhrase(text string) bool {
return len([]rune(text)) <= 8
}
func (s *answerabilityGateState) recordAnswerability(status string, reason string, err error) {
s.recordAnswerabilityWithLatency(status, reason, err, time.Time{})
}
func (s *answerabilityGateState) recordAnswerabilityWithLatency(status string, reason string, err error, started time.Time) {
if s == nil || s.Input.Collector == nil {
return
}
errorMessage := strings.TrimSpace(s.ErrorMessage)
if err != nil {
errorMessage = err.Error()
}
data := callbacks.AnswerabilityTraceData{
Status: status,
Reason: strings.TrimSpace(reason),
ErrorMessage: errorMessage,
}
if !started.IsZero() {
data.LatencyMs = time.Since(started).Milliseconds()
}
s.Input.Collector.SetAnswerability(data)
}
@@ -1,301 +0,0 @@
package executor
import (
"context"
"errors"
"strings"
"testing"
"agent-desk/internal/ai/rag"
"agent-desk/internal/ai/runtime/internal/impl/callbacks"
"agent-desk/internal/ai/runtime/internal/impl/retrievers"
"agent-desk/internal/models"
"agent-desk/internal/pkg/enums"
"github.com/cloudwego/eino/schema"
)
type fakeKnowledgeContextRetriever struct {
knowledgeBaseIDs []int64
result *retrievers.KnowledgeRetrieveResult
err error
called bool
}
func (r *fakeKnowledgeContextRetriever) KnowledgeBaseIDs() []int64 {
return append([]int64(nil), r.knowledgeBaseIDs...)
}
func (r *fakeKnowledgeContextRetriever) RetrieveContextByOptions(ctx context.Context, opts retrievers.KnowledgeRetrieveOptions, query string) (*retrievers.KnowledgeRetrieveResult, error) {
r.called = true
if r.err != nil {
return nil, r.err
}
if r.result != nil {
return r.result, nil
}
return &retrievers.KnowledgeRetrieveResult{
KnowledgeBaseIDs: append([]int64(nil), r.knowledgeBaseIDs...),
Query: query,
}, nil
}
func newTestKnowledgePolicyGate(retriever knowledgeContextRetriever) *KnowledgeAnswerabilityGate {
return &KnowledgeAnswerabilityGate{
newRetriever: func(aiAgent models.AIAgent) knowledgeContextRetriever {
return retriever
},
}
}
func newKnowledgePolicyRunInput(content string, knowledgeIDs string) RunInput {
return RunInput{
UserMessage: models.Message{Content: content},
AIAgent: models.AIAgent{
KnowledgeIDs: knowledgeIDs,
FallbackMode: enums.AIAgentFallbackModeSuggestRetry,
FallbackMessage: "我暂时没有找到足够准确的信息。你可以补充更具体的问题,我再继续帮你查。",
AllowedMCPTools: "[]",
},
AIConfig: models.AIConfig{ModelName: "fake-model"},
}
}
func messagesContainContent(messages []*schema.Message, needle string) bool {
for _, message := range messages {
if message != nil && strings.Contains(message.Content, needle) {
return true
}
}
return false
}
func TestKnowledgePolicyEvaluateInjectsNoContextInstructionWithoutFallback(t *testing.T) {
collector := callbacks.NewRuntimeTraceCollector()
gate := newTestKnowledgePolicyGate(&fakeKnowledgeContextRetriever{
knowledgeBaseIDs: []int64{1},
result: &retrievers.KnowledgeRetrieveResult{
KnowledgeBaseIDs: []int64{1},
},
})
state, err := gate.Evaluate(context.Background(), answerabilityGateInput{
Request: newKnowledgePolicyRunInput("你好", "1"),
Summary: &RunResult{},
Collector: collector,
})
if err != nil {
t.Fatalf("Evaluate returned error: %v", err)
}
if state.FallbackReply != "" {
t.Fatalf("expected no direct fallback, got %q", state.FallbackReply)
}
if state.SkipGate {
t.Fatal("expected configured knowledge to inject policy, not skip")
}
if len(state.Decision.Instructions) != 1 {
t.Fatalf("expected one no-context instruction, got %d", len(state.Decision.Instructions))
}
if !strings.Contains(state.Decision.Instructions[0].Content, "当前没有从知识库检索到可用资料") {
t.Fatalf("unexpected no-context instruction: %q", state.Decision.Instructions[0].Content)
}
if !strings.Contains(state.Decision.Instructions[0].Content, "不得编造") {
t.Fatalf("expected anti-hallucination policy, got %q", state.Decision.Instructions[0].Content)
}
if collector.Data.Answerability.Status != answerabilityStatusNoContext {
t.Fatalf("unexpected policy status: %q", collector.Data.Answerability.Status)
}
}
func TestBuildRunMessagesContinuesAgentFlowWhenNoContext(t *testing.T) {
summary := &RunResult{}
gate := newTestKnowledgePolicyGate(&fakeKnowledgeContextRetriever{
knowledgeBaseIDs: []int64{1},
result: &retrievers.KnowledgeRetrieveResult{
KnowledgeBaseIDs: []int64{1},
},
})
messages := buildRunMessages(context.Background(), newKnowledgePolicyRunInput("你好", "1"), summary, nil, gate)
if summary.ReplyText != "" {
t.Fatalf("expected no early fallback reply, got %q", summary.ReplyText)
}
if !messagesContainContent(messages, "当前没有从知识库检索到可用资料") {
t.Fatalf("expected no-context instruction in messages: %#v", messages)
}
if !messagesContainContent(messages, "你好") {
t.Fatalf("expected current user message to remain in messages: %#v", messages)
}
}
func TestKnowledgePolicyEvaluateInjectsGroundedInstructionAndContext(t *testing.T) {
collector := callbacks.NewRuntimeTraceCollector()
gate := newTestKnowledgePolicyGate(&fakeKnowledgeContextRetriever{
knowledgeBaseIDs: []int64{1},
result: &retrievers.KnowledgeRetrieveResult{
KnowledgeBaseIDs: []int64{1},
Hits: []rag.RetrieveResult{
{KnowledgeBaseID: 1, DocumentID: 10, ChunkID: 101, Content: "退款规则:订单发货前可以申请退款。", Score: 0.91},
},
ContextResults: []rag.RetrieveResult{
{KnowledgeBaseID: 1, DocumentID: 10, ChunkID: 101, Content: "退款规则:订单发货前可以申请退款。", Score: 0.91},
},
ContextText: "知识库片段:退款规则:订单发货前可以申请退款。",
AnswerMode: enums.KnowledgeAnswerModeStrict,
},
})
state, err := gate.Evaluate(context.Background(), answerabilityGateInput{
Request: newKnowledgePolicyRunInput("怎么退款", "1"),
Summary: &RunResult{},
Collector: collector,
})
if err != nil {
t.Fatalf("Evaluate returned error: %v", err)
}
if state.FallbackReply != "" {
t.Fatalf("expected no direct fallback, got %q", state.FallbackReply)
}
if len(state.Decision.Instructions) != 1 {
t.Fatalf("expected one grounded instruction, got %d", len(state.Decision.Instructions))
}
if !strings.Contains(state.Decision.Instructions[0].Content, "知识库回答约束") {
t.Fatalf("unexpected grounded instruction: %q", state.Decision.Instructions[0].Content)
}
if collector.Data.Answerability.Status != answerabilityStatusHasContext {
t.Fatalf("unexpected policy status: %q", collector.Data.Answerability.Status)
}
}
func TestBuildRunMessagesInjectsRetrievedContextWhenHasContext(t *testing.T) {
summary := &RunResult{}
gate := newTestKnowledgePolicyGate(&fakeKnowledgeContextRetriever{
knowledgeBaseIDs: []int64{1},
result: &retrievers.KnowledgeRetrieveResult{
KnowledgeBaseIDs: []int64{1},
Hits: []rag.RetrieveResult{
{KnowledgeBaseID: 1, DocumentID: 10, ChunkID: 101, Content: "退款规则:订单发货前可以申请退款。", Score: 0.91},
},
ContextText: "知识库片段:退款规则:订单发货前可以申请退款。",
AnswerMode: enums.KnowledgeAnswerModeStrict,
},
})
messages := buildRunMessages(context.Background(), newKnowledgePolicyRunInput("怎么退款", "1"), summary, nil, gate)
if summary.ReplyText != "" {
t.Fatalf("expected no fallback, got %q", summary.ReplyText)
}
if !messagesContainContent(messages, "知识库回答约束") {
t.Fatalf("expected knowledge instruction in messages: %#v", messages)
}
if !messagesContainContent(messages, "退款规则") {
t.Fatalf("expected retrieved context in messages: %#v", messages)
}
if !messagesContainContent(messages, "怎么退款") {
t.Fatalf("expected current user message in messages: %#v", messages)
}
}
func TestKnowledgePolicyEvaluateSkipsWhenNoKnowledgeConfigured(t *testing.T) {
retriever := &fakeKnowledgeContextRetriever{}
collector := callbacks.NewRuntimeTraceCollector()
gate := newTestKnowledgePolicyGate(retriever)
state, err := gate.Evaluate(context.Background(), answerabilityGateInput{
Request: newKnowledgePolicyRunInput("你好", ""),
Collector: collector,
})
if err != nil {
t.Fatalf("Evaluate returned error: %v", err)
}
if !state.SkipGate {
t.Fatal("expected skip without knowledge")
}
if retriever.called {
t.Fatal("expected retriever not to run without configured knowledge")
}
if collector.Data.Answerability.Status != answerabilityStatusSkipped {
t.Fatalf("unexpected status: %q", collector.Data.Answerability.Status)
}
}
func TestKnowledgePolicyEvaluateSkipsRuntimeActionIntent(t *testing.T) {
retriever := &fakeKnowledgeContextRetriever{knowledgeBaseIDs: []int64{1}}
collector := callbacks.NewRuntimeTraceCollector()
gate := newTestKnowledgePolicyGate(retriever)
state, err := gate.Evaluate(context.Background(), answerabilityGateInput{
Request: newKnowledgePolicyRunInput("帮我转人工", "1"),
Collector: collector,
})
if err != nil {
t.Fatalf("Evaluate returned error: %v", err)
}
if !state.SkipGate {
t.Fatal("expected runtime action to skip knowledge policy")
}
if retriever.called {
t.Fatal("expected retriever not to run for runtime action")
}
if collector.Data.Answerability.Status != answerabilityStatusSkipped {
t.Fatalf("unexpected status: %q", collector.Data.Answerability.Status)
}
}
func TestKnowledgePolicyEvaluateInjectsRetrievalErrorInstructionWithoutFallback(t *testing.T) {
collector := callbacks.NewRuntimeTraceCollector()
gate := newTestKnowledgePolicyGate(&fakeKnowledgeContextRetriever{
knowledgeBaseIDs: []int64{1},
err: errors.New("vector store unavailable"),
})
state, err := gate.Evaluate(context.Background(), answerabilityGateInput{
Request: newKnowledgePolicyRunInput("怎么退款", "1"),
Collector: collector,
})
if err != nil {
t.Fatalf("Evaluate returned error: %v", err)
}
if state.FallbackReply != "" {
t.Fatalf("expected no direct fallback on retrieval error, got %q", state.FallbackReply)
}
if len(state.Decision.Instructions) != 1 {
t.Fatalf("expected one retrieval-error instruction, got %d", len(state.Decision.Instructions))
}
if !strings.Contains(state.Decision.Instructions[0].Content, "知识库检索暂时不可用") {
t.Fatalf("unexpected retrieval-error instruction: %q", state.Decision.Instructions[0].Content)
}
if collector.Data.Answerability.Status != answerabilityStatusUnanswerable {
t.Fatalf("unexpected status: %q", collector.Data.Answerability.Status)
}
if collector.Data.Answerability.Reason != "knowledge retrieval failed" {
t.Fatalf("unexpected reason: %q", collector.Data.Answerability.Reason)
}
}
func TestBuildRunMessagesContinuesAgentFlowWhenRetrievalFails(t *testing.T) {
summary := &RunResult{}
gate := newTestKnowledgePolicyGate(&fakeKnowledgeContextRetriever{
knowledgeBaseIDs: []int64{1},
err: errors.New("vector store unavailable"),
})
messages := buildRunMessages(context.Background(), newKnowledgePolicyRunInput("你好", "1"), summary, nil, gate)
if summary.ReplyText != "" {
t.Fatalf("expected no early fallback reply, got %q", summary.ReplyText)
}
if !messagesContainContent(messages, "知识库检索暂时不可用") {
t.Fatalf("expected retrieval-error instruction in messages: %#v", messages)
}
if !messagesContainContent(messages, "你好") {
t.Fatalf("expected current user message to remain in messages: %#v", messages)
}
}
@@ -1,78 +0,0 @@
package executor
import (
"context"
"strings"
"agent-desk/internal/ai/runtime/internal/impl/adapter"
"agent-desk/internal/ai/runtime/internal/impl/callbacks"
"agent-desk/internal/pkg/utils"
"github.com/cloudwego/eino/schema"
)
func buildRunMessages(ctx context.Context, req RunInput, summary *RunResult, collector *callbacks.RuntimeTraceCollector, gate *KnowledgeAnswerabilityGate) []*schema.Message {
history := adapter.BuildHistoryMessages(req.Conversation.ID, req.UserMessage.ID, 12)
if summary != nil {
summary.HistoryMessageCount = len(history.Messages)
}
if collector != nil {
collector.Data.Input.HistoryMessageCount = len(history.Messages)
collector.Data.Input.KnowledgeBaseIDs = utils.SplitInt64s(req.AIAgent.KnowledgeIDs)
collector.Data.Input.CurrentUserMessagePreview = preview(req.UserMessage.Content, 120)
}
messages := make([]*schema.Message, 0, len(history.Messages)+3)
messages = append(messages, history.Messages...)
decision := appendRetrievedContext(ctx, req, summary, collector, gate, &messages)
if strings.TrimSpace(decision.FallbackReply) != "" {
if summary != nil {
summary.ReplyText = decision.FallbackReply
}
return messages
}
messages = append(messages, schema.UserMessage(strings.TrimSpace(req.UserMessage.Content)))
return messages
}
func appendRetrievedContext(ctx context.Context, req RunInput, summary *RunResult, collector *callbacks.RuntimeTraceCollector, gate *KnowledgeAnswerabilityGate, messages *[]*schema.Message) knowledgeGuardDecision {
if messages == nil {
return knowledgeGuardDecision{}
}
if gate == nil {
gate = NewKnowledgeAnswerabilityGate()
}
state, err := gate.Evaluate(ctx, answerabilityGateInput{
Request: req,
Summary: summary,
Collector: collector,
Messages: append([]*schema.Message(nil), (*messages)...),
})
if err != nil || state == nil {
errorMessage := ""
if err != nil {
errorMessage = err.Error()
} else {
errorMessage = "answerability gate returned nil state"
}
if collector != nil {
collector.SetAnswerability(callbacks.AnswerabilityTraceData{
Status: answerabilityStatusUnanswerable,
Reason: "answerability gate failed",
ErrorMessage: errorMessage,
})
}
decision := buildKnowledgeUnavailableDecision(req.AIAgent, utils.SplitInt64s(req.AIAgent.KnowledgeIDs))
if strings.TrimSpace(decision.FallbackReply) != "" {
decision.FallbackReply = resolveKnowledgeHumanSupportFallback(req.AIAgent)
}
return decision
}
if strings.TrimSpace(state.FallbackReply) != "" {
return knowledgeGuardDecision{FallbackReply: state.FallbackReply}
}
if state.SkipGate {
return knowledgeGuardDecision{}
}
*messages = append((*messages)[:0], state.Input.Messages...)
return state.Decision
}
@@ -1,121 +0,0 @@
package executor
import (
"strings"
"agent-desk/internal/ai/runtime/internal/impl/callbacks"
"agent-desk/internal/ai/runtime/tooling"
"agent-desk/internal/pkg/enums"
"agent-desk/internal/pkg/toolx"
"github.com/cloudwego/eino/adk"
"github.com/cloudwego/eino/schema"
)
func consumeAgentEvents(events *adk.AsyncIterator[*adk.AgentEvent], summary *RunResult, collector *callbacks.RuntimeTraceCollector, toolDefsByModelName map[string]string) {
if summary == nil {
return
}
if collector == nil {
collector = callbacks.NewRuntimeTraceCollector()
}
suppressAssistantReply := false
for {
event, ok := events.Next()
if !ok {
break
}
if event == nil {
continue
}
if event.Action != nil && event.Action.Interrupted != nil {
summary.Status = "interrupted"
summary.Interrupted = true
summary.Interrupts = buildInterruptSummaries(event)
}
if event.Err != nil {
errMsg := strings.TrimSpace(event.Err.Error())
if errMsg != "" {
summary.Status = "error"
summary.ErrorMessage = errMsg
}
}
if event.Output == nil || event.Output.MessageOutput == nil {
continue
}
messageOutput := event.Output.MessageOutput
switch messageOutput.Role {
case schema.Assistant:
if suppressAssistantReply {
continue
}
replyText := strings.TrimSpace(messageOutput.Message.Content)
if replyText != "" {
summary.ReplyText = replyText
}
case schema.Tool:
toolName := strings.TrimSpace(messageOutput.ToolName)
if toolName == "" {
continue
}
toolCode := toolName
if mappedCode, ok := toolDefsByModelName[toolName]; ok && strings.TrimSpace(mappedCode) != "" {
toolCode = strings.TrimSpace(mappedCode)
}
summary.InvokedToolCodes = appendIfMissing(summary.InvokedToolCodes, toolCode)
if strings.TrimSpace(summary.ReplyText) == "" && toolx.ResolveToolSourceType(toolCode) == enums.ToolSourceTypeGraph {
toolReplyText := strings.TrimSpace(messageOutput.Message.Content)
if result, ok := tooling.ParseToolResult(toolReplyText); ok {
if result.ReplyText != "" && !result.ReplySent {
summary.ReplyText = result.ReplyText
}
if result.Terminal && !result.ShouldRetry {
suppressAssistantReply = true
}
}
}
}
}
if summary.Status == "started" {
switch {
case strings.TrimSpace(summary.ErrorMessage) != "":
summary.Status = "error"
case summary.Interrupted:
summary.Status = "interrupted"
case strings.TrimSpace(summary.ReplyText) != "":
summary.Status = "completed"
case hasInvokedGraphTool(summary.InvokedToolCodes):
summary.Status = "completed"
default:
summary.Status = "fallback"
}
}
summary.ToolCallCount = len(summary.InvokedToolCodes)
}
func hasInvokedGraphTool(toolCodes []string) bool {
for _, toolCode := range toolCodes {
if toolx.ResolveToolSourceType(toolCode) == enums.ToolSourceTypeGraph {
return true
}
}
return false
}
func buildInterruptSummaries(event *adk.AgentEvent) []InterruptContextSummary {
if event == nil || event.Action == nil || event.Action.Interrupted == nil {
return nil
}
interrupts := event.Action.Interrupted.InterruptContexts
result := make([]InterruptContextSummary, 0, len(interrupts))
for _, item := range interrupts {
if item == nil {
continue
}
result = append(result, InterruptContextSummary{
ID: strings.TrimSpace(item.ID),
InfoPreview: previewInterruptInfo(item.Info),
})
}
return result
}
@@ -1,177 +0,0 @@
package executor
import (
"encoding/json"
"testing"
"agent-desk/internal/ai/runtime/tooling"
"agent-desk/internal/pkg/toolx"
"github.com/cloudwego/eino/adk"
"github.com/cloudwego/eino/schema"
)
func TestConsumeAgentEventsIgnoresPlainGraphToolText(t *testing.T) {
summary := &RunResult{
Status: "started",
InvokedToolCodes: make([]string, 0),
}
events, gen := adk.NewAsyncIteratorPair[*adk.AgentEvent]()
gen.Send(&adk.AgentEvent{
Output: &adk.AgentOutput{
MessageOutput: &adk.MessageVariant{
Role: schema.Tool,
ToolName: toolx.GraphHandoffConversation.Name,
Message: &schema.Message{
Content: "已为你转接人工客服,请稍候。,请稍候。",
},
},
},
})
gen.Close()
consumeAgentEvents(events, summary, nil, map[string]string{
toolx.GraphHandoffConversation.Name: toolx.GraphHandoffConversation.Code,
})
if summary.ReplyText != "" {
t.Fatalf("unexpected reply text: %q", summary.ReplyText)
}
if summary.Status != "completed" {
t.Fatalf("unexpected summary status: %q", summary.Status)
}
}
func TestConsumeAgentEventsUsesGraphToolResultReplyText(t *testing.T) {
summary := &RunResult{
Status: "started",
InvokedToolCodes: make([]string, 0),
}
payload, err := json.Marshal(tooling.ToolResult{
Handled: true,
Terminal: true,
Action: "off_hours_handoff",
ReplyText: "当前暂不在人工客服服务时间内,你可以先继续描述问题。",
ShouldRetry: false,
})
if err != nil {
t.Fatalf("marshal graph tool result: %v", err)
}
events, gen := adk.NewAsyncIteratorPair[*adk.AgentEvent]()
gen.Send(&adk.AgentEvent{
Output: &adk.AgentOutput{
MessageOutput: &adk.MessageVariant{
Role: schema.Tool,
ToolName: toolx.GraphHandoffConversation.Name,
Message: &schema.Message{
Content: string(payload),
},
},
},
})
gen.Send(&adk.AgentEvent{
Output: &adk.AgentOutput{
MessageOutput: &adk.MessageVariant{
Role: schema.Assistant,
Message: &schema.Message{
Content: "我再试一次转人工。",
},
},
},
})
gen.Close()
consumeAgentEvents(events, summary, nil, map[string]string{
toolx.GraphHandoffConversation.Name: toolx.GraphHandoffConversation.Code,
})
if summary.ReplyText != "当前暂不在人工客服服务时间内,你可以先继续描述问题。" {
t.Fatalf("unexpected reply text: %q", summary.ReplyText)
}
if summary.Status != "completed" {
t.Fatalf("unexpected summary status: %q", summary.Status)
}
}
func TestConsumeAgentEventsSuppressesGraphToolResultWhenReplyAlreadySent(t *testing.T) {
summary := &RunResult{
Status: "started",
InvokedToolCodes: make([]string, 0),
}
payload, err := json.Marshal(tooling.ToolResult{
Handled: true,
Terminal: true,
Action: "off_hours_handoff",
ReplyText: "当前暂不在人工客服服务时间内,你可以先继续描述问题。",
ReplySent: true,
ShouldRetry: false,
})
if err != nil {
t.Fatalf("marshal graph tool result: %v", err)
}
events, gen := adk.NewAsyncIteratorPair[*adk.AgentEvent]()
gen.Send(&adk.AgentEvent{
Output: &adk.AgentOutput{
MessageOutput: &adk.MessageVariant{
Role: schema.Tool,
ToolName: toolx.GraphHandoffConversation.Name,
Message: &schema.Message{
Content: string(payload),
},
},
},
})
gen.Send(&adk.AgentEvent{
Output: &adk.AgentOutput{
MessageOutput: &adk.MessageVariant{
Role: schema.Assistant,
Message: &schema.Message{
Content: "我再试一次转人工。",
},
},
},
})
gen.Close()
consumeAgentEvents(events, summary, nil, map[string]string{
toolx.GraphHandoffConversation.Name: toolx.GraphHandoffConversation.Code,
})
if summary.ReplyText != "" {
t.Fatalf("expected no committed reply because graph already sent it, got %q", summary.ReplyText)
}
if summary.Status != "completed" {
t.Fatalf("unexpected summary status: %q", summary.Status)
}
}
func TestConsumeAgentEventsCompletesGraphToolWithNoVisibleReply(t *testing.T) {
summary := &RunResult{
Status: "started",
InvokedToolCodes: make([]string, 0),
}
events, gen := adk.NewAsyncIteratorPair[*adk.AgentEvent]()
gen.Send(&adk.AgentEvent{
Output: &adk.AgentOutput{
MessageOutput: &adk.MessageVariant{
Role: schema.Tool,
ToolName: toolx.GraphHandoffConversation.Name,
Message: &schema.Message{
Content: "",
},
},
},
})
gen.Close()
consumeAgentEvents(events, summary, nil, map[string]string{
toolx.GraphHandoffConversation.Name: toolx.GraphHandoffConversation.Code,
})
if summary.ReplyText != "" {
t.Fatalf("expected no reply text, got %q", summary.ReplyText)
}
if summary.Status != "completed" {
t.Fatalf("unexpected summary status: %q", summary.Status)
}
}
-28
View File
@@ -1,28 +0,0 @@
package executor
import "strings"
func appendIfMissing(items []string, item string) []string {
item = strings.TrimSpace(item)
if item == "" {
return items
}
for _, existing := range items {
if strings.TrimSpace(existing) == item {
return items
}
}
return append(items, item)
}
func preview(text string, limit int) string {
text = strings.TrimSpace(text)
if text == "" || limit <= 0 {
return ""
}
runes := []rune(text)
if len(runes) <= limit {
return string(runes)
}
return string(runes[:limit]) + "..."
}
@@ -1,124 +0,0 @@
package executor
import (
"strings"
"agent-desk/internal/ai/runtime/internal/impl/retrievers"
"agent-desk/internal/models"
"agent-desk/internal/pkg/enums"
"github.com/cloudwego/eino/schema"
"github.com/mlogclub/simple/common/strs"
)
type knowledgeGuardDecision struct {
FallbackReply string
Instructions []*schema.Message
}
func buildKnowledgeUnavailableDecision(aiAgent models.AIAgent, knowledgeBaseIDs []int64) knowledgeGuardDecision {
if len(knowledgeBaseIDs) == 0 {
return knowledgeGuardDecision{}
}
return knowledgeGuardDecision{FallbackReply: resolveKnowledgeFallbackReply(aiAgent)}
}
func buildKnowledgeGuardDecision(aiAgent models.AIAgent, retrieveResult *retrievers.KnowledgeRetrieveResult) knowledgeGuardDecision {
if retrieveResult == nil || len(retrieveResult.KnowledgeBaseIDs) == 0 {
return knowledgeGuardDecision{}
}
fallbackReply := resolveKnowledgeFallbackReply(aiAgent)
if len(retrieveResult.Hits) == 0 || strings.TrimSpace(retrieveResult.ContextText) == "" {
return knowledgeGuardDecision{FallbackReply: fallbackReply}
}
instruction := buildKnowledgeRuntimeInstruction(retrieveResult.AnswerMode, fallbackReply)
if instruction == "" {
return knowledgeGuardDecision{}
}
return knowledgeGuardDecision{
Instructions: []*schema.Message{schema.SystemMessage(instruction)},
}
}
func buildKnowledgeNoContextDecision(aiAgent models.AIAgent, knowledgeBaseIDs []int64) knowledgeGuardDecision {
if len(knowledgeBaseIDs) == 0 {
return knowledgeGuardDecision{}
}
instruction := buildKnowledgeNoContextInstruction(resolveKnowledgeFallbackReply(aiAgent))
if instruction == "" {
return knowledgeGuardDecision{}
}
return knowledgeGuardDecision{
Instructions: []*schema.Message{schema.SystemMessage(instruction)},
}
}
func buildKnowledgeRetrievalErrorDecision(aiAgent models.AIAgent, knowledgeBaseIDs []int64) knowledgeGuardDecision {
if len(knowledgeBaseIDs) == 0 {
return knowledgeGuardDecision{}
}
instruction := buildKnowledgeRetrievalErrorInstruction(resolveKnowledgeFallbackReply(aiAgent))
if instruction == "" {
return knowledgeGuardDecision{}
}
return knowledgeGuardDecision{
Instructions: []*schema.Message{schema.SystemMessage(instruction)},
}
}
func resolveKnowledgeFallbackReply(aiAgent models.AIAgent) string {
if reply := strings.TrimSpace(aiAgent.FallbackMessage); reply != "" {
return reply
}
switch aiAgent.FallbackMode {
case enums.AIAgentFallbackModeSuggestRetry:
return "当前知识库里没有找到足够明确的信息,你可以换个更具体的问法再试一次。"
default:
return "当前知识库暂无明确信息。"
}
}
func resolveKnowledgeHumanSupportFallback(aiAgent models.AIAgent) string {
base := strings.TrimSpace(resolveKnowledgeFallbackReply(aiAgent))
if strs.IsBlank(base) {
base = "当前知识库暂无明确信息。"
}
return base
}
func buildKnowledgeRuntimeInstruction(answerMode enums.KnowledgeAnswerMode, fallbackReply string) string {
fallbackReply = strings.TrimSpace(fallbackReply)
if fallbackReply == "" {
fallbackReply = "当前知识库暂无明确信息。"
}
if answerMode == enums.KnowledgeAnswerModeAssist {
return "知识库回答约束:优先依据后续提供的知识片段回答,可以做轻度归纳,但不要编造片段中未提供的事实。回答中的具体事实、步骤、承诺、价格、时效、政策必须能被知识片段直接支持;若知识片段不足以直接支持答案,必须明确回复:" + fallbackReply
}
return "知识库回答约束:本轮只能依据后续提供的知识片段回答,不得使用模型常识补充未提供的事实,不得输出知识片段外的具体事实、步骤、承诺、建议、价格、时效或政策。若知识片段不足以支持回答,必须明确回复:" + fallbackReply
}
func buildKnowledgeNoContextInstruction(fallbackReply string) string {
fallbackReply = strings.TrimSpace(fallbackReply)
if fallbackReply == "" {
fallbackReply = "当前知识库暂无明确信息。"
}
return "知识库检索状态:当前没有从知识库检索到可用资料。\n" +
"回复策略:\n" +
"1. 如果用户只是寒暄、问候、感谢、确认或结束语,可以自然、简短地回复。\n" +
"2. 如果用户表达不清楚或缺少上下文,应追问具体场景、对象、报错信息或操作步骤。\n" +
"3. 如果用户询问业务事实、规则、价格、流程、配置、时效、承诺、售后、退款、权限或政策,不得编造答案,必须明确回复:" + fallbackReply + "\n" +
"4. 不得输出知识库未提供的具体事实、流程、承诺、价格、时效或政策。"
}
func buildKnowledgeRetrievalErrorInstruction(fallbackReply string) string {
fallbackReply = strings.TrimSpace(fallbackReply)
if fallbackReply == "" {
fallbackReply = "当前知识库暂无明确信息。"
}
return "知识库检索状态:知识库检索暂时不可用,当前没有可用的知识库资料。\n" +
"回复策略:\n" +
"1. 如果用户只是寒暄、问候、感谢、确认或结束语,可以自然、简短地回复,不要使用知识库兜底话术。\n" +
"2. 如果用户表达不清楚或缺少上下文,应追问具体场景、对象、报错信息或操作步骤。\n" +
"3. 如果用户询问业务事实、规则、价格、流程、配置、时效、承诺、售后、退款、权限或政策,不得编造答案,必须明确回复:" + fallbackReply + "\n" +
"4. 不得输出知识库未提供的具体事实、流程、承诺、价格、时效或政策。"
}
@@ -1,75 +0,0 @@
package executor
import (
"encoding/json"
"fmt"
"strings"
"github.com/cloudwego/eino/adk"
"github.com/cloudwego/eino/schema"
)
func resolveCheckPointID(input string, runID string) string {
checkPointID := strings.TrimSpace(input)
if checkPointID != "" {
return checkPointID
}
return "eino_cp_" + strings.TrimSpace(runID)
}
func buildResumeDataMessage(resumeData map[string]string) *schema.Message {
if len(resumeData) == 0 {
return nil
}
data, err := json.Marshal(resumeData)
if err != nil {
return schema.UserMessage(fmt.Sprint(resumeData))
}
return schema.UserMessage(string(data))
}
func buildResumeTargets(resumeData map[string]string) map[string]any {
if len(resumeData) == 0 {
return nil
}
targets := make(map[string]any, len(resumeData))
for key, value := range resumeData {
key = strings.TrimSpace(key)
if key == "" {
continue
}
targets[key] = value
}
if len(targets) == 0 {
return nil
}
return targets
}
func buildRunOptions(checkPointID string) []adk.AgentRunOption {
options := make([]adk.AgentRunOption, 0, 1)
if strings.TrimSpace(checkPointID) != "" {
options = append(options, adk.WithCheckPointID(checkPointID))
}
return options
}
func buildResumeOptions(checkPointID string, resumeData *schema.Message) []adk.AgentRunOption {
options := make([]adk.AgentRunOption, 0, 1)
if strings.TrimSpace(checkPointID) != "" {
options = append(options, adk.WithCheckPointID(checkPointID))
}
_ = resumeData
return options
}
func previewInterruptInfo(info any) string {
if info == nil {
return ""
}
data, err := json.Marshal(info)
if err != nil {
return fmt.Sprint(info)
}
return string(data)
}
@@ -1,34 +0,0 @@
package executor
import "testing"
func TestBuildResumeTargets(t *testing.T) {
targets := buildResumeTargets(map[string]string{
" interrupt-1 ": "确认",
"": "ignored",
" ": "ignored",
"interrupt-2": "取消",
})
if len(targets) != 2 {
t.Fatalf("expected 2 resume targets, got %d", len(targets))
}
if got := targets["interrupt-1"]; got != "确认" {
t.Fatalf("unexpected target data for interrupt-1: %#v", got)
}
if got := targets["interrupt-2"]; got != "取消" {
t.Fatalf("unexpected target data for interrupt-2: %#v", got)
}
}
func TestBuildResumeTargetsEmpty(t *testing.T) {
if got := buildResumeTargets(nil); got != nil {
t.Fatalf("expected nil targets for nil input, got %#v", got)
}
if got := buildResumeTargets(map[string]string{
"": "ignored",
" ": "ignored",
}); got != nil {
t.Fatalf("expected nil targets for blank keys, got %#v", got)
}
}
-224
View File
@@ -1,224 +0,0 @@
package executor
import (
"context"
"fmt"
"strings"
"agent-desk/internal/ai/runtime/internal/impl/callbacks"
"agent-desk/internal/ai/runtime/internal/impl/factory"
"github.com/cloudwego/eino/adk"
"github.com/google/uuid"
)
type Service struct {
agentFactory *factory.AgentFactory
runnerFactory *factory.RunnerFactory
answerabilityGate *KnowledgeAnswerabilityGate
}
func NewService() *Service {
return &Service{
agentFactory: factory.NewAgentFactory(),
runnerFactory: factory.NewRunnerFactory(),
answerabilityGate: NewKnowledgeAnswerabilityGate(),
}
}
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),
}
collector := callbacks.NewRuntimeTraceCollector()
collector.Data.RunID = summary.RunID
summary.ModelName = req.AIConfig.ModelName
collector.Data.Model.Provider = string(req.AIConfig.Provider)
collector.Data.Model.Name = req.AIConfig.ModelName
checkPointID := resolveCheckPointID(req.CheckPointID, summary.RunID)
summary.CheckPointID = checkPointID
messages := buildRunMessages(ctx, req, summary, collector, s.answerabilityGate)
if strings.TrimSpace(summary.ReplyText) != "" {
summary.Status = "completed"
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
}
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
}
hasVisibleSkills := factory.HasVisibleSkills(req.AIAgent)
tooling := prepareTooling(toolDefs, nil, req.ToolSet, hasVisibleSkills)
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)
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 = "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 = "prepare"
summary.TraceData = collector.Marshal()
return summary, fmt.Errorf("%s", summary.ErrorMessage)
}
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
syncSkillSummaryFromCollector(summary, collector)
summary.TraceData = collector.Marshal()
return summary, nil
}
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),
}
collector := callbacks.NewRuntimeTraceCollector()
collector.Data.RunID = summary.RunID
collector.Data.Interrupt.CheckPointID = summary.CheckPointID
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
}
hasVisibleSkills := factory.HasVisibleSkills(req.AIAgent)
tooling := prepareTooling(toolDefs, nil, req.ToolSet, hasVisibleSkills)
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)
resumeTargets := buildResumeTargets(req.ResumeData)
var (
iter *adk.AsyncIterator[*adk.AgentEvent]
)
if len(resumeTargets) > 0 {
iter, err = runner.ResumeWithParams(ctx, summary.CheckPointID, &adk.ResumeParams{
Targets: resumeTargets,
}, buildResumeOptions(summary.CheckPointID, resumeData)...)
} else {
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
syncSkillSummaryFromCollector(summary, collector)
summary.TraceData = collector.Marshal()
return summary, nil
}
func syncSkillSummaryFromCollector(summary *RunResult, collector *callbacks.RuntimeTraceCollector) {
if summary == nil || collector == nil {
return
}
trace := collector.Data.Skill
summary.SelectedSkillID = trace.ID
summary.SelectedSkillName = strings.TrimSpace(trace.Name)
summary.SkillRouteReason = strings.TrimSpace(trace.RouteReason)
summary.SkillRouteTrace = strings.TrimSpace(trace.RouteTrace)
summary.SkillAllowedToolCodes = append([]string(nil), trace.AllowedToolCodes...)
if len(trace.FilteredToolCodes) > 0 {
summary.ToolCodes = append([]string(nil), trace.FilteredToolCodes...)
}
}
-185
View File
@@ -1,185 +0,0 @@
package executor
import (
"encoding/json"
"strings"
"agent-desk/internal/ai/runtime/registry"
runtimetooling "agent-desk/internal/ai/runtime/tooling"
"agent-desk/internal/models"
"agent-desk/internal/pkg/toolx"
einotool "github.com/cloudwego/eino/components/tool"
)
type preparedTooling struct {
definitions []runtimetooling.MCPToolDefinition
toolCodes []string
toolDefsByModelName map[string]string
staticToolCodes []string
staticTools []einotool.BaseTool
staticToolCodeMap map[string]string
staticToolMetadata map[string]registry.ToolMetadata
}
func prepareTooling(defs []runtimetooling.MCPToolDefinition, selectedSkill *models.SkillDefinition, toolSet *registry.ToolSet, includeSkillTool bool) preparedTooling {
filteredDefs := filterToolDefinitionsBySkill(defs, selectedSkill)
ret := preparedTooling{
definitions: filteredDefs,
toolCodes: make([]string, 0, len(filteredDefs)+2),
toolDefsByModelName: make(map[string]string, len(filteredDefs)),
staticToolCodes: staticToolCodeList(toolSet),
staticTools: toolSetStaticTools(toolSet),
staticToolCodeMap: toolSetStaticToolCodes(toolSet),
staticToolMetadata: toolSetStaticToolMetadata(toolSet),
}
for _, item := range filteredDefs {
toolCode := strings.TrimSpace(item.ToolCode)
modelName := strings.TrimSpace(item.ModelName)
if toolCode == "" || modelName == "" {
continue
}
ret.toolCodes = appendIfMissing(ret.toolCodes, toolCode)
ret.toolDefsByModelName[modelName] = toolCode
}
if len(filteredDefs) > 0 {
ret.toolCodes = appendIfMissing(ret.toolCodes, toolx.BuiltinToolSearch.Code)
ret.toolDefsByModelName[toolx.BuiltinToolSearch.Name] = toolx.BuiltinToolSearch.Code
}
if includeSkillTool {
ret.toolCodes = appendIfMissing(ret.toolCodes, toolx.BuiltinSkill.Code)
ret.toolDefsByModelName[toolx.BuiltinSkill.Name] = toolx.BuiltinSkill.Code
}
for modelName, toolCode := range ret.staticToolCodeMap {
modelName = strings.TrimSpace(modelName)
toolCode = strings.TrimSpace(toolCode)
if modelName == "" || toolCode == "" {
continue
}
ret.toolCodes = appendIfMissing(ret.toolCodes, toolCode)
ret.toolDefsByModelName[modelName] = toolCode
}
return ret
}
func toolSetStaticTools(toolSet *registry.ToolSet) []einotool.BaseTool {
if toolSet == nil {
return nil
}
return append([]einotool.BaseTool(nil), toolSet.StaticTools...)
}
func toolSetStaticToolCodes(toolSet *registry.ToolSet) map[string]string {
if toolSet == nil || len(toolSet.StaticToolCodes) == 0 {
return nil
}
ret := make(map[string]string, len(toolSet.StaticToolCodes))
for name, code := range toolSet.StaticToolCodes {
ret[strings.TrimSpace(name)] = strings.TrimSpace(code)
}
return ret
}
func toolSetStaticToolMetadata(toolSet *registry.ToolSet) map[string]registry.ToolMetadata {
if toolSet == nil || len(toolSet.StaticToolMetadata) == 0 {
return nil
}
ret := make(map[string]registry.ToolMetadata, len(toolSet.StaticToolMetadata))
for name, item := range toolSet.StaticToolMetadata {
trimmedName := strings.TrimSpace(name)
if trimmedName == "" {
continue
}
item.ToolCode = strings.TrimSpace(item.ToolCode)
item.ServerCode = strings.TrimSpace(item.ServerCode)
item.ToolName = strings.TrimSpace(item.ToolName)
ret[trimmedName] = item
}
return ret
}
func staticToolCodeList(toolSet *registry.ToolSet) []string {
if toolSet == nil {
return nil
}
metadata := toolSetStaticToolMetadata(toolSet)
ret := make([]string, 0, len(metadata))
for _, item := range metadata {
code := strings.TrimSpace(item.ToolCode)
if code == "" {
continue
}
ret = appendIfMissing(ret, code)
}
if len(ret) > 0 {
return ret
}
for _, code := range toolSetStaticToolCodes(toolSet) {
code = strings.TrimSpace(code)
if code == "" {
continue
}
ret = appendIfMissing(ret, code)
}
return ret
}
func definitionToolCodes(defs []runtimetooling.MCPToolDefinition) []string {
ret := make([]string, 0, len(defs))
for _, item := range defs {
code := strings.TrimSpace(item.ToolCode)
if code == "" {
continue
}
ret = append(ret, code)
}
return ret
}
func filterToolDefinitionsBySkill(defs []runtimetooling.MCPToolDefinition, skill *models.SkillDefinition) []runtimetooling.MCPToolDefinition {
if skill == nil || strings.TrimSpace(skill.ToolWhitelist) == "" {
return defs
}
var allowed []string
if err := json.Unmarshal([]byte(skill.ToolWhitelist), &allowed); err != nil {
return defs
}
allowedSet := make(map[string]struct{}, len(allowed))
for _, item := range allowed {
item = toolx.NormalizeToolCodeAlias(item)
if strings.TrimSpace(item) == "" {
continue
}
allowedSet[strings.TrimSpace(item)] = struct{}{}
}
if len(allowedSet) == 0 {
return defs
}
ret := make([]runtimetooling.MCPToolDefinition, 0, len(defs))
for _, item := range defs {
if _, ok := allowedSet[strings.TrimSpace(item.ToolCode)]; ok {
ret = append(ret, item)
}
}
return ret
}
func parseJSONArrayList(raw string) []string {
raw = strings.TrimSpace(raw)
if raw == "" {
return nil
}
var items []string
if err := json.Unmarshal([]byte(raw), &items); err != nil {
return nil
}
ret := make([]string, 0, len(items))
for _, item := range items {
item = strings.TrimSpace(item)
if item == "" {
continue
}
ret = append(ret, item)
}
return ret
}
-54
View File
@@ -1,54 +0,0 @@
package executor
import (
"agent-desk/internal/ai/runtime/registry"
"agent-desk/internal/models"
)
type RunInput struct {
Conversation models.Conversation
UserMessage models.Message
AIAgent models.AIAgent
AIConfig models.AIConfig
CheckPointID string
ToolSet *registry.ToolSet
}
type ResumeInput struct {
Conversation models.Conversation
AIAgent models.AIAgent
AIConfig models.AIConfig
CheckPointID string
ResumeData map[string]string
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
SelectedSkillID int64
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
}
@@ -1,26 +0,0 @@
package adapter
import "agent-desk/internal/models"
type AIConfigSnapshot struct {
ID int64
Provider string
ModelName string
BaseURL string
MaxOutputTokens int
TimeoutMS int
}
func BuildAIConfigSnapshot(item *models.AIConfig) *AIConfigSnapshot {
if item == nil {
return nil
}
return &AIConfigSnapshot{
ID: item.ID,
Provider: string(item.Provider),
ModelName: item.ModelName,
BaseURL: item.BaseURL,
MaxOutputTokens: item.MaxOutputTokens,
TimeoutMS: item.TimeoutMS,
}
}
@@ -1,22 +0,0 @@
package adapter
import "agent-desk/internal/models"
type ConversationSnapshot struct {
ID int64
AIAgentID int64
LastMessageID int64
CurrentAssigneeID int64
}
func BuildConversationSnapshot(item *models.Conversation) *ConversationSnapshot {
if item == nil {
return nil
}
return &ConversationSnapshot{
ID: item.ID,
AIAgentID: item.AIAgentID,
LastMessageID: item.LastMessageID,
CurrentAssigneeID: item.CurrentAssigneeID,
}
}
@@ -1,134 +0,0 @@
package adapter
import (
"context"
"encoding/json"
"fmt"
"strings"
"agent-desk/internal/ai/mcps"
"agent-desk/internal/ai/runtime/tooling"
"github.com/eino-contrib/jsonschema"
"github.com/cloudwego/eino/components/tool"
"github.com/cloudwego/eino/schema"
)
type MCPTool struct {
definition tooling.MCPToolDefinition
info *schema.ToolInfo
}
func NewMCPTool(definition tooling.MCPToolDefinition, metadata *mcps.ToolInfo) *MCPTool {
return &MCPTool{
definition: definition,
info: buildToolInfo(definition, metadata),
}
}
var _ tool.InvokableTool = (*MCPTool)(nil)
func (t *MCPTool) Info(ctx context.Context) (*schema.ToolInfo, error) {
if t == nil || t.info == nil {
return nil, nil
}
return t.info, nil
}
func (t *MCPTool) InvokableRun(ctx context.Context, argumentsInJSON string, opts ...tool.Option) (string, error) {
if t == nil {
return "", fmt.Errorf("mcp tool is nil")
}
arguments, err := parseArguments(argumentsInJSON)
if err != nil {
return "", err
}
arguments = mergeFixedArguments(arguments, t.definition.FixedArgs)
result, err := mcps.Runtime.CallTool(ctx, t.definition.ServerCode, t.definition.ToolName, arguments)
if err != nil {
return "", err
}
return BuildReducedToolResultSummary(result), nil
}
func buildToolInfo(definition tooling.MCPToolDefinition, metadata *mcps.ToolInfo) *schema.ToolInfo {
desc := strings.TrimSpace(definition.Description)
if desc == "" && metadata != nil {
desc = strings.TrimSpace(metadata.Description)
}
title := strings.TrimSpace(definition.Title)
if title == "" && metadata != nil {
title = strings.TrimSpace(metadata.Title)
}
if title != "" && desc != "" {
desc = title + "\n\n" + desc
} else if title != "" {
desc = title
}
if desc == "" {
desc = "Call MCP tool " + strings.TrimSpace(definition.ToolCode)
}
info := &schema.ToolInfo{
Name: tooling.BuildModelToolName(definition),
Desc: desc,
Extra: map[string]any{
"toolCode": definition.ToolCode,
"serverCode": definition.ServerCode,
"toolName": definition.ToolName,
},
}
if js := buildParamsSchema(metadata); js != nil {
info.ParamsOneOf = schema.NewParamsOneOfByJSONSchema(js)
}
return info
}
func buildParamsSchema(metadata *mcps.ToolInfo) *jsonschema.Schema {
if metadata == nil || metadata.InputSchema == nil {
return genericObjectSchema()
}
raw, err := json.Marshal(metadata.InputSchema)
if err != nil || len(raw) == 0 {
return genericObjectSchema()
}
js := &jsonschema.Schema{}
if err := json.Unmarshal(raw, js); err != nil {
return genericObjectSchema()
}
return js
}
func genericObjectSchema() *jsonschema.Schema {
return &jsonschema.Schema{
Version: jsonschema.Version,
Type: "object",
AdditionalProperties: &jsonschema.Schema{},
}
}
func parseArguments(argumentsInJSON string) (map[string]any, error) {
argumentsInJSON = strings.TrimSpace(argumentsInJSON)
if argumentsInJSON == "" {
return map[string]any{}, nil
}
args := make(map[string]any)
if err := json.Unmarshal([]byte(argumentsInJSON), &args); err != nil {
return nil, fmt.Errorf("invalid tool arguments: %w", err)
}
return args, nil
}
func mergeFixedArguments(arguments map[string]any, fixedArgs map[string]string) map[string]any {
if len(arguments) == 0 && len(fixedArgs) == 0 {
return map[string]any{}
}
ret := make(map[string]any, len(arguments)+len(fixedArgs))
for key, value := range arguments {
ret[key] = value
}
for key, value := range fixedArgs {
ret[key] = strings.TrimSpace(value)
}
return ret
}
@@ -1,68 +0,0 @@
package adapter
import (
"agent-desk/internal/models"
"agent-desk/internal/pkg/enums"
"agent-desk/internal/pkg/utils"
"agent-desk/internal/repositories"
"github.com/cloudwego/eino/schema"
"github.com/mlogclub/simple/sqls"
)
const defaultHistoryLimit = 12
type HistoryBuildResult struct {
Messages []*schema.Message
RawItems []models.Message
}
func BuildHistoryMessages(conversationID int64, currentMessageID int64, limit int) HistoryBuildResult {
if conversationID <= 0 {
return HistoryBuildResult{}
}
if limit <= 0 {
limit = defaultHistoryLimit
}
items := repositories.MessageRepository.Find(sqls.DB(), sqls.NewCnd().
Eq("conversation_id", conversationID).
Desc("id").
Limit(limit+1))
for i, j := 0, len(items)-1; i < j; i, j = i+1, j-1 {
items[i], items[j] = items[j], items[i]
}
ret := HistoryBuildResult{
Messages: make([]*schema.Message, 0, len(items)),
RawItems: make([]models.Message, 0, len(items)),
}
for _, item := range items {
if item.ID == currentMessageID {
continue
}
msg := BuildSchemaMessage(&item)
if msg == nil {
continue
}
ret.RawItems = append(ret.RawItems, item)
ret.Messages = append(ret.Messages, msg)
}
return ret
}
func BuildSchemaMessage(item *models.Message) *schema.Message {
if item == nil {
return nil
}
content := utils.BuildRuntimeMessageText(item.MessageType, item.Content)
if content == "" {
return nil
}
switch item.SenderType {
case enums.IMSenderTypeCustomer:
return schema.UserMessage(content)
case enums.IMSenderTypeAI, enums.IMSenderTypeAgent:
return schema.AssistantMessage(content, nil)
default:
return nil
}
}
@@ -1,55 +0,0 @@
package agents
import (
"context"
"fmt"
"github.com/cloudwego/eino/adk"
)
type CustomerServiceAgent struct {
Inner adk.Agent
}
var _ adk.ResumableAgent = (*CustomerServiceAgent)(nil)
func (a *CustomerServiceAgent) Name(ctx context.Context) string {
if a == nil || a.Inner == nil {
return "customer_service_agent"
}
return a.Inner.Name(ctx)
}
func (a *CustomerServiceAgent) Description(ctx context.Context) string {
if a == nil || a.Inner == nil {
return "customer service chat agent"
}
return a.Inner.Description(ctx)
}
func (a *CustomerServiceAgent) Run(ctx context.Context, input *adk.AgentInput, options ...adk.AgentRunOption) *adk.AsyncIterator[*adk.AgentEvent] {
if a == nil || a.Inner == nil {
iter, gen := adk.NewAsyncIteratorPair[*adk.AgentEvent]()
gen.Send(&adk.AgentEvent{Err: context.Canceled})
gen.Close()
return iter
}
return a.Inner.Run(ctx, input, options...)
}
func (a *CustomerServiceAgent) Resume(ctx context.Context, info *adk.ResumeInfo, options ...adk.AgentRunOption) *adk.AsyncIterator[*adk.AgentEvent] {
if a == nil || a.Inner == nil {
iter, gen := adk.NewAsyncIteratorPair[*adk.AgentEvent]()
gen.Send(&adk.AgentEvent{Err: fmt.Errorf("customer service agent is not initialized")})
gen.Close()
return iter
}
ra, ok := a.Inner.(adk.ResumableAgent)
if !ok {
iter, gen := adk.NewAsyncIteratorPair[*adk.AgentEvent]()
gen.Send(&adk.AgentEvent{Err: fmt.Errorf("inner agent %q does not implement resumable agent", a.Inner.Name(ctx))})
gen.Close()
return iter
}
return ra.Resume(ctx, info, options...)
}
@@ -1,324 +0,0 @@
package callbacks
import (
"context"
"encoding/json"
"strconv"
"strings"
"time"
impladapter "agent-desk/internal/ai/runtime/internal/impl/adapter"
"agent-desk/internal/pkg/enums"
"agent-desk/internal/pkg/toolx"
einotool "github.com/cloudwego/eino/components/tool"
"github.com/cloudwego/eino/adk"
)
type ToolMetadata struct {
ToolCode string
ServerCode string
ToolName string
SourceType enums.ToolSourceType
}
type RuntimeTraceHandler struct {
*adk.BaseChatModelAgentMiddleware
collector *RuntimeTraceCollector
toolMetadataBy map[string]ToolMetadata
skillMetadataBy map[string]SkillMetadata
}
type graphAnalyzeConversationResult struct {
RecommendedNextAction string `json:"recommendedNextAction"`
RiskLevel string `json:"riskLevel"`
}
type graphTriageAnalysisResult struct {
RiskLevel string `json:"riskLevel"`
}
type graphTriageTicketDraftResult struct {
Ready bool `json:"ready"`
}
type graphTriageServiceRequestResult struct {
RecommendedAction string `json:"recommendedAction"`
Analysis graphTriageAnalysisResult `json:"analysis"`
TicketDraft *graphTriageTicketDraftResult `json:"ticketDraft"`
}
type toolSearchArguments struct {
Query string `json:"query"`
RegexPattern string `json:"regex_pattern"`
ToolCode string `json:"toolCode"`
}
type toolSearchCandidateResult struct {
ToolCode string `json:"toolCode"`
}
type toolSearchInvokeResult struct {
SelectedTools []string `json:"selectedTools"`
}
type toolSearchSearchResult struct {
Candidates []toolSearchCandidateResult `json:"candidates"`
}
func NewRuntimeTraceHandler(collector *RuntimeTraceCollector, toolMetadataBy map[string]ToolMetadata, skillMetadataBy map[string]SkillMetadata) *RuntimeTraceHandler {
return &RuntimeTraceHandler{
BaseChatModelAgentMiddleware: &adk.BaseChatModelAgentMiddleware{},
collector: collector,
toolMetadataBy: toolMetadataBy,
skillMetadataBy: skillMetadataBy,
}
}
func (h *RuntimeTraceHandler) WrapInvokableToolCall(_ context.Context, endpoint adk.InvokableToolCallEndpoint, tCtx *adk.ToolContext) (adk.InvokableToolCallEndpoint, error) {
return func(ctx context.Context, argumentsInJSON string, opts ...einotool.Option) (string, error) {
startedAt := time.Now()
result, err := endpoint(ctx, argumentsInJSON, opts...)
item := ToolTraceItem{
ResultPreview: previewToolText(result, 300),
LatencyMs: time.Since(startedAt).Milliseconds(),
Status: "ok",
}
reductionInfo := impladapter.ParseReductionInfo(result)
item.ResultReduced = reductionInfo.Reduced
item.OriginalChars = reductionInfo.OriginalChars
item.KeptChars = reductionInfo.KeptChars
if tCtx != nil {
item.ToolName = strings.TrimSpace(tCtx.Name)
if metadata, ok := h.toolMetadataBy[item.ToolName]; ok {
item.ToolCode = metadata.ToolCode
item.ServerCode = metadata.ServerCode
item.ToolName = metadata.ToolName
}
}
if arguments := parseToolArguments(argumentsInJSON); len(arguments) > 0 {
item.Arguments = arguments
}
if err != nil {
item.Status = "error"
item.ErrorMessage = err.Error()
}
h.collector.AddToolItem(item)
if metadata, ok := h.resolveToolMetadata(item.ToolName); ok && metadata.SourceType == enums.ToolSourceTypeGraph {
recommendedAction, riskLevel, ticketDraftReady := parseGraphToolOutcome(item.ToolCode, result)
h.collector.AddGraphToolItem(GraphToolTraceItem{
ToolCode: item.ToolCode,
ToolName: item.ToolName,
Arguments: item.Arguments,
ResultPreview: item.ResultPreview,
ResultReduced: item.ResultReduced,
OriginalChars: item.OriginalChars,
KeptChars: item.KeptChars,
LatencyMs: item.LatencyMs,
Status: item.Status,
ErrorMessage: item.ErrorMessage,
RecommendedAction: recommendedAction,
RiskLevel: riskLevel,
TicketDraftReady: ticketDraftReady,
})
}
if metadata, ok := h.resolveToolMetadata(item.ToolName); ok && strings.TrimSpace(metadata.ToolCode) == toolx.BuiltinToolSearch.Code {
h.collector.AddToolSearchItem(h.buildToolSearchTraceItem(argumentsInJSON, result, err))
}
if metadata, ok := h.resolveToolMetadata(item.ToolName); ok && strings.TrimSpace(metadata.ToolCode) == toolx.BuiltinSkill.Code {
h.tryActivateSkill(argumentsInJSON)
}
return result, err
}, nil
}
func (h *RuntimeTraceHandler) tryActivateSkill(argumentsInJSON string) {
if h == nil || h.collector == nil {
return
}
var args struct {
Skill string `json:"skill"`
}
if err := json.Unmarshal([]byte(strings.TrimSpace(argumentsInJSON)), &args); err != nil {
return
}
skillKey := strings.TrimSpace(args.Skill)
if skillKey == "" {
return
}
meta, ok := h.skillMetadataBy[skillKey]
if !ok {
if id, err := strconv.ParseInt(skillKey, 10, 64); err == nil {
meta = SkillMetadata{ID: id}
}
}
buf, err := json.Marshal(map[string]any{
"source": "eino_skill_tool",
"skillId": skillKey,
})
routeTrace := ""
if err == nil {
routeTrace = string(buf)
}
h.collector.ActivateSkill(meta, "eino_skill_tool", routeTrace)
}
func parseGraphToolOutcome(toolCode string, result string) (recommendedAction, riskLevel string, ticketDraftReady bool) {
toolCode = strings.TrimSpace(toolCode)
if toolCode == "" || strings.TrimSpace(result) == "" {
return "", "", false
}
switch toolCode {
case toolx.GraphAnalyzeConversation.Code:
var payload graphAnalyzeConversationResult
if err := json.Unmarshal([]byte(strings.TrimSpace(result)), &payload); err != nil {
return "", "", false
}
return strings.TrimSpace(payload.RecommendedNextAction), strings.TrimSpace(payload.RiskLevel), false
case toolx.GraphTriageServiceRequest.Code:
var payload graphTriageServiceRequestResult
if err := json.Unmarshal([]byte(strings.TrimSpace(result)), &payload); err != nil {
return "", "", false
}
recommendedAction = strings.TrimSpace(payload.RecommendedAction)
riskLevel = strings.TrimSpace(payload.Analysis.RiskLevel)
if payload.TicketDraft != nil {
ticketDraftReady = payload.TicketDraft.Ready
}
return recommendedAction, riskLevel, ticketDraftReady
default:
return "", "", false
}
}
func (h *RuntimeTraceHandler) resolveToolMetadata(modelToolName string) (ToolMetadata, bool) {
if h == nil || h.toolMetadataBy == nil {
return ToolMetadata{}, false
}
modelToolName = strings.TrimSpace(modelToolName)
if modelToolName == "" {
return ToolMetadata{}, false
}
if spec, ok := toolx.GetRegisteredToolSpecByName(modelToolName); ok {
resolved := toolx.ResolveToolMetadata(spec.Code, spec.Name)
return ToolMetadata{
ToolCode: resolved.ToolCode,
ServerCode: resolved.ServerCode,
ToolName: resolved.ToolName,
SourceType: resolved.SourceType,
}, true
}
metadata, ok := h.toolMetadataBy[modelToolName]
return metadata, ok
}
func parseToolArguments(argumentsInJSON string) map[string]any {
argumentsInJSON = strings.TrimSpace(argumentsInJSON)
if argumentsInJSON == "" {
return nil
}
ret := make(map[string]any)
if err := json.Unmarshal([]byte(argumentsInJSON), &ret); err != nil {
return nil
}
return ret
}
func previewToolText(text string, limit int) string {
if limit <= 0 {
return ""
}
text = strings.TrimSpace(text)
runes := []rune(text)
if len(runes) <= limit {
return text
}
return string(runes[:limit]) + "..."
}
func (h *RuntimeTraceHandler) buildToolSearchTraceItem(argumentsInJSON string, result string, runErr error) ToolSearchTraceItem {
item := ToolSearchTraceItem{Status: "ok"}
var args toolSearchArguments
if strings.TrimSpace(argumentsInJSON) != "" {
_ = json.Unmarshal([]byte(argumentsInJSON), &args)
}
item.Query = strings.TrimSpace(firstNonBlank(args.Query, args.RegexPattern))
item.TargetToolCode = strings.TrimSpace(args.ToolCode)
item.TargetServerCode, item.TargetToolName = toolx.SplitMCPToolCode(item.TargetToolCode)
if item.TargetToolCode != "" {
item.Action = "invoke"
} else {
item.Action = "search"
}
if runErr != nil {
item.Status = "error"
item.ErrorMessage = runErr.Error()
return item
}
item.CandidateToolCodes = h.extractCandidateToolCodes(result)
return item
}
func firstNonBlank(values ...string) string {
for _, value := range values {
value = strings.TrimSpace(value)
if value != "" {
return value
}
}
return ""
}
func (h *RuntimeTraceHandler) extractCandidateToolCodes(result string) []string {
result = strings.TrimSpace(result)
if result == "" {
return nil
}
var invokePayload toolSearchInvokeResult
if err := json.Unmarshal([]byte(result), &invokePayload); err == nil && len(invokePayload.SelectedTools) > 0 {
return h.extractSelectedToolCodes(invokePayload.SelectedTools)
}
var searchPayload toolSearchSearchResult
if err := json.Unmarshal([]byte(result), &searchPayload); err == nil && len(searchPayload.Candidates) > 0 {
return extractCandidateObjectCodes(searchPayload.Candidates)
}
return nil
}
func (h *RuntimeTraceHandler) extractSelectedToolCodes(items []string) []string {
if len(items) == 0 {
return nil
}
ret := make([]string, 0, len(items))
for _, item := range items {
toolName := strings.TrimSpace(item)
if toolName == "" {
continue
}
toolCode := toolName
if metadata, ok := h.resolveToolMetadata(toolName); ok && strings.TrimSpace(metadata.ToolCode) != "" {
toolCode = strings.TrimSpace(metadata.ToolCode)
}
if toolCode == "" {
continue
}
ret = append(ret, toolCode)
}
return ret
}
func extractCandidateObjectCodes(items []toolSearchCandidateResult) []string {
if len(items) == 0 {
return nil
}
ret := make([]string, 0, len(items))
for _, item := range items {
toolCode := strings.TrimSpace(item.ToolCode)
if toolCode == "" {
continue
}
ret = append(ret, toolCode)
}
return ret
}
@@ -1,67 +0,0 @@
package callbacks
import (
"testing"
"agent-desk/internal/pkg/toolx"
)
func TestParseGraphToolOutcome(t *testing.T) {
action, risk, ready := parseGraphToolOutcome(toolx.GraphAnalyzeConversation.Code, `{"recommendedNextAction":"handoff_to_human","riskLevel":"high"}`)
if action != "handoff_to_human" || risk != "high" || ready {
t.Fatalf("unexpected analyze graph outcome: %q %q %v", action, risk, ready)
}
action, risk, ready = parseGraphToolOutcome(toolx.GraphTriageServiceRequest.Code, `{"recommendedAction":"prepare_ticket","analysis":{"riskLevel":"medium"},"ticketDraft":{"ready":true}}`)
if action != "prepare_ticket" || risk != "medium" || !ready {
t.Fatalf("unexpected triage graph outcome: %q %q %v", action, risk, ready)
}
}
func TestExtractCandidateToolCodes(t *testing.T) {
handler := &RuntimeTraceHandler{
toolMetadataBy: map[string]ToolMetadata{
"tool_search": {ToolCode: toolx.BuiltinToolSearch.Code, ToolName: toolx.BuiltinToolSearch.Name},
"foo_model": {ToolCode: "mcp/server/foo", ToolName: "foo"},
},
}
got := handler.extractCandidateToolCodes(`{"selectedTools":["foo_model"]}`)
if len(got) != 1 || got[0] != "mcp/server/foo" {
t.Fatalf("unexpected selectedTools codes: %#v", got)
}
got = handler.extractCandidateToolCodes(`{"candidates":[{"toolCode":"mcp/server/bar"}]}`)
if len(got) != 1 || got[0] != "mcp/server/bar" {
t.Fatalf("unexpected candidate codes: %#v", got)
}
}
func TestTryActivateSkill(t *testing.T) {
collector := NewRuntimeTraceCollector()
handler := &RuntimeTraceHandler{
collector: collector,
skillMetadataBy: map[string]SkillMetadata{
"44": {
ID: 44,
Name: "售后升级",
AllowedToolCodes: []string{"graph/handoff_to_human"},
},
},
}
handler.tryActivateSkill(`{"skill":"44"}`)
if collector.Data.Skill.ID != 44 {
t.Fatalf("unexpected skill id: %#v", collector.Data.Skill)
}
if collector.Data.Skill.Name != "售后升级" {
t.Fatalf("unexpected skill name: %#v", collector.Data.Skill)
}
if collector.Data.Skill.RouteReason != "eino_skill_tool" {
t.Fatalf("unexpected route reason: %#v", collector.Data.Skill)
}
if len(collector.Data.Skill.AllowedToolCodes) != 1 || collector.Data.Skill.AllowedToolCodes[0] != "graph/handoff_to_human" {
t.Fatalf("unexpected allowed tools: %#v", collector.Data.Skill.AllowedToolCodes)
}
}
@@ -1,150 +0,0 @@
package callbacks
import (
"encoding/json"
"sync"
)
type RuntimeTraceCollector struct {
mu sync.Mutex
Data RuntimeTraceData
}
func NewRuntimeTraceCollector() *RuntimeTraceCollector {
ret := &RuntimeTraceCollector{}
ret.Data.Version = "v1"
ret.Data.Status = "started"
return ret
}
func (c *RuntimeTraceCollector) Marshal() string {
c.mu.Lock()
defer c.mu.Unlock()
buf, err := json.Marshal(c.Data)
if err != nil {
return ""
}
return string(buf)
}
func (c *RuntimeTraceCollector) SetTooling(staticToolCodes []string, dynamicToolCodes []string, toolSearchEnabled bool) {
c.mu.Lock()
defer c.mu.Unlock()
c.Data.Input.StaticToolCodes = append([]string(nil), staticToolCodes...)
c.Data.Input.DynamicToolCodes = append([]string(nil), dynamicToolCodes...)
c.Data.Input.ToolSearchEnabled = toolSearchEnabled
}
func (c *RuntimeTraceCollector) SetInstructionSummary(summary InstructionTraceSummary) {
c.mu.Lock()
defer c.mu.Unlock()
c.Data.Instruction.SectionTitles = append([]string(nil), summary.SectionTitles...)
c.Data.Instruction.HasAgentRule = summary.HasAgentRule
c.Data.Instruction.HasSkillRule = summary.HasSkillRule
c.Data.Instruction.HasToolRule = summary.HasToolRule
}
func (c *RuntimeTraceCollector) SetSkillMiddleware(enabled bool, toolName string) {
c.mu.Lock()
defer c.mu.Unlock()
c.Data.Skill.MiddlewareEnabled = enabled
c.Data.Skill.MiddlewareToolName = toolName
}
type SkillMetadata struct {
ID int64
Name string
Description string
AllowedToolCodes []string
}
func (c *RuntimeTraceCollector) SetVisibleSkills(skills map[string]SkillMetadata) {
if len(skills) == 0 {
return
}
c.mu.Lock()
defer c.mu.Unlock()
ids := make([]int64, 0, len(skills))
for _, skill := range skills {
if skill.ID <= 0 {
continue
}
ids = append(ids, skill.ID)
}
c.Data.Skill.VisibleIDs = append([]int64(nil), ids...)
}
func (c *RuntimeTraceCollector) ActivateSkill(skill SkillMetadata, routeReason string, routeTrace string) {
c.mu.Lock()
defer c.mu.Unlock()
c.Data.Skill.ID = skill.ID
c.Data.Skill.Name = skill.Name
c.Data.Skill.Description = skill.Description
c.Data.Skill.AllowedToolCodes = append([]string(nil), skill.AllowedToolCodes...)
c.Data.Skill.RouteReason = routeReason
c.Data.Skill.RouteTrace = routeTrace
}
func (c *RuntimeTraceCollector) SetFilteredToolCodes(toolCodes []string) {
c.mu.Lock()
defer c.mu.Unlock()
c.Data.Skill.FilteredToolCodes = append([]string(nil), toolCodes...)
}
func (c *RuntimeTraceCollector) SetRetrieverSummary(summary RetrieverTraceSummary) {
c.mu.Lock()
defer c.mu.Unlock()
c.Data.Retriever.TopK = summary.TopK
c.Data.Retriever.ScoreThreshold = summary.ScoreThreshold
c.Data.Retriever.ContextMaxTokens = summary.ContextMaxTokens
c.Data.Retriever.MaxContextItems = summary.MaxContextItems
c.Data.Retriever.Count = summary.HitCount
c.Data.Retriever.ContextCount = summary.ContextCount
c.Data.Retriever.EmbeddingMs = summary.EmbeddingMs
c.Data.Retriever.VectorSearchMs = summary.VectorSearchMs
c.Data.Retriever.HydrateMs = summary.HydrateMs
c.Data.Retriever.Policies = append([]RetrieverPolicyTraceItem(nil), summary.Policies...)
}
func (c *RuntimeTraceCollector) AddRetrieverItems(items []RetrieverTraceItem) {
if len(items) == 0 {
return
}
c.mu.Lock()
defer c.mu.Unlock()
c.Data.Retriever.Items = append(c.Data.Retriever.Items, items...)
}
func (c *RuntimeTraceCollector) SetAnswerability(data AnswerabilityTraceData) {
c.mu.Lock()
defer c.mu.Unlock()
c.Data.Answerability = AnswerabilityTraceData{
Status: data.Status,
Reason: data.Reason,
SupportingChunkIDs: append([]string(nil), data.SupportingChunkIDs...),
MissingInfo: append([]string(nil), data.MissingInfo...),
LatencyMs: data.LatencyMs,
ErrorMessage: data.ErrorMessage,
}
}
func (c *RuntimeTraceCollector) AddToolItem(item ToolTraceItem) {
c.mu.Lock()
defer c.mu.Unlock()
c.Data.Tools.Count++
c.Data.Tools.Items = append(c.Data.Tools.Items, item)
}
func (c *RuntimeTraceCollector) AddToolSearchItem(item ToolSearchTraceItem) {
c.mu.Lock()
defer c.mu.Unlock()
c.Data.ToolSearch.Count++
c.Data.ToolSearch.Items = append(c.Data.ToolSearch.Items, item)
}
func (c *RuntimeTraceCollector) AddGraphToolItem(item GraphToolTraceItem) {
c.mu.Lock()
defer c.mu.Unlock()
c.Data.GraphTools.Count++
c.Data.GraphTools.Items = append(c.Data.GraphTools.Items, item)
}
@@ -1,20 +0,0 @@
package callbacks
import "testing"
func TestRuntimeTraceCollectorAddRetrieverItems(t *testing.T) {
collector := NewRuntimeTraceCollector()
collector.AddRetrieverItems([]RetrieverTraceItem{
{KnowledgeBaseID: 1, DocumentID: 10, DocumentTitle: "doc-1"},
{KnowledgeBaseID: 2, DocumentID: 20, DocumentTitle: "doc-2"},
})
collector.AddRetrieverItems(nil)
if got := len(collector.Data.Retriever.Items); got != 2 {
t.Fatalf("expected two retriever items, got %d", got)
}
if collector.Data.Retriever.Items[0].DocumentTitle != "doc-1" || collector.Data.Retriever.Items[1].DocumentTitle != "doc-2" {
t.Fatalf("unexpected retriever items: %#v", collector.Data.Retriever.Items)
}
}
@@ -1,105 +0,0 @@
package factory
import (
"context"
"strings"
"agent-desk/internal/ai/runtime/instruction"
"agent-desk/internal/ai/runtime/internal/impl/agents"
"agent-desk/internal/ai/runtime/internal/impl/callbacks"
"agent-desk/internal/ai/runtime/registry"
"agent-desk/internal/ai/runtime/tooling"
"agent-desk/internal/models"
"github.com/cloudwego/eino/adk"
"github.com/cloudwego/eino/components/tool"
"github.com/cloudwego/eino/compose"
)
type AgentFactory struct {
chatModelFactory *ChatModelFactory
toolFactory *ToolFactory
instructionService *instruction.Service
handlerService *AgentHandlerService
}
// BuildCustomerServiceAgentInput 定义客服 Agent 的装配输入。
//
// 之所以收敛成单一输入对象,而不是继续堆叠函数参数,是为了避免:
// 1. 调用点无法看懂每个位置参数的语义;
// 2. instruction 用工具、动态工具、中间件工具之间职责混淆;
// 3. 后续扩展装配项时继续拉长函数签名。
type BuildCustomerServiceAgentInput struct {
// AIAgent 为当前运行的业务 Agent 配置,提供名称、描述、系统提示词等基础信息。
AIAgent models.AIAgent
// AIConfig 为模型配置,决定底层使用哪个 ChatModel。
AIConfig models.AIConfig
// InstructionToolDefinitions 用于生成 instruction 中的工具说明。
// 它描述“当前允许模型理解和使用的 MCP 工具范围”。
InstructionToolDefinitions []tooling.MCPToolDefinition
// DynamicMCPToolDefinitions 用于接入 Eino tool_search middleware 的动态工具集合。
// 这些工具默认不直接挂在 ToolsNode 上,而是经 tool_search 选择后再暴露给模型。
DynamicMCPToolDefinitions []tooling.MCPToolDefinition
// StaticTools 为当前运行时直接挂载到 ToolsNode 的固定工具,例如 Graph Tool。
StaticTools []tool.BaseTool
// StaticToolCodes 为固定工具的 modelName -> toolCode 映射,用于 trace 和运行日志归因。
StaticToolCodes map[string]string
// StaticToolMetadata 为固定工具的 modelName -> metadata 映射,用于 trace 和运行日志归因。
StaticToolMetadata map[string]registry.ToolMetadata
// Collector 用于收集运行链路中的 tool trace、graph trace 等调试信息。
Collector *callbacks.RuntimeTraceCollector
}
func NewAgentFactory() *AgentFactory {
return &AgentFactory{
chatModelFactory: NewChatModelFactory(),
toolFactory: NewToolFactory(),
instructionService: instruction.NewService(nil, nil, nil),
handlerService: NewAgentHandlerService(NewSkillMiddlewareService()),
}
}
// BuildCustomerServiceAgent 根据装配输入构建客服 ChatModelAgent。
func (f *AgentFactory) BuildCustomerServiceAgent(ctx context.Context, input BuildCustomerServiceAgentInput) (*agents.CustomerServiceAgent, error) {
chatModel, err := f.chatModelFactory.Build(ctx, input.AIConfig)
if err != nil {
return nil, err
}
dynamicTools, err := f.toolFactory.BuildBaseToolsByDefinitions(ctx, input.DynamicMCPToolDefinitions)
if err != nil {
return nil, err
}
allTools := make([]tool.BaseTool, 0, len(input.StaticTools))
allTools = append(allTools, input.StaticTools...)
instructionResult := f.instructionService.Build(input.AIAgent, nil, input.InstructionToolDefinitions, input.StaticToolCodes)
handlers := make([]adk.ChatModelAgentMiddleware, 0, 3)
builtHandlers, err := f.handlerService.Build(ctx, BuildAgentHandlersInput{
AIAgent: input.AIAgent,
InstructionToolDefinitions: input.InstructionToolDefinitions,
DynamicToolDefinitions: input.DynamicMCPToolDefinitions,
DynamicTools: dynamicTools,
StaticToolMetadata: input.StaticToolMetadata,
Collector: input.Collector,
InstructionSummary: buildInstructionTraceSummary(instructionResult.Summary),
})
if err != nil {
return nil, err
}
handlers = append(handlers, builtHandlers...)
inner, err := adk.NewChatModelAgent(ctx, &adk.ChatModelAgentConfig{
Name: strings.TrimSpace(input.AIAgent.Name),
Description: strings.TrimSpace(input.AIAgent.Description),
Instruction: instructionResult.Text,
Model: chatModel,
ToolsConfig: adk.ToolsConfig{
ToolsNodeConfig: compose.ToolsNodeConfig{
Tools: allTools,
},
},
Handlers: handlers,
})
if err != nil {
return nil, err
}
return &agents.CustomerServiceAgent{Inner: inner}, nil
}
@@ -1,98 +0,0 @@
package factory
import (
"context"
"strings"
einocallbacks "agent-desk/internal/ai/runtime/internal/impl/callbacks"
"agent-desk/internal/ai/runtime/registry"
runtimetooling "agent-desk/internal/ai/runtime/tooling"
"agent-desk/internal/models"
"agent-desk/internal/pkg/toolx"
"github.com/cloudwego/eino/adk"
einotoolsearch "github.com/cloudwego/eino/adk/middlewares/dynamictool/toolsearch"
einobasetool "github.com/cloudwego/eino/components/tool"
)
type AgentHandlerService struct {
skillMiddleware *SkillMiddlewareService
}
type BuildAgentHandlersInput struct {
AIAgent models.AIAgent
InstructionToolDefinitions []runtimetooling.MCPToolDefinition
DynamicToolDefinitions []runtimetooling.MCPToolDefinition
DynamicTools []einobasetool.BaseTool
StaticToolMetadata map[string]registry.ToolMetadata
Collector *einocallbacks.RuntimeTraceCollector
InstructionSummary einocallbacks.InstructionTraceSummary
}
func NewAgentHandlerService(skillMiddleware *SkillMiddlewareService) *AgentHandlerService {
if skillMiddleware == nil {
panic("skill middleware is required")
}
return &AgentHandlerService{skillMiddleware: skillMiddleware}
}
func (s *AgentHandlerService) Build(ctx context.Context, input BuildAgentHandlersInput) ([]adk.ChatModelAgentMiddleware, error) {
handlers := make([]adk.ChatModelAgentMiddleware, 0, 4)
skillMetadataByID := buildRuntimeSkillMetadataMap(input.AIAgent)
toolMetadataBy := buildRuntimeTraceToolMetadata(input.DynamicToolDefinitions, input.StaticToolMetadata, len(skillMetadataByID) > 0)
traceSkillMetadata := make(map[string]einocallbacks.SkillMetadata, len(skillMetadataByID))
for id, item := range skillMetadataByID {
traceSkillMetadata[id] = einocallbacks.SkillMetadata{
ID: item.ID,
Name: item.Name,
Description: item.Description,
AllowedToolCodes: append([]string(nil), item.AllowedToolCodes...),
}
}
if input.Collector != nil {
if len(skillMetadataByID) > 0 {
input.Collector.SetSkillMiddleware(true, toolx.BuiltinSkill.Name)
}
input.Collector.SetVisibleSkills(traceSkillMetadata)
input.Collector.SetInstructionSummary(input.InstructionSummary)
handlers = append(handlers, einocallbacks.NewRuntimeTraceHandler(input.Collector, toolMetadataBy, traceSkillMetadata))
}
if len(input.DynamicTools) > 0 {
toolSearchHandler, err := einotoolsearch.New(ctx, &einotoolsearch.Config{
DynamicTools: input.DynamicTools,
})
if err != nil {
return nil, err
}
handlers = append(handlers, toolSearchHandler)
}
if len(skillMetadataByID) > 0 {
skillHandler, err := s.skillMiddleware.Build(ctx, input.AIAgent, input.InstructionToolDefinitions)
if err != nil {
return nil, err
}
handlers = append(handlers, skillHandler)
handlers = append(handlers, NewRuntimeToolFilterMiddleware(
input.Collector,
toolMetadataBy,
traceSkillMetadata,
dynamicToolModelNames(input.DynamicToolDefinitions),
))
}
return handlers, nil
}
func dynamicToolModelNames(definitions []runtimetooling.MCPToolDefinition) []string {
ret := make([]string, 0, len(definitions))
for _, item := range definitions {
modelName := strings.TrimSpace(item.ModelName)
if modelName == "" {
modelName = strings.TrimSpace(runtimetooling.BuildModelToolName(item))
}
if modelName == "" {
continue
}
ret = append(ret, modelName)
}
return ret
}
@@ -1,45 +0,0 @@
package factory
import (
"context"
"testing"
einocallbacks "agent-desk/internal/ai/runtime/internal/impl/callbacks"
)
func TestAgentHandlerServiceBuildWithCollectorOnly(t *testing.T) {
collector := einocallbacks.NewRuntimeTraceCollector()
service := NewAgentHandlerService(NewSkillMiddlewareService())
handlers, err := service.Build(context.Background(), BuildAgentHandlersInput{
Collector: collector,
InstructionSummary: einocallbacks.InstructionTraceSummary{
SectionTitles: []string{"Agent 规则"},
HasAgentRule: true,
},
})
if err != nil {
t.Fatalf("Build returned error: %v", err)
}
if len(handlers) != 1 {
t.Fatalf("expected 1 handler, got %d", len(handlers))
}
if !collector.Data.Instruction.HasAgentRule {
t.Fatalf("instruction summary was not written to collector: %#v", collector.Data.Instruction)
}
if len(collector.Data.Instruction.SectionTitles) != 1 {
t.Fatalf("unexpected section titles: %#v", collector.Data.Instruction.SectionTitles)
}
}
func TestAgentHandlerServiceBuildWithEmptyInput(t *testing.T) {
service := NewAgentHandlerService(NewSkillMiddlewareService())
handlers, err := service.Build(context.Background(), BuildAgentHandlersInput{})
if err != nil {
t.Fatalf("Build returned error: %v", err)
}
if len(handlers) != 0 {
t.Fatalf("expected no handlers, got %d", len(handlers))
}
}
@@ -1,58 +0,0 @@
package factory
import (
"context"
"strings"
"time"
"agent-desk/internal/models"
"agent-desk/internal/pkg/enums"
openai "github.com/cloudwego/eino-ext/components/model/openai"
"github.com/cloudwego/eino/components/model"
)
type ChatModelFactory struct{}
func NewChatModelFactory() *ChatModelFactory {
return &ChatModelFactory{}
}
func (f *ChatModelFactory) Build(ctx context.Context, aiConfig models.AIConfig) (model.ToolCallingChatModel, error) {
conf := &openai.ChatModelConfig{
APIKey: strings.TrimSpace(aiConfig.APIKey),
BaseURL: strings.TrimSpace(aiConfig.BaseURL),
Model: strings.TrimSpace(aiConfig.ModelName),
}
if aiConfig.TimeoutMS > 0 {
conf.Timeout = time.Duration(aiConfig.TimeoutMS) * time.Millisecond
}
if aiConfig.MaxOutputTokens > 0 {
maxCompletionTokens := aiConfig.MaxOutputTokens
conf.MaxCompletionTokens = &maxCompletionTokens
}
if aiConfig.Provider == enums.AIProviderOpenAI && isAzureOpenAIBaseURL(aiConfig.BaseURL) {
conf.ByAzure = true
conf.APIVersion = "2024-06-01"
}
if extraFields := providerExtraFields(aiConfig); len(extraFields) > 0 {
conf.ExtraFields = extraFields
}
return openai.NewChatModel(ctx, conf)
}
func isAzureOpenAIBaseURL(baseURL string) bool {
baseURL = strings.ToLower(strings.TrimSpace(baseURL))
return strings.Contains(baseURL, ".openai.azure.com")
}
func providerExtraFields(aiConfig models.AIConfig) map[string]any {
baseURL := strings.ToLower(strings.TrimSpace(aiConfig.BaseURL))
modelName := strings.ToLower(strings.TrimSpace(aiConfig.ModelName))
if strings.Contains(baseURL, "dashscope.aliyuncs.com") && strings.HasPrefix(modelName, "qwen3") {
return map[string]any{
"enable_thinking": false,
}
}
return nil
}
@@ -1,27 +0,0 @@
package factory
import (
"context"
einostore "agent-desk/internal/ai/runtime/internal/impl/store"
"github.com/cloudwego/eino/adk"
)
type RunnerFactory struct{}
func NewRunnerFactory() *RunnerFactory {
return &RunnerFactory{}
}
func (f *RunnerFactory) Build(ctx context.Context, agent adk.Agent, enableStreaming bool, enableCheckpoint bool) *adk.Runner {
var checkpointStore adk.CheckPointStore
if enableCheckpoint {
checkpointStore = einostore.DefaultCheckPointStore
}
return adk.NewRunner(ctx, adk.RunnerConfig{
Agent: agent,
EnableStreaming: enableStreaming,
CheckPointStore: checkpointStore,
})
}
@@ -1,188 +0,0 @@
package factory
import (
"context"
"encoding/json"
"fmt"
"strconv"
"strings"
runtimeinstruction "agent-desk/internal/ai/runtime/instruction"
runtimetooling "agent-desk/internal/ai/runtime/tooling"
"agent-desk/internal/models"
"agent-desk/internal/pkg/enums"
"agent-desk/internal/pkg/utils"
"agent-desk/internal/services"
einoskill "github.com/cloudwego/eino/adk/middlewares/skill"
)
type runtimeSkillMetadata struct {
ID int64
Name string
Description string
AllowedToolCodes []string
}
type databaseSkillBackend struct {
toolDefinitions []runtimetooling.MCPToolDefinition
skillsByID map[string]models.SkillDefinition
order []string
}
func newDatabaseSkillBackend(aiAgent models.AIAgent, toolDefinitions []runtimetooling.MCPToolDefinition) (*databaseSkillBackend, error) {
visibleSkills := loadVisibleSkills(aiAgent)
if len(visibleSkills) == 0 {
return nil, fmt.Errorf("no visible skills available")
}
ret := &databaseSkillBackend{
toolDefinitions: append([]runtimetooling.MCPToolDefinition(nil), toolDefinitions...),
skillsByID: make(map[string]models.SkillDefinition, len(visibleSkills)),
order: make([]string, 0, len(visibleSkills)),
}
for _, item := range visibleSkills {
id := strconv.FormatInt(item.ID, 10)
if id == "" {
continue
}
ret.skillsByID[id] = item
ret.order = append(ret.order, id)
}
if len(ret.skillsByID) == 0 {
return nil, fmt.Errorf("no visible skills available")
}
return ret, nil
}
func (b *databaseSkillBackend) List(_ context.Context) ([]einoskill.FrontMatter, error) {
if b == nil || len(b.order) == 0 {
return nil, nil
}
ret := make([]einoskill.FrontMatter, 0, len(b.order))
for _, id := range b.order {
item, ok := b.skillsByID[id]
if !ok {
continue
}
ret = append(ret, einoskill.FrontMatter{
Name: strconv.FormatInt(item.ID, 10),
Description: skillListDescription(item),
})
}
return ret, nil
}
func (b *databaseSkillBackend) Get(_ context.Context, name string) (einoskill.Skill, error) {
if b == nil {
return einoskill.Skill{}, fmt.Errorf("database skill backend is nil")
}
name = strings.TrimSpace(name)
if name == "" {
return einoskill.Skill{}, fmt.Errorf("skill name is empty")
}
item, ok := b.skillsByID[name]
if !ok {
return einoskill.Skill{}, fmt.Errorf("skill %q not found", name)
}
return einoskill.Skill{
FrontMatter: einoskill.FrontMatter{
Name: strconv.FormatInt(item.ID, 10),
Description: skillListDescription(item),
},
Content: runtimeinstruction.BuildSkillDocument(&item, filterSkillToolDefinitions(b.toolDefinitions, &item)),
BaseDirectory: "",
}, nil
}
func loadVisibleSkills(aiAgent models.AIAgent) []models.SkillDefinition {
ids := utils.SplitInt64s(strings.TrimSpace(aiAgent.SkillIDs))
if len(ids) == 0 {
return nil
}
byID := services.SkillDefinitionService.GetByIDs(ids)
if len(byID) == 0 {
return nil
}
ret := make([]models.SkillDefinition, 0, len(ids))
for _, id := range ids {
item, ok := byID[id]
if !ok || item.Status != enums.StatusOk || item.ID <= 0 {
continue
}
ret = append(ret, item)
}
return ret
}
func buildRuntimeSkillMetadataMap(aiAgent models.AIAgent) map[string]runtimeSkillMetadata {
visibleSkills := loadVisibleSkills(aiAgent)
if len(visibleSkills) == 0 {
return nil
}
ret := make(map[string]runtimeSkillMetadata, len(visibleSkills))
for _, item := range visibleSkills {
if item.ID <= 0 {
continue
}
id := strconv.FormatInt(item.ID, 10)
ret[id] = runtimeSkillMetadata{
ID: item.ID,
Name: strings.TrimSpace(item.Name),
Description: skillListDescription(item),
AllowedToolCodes: parseSkillToolWhitelist(item.ToolWhitelist),
}
}
return ret
}
func HasVisibleSkills(aiAgent models.AIAgent) bool {
return len(buildRuntimeSkillMetadataMap(aiAgent)) > 0
}
func skillListDescription(item models.SkillDefinition) string {
if desc := strings.TrimSpace(item.Description); desc != "" {
return desc
}
if name := strings.TrimSpace(item.Name); name != "" {
return name
}
return fmt.Sprintf("Skill %d", item.ID)
}
func parseSkillToolWhitelist(raw string) []string {
raw = strings.TrimSpace(raw)
if raw == "" {
return nil
}
var items []string
if err := json.Unmarshal([]byte(raw), &items); err != nil {
return nil
}
ret := make([]string, 0, len(items))
for _, item := range items {
item = strings.TrimSpace(item)
if item == "" {
continue
}
ret = append(ret, item)
}
return ret
}
func filterSkillToolDefinitions(defs []runtimetooling.MCPToolDefinition, skill *models.SkillDefinition) []runtimetooling.MCPToolDefinition {
allowed := parseSkillToolWhitelist(skill.ToolWhitelist)
if len(allowed) == 0 {
return defs
}
allowedSet := make(map[string]struct{}, len(allowed))
for _, item := range allowed {
allowedSet[item] = struct{}{}
}
ret := make([]runtimetooling.MCPToolDefinition, 0, len(defs))
for _, item := range defs {
if _, ok := allowedSet[strings.TrimSpace(item.ToolCode)]; ok {
ret = append(ret, item)
}
}
return ret
}
@@ -1,124 +0,0 @@
package factory
import (
"context"
"strings"
"testing"
"time"
runtimetooling "agent-desk/internal/ai/runtime/tooling"
"agent-desk/internal/models"
"agent-desk/internal/pkg/enums"
"github.com/glebarez/sqlite"
"github.com/mlogclub/simple/sqls"
"gorm.io/gorm"
)
func TestDatabaseSkillBackendListAndGet(t *testing.T) {
setupSkillBackendTestDB(t)
createSkillDefinitionForTest(t, models.SkillDefinition{
ID: 1,
Name: "售后升级",
Description: "处理转人工和升级诉求",
Instruction: "请优先判断是否需要转人工。",
ToolWhitelist: `["graph/handoff_to_human"]`,
Status: enums.StatusOk,
})
createSkillDefinitionForTest(t, models.SkillDefinition{
ID: 2,
Name: "禁用技能",
Description: "不会被暴露",
Instruction: "noop",
Status: enums.StatusDeleted,
})
backend, err := newDatabaseSkillBackend(models.AIAgent{SkillIDs: "1,2"}, []runtimetooling.MCPToolDefinition{
{ToolCode: "graph/handoff_to_human", Title: "转人工确认流程"},
{ToolCode: "graph/prepare_ticket_draft", Title: "整理工单草稿"},
})
if err != nil {
t.Fatalf("newDatabaseSkillBackend returned error: %v", err)
}
matters, err := backend.List(context.Background())
if err != nil {
t.Fatalf("List returned error: %v", err)
}
if len(matters) != 1 || matters[0].Name != "1" {
t.Fatalf("unexpected matters: %#v", matters)
}
skill, err := backend.Get(context.Background(), "1")
if err != nil {
t.Fatalf("Get returned error: %v", err)
}
if skill.Name != "1" {
t.Fatalf("unexpected skill name: %#v", skill)
}
if skill.Content == "" || !containsAll(skill.Content, "处理转人工和升级诉求", "graph/handoff_to_human") {
t.Fatalf("unexpected skill content: %q", skill.Content)
}
}
func TestHasVisibleSkills(t *testing.T) {
setupSkillBackendTestDB(t)
createSkillDefinitionForTest(t, models.SkillDefinition{
ID: 3,
Name: "启用技能",
Description: "可见",
Instruction: "noop",
Status: enums.StatusOk,
})
createSkillDefinitionForTest(t, models.SkillDefinition{
ID: 4,
Name: "删除技能",
Description: "不可见",
Instruction: "noop",
Status: enums.StatusDeleted,
})
if !HasVisibleSkills(models.AIAgent{SkillIDs: "3,4"}) {
t.Fatalf("expected visible skills")
}
if HasVisibleSkills(models.AIAgent{SkillIDs: "4"}) {
t.Fatalf("expected no visible skills")
}
}
func setupSkillBackendTestDB(t *testing.T) {
t.Helper()
db, err := gorm.Open(sqlite.Open("file:skill_backend_test?mode=memory&cache=shared"), &gorm.Config{})
if err != nil {
t.Fatalf("open sqlite failed: %v", err)
}
if err := db.AutoMigrate(&models.SkillDefinition{}); err != nil {
t.Fatalf("auto migrate skill definition failed: %v", err)
}
if err := db.Exec("DELETE FROM skill_definitions").Error; err != nil {
t.Fatalf("cleanup skill definitions failed: %v", err)
}
sqls.SetDB(db)
}
func createSkillDefinitionForTest(t *testing.T, item models.SkillDefinition) {
t.Helper()
now := time.Now()
if item.CreatedAt.IsZero() {
item.CreatedAt = now
}
if item.UpdatedAt.IsZero() {
item.UpdatedAt = now
}
if err := sqls.DB().Create(&item).Error; err != nil {
t.Fatalf("create skill definition failed: %v", err)
}
}
func containsAll(text string, items ...string) bool {
for _, item := range items {
if item != "" && !strings.Contains(text, item) {
return false
}
}
return true
}
@@ -1,35 +0,0 @@
package factory
import (
"context"
runtimetooling "agent-desk/internal/ai/runtime/tooling"
"agent-desk/internal/models"
"agent-desk/internal/pkg/toolx"
"github.com/cloudwego/eino/adk"
einoskill "github.com/cloudwego/eino/adk/middlewares/skill"
)
type SkillMiddlewareService struct{}
func NewSkillMiddlewareService() *SkillMiddlewareService {
return &SkillMiddlewareService{}
}
func (s *SkillMiddlewareService) Build(
ctx context.Context,
aiAgent models.AIAgent,
toolDefinitions []runtimetooling.MCPToolDefinition,
) (adk.ChatModelAgentMiddleware, error) {
backend, err := newDatabaseSkillBackend(aiAgent, toolDefinitions)
if err != nil {
return nil, err
}
toolName := toolx.BuiltinSkill.Name
return einoskill.NewMiddleware(ctx, &einoskill.Config{
Backend: backend,
SkillToolName: &toolName,
UseChinese: true,
})
}
@@ -1,109 +0,0 @@
package factory
import (
"context"
"strings"
"agent-desk/internal/ai/mcps"
impladapter "agent-desk/internal/ai/runtime/internal/impl/adapter"
runtimetooling "agent-desk/internal/ai/runtime/tooling"
"agent-desk/internal/models"
"agent-desk/internal/pkg/enums"
"agent-desk/internal/pkg/toolx"
einotool "github.com/cloudwego/eino/components/tool"
)
type ToolFactory struct{}
func NewToolFactory() *ToolFactory {
return &ToolFactory{}
}
func (f *ToolFactory) BuildMCPTools(aiAgent models.AIAgent) ([]runtimetooling.MCPToolDefinition, error) {
raw, err := toolx.ParseAgentMCPToolsJSON(aiAgent.AllowedMCPTools)
if err != nil {
return nil, err
}
ret := make([]runtimetooling.MCPToolDefinition, 0, len(raw))
for _, item := range raw {
toolCode := strings.TrimSpace(item.ToolCode)
toolCode = toolx.NormalizeToolCodeAlias(toolCode)
if toolCode == "" {
toolCode = toolx.BuildMCPToolCode(item.ServerCode, item.ToolName)
}
if toolx.ResolveToolSourceType(toolCode) != enums.ToolSourceTypeMCP {
continue
}
serverCode, toolName := toolx.SplitMCPToolCode(toolCode)
if serverCode == "" || toolName == "" {
continue
}
definition := runtimetooling.MCPToolDefinition{
ToolCode: toolCode,
ServerCode: serverCode,
ToolName: toolName,
Title: strings.TrimSpace(item.Title),
Description: strings.TrimSpace(item.Description),
FixedArgs: cloneStringMap(item.Arguments),
}
definition.ModelName = runtimetooling.BuildModelToolName(definition)
ret = append(ret, definition)
}
return ret, nil
}
func (f *ToolFactory) BuildBaseTools(ctx context.Context, aiAgent models.AIAgent) ([]einotool.BaseTool, error) {
definitions, err := f.BuildMCPTools(aiAgent)
if err != nil {
return nil, err
}
return f.BuildBaseToolsByDefinitions(ctx, definitions)
}
func (f *ToolFactory) BuildBaseToolsByDefinitions(ctx context.Context, definitions []runtimetooling.MCPToolDefinition) ([]einotool.BaseTool, error) {
if len(definitions) == 0 {
return nil, nil
}
metadataByCode, err := f.loadToolMetadata(ctx, definitions)
if err != nil {
return nil, err
}
ret := make([]einotool.BaseTool, 0, len(definitions))
for _, item := range definitions {
ret = append(ret, impladapter.NewMCPTool(item, metadataByCode[item.ToolCode]))
}
return ret, nil
}
func (f *ToolFactory) loadToolMetadata(ctx context.Context, definitions []runtimetooling.MCPToolDefinition) (map[string]*mcps.ToolInfo, error) {
toolsByCode := make(map[string]*mcps.ToolInfo, len(definitions))
serverCodes := make(map[string]struct{})
for _, item := range definitions {
serverCodes[item.ServerCode] = struct{}{}
}
for serverCode := range serverCodes {
toolInfos, err := mcps.Runtime.ListTools(ctx, serverCode)
if err != nil {
return nil, err
}
for i := range toolInfos {
toolInfo := toolInfos[i]
toolCode := toolx.BuildMCPToolCode(serverCode, toolInfo.Name)
toolInfoCopy := toolInfo
toolsByCode[toolCode] = &toolInfoCopy
}
}
return toolsByCode, nil
}
func cloneStringMap(input map[string]string) map[string]string {
if len(input) == 0 {
return nil
}
ret := make(map[string]string, len(input))
for key, value := range input {
ret[key] = value
}
return ret
}
@@ -1,31 +0,0 @@
package factory
import (
"testing"
"agent-desk/internal/models"
)
func TestBuildMCPToolsSkipsGraphAndBuiltinTools(t *testing.T) {
aiAgent := models.AIAgent{
AllowedMCPTools: `[
{"toolCode":"graph/create_ticket_with_confirmation","serverCode":"graph","toolName":"create_ticket_with_confirmation"},
{"toolCode":"builtin/tool_search","serverCode":"builtin","toolName":"tool_search"},
{"toolCode":"system/list_agents","serverCode":"system","toolName":"list_agents"}
]`,
}
got, err := NewToolFactory().BuildMCPTools(aiAgent)
if err != nil {
t.Fatalf("BuildMCPTools returned error: %v", err)
}
if len(got) != 1 {
t.Fatalf("expected 1 dynamic mcp tool, got %d: %#v", len(got), got)
}
if got[0].ToolCode != "system/list_agents" {
t.Fatalf("unexpected tool code: %#v", got[0])
}
if got[0].ServerCode != "system" || got[0].ToolName != "list_agents" {
t.Fatalf("unexpected tool identity: %#v", got[0])
}
}
@@ -1,402 +0,0 @@
package factory
import (
"context"
"encoding/json"
"fmt"
"strings"
einocallbacks "agent-desk/internal/ai/runtime/internal/impl/callbacks"
"agent-desk/internal/pkg/toolx"
"github.com/cloudwego/eino/adk"
"github.com/cloudwego/eino/components/model"
einotool "github.com/cloudwego/eino/components/tool"
"github.com/cloudwego/eino/schema"
)
const activeSkillRunLocalKey = "runtime_active_skill_id"
type RuntimeToolFilterMiddleware struct {
*adk.BaseChatModelAgentMiddleware
collector *einocallbacks.RuntimeTraceCollector
toolMetadataByName map[string]einocallbacks.ToolMetadata
skillMetadataBy map[string]einocallbacks.SkillMetadata
dynamicToolNames []string
}
func NewRuntimeToolFilterMiddleware(
collector *einocallbacks.RuntimeTraceCollector,
toolMetadataByName map[string]einocallbacks.ToolMetadata,
skillMetadataBy map[string]einocallbacks.SkillMetadata,
dynamicToolNames []string,
) *RuntimeToolFilterMiddleware {
return &RuntimeToolFilterMiddleware{
BaseChatModelAgentMiddleware: &adk.BaseChatModelAgentMiddleware{},
collector: collector,
toolMetadataByName: cloneToolMetadataMap(toolMetadataByName),
skillMetadataBy: cloneSkillMetadataMap(skillMetadataBy),
dynamicToolNames: append([]string(nil), dynamicToolNames...),
}
}
func (m *RuntimeToolFilterMiddleware) WrapModel(_ context.Context, cm model.BaseChatModel, mc *adk.ModelContext) (model.BaseChatModel, error) {
if mc == nil {
return cm, nil
}
return &runtimeToolFilterModelWrapper{
cm: cm,
allTools: append([]*schema.ToolInfo(nil), mc.Tools...),
collector: m.collector,
toolMetadataByName: m.toolMetadataByName,
skillMetadataBy: m.skillMetadataBy,
dynamicToolNames: append([]string(nil), m.dynamicToolNames...),
}, nil
}
func (m *RuntimeToolFilterMiddleware) WrapInvokableToolCall(_ context.Context, endpoint adk.InvokableToolCallEndpoint, tCtx *adk.ToolContext) (adk.InvokableToolCallEndpoint, error) {
return func(ctx context.Context, argumentsInJSON string, opts ...einotool.Option) (string, error) {
toolName := ""
if tCtx != nil {
toolName = strings.TrimSpace(tCtx.Name)
}
metadata, _ := resolveRuntimeToolMetadata(toolName, m.toolMetadataByName)
if !isRuntimeBuiltinAlwaysAllowed(metadata.ToolCode) {
activeSkill, restricted := m.resolveActiveSkill(ctx)
if restricted && !isToolCodeAllowedForSkill(metadata.ToolCode, activeSkill.AllowedToolCodes) {
return "", m.blockToolCall(metadata, argumentsInJSON, activeSkill)
}
}
result, err := endpoint(ctx, argumentsInJSON, opts...)
if err != nil {
return result, err
}
if strings.TrimSpace(metadata.ToolCode) == toolx.BuiltinSkill.Code {
_ = m.setActiveSkill(ctx, skillIDFromArguments(argumentsInJSON))
return result, nil
}
if strings.TrimSpace(metadata.ToolCode) == toolx.BuiltinToolSearch.Code {
activeSkill, restricted := m.resolveActiveSkill(ctx)
if !restricted {
return result, nil
}
filtered, filterErr := filterToolSearchResult(result, activeSkill.AllowedToolCodes, m.toolMetadataByName)
if filterErr == nil {
return filtered, nil
}
}
return result, nil
}, nil
}
func (m *RuntimeToolFilterMiddleware) blockToolCall(metadata einocallbacks.ToolMetadata, argumentsInJSON string, activeSkill einocallbacks.SkillMetadata) error {
err := fmt.Errorf("tool %s is not allowed for active skill %d", strings.TrimSpace(metadata.ToolCode), activeSkill.ID)
if m.collector != nil {
m.collector.AddToolItem(einocallbacks.ToolTraceItem{
ToolCode: strings.TrimSpace(metadata.ToolCode),
ServerCode: strings.TrimSpace(metadata.ServerCode),
ToolName: strings.TrimSpace(metadata.ToolName),
Arguments: parseRuntimeToolArguments(argumentsInJSON),
Status: "error",
ErrorMessage: err.Error(),
Blocked: true,
BlockedReason: "skill_tool_not_allowed",
})
}
return err
}
func (m *RuntimeToolFilterMiddleware) setActiveSkill(ctx context.Context, skillID string) error {
skillID = strings.TrimSpace(skillID)
if skillID == "" {
return nil
}
return adk.SetRunLocalValue(ctx, activeSkillRunLocalKey, skillID)
}
func (m *RuntimeToolFilterMiddleware) resolveActiveSkill(ctx context.Context) (einocallbacks.SkillMetadata, bool) {
if len(m.skillMetadataBy) == 0 {
return einocallbacks.SkillMetadata{}, false
}
value, found, err := adk.GetRunLocalValue(ctx, activeSkillRunLocalKey)
if err != nil || !found {
return einocallbacks.SkillMetadata{}, false
}
skillID, ok := value.(string)
if !ok {
return einocallbacks.SkillMetadata{}, false
}
skillID = strings.TrimSpace(skillID)
if skillID == "" {
return einocallbacks.SkillMetadata{}, false
}
skill, ok := m.skillMetadataBy[skillID]
if !ok {
return einocallbacks.SkillMetadata{}, false
}
if len(skill.AllowedToolCodes) == 0 {
return skill, false
}
return skill, true
}
type runtimeToolFilterModelWrapper struct {
cm model.BaseChatModel
allTools []*schema.ToolInfo
collector *einocallbacks.RuntimeTraceCollector
toolMetadataByName map[string]einocallbacks.ToolMetadata
skillMetadataBy map[string]einocallbacks.SkillMetadata
dynamicToolNames []string
}
func (w *runtimeToolFilterModelWrapper) Generate(ctx context.Context, input []*schema.Message, opts ...model.Option) (*schema.Message, error) {
tools := w.filteredTools(ctx, input)
return w.cm.Generate(ctx, input, append(opts, model.WithTools(tools))...)
}
func (w *runtimeToolFilterModelWrapper) Stream(ctx context.Context, input []*schema.Message, opts ...model.Option) (*schema.StreamReader[*schema.Message], error) {
tools := w.filteredTools(ctx, input)
return w.cm.Stream(ctx, input, append(opts, model.WithTools(tools))...)
}
func (w *runtimeToolFilterModelWrapper) filteredTools(ctx context.Context, input []*schema.Message) []*schema.ToolInfo {
tools := filterDynamicToolInfos(w.allTools, w.dynamicToolNames, input)
activeSkill, restricted := resolveActiveSkillMetadata(ctx, w.skillMetadataBy)
if restricted {
tools = filterToolInfosBySkill(tools, w.toolMetadataByName, activeSkill.AllowedToolCodes)
}
if w.collector != nil {
w.collector.SetFilteredToolCodes(extractToolCodesFromInfos(tools, w.toolMetadataByName))
}
return tools
}
func resolveActiveSkillMetadata(ctx context.Context, skills map[string]einocallbacks.SkillMetadata) (einocallbacks.SkillMetadata, bool) {
if len(skills) == 0 {
return einocallbacks.SkillMetadata{}, false
}
value, found, err := adk.GetRunLocalValue(ctx, activeSkillRunLocalKey)
if err != nil || !found {
return einocallbacks.SkillMetadata{}, false
}
skillID, ok := value.(string)
if !ok {
return einocallbacks.SkillMetadata{}, false
}
skillID = strings.TrimSpace(skillID)
skill, ok := skills[skillID]
if !ok || len(skill.AllowedToolCodes) == 0 {
return skill, false
}
return skill, true
}
func filterDynamicToolInfos(allTools []*schema.ToolInfo, dynamicToolNames []string, messages []*schema.Message) []*schema.ToolInfo {
if len(allTools) == 0 {
return nil
}
selectedToolNames := extractSelectedDynamicToolNames(messages)
if len(dynamicToolNames) == 0 {
return append([]*schema.ToolInfo(nil), allTools...)
}
removeMap := invertStringSelection(dynamicToolNames, selectedToolNames)
ret := make([]*schema.ToolInfo, 0, len(allTools))
for _, info := range allTools {
if info == nil {
continue
}
if _, ok := removeMap[strings.TrimSpace(info.Name)]; ok {
continue
}
ret = append(ret, info)
}
return ret
}
func extractSelectedDynamicToolNames(messages []*schema.Message) []string {
if len(messages) == 0 {
return nil
}
selected := make([]string, 0)
for _, message := range messages {
if message == nil || message.Role != schema.Tool || strings.TrimSpace(message.ToolName) != toolx.BuiltinToolSearch.Name {
continue
}
var payload struct {
SelectedTools []string `json:"selectedTools"`
}
if err := json.Unmarshal([]byte(strings.TrimSpace(message.Content)), &payload); err != nil {
continue
}
for _, item := range payload.SelectedTools {
item = strings.TrimSpace(item)
if item == "" {
continue
}
selected = append(selected, item)
}
}
return selected
}
func invertStringSelection(all []string, selected []string) map[string]struct{} {
selectedSet := make(map[string]struct{}, len(selected))
for _, item := range selected {
item = strings.TrimSpace(item)
if item == "" {
continue
}
selectedSet[item] = struct{}{}
}
ret := make(map[string]struct{})
for _, item := range all {
item = strings.TrimSpace(item)
if item == "" {
continue
}
if _, ok := selectedSet[item]; ok {
continue
}
ret[item] = struct{}{}
}
return ret
}
func filterToolInfosBySkill(allTools []*schema.ToolInfo, toolMetadataByName map[string]einocallbacks.ToolMetadata, allowedToolCodes []string) []*schema.ToolInfo {
if len(allTools) == 0 || len(allowedToolCodes) == 0 {
return append([]*schema.ToolInfo(nil), allTools...)
}
ret := make([]*schema.ToolInfo, 0, len(allTools))
for _, info := range allTools {
if info == nil {
continue
}
metadata, _ := resolveRuntimeToolMetadata(strings.TrimSpace(info.Name), toolMetadataByName)
if isRuntimeBuiltinAlwaysAllowed(metadata.ToolCode) || isToolCodeAllowedForSkill(metadata.ToolCode, allowedToolCodes) {
ret = append(ret, info)
}
}
return ret
}
func extractToolCodesFromInfos(infos []*schema.ToolInfo, toolMetadataByName map[string]einocallbacks.ToolMetadata) []string {
ret := make([]string, 0, len(infos))
for _, info := range infos {
if info == nil {
continue
}
metadata, ok := resolveRuntimeToolMetadata(strings.TrimSpace(info.Name), toolMetadataByName)
if !ok || strings.TrimSpace(metadata.ToolCode) == "" {
continue
}
ret = append(ret, metadata.ToolCode)
}
return toolx.NormalizeToolCodes(ret)
}
func isToolCodeAllowedForSkill(toolCode string, allowedToolCodes []string) bool {
toolCode = toolx.NormalizeToolCodeAlias(strings.TrimSpace(toolCode))
if toolCode == "" {
return true
}
for _, item := range toolx.NormalizeToolCodes(allowedToolCodes) {
if item == toolCode {
return true
}
}
return false
}
func isRuntimeBuiltinAlwaysAllowed(toolCode string) bool {
toolCode = toolx.NormalizeToolCodeAlias(strings.TrimSpace(toolCode))
return toolCode == toolx.BuiltinSkill.Code || toolCode == toolx.BuiltinToolSearch.Code
}
func resolveRuntimeToolMetadata(toolName string, toolMetadataByName map[string]einocallbacks.ToolMetadata) (einocallbacks.ToolMetadata, bool) {
toolName = strings.TrimSpace(toolName)
if toolName == "" {
return einocallbacks.ToolMetadata{}, false
}
if spec, ok := toolx.GetRegisteredToolSpecByName(toolName); ok {
resolved := toolx.ResolveToolMetadata(spec.Code, spec.Name)
return einocallbacks.ToolMetadata{
ToolCode: resolved.ToolCode,
ServerCode: resolved.ServerCode,
ToolName: resolved.ToolName,
SourceType: resolved.SourceType,
}, true
}
metadata, ok := toolMetadataByName[toolName]
return metadata, ok
}
func skillIDFromArguments(argumentsInJSON string) string {
var args struct {
Skill string `json:"skill"`
}
if err := json.Unmarshal([]byte(strings.TrimSpace(argumentsInJSON)), &args); err != nil {
return ""
}
return strings.TrimSpace(args.Skill)
}
func parseRuntimeToolArguments(argumentsInJSON string) map[string]any {
argumentsInJSON = strings.TrimSpace(argumentsInJSON)
if argumentsInJSON == "" {
return nil
}
ret := make(map[string]any)
if err := json.Unmarshal([]byte(argumentsInJSON), &ret); err != nil {
return nil
}
return ret
}
func filterToolSearchResult(result string, allowedToolCodes []string, toolMetadataByName map[string]einocallbacks.ToolMetadata) (string, error) {
result = strings.TrimSpace(result)
if result == "" {
return result, nil
}
var payload struct {
SelectedTools []string `json:"selectedTools"`
}
if err := json.Unmarshal([]byte(result), &payload); err != nil {
return result, err
}
filtered := make([]string, 0, len(payload.SelectedTools))
for _, toolName := range payload.SelectedTools {
metadata, _ := resolveRuntimeToolMetadata(toolName, toolMetadataByName)
if isToolCodeAllowedForSkill(metadata.ToolCode, allowedToolCodes) {
filtered = append(filtered, strings.TrimSpace(toolName))
}
}
payload.SelectedTools = filtered
buf, err := json.Marshal(payload)
if err != nil {
return "", err
}
return string(buf), nil
}
func cloneToolMetadataMap(input map[string]einocallbacks.ToolMetadata) map[string]einocallbacks.ToolMetadata {
if len(input) == 0 {
return nil
}
ret := make(map[string]einocallbacks.ToolMetadata, len(input))
for key, value := range input {
ret[key] = value
}
return ret
}
func cloneSkillMetadataMap(input map[string]einocallbacks.SkillMetadata) map[string]einocallbacks.SkillMetadata {
if len(input) == 0 {
return nil
}
ret := make(map[string]einocallbacks.SkillMetadata, len(input))
for key, value := range input {
value.AllowedToolCodes = append([]string(nil), value.AllowedToolCodes...)
ret[key] = value
}
return ret
}
@@ -1,84 +0,0 @@
package factory
import (
"testing"
einocallbacks "agent-desk/internal/ai/runtime/internal/impl/callbacks"
"agent-desk/internal/pkg/toolx"
"github.com/cloudwego/eino/schema"
)
func TestFilterDynamicToolInfos(t *testing.T) {
allTools := []*schema.ToolInfo{
{Name: toolx.BuiltinToolSearch.Name},
{Name: "mcp_server_a"},
{Name: "mcp_server_b"},
}
messages := []*schema.Message{
{Role: schema.Tool, ToolName: toolx.BuiltinToolSearch.Name, Content: `{"selectedTools":["mcp_server_b"]}`},
}
filtered := filterDynamicToolInfos(allTools, []string{"mcp_server_a", "mcp_server_b"}, messages)
if len(filtered) != 2 {
t.Fatalf("unexpected filtered tool count: %d", len(filtered))
}
if filtered[0].Name != toolx.BuiltinToolSearch.Name || filtered[1].Name != "mcp_server_b" {
t.Fatalf("unexpected filtered tools: %#v", filtered)
}
}
func TestFilterToolInfosBySkill(t *testing.T) {
allTools := []*schema.ToolInfo{
{Name: toolx.BuiltinSkill.Name},
{Name: toolx.BuiltinToolSearch.Name},
{Name: toolx.GraphHandoffConversation.Name},
{Name: "mcp_server_refund"},
}
toolMetadataByName := map[string]einocallbacks.ToolMetadata{
toolx.GraphHandoffConversation.Name: {ToolCode: toolx.GraphHandoffConversation.Code, ToolName: toolx.GraphHandoffConversation.Name},
"mcp_server_refund": {ToolCode: "mcp/refund", ToolName: "mcp_server_refund"},
}
filtered := filterToolInfosBySkill(allTools, toolMetadataByName, []string{toolx.GraphHandoffConversation.Code})
if len(filtered) != 3 {
t.Fatalf("unexpected filtered tool count: %d", len(filtered))
}
if filtered[0].Name != toolx.BuiltinSkill.Name || filtered[1].Name != toolx.BuiltinToolSearch.Name || filtered[2].Name != toolx.GraphHandoffConversation.Name {
t.Fatalf("unexpected filtered tools: %#v", filtered)
}
}
func TestFilterToolSearchResult(t *testing.T) {
toolMetadataByName := map[string]einocallbacks.ToolMetadata{
"mcp_server_refund": {ToolCode: "mcp/refund", ToolName: "mcp_server_refund"},
"mcp_server_order": {ToolCode: "mcp/order", ToolName: "mcp_server_order"},
}
got, err := filterToolSearchResult(`{"selectedTools":["mcp_server_refund","mcp_server_order"]}`, []string{"mcp/order"}, toolMetadataByName)
if err != nil {
t.Fatalf("filterToolSearchResult returned error: %v", err)
}
if got != `{"selectedTools":["mcp_server_order"]}` {
t.Fatalf("unexpected filtered result: %s", got)
}
}
func TestExtractToolCodesFromInfos(t *testing.T) {
infos := []*schema.ToolInfo{
{Name: toolx.BuiltinSkill.Name},
{Name: toolx.GraphPrepareTicketDraft.Name},
{Name: "mcp_server_refund"},
}
toolMetadataByName := map[string]einocallbacks.ToolMetadata{
toolx.GraphPrepareTicketDraft.Name: {ToolCode: toolx.GraphPrepareTicketDraft.Code, ToolName: toolx.GraphPrepareTicketDraft.Name},
"mcp_server_refund": {ToolCode: "mcp/refund", ToolName: "mcp_server_refund"},
}
got := extractToolCodesFromInfos(infos, toolMetadataByName)
if len(got) != 3 {
t.Fatalf("unexpected tool codes: %#v", got)
}
if got[0] != toolx.BuiltinSkill.Code || got[1] != toolx.GraphPrepareTicketDraft.Code || got[2] != "mcp/refund" {
t.Fatalf("unexpected tool codes order: %#v", got)
}
}
@@ -1,66 +0,0 @@
package factory
import (
"strings"
runtimeinstruction "agent-desk/internal/ai/runtime/instruction"
einocallbacks "agent-desk/internal/ai/runtime/internal/impl/callbacks"
"agent-desk/internal/ai/runtime/registry"
runtimetooling "agent-desk/internal/ai/runtime/tooling"
"agent-desk/internal/pkg/enums"
"agent-desk/internal/pkg/toolx"
)
func buildInstructionTraceSummary(summary runtimeinstruction.AssemblySummary) einocallbacks.InstructionTraceSummary {
return einocallbacks.InstructionTraceSummary{
SectionTitles: append([]string(nil), summary.SectionTitles...),
HasAgentRule: summary.HasAgentRule,
HasSkillRule: summary.HasSkillRule,
HasToolRule: summary.HasToolRule,
}
}
func buildRuntimeTraceToolMetadata(
dynamicToolDefinitions []runtimetooling.MCPToolDefinition,
staticToolMetadata map[string]registry.ToolMetadata,
includeSkillTool bool,
) map[string]einocallbacks.ToolMetadata {
ret := make(map[string]einocallbacks.ToolMetadata, len(dynamicToolDefinitions)+len(staticToolMetadata)+1)
for _, item := range dynamicToolDefinitions {
modelName := strings.TrimSpace(item.ModelName)
if modelName == "" {
continue
}
ret[modelName] = einocallbacks.ToolMetadata{
ToolCode: strings.TrimSpace(item.ToolCode),
ServerCode: strings.TrimSpace(item.ServerCode),
ToolName: strings.TrimSpace(item.ToolName),
SourceType: enums.ToolSourceTypeMCP,
}
}
for modelName, metadata := range staticToolMetadata {
modelName = strings.TrimSpace(modelName)
metadata.ToolCode = strings.TrimSpace(metadata.ToolCode)
metadata.ServerCode = strings.TrimSpace(metadata.ServerCode)
metadata.ToolName = strings.TrimSpace(metadata.ToolName)
if modelName == "" || metadata.ToolCode == "" {
continue
}
ret[modelName] = einocallbacks.ToolMetadata{
ToolCode: metadata.ToolCode,
ServerCode: metadata.ServerCode,
ToolName: metadata.ToolName,
SourceType: metadata.SourceType,
}
}
if includeSkillTool {
resolved := toolx.ResolveToolMetadata(toolx.BuiltinSkill.Code, toolx.BuiltinSkill.Name)
ret[toolx.BuiltinSkill.Name] = einocallbacks.ToolMetadata{
ToolCode: resolved.ToolCode,
ServerCode: resolved.ServerCode,
ToolName: resolved.ToolName,
SourceType: resolved.SourceType,
}
}
return ret
}
@@ -1,26 +0,0 @@
package factory
import (
"testing"
runtimeinstruction "agent-desk/internal/ai/runtime/instruction"
)
func TestBuildInstructionTraceSummary(t *testing.T) {
got := buildInstructionTraceSummary(runtimeinstruction.AssemblySummary{
SectionTitles: []string{"Agent 规则", "当前技能上下文"},
HasAgentRule: true,
HasSkillRule: true,
HasToolRule: false,
})
if len(got.SectionTitles) != 2 {
t.Fatalf("unexpected section titles: %#v", got.SectionTitles)
}
if !got.HasAgentRule || !got.HasSkillRule {
t.Fatalf("unexpected summary flags: %#v", got)
}
if got.HasToolRule {
t.Fatalf("expected HasToolRule false, got %#v", got)
}
}
@@ -1,69 +0,0 @@
package store
import (
"context"
"encoding/base64"
"time"
"agent-desk/internal/models"
"agent-desk/internal/repositories"
"github.com/cloudwego/eino/adk"
"github.com/mlogclub/simple/sqls"
)
var DefaultCheckPointStore adk.CheckPointStore = NewDBCheckPointStore()
type DBCheckPointStore struct{}
func NewDBCheckPointStore() *DBCheckPointStore {
return &DBCheckPointStore{}
}
func (s *DBCheckPointStore) Get(_ context.Context, checkPointID string) ([]byte, bool, error) {
item := repositories.ConversationInterruptRepository.GetByCheckPointID(sqls.DB(), checkPointID)
if item == nil || item.CheckPointData == "" {
return nil, false, nil
}
return decodeCheckPointData(item.CheckPointData)
}
func (s *DBCheckPointStore) Set(_ context.Context, checkPointID string, checkPoint []byte) error {
item := repositories.ConversationInterruptRepository.GetByCheckPointID(sqls.DB(), checkPointID)
if item == nil {
item = buildEmptyInterrupt(checkPointID)
}
item.CheckPointData = encodeCheckPointData(checkPoint)
if item.ConversationID == 0 && item.AIAgentID == 0 && item.SourceMessageID == 0 && item.Status == "" {
return repositories.ConversationInterruptRepository.Create(sqls.DB(), item)
}
return repositories.ConversationInterruptRepository.UpsertByCheckPointID(sqls.DB(), item)
}
func encodeCheckPointData(data []byte) string {
if len(data) == 0 {
return ""
}
return base64.StdEncoding.EncodeToString(data)
}
func decodeCheckPointData(value string) ([]byte, bool, error) {
if value == "" {
return nil, false, nil
}
data, err := base64.StdEncoding.DecodeString(value)
if err != nil {
return nil, false, err
}
return data, true, nil
}
func buildEmptyInterrupt(checkPointID string) *models.ConversationInterrupt {
now := time.Now()
return &models.ConversationInterrupt{
CheckPointID: checkPointID,
Status: "checkpointed",
CreatedAt: now,
UpdatedAt: now,
}
}
@@ -5,7 +5,7 @@ import (
"strings"
"agent-desk/internal/ai/rag"
"agent-desk/internal/ai/runtime/internal/impl/callbacks"
"agent-desk/internal/ai/runtime/traces"
"agent-desk/internal/models"
"agent-desk/internal/pkg/enums"
"agent-desk/internal/pkg/utils"
@@ -47,8 +47,8 @@ type KnowledgeRetrieveResult struct {
TopScore float64
AnswerMode enums.KnowledgeAnswerMode
Trace *rag.RetrieveTrace
TraceItems []callbacks.RetrieverTraceItem
TraceSummary callbacks.RetrieverTraceSummary
TraceItems []traces.RetrieverTraceItem
TraceSummary traces.RetrieverTraceSummary
Policies []KnowledgeBaseRetrievePolicy
}
@@ -231,7 +231,7 @@ func loadRuntimeKnowledgeBases(ids []int64) map[int64]models.KnowledgeBase {
return ret
}
func buildRetrieverTraceItems(queryPreview string, results []rag.RetrieveResult, trace *rag.RetrieveTrace) []callbacks.RetrieverTraceItem {
func buildRetrieverTraceItems(queryPreview string, results []rag.RetrieveResult, trace *rag.RetrieveTrace) []traces.RetrieverTraceItem {
if len(results) == 0 {
return nil
}
@@ -239,9 +239,9 @@ func buildRetrieverTraceItems(queryPreview string, results []rag.RetrieveResult,
if trace != nil {
latencyMs = trace.EmbeddingMs + trace.VectorSearchMs + trace.HydrateMs
}
ret := make([]callbacks.RetrieverTraceItem, 0, len(results))
ret := make([]traces.RetrieverTraceItem, 0, len(results))
for _, item := range results {
ret = append(ret, callbacks.RetrieverTraceItem{
ret = append(ret, traces.RetrieverTraceItem{
Query: queryPreview,
KnowledgeBaseID: item.KnowledgeBaseID,
DocumentID: item.DocumentID,
@@ -253,8 +253,8 @@ func buildRetrieverTraceItems(queryPreview string, results []rag.RetrieveResult,
return ret
}
func buildRetrieverTraceSummary(opts KnowledgeRetrieveOptions, policies []KnowledgeBaseRetrievePolicy, contextResults []rag.RetrieveResult, results []rag.RetrieveResult, trace *rag.RetrieveTrace) callbacks.RetrieverTraceSummary {
ret := callbacks.RetrieverTraceSummary{
func buildRetrieverTraceSummary(opts KnowledgeRetrieveOptions, policies []KnowledgeBaseRetrievePolicy, contextResults []rag.RetrieveResult, results []rag.RetrieveResult, trace *rag.RetrieveTrace) traces.RetrieverTraceSummary {
ret := traces.RetrieverTraceSummary{
TopK: opts.TopK,
ScoreThreshold: opts.ScoreThreshold,
ContextMaxTokens: opts.ContextMaxTokens,
@@ -277,13 +277,13 @@ func buildRetrieverTraceSummary(opts KnowledgeRetrieveOptions, policies []Knowle
return ret
}
func buildRetrieverPolicyTraceItems(policies []KnowledgeBaseRetrievePolicy) []callbacks.RetrieverPolicyTraceItem {
func buildRetrieverPolicyTraceItems(policies []KnowledgeBaseRetrievePolicy) []traces.RetrieverPolicyTraceItem {
if len(policies) == 0 {
return nil
}
ret := make([]callbacks.RetrieverPolicyTraceItem, 0, len(policies))
ret := make([]traces.RetrieverPolicyTraceItem, 0, len(policies))
for _, item := range policies {
ret = append(ret, callbacks.RetrieverPolicyTraceItem{
ret = append(ret, traces.RetrieverPolicyTraceItem{
KnowledgeBaseID: item.KnowledgeBaseID,
TopK: item.TopK,
ScoreThreshold: item.ScoreThreshold,
@@ -1,4 +1,4 @@
package adapter
package tooling
import (
"encoding/json"
@@ -8,8 +8,8 @@ import (
"strings"
"agent-desk/internal/ai/mcps"
impladapter "agent-desk/internal/ai/runtime/internal/impl/adapter"
"agent-desk/internal/ai/runtime/registry"
"agent-desk/internal/ai/runtime/tooling"
"agent-desk/internal/pkg/i18nx"
"agent-desk/internal/pkg/toolx"
@@ -277,5 +277,5 @@ func cloneArguments(input map[string]any) map[string]any {
}
func buildToolCallResultSummary(result *mcps.ToolCallResult) string {
return impladapter.BuildReducedToolResultSummary(result)
return tooling.BuildReducedToolResultSummary(result)
}
@@ -1,4 +1,4 @@
package callbacks
package traces
type ToolTraceItem struct {
ToolCode string `json:"toolCode"`
+1 -1
View File
@@ -13,7 +13,7 @@ import (
"agent-desk/internal/ai"
"agent-desk/internal/ai/runtime/graphs"
"agent-desk/internal/ai/runtime/internal/impl/retrievers"
"agent-desk/internal/ai/runtime/retrievers"
"agent-desk/internal/ai/workflow/dsl"
workflowregistry "agent-desk/internal/ai/workflow/registry"
"agent-desk/internal/models"