package factcheck import ( "context" "encoding/json" "errors" "strconv" "strings" "time" platformid "aigateway.local/core/internal/platform/id" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" ) var ErrNotFound = errors.New("fact-check resource not found") type Service struct{ pool *pgxpool.Pool } func NewService(pool *pgxpool.Pool) *Service { return &Service{pool: pool} } type Settings struct { ProviderID *string `json:"provider_id"` ProviderCode string `json:"provider_code"` Model string `json:"model"` TimeoutSeconds int `json:"timeout_seconds"` UpdatedAt time.Time `json:"updated_at"` } func (s *Service) Settings(ctx context.Context) (Settings, error) { var x Settings err := s.pool.QueryRow(ctx, `SELECT f.provider_id::text,coalesce(p.code,''),f.model,f.timeout_seconds,f.updated_at FROM gateway.fact_check_settings f LEFT JOIN gateway.providers p ON p.id=f.provider_id WHERE singleton`).Scan(&x.ProviderID, &x.ProviderCode, &x.Model, &x.TimeoutSeconds, &x.UpdatedAt) return x, err } func (s *Service) SaveSettings(ctx context.Context, x Settings, actor string) (Settings, error) { x.Model = strings.TrimSpace(x.Model) if x.TimeoutSeconds < 3 || x.TimeoutSeconds > 60 { return Settings{}, errors.New("超时必须在 3-60 秒之间") } if x.ProviderID != nil && strings.TrimSpace(*x.ProviderID) == "" { x.ProviderID = nil } if x.ProviderID != nil && x.Model == "" { return Settings{}, errors.New("配置供应商时模型不能为空") } _, err := s.pool.Exec(ctx, `UPDATE gateway.fact_check_settings SET provider_id=$1,model=$2,timeout_seconds=$3,updated_by=$4,updated_at=clock_timestamp() WHERE singleton`, x.ProviderID, x.Model, x.TimeoutSeconds, actor) if err != nil { return Settings{}, err } return s.Settings(ctx) } type Policy struct { ID string `json:"id"` Scope string `json:"scope"` Enabled bool `json:"enabled"` Mode string `json:"mode"` Action string `json:"action"` KnowledgeBaseIDs []string `json:"knowledge_base_ids"` SupportThreshold int `json:"support_threshold"` EvidenceThreshold float64 `json:"evidence_threshold"` TopK int `json:"top_k"` MaxClaims int `json:"max_claims"` Revision int64 `json:"revision"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` } func validatePolicy(x *Policy) error { x.Scope = strings.ToLower(strings.TrimSpace(x.Scope)) if x.Scope == "" { x.Scope = "global" } if x.Mode == "" { x.Mode = "async" } if x.Action == "" { x.Action = "observe" } if x.Mode != "async" && x.Mode != "sync" { return errors.New("模式无效") } if x.Action != "observe" && x.Action != "annotate" && x.Action != "block" { return errors.New("处置动作无效") } if x.Mode == "async" && x.Action != "observe" { return errors.New("异步模式只能观察") } if x.Enabled && len(x.KnowledgeBaseIDs) == 0 { return errors.New("启用策略前至少选择一个知识库") } if x.TopK == 0 { x.TopK = 4 } if x.MaxClaims == 0 { x.MaxClaims = 8 } if x.SupportThreshold == 0 { x.SupportThreshold = 70 } if x.EvidenceThreshold == 0 { x.EvidenceThreshold = .35 } if x.TopK < 1 || x.TopK > 10 || x.MaxClaims < 1 || x.MaxClaims > 20 || x.SupportThreshold < 0 || x.SupportThreshold > 100 || x.EvidenceThreshold < 0 || x.EvidenceThreshold > 1 { return errors.New("事实核验阈值无效") } return nil } const policySelect = `SELECT id::text,scope,enabled,mode,action,knowledge_base_ids::text[],support_threshold,evidence_threshold,top_k,max_claims,revision,created_at,updated_at FROM gateway.fact_check_policies` func scanPolicy(row pgx.Row) (Policy, error) { var x Policy err := row.Scan(&x.ID, &x.Scope, &x.Enabled, &x.Mode, &x.Action, &x.KnowledgeBaseIDs, &x.SupportThreshold, &x.EvidenceThreshold, &x.TopK, &x.MaxClaims, &x.Revision, &x.CreatedAt, &x.UpdatedAt) if errors.Is(err, pgx.ErrNoRows) { err = ErrNotFound } return x, err } func (s *Service) Policies(ctx context.Context) ([]Policy, error) { rows, err := s.pool.Query(ctx, policySelect+` ORDER BY scope`) if err != nil { return nil, err } defer rows.Close() items := []Policy{} for rows.Next() { x, err := scanPolicy(rows) if err != nil { return nil, err } items = append(items, x) } return items, rows.Err() } func (s *Service) SavePolicy(ctx context.Context, x Policy, actor string) (Policy, error) { if err := validatePolicy(&x); err != nil { return Policy{}, err } if x.KnowledgeBaseIDs == nil { x.KnowledgeBaseIDs = []string{} } if x.ID == "" { x.ID, _ = platformid.NewUUID() _, err := s.pool.Exec(ctx, `INSERT INTO gateway.fact_check_policies(id,scope,enabled,mode,action,knowledge_base_ids,support_threshold,evidence_threshold,top_k,max_claims,created_by) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)`, x.ID, x.Scope, x.Enabled, x.Mode, x.Action, x.KnowledgeBaseIDs, x.SupportThreshold, x.EvidenceThreshold, x.TopK, x.MaxClaims, actor) if err != nil { return Policy{}, err } } else { tag, err := s.pool.Exec(ctx, `UPDATE gateway.fact_check_policies SET scope=$2,enabled=$3,mode=$4,action=$5,knowledge_base_ids=$6,support_threshold=$7,evidence_threshold=$8,top_k=$9,max_claims=$10,revision=revision+1,updated_at=clock_timestamp() WHERE id=$1`, x.ID, x.Scope, x.Enabled, x.Mode, x.Action, x.KnowledgeBaseIDs, x.SupportThreshold, x.EvidenceThreshold, x.TopK, x.MaxClaims) if err != nil { return Policy{}, err } if tag.RowsAffected() == 0 { return Policy{}, ErrNotFound } } return scanPolicy(s.pool.QueryRow(ctx, policySelect+` WHERE id=$1`, x.ID)) } func (s *Service) DeletePolicy(ctx context.Context, id string) error { tag, err := s.pool.Exec(ctx, `DELETE FROM gateway.fact_check_policies WHERE id=$1`, id) if err != nil { return err } if tag.RowsAffected() == 0 { return ErrNotFound } return nil } type Event struct { ID string `json:"id"` PolicyID *string `json:"policy_id"` RequestID string `json:"request_id"` Model string `json:"model"` Mode string `json:"mode"` Action string `json:"action"` Verdict string `json:"verdict"` SupportScore *int `json:"support_score"` LatencyMS int `json:"latency_ms"` Question string `json:"question,omitempty"` Answer string `json:"answer,omitempty"` Claims json.RawMessage `json:"claims,omitempty"` Evidence json.RawMessage `json:"evidence,omitempty"` Error string `json:"error"` CreatedAt time.Time `json:"created_at"` } const eventSelect = `SELECT id::text,policy_id::text,request_id,model,mode,action,verdict,support_score,latency_ms,question,answer,claims,evidence,error,created_at FROM gateway.fact_check_events` func scanEvent(row pgx.Row) (Event, error) { var x Event err := row.Scan(&x.ID, &x.PolicyID, &x.RequestID, &x.Model, &x.Mode, &x.Action, &x.Verdict, &x.SupportScore, &x.LatencyMS, &x.Question, &x.Answer, &x.Claims, &x.Evidence, &x.Error, &x.CreatedAt) if errors.Is(err, pgx.ErrNoRows) { err = ErrNotFound } return x, err } func (s *Service) Events(ctx context.Context, verdict string, limit int) ([]Event, error) { if limit < 1 { limit = 50 } if limit > 200 { limit = 200 } query := eventSelect args := []any{} if verdict != "" { args = append(args, verdict) query += ` WHERE verdict=$1` } args = append(args, limit) query += ` ORDER BY created_at DESC LIMIT $` + strconv.Itoa(len(args)) rows, err := s.pool.Query(ctx, query, args...) if err != nil { return nil, err } defer rows.Close() items := []Event{} for rows.Next() { x, err := scanEvent(rows) if err != nil { return nil, err } x.Question = "" x.Answer = "" x.Claims = nil x.Evidence = nil items = append(items, x) } return items, rows.Err() } func (s *Service) Event(ctx context.Context, id string) (Event, error) { return scanEvent(s.pool.QueryRow(ctx, eventSelect+` WHERE id=$1`, id)) }