ai-agent/internal/ai/application/runtime/eino_agent_loop.go

package runtime

import (
	"context"
	"crypto/sha256"
	"encoding/json"
	"fmt"
	"strings"
	"sync/atomic"
	"time"

	ai "code.tczkiot.com/wlw/ai-agent/internal/ai"
	"code.tczkiot.com/wlw/ai-agent/internal/models"

	einoopenai "github.com/cloudwego/eino-ext/components/model/openai"
	einomodel "github.com/cloudwego/eino/components/model"
	einotool "github.com/cloudwego/eino/components/tool"
	"github.com/cloudwego/eino/compose"
	"github.com/cloudwego/eino/flow/agent/react"
	"github.com/cloudwego/eino/schema"
	einojsonschema "github.com/eino-contrib/jsonschema"
	"github.com/google/uuid"
)

const visionUnavailableInstruction = "The current customer message is an image, but this model did not receive usable image pixels. Never claim that you saw, read, or identified anything in the photo. You may still use verified business tools for the bound device, but for visual details ask the customer to describe the visible symptom or offer human support."

// einoAgentLoop is the production model/tool loop. AgentDesk still owns tool
// authorization, business execution, interrupts, idempotency, and auditing.
func einoAgentLoop(
	ctx context.Context,
	config models.AIConfig,
	systemPrompt string,
	userPrompt string,
	images []ai.ImageInput,
	definitions []ai.ToolDefinition,
	maxSteps int,
	execute ai.ToolCallExecutor,
) (*ai.ToolLoopResult, error) {
	model, err := newEinoChatModel(ctx, config)
	if err != nil {
		return nil, err
	}
	if maxSteps <= 0 {
		maxSteps = 6
	}
	tools := make([]einotool.BaseTool, 0, len(definitions))
	for _, definition := range definitions {
		tool, buildErr := newEinoFunctionTool(definition, execute)
		if buildErr != nil {
			return nil, buildErr
		}
		tools = append(tools, tool)
	}
	agent, err := react.NewAgent(ctx, &react.AgentConfig{
		ToolCallingModel: model,
		ToolsConfig:      compose.ToolsNodeConfig{Tools: tools},
		MaxStep:          maxSteps,
	})
	if err != nil {
		return nil, fmt.Errorf("create Eino agent loop: %w", err)
	}
	messages := make([]*schema.Message, 0, 2)
	if value := strings.TrimSpace(systemPrompt); value != "" {
		messages = append(messages, schema.SystemMessage(value))
	}
	messages = append(messages, buildEinoUserMessage(userPrompt, images))
	result, err := agent.Generate(ctx, messages)
	if err != nil && len(images) > 0 && isVisionUnsupportedError(err) {
		// Some OpenAI-compatible endpoints expose text-only models behind the
		// same API. Retry once without image parts so the customer still gets a
		// useful text response instead of a failed conversation turn.
		messages = buildVisionFallbackMessages(messages, userPrompt)
		result, err = agent.Generate(ctx, messages)
	}
	if err != nil {
		return nil, err
	}
	if result == nil {
		return nil, fmt.Errorf("Eino agent loop returned no result")
	}
	ret := &ai.ToolLoopResult{ChatCompletionResult: ai.ChatCompletionResult{
		Content:   strings.TrimSpace(result.Content),
		ModelName: config.ModelName,
	}}
	if result.ResponseMeta != nil && result.ResponseMeta.Usage != nil {
		ret.PromptTokens = result.ResponseMeta.Usage.PromptTokens
		ret.CompletionTokens = result.ResponseMeta.Usage.CompletionTokens
	}
	return ret, nil
}

func buildVisionFallbackMessages(messages []*schema.Message, userPrompt string) []*schema.Message {
	ret := append([]*schema.Message(nil), messages...)
	if len(ret) > 0 {
		ret = ret[:len(ret)-1]
	}
	ret = append(ret, schema.SystemMessage(visionUnavailableInstruction), schema.UserMessage(strings.TrimSpace(userPrompt)))
	return ret
}

