feat(runtime): IM 主人私聊补历史转录（与 Web 主会话同源：抽共享 transcript 包 + OwnerEnvFor 单入口）
This commit is contained in:
AskaEth committed
1 parent
df8fc660c0
commit
9b903a89bc
14 files changed
+676
-181
No files matched your search
@@ -0,0 +1,51 @@
|
||||
package transcript
|
||||
|
||||
import (
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
)
|
||||
|
||||
// DocumentPathPrefix 是文档读取的版本化路径前缀。转录还原出的文档部件必须与网关读侧
|
||||
// （`GET /api/v1/documents/{name}`）指向同一个 URL，否则续接的回合会把模型指到一个取不到的
|
||||
// 地址。它是这个契约字符串的**唯一字面量**：网关的 `documentFilePrefix` 直接取自这里。
|
||||
const DocumentPathPrefix = "/api/v1/documents/"
|
||||
|
||||
// documentParts 把已存 document 列里的**全部**引用还原为该回合的文档部件（design/24 §9.5）。
|
||||
// 部件只交给 RoleUser 回合（§2.3-③）：文档在任何别的角色上都会让编码器 fail-loud，而一行
|
||||
// 坏数据不该把整个会话卡死。
|
||||
//
|
||||
// 列里是引用数组（一个回合最多 5 个附件），因此这里逐条还原、保持部件顺序 —— 少还原一条，
|
||||
// 那个附件就会在发起它的回合之后的所有回合里消失，与「清单恒在」的承诺自相矛盾。
|
||||
//
|
||||
// 列被写坏（或附件落在非用户行上）时**不静默**：打一条带来源的 WARN 并省掉该部件，回合照常
|
||||
// 继续 —— 附件凭空消失正是这条清单要修的缺陷，因此必须留下痕迹。日志只带消息 id 与错误，
|
||||
// 绝不带正文。
|
||||
func documentParts(role model.Role, h history.Message) []model.Content {
|
||||
if h.Document == "" {
|
||||
return nil
|
||||
}
|
||||
if role != model.RoleUser {
|
||||
transcriptLog().Warn("transcript document on non-user row", "message", h.ID, "role", h.Role)
|
||||
return nil
|
||||
}
|
||||
refs, err := history.DecodeDocuments(h.Document)
|
||||
if err != nil {
|
||||
transcriptLog().Warn("transcript document decode", "message", h.ID, "err", err)
|
||||
return nil
|
||||
}
|
||||
parts := make([]model.Content, 0, len(refs))
|
||||
for _, ref := range refs {
|
||||
if ref.Name == "" {
|
||||
transcriptLog().Warn("transcript document without name", "message", h.ID)
|
||||
continue
|
||||
}
|
||||
doc := model.Document{
|
||||
Name: ref.Name,
|
||||
MIME: ref.MIME,
|
||||
URL: DocumentPathPrefix + ref.Name,
|
||||
Size: ref.Size,
|
||||
}
|
||||
parts = append(parts, model.DocumentMessage(model.RoleUser, "", doc).Content...)
|
||||
}
|
||||
return parts
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
// Package transcript 拥有「把持久化的聊天转录重建为一个回合的模型上下文」这一件事
|
||||
// （design/15 §2.2/§4 的上下文有界、design/02 §1 主会话承载完整上下文、design/22 §7-5）。
|
||||
//
|
||||
// 两条编排路径都要在回合开始时带上该会话最近的若干条已落库消息：网关的会话编排器
|
||||
// （internal/services/gateway/session_transcript.go）与渠道桥的 IM 回合
|
||||
// （internal/services/channelbridge）。IM 与 Web 只有**呈现形态**不同（IM 只发 chat 文本），
|
||||
// 模型看到的对话历史必须是同一份 —— 两边各写一套映射，迟早会漂移成「消息在同一条会话流
|
||||
// 里，模型看到的历史却不同」，也就是同一个她记得的事不一样。
|
||||
//
|
||||
// 本包只做纯映射：读窄接口 Recent，把 history.Message 翻成 model.Message。**该不该播种、
|
||||
// 播在哪条车道**由调用方按各自的会话与身份边界判定（网关按会话，渠道桥按 private/受限），
|
||||
// 因此本包不持有任何编排状态，也不做身份判断（项目红线：身份在 channel 层的 actor 元数据上）。
|
||||
package transcript
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
)
|
||||
|
||||
// ContextCount 限定一个新回合带入模型上下文的持久化消息数。它与编排器的续跑尾部上限
|
||||
// （网关的 defaultMaxTail）一致，使播种回合与续跑回合保持同样大小；超出之后仍由 loop
|
||||
// 自身的上下文预算治理压缩。**两条车道都从这里取**：这是条数的唯一来源，不许各自另立常量。
|
||||
const ContextCount = 20
|
||||
|
||||
// Reader 是播种所需的窄历史视图。*history.Store 满足它；未满足的存储（测试替身，或
|
||||
// 未接持久化构建的调用方）只是得不到播种上下文，因此回合完全照旧继续运行。
|
||||
type Reader interface {
|
||||
Recent(ctx context.Context, sessionID string, n int) ([]history.Message, error)
|
||||
}
|
||||
|
||||
// Read 返回 sessionID 最近的持久化消息映射成的模型消息（最旧优先）。store 为 nil、
|
||||
// 未实现 Reader 或读取出错都得到 nil，因此聊天绝不被一次元数据查找阻塞；没有历史的
|
||||
// 会话同样什么都不产出 —— 这些情形都退化为既有的「只有本次消息」上下文。
|
||||
func Read(ctx context.Context, store Reader, sessionID string) []model.Message {
|
||||
if store == nil || sessionID == "" {
|
||||
return nil
|
||||
}
|
||||
recent, err := store.Recent(ctx, sessionID, ContextCount)
|
||||
if err != nil {
|
||||
transcriptLog().Debug("turn transcript read", "err", err, "session", sessionID)
|
||||
return nil
|
||||
}
|
||||
out := make([]model.Message, 0, len(recent))
|
||||
for _, h := range recent {
|
||||
out = append(out, Message(h))
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Message 把一条持久化聊天消息映射为回合延续所依据的模型消息。助手消息携带说话的角色
|
||||
// （该回合已存的 persona_id）作为其说话者标签，因此即使会话被切换过，上下文也能把它归属
|
||||
// 到那个角色；非助手消息保持其已存发送者标签（design/16 §4.6 的第三方显示名），绝不
|
||||
// 被重新标为说话角色。文档引用**全部**一并还原（design/24 §9.5）：少了它，附件会在发起
|
||||
// 它的那一回合之后的所有回合里消失。
|
||||
func Message(h history.Message) model.Message {
|
||||
role := RoleOf(h.Role)
|
||||
msg := model.Message{
|
||||
Role: role,
|
||||
Content: []model.Content{{Type: model.ContentTypeText, Text: h.Content}},
|
||||
}
|
||||
if h.Image != "" {
|
||||
msg.Content = append(msg.Content, model.Content{Type: model.ContentTypeImage, ImageURL: h.Image})
|
||||
}
|
||||
msg.Content = append(msg.Content, documentParts(role, h)...)
|
||||
if role == model.RoleAssistant {
|
||||
msg.SenderName = h.PersonaID
|
||||
return msg
|
||||
}
|
||||
msg.SenderName = h.SenderName
|
||||
return msg
|
||||
}
|
||||
|
||||
// RoleOf 把已存聊天角色映射为模型的角色词汇，未知角色默认归到用户通道。
|
||||
func RoleOf(role string) model.Role {
|
||||
switch role {
|
||||
case history.RoleAssistant:
|
||||
return model.RoleAssistant
|
||||
case history.RoleSystem:
|
||||
return model.RoleSystem
|
||||
default:
|
||||
return model.RoleUser
|
||||
}
|
||||
}
|
||||
|
||||
// logSource 是转录重建的日志来源标签。
|
||||
const logSource = "runtime/transcript"
|
||||
|
||||
// transcriptLog 返回标注来源的子日志器。转录重建是纯函数（两条车道的调用点都只拿得到
|
||||
// history.Message，作用域里没有各自编排器的 logger），因此以进程默认 logger 为基底再盖上
|
||||
// 来源标签，使读取失败与坏数据的痕迹既可见又可追溯（与 internal/auth 的 apikey/jwt 同一手法）。
|
||||
// 它只在冷路径上打日志，不在每回合的正常路径上打任何日志，且绝不记消息正文。
|
||||
func transcriptLog() *slog.Logger {
|
||||
return logger.WithSource(slog.Default(), logSource)
|
||||
}
|
||||
@@ -0,0 +1,167 @@
|
||||
package transcript
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
)
|
||||
|
||||
// 本文件钉住共享转录映射的行为（网关与渠道桥的 IM 主人私聊回合都经它读历史）：
|
||||
// 顺序最旧优先、角色翻译、说话角色归属、文档引用全量还原、条数上限取自本包的
|
||||
// ContextCount，以及「读不出来就退化为无转录」的降级口径。
|
||||
|
||||
// fakeReader 是只实现 Recent 的窄历史视图，并记录每次请求的会话与条数上限。
|
||||
type fakeReader struct {
|
||||
rows []history.Message
|
||||
err error
|
||||
session string
|
||||
n int
|
||||
callCount int
|
||||
}
|
||||
|
||||
func (f *fakeReader) Recent(_ context.Context, sessionID string, n int) ([]history.Message, error) {
|
||||
f.session = sessionID
|
||||
f.n = n
|
||||
f.callCount++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
out := make([]history.Message, 0, len(f.rows))
|
||||
for _, m := range f.rows {
|
||||
if m.SessionID == sessionID {
|
||||
out = append(out, m)
|
||||
}
|
||||
}
|
||||
if n <= 0 {
|
||||
return nil, nil
|
||||
}
|
||||
if len(out) > n {
|
||||
out = out[len(out)-n:]
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// textOf 拼接一条模型消息里的文本部件。
|
||||
func textOf(m model.Message) string {
|
||||
out := ""
|
||||
for _, c := range m.Content {
|
||||
if c.Type == model.ContentTypeText {
|
||||
out += c.Text
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func TestReadMapsRecentRowsOldestFirstWithSpeakerAttribution(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
rd := &fakeReader{rows: []history.Message{
|
||||
{ID: "m1", SessionID: "web", Role: history.RoleUser, Content: "喵"},
|
||||
{ID: "m2", SessionID: "web", Role: history.RoleAssistant, Content: "又用喵来敲门啦", PersonaID: "cyrene"},
|
||||
{ID: "m3", SessionID: "web", Role: history.RoleSystem, Content: "系统行"},
|
||||
{ID: "x1", SessionID: "sess-other", Role: history.RoleUser, Content: "别的会话"},
|
||||
}}
|
||||
|
||||
got := Read(context.Background(), rd, "web")
|
||||
if len(got) != 3 {
|
||||
t.Fatalf("messages = %d, want 3 (only the requested session): %+v", len(got), got)
|
||||
}
|
||||
if got[0].Role != model.RoleUser || textOf(got[0]) != "喵" {
|
||||
t.Fatalf("messages[0] = %+v, want the oldest user turn first", got[0])
|
||||
}
|
||||
// 助手话语带上说出它的角色，非助手消息保持已存发送者标签。
|
||||
if got[1].Role != model.RoleAssistant || got[1].SenderName != "cyrene" {
|
||||
t.Fatalf("messages[1] = %+v, want her reply attributed to cyrene", got[1])
|
||||
}
|
||||
if got[2].Role != model.RoleSystem {
|
||||
t.Fatalf("messages[2] = %+v, want the persisted system row", got[2])
|
||||
}
|
||||
}
|
||||
|
||||
func TestReadWindowIsTheSharedContextCount(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
rows := make([]history.Message, 0, ContextCount+5)
|
||||
for i := 0; i < ContextCount+5; i++ {
|
||||
rows = append(rows, history.Message{SessionID: "web", Role: history.RoleUser, Content: "第" + string(rune('a'+i)) + "句"})
|
||||
}
|
||||
rd := &fakeReader{rows: rows}
|
||||
|
||||
got := Read(context.Background(), rd, "web")
|
||||
// 条数上限只有一个来源：本包的 ContextCount（网关与 IM 车道都从这里取，不许各自造数）。
|
||||
if rd.n != ContextCount {
|
||||
t.Fatalf("requested window = %d, want the shared ContextCount (%d)", rd.n, ContextCount)
|
||||
}
|
||||
if len(got) != ContextCount {
|
||||
t.Fatalf("messages = %d, want %d (the newest window)", len(got), ContextCount)
|
||||
}
|
||||
}
|
||||
|
||||
func TestReadDegradesToNil(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
if got := Read(context.Background(), nil, "web"); got != nil {
|
||||
t.Fatalf("nil store = %+v, want nil", got)
|
||||
}
|
||||
rd := &fakeReader{rows: []history.Message{{SessionID: "web", Role: history.RoleUser, Content: "喵"}}}
|
||||
if got := Read(context.Background(), rd, ""); got != nil {
|
||||
t.Fatalf("empty session = %+v, want nil", got)
|
||||
}
|
||||
if rd.callCount != 0 {
|
||||
t.Fatalf("empty session read the store %d times, want 0", rd.callCount)
|
||||
}
|
||||
bad := &fakeReader{err: errors.New("boom")}
|
||||
if got := Read(context.Background(), bad, "web"); got != nil {
|
||||
t.Fatalf("read error = %+v, want nil (the turn must still run)", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMessageRestoresEveryDocumentPartOnUserRows(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
raw, err := history.EncodeDocuments([]history.DocumentRef{
|
||||
{Name: "0123456789abcdef0123456789abcdef.txt", MIME: "text/plain", Size: 12},
|
||||
{Name: "fedcba9876543210fedcba9876543210.pdf", MIME: "application/pdf", Size: 34},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("EncodeDocuments: %v", err)
|
||||
}
|
||||
|
||||
msg := Message(history.Message{ID: "m1", Role: history.RoleUser, Content: "读一下", Document: raw})
|
||||
if len(msg.Content) != 3 {
|
||||
t.Fatalf("parts = %d, want text + 2 documents: %+v", len(msg.Content), msg.Content)
|
||||
}
|
||||
for i, want := range []history.DocumentRef{
|
||||
{Name: "0123456789abcdef0123456789abcdef.txt", MIME: "text/plain", Size: 12},
|
||||
{Name: "fedcba9876543210fedcba9876543210.pdf", MIME: "application/pdf", Size: 34},
|
||||
} {
|
||||
part := msg.Content[i+1]
|
||||
if part.DocumentURL != DocumentPathPrefix+want.Name || part.DocumentMIME != want.MIME {
|
||||
t.Fatalf("document part[%d] = %+v, want %s%s", i, part, DocumentPathPrefix, want.Name)
|
||||
}
|
||||
}
|
||||
// 非用户行上的文档部件会让编码器 fail-loud，因此刻意省掉（坏数据不该卡死会话）。
|
||||
other := Message(history.Message{ID: "m2", Role: history.RoleAssistant, Content: "好", Document: raw})
|
||||
for _, c := range other.Content {
|
||||
if c.DocumentURL != "" {
|
||||
t.Fatalf("assistant row carries a document part: %+v", c)
|
||||
}
|
||||
}
|
||||
// 坏列不 panic、也不产出部件（调用方另行记录 WARN）。
|
||||
broken := Message(history.Message{ID: "m3", Role: history.RoleUser, Content: "坏列", Document: "{not json"})
|
||||
if len(broken.Content) != 1 {
|
||||
t.Fatalf("broken document column produced %d parts, want just the text: %+v", len(broken.Content), broken.Content)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMessageRestoresImages(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
msg := Message(history.Message{ID: "m1", Role: history.RoleUser, Content: "看图", Image: "data:image/png;base64,AAAA"})
|
||||
if len(msg.Content) != 2 || msg.Content[1].ImageURL != "data:image/png;base64,AAAA" {
|
||||
t.Fatalf("image part = %+v, want the stored data URI restored", msg.Content)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user