feat: Enhance AI Agent and Channel Management
- Updated labels in the AI Agents dashboard for clarity, changing "流程状态" to "Playbook 状态" and "未发布流程" to "未发布 Playbook". - Introduced AI Agent rollout percentage management in channel editing, allowing users to set and rollback rollout percentages. - Added new API endpoints for rolling back AI Agent rollout and fetching agent run metrics. - Implemented new UI components for displaying agent run details, including status, duration, and input/output tokens. - Enhanced type definitions for AdminChannel and AIAgent to include rollout percentages and runtime modes. - Updated navigation to include a section for agent runs. - Added new translations for agent run features in both English and Chinese.
This commit is contained in:
mlogclub
@@ -0,0 +1,102 @@
|
||||
package einoexperiment
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
applicationruntime "agent-desk/internal/ai/application/runtime"
|
||||
"agent-desk/internal/ai/runtime/graphs"
|
||||
)
|
||||
|
||||
const confirmationInterruptType = "human_confirm"
|
||||
|
||||
// ConfirmationRequest represents a high-risk Eino tool action that must pause
|
||||
// at AgentDesk's existing conversation-interrupt boundary.
|
||||
type ConfirmationRequest struct {
|
||||
InterruptID string
|
||||
ToolCode string
|
||||
Prompt string
|
||||
Arguments map[string]any
|
||||
}
|
||||
|
||||
type confirmationCheckpoint struct {
|
||||
Version int `json:"version"`
|
||||
Engine string `json:"engine"`
|
||||
InterruptID string `json:"interruptId"`
|
||||
ToolCode string `json:"toolCode"`
|
||||
Arguments map[string]any `json:"arguments"`
|
||||
}
|
||||
|
||||
// BuildConfirmationResult returns the generic interrupted result consumed by
|
||||
// replyInterruptService. That service persists ConversationInterrupt from the
|
||||
// result, so this package remains independent of database writes.
|
||||
func BuildConfirmationResult(input applicationruntime.RunInput, request ConfirmationRequest) (*applicationruntime.RunResult, error) {
|
||||
interruptID := strings.TrimSpace(request.InterruptID)
|
||||
if interruptID == "" {
|
||||
interruptID = "eino_confirm"
|
||||
}
|
||||
toolCode := strings.TrimSpace(request.ToolCode)
|
||||
if toolCode == "" {
|
||||
return nil, fmt.Errorf("confirmation tool code is required")
|
||||
}
|
||||
prompt := strings.TrimSpace(request.Prompt)
|
||||
if prompt == "" {
|
||||
return nil, fmt.Errorf("confirmation prompt is required")
|
||||
}
|
||||
checkpointData, err := json.Marshal(confirmationCheckpoint{
|
||||
Version: 1, Engine: "eino", InterruptID: interruptID, ToolCode: toolCode, Arguments: cloneConfirmationArguments(request.Arguments),
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("encode Eino confirmation checkpoint: %w", err)
|
||||
}
|
||||
return &applicationruntime.RunResult{
|
||||
Status: "interrupted",
|
||||
Interrupted: true,
|
||||
CheckPointID: confirmationCheckpointID(input, interruptID, toolCode, checkpointData),
|
||||
CheckPointData: string(checkpointData),
|
||||
Interrupts: []applicationruntime.InterruptContextSummary{{
|
||||
Type: confirmationInterruptType, ID: interruptID, InfoPreview: string(mustMarshalConfirmationPrompt(prompt)),
|
||||
}},
|
||||
}, nil
|
||||
}
|
||||
|
||||
// ResumeConfirmation reads the generic ResumeInput populated by the existing
|
||||
// AgentApplicationService and validates that it belongs to this checkpoint.
|
||||
func ResumeConfirmation(checkPointData string, input applicationruntime.ResumeInput) (string, confirmationCheckpoint, error) {
|
||||
checkpoint := confirmationCheckpoint{}
|
||||
if err := json.Unmarshal([]byte(strings.TrimSpace(checkPointData)), &checkpoint); err != nil {
|
||||
return "", checkpoint, fmt.Errorf("decode Eino confirmation checkpoint: %w", err)
|
||||
}
|
||||
if checkpoint.Version != 1 || checkpoint.Engine != "eino" || strings.TrimSpace(checkpoint.InterruptID) == "" || strings.TrimSpace(checkpoint.ToolCode) == "" {
|
||||
return "", checkpoint, fmt.Errorf("invalid Eino confirmation checkpoint")
|
||||
}
|
||||
decision := graphs.ParseConfirmationDecision(strings.TrimSpace(input.ResumeData[checkpoint.InterruptID]))
|
||||
if decision == "" {
|
||||
return "", checkpoint, fmt.Errorf("Eino confirmation decision is required")
|
||||
}
|
||||
return string(decision), checkpoint, nil
|
||||
}
|
||||
|
||||
func confirmationCheckpointID(input applicationruntime.RunInput, interruptID, toolCode string, data []byte) string {
|
||||
digest := sha256.Sum256(append([]byte(strings.TrimSpace(toolCode)+":"+strings.TrimSpace(interruptID)+":"), data...))
|
||||
return fmt.Sprintf("eino:%d:%d:%s", input.Conversation.ID, input.UserMessage.ID, hex.EncodeToString(digest[:8]))
|
||||
}
|
||||
|
||||
func cloneConfirmationArguments(input map[string]any) map[string]any {
|
||||
if len(input) == 0 {
|
||||
return map[string]any{}
|
||||
}
|
||||
ret := make(map[string]any, len(input))
|
||||
for key, value := range input {
|
||||
ret[key] = value
|
||||
}
|
||||
return ret
|
||||
}
|
||||
|
||||
func mustMarshalConfirmationPrompt(prompt string) []byte {
|
||||
data, _ := json.Marshal(map[string]string{"message": prompt})
|
||||
return data
|
||||
}
|
||||
@@ -0,0 +1,94 @@
|
||||
package einoexperiment
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
aitooling "agent-desk/internal/ai/tooling"
|
||||
|
||||
einotool "github.com/cloudwego/eino/components/tool"
|
||||
"github.com/cloudwego/eino/schema"
|
||||
)
|
||||
|
||||
// ToolHandler is the adapter point from an approved Eino experiment tool to
|
||||
// AgentDesk business services. Production handlers must still call services,
|
||||
// never repositories.
|
||||
type ToolHandler func(ctx context.Context, arguments map[string]any) (string, error)
|
||||
|
||||
// ToolTrace is emitted for every guarded invocation. A future Engine adapter
|
||||
// can translate it into AgentRun tool-call audit records without coupling this
|
||||
// experiment package to the service layer.
|
||||
type ToolTrace struct {
|
||||
ToolCode string
|
||||
Arguments map[string]any
|
||||
Status string
|
||||
Result string
|
||||
Err error
|
||||
Duration time.Duration
|
||||
}
|
||||
|
||||
type ToolTraceHook func(ToolTrace)
|
||||
|
||||
// GuardedTool adapts an Eino InvokableTool to the shared ToolPolicyGuard. It is
|
||||
// deliberately generic so Tool Registry semantics are checked before a tool
|
||||
// handler is invoked.
|
||||
type GuardedTool struct {
|
||||
InfoDefinition *schema.ToolInfo
|
||||
Definition aitooling.Definition
|
||||
Policy aitooling.Policy
|
||||
Handler ToolHandler
|
||||
Trace ToolTraceHook
|
||||
}
|
||||
|
||||
var _ einotool.InvokableTool = (*GuardedTool)(nil)
|
||||
|
||||
func (t *GuardedTool) Info(context.Context) (*schema.ToolInfo, error) {
|
||||
if t == nil || t.InfoDefinition == nil {
|
||||
return nil, fmt.Errorf("eino experiment tool info is required")
|
||||
}
|
||||
return t.InfoDefinition, nil
|
||||
}
|
||||
|
||||
func (t *GuardedTool) InvokableRun(ctx context.Context, argumentsInJSON string, _ ...einotool.Option) (string, error) {
|
||||
if t == nil || t.Handler == nil {
|
||||
return "", fmt.Errorf("eino experiment tool handler is required")
|
||||
}
|
||||
startedAt := time.Now()
|
||||
arguments := map[string]any{}
|
||||
if err := json.Unmarshal([]byte(argumentsInJSON), &arguments); err != nil {
|
||||
t.emitTrace(arguments, "failed", "", err, startedAt)
|
||||
return "", fmt.Errorf("decode tool arguments: %w", err)
|
||||
}
|
||||
if err := aitooling.DefaultPolicyGuard.Authorize(aitooling.Invocation{
|
||||
Definition: t.Definition,
|
||||
Arguments: arguments,
|
||||
Policy: t.Policy,
|
||||
}); err != nil {
|
||||
t.emitTrace(arguments, "failed", "", err, startedAt)
|
||||
return "", err
|
||||
}
|
||||
if t.Definition.TimeoutMS > 0 {
|
||||
var cancel context.CancelFunc
|
||||
ctx, cancel = context.WithTimeout(ctx, time.Duration(t.Definition.TimeoutMS)*time.Millisecond)
|
||||
defer cancel()
|
||||
}
|
||||
result, err := t.Handler(ctx, arguments)
|
||||
status := "completed"
|
||||
if err != nil {
|
||||
status = "failed"
|
||||
}
|
||||
t.emitTrace(arguments, status, result, err, startedAt)
|
||||
return result, err
|
||||
}
|
||||
|
||||
func (t *GuardedTool) emitTrace(arguments map[string]any, status, result string, err error, startedAt time.Time) {
|
||||
if t == nil || t.Trace == nil {
|
||||
return
|
||||
}
|
||||
t.Trace(ToolTrace{
|
||||
ToolCode: t.Definition.Code, Arguments: arguments, Status: status, Result: result, Err: err,
|
||||
Duration: time.Since(startedAt),
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
package einoexperiment
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"agent-desk/internal/ai/mcps"
|
||||
runtimetooling "agent-desk/internal/ai/runtime/tooling"
|
||||
aitooling "agent-desk/internal/ai/tooling"
|
||||
)
|
||||
|
||||
// MCPToolExecutor is the narrow execution boundary used by the Eino
|
||||
// experiment. The production MCP executor remains responsible for dynamic
|
||||
// registry resolution, policy enforcement, timeout, and transport lifecycle.
|
||||
type MCPToolExecutor interface {
|
||||
Execute(context.Context, string, map[string]any, aitooling.Policy) (aitooling.Definition, *mcps.ToolCallResult, error)
|
||||
}
|
||||
|
||||
// NewMCPToolHandler adapts a dynamically discovered MCP tool to GuardedTool.
|
||||
// Callers must still configure GuardedTool.Definition and Policy so its
|
||||
// pre-handler guard provides a deterministic rejection before MCP transport.
|
||||
func NewMCPToolHandler(executor MCPToolExecutor, toolCode string, policy aitooling.Policy) ToolHandler {
|
||||
return func(ctx context.Context, arguments map[string]any) (string, error) {
|
||||
if executor == nil {
|
||||
return "", fmt.Errorf("eino experiment MCP executor is required")
|
||||
}
|
||||
definition, result, err := executor.Execute(ctx, strings.TrimSpace(toolCode), arguments, policy)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if definition.Code == "" {
|
||||
return "", fmt.Errorf("MCP executor returned an empty tool definition")
|
||||
}
|
||||
return runtimetooling.BuildReducedToolResultSummary(result), nil
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
// Package einoexperiment contains an isolated Eino ReAct verification path.
|
||||
// It must not be registered in the production Agent Engine registry.
|
||||
package einoexperiment
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"agent-desk/internal/models"
|
||||
|
||||
einoopenai "github.com/cloudwego/eino-ext/components/model/openai"
|
||||
"github.com/cloudwego/eino/components/model"
|
||||
"github.com/cloudwego/eino/components/tool"
|
||||
"github.com/cloudwego/eino/compose"
|
||||
"github.com/cloudwego/eino/flow/agent/react"
|
||||
"github.com/cloudwego/eino/schema"
|
||||
)
|
||||
|
||||
// ReActConfig keeps the experiment dependency-injected. The caller owns model
|
||||
// construction, connection reuse, and all production configuration decisions.
|
||||
type ReActConfig struct {
|
||||
Model model.ToolCallingChatModel
|
||||
Tools []tool.BaseTool
|
||||
MaxSteps int
|
||||
}
|
||||
|
||||
// NewOpenAICompatibleModel adapts an existing AgentDesk AI configuration to
|
||||
// Eino's OpenAI-compatible chat model. It is intentionally not wired into any
|
||||
// production Engine; the experiment owns the adoption decision.
|
||||
func NewOpenAICompatibleModel(ctx context.Context, config models.AIConfig) (model.ToolCallingChatModel, error) {
|
||||
if strings.TrimSpace(config.APIKey) == "" || strings.TrimSpace(config.BaseURL) == "" || strings.TrimSpace(config.ModelName) == "" {
|
||||
return nil, fmt.Errorf("ai config base URL, API key, and model name are required")
|
||||
}
|
||||
modelConfig := &einoopenai.ChatModelConfig{
|
||||
APIKey: strings.TrimSpace(config.APIKey),
|
||||
BaseURL: strings.TrimSpace(config.BaseURL),
|
||||
Model: strings.TrimSpace(config.ModelName),
|
||||
}
|
||||
if config.TimeoutMS > 0 {
|
||||
modelConfig.Timeout = time.Duration(config.TimeoutMS) * time.Millisecond
|
||||
}
|
||||
if config.MaxOutputTokens > 0 {
|
||||
maxTokens := config.MaxOutputTokens
|
||||
modelConfig.MaxCompletionTokens = &maxTokens
|
||||
}
|
||||
return einoopenai.NewChatModel(ctx, modelConfig)
|
||||
}
|
||||
|
||||
// NewReAct creates an Eino ReAct agent without registering it with AgentDesk's
|
||||
// runtime. It is deliberately suitable only for technical verification.
|
||||
func NewReAct(ctx context.Context, config ReActConfig) (*react.Agent, error) {
|
||||
if config.Model == nil {
|
||||
return nil, fmt.Errorf("eino experiment model is required")
|
||||
}
|
||||
maxSteps := config.MaxSteps
|
||||
if maxSteps <= 0 {
|
||||
maxSteps = 5
|
||||
}
|
||||
return react.NewAgent(ctx, &react.AgentConfig{
|
||||
ToolCallingModel: config.Model,
|
||||
ToolsConfig: compose.ToolsNodeConfig{Tools: config.Tools},
|
||||
MaxStep: maxSteps,
|
||||
})
|
||||
}
|
||||
|
||||
// Run performs one non-streaming experiment. Context cancellation is passed
|
||||
// directly to Eino and the injected model/tools.
|
||||
func Run(ctx context.Context, config ReActConfig, input []*schema.Message) (*schema.Message, error) {
|
||||
agent, err := NewReAct(ctx, config)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return agent.Generate(ctx, input)
|
||||
}
|
||||
|
||||
// Stream performs one streaming experiment. The caller must close the returned
|
||||
// reader after consuming it.
|
||||
func Stream(ctx context.Context, config ReActConfig, input []*schema.Message) (*schema.StreamReader[*schema.Message], error) {
|
||||
agent, err := NewReAct(ctx, config)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return agent.Stream(ctx, input)
|
||||
}
|
||||
@@ -0,0 +1,376 @@
|
||||
package einoexperiment
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
applicationruntime "agent-desk/internal/ai/application/runtime"
|
||||
"agent-desk/internal/ai/mcps"
|
||||
"agent-desk/internal/ai/runtime/graphs"
|
||||
aitooling "agent-desk/internal/ai/tooling"
|
||||
"agent-desk/internal/models"
|
||||
|
||||
"github.com/cloudwego/eino/components/model"
|
||||
"github.com/cloudwego/eino/components/tool"
|
||||
"github.com/cloudwego/eino/schema"
|
||||
)
|
||||
|
||||
type scriptedToolCallingModel struct {
|
||||
responses []*schema.Message
|
||||
calls int
|
||||
err error
|
||||
block bool
|
||||
lastInput []*schema.Message
|
||||
}
|
||||
|
||||
type fakeMCPToolExecutor struct {
|
||||
toolCode string
|
||||
arguments map[string]any
|
||||
policy aitooling.Policy
|
||||
result *mcps.ToolCallResult
|
||||
err error
|
||||
}
|
||||
|
||||
type concurrentToolCallingModel struct {
|
||||
calls atomic.Int32
|
||||
}
|
||||
|
||||
var _ model.ToolCallingChatModel = (*concurrentToolCallingModel)(nil)
|
||||
|
||||
func (m *concurrentToolCallingModel) Generate(ctx context.Context, _ []*schema.Message, _ ...model.Option) (*schema.Message, error) {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
m.calls.Add(1)
|
||||
return schema.AssistantMessage("并发调用完成。", nil), nil
|
||||
}
|
||||
|
||||
func (m *concurrentToolCallingModel) Stream(ctx context.Context, input []*schema.Message, opts ...model.Option) (*schema.StreamReader[*schema.Message], error) {
|
||||
message, err := m.Generate(ctx, input, opts...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return schema.StreamReaderFromArray([]*schema.Message{message}), nil
|
||||
}
|
||||
|
||||
func (m *concurrentToolCallingModel) WithTools(_ []*schema.ToolInfo) (model.ToolCallingChatModel, error) {
|
||||
return m, nil
|
||||
}
|
||||
|
||||
func (e *fakeMCPToolExecutor) Execute(_ context.Context, toolCode string, arguments map[string]any, policy aitooling.Policy) (aitooling.Definition, *mcps.ToolCallResult, error) {
|
||||
e.toolCode = toolCode
|
||||
e.arguments = arguments
|
||||
e.policy = policy
|
||||
return aitooling.Definition{Code: toolCode, RiskLevel: aitooling.RiskLevelSensitive}, e.result, e.err
|
||||
}
|
||||
|
||||
var _ model.ToolCallingChatModel = (*scriptedToolCallingModel)(nil)
|
||||
|
||||
func (m *scriptedToolCallingModel) Generate(ctx context.Context, input []*schema.Message, _ ...model.Option) (*schema.Message, error) {
|
||||
m.lastInput = append([]*schema.Message(nil), input...)
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if m.block {
|
||||
<-ctx.Done()
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
if m.err != nil {
|
||||
return nil, m.err
|
||||
}
|
||||
if m.calls >= len(m.responses) {
|
||||
return nil, errors.New("unexpected model call")
|
||||
}
|
||||
result := m.responses[m.calls]
|
||||
m.calls++
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (m *scriptedToolCallingModel) Stream(ctx context.Context, input []*schema.Message, opts ...model.Option) (*schema.StreamReader[*schema.Message], error) {
|
||||
message, err := m.Generate(ctx, input, opts...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return schema.StreamReaderFromArray([]*schema.Message{message}), nil
|
||||
}
|
||||
|
||||
func (m *scriptedToolCallingModel) WithTools(_ []*schema.ToolInfo) (model.ToolCallingChatModel, error) {
|
||||
return m, nil
|
||||
}
|
||||
|
||||
func TestRunExecutesGuardedToolThenReturnsFinalAnswer(t *testing.T) {
|
||||
called := false
|
||||
guardedTool := &GuardedTool{
|
||||
InfoDefinition: &schema.ToolInfo{Name: "customer_lookup", Desc: "Read customer data"},
|
||||
Definition: aitooling.Definition{Code: "builtin/customer_lookup", RiskLevel: aitooling.RiskLevelRead},
|
||||
Policy: aitooling.Policy{AllowedToolCodes: []string{"builtin/customer_lookup"}},
|
||||
Handler: func(_ context.Context, arguments map[string]any) (string, error) {
|
||||
called = arguments["customerId"] == "42"
|
||||
return "customer: Ada", nil
|
||||
},
|
||||
}
|
||||
model := &scriptedToolCallingModel{responses: []*schema.Message{
|
||||
schema.AssistantMessage("", []schema.ToolCall{{ID: "call-1", Type: "function", Function: schema.FunctionCall{Name: "customer_lookup", Arguments: `{"customerId":"42"}`}}}),
|
||||
schema.AssistantMessage("已找到客户资料。", nil),
|
||||
}}
|
||||
|
||||
result, err := Run(context.Background(), ReActConfig{Model: model, Tools: []tool.BaseTool{guardedTool}, MaxSteps: 4}, []*schema.Message{schema.UserMessage("查询客户")})
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
if !called || result == nil || result.Content != "已找到客户资料。" || model.calls != 2 {
|
||||
t.Fatalf("unexpected ReAct result: called=%t result=%#v modelCalls=%d", called, result, model.calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunInjectsProvidedConversationContext(t *testing.T) {
|
||||
model := &scriptedToolCallingModel{responses: []*schema.Message{schema.AssistantMessage("已理解上下文。", nil)}}
|
||||
input := []*schema.Message{
|
||||
schema.SystemMessage("你是客服助手，优先引用知识库。"),
|
||||
schema.UserMessage("我的订单状态如何？"),
|
||||
}
|
||||
if _, err := Run(context.Background(), ReActConfig{Model: model}, input); err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
if len(model.lastInput) != len(input) || model.lastInput[0].Content != input[0].Content || model.lastInput[1].Content != input[1].Content {
|
||||
t.Fatalf("conversation context was not passed to model: %#v", model.lastInput)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewOpenAICompatibleModelValidatesExistingAIConfig(t *testing.T) {
|
||||
if _, err := NewOpenAICompatibleModel(context.Background(), models.AIConfig{}); err == nil {
|
||||
t.Fatal("expected incomplete AI config error")
|
||||
}
|
||||
configured, err := NewOpenAICompatibleModel(context.Background(), models.AIConfig{
|
||||
BaseURL: "https://api.example.test/v1", APIKey: "test-key", ModelName: "test-model", TimeoutMS: 1200, MaxOutputTokens: 256,
|
||||
})
|
||||
if err != nil || configured == nil {
|
||||
t.Fatalf("expected OpenAI-compatible model adapter, model=%#v err=%v", configured, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGuardedToolRejectsDisallowedPolicyBeforeHandler(t *testing.T) {
|
||||
called := false
|
||||
guardedTool := &GuardedTool{
|
||||
InfoDefinition: &schema.ToolInfo{Name: "restricted_lookup", Desc: "Read restricted data"},
|
||||
Definition: aitooling.Definition{Code: "builtin/restricted_lookup", RiskLevel: aitooling.RiskLevelRead},
|
||||
Policy: aitooling.Policy{AllowedToolCodes: []string{"builtin/customer_lookup"}},
|
||||
Handler: func(context.Context, map[string]any) (string, error) {
|
||||
called = true
|
||||
return "unexpected", nil
|
||||
},
|
||||
}
|
||||
if _, err := guardedTool.InvokableRun(context.Background(), `{}`); err == nil {
|
||||
t.Fatal("expected policy rejection")
|
||||
}
|
||||
if called {
|
||||
t.Fatal("handler must not run after policy rejection")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunPropagatesCancellation(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
model := &scriptedToolCallingModel{responses: []*schema.Message{schema.AssistantMessage("unused", nil)}}
|
||||
if _, err := Run(ctx, ReActConfig{Model: model}, []*schema.Message{schema.UserMessage("查询")}); !errors.Is(err, context.Canceled) {
|
||||
t.Fatalf("expected cancellation, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunPropagatesDeadlineDuringModelCall(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
|
||||
defer cancel()
|
||||
model := &scriptedToolCallingModel{block: true}
|
||||
if _, err := Run(ctx, ReActConfig{Model: model}, []*schema.Message{schema.UserMessage("查询")}); !errors.Is(err, context.DeadlineExceeded) {
|
||||
t.Fatalf("expected deadline propagation, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunPropagatesModelFailure(t *testing.T) {
|
||||
modelErr := errors.New("model unavailable")
|
||||
model := &scriptedToolCallingModel{err: modelErr}
|
||||
if _, err := Run(context.Background(), ReActConfig{Model: model}, []*schema.Message{schema.UserMessage("查询")}); !errors.Is(err, modelErr) {
|
||||
t.Fatalf("expected model error propagation, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunPropagatesToolFailure(t *testing.T) {
|
||||
toolErr := errors.New("customer service unavailable")
|
||||
guardedTool := &GuardedTool{
|
||||
InfoDefinition: &schema.ToolInfo{Name: "failing_lookup", Desc: "Read customer data"},
|
||||
Definition: aitooling.Definition{Code: "builtin/failing_lookup", RiskLevel: aitooling.RiskLevelRead},
|
||||
Policy: aitooling.Policy{AllowedToolCodes: []string{"builtin/failing_lookup"}},
|
||||
Handler: func(context.Context, map[string]any) (string, error) {
|
||||
return "", toolErr
|
||||
},
|
||||
}
|
||||
model := &scriptedToolCallingModel{responses: []*schema.Message{
|
||||
schema.AssistantMessage("", []schema.ToolCall{{ID: "call-1", Type: "function", Function: schema.FunctionCall{Name: "failing_lookup", Arguments: `{}`}}}),
|
||||
}}
|
||||
if _, err := Run(context.Background(), ReActConfig{Model: model, Tools: []tool.BaseTool{guardedTool}}, []*schema.Message{schema.UserMessage("查询")}); !errors.Is(err, toolErr) {
|
||||
t.Fatalf("expected tool error propagation, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGuardedToolEnforcesTimeout(t *testing.T) {
|
||||
guardedTool := &GuardedTool{
|
||||
InfoDefinition: &schema.ToolInfo{Name: "slow_lookup", Desc: "Read customer data"},
|
||||
Definition: aitooling.Definition{Code: "builtin/slow_lookup", RiskLevel: aitooling.RiskLevelRead, TimeoutMS: 20},
|
||||
Policy: aitooling.Policy{AllowedToolCodes: []string{"builtin/slow_lookup"}},
|
||||
Handler: func(ctx context.Context, _ map[string]any) (string, error) {
|
||||
<-ctx.Done()
|
||||
return "", ctx.Err()
|
||||
},
|
||||
}
|
||||
if _, err := guardedTool.InvokableRun(context.Background(), `{}`); !errors.Is(err, context.DeadlineExceeded) {
|
||||
t.Fatalf("expected tool timeout, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGuardedToolEmitsTraceForPolicyFailure(t *testing.T) {
|
||||
var trace ToolTrace
|
||||
guardedTool := &GuardedTool{
|
||||
InfoDefinition: &schema.ToolInfo{Name: "restricted_lookup", Desc: "Read restricted data"},
|
||||
Definition: aitooling.Definition{Code: "builtin/restricted_lookup", RiskLevel: aitooling.RiskLevelRead},
|
||||
Policy: aitooling.Policy{AllowedToolCodes: []string{"builtin/other_lookup"}},
|
||||
Handler: func(context.Context, map[string]any) (string, error) {
|
||||
return "unexpected", nil
|
||||
},
|
||||
Trace: func(item ToolTrace) { trace = item },
|
||||
}
|
||||
if _, err := guardedTool.InvokableRun(context.Background(), `{"customerId":"42"}`); err == nil {
|
||||
t.Fatal("expected policy rejection")
|
||||
}
|
||||
if trace.ToolCode != "builtin/restricted_lookup" || trace.Status != "failed" || trace.Err == nil || trace.Arguments["customerId"] != "42" || trace.Duration < 0 {
|
||||
t.Fatalf("unexpected trace: %#v", trace)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMCPToolHandlerUsesSharedExecutorAndReducesResult(t *testing.T) {
|
||||
executor := &fakeMCPToolExecutor{result: &mcps.ToolCallResult{Content: []mcps.ToolResultContent{{Type: "text", Text: "customer: Ada"}}}}
|
||||
policy := aitooling.Policy{AllowedToolCodes: []string{"crm/customer_lookup"}, Confirmed: true}
|
||||
handler := NewMCPToolHandler(executor, "crm/customer_lookup", policy)
|
||||
result, err := handler(context.Background(), map[string]any{"customerId": "42"})
|
||||
if err != nil || result != "customer: Ada" {
|
||||
t.Fatalf("unexpected MCP handler result=%q err=%v", result, err)
|
||||
}
|
||||
if executor.toolCode != "crm/customer_lookup" || executor.arguments["customerId"] != "42" || !executor.policy.Confirmed {
|
||||
t.Fatalf("unexpected MCP execution: %#v", executor)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunStopsAtConfiguredMaxSteps(t *testing.T) {
|
||||
guardedTool := &GuardedTool{
|
||||
InfoDefinition: &schema.ToolInfo{Name: "loop_lookup", Desc: "Read loop data"},
|
||||
Definition: aitooling.Definition{Code: "builtin/loop_lookup", RiskLevel: aitooling.RiskLevelRead},
|
||||
Policy: aitooling.Policy{AllowedToolCodes: []string{"builtin/loop_lookup"}},
|
||||
Handler: func(context.Context, map[string]any) (string, error) {
|
||||
return "keep going", nil
|
||||
},
|
||||
}
|
||||
responses := make([]*schema.Message, 8)
|
||||
for i := range responses {
|
||||
responses[i] = schema.AssistantMessage("", []schema.ToolCall{{
|
||||
ID: "loop-call", Type: "function", Function: schema.FunctionCall{Name: "loop_lookup", Arguments: `{}`},
|
||||
}})
|
||||
}
|
||||
model := &scriptedToolCallingModel{responses: responses}
|
||||
if _, err := Run(context.Background(), ReActConfig{Model: model, Tools: []tool.BaseTool{guardedTool}, MaxSteps: 2}, []*schema.Message{schema.UserMessage("循环查询")}); err == nil {
|
||||
t.Fatal("expected configured maximum step limit to stop the loop")
|
||||
}
|
||||
}
|
||||
|
||||
func TestStreamReturnsModelOutput(t *testing.T) {
|
||||
model := &scriptedToolCallingModel{responses: []*schema.Message{schema.AssistantMessage("流式回复", nil)}}
|
||||
stream, err := Stream(context.Background(), ReActConfig{Model: model}, []*schema.Message{schema.UserMessage("查询")})
|
||||
if err != nil {
|
||||
t.Fatalf("Stream: %v", err)
|
||||
}
|
||||
defer stream.Close()
|
||||
result, err := schema.ConcatMessageStream(stream)
|
||||
if err != nil {
|
||||
t.Fatalf("ConcatMessageStream: %v", err)
|
||||
}
|
||||
if result.Content != "流式回复" {
|
||||
t.Fatalf("unexpected stream result: %#v", result)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStreamPropagatesCancellation(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
model := &scriptedToolCallingModel{responses: []*schema.Message{schema.AssistantMessage("unused", nil)}}
|
||||
if _, err := Stream(ctx, ReActConfig{Model: model}, []*schema.Message{schema.UserMessage("查询")}); !errors.Is(err, context.Canceled) {
|
||||
t.Fatalf("expected stream cancellation, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunSupportsConcurrentIndependentCalls(t *testing.T) {
|
||||
model := &concurrentToolCallingModel{}
|
||||
const workers = 16
|
||||
errs := make(chan error, workers)
|
||||
var group sync.WaitGroup
|
||||
for range workers {
|
||||
group.Add(1)
|
||||
go func() {
|
||||
defer group.Done()
|
||||
result, err := Run(context.Background(), ReActConfig{Model: model, MaxSteps: 3}, []*schema.Message{schema.UserMessage("并发查询")})
|
||||
if err != nil {
|
||||
errs <- err
|
||||
return
|
||||
}
|
||||
if result == nil || result.Content != "并发调用完成。" {
|
||||
errs <- errors.New("unexpected concurrent result")
|
||||
}
|
||||
}()
|
||||
}
|
||||
group.Wait()
|
||||
close(errs)
|
||||
for err := range errs {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if model.calls.Load() != workers {
|
||||
t.Fatalf("model calls = %d, want %d", model.calls.Load(), workers)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConfirmationBridgeUsesGenericInterruptAndResumeContracts(t *testing.T) {
|
||||
input := applicationruntime.RunInput{
|
||||
Conversation: models.Conversation{ID: 11}, UserMessage: models.Message{ID: 22},
|
||||
}
|
||||
result, err := BuildConfirmationResult(input, ConfirmationRequest{
|
||||
InterruptID: "confirm_refund", ToolCode: "graph/create_ticket_with_confirmation", Prompt: "是否确认提交退款工单？",
|
||||
Arguments: map[string]any{"title": "退款申请"},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("BuildConfirmationResult: %v", err)
|
||||
}
|
||||
if !result.Interrupted || result.Status != "interrupted" || result.CheckPointID == "" || len(result.Interrupts) != 1 || result.Interrupts[0].Type != confirmationInterruptType || result.Interrupts[0].ID != "confirm_refund" {
|
||||
t.Fatalf("unexpected confirmation result: %#v", result)
|
||||
}
|
||||
decision, checkpoint, err := ResumeConfirmation(result.CheckPointData, applicationruntime.ResumeInput{ResumeData: map[string]string{"confirm_refund": "确认"}})
|
||||
if err != nil || decision != string(graphs.ConfirmationDecisionConfirm) || checkpoint.ToolCode != "graph/create_ticket_with_confirmation" || checkpoint.Arguments["title"] != "退款申请" {
|
||||
t.Fatalf("unexpected resume bridge decision=%q checkpoint=%#v err=%v", decision, checkpoint, err)
|
||||
}
|
||||
decision, _, err = ResumeConfirmation(result.CheckPointData, applicationruntime.ResumeInput{ResumeData: map[string]string{"confirm_refund": "取消"}})
|
||||
if err != nil || decision != string(graphs.ConfirmationDecisionCancel) {
|
||||
t.Fatalf("unexpected cancellation decision=%q err=%v", decision, err)
|
||||
}
|
||||
}
|
||||
|
||||
func BenchmarkRunWithInjectedModel(b *testing.B) {
|
||||
model := &concurrentToolCallingModel{}
|
||||
input := []*schema.Message{schema.SystemMessage("你是客服助手。"), schema.UserMessage("查询订单状态")}
|
||||
b.ReportAllocs()
|
||||
b.ResetTimer()
|
||||
for range b.N {
|
||||
result, err := Run(context.Background(), ReActConfig{Model: model, MaxSteps: 3}, input)
|
||||
if err != nil || result == nil || result.Content == "" {
|
||||
b.Fatalf("Run result=%#v err=%v", result, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,141 @@
|
||||
package einoexperiment
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"agent-desk/internal/bootstrap"
|
||||
"agent-desk/internal/models"
|
||||
"agent-desk/internal/pkg/config"
|
||||
"agent-desk/internal/pkg/enums"
|
||||
|
||||
einoopenai "github.com/cloudwego/eino-ext/components/model/openai"
|
||||
einomodel "github.com/cloudwego/eino/components/model"
|
||||
"github.com/cloudwego/eino/schema"
|
||||
)
|
||||
|
||||
// TestRealOpenAICompatibleEndpoint is intentionally opt-in because it spends
|
||||
// a small amount of configured model quota. It verifies the production-shaped
|
||||
// OpenAI-compatible adapter without exposing credentials in test output.
|
||||
func TestRealOpenAICompatibleEndpoint(t *testing.T) {
|
||||
if os.Getenv("EINO_EXPERIMENT_REAL") != "1" {
|
||||
t.Skip("set EINO_EXPERIMENT_REAL=1 to run against the configured endpoint")
|
||||
}
|
||||
configPath := strings.TrimSpace(os.Getenv("EINO_EXPERIMENT_CONFIG"))
|
||||
var err error
|
||||
if configPath == "" {
|
||||
configPath, err = findExperimentConfigPath()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
workingDir, err := os.Getwd()
|
||||
if err != nil {
|
||||
t.Fatalf("get working directory: %v", err)
|
||||
}
|
||||
repoRoot := filepath.Dir(filepath.Dir(configPath))
|
||||
if err := os.Chdir(repoRoot); err != nil {
|
||||
t.Fatalf("change to config root: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = os.Chdir(workingDir) })
|
||||
cfg, err := config.Load(configPath)
|
||||
if err != nil {
|
||||
t.Fatalf("load config: %v", err)
|
||||
}
|
||||
db, err := bootstrap.InitDB(cfg.DB)
|
||||
if err != nil {
|
||||
t.Fatalf("open configured database: %v", err)
|
||||
}
|
||||
sqlDB, err := db.DB()
|
||||
if err == nil {
|
||||
defer sqlDB.Close()
|
||||
}
|
||||
var aiConfig models.AIConfig
|
||||
if err := db.Where("model_type = ? AND status = ?", enums.AIModelTypeLLM, enums.StatusOk).Order("id").First(&aiConfig).Error; err != nil {
|
||||
t.Fatalf("load enabled LLM config: %v", err)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(maxInt(aiConfig.TimeoutMS, 30000))*time.Millisecond)
|
||||
defer cancel()
|
||||
model, err := NewOpenAICompatibleModel(ctx, aiConfig)
|
||||
if err != nil {
|
||||
t.Fatalf("create Eino model adapter: %v", err)
|
||||
}
|
||||
input := []*schema.Message{schema.SystemMessage("You are a terse service assistant."), schema.UserMessage("Reply with exactly: OK")}
|
||||
startedAt := time.Now()
|
||||
result, err := Run(ctx, ReActConfig{Model: model, MaxSteps: 2}, input)
|
||||
if err != nil {
|
||||
t.Fatalf("Eino ReAct request: %v", err)
|
||||
}
|
||||
if result == nil || strings.TrimSpace(result.Content) == "" {
|
||||
t.Fatal("Eino endpoint returned an empty response")
|
||||
}
|
||||
if result.ResponseMeta == nil || result.ResponseMeta.Usage == nil {
|
||||
t.Fatal("Eino endpoint did not return token usage")
|
||||
}
|
||||
t.Logf("real endpoint verified: latency=%s promptTokens=%d completionTokens=%d", time.Since(startedAt).Round(time.Millisecond), result.ResponseMeta.Usage.PromptTokens, result.ResponseMeta.Usage.CompletionTokens)
|
||||
|
||||
toolModel, err := model.WithTools([]*schema.ToolInfo{{
|
||||
Name: "eino_echo",
|
||||
Desc: "Echoes a short input. Always call this tool when asked to verify tool calling.",
|
||||
ParamsOneOf: schema.NewParamsOneOfByParams(map[string]*schema.ParameterInfo{
|
||||
"text": {Type: schema.String, Desc: "Short text to echo", Required: true},
|
||||
}),
|
||||
}})
|
||||
if err != nil {
|
||||
t.Fatalf("bind Eino tool: %v", err)
|
||||
}
|
||||
toolResult, err := toolModel.Generate(ctx, []*schema.Message{schema.UserMessage("Verify tool calling by invoking eino_echo with text OK.")},
|
||||
einomodel.WithToolChoice(schema.ToolChoiceForced, "eino_echo"),
|
||||
einoopenai.WithExtraFields(map[string]any{"enable_thinking": false}),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("real endpoint tool call: %v", err)
|
||||
}
|
||||
if toolResult == nil || len(toolResult.ToolCalls) != 1 || toolResult.ToolCalls[0].Function.Name != "eino_echo" {
|
||||
t.Fatalf("expected one eino_echo tool call, got %#v", toolResult)
|
||||
}
|
||||
|
||||
stream, err := model.Stream(ctx, []*schema.Message{schema.UserMessage("Reply with exactly: STREAM_OK")})
|
||||
if err != nil {
|
||||
t.Fatalf("real endpoint stream: %v", err)
|
||||
}
|
||||
// ConcatMessageStream consumes and closes the Eino reader. Do not close it
|
||||
// again here: v0.9.6 treats a second close as a panic.
|
||||
streamResult, err := schema.ConcatMessageStream(stream)
|
||||
if err != nil {
|
||||
t.Fatalf("concat real stream: %v", err)
|
||||
}
|
||||
if streamResult == nil || strings.TrimSpace(streamResult.Content) == "" {
|
||||
t.Fatal("Eino endpoint stream returned an empty response")
|
||||
}
|
||||
t.Logf("real endpoint tool and stream verified: toolCalls=%d streamChars=%d", len(toolResult.ToolCalls), len([]rune(streamResult.Content)))
|
||||
}
|
||||
|
||||
func findExperimentConfigPath() (string, error) {
|
||||
dir, err := os.Getwd()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
for {
|
||||
candidate := filepath.Join(dir, "config", "config.yaml")
|
||||
if _, err := os.Stat(candidate); err == nil {
|
||||
return candidate, nil
|
||||
}
|
||||
parent := filepath.Dir(dir)
|
||||
if parent == dir {
|
||||
return "", os.ErrNotExist
|
||||
}
|
||||
dir = parent
|
||||
}
|
||||
}
|
||||
|
||||
func maxInt(value, fallback int) int {
|
||||
if value > 0 {
|
||||
return value
|
||||
}
|
||||
return fallback
|
||||
}
|
||||
Reference in New Issue
Block a user