func buildEinoUserMessage(userPrompt string, images []ai.ImageInput) *schema.Message {
	prompt := strings.TrimSpace(userPrompt)
	if len(images) == 0 {
		return schema.UserMessage(prompt)
	}
	parts := make([]schema.MessageInputPart, 0, len(images)+2)
	parts = append(parts, schema.MessageInputPart{Type: schema.ChatMessagePartTypeText, Text: prompt})
	parts = append(parts, schema.MessageInputPart{
		Type: schema.ChatMessagePartTypeText,
		Text: "以下是客户本次同一批上传的图片，按顺序编号为图1、图2……。请先逐图核验，再结合客户文字和实时业务工具判断；看不清时明确说明，不要臆测。",
	})
	for index, image := range images {
		base64Data := strings.TrimSpace(image.Base64Data)
		mimeType := strings.TrimSpace(image.MIMEType)
		if base64Data == "" || mimeType == "" {
			continue
		}
		parts = append(parts, schema.MessageInputPart{
			Type: schema.ChatMessagePartTypeText,
			Text: fmt.Sprintf("图%d（%s）：", index+1, fallbackVisionFilename(image.Filename)),
		})
		parts = append(parts, schema.MessageInputPart{
			Type: schema.ChatMessagePartTypeImageURL,
			Image: &schema.MessageInputImage{
				MessagePartCommon: schema.MessagePartCommon{Base64Data: &base64Data, MIMEType: mimeType},
				Detail:            schema.ImageURLDetailHigh,
			},
		})
	}
	if len(parts) == 2 {
		return schema.UserMessage(prompt)
	}
	return &schema.Message{Role: schema.User, UserInputMultiContent: parts}
}

func fallbackVisionFilename(filename string) string {
	if value := strings.TrimSpace(filename); value != "" {
		return value
	}
	return "未命名图片"
}

func supportsVisionInput(config models.AIConfig) bool {
	// The managed platform gateway inspects multimodal content and routes image
	// turns to its dedicated vision model, independently of the text model name
	// exposed in the tenant snapshot.
	if config.Platform {
		return config.VisionEnabled && strings.TrimSpace(config.VisionModel) != ""
	}
	name := strings.ToLower(strings.TrimSpace(config.ModelName))
	if name == "" {
		return false
	}
	for _, marker := range []string{
		"qwen-vl", "qwen2-vl", "qwen2.5-vl", "qwen3-vl", "qwen-omni",
		"gpt-4o", "gpt-4.1", "gpt-5", "gemini", "claude-3", "claude-4",
		"vision", "multimodal", "multi-modal",
	} {
		if strings.Contains(name, marker) {
			return true
		}
	}
	return false
}

func isVisionUnsupportedError(err error) bool {
	if err == nil {
		return false
	}
	value := strings.ToLower(err.Error())
	if !strings.Contains(value, "image") && !strings.Contains(value, "vision") && !strings.Contains(value, "multimodal") && !strings.Contains(value, "multi-modal") {
		return false
	}
	for _, marker := range []string{"unsupported", "not support", "does not support", "invalid content", "content must be", "unknown content"} {
		if strings.Contains(value, marker) {
			return true
		}
	}
	return false
}

func newEinoChatModel(ctx context.Context, config models.AIConfig) (einomodel.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),
		HTTPClient: config.HTTPClient,
	}
	if config.TimeoutMS > 0 {
		modelConfig.Timeout = time.Duration(config.TimeoutMS) * time.Millisecond
	}
	if config.MaxOutputTokens > 0 {
		maxTokens := config.MaxOutputTokens
		modelConfig.MaxCompletionTokens = &maxTokens
	}
	if isDeepSeekV4Model(config) {
		modelConfig.ExtraFields = map[string]any{
			"thinking": map[string]any{"type": "disabled"},
		}
	} else if isDashScopeQwenThinkingModel(config) {
		modelConfig.ExtraFields = map[string]any{"enable_thinking": false}
	}
	model, err := einoopenai.NewChatModel(ctx, modelConfig)
	if err != nil {
		return nil, fmt.Errorf("create Eino OpenAI-compatible model: %w", err)
	}
	if config.Platform {
		return &platformRequestIDChatModel{inner: model, requestIDBase: platformRequestIDBase(ctx), callIndex: &atomic.Uint64{}}, nil
	}
	return model, nil
}

// platformRequestIDChatModel gives every Eino model step its own idempotency
// key. A ReAct run can call the model multiple times, so the key must be fresh
// per Generate/Stream invocation rather than shared by the whole agent run.
type platformRequestIDChatModel struct {
	inner         einomodel.ToolCallingChatModel
	requestIDBase string
	callIndex     *atomic.Uint64
}

func (m *platformRequestIDChatModel) Generate(ctx context.Context, input []*schema.Message, opts ...einomodel.Option) (*schema.Message, error) {
	opts = append(opts, einoopenai.WithExtraHeader(map[string]string{
		"X-AI-Request-ID": m.nextRequestID(),
	}))
	return m.inner.Generate(ctx, input, opts...)
}

func (m *platformRequestIDChatModel) Stream(ctx context.Context, input []*schema.Message, opts ...einomodel.Option) (*schema.StreamReader[*schema.Message], error) {
	opts = append(opts, einoopenai.WithExtraHeader(map[string]string{
		"X-AI-Request-ID": m.nextRequestID(),
	}))
	return m.inner.Stream(ctx, input, opts...)
}

func (m *platformRequestIDChatModel) WithTools(tools []*schema.ToolInfo) (einomodel.ToolCallingChatModel, error) {
	inner, err := m.inner.WithTools(tools)
	if err != nil {
		return nil, err
	}
	return &platformRequestIDChatModel{inner: inner, requestIDBase: m.requestIDBase, callIndex: m.callIndex}, nil
}

