「产品矩阵平台」平台级 AI 支撑架构

type ModelClient interface {
    Chat(ctx context.Context, req ChatRequest) (*ChatResponse, error)
    Embed(ctx context.Context, req EmbedRequest) (*EmbedResponse, error)
}

文档上传 -> 切分 chunk -> 生成 embedding -> 写入向量库
用户问题 -> 生成 query embedding -> 检索 TopK -> 重排序 -> 拼接上下文 -> 调用模型

package aigateway

import (
    "context"
    "fmt"
    "time"
)

// AIRequest 统一请求结构
type AIRequest struct {
    TenantID        string                 `json:"tenant_id"`
    AppID           string                 `json:"app_id"`
    TaskType        string                 `json:"task_type"`        // chat / embed / summarize
    PromptTemplate  string                 `json:"prompt_template"`
    Variables       map[string]interface{} `json:"variables"`
    Temperature     float64                `json:"temperature"`
    MaxTokens       int                    `json:"max_tokens"`
    Timeout         time.Duration          `json:"timeout"`
    SafetyLevel     string                 `json:"safety_level"`     // low / medium / high
}

type AIResponse struct {
    Content      string  `json:"content"`
    ModelUsed    string  `json:"model_used"`
    InputTokens  int     `json:"input_tokens"`
    OutputTokens int     `json:"output_tokens"`
    CostUSD      float64 `json:"cost_usd"`
    Cached       bool    `json:"cached"`
}

type Gateway struct {
    router      ModelRouter
    meter       UsageMeter
    safety      SafetyGuard
    cache       Cache
    templates   PromptTemplateStore
}

func (g *Gateway) Call(ctx context.Context, req AIRequest) (*AIResponse, error) {
    // 1. 检查租户配额
    if err := g.meter.CheckQuota(ctx, req.TenantID, req.TaskType); err != nil {
        return nil, fmt.Errorf("quota exceeded: %w", err)
    }

    // 2. 加载 Prompt 模板并填充变量
    prompt, err := g.templates.Render(ctx, req.PromptTemplate, req.Variables)
    if err != nil {
        return nil, err
    }

    // 3. 安全审查（输入过滤）
    if req.SafetyLevel == "high" {
        if blocked := g.safety.CheckInput(ctx, prompt); blocked {
            return nil, fmt.Errorf("input blocked by safety policy")
        }
    }

    // 4. 缓存命中检查
    cacheKey := Hash(prompt + req.TaskType)
    if cached := g.cache.Get(ctx, cacheKey); cached != nil {
        return cached.(*AIResponse), nil
    }

    // 5. 模型路由
    client, modelConfig := g.router.Select(ctx, req.TaskType, req.TenantID)

    // 6. 调用模型（带超时）
    callCtx, cancel := context.WithTimeout(ctx, req.Timeout)
    defer cancel()

    resp, err := client.Chat(callCtx, ChatRequest{
        Messages:    []Message{{Role: "user", Content: prompt}},
        Temperature: req.Temperature,
        MaxTokens:   req.MaxTokens,
    })
    if err != nil {
        return nil, fmt.Errorf("model call failed: %w", err)
    }

    // 7. 安全审查（输出过滤）
    if req.SafetyLevel != "low" {
        if flagged := g.safety.CheckOutput(ctx, resp.Content); flagged {
            return nil, fmt.Errorf("output flagged by safety policy")
        }
    }

    result := &AIResponse{
        Content:      resp.Content,
        ModelUsed:    modelConfig.Name,
        InputTokens:  resp.Usage.PromptTokens,
        OutputTokens: resp.Usage.CompletionTokens,
        CostUSD:      modelConfig.CalcCost(resp.Usage.PromptTokens, resp.Usage.CompletionTokens),
        Cached:       false,
    }

    // 8. 记录用量
    g.meter.Record(ctx, req.TenantID, req.AppID, req.TaskType, result)

    // 9. 缓存结果
    g.cache.Set(ctx, cacheKey, result, 5*time.Minute)

    return result, nil
}

package modelclient

import (
    "context"
    "fmt"
)

// ModelClient 统一模型调用接口
type ModelClient interface {
    Chat(ctx context.Context, req ChatRequest) (*ChatResponse, error)
    Embed(ctx context.Context, req EmbedRequest) (*EmbedResponse, error)
    Name() string // 供应商标识
}

