Your NameandClaude Opus 5
a024d59827
按协议而不是按厂商做适配，与现有短信、对象存储的做法一致：openai 兼容格式 一个适配器即可覆盖 DeepSeek、通义、智谱、火山方舟、Ollama 等，新增厂商通常 只需在后台加一条模型记录；anthropic、gemini、webhook、debug 各一个适配器。 密钥沿用 integration.go 的 AES-GCM 加密存储。 测试账号托管的回复走 persistMessage 同一条落库和推送路径，因此未读数、 WebSocket 推送和会话排序全部复用现有逻辑，客户端无需改动。为此把 persistMessage 抽出 persistMessageContext，因为 worker 没有 *http.Request。 任务队列用数据库表而非内存：重启不丢回复，多实例不重复消费。领取用 UPDATE 打 claim_token 再回读，没有用 SELECT ... FOR UPDATE SKIP LOCKED——生产存在 MySQL 5.7 环境，那里该语法无法解析。同一会话同时只允许一个待处理任务 （pending_key 生成列 + 唯一键），所以用户连发多条消息只会得到一条回复。 闸门：仅 is_test=1 且在允许批次内的账号生效，托管账号之间不互相触发，三层 配额，调用失败默认静默。会话内是否标注 AI 身份由 ai.disclose_in_chat 控制 并默认开启，关闭前需确认所在地区的监管要求。 app.go、integration.go、im.go 三个文件同时包含本次改动之前工作区里就已存在 的未提交修改，一并带入。 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
425 lines
15 KiB
Go
425 lines
15 KiB
Go
package app
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha256"
|
||
"database/sql"
|
||
"encoding/hex"
|
||
"errors"
|
||
"fmt"
|
||
"net/http"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/go-sql-driver/mysql"
|
||
)
|
||
|
||
type aiModelRequest struct {
|
||
Name string `json:"name"`
|
||
Protocol string `json:"protocol"`
|
||
BaseURL string `json:"baseUrl"`
|
||
ModelName string `json:"modelName"`
|
||
APIKey string `json:"apiKey"`
|
||
ClearAPIKey bool `json:"clearApiKey"`
|
||
Temperature *float64 `json:"temperature"`
|
||
MaxTokens *int `json:"maxTokens"`
|
||
TimeoutMS *int `json:"timeoutMs"`
|
||
Status *int `json:"status"`
|
||
IsDefault *bool `json:"isDefault"`
|
||
}
|
||
|
||
const aiModelColumns = `id,name,protocol,base_url,model_name,api_key_cipher,temperature,max_tokens,timeout_ms,status,is_default,created_at,updated_at`
|
||
|
||
func (a *App) scanAIModel(scan func(dest ...any) error) (aiModel, error) {
|
||
var model aiModel
|
||
var cipher string
|
||
var created, updated time.Time
|
||
if err := scan(&model.ID, &model.Name, &model.Protocol, &model.BaseURL, &model.ModelName, &cipher,
|
||
&model.Temperature, &model.MaxTokens, &model.TimeoutMS, &model.Status, &model.IsDefault, &created, &updated); err != nil {
|
||
return aiModel{}, err
|
||
}
|
||
model.HasAPIKey = cipher != ""
|
||
if cipher != "" {
|
||
plain, err := a.decryptSecret(cipher)
|
||
if err != nil {
|
||
// A key that cannot be decrypted (rotated encryption key) must not
|
||
// silently become an empty key that reaches the provider.
|
||
return model, fmt.Errorf("模型 %s 的密钥无法解密，请重新填写", model.Name)
|
||
}
|
||
model.APIKey = plain
|
||
}
|
||
model.CreatedAt = created.Format(time.RFC3339)
|
||
model.UpdatedAt = updated.Format(time.RFC3339)
|
||
return model, nil
|
||
}
|
||
|
||
func (a *App) loadAIModel(ctx context.Context, id int64) (aiModel, error) {
|
||
row := a.db.QueryRowContext(ctx, `SELECT `+aiModelColumns+` FROM ai_models WHERE id=?`, id)
|
||
return a.scanAIModel(row.Scan)
|
||
}
|
||
|
||
// The default model is what the reply worker uses when an agent has no explicit
|
||
// binding: the configured id first, then the flagged row.
|
||
func (a *App) defaultAIModel(ctx context.Context) (aiModel, error) {
|
||
if configured := strings.TrimSpace(a.configPlain(ctx, "ai.default_model_id", "")); configured != "" {
|
||
row := a.db.QueryRowContext(ctx, `SELECT `+aiModelColumns+` FROM ai_models WHERE id=? AND status=1`, configured)
|
||
if model, err := a.scanAIModel(row.Scan); err == nil {
|
||
return model, nil
|
||
}
|
||
}
|
||
row := a.db.QueryRowContext(ctx, `SELECT `+aiModelColumns+` FROM ai_models WHERE status=1 ORDER BY is_default DESC,id ASC LIMIT 1`)
|
||
model, err := a.scanAIModel(row.Scan)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return aiModel{}, fmt.Errorf("尚未配置可用的 AI 模型")
|
||
}
|
||
return model, err
|
||
}
|
||
|
||
func (a *App) adminAIModels(w http.ResponseWriter, r *http.Request) {
|
||
rows, err := a.db.QueryContext(r.Context(), `SELECT `+aiModelColumns+` FROM ai_models ORDER BY is_default DESC,id ASC`)
|
||
if err != nil {
|
||
fail(w, http.StatusInternalServerError, 50001, "查询模型列表失败")
|
||
return
|
||
}
|
||
defer rows.Close()
|
||
items := []aiModel{}
|
||
for rows.Next() {
|
||
model, scanErr := a.scanAIModel(rows.Scan)
|
||
if scanErr != nil && model.ID == 0 {
|
||
fail(w, http.StatusInternalServerError, 50001, "查询模型列表失败")
|
||
return
|
||
}
|
||
model.APIKey = ""
|
||
items = append(items, model)
|
||
}
|
||
if rows.Err() != nil {
|
||
fail(w, http.StatusInternalServerError, 50001, "查询模型列表失败")
|
||
return
|
||
}
|
||
protocols := make([]map[string]string, 0, len(aiProtocols))
|
||
for _, protocol := range aiProtocols {
|
||
protocols = append(protocols, map[string]string{"value": protocol, "label": aiProtocolName(protocol)})
|
||
}
|
||
reply(w, map[string]any{
|
||
"items": items,
|
||
"protocols": protocols,
|
||
"enabled": a.configBool(r.Context(), "ai.enabled", false),
|
||
"secretMask": maskedSecret,
|
||
})
|
||
}
|
||
|
||
func (a *App) normalizeAIModelRequest(req *aiModelRequest, existing *aiModel) error {
|
||
req.Name = strings.TrimSpace(req.Name)
|
||
req.Protocol = strings.ToLower(strings.TrimSpace(req.Protocol))
|
||
req.BaseURL = strings.TrimSpace(req.BaseURL)
|
||
req.ModelName = strings.TrimSpace(req.ModelName)
|
||
if existing != nil {
|
||
if req.Name == "" {
|
||
req.Name = existing.Name
|
||
}
|
||
if req.Protocol == "" {
|
||
req.Protocol = existing.Protocol
|
||
}
|
||
}
|
||
if req.Name == "" || len([]rune(req.Name)) > 60 {
|
||
return fmt.Errorf("模型名称须为 1–60 字")
|
||
}
|
||
if !containsString(aiProtocols, req.Protocol) {
|
||
return fmt.Errorf("不支持的模型协议")
|
||
}
|
||
if a.config.Environment == "production" && req.Protocol == "debug" {
|
||
return fmt.Errorf("生产环境禁止使用本地调试模型")
|
||
}
|
||
if len(req.BaseURL) > 300 || len([]rune(req.ModelName)) > 120 {
|
||
return fmt.Errorf("接口地址最多 300 字，模型名称最多 120 字")
|
||
}
|
||
if _, err := parseAIBaseURL(req.Protocol, req.BaseURL, a.config.Environment == "production"); err != nil {
|
||
return err
|
||
}
|
||
if req.Protocol != "debug" && req.Protocol != "webhook" && req.ModelName == "" {
|
||
return fmt.Errorf("请填写模型名称，例如 gpt-4o-mini 或 deepseek-chat")
|
||
}
|
||
if req.Temperature != nil && (*req.Temperature < 0 || *req.Temperature > 2) {
|
||
return fmt.Errorf("温度必须在 0 到 2 之间")
|
||
}
|
||
if req.MaxTokens != nil && (*req.MaxTokens < 16 || *req.MaxTokens > 4096) {
|
||
return fmt.Errorf("最大回复长度必须在 16 到 4096 之间")
|
||
}
|
||
if req.TimeoutMS != nil && (*req.TimeoutMS < 1000 || *req.TimeoutMS > 60000) {
|
||
return fmt.Errorf("超时必须在 1000 到 60000 毫秒之间")
|
||
}
|
||
if req.Status != nil && *req.Status != 0 && *req.Status != 1 {
|
||
return fmt.Errorf("状态无效")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (a *App) adminCreateAIModel(w http.ResponseWriter, r *http.Request) {
|
||
var req aiModelRequest
|
||
if decode(r, &req) != nil {
|
||
fail(w, http.StatusBadRequest, 20001, "模型配置格式不正确")
|
||
return
|
||
}
|
||
if err := a.normalizeAIModelRequest(&req, nil); err != nil {
|
||
fail(w, http.StatusBadRequest, 20001, err.Error())
|
||
return
|
||
}
|
||
cipher, err := a.encryptSecret(strings.TrimSpace(req.APIKey))
|
||
if err != nil {
|
||
fail(w, http.StatusInternalServerError, 50001, "加密模型密钥失败")
|
||
return
|
||
}
|
||
temperature, maxTokens, timeout, status := 0.8, 256, 15000, 1
|
||
if req.Temperature != nil {
|
||
temperature = *req.Temperature
|
||
}
|
||
if req.MaxTokens != nil {
|
||
maxTokens = *req.MaxTokens
|
||
}
|
||
if req.TimeoutMS != nil {
|
||
timeout = *req.TimeoutMS
|
||
}
|
||
if req.Status != nil {
|
||
status = *req.Status
|
||
}
|
||
isDefault := req.IsDefault != nil && *req.IsDefault
|
||
tx, err := a.db.BeginTx(r.Context(), nil)
|
||
if err != nil {
|
||
fail(w, http.StatusInternalServerError, 50001, "保存模型失败")
|
||
return
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
if isDefault {
|
||
if _, err = tx.ExecContext(r.Context(), `UPDATE ai_models SET is_default=0 WHERE is_default=1`); err != nil {
|
||
fail(w, http.StatusInternalServerError, 50001, "保存模型失败")
|
||
return
|
||
}
|
||
}
|
||
result, err := tx.ExecContext(r.Context(), `INSERT INTO ai_models
|
||
(name,protocol,base_url,model_name,api_key_cipher,temperature,max_tokens,timeout_ms,status,is_default)
|
||
VALUES(?,?,?,?,?,?,?,?,?,?)`,
|
||
req.Name, req.Protocol, req.BaseURL, req.ModelName, cipher, temperature, maxTokens, timeout, status, btoi(isDefault))
|
||
if err != nil {
|
||
if isDuplicateAIModelName(err) {
|
||
fail(w, http.StatusBadRequest, 20001, "模型名称已存在")
|
||
return
|
||
}
|
||
fail(w, http.StatusInternalServerError, 50001, "保存模型失败")
|
||
return
|
||
}
|
||
id, _ := result.LastInsertId()
|
||
if err = tx.Commit(); err != nil {
|
||
fail(w, http.StatusInternalServerError, 50001, "保存模型失败")
|
||
return
|
||
}
|
||
a.audit(r, "create", "ai_model", id, map[string]any{"name": req.Name, "protocol": req.Protocol})
|
||
reply(w, map[string]any{"id": id})
|
||
}
|
||
|
||
func (a *App) adminUpdateAIModel(w http.ResponseWriter, r *http.Request) {
|
||
id, err := pathID(r)
|
||
if err != nil {
|
||
fail(w, http.StatusBadRequest, 20001, "模型不存在")
|
||
return
|
||
}
|
||
existing, err := a.loadAIModel(r.Context(), id)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
fail(w, http.StatusNotFound, 30001, "模型不存在")
|
||
return
|
||
}
|
||
var req aiModelRequest
|
||
if decode(r, &req) != nil {
|
||
fail(w, http.StatusBadRequest, 20001, "模型配置格式不正确")
|
||
return
|
||
}
|
||
if err = a.normalizeAIModelRequest(&req, &existing); err != nil {
|
||
fail(w, http.StatusBadRequest, 20001, err.Error())
|
||
return
|
||
}
|
||
// An empty key means "keep the stored one"; clearing is explicit.
|
||
cipherClause, cipherArgs := "", []any{}
|
||
if req.ClearAPIKey {
|
||
cipherClause = ",api_key_cipher=''"
|
||
} else if strings.TrimSpace(req.APIKey) != "" {
|
||
cipher, encryptErr := a.encryptSecret(strings.TrimSpace(req.APIKey))
|
||
if encryptErr != nil {
|
||
fail(w, http.StatusInternalServerError, 50001, "加密模型密钥失败")
|
||
return
|
||
}
|
||
cipherClause = ",api_key_cipher=?"
|
||
cipherArgs = append(cipherArgs, cipher)
|
||
}
|
||
temperature, maxTokens, timeout, status := existing.Temperature, existing.MaxTokens, existing.TimeoutMS, existing.Status
|
||
if req.Temperature != nil {
|
||
temperature = *req.Temperature
|
||
}
|
||
if req.MaxTokens != nil {
|
||
maxTokens = *req.MaxTokens
|
||
}
|
||
if req.TimeoutMS != nil {
|
||
timeout = *req.TimeoutMS
|
||
}
|
||
if req.Status != nil {
|
||
status = *req.Status
|
||
}
|
||
isDefault := existing.IsDefault
|
||
if req.IsDefault != nil {
|
||
isDefault = *req.IsDefault
|
||
}
|
||
tx, err := a.db.BeginTx(r.Context(), nil)
|
||
if err != nil {
|
||
fail(w, http.StatusInternalServerError, 50001, "保存模型失败")
|
||
return
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
if isDefault {
|
||
if _, err = tx.ExecContext(r.Context(), `UPDATE ai_models SET is_default=0 WHERE is_default=1 AND id<>?`, id); err != nil {
|
||
fail(w, http.StatusInternalServerError, 50001, "保存模型失败")
|
||
return
|
||
}
|
||
}
|
||
args := append([]any{req.Name, req.Protocol, req.BaseURL, req.ModelName, temperature, maxTokens, timeout, status, btoi(isDefault)}, cipherArgs...)
|
||
args = append(args, id)
|
||
if _, err = tx.ExecContext(r.Context(), `UPDATE ai_models SET name=?,protocol=?,base_url=?,model_name=?,temperature=?,max_tokens=?,timeout_ms=?,status=?,is_default=?`+cipherClause+` WHERE id=?`, args...); err != nil {
|
||
if isDuplicateAIModelName(err) {
|
||
fail(w, http.StatusBadRequest, 20001, "模型名称已存在")
|
||
return
|
||
}
|
||
fail(w, http.StatusInternalServerError, 50001, "保存模型失败")
|
||
return
|
||
}
|
||
if err = tx.Commit(); err != nil {
|
||
fail(w, http.StatusInternalServerError, 50001, "保存模型失败")
|
||
return
|
||
}
|
||
a.audit(r, "update", "ai_model", id, map[string]any{"name": req.Name, "protocol": req.Protocol, "status": status})
|
||
reply(w, map[string]any{"success": true})
|
||
}
|
||
|
||
func (a *App) adminDeleteAIModel(w http.ResponseWriter, r *http.Request) {
|
||
id, err := pathID(r)
|
||
if err != nil {
|
||
fail(w, http.StatusBadRequest, 20001, "模型不存在")
|
||
return
|
||
}
|
||
if _, err = a.db.ExecContext(r.Context(), `DELETE FROM ai_models WHERE id=?`, id); err != nil {
|
||
fail(w, http.StatusInternalServerError, 50001, "删除模型失败")
|
||
return
|
||
}
|
||
a.audit(r, "delete", "ai_model", id, nil)
|
||
reply(w, map[string]any{"success": true})
|
||
}
|
||
|
||
// A probe call from the console: the operator sees latency, tokens and the
|
||
// actual reply before any conversation depends on the configuration.
|
||
func (a *App) adminTestAIModel(w http.ResponseWriter, r *http.Request) {
|
||
id, err := pathID(r)
|
||
if err != nil {
|
||
fail(w, http.StatusBadRequest, 20001, "模型不存在")
|
||
return
|
||
}
|
||
model, err := a.loadAIModel(r.Context(), id)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
fail(w, http.StatusNotFound, 30001, "模型不存在")
|
||
return
|
||
}
|
||
if err != nil {
|
||
fail(w, http.StatusBadRequest, 20001, err.Error())
|
||
return
|
||
}
|
||
provider, err := newAIProvider(model, a.config.Environment == "production")
|
||
if err != nil {
|
||
fail(w, http.StatusBadRequest, 20001, err.Error())
|
||
return
|
||
}
|
||
request := aiChatRequest{
|
||
System: "你在参与一次接口连通性测试，请直接给出简短的中文回复。",
|
||
Messages: []aiChatMessage{{Role: "user", Text: aiProbePrompt}},
|
||
}
|
||
started := time.Now()
|
||
result, err := provider.Chat(r.Context(), request)
|
||
latency := int(time.Since(started).Milliseconds())
|
||
a.logAICall(r.Context(), model.ID, "probe", 0, 0, latency, result, err)
|
||
if err != nil {
|
||
fail(w, http.StatusBadGateway, 50002, err.Error())
|
||
return
|
||
}
|
||
a.audit(r, "test", "ai_model", id, map[string]any{"latencyMs": latency})
|
||
reply(w, map[string]any{
|
||
"reply": result.Text,
|
||
"latencyMs": latency,
|
||
"promptTokens": result.PromptTokens,
|
||
"completionTokens": result.CompletionTokens,
|
||
"finishReason": result.FinishReason,
|
||
})
|
||
}
|
||
|
||
func (a *App) logAICall(ctx context.Context, modelID int64, scene string, agentUserID, conversationID int64, latencyMS int, result aiChatResult, callErr error) {
|
||
status, message := 1, ""
|
||
if callErr != nil {
|
||
status = 0
|
||
message = callErr.Error()
|
||
if len([]rune(message)) > 500 {
|
||
message = string([]rune(message)[:500])
|
||
}
|
||
}
|
||
text := result.Text
|
||
if len([]rune(text)) > 1000 {
|
||
text = string([]rune(text)[:1000])
|
||
}
|
||
digest := ""
|
||
if scene != "" {
|
||
sum := sha256.Sum256([]byte(fmt.Sprintf("%s:%d:%d", scene, agentUserID, conversationID)))
|
||
digest = hex.EncodeToString(sum[:])[:32]
|
||
}
|
||
_, _ = a.db.ExecContext(ctx, `INSERT INTO ai_call_logs
|
||
(model_id,scene,agent_user_id,conversation_id,latency_ms,prompt_tokens,completion_tokens,status,error,prompt_digest,reply_text)
|
||
VALUES(?,?,?,?,?,?,?,?,?,?,?)`,
|
||
modelID, scene, agentUserID, conversationID, latencyMS, result.PromptTokens, result.CompletionTokens, status, message, digest, text)
|
||
}
|
||
|
||
func isDuplicateAIModelName(err error) bool {
|
||
var mysqlErr *mysql.MySQLError
|
||
return errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 && strings.Contains(mysqlErr.Message, "uk_ai_models_name")
|
||
}
|
||
|
||
// Global switches are validated together because several of them only make
|
||
// sense as a pair, and the console tests them before the worker ever runs.
|
||
func (a *App) validateAIConfig(ctx context.Context) error {
|
||
if !a.configBool(ctx, "ai.enabled", false) {
|
||
return fmt.Errorf("AI 托管当前未启用")
|
||
}
|
||
minDelay := a.configInt(ctx, "ai.reply_delay_min_ms", 2000)
|
||
maxDelay := a.configInt(ctx, "ai.reply_delay_max_ms", 6000)
|
||
if minDelay < 0 || maxDelay < minDelay || maxDelay > 120000 {
|
||
return fmt.Errorf("回复延迟区间无效，最大延迟须不小于最小延迟且不超过 120000 毫秒")
|
||
}
|
||
if concurrency := a.configInt(ctx, "ai.max_concurrency", 4); concurrency < 1 || concurrency > 64 {
|
||
return fmt.Errorf("并发调用上限必须在 1 到 64 之间")
|
||
}
|
||
if limit := a.configInt(ctx, "ai.daily_call_limit", 0); limit < 0 {
|
||
return fmt.Errorf("每日调用上限不能为负数")
|
||
}
|
||
if limit := a.configInt(ctx, "ai.agent_daily_reply_limit", 0); limit < 0 {
|
||
return fmt.Errorf("单账号每日回复上限不能为负数")
|
||
}
|
||
if days := a.configInt(ctx, "ai.log_retention_days", 30); days < 1 || days > 365 {
|
||
return fmt.Errorf("调用日志保留天数必须在 1 到 365 之间")
|
||
}
|
||
for _, batch := range strings.Split(a.configPlain(ctx, "ai.allow_batches", ""), ",") {
|
||
if len(strings.TrimSpace(batch)) > 64 {
|
||
return fmt.Errorf("测试批次名称长度不能超过 64 位")
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (a *App) configInt(ctx context.Context, key string, fallback int) int {
|
||
value, err := strconv.Atoi(strings.TrimSpace(a.configPlain(ctx, key, "")))
|
||
if err != nil {
|
||
return fallback
|
||
}
|
||
return value
|
||
}
|