func (m *platformRequestIDChatModel) nextRequestID() string {
	if m.callIndex == nil {
		m.callIndex = &atomic.Uint64{}
	}
	step := m.callIndex.Add(1)
	if strings.TrimSpace(m.requestIDBase) == "" {
		return uuid.NewString()
	}
	// The same persisted message/revision starts from the same step sequence on
	// recovery. This lets the gateway deduplicate a response-lost retry without
	// collapsing distinct ReAct model steps into one billable request.
	return uuid.NewSHA1(uuid.NameSpaceOID, []byte(fmt.Sprintf("%s:step:%d", m.requestIDBase, step))).String()
}

type platformRequestIDBaseContextKey struct{}

func withPlatformRequestIDBase(ctx context.Context, base string) context.Context {
	return context.WithValue(ctx, platformRequestIDBaseContextKey{}, strings.TrimSpace(base))
}

func platformRequestIDBase(ctx context.Context) string {
	if ctx == nil {
		return ""
	}
	value, _ := ctx.Value(platformRequestIDBaseContextKey{}).(string)
	return strings.TrimSpace(value)
}

func isDeepSeekV4Model(config models.AIConfig) bool {
	baseURL := strings.ToLower(strings.TrimSpace(config.BaseURL))
	modelName := strings.ToLower(strings.TrimSpace(config.ModelName))
	return strings.Contains(baseURL, "api.deepseek.com") && strings.HasPrefix(modelName, "deepseek-v4-")
}

func isDashScopeQwenThinkingModel(config models.AIConfig) bool {
	baseURL := strings.ToLower(strings.TrimSpace(config.BaseURL))
	modelName := strings.ToLower(strings.TrimSpace(config.ModelName))
	return strings.Contains(baseURL, "dashscope.aliyuncs.com") && strings.HasPrefix(modelName, "qwen3")
}

type einoFunctionTool struct {
	info         *schema.ToolInfo
	originalName string
	execute      ai.ToolCallExecutor
}

var _ einotool.InvokableTool = (*einoFunctionTool)(nil)

func newEinoFunctionTool(definition ai.ToolDefinition, execute ai.ToolCallExecutor) (*einoFunctionTool, error) {
	originalName := strings.TrimSpace(definition.Name)
	if originalName == "" || execute == nil {
		return nil, fmt.Errorf("Eino tool name and executor are required")
	}
	info := &schema.ToolInfo{
		Name: normalizeEinoToolName(originalName),
		Desc: strings.TrimSpace(definition.Description),
	}
	if len(definition.Parameters) > 0 {
		data, err := json.Marshal(definition.Parameters)
		if err != nil {
			return nil, fmt.Errorf("encode Eino tool schema: %w", err)
		}
		var params einojsonschema.Schema
		if err := json.Unmarshal(data, &params); err != nil {
			return nil, fmt.Errorf("decode Eino tool schema: %w", err)
		}
		info.ParamsOneOf = schema.NewParamsOneOfByJSONSchema(&params)
	}
	return &einoFunctionTool{info: info, originalName: originalName, execute: execute}, nil
}

func normalizeEinoToolName(name string) string {
	name = strings.TrimSpace(name)
	valid := name != "" && len(name) <= 64
	for _, char := range name {
		if !isEinoToolNameCharacter(char) {
			valid = false
			break
		}
	}
	if valid {
		return name
	}

	var normalized strings.Builder
	for _, char := range name {
		if isEinoToolNameCharacter(char) {
			normalized.WriteRune(char)
		} else {
			normalized.WriteByte('_')
		}
	}
	base := strings.Trim(normalized.String(), "_")
	if base == "" {
		base = "tool"
	}
	hash := sha256.Sum256([]byte(name))
	suffix := fmt.Sprintf("_%x", hash[:6])
	maxBaseLength := 64 - len(suffix)
	if len(base) > maxBaseLength {
		base = base[:maxBaseLength]
	}
	return base + suffix
}

func isEinoToolNameCharacter(char rune) bool {
	return char >= 'a' && char <= 'z' ||
		char >= 'A' && char <= 'Z' ||
		char >= '0' && char <= '9' ||
		char == '_' || char == '-'
}

func (t *einoFunctionTool) Info(context.Context) (*schema.ToolInfo, error) {
	return t.info, nil
}

func (t *einoFunctionTool) InvokableRun(ctx context.Context, arguments string, _ ...einotool.Option) (string, error) {
	result, err := t.execute(ctx, ai.ToolCall{Name: t.originalName, Arguments: arguments})
	if err == nil {
		return result, nil
	}
	observation, marshalErr := json.Marshal(map[string]string{"error": err.Error()})
	if marshalErr != nil {
		return "", err
	}
	return string(observation), nil
}