// OpenAIAdapter OpenAI / Azure 适配器
type OpenAIAdapter struct {
    client *openai.Client
    model  string
}

func (a *OpenAIAdapter) Name() string { return "openai:" + a.model }

func (a *OpenAIAdapter) Chat(ctx context.Context, req ChatRequest) (*ChatResponse, error) {
    msgs := make([]openai.ChatCompletionMessageParamUnion, len(req.Messages))
    for i, m := range req.Messages {
        msgs[i] = openai.UserMessage(m.Content)
    }
    resp, err := a.client.Chat.Completions.New(ctx, openai.ChatCompletionNewParams{
        Model:    openai.F(a.model),
        Messages: openai.F(msgs),
    })
    if err != nil {
        return nil, err
    }
    return &ChatResponse{
        Content: resp.Choices[0].Message.Content,
        Usage:   Usage{PromptTokens: int(resp.Usage.PromptTokens), CompletionTokens: int(resp.Usage.CompletionTokens)},
    }, nil
}

// LocalAdapter 本地 Ollama / vLLM 适配器
type LocalAdapter struct {
    baseURL string
    model   string
}

func (a *LocalAdapter) Name() string { return "local:" + a.model }

func (a *LocalAdapter) Chat(ctx context.Context, req ChatRequest) (*ChatResponse, error) {
    // 调用本地 HTTP 接口
    // 实现略，结构与 OpenAIAdapter 类似
    return &ChatResponse{}, fmt.Errorf("not implemented")
}

// Router 按任务类型和成本策略选择模型
type Router struct {
    clients map[string][]ModelClient // task_type -> clients
    configs map[string]ModelConfig
}

type ModelConfig struct {
    Name            string
    MaxTokens       int
    CostPer1KInput  float64
    CostPer1KOutput float64
    LatencyMS       int
}

func (r *Router) Select(ctx context.Context, taskType, tenantID string) (ModelClient, ModelConfig) {
    candidates := r.clients[taskType]
    // 简化策略：优先成本低，兜底稳定性
    return candidates[0], r.configs[candidates[0].Name()]
}

func (c ModelConfig) CalcCost(inputTokens, outputTokens int) float64 {
    return float64(inputTokens)*c.CostPer1KInput/1000 + float64(outputTokens)*c.CostPer1KOutput/1000
}

package prompt

import (
    "bytes"
    "context"
    "encoding/json"
    "fmt"
    "html/template"
    "time"
)

type Template struct {
    ID          string    `json:"id"`
    Version     string    `json:"version"`
    Content     string    `json:"content"`     // Go template syntax
    Variables   []string  `json:"variables"`
    ModelPolicy string    `json:"model_policy"` // chat / gpt-4 / local-llama
    SafetyLevel string    `json:"safety_level"`
    Owner       string    `json:"owner"`
    CreatedAt   time.Time `json:"created_at"`
}

func (t *Template) Render(variables map[string]interface{}) (string, error) {
    tmpl, err := template.New("prompt").Parse(t.Content)
    if err != nil {
        return "", fmt.Errorf("invalid template: %w", err)
    }
    var buf bytes.Buffer
    if err := tmpl.Execute(&buf, variables); err != nil {
        return "", fmt.Errorf("render failed: %w", err)
    }
    return buf.String(), nil
}

// TemplateStore 模板存储与版本管理
type TemplateStore struct {
    db *gorm.DB
}

func (s *TemplateStore) Create(ctx context.Context, tmpl *Template) error {
    // 新模板默认 v1
    tmpl.Version = "v1"
    tmpl.CreatedAt = time.Now()
    return s.db.Create(tmpl).Error
}

func (s *TemplateStore) Update(ctx context.Context, tmpl *Template) error {
    // 版本号自增
    tmpl.Version = bumpVersion(tmpl.Version)
    return s.db.Create(tmpl).Error // 保留历史版本
}

func (s *TemplateStore) GetLatest(ctx context.Context, templateID, tenantID string) (*Template, error) {
    var tmpl Template
    err := s.db.Where("id = ? AND (target_tenant IS NULL OR target_tenant = ?)", templateID, tenantID).
        Order("created_at DESC").
        First(&tmpl).Error
    return &tmpl, err
}

package rag

import (
    "context"
    "fmt"
    "sort"
)

