t
18c9354095
- 注入数据库、运行时配置、统一响应、文件存储和平台 AI 能力，补充业务读写工具与客户快捷操作契约。 - 移除模块内重复的组织、客户、工单、标签、技能、旧工作流、MCP 和迁移实现，将身份权限与业务主体交由宿主管理。 - 使用 libSQL 重构向量存储，并完善图片消息、访客身份、排队调度、企业微信和支持聊天页面。 - 统一 HTTP、DTO 与 WebSocket 的 snake_case 协议，补齐模块初始化、业务动作和公共载荷等回归测试。
374 lines
13 KiB
Go
374 lines
13 KiB
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, ¶ms); err != nil {
|
||
return nil, fmt.Errorf("decode Eino tool schema: %w", err)
|
||
}
|
||
info.ParamsOneOf = schema.NewParamsOneOfByJSONSchema(¶ms)
|
||
}
|
||
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
|
||
}
|