jackyu66gitandCursor
89756f65b4
落地输入合规、探索题库、报告日/时辰刷新、账号头像、OEJTS 量表，并补齐 H5 埋点与 Admin 漏斗；同步 ESS 工件、切至自建 Git、清理 GitHub Actions。 Co-authored-by: Cursor <cursoragent@cursor.com>
526 lines
14 KiB
Go
526 lines
14 KiB
Go
package ask
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"log"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/google/uuid"
|
||
|
||
eng "github.com/yuxingu/digital-psychology/apps/api/internal/ask"
|
||
"github.com/yuxingu/digital-psychology/apps/api/internal/llm/deepseek"
|
||
"github.com/yuxingu/digital-psychology/apps/api/internal/model"
|
||
"github.com/yuxingu/digital-psychology/apps/api/internal/repository"
|
||
"github.com/yuxingu/digital-psychology/apps/api/internal/textsafe"
|
||
)
|
||
|
||
// FreeQuota is the number of assistant replies allowed without membership.
|
||
const FreeQuota = 3
|
||
|
||
const historyLimit = 10
|
||
|
||
// Service handles ask threads, quota, and replies.
|
||
type Service struct {
|
||
Profiles *repository.ProfileRepo
|
||
Reports *repository.ReportRepo
|
||
Ask *repository.AskRepo
|
||
LLM *deepseek.Client // optional; nil or disabled → rule engine
|
||
}
|
||
|
||
// CreateThreadInput for POST /ask/threads.
|
||
type CreateThreadInput struct {
|
||
ProfileID uuid.UUID
|
||
Scene string
|
||
}
|
||
|
||
// CreateThread binds a conversation to an owned profile.
|
||
func (s *Service) CreateThread(ctx context.Context, userID uuid.UUID, in CreateThreadInput) (*model.AskThread, error) {
|
||
if _, err := s.Profiles.GetForUser(ctx, userID, in.ProfileID); err != nil {
|
||
return nil, errors.New("profile not found")
|
||
}
|
||
var scene *string
|
||
if strings.TrimSpace(in.Scene) != "" {
|
||
sc, err := textsafe.Check(textsafe.Scene, in.Scene)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
scene = &sc
|
||
}
|
||
return s.Ask.CreateThread(ctx, userID, in.ProfileID, scene)
|
||
}
|
||
|
||
// ClearThread soft-deletes an owned thread and its messages.
|
||
func (s *Service) ClearThread(ctx context.Context, userID, threadID uuid.UUID) error {
|
||
return s.Ask.SoftDeleteThread(ctx, userID, threadID)
|
||
}
|
||
|
||
// QuotaStatus describes remaining ask allowance.
|
||
type QuotaStatus struct {
|
||
ActiveMembership bool `json:"active_membership"`
|
||
Remaining int `json:"remaining"`
|
||
FreeLimit int `json:"free_limit"`
|
||
PaidLeft int `json:"paid_left"`
|
||
Source string `json:"source"` // membership | free | paid | mixed
|
||
}
|
||
|
||
// GetQuota returns remaining ask replies.
|
||
func (s *Service) GetQuota(ctx context.Context, userID uuid.UUID) (*QuotaStatus, error) {
|
||
paid, err := s.Ask.GetAskPaidQuota(ctx, userID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
vip, err := s.Reports.HasActiveMembership(ctx, userID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if vip {
|
||
me, err := s.Reports.GetMembership(ctx, userID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
src := "membership"
|
||
if paid > 0 && me.AskQuotaLeft > 0 {
|
||
src = "mixed"
|
||
} else if me.AskQuotaLeft <= 0 && paid > 0 {
|
||
src = "paid"
|
||
}
|
||
return &QuotaStatus{
|
||
ActiveMembership: true,
|
||
Remaining: me.AskQuotaLeft + paid,
|
||
FreeLimit: FreeQuota,
|
||
PaidLeft: paid,
|
||
Source: src,
|
||
}, nil
|
||
}
|
||
used, err := s.Ask.CountUserAssistantMessages(ctx, userID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
// Free tier only counts assistant replies that were not covered by paid packs.
|
||
// Approximate: free used = min(used, FreeQuota) when no paid history is tracked separately.
|
||
// Paid replies decrement ask_paid_quota_left; free replies increase assistant count.
|
||
// Remaining free = max(0, FreeQuota - max(0, used - lifetimePaidConsumed)).
|
||
// Without lifetime paid consumed counter, treat free as: FreeQuota - used, floored at 0,
|
||
// and add current paid left (purchased top-ups work after free exhausted).
|
||
left := FreeQuota - used
|
||
if left < 0 {
|
||
left = 0
|
||
}
|
||
src := "free"
|
||
if left == 0 && paid > 0 {
|
||
src = "paid"
|
||
} else if left > 0 && paid > 0 {
|
||
src = "mixed"
|
||
}
|
||
return &QuotaStatus{
|
||
ActiveMembership: false,
|
||
Remaining: left + paid,
|
||
FreeLimit: FreeQuota,
|
||
PaidLeft: paid,
|
||
Source: src,
|
||
}, nil
|
||
}
|
||
|
||
// ListMessages returns history for an owned thread.
|
||
func (s *Service) ListMessages(ctx context.Context, userID, threadID uuid.UUID) ([]model.AskMessage, error) {
|
||
if _, err := s.Ask.GetThreadForUser(ctx, userID, threadID); err != nil {
|
||
return nil, errors.New("thread not found")
|
||
}
|
||
return s.Ask.ListMessages(ctx, threadID)
|
||
}
|
||
|
||
// SendResult is the user+assistant turn after a send.
|
||
type SendResult struct {
|
||
UserMessage *model.AskMessage `json:"user_message"`
|
||
AssistantMessage *model.AskMessage `json:"assistant_message"`
|
||
Quota *QuotaStatus `json:"quota"`
|
||
}
|
||
|
||
// SendMessage stores user content, consumes quota, generates assistant reply.
|
||
func (s *Service) SendMessage(ctx context.Context, userID, threadID uuid.UUID, content string) (*SendResult, error) {
|
||
var err error
|
||
content, err = textsafe.Check(textsafe.AskContent, content)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
thread, err := s.Ask.GetThreadForUser(ctx, userID, threadID)
|
||
if err != nil {
|
||
return nil, errors.New("thread not found")
|
||
}
|
||
profile, err := s.Profiles.GetForUser(ctx, userID, thread.ProfileID)
|
||
if err != nil {
|
||
return nil, errors.New("profile not found")
|
||
}
|
||
|
||
quota, err := s.GetQuota(ctx, userID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if quota.Remaining <= 0 {
|
||
return nil, ErrQuotaExhausted
|
||
}
|
||
|
||
bucket, err := s.pickQuotaBucket(ctx, userID, quota)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
userMsg, err := s.Ask.InsertMessage(ctx, threadID, "user", content)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
scene := ""
|
||
if thread.Scene != nil {
|
||
scene = *thread.Scene
|
||
}
|
||
|
||
hist, _ := s.Ask.ListMessages(ctx, threadID)
|
||
reply := s.generateReply(ctx, userID, profile, scene, content, hist)
|
||
|
||
asst, err := s.Ask.InsertMessage(ctx, threadID, "assistant", reply)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
if err := s.consumeQuotaBucket(ctx, userID, bucket); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
q2, err := s.GetQuota(ctx, userID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &SendResult{UserMessage: userMsg, AssistantMessage: asst, Quota: q2}, nil
|
||
}
|
||
|
||
type quotaBucket string
|
||
|
||
const (
|
||
bucketFree quotaBucket = "free"
|
||
bucketMembership quotaBucket = "membership"
|
||
bucketPaid quotaBucket = "paid"
|
||
)
|
||
|
||
func (s *Service) pickQuotaBucket(ctx context.Context, userID uuid.UUID, quota *QuotaStatus) (quotaBucket, error) {
|
||
if quota.ActiveMembership {
|
||
me, err := s.Reports.GetMembership(ctx, userID)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
if me.AskQuotaLeft > 0 {
|
||
return bucketMembership, nil
|
||
}
|
||
if quota.PaidLeft > 0 {
|
||
return bucketPaid, nil
|
||
}
|
||
return "", ErrQuotaExhausted
|
||
}
|
||
used, err := s.Ask.CountUserAssistantMessages(ctx, userID)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
if FreeQuota-used > 0 {
|
||
return bucketFree, nil
|
||
}
|
||
if quota.PaidLeft > 0 {
|
||
return bucketPaid, nil
|
||
}
|
||
return "", ErrQuotaExhausted
|
||
}
|
||
|
||
func (s *Service) consumeQuotaBucket(ctx context.Context, userID uuid.UUID, bucket quotaBucket) error {
|
||
switch bucket {
|
||
case bucketFree:
|
||
return nil // counted by assistant message total
|
||
case bucketMembership:
|
||
ok, _, err := s.Ask.ConsumeMembershipQuota(ctx, userID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !ok {
|
||
// race: fall through to paid if possible
|
||
ok2, _, err2 := s.Ask.ConsumeAskPaidQuota(ctx, userID)
|
||
if err2 != nil {
|
||
return err2
|
||
}
|
||
if !ok2 {
|
||
return ErrQuotaExhausted
|
||
}
|
||
}
|
||
return nil
|
||
case bucketPaid:
|
||
ok, _, err := s.Ask.ConsumeAskPaidQuota(ctx, userID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !ok {
|
||
return ErrQuotaExhausted
|
||
}
|
||
return nil
|
||
default:
|
||
return ErrQuotaExhausted
|
||
}
|
||
}
|
||
|
||
func (s *Service) generateReply(ctx context.Context, userID uuid.UUID, profile *model.Profile, scene, userContent string, hist []model.AskMessage) string {
|
||
out, err := s.generateReplyStream(ctx, userID, profile, scene, userContent, hist, nil)
|
||
if err != nil {
|
||
log.Printf("ask: generateReply: %v", err)
|
||
}
|
||
return out
|
||
}
|
||
|
||
// StreamEmit writes one SSE-style event to the client.
|
||
type StreamEmit func(event string, payload any) error
|
||
|
||
// StreamMessage is like SendMessage but streams assistant deltas via emit.
|
||
func (s *Service) StreamMessage(ctx context.Context, userID, threadID uuid.UUID, content string, emit StreamEmit) error {
|
||
if emit == nil {
|
||
return errors.New("emit required")
|
||
}
|
||
var err error
|
||
content, err = textsafe.Check(textsafe.AskContent, content)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
thread, err := s.Ask.GetThreadForUser(ctx, userID, threadID)
|
||
if err != nil {
|
||
return errors.New("thread not found")
|
||
}
|
||
profile, err := s.Profiles.GetForUser(ctx, userID, thread.ProfileID)
|
||
if err != nil {
|
||
return errors.New("profile not found")
|
||
}
|
||
|
||
quota, err := s.GetQuota(ctx, userID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if quota.Remaining <= 0 {
|
||
return ErrQuotaExhausted
|
||
}
|
||
|
||
bucket, err := s.pickQuotaBucket(ctx, userID, quota)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
userMsg, err := s.Ask.InsertMessage(ctx, threadID, "user", content)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if err := emit("meta", map[string]any{"user_message": userMsg}); err != nil {
|
||
return err
|
||
}
|
||
|
||
scene := ""
|
||
if thread.Scene != nil {
|
||
scene = *thread.Scene
|
||
}
|
||
hist, _ := s.Ask.ListMessages(ctx, threadID)
|
||
|
||
reply, err := s.generateReplyStream(ctx, userID, profile, scene, content, hist, func(delta string) error {
|
||
return emit("delta", map[string]any{"text": delta})
|
||
})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if strings.TrimSpace(reply) == "" {
|
||
reply = "暂时没能生成回复，请稍后再试。"
|
||
_ = emit("delta", map[string]any{"text": reply})
|
||
}
|
||
|
||
asst, err := s.Ask.InsertMessage(ctx, threadID, "assistant", reply)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if err := s.consumeQuotaBucket(ctx, userID, bucket); err != nil {
|
||
return err
|
||
}
|
||
q2, err := s.GetQuota(ctx, userID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return emit("done", map[string]any{
|
||
"assistant_message": asst,
|
||
"quota": q2,
|
||
})
|
||
}
|
||
|
||
func (s *Service) generateReplyStream(
|
||
ctx context.Context,
|
||
userID uuid.UUID,
|
||
profile *model.Profile,
|
||
scene, userContent string,
|
||
hist []model.AskMessage,
|
||
onDelta func(string) error,
|
||
) (string, error) {
|
||
fallback := eng.BuildReply(eng.ReplyInput{
|
||
DisplayName: profile.DisplayName,
|
||
BirthDate: profile.BirthDate,
|
||
Relation: profile.Relation,
|
||
Scene: scene,
|
||
UserMessage: userContent,
|
||
})
|
||
emitFallback := func(text string) (string, error) {
|
||
if onDelta == nil {
|
||
return text, nil
|
||
}
|
||
return text, streamFake(ctx, text, onDelta)
|
||
}
|
||
|
||
if s.LLM == nil || !s.LLM.Enabled() {
|
||
return emitFallback(fallback)
|
||
}
|
||
|
||
msgs := []deepseek.Message{{Role: "system", Content: systemPrompt(profile, scene, s.profileContext(ctx, userID, profile))}}
|
||
start := 0
|
||
if len(hist) > historyLimit*2 {
|
||
start = len(hist) - historyLimit*2
|
||
}
|
||
for _, m := range hist[start:] {
|
||
role := m.Role
|
||
if role != "user" && role != "assistant" {
|
||
continue
|
||
}
|
||
msgs = append(msgs, deepseek.Message{Role: role, Content: m.Content})
|
||
}
|
||
|
||
var assembled strings.Builder
|
||
out, err := s.LLM.ChatStream(ctx, msgs, func(delta string) error {
|
||
assembled.WriteString(delta)
|
||
if onDelta != nil {
|
||
return onDelta(delta)
|
||
}
|
||
return nil
|
||
})
|
||
if err != nil {
|
||
if assembled.Len() > 0 {
|
||
return assembled.String(), nil
|
||
}
|
||
log.Printf("ask: deepseek stream failed, fallback to rules: %v", err)
|
||
return emitFallback(fallback)
|
||
}
|
||
if out == "" {
|
||
out = assembled.String()
|
||
}
|
||
if onDelta != nil && assembled.Len() == 0 && out != "" {
|
||
_ = streamFake(ctx, out, onDelta)
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// streamFake chunks text for rule-engine replies so UI still feels streamed.
|
||
func streamFake(ctx context.Context, text string, onDelta func(string) error) error {
|
||
runes := []rune(text)
|
||
const chunk = 2
|
||
for i := 0; i < len(runes); i += chunk {
|
||
select {
|
||
case <-ctx.Done():
|
||
return ctx.Err()
|
||
default:
|
||
}
|
||
end := i + chunk
|
||
if end > len(runes) {
|
||
end = len(runes)
|
||
}
|
||
if err := onDelta(string(runes[i:end])); err != nil {
|
||
return err
|
||
}
|
||
time.Sleep(18 * time.Millisecond)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *Service) profileContext(ctx context.Context, userID uuid.UUID, profile *model.Profile) string {
|
||
if s.Reports == nil || profile == nil {
|
||
return ""
|
||
}
|
||
var parts []string
|
||
for _, typ := range []string{"portrait", "star", "rhythm"} {
|
||
rep, err := s.Reports.GetLatest(ctx, userID, profile.ID, typ, nil)
|
||
if err != nil || rep == nil || len(rep.Summary) == 0 {
|
||
continue
|
||
}
|
||
var sum map[string]any
|
||
if json.Unmarshal(rep.Summary, &sum) != nil {
|
||
continue
|
||
}
|
||
label := map[string]string{"portrait": "愈心解码", "star": "星座探索", "rhythm": "身心节律"}[typ]
|
||
line := strings.TrimSpace(fmt.Sprintf("%s｜%v｜%v",
|
||
label, sum["headline"], sum["one_liner"]))
|
||
if kw, ok := sum["keywords"].([]any); ok && len(kw) > 0 {
|
||
var ks []string
|
||
for i, k := range kw {
|
||
if i >= 5 {
|
||
break
|
||
}
|
||
ks = append(ks, fmt.Sprint(k))
|
||
}
|
||
if len(ks) > 0 {
|
||
line += "｜关键词：" + strings.Join(ks, "、")
|
||
}
|
||
}
|
||
parts = append(parts, line)
|
||
}
|
||
return strings.Join(parts, "\n")
|
||
}
|
||
|
||
func systemPrompt(profile *model.Profile, scene, reportCtx string) string {
|
||
name := profile.DisplayName
|
||
if name == "" {
|
||
if profile.Relation == "other" {
|
||
name = "TA"
|
||
} else {
|
||
name = "你"
|
||
}
|
||
}
|
||
birth := profile.BirthDate.Format("2006-01-02")
|
||
who := "用户本人的个人档案"
|
||
if profile.Relation == "other" {
|
||
who = "用户添加的关系对象（TA）档案"
|
||
}
|
||
sc := strings.TrimSpace(scene)
|
||
if sc == "" {
|
||
sc = "综合成长对话"
|
||
}
|
||
ctxBlock := "（暂无已生成的解码/星座/节律摘要；请主要依据生日与对话内容温和探索，不要假装已经测过完整报告。）"
|
||
if strings.TrimSpace(reportCtx) != "" {
|
||
ctxBlock = reportCtx
|
||
}
|
||
|
||
return fmt.Sprintf(`你是「愈心谷」的 AI 成长助手。
|
||
|
||
【定位】
|
||
把愈心解码、星座、人格匹配、身心节律等探索结果，转成短而可执行的陪伴。你不是占卜师/医生；禁止吉凶断语、改命恐吓、医疗诊断。
|
||
|
||
【当前对象】
|
||
称呼：%s｜档案：%s｜生日：%s｜场景：%s
|
||
摘要（有则引用，无则勿编）：
|
||
%s
|
||
|
||
【回复（务必短）】
|
||
1. 先用 1 句接住情绪/问题。
|
||
2. 只挑与问题最相关的 1 个档案洞察，讲清即可。
|
||
3. 给 1–2 条本周就能做的小行动（可附一句可说出口的话）。
|
||
4. 用中文；全文控制在 80–160 字；不要分大段、不要列表堆砌、不要长篇铺垫。
|
||
5. 不要在文末重复免责声明（界面已展示）。
|
||
|
||
今天是 %s。`, name, who, birth, sc, ctxBlock, time.Now().Format("2006-01-02"))
|
||
}
|
||
|
||
// ErrQuotaExhausted when free or membership ask quota is 0.
|
||
var ErrQuotaExhausted = errors.New("ask quota exhausted")
|
||
|
||
// IsQuotaExhausted reports ErrQuotaExhausted.
|
||
func IsQuotaExhausted(err error) bool {
|
||
return errors.Is(err, ErrQuotaExhausted)
|
||
}
|