18c9354095
- 注入数据库、运行时配置、统一响应、文件存储和平台 AI 能力,补充业务读写工具与客户快捷操作契约。 - 移除模块内重复的组织、客户、工单、标签、技能、旧工作流、MCP 和迁移实现,将身份权限与业务主体交由宿主管理。 - 使用 libSQL 重构向量存储,并完善图片消息、访客身份、排队调度、企业微信和支持聊天页面。 - 统一 HTTP、DTO 与 WebSocket 的 snake_case 协议,补齐模块初始化、业务动作和公共载荷等回归测试。
133 lines
4.3 KiB
Go
133 lines
4.3 KiB
Go
package ai
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/mlogclub/simple/sqls"
|
|
openai "github.com/openai/openai-go/v3"
|
|
"github.com/openai/openai-go/v3/option"
|
|
|
|
"code.tczkiot.com/wlw/ai-agent/internal/models"
|
|
"code.tczkiot.com/wlw/ai-agent/internal/pkg/enums"
|
|
"code.tczkiot.com/wlw/ai-agent/internal/pkg/errorsx"
|
|
"code.tczkiot.com/wlw/ai-agent/internal/pkg/tracex"
|
|
"code.tczkiot.com/wlw/ai-agent/internal/repositories"
|
|
)
|
|
|
|
func newOpenAIClient(config models.AIConfig) openai.Client {
|
|
opts := []option.RequestOption{
|
|
option.WithAPIKey(config.APIKey),
|
|
option.WithBaseURL(config.BaseURL),
|
|
}
|
|
if config.TimeoutMS > 0 {
|
|
opts = append(opts, option.WithRequestTimeout(time.Duration(config.TimeoutMS)*time.Millisecond))
|
|
}
|
|
if config.HTTPClient != nil {
|
|
opts = append(opts, option.WithHTTPClient(config.HTTPClient))
|
|
}
|
|
if config.MaxRetryCount >= 0 {
|
|
opts = append(opts, option.WithMaxRetries(config.MaxRetryCount))
|
|
}
|
|
return openai.NewClient(opts...)
|
|
}
|
|
|
|
type platformAIRequestScopeContextKey struct{}
|
|
type platformAIRequestPurposeContextKey struct{}
|
|
|
|
type platformAIRequestScope struct {
|
|
base string
|
|
mu sync.Mutex
|
|
next map[string]uint64
|
|
}
|
|
|
|
// WithPlatformAIRequestScope binds a persisted business operation identity to
|
|
// platform AI calls. Recreating a scope with the same base during recovery
|
|
// reproduces the same purpose/ordinal request IDs, while one live scope gives
|
|
// every logical upstream call a distinct ordinal.
|
|
func WithPlatformAIRequestScope(ctx context.Context, base string) context.Context {
|
|
if ctx == nil {
|
|
ctx = context.Background()
|
|
}
|
|
base = strings.TrimSpace(base)
|
|
if base == "" {
|
|
return ctx
|
|
}
|
|
return context.WithValue(ctx, platformAIRequestScopeContextKey{}, &platformAIRequestScope{
|
|
base: base,
|
|
next: make(map[string]uint64),
|
|
})
|
|
}
|
|
|
|
// WithPlatformAIRequestPurpose separates otherwise identical calls belonging
|
|
// to different stages such as retrieval and document indexing.
|
|
func WithPlatformAIRequestPurpose(ctx context.Context, purpose string) context.Context {
|
|
if ctx == nil {
|
|
ctx = context.Background()
|
|
}
|
|
purpose = strings.TrimSpace(purpose)
|
|
if purpose == "" {
|
|
return ctx
|
|
}
|
|
return context.WithValue(ctx, platformAIRequestPurposeContextKey{}, purpose)
|
|
}
|
|
|
|
func ensurePlatformAIRequestScope(ctx context.Context) context.Context {
|
|
if ctx == nil {
|
|
ctx = context.Background()
|
|
}
|
|
if scope, _ := ctx.Value(platformAIRequestScopeContextKey{}).(*platformAIRequestScope); scope != nil && strings.TrimSpace(scope.base) != "" {
|
|
return ctx
|
|
}
|
|
if requestID := tracex.RequestIDFromContext(ctx); requestID != "" {
|
|
return WithPlatformAIRequestScope(ctx, "request:"+requestID)
|
|
}
|
|
// No durable business identity is available (for example a one-off debug
|
|
// call), so create one scope for the public operation. Callers with recovery
|
|
// semantics must bind their persisted identity explicitly.
|
|
return WithPlatformAIRequestScope(ctx, "operation:"+uuid.NewString())
|
|
}
|
|
|
|
func nextPlatformAIRequestID(ctx context.Context, defaultPurpose string) string {
|
|
ctx = ensurePlatformAIRequestScope(ctx)
|
|
scope, _ := ctx.Value(platformAIRequestScopeContextKey{}).(*platformAIRequestScope)
|
|
purpose, _ := ctx.Value(platformAIRequestPurposeContextKey{}).(string)
|
|
purpose = strings.TrimSpace(purpose)
|
|
if purpose == "" {
|
|
purpose = strings.TrimSpace(defaultPurpose)
|
|
}
|
|
if purpose == "" {
|
|
purpose = "request"
|
|
}
|
|
|
|
scope.mu.Lock()
|
|
scope.next[purpose]++
|
|
ordinal := scope.next[purpose]
|
|
scope.mu.Unlock()
|
|
return uuid.NewSHA1(uuid.NameSpaceOID, []byte(fmt.Sprintf("%s:purpose:%s:call:%d", scope.base, purpose, ordinal))).String()
|
|
}
|
|
|
|
// platformRequestOptions creates one deterministic idempotency key for one
|
|
// logical upstream call. SDK retries reuse these options. A later call in the
|
|
// same operation gets the next ordinal; recovery recreates the same sequence.
|
|
func platformRequestOptions(ctx context.Context, config models.AIConfig, purpose string) []option.RequestOption {
|
|
if !config.Platform {
|
|
return nil
|
|
}
|
|
return []option.RequestOption{
|
|
option.WithHeader("X-AI-Request-ID", nextPlatformAIRequestID(ctx, purpose)),
|
|
}
|
|
}
|
|
|
|
func GetEnabledAIConfig(modelType enums.AIModelType) (*models.AIConfig, error) {
|
|
item := repositories.AIConfigRepository.GetEnabled(sqls.DB(), modelType)
|
|
if item == nil {
|
|
return nil, errorsx.BusinessErrorI18n(2005, "error.aiConfig.noneEnabled")
|
|
}
|
|
return item, nil
|
|
}
|