// Retriever 执行向量检索 + 重排序 + 上下文拼接
type Retriever struct {
    embedder Embedder
    vectorDB VectorStore
    reRanker ReRanker
}

func (r *Retriever) Retrieve(ctx context.Context, query string, topK int) (*Context, error) {
    // 1. 生成 query embedding
    embedding, err := r.embedder.Embed(ctx, query)
    if err != nil {
        return nil, fmt.Errorf("embedding failed: %w", err)
    }

    // 2. 向量检索（粗排）
    candidates, err := r.vectorDB.Search(ctx, embedding, topK*3)
    if err != nil {
        return nil, fmt.Errorf("vector search failed: %w", err)
    }

    // 3. 重排序（精排）
    scored := r.reRanker.Rank(query, candidates)
    sort.Slice(scored, func(i, j int) bool { return scored[i].Score > scored[j].Score })
    if len(scored) > topK {
        scored = scored[:topK]
    }

    // 4. 组装上下文
    ctx := &Context{
        Query:       query,
        References:  make([]Reference, len(scored)),
        TotalTokens: 0,
    }
    for i, s := range scored {
        ctx.References[i] = Reference{
            Document: s.Document,
            Score:    s.Score,
            Source:   s.Document.Meta.Source,
            UpdatedAt: s.Document.Meta.UpdatedAt,
        }
        ctx.TotalTokens += s.Document.TokenCount
    }
    return ctx, nil
}

// BuildPrompt 将检索结果和系统指令拼接为最终 Prompt
func BuildPrompt(query string, ctx *Context, systemPrompt string) string {
    var buf bytes.Buffer
    buf.WriteString(systemPrompt)
    buf.WriteString("\n\n## 参考资料（按相关性排序）：\n")
    for i, ref := range ctx.References {
        buf.WriteString(fmt.Sprintf("[%d] %s（来源：%s，更新时间：%s）\n%s\n\n",
            i+1, ref.Document.Title, ref.Source, ref.UpdatedAt.Format("2006-01-02"), ref.Document.Content))
    }
    buf.WriteString("## 用户问题：\n")
    buf.WriteString(query)
    buf.WriteString("\n\n请基于参考资料回答，如果参考资料不足以回答，请说明。")
    return buf.String()
}

package aitask

import (
    "context"
    "encoding/json"
    "time"
)

// AITask 异步任务模型
type AITask struct {
    ID          string          `json:"id"`
    TenantID    string          `json:"tenant_id"`
    AppID       string          `json:"app_id"`
    Type        string          `json:"type"`         // summarize / generate / classify
    Status      string          `json:"status"`       // pending / running / completed / failed / review
    Input       json.RawMessage `json:"input"`
    Output      string          `json:"output"`
    ModelUsed   string          `json:"model_used"`
    ReviewFlag  string          `json:"review_flag"`  // auto / manual / rejected
    ReviewerID  string          `json:"reviewer_id"`
    CostUSD     float64         `json:"cost_usd"`
    CreatedAt   time.Time       `json:"created_at"`
    CompletedAt *time.Time      `json:"completed_at"`
}

// TaskWorker 消费异步 AI 任务队列
func (w *TaskWorker) Process(ctx context.Context, taskID string) error {
    var task AITask
    if err := w.db.First(&task, "id = ?", taskID).Error; err != nil {
        return err
    }

    // 更新状态为 running
    w.db.Model(&task).Update("status", "running")

    var req aigateway.AIRequest
    _ = json.Unmarshal(task.Input, &req)

    resp, err := w.gateway.Call(ctx, req)
    if err != nil {
        w.db.Model(&task).Updates(map[string]interface{}{
            "status": "failed",
            "output": err.Error(),
        })
        return err
    }

    // 判断是否需要人工复核
    reviewFlag := "auto"
    if req.SafetyLevel == "high" || ContainsHighRiskKeywords(resp.Content) {
        reviewFlag = "manual"
    }

    completedAt := time.Now()
    w.db.Model(&task).Updates(map[string]interface{}{
        "status":        "completed",
        "output":        resp.Content,
        "model_used":    resp.ModelUsed,
        "cost_usd":      resp.CostUSD,
        "review_flag":   reviewFlag,
        "completed_at":  &completedAt,
    })

    // 触发通知
    if reviewFlag == "manual" {
        w.notify.ReviewPending(task.TenantID, task.ID)
    }
    return nil
}