feat: enhance agent loop tool handling and add tests for direct capability aliases
This commit is contained in:
@@ -69,7 +69,7 @@ func (e *AgentLoopEngine) Run(ctx context.Context, req RunInput) (*RunResult, er
|
|||||||
turn := e.prepareTurn(ctx, req, snapshot)
|
turn := e.prepareTurn(ctx, req, snapshot)
|
||||||
var toolCalls []svc.AgentLoopToolCallInput
|
var toolCalls []svc.AgentLoopToolCallInput
|
||||||
state := agentLoopExecutionState{}
|
state := agentLoopExecutionState{}
|
||||||
loopResult, loopErr := e.loop(ctx, req.AIConfig, turn.SystemPrompt, turn.UserPrompt, []ai.ToolDefinition{agentLoopToolSearchDefinition()}, req.AIAgent.MaxSteps,
|
loopResult, loopErr := e.loop(ctx, req.AIConfig, turn.SystemPrompt, turn.UserPrompt, agentLoopToolDefinitions(turn), req.AIAgent.MaxSteps,
|
||||||
e.toolSearchExecutor(req, turn, &state, &toolCalls))
|
e.toolSearchExecutor(req, turn, &state, &toolCalls))
|
||||||
if state.Interrupted != nil {
|
if state.Interrupted != nil {
|
||||||
result := state.Interrupted
|
result := state.Interrupted
|
||||||
@@ -485,6 +485,33 @@ func agentLoopToolSearchDefinition() ai.ToolDefinition {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// agentLoopToolDefinitions registers the capability codes as compatibility
|
||||||
|
// aliases in addition to tool_search. Some OpenAI-compatible providers invoke
|
||||||
|
// a capability code mentioned in the prompt directly instead of wrapping it in
|
||||||
|
// tool_search. Eino validates the function name before our executor runs, so
|
||||||
|
// those calls must be registered here and then routed through the same policy
|
||||||
|
// boundary below.
|
||||||
|
func agentLoopToolDefinitions(turn agentLoopTurn) []ai.ToolDefinition {
|
||||||
|
definitions := []ai.ToolDefinition{agentLoopToolSearchDefinition()}
|
||||||
|
seen := map[string]struct{}{"tool_search": {}}
|
||||||
|
for _, code := range turn.AllowedTools {
|
||||||
|
code = strings.TrimSpace(code)
|
||||||
|
if code == "" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if _, exists := seen[code]; exists {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
seen[code] = struct{}{}
|
||||||
|
definitions = append(definitions, ai.ToolDefinition{
|
||||||
|
Name: code,
|
||||||
|
Description: "Execute the configured capability " + code + " with its arguments.",
|
||||||
|
Parameters: map[string]any{"type": "object", "additionalProperties": true},
|
||||||
|
})
|
||||||
|
}
|
||||||
|
return definitions
|
||||||
|
}
|
||||||
|
|
||||||
func agentLoopSafeBuiltinCodes() []string {
|
func agentLoopSafeBuiltinCodes() []string {
|
||||||
return []string{
|
return []string{
|
||||||
toolx.BuiltinConversationContext.Code,
|
toolx.BuiltinConversationContext.Code,
|
||||||
@@ -538,14 +565,10 @@ func (e *agentLoopInterruptError) Error() string {
|
|||||||
func (e *AgentLoopEngine) toolSearchExecutor(runInput RunInput, turn agentLoopTurn, state *agentLoopExecutionState, records *[]svc.AgentLoopToolCallInput) ai.ToolCallExecutor {
|
func (e *AgentLoopEngine) toolSearchExecutor(runInput RunInput, turn agentLoopTurn, state *agentLoopExecutionState, records *[]svc.AgentLoopToolCallInput) ai.ToolCallExecutor {
|
||||||
return func(ctx context.Context, call ai.ToolCall) (string, error) {
|
return func(ctx context.Context, call ai.ToolCall) (string, error) {
|
||||||
startedAt := time.Now()
|
startedAt := time.Now()
|
||||||
if call.Name != "tool_search" {
|
toolCode, arguments, err := resolveAgentLoopToolCall(call)
|
||||||
return "", fmt.Errorf("unsupported agent loop tool: %s", call.Name)
|
if err != nil {
|
||||||
|
return "", err
|
||||||
}
|
}
|
||||||
var toolRequest agentLoopToolSearchRequest
|
|
||||||
if err := json.Unmarshal([]byte(call.Arguments), &toolRequest); err != nil {
|
|
||||||
return "", fmt.Errorf("invalid tool_search arguments: %w", err)
|
|
||||||
}
|
|
||||||
toolCode := strings.TrimSpace(toolRequest.ToolCode)
|
|
||||||
if !slices.Contains(turn.AllowedTools, toolCode) {
|
if !slices.Contains(turn.AllowedTools, toolCode) {
|
||||||
return "", fmt.Errorf("capability is not configured for this Agent: %s", toolCode)
|
return "", fmt.Errorf("capability is not configured for this Agent: %s", toolCode)
|
||||||
}
|
}
|
||||||
@@ -563,22 +586,22 @@ func (e *AgentLoopEngine) toolSearchExecutor(runInput RunInput, turn agentLoopTu
|
|||||||
}
|
}
|
||||||
definition := aitooling.Definition{Code: toolCode, RiskLevel: aitooling.RiskLevelRead}
|
definition := aitooling.Definition{Code: toolCode, RiskLevel: aitooling.RiskLevelRead}
|
||||||
var resultPreview string
|
var resultPreview string
|
||||||
var err error
|
var executeErr error
|
||||||
switch {
|
switch {
|
||||||
case strings.HasPrefix(toolCode, "skill/"):
|
case strings.HasPrefix(toolCode, "skill/"):
|
||||||
resultPreview, err = activateAgentLoopSkill(toolCode, turn.Skills, state)
|
resultPreview, executeErr = activateAgentLoopSkill(toolCode, turn.Skills, state)
|
||||||
case strings.HasPrefix(toolCode, "workflow/"):
|
case strings.HasPrefix(toolCode, "workflow/"):
|
||||||
definition.RiskLevel = aitooling.RiskLevelWrite
|
definition.RiskLevel = aitooling.RiskLevelWrite
|
||||||
definition.RequireConfirmation = true
|
definition.RequireConfirmation = true
|
||||||
workflowPolicy := policy
|
workflowPolicy := policy
|
||||||
workflowPolicy.Confirmed = true
|
workflowPolicy.Confirmed = true
|
||||||
if err = aitooling.DefaultRegistry.Authorize(definition, workflowPolicy); err == nil {
|
if executeErr = aitooling.DefaultRegistry.Authorize(definition, workflowPolicy); executeErr == nil {
|
||||||
resultPreview, err = executeAgentLoopWorkflow(ctx, runInput, toolCode, turn.Workflows, state)
|
resultPreview, executeErr = executeAgentLoopWorkflow(ctx, runInput, toolCode, turn.Workflows, state)
|
||||||
}
|
}
|
||||||
default:
|
default:
|
||||||
definition, resultPreview, err = executeAgentLoopReadTool(ctx, runInput.Conversation, runInput.AIAgent, toolCode, toolRequest.Arguments, policy)
|
definition, resultPreview, executeErr = executeAgentLoopReadTool(ctx, runInput.Conversation, runInput.AIAgent, toolCode, arguments, policy)
|
||||||
if err != nil && definition.Code == "" {
|
if executeErr != nil && definition.Code == "" {
|
||||||
definition, resultPreview, err = executeAgentLoopMCP(ctx, runInput, toolCode, toolRequest.Arguments, policy, state)
|
definition, resultPreview, executeErr = executeAgentLoopMCP(ctx, runInput, toolCode, arguments, policy, state)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
durationMS := int(time.Since(startedAt).Milliseconds())
|
durationMS := int(time.Since(startedAt).Milliseconds())
|
||||||
@@ -590,15 +613,15 @@ func (e *AgentLoopEngine) toolSearchExecutor(runInput RunInput, turn agentLoopTu
|
|||||||
record.RiskLevel = definition.RiskLevel
|
record.RiskLevel = definition.RiskLevel
|
||||||
record.RequireConfirm = definition.RequireConfirmation
|
record.RequireConfirm = definition.RequireConfirmation
|
||||||
}
|
}
|
||||||
if err != nil {
|
if executeErr != nil {
|
||||||
record.Status = "failed"
|
record.Status = "failed"
|
||||||
var interruptErr *agentLoopInterruptError
|
var interruptErr *agentLoopInterruptError
|
||||||
if errors.As(err, &interruptErr) {
|
if errors.As(executeErr, &interruptErr) {
|
||||||
record.Status = "interrupted"
|
record.Status = "interrupted"
|
||||||
}
|
}
|
||||||
record.ErrorMessage = err.Error()
|
record.ErrorMessage = executeErr.Error()
|
||||||
*records = append(*records, record)
|
*records = append(*records, record)
|
||||||
return "", err
|
return "", executeErr
|
||||||
}
|
}
|
||||||
record.ResultPreview = aitooling.SanitizePreview(resultPreview)
|
record.ResultPreview = aitooling.SanitizePreview(resultPreview)
|
||||||
*records = append(*records, record)
|
*records = append(*records, record)
|
||||||
@@ -606,6 +629,28 @@ func (e *AgentLoopEngine) toolSearchExecutor(runInput RunInput, turn agentLoopTu
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func resolveAgentLoopToolCall(call ai.ToolCall) (string, map[string]any, error) {
|
||||||
|
if call.Name == "tool_search" {
|
||||||
|
var request agentLoopToolSearchRequest
|
||||||
|
if err := json.Unmarshal([]byte(call.Arguments), &request); err != nil {
|
||||||
|
return "", nil, fmt.Errorf("invalid tool_search arguments: %w", err)
|
||||||
|
}
|
||||||
|
return strings.TrimSpace(request.ToolCode), request.Arguments, nil
|
||||||
|
}
|
||||||
|
toolCode := strings.TrimSpace(call.Name)
|
||||||
|
if toolCode == "" {
|
||||||
|
return "", nil, fmt.Errorf("agent loop tool name is required")
|
||||||
|
}
|
||||||
|
arguments := map[string]any{}
|
||||||
|
if strings.TrimSpace(call.Arguments) == "" {
|
||||||
|
return toolCode, arguments, nil
|
||||||
|
}
|
||||||
|
if err := json.Unmarshal([]byte(call.Arguments), &arguments); err != nil {
|
||||||
|
return "", nil, fmt.Errorf("invalid direct capability arguments for %s: %w", toolCode, err)
|
||||||
|
}
|
||||||
|
return toolCode, arguments, nil
|
||||||
|
}
|
||||||
|
|
||||||
func activateAgentLoopSkill(code string, skills map[int64]models.SkillDefinition, state *agentLoopExecutionState) (string, error) {
|
func activateAgentLoopSkill(code string, skills map[int64]models.SkillDefinition, state *agentLoopExecutionState) (string, error) {
|
||||||
id, err := strconv.ParseInt(strings.TrimPrefix(code, "skill/"), 10, 64)
|
id, err := strconv.ParseInt(strings.TrimPrefix(code, "skill/"), 10, 64)
|
||||||
if err != nil || id <= 0 {
|
if err != nil || id <= 0 {
|
||||||
|
|||||||
@@ -45,6 +45,55 @@ func TestAgentLoopActivatesSkillInsideSameToolLoop(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestAgentLoopRegistersDirectCapabilityAliases(t *testing.T) {
|
||||||
|
turn := agentLoopTurn{AllowedTools: []string{
|
||||||
|
"builtin/conversation_context",
|
||||||
|
"graph/triage_service_request",
|
||||||
|
"graph/triage_service_request",
|
||||||
|
"workflow/47",
|
||||||
|
}}
|
||||||
|
definitions := agentLoopToolDefinitions(turn)
|
||||||
|
names := make(map[string]bool, len(definitions))
|
||||||
|
for _, definition := range definitions {
|
||||||
|
names[definition.Name] = true
|
||||||
|
}
|
||||||
|
for _, expected := range []string{
|
||||||
|
"tool_search",
|
||||||
|
"builtin/conversation_context",
|
||||||
|
"graph/triage_service_request",
|
||||||
|
"workflow/47",
|
||||||
|
} {
|
||||||
|
if !names[expected] {
|
||||||
|
t.Fatalf("missing registered function alias %q: %#v", expected, definitions)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestAgentLoopDirectCapabilityAliasUsesSamePolicyBoundary(t *testing.T) {
|
||||||
|
skill := models.SkillDefinition{
|
||||||
|
ID: 7, Name: "售后升级处理", Instruction: "先确认升级诉求。", Status: enums.StatusOk,
|
||||||
|
}
|
||||||
|
turn := agentLoopTurn{
|
||||||
|
AllowedTools: []string{"skill/7"},
|
||||||
|
ToolPolicy: parseAgentLoopToolPolicy(""),
|
||||||
|
Skills: map[int64]models.SkillDefinition{skill.ID: skill},
|
||||||
|
}
|
||||||
|
state := agentLoopExecutionState{}
|
||||||
|
var calls []svc.AgentLoopToolCallInput
|
||||||
|
execute := NewAgentLoopEngine().toolSearchExecutor(RunInput{}, turn, &state, &calls)
|
||||||
|
|
||||||
|
result, err := execute(context.Background(), ai.ToolCall{Name: "skill/7", Arguments: `{}`})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("execute direct capability alias: %v", err)
|
||||||
|
}
|
||||||
|
if state.SkillContext.SkillID() != skill.ID || !strings.Contains(result, skill.Instruction) {
|
||||||
|
t.Fatalf("direct capability was not routed through Skill activation: state=%#v result=%q", state, result)
|
||||||
|
}
|
||||||
|
if len(calls) != 1 || calls[0].ToolCode != "skill/7" || calls[0].Status != "completed" {
|
||||||
|
t.Fatalf("unexpected direct capability audit: %#v", calls)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestAgentLoopInterruptsBeforeWriteMCPTool(t *testing.T) {
|
func TestAgentLoopInterruptsBeforeWriteMCPTool(t *testing.T) {
|
||||||
configured, err := json.Marshal([]request.AIAgentMCPToolRequest{{
|
configured, err := json.Marshal([]request.AIAgentMCPToolRequest{{
|
||||||
ToolCode: "crm/update_customer", ServerCode: "crm", ToolName: "update_customer",
|
ToolCode: "crm/update_customer", ServerCode: "crm", ToolName: "update_customer",
|
||||||
|
|||||||
@@ -36,6 +36,7 @@ func (e *AgentLoopEngine) prepareTurn(ctx context.Context, req RunInput, snapsho
|
|||||||
skills := svc.SkillDefinitionService.GetByIDs(utils.SplitInt64s(req.AIAgent.SkillIDs))
|
skills := svc.SkillDefinitionService.GetByIDs(utils.SplitInt64s(req.AIAgent.SkillIDs))
|
||||||
workflows := make(map[int64]svc.AgentRevisionWorkflowBinding, len(snapshot.WorkflowBindings))
|
workflows := make(map[int64]svc.AgentRevisionWorkflowBinding, len(snapshot.WorkflowBindings))
|
||||||
allowedTools := agentLoopSafeBuiltinCodes()
|
allowedTools := agentLoopSafeBuiltinCodes()
|
||||||
|
// TODO 这么实现我觉得不太好,最好是能够有个统一的能力目录
|
||||||
catalog := []string{
|
catalog := []string{
|
||||||
"- " + toolx.BuiltinConversationContext.Code + " | Builtin | 读取当前会话和客户上下文",
|
"- " + toolx.BuiltinConversationContext.Code + " | Builtin | 读取当前会话和客户上下文",
|
||||||
"- " + toolx.BuiltinKnowledgeRetrieve.Code + " | Builtin | 按需再次检索已绑定知识库",
|
"- " + toolx.BuiltinKnowledgeRetrieve.Code + " | Builtin | 按需再次检索已绑定知识库",
|
||||||
|
|||||||
@@ -42,7 +42,7 @@ func TestConversationHumanDispatchAIHandoffOffHoursKeepsAIServingAndSendsNotice(
|
|||||||
if message == nil {
|
if message == nil {
|
||||||
t.Fatalf("expected off-hours notice message")
|
t.Fatalf("expected off-hours notice message")
|
||||||
}
|
}
|
||||||
if message.SenderType != enums.IMSenderTypeAI || !strings.Contains(message.Content, "Human support is currently outside service hours") {
|
if message.SenderType != enums.IMSenderTypeAI || message.Content != services.HandoffOffHoursMessage {
|
||||||
t.Fatalf("unexpected off-hours message: %+v", message)
|
t.Fatalf("unexpected off-hours message: %+v", message)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user