Version: 0.9.13.dev.260410

后端:
1. Memory Day1 链路打通(chat_history -> outbox -> memory_jobs)
   - 更新 service/events/chat_history_persist.go:聊天消息落库同事务追加 memory.extract.requested 事件(仅 user 消息,失败回滚后由 outbox 重试)
   - 新建 service/events/memory_extract_requested.go:消费 memory.extract.requested 并幂等入队 memory_jobs,补齐 payload 校验、文本截断与 idempotency key
   - 更新 cmd/start.go:注册 RegisterMemoryExtractRequestedHandler
2. Memory 模块骨架落地(先跑通状态机,再接入真实抽取)
   - 新建 memory/model、repo、service、orchestrator、worker、utils 目录与 Day1 mock 抽取执行链
   - 新建 model/memory.go:补齐 memory_items / memory_jobs / memory_audit_logs / memory_user_settings 与事件 payload 模型
   - 更新 inits/mysql.go:接入 4 张 memory 相关表 AutoMigrate
3. RAG 复用基础设施预埋(依赖可替换)
   - 新建 infra/rag:core pipeline + chunk/embed/retrieve/rerank/store/corpus/config 分层实现
   - 默认接入 MockEmbedder + InMemoryStore,预留 Milvus / Eino 适配实现
   - 新增 infra/rag/RAG复用接口实施计划.md
4. 本地依赖与交接文档同步
   - 更新 docker-compose.yml:新增 etcd / minio / milvus / attu 服务与数据卷
   - 删除 newAgent/HANDOFF_工具研究与运行态重置.md、newAgent/阶段3_上下文瘦身设计.md
   - 新增 newAgent/HANDOFF_WebSearch两阶段实施计划.md、memory/HANDOFF-RAG复用后续实施计划.md、memory/README.md
前端:无 仓库:无
This commit is contained in:
LoveLosita
2026-04-10 13:07:54 +08:00
parent ee34d5f111
commit fae162162a
47 changed files with 3244 additions and 1280 deletions

View File

@@ -90,6 +90,9 @@ func Start() {
if err = eventsvc.RegisterAgentStateSnapshotHandler(eventBus, outboxRepo, manager); err != nil {
log.Fatalf("Failed to register agent state snapshot event handler: %v", err)
}
if err = eventsvc.RegisterMemoryExtractRequestedHandler(eventBus, outboxRepo); err != nil {
log.Fatalf("Failed to register memory extract event handler: %v", err)
}
eventBus.Start(context.Background())
defer eventBus.Close()
log.Println("Outbox event bus started")

View File

@@ -0,0 +1,191 @@
# RAG 复用接口实施计划Memory + WebSearch 统一底座)
## 1. 目标与原则
1.`backend/infra/rag` 抽离共享 RAG Core统一 `chunk/embed/retrieve/rerank` 能力。
2. 先接入 `MemoryCorpus``WebCorpus` 两个适配器,避免后续重复造轮子。
3. 保持“并行迁移”策略:新老链路并存,先接入、再灰度、再切流、最后删除旧实现。
4. 不阻塞现有主链路;任何 RAG 子能力失败都必须可降级。
## 2. 本轮范围与非目标
### 2.1 本轮范围
1. 定义 RAG Core 接口、标准数据结构、错误码和回退语义。
2. 提供 `MemoryCorpus``WebCorpus` 适配层设计。
3. 给出分阶段落地步骤、验收标准、风险控制。
### 2.2 本轮非目标
1. 不在本轮实现完整生产级向量检索细节Milvus 连接器可先占位)。
2. 不在本轮统一改造所有调用方,只做首批接入点。
3. 不在本轮引入多 Provider 工厂(先保证单 Provider 可替换)。
## 3. 目录与模块规划
建议目录(先建骨架,逐轮填实):
```text
backend/infra/rag/
core/
types.go
interfaces.go
pipeline.go
errors.go
chunk/
text_chunker.go
embed/
eino_embedder.go
retrieve/
vector_retriever.go
rerank/
eino_reranker.go
store/
vector_store.go
milvus_store.go
corpus/
memory_corpus.go
web_corpus.go
config/
config.go
```
## 4. 核心接口设计(建议签名)
```go
type Chunker interface {
Chunk(ctx context.Context, doc SourceDocument, opt ChunkOption) ([]Chunk, error)
}
type Embedder interface {
Embed(ctx context.Context, texts []string, action string) ([][]float32, error)
}
type Retriever interface {
Retrieve(ctx context.Context, req RetrieveRequest) ([]ScoredChunk, error)
}
type Reranker interface {
Rerank(ctx context.Context, query string, candidates []ScoredChunk, topK int) ([]ScoredChunk, error)
}
type VectorStore interface {
Upsert(ctx context.Context, rows []VectorRow) error
Search(ctx context.Context, req VectorSearchRequest) ([]ScoredVectorRow, error)
Delete(ctx context.Context, ids []string) error
Get(ctx context.Context, ids []string) ([]VectorRow, error)
}
type CorpusAdapter interface {
Name() string
BuildIngestDocuments(ctx context.Context, input any) ([]SourceDocument, error)
BuildRetrieveFilter(ctx context.Context, req any) (map[string]any, error)
}
```
## 5. 统一流程约定
### 5.1 Ingest 流程
1. `CorpusAdapter.BuildIngestDocuments` 生成标准文档。
2. `Chunker.Chunk` 切块(固定 chunk_size + overlap
3. `Embedder.Embed(action=add/update)` 生成向量。
4. `VectorStore.Upsert` 写入。
5. 任一步失败按“可补偿”记录状态,不影响主业务成功返回。
### 5.2 Retrieve 流程
1. `CorpusAdapter.BuildRetrieveFilter` 构建过滤条件。
2. `Embedder.Embed(action=search)` 向量化 query。
3. `VectorStore.Search` 召回候选。
4. `threshold` 过滤。
5. 可选 `Reranker` 重排;失败则 fallback 到原排序并记录原因码。
## 6. 两类 Corpus 适配器设计
### 6.1 MemoryCorpus
1. 数据源:`memory_items`(结构化记忆事实)。
2. 强约束过滤:`user_id + assistant_id + conversation_id`
3. 元数据:`memory_type/confidence/sensitivity_level/ttl_at/source_event_id`
4. 注入优先级:`constraint/preference` 高于 `fact/todo_hint`
### 6.2 WebCorpus
1. 数据源websearch 抓取结果(`url/title/snippet/content`)。
2. 强约束过滤:`query_id/session_id`,避免跨问题污染。
3. 元数据:`domain/published_at/fetched_at/language/source_rank`
4. 检索策略:先向量召回,再结合域名可信度做轻量加权。
## 7. 与 Eino 的集成方式
1. `embed/eino_embedder.go`:封装 Eino embedding 调用。
2. `rerank/eino_reranker.go`:封装 Eino 重排调用。
3. 统一配置入口:`rag.enabled/top_k/threshold/reranker_enabled/timeout`
4. 统一日志字段:`trace_id/corpus/action/fallback_reason/latency_ms/hit_count`
## 8. 分阶段实施(建议 4 轮)
### Round 1基础骨架不切流
1.`infra/rag` 目录与接口、类型、错误码。
2. 提供 `NoopReranker``MockEmbedder` 兜底实现。
3. 验收:编译通过,主链路行为不变。
### Round 2MemoryCorpus 接入(灰度)
1. 把记忆检索从“模块内直连”改为调用 RAG Core。
2. 保留旧路径开关 `memory.rag.enabled`,默认关闭。
3. 验收:开启开关后功能等价,失败可自动降级旧链路。
### Round 3WebCorpus 接入(灰度)
1. websearch 召回改走 RAG Core。
2. 加入 `web.rag.enabled` 灰度开关。
3. 验收:检索可复用同一 pipeline质量不低于旧实现。
### Round 4统一切流与清理
1. 默认开启 RAG Core旧链路保留一段观察窗口。
2. 指标稳定后删除旧实现。
3. 验收:两条业务链路均通过统一接口,文档与监控齐全。
## 9. 配置建议
```yaml
rag:
enabled: true
topK: 8
threshold: 0.55
reranker:
enabled: true
timeoutMs: 1200
ingest:
chunkSize: 400
chunkOverlap: 80
retrieve:
timeoutMs: 1500
```
## 10. 验收标准DoD
1. 同一套 Core 能同时服务 Memory 与 WebSearch。
2. `rerank` 异常时可观测地降级,不影响主功能可用性。
3. 支持按 corpus 维度查看命中率、耗时、降级率。
4. 新老链路可开关切换,回滚路径明确。
## 11. 风险与应对
1. 风险:一次性切流影响面大。
应对:按 corpus 分轮灰度,先 Memory 后 Web。
2. 风险:向量检索延迟波动。
应对:超时控制 + fallback + 本地缓存热点 query。
3. 风险:跨域检索串数据。
应对:强制 filter 校验,不满足维度直接拒绝检索。
## 12. 下一步执行清单(紧接实现)
1. 先补 `core/interfaces.go + core/types.go + core/pipeline.go`
2. 再补 `corpus/memory_corpus.go`(首个适配器)。
3. 然后给 websearch 接 `corpus/web_corpus.go` 占位适配器。
4. 最后补 `store/milvus_store.go` 与配置接线(当前 docker compose 已准备 Milvus 依赖)。

View File

@@ -0,0 +1,85 @@
package chunk
import (
"context"
"fmt"
"strings"
"github.com/LoveLosita/smartflow/backend/infra/rag/core"
)
// TextChunker 是默认文本切块器。
type TextChunker struct{}
func NewTextChunker() *TextChunker {
return &TextChunker{}
}
// Chunk 对文本执行固定窗口切块。
//
// 步骤化说明:
// 1. 先做空白归一,避免无效块进入向量库;
// 2. 再按 chunk_size/overlap 滑窗切割;
// 3. 每块继承原文 metadata并补充 chunk 序号。
func (c *TextChunker) Chunk(_ context.Context, doc core.SourceDocument, opt core.ChunkOption) ([]core.Chunk, error) {
if strings.TrimSpace(doc.ID) == "" {
return nil, fmt.Errorf("empty document id")
}
text := strings.TrimSpace(doc.Text)
if text == "" {
return nil, nil
}
if opt.ChunkSize <= 0 {
opt.ChunkSize = 400
}
if opt.ChunkOverlap < 0 {
opt.ChunkOverlap = 0
}
if opt.ChunkOverlap >= opt.ChunkSize {
opt.ChunkOverlap = opt.ChunkSize / 5
}
runes := []rune(text)
step := opt.ChunkSize - opt.ChunkOverlap
if step <= 0 {
step = opt.ChunkSize
}
result := make([]core.Chunk, 0, len(runes)/step+1)
order := 0
for start := 0; start < len(runes); start += step {
end := start + opt.ChunkSize
if end > len(runes) {
end = len(runes)
}
chunkText := strings.TrimSpace(string(runes[start:end]))
if chunkText == "" {
continue
}
metadata := cloneMap(doc.Metadata)
metadata["chunk_order"] = order
result = append(result, core.Chunk{
ID: fmt.Sprintf("%s#%d", doc.ID, order),
DocumentID: doc.ID,
Text: chunkText,
Order: order,
Metadata: metadata,
})
order++
if end == len(runes) {
break
}
}
return result, nil
}
func cloneMap(src map[string]any) map[string]any {
if len(src) == 0 {
return map[string]any{}
}
dst := make(map[string]any, len(src))
for k, v := range src {
dst[k] = v
}
return dst
}

View File

@@ -0,0 +1,52 @@
package config
import "github.com/spf13/viper"
// Config 是 RAG Core 运行配置。
type Config struct {
Enabled bool
TopK int
Threshold float64
RerankerEnabled bool
RerankerTimeoutMS int
ChunkSize int
ChunkOverlap int
RetrieveTimeoutMS int
}
// LoadFromViper 读取 rag 配置并补默认值。
func LoadFromViper() Config {
cfg := Config{
Enabled: viper.GetBool("rag.enabled"),
TopK: viper.GetInt("rag.topK"),
Threshold: viper.GetFloat64("rag.threshold"),
RerankerEnabled: viper.GetBool("rag.reranker.enabled"),
RerankerTimeoutMS: viper.GetInt("rag.reranker.timeoutMs"),
ChunkSize: viper.GetInt("rag.ingest.chunkSize"),
ChunkOverlap: viper.GetInt("rag.ingest.chunkOverlap"),
RetrieveTimeoutMS: viper.GetInt("rag.retrieve.timeoutMs"),
}
if cfg.TopK <= 0 {
cfg.TopK = 8
}
if cfg.Threshold < 0 {
cfg.Threshold = 0
}
if cfg.RerankerTimeoutMS <= 0 {
cfg.RerankerTimeoutMS = 1200
}
if cfg.ChunkSize <= 0 {
cfg.ChunkSize = 400
}
if cfg.ChunkOverlap < 0 {
cfg.ChunkOverlap = 80
}
if cfg.RetrieveTimeoutMS <= 0 {
cfg.RetrieveTimeoutMS = 1500
}
return cfg
}

View File

@@ -0,0 +1,17 @@
package core
import "errors"
var (
// ErrInvalidQuery 表示检索请求缺少有效 query。
ErrInvalidQuery = errors.New("invalid query")
// ErrInvalidTopK 表示 topK 非法。
ErrInvalidTopK = errors.New("invalid top_k")
// ErrNilDependency 表示 pipeline 关键依赖未注入。
ErrNilDependency = errors.New("nil dependency")
)
const (
// FallbackReasonRerankFailed 表示 rerank 失败后降级。
FallbackReasonRerankFailed = "RERANK_FAILED"
)

View File

@@ -0,0 +1,38 @@
package core
import "context"
// Chunker 负责文本切块。
type Chunker interface {
Chunk(ctx context.Context, doc SourceDocument, opt ChunkOption) ([]Chunk, error)
}
// Embedder 负责向量化。
type Embedder interface {
Embed(ctx context.Context, texts []string, action string) ([][]float32, error)
}
// Retriever 负责召回候选。
type Retriever interface {
Retrieve(ctx context.Context, req RetrieveRequest) ([]ScoredChunk, error)
}
// Reranker 负责重排候选。
type Reranker interface {
Rerank(ctx context.Context, query string, candidates []ScoredChunk, topK int) ([]ScoredChunk, error)
}
// VectorStore 负责向量库读写。
type VectorStore interface {
Upsert(ctx context.Context, rows []VectorRow) error
Search(ctx context.Context, req VectorSearchRequest) ([]ScoredVectorRow, error)
Delete(ctx context.Context, ids []string) error
Get(ctx context.Context, ids []string) ([]VectorRow, error)
}
// CorpusAdapter 负责把业务语料映射成统一文档/过滤条件。
type CorpusAdapter interface {
Name() string
BuildIngestDocuments(ctx context.Context, input any) ([]SourceDocument, error)
BuildRetrieveFilter(ctx context.Context, req any) (map[string]any, error)
}

View File

@@ -0,0 +1,266 @@
package core
import (
"context"
"errors"
"fmt"
"log"
"strings"
"time"
)
const (
defaultTopK = 8
defaultThreshold = 0
defaultChunkSize = 400
defaultChunkOvLap = 80
)
// Pipeline 是 RAG Core 编排器。
//
// 职责边界:
// 1. 负责统一 chunk/embed/retrieve/rerank 流程;
// 2. 负责失败降级语义;
// 3. 不承载任何具体业务语义(由 CorpusAdapter 提供)。
type Pipeline struct {
chunker Chunker
embedder Embedder
store VectorStore
reranker Reranker
logger *log.Logger
}
func NewPipeline(chunker Chunker, embedder Embedder, store VectorStore, reranker Reranker) *Pipeline {
return &Pipeline{
chunker: chunker,
embedder: embedder,
store: store,
reranker: reranker,
logger: log.Default(),
}
}
// Ingest 执行统一入库流程。
//
// 步骤化说明:
// 1. 先由 CorpusAdapter 生成统一文档,确保不同语料入口一致;
// 2. 再统一切块与向量化,避免业务侧重复实现;
// 3. 最后一次性 Upsert失败直接返回交由上层决定是否重试。
func (p *Pipeline) Ingest(
ctx context.Context,
corpus CorpusAdapter,
input any,
opt IngestOption,
) (*IngestResult, error) {
if p == nil || p.chunker == nil || p.embedder == nil || p.store == nil {
return nil, ErrNilDependency
}
if corpus == nil {
return nil, errors.New("nil corpus adapter")
}
docs, err := corpus.BuildIngestDocuments(ctx, input)
if err != nil {
return nil, err
}
if len(docs) == 0 {
return &IngestResult{DocumentCount: 0, ChunkCount: 0}, nil
}
chunkOpt := normalizeChunkOption(opt.Chunk)
chunks := make([]Chunk, 0, len(docs)*2)
for _, doc := range docs {
// 1. 对每个文档独立切块,失败直接中断,避免写入半成品。
docChunks, chunkErr := p.chunker.Chunk(ctx, doc, chunkOpt)
if chunkErr != nil {
return nil, chunkErr
}
chunks = append(chunks, docChunks...)
}
if len(chunks) == 0 {
return &IngestResult{DocumentCount: len(docs), ChunkCount: 0}, nil
}
texts := make([]string, 0, len(chunks))
for _, chunk := range chunks {
texts = append(texts, chunk.Text)
}
action := strings.TrimSpace(opt.Action)
if action == "" {
action = "add"
}
vectors, err := p.embedder.Embed(ctx, texts, action)
if err != nil {
return nil, err
}
if len(vectors) != len(chunks) {
return nil, fmt.Errorf("embedding result length mismatch: chunks=%d vectors=%d", len(chunks), len(vectors))
}
rows := make([]VectorRow, 0, len(chunks))
now := time.Now()
for i, chunk := range chunks {
metadata := cloneMap(chunk.Metadata)
metadata["corpus"] = corpus.Name()
metadata["document_id"] = chunk.DocumentID
metadata["chunk_order"] = chunk.Order
rows = append(rows, VectorRow{
ID: chunk.ID,
Vector: vectors[i],
Text: chunk.Text,
Metadata: metadata,
CreatedAt: now,
UpdatedAt: now,
})
}
if err = p.store.Upsert(ctx, rows); err != nil {
return nil, err
}
return &IngestResult{
DocumentCount: len(docs),
ChunkCount: len(chunks),
}, nil
}
// Retrieve 执行统一检索流程。
//
// 步骤化说明:
// 1. 先做 query 向量化与向量检索;
// 2. 再执行阈值过滤,减少低质量候选;
// 3. 最后可选 rerank若失败则降级回原排序并打日志。
func (p *Pipeline) Retrieve(
ctx context.Context,
corpus CorpusAdapter,
req RetrieveRequest,
) (*RetrieveResult, error) {
if p == nil || p.embedder == nil || p.store == nil {
return nil, ErrNilDependency
}
query := strings.TrimSpace(req.Query)
if query == "" {
return nil, ErrInvalidQuery
}
topK := req.TopK
if topK <= 0 {
topK = defaultTopK
}
threshold := req.Threshold
if threshold < 0 {
threshold = defaultThreshold
}
filter := cloneMap(req.Filter)
if corpus != nil {
// 1. 先拼接 corpus 过滤条件,避免跨语料串召回。
corpusFilter, err := corpus.BuildRetrieveFilter(ctx, req.CorpusInput)
if err != nil {
return nil, err
}
filter = mergeMap(filter, corpusFilter)
filter["corpus"] = corpus.Name()
}
action := strings.TrimSpace(req.Action)
if action == "" {
action = "search"
}
vectors, err := p.embedder.Embed(ctx, []string{query}, action)
if err != nil {
return nil, err
}
if len(vectors) != 1 {
return nil, fmt.Errorf("embedding query length mismatch: %d", len(vectors))
}
scoredRows, err := p.store.Search(ctx, VectorSearchRequest{
QueryVector: vectors[0],
TopK: topK,
Filter: filter,
})
if err != nil {
return nil, err
}
rawCount := len(scoredRows)
candidates := make([]ScoredChunk, 0, len(scoredRows))
for _, row := range scoredRows {
if row.Score < threshold {
continue
}
candidates = append(candidates, ScoredChunk{
ChunkID: row.Row.ID,
DocumentID: asString(row.Row.Metadata["document_id"]),
Text: row.Row.Text,
Score: row.Score,
Metadata: cloneMap(row.Row.Metadata),
})
}
result := &RetrieveResult{
Items: candidates,
RawCount: rawCount,
FallbackUsed: false,
}
if len(candidates) == 0 || p.reranker == nil {
return result, nil
}
reranked, rerankErr := p.reranker.Rerank(ctx, query, candidates, topK)
if rerankErr != nil {
// 2. rerank 异常不终止主流程,统一降级为原排序。
result.FallbackUsed = true
result.FallbackReason = FallbackReasonRerankFailed
p.logger.Printf("rag rerank fallback: reason=%s err=%v", FallbackReasonRerankFailed, rerankErr)
return result, nil
}
result.Items = reranked
return result, nil
}
func normalizeChunkOption(opt ChunkOption) ChunkOption {
if opt.ChunkSize <= 0 {
opt.ChunkSize = defaultChunkSize
}
if opt.ChunkOverlap < 0 {
opt.ChunkOverlap = 0
}
if opt.ChunkOverlap >= opt.ChunkSize {
opt.ChunkOverlap = defaultChunkOvLap
if opt.ChunkOverlap >= opt.ChunkSize {
opt.ChunkOverlap = opt.ChunkSize / 5
}
}
return opt
}
func cloneMap(src map[string]any) map[string]any {
if len(src) == 0 {
return map[string]any{}
}
dst := make(map[string]any, len(src))
for key, value := range src {
dst[key] = value
}
return dst
}
func mergeMap(base map[string]any, ext map[string]any) map[string]any {
if base == nil {
base = map[string]any{}
}
for key, value := range ext {
base[key] = value
}
return base
}
func asString(v any) string {
if v == nil {
return ""
}
return fmt.Sprintf("%v", v)
}

View File

@@ -0,0 +1,94 @@
package core
import "time"
// SourceDocument 是统一语料文档模型。
//
// 职责边界:
// 1. 只描述“可被切块与索引”的原始文档;
// 2. 不承载业务流程状态。
type SourceDocument struct {
ID string
Text string
Title string
Metadata map[string]any
CreatedAt time.Time
}
// Chunk 是标准切块结果。
type Chunk struct {
ID string
DocumentID string
Text string
Order int
Metadata map[string]any
}
// ChunkOption 控制切块参数。
type ChunkOption struct {
ChunkSize int
ChunkOverlap int
}
// IngestOption 控制入库参数。
type IngestOption struct {
Chunk ChunkOption
// Action 用于 embedding 分型add/update/search
Action string
}
// IngestResult 描述一次入库执行摘要。
type IngestResult struct {
DocumentCount int
ChunkCount int
}
// RetrieveRequest 是统一检索请求。
type RetrieveRequest struct {
Query string
TopK int
Threshold float64
Action string
Filter map[string]any
CorpusInput any
}
// ScoredChunk 是统一召回结果。
type ScoredChunk struct {
ChunkID string
DocumentID string
Text string
Score float64
Metadata map[string]any
}
// RetrieveResult 是检索链路执行摘要。
type RetrieveResult struct {
Items []ScoredChunk
RawCount int
FallbackUsed bool
FallbackReason string
}
// VectorRow 是向量存储标准行。
type VectorRow struct {
ID string
Vector []float32
Text string
Metadata map[string]any
CreatedAt time.Time
UpdatedAt time.Time
}
// VectorSearchRequest 是向量检索请求。
type VectorSearchRequest struct {
QueryVector []float32
TopK int
Filter map[string]any
}
// ScoredVectorRow 是向量检索结果。
type ScoredVectorRow struct {
Row VectorRow
Score float64
}

View File

@@ -0,0 +1,13 @@
package corpus
import (
"crypto/sha256"
"encoding/hex"
"strings"
)
func hashLikeText(text string) string {
normalized := strings.TrimSpace(strings.ToLower(text))
sum := sha256.Sum256([]byte(normalized))
return hex.EncodeToString(sum[:8])
}

View File

@@ -0,0 +1,149 @@
package corpus
import (
"context"
"errors"
"fmt"
"strings"
"time"
"github.com/LoveLosita/smartflow/backend/infra/rag/core"
)
const memoryCorpusName = "memory"
// MemoryIngestItem 是记忆语料入库项。
type MemoryIngestItem struct {
MemoryID int64
UserID int
ConversationID string
AssistantID string
RunID string
MemoryType string
Title string
Content string
SensitivityLevel int
TTLAt *time.Time
CreatedAt *time.Time
}
// MemoryRetrieveInput 是记忆检索过滤输入。
type MemoryRetrieveInput struct {
UserID int
ConversationID string
AssistantID string
RunID string
MemoryType string
}
// MemoryCorpus 是记忆语料适配器。
type MemoryCorpus struct{}
func NewMemoryCorpus() *MemoryCorpus {
return &MemoryCorpus{}
}
func (c *MemoryCorpus) Name() string {
return memoryCorpusName
}
func (c *MemoryCorpus) BuildIngestDocuments(_ context.Context, input any) ([]core.SourceDocument, error) {
items, err := toMemoryItems(input)
if err != nil {
return nil, err
}
result := make([]core.SourceDocument, 0, len(items))
for _, item := range items {
if item.UserID <= 0 {
return nil, errors.New("memory ingest item user_id is invalid")
}
text := strings.TrimSpace(item.Content)
if text == "" {
continue
}
docID := fmt.Sprintf("memory:%d", item.MemoryID)
if item.MemoryID <= 0 {
docID = fmt.Sprintf("memory:uid:%d:%s", item.UserID, hashLikeText(text))
}
metadata := map[string]any{
"user_id": item.UserID,
"conversation_id": strings.TrimSpace(item.ConversationID),
"assistant_id": strings.TrimSpace(item.AssistantID),
"run_id": strings.TrimSpace(item.RunID),
"memory_type": strings.TrimSpace(strings.ToLower(item.MemoryType)),
"sensitivity_level": item.SensitivityLevel,
}
if item.TTLAt != nil {
metadata["ttl_at"] = item.TTLAt.Format(time.RFC3339)
}
createdAt := time.Now()
if item.CreatedAt != nil {
createdAt = *item.CreatedAt
}
result = append(result, core.SourceDocument{
ID: docID,
Text: text,
Title: strings.TrimSpace(item.Title),
Metadata: metadata,
CreatedAt: createdAt,
})
}
return result, nil
}
func (c *MemoryCorpus) BuildRetrieveFilter(_ context.Context, req any) (map[string]any, error) {
input, ok := req.(MemoryRetrieveInput)
if !ok {
if ptr, isPtr := req.(*MemoryRetrieveInput); isPtr && ptr != nil {
input = *ptr
} else if req == nil {
return nil, errors.New("memory retrieve input is nil")
} else {
return nil, errors.New("invalid memory retrieve input")
}
}
if input.UserID <= 0 {
return nil, errors.New("memory retrieve user_id is invalid")
}
filter := map[string]any{
"user_id": input.UserID,
}
if v := strings.TrimSpace(input.ConversationID); v != "" {
filter["conversation_id"] = v
}
if v := strings.TrimSpace(input.AssistantID); v != "" {
filter["assistant_id"] = v
}
if v := strings.TrimSpace(input.RunID); v != "" {
filter["run_id"] = v
}
if v := strings.TrimSpace(strings.ToLower(input.MemoryType)); v != "" {
filter["memory_type"] = v
}
return filter, nil
}
func toMemoryItems(input any) ([]MemoryIngestItem, error) {
switch value := input.(type) {
case MemoryIngestItem:
return []MemoryIngestItem{value}, nil
case *MemoryIngestItem:
if value == nil {
return nil, errors.New("memory ingest item is nil")
}
return []MemoryIngestItem{*value}, nil
case []MemoryIngestItem:
return value, nil
case []*MemoryIngestItem:
items := make([]MemoryIngestItem, 0, len(value))
for _, ptr := range value {
if ptr == nil {
continue
}
items = append(items, *ptr)
}
return items, nil
default:
return nil, errors.New("invalid memory ingest input")
}
}

View File

@@ -0,0 +1,163 @@
package corpus
import (
"context"
"errors"
"fmt"
"strings"
"time"
"github.com/LoveLosita/smartflow/backend/infra/rag/core"
)
const webCorpusName = "web"
// WebIngestItem 是网页语料入库项。
type WebIngestItem struct {
URL string
Title string
Content string
Snippet string
Domain string
QueryID string
SessionID string
PublishedAt *time.Time
FetchedAt *time.Time
SourceRank int
}
// WebRetrieveInput 是网页检索过滤输入。
type WebRetrieveInput struct {
QueryID string
SessionID string
Domain string
}
// WebCorpus 是网页语料适配器。
type WebCorpus struct{}
func NewWebCorpus() *WebCorpus {
return &WebCorpus{}
}
func (c *WebCorpus) Name() string {
return webCorpusName
}
func (c *WebCorpus) BuildIngestDocuments(_ context.Context, input any) ([]core.SourceDocument, error) {
items, err := toWebItems(input)
if err != nil {
return nil, err
}
result := make([]core.SourceDocument, 0, len(items))
for _, item := range items {
url := strings.TrimSpace(item.URL)
if url == "" {
return nil, errors.New("web ingest item url is empty")
}
mainText := buildWebText(item)
if strings.TrimSpace(mainText) == "" {
continue
}
docID := fmt.Sprintf("web:%s", hashLikeText(url+"|"+mainText))
metadata := map[string]any{
"url": url,
"domain": strings.TrimSpace(item.Domain),
"query_id": strings.TrimSpace(item.QueryID),
"session_id": strings.TrimSpace(item.SessionID),
"source_rank": item.SourceRank,
}
if item.PublishedAt != nil {
metadata["published_at"] = item.PublishedAt.Format(time.RFC3339)
}
if item.FetchedAt != nil {
metadata["fetched_at"] = item.FetchedAt.Format(time.RFC3339)
}
createdAt := time.Now()
if item.FetchedAt != nil {
createdAt = *item.FetchedAt
}
result = append(result, core.SourceDocument{
ID: docID,
Text: mainText,
Title: strings.TrimSpace(item.Title),
Metadata: metadata,
CreatedAt: createdAt,
})
}
return result, nil
}
func (c *WebCorpus) BuildRetrieveFilter(_ context.Context, req any) (map[string]any, error) {
input, ok := req.(WebRetrieveInput)
if !ok {
if ptr, isPtr := req.(*WebRetrieveInput); isPtr && ptr != nil {
input = *ptr
} else if req == nil {
return nil, errors.New("web retrieve input is nil")
} else {
return nil, errors.New("invalid web retrieve input")
}
}
// 1. query_id/session_id 至少要有一个,避免跨问题串数据。
queryID := strings.TrimSpace(input.QueryID)
sessionID := strings.TrimSpace(input.SessionID)
if queryID == "" && sessionID == "" {
return nil, errors.New("web retrieve filter requires query_id or session_id")
}
filter := map[string]any{}
if queryID != "" {
filter["query_id"] = queryID
}
if sessionID != "" {
filter["session_id"] = sessionID
}
if domain := strings.TrimSpace(input.Domain); domain != "" {
filter["domain"] = domain
}
return filter, nil
}
func toWebItems(input any) ([]WebIngestItem, error) {
switch value := input.(type) {
case WebIngestItem:
return []WebIngestItem{value}, nil
case *WebIngestItem:
if value == nil {
return nil, errors.New("web ingest item is nil")
}
return []WebIngestItem{*value}, nil
case []WebIngestItem:
return value, nil
case []*WebIngestItem:
items := make([]WebIngestItem, 0, len(value))
for _, ptr := range value {
if ptr == nil {
continue
}
items = append(items, *ptr)
}
return items, nil
default:
return nil, errors.New("invalid web ingest input")
}
}
func buildWebText(item WebIngestItem) string {
parts := make([]string, 0, 3)
if title := strings.TrimSpace(item.Title); title != "" {
parts = append(parts, title)
}
if snippet := strings.TrimSpace(item.Snippet); snippet != "" {
parts = append(parts, snippet)
}
if content := strings.TrimSpace(item.Content); content != "" {
parts = append(parts, content)
}
return strings.Join(parts, "\n\n")
}

View File

@@ -0,0 +1,21 @@
package embed
import (
"context"
"errors"
)
// EinoEmbedder 是 Eino embedding 的占位实现。
//
// 说明:
// 1. 本轮先占位接口,避免过早耦合具体 Provider
// 2. 后续接入真实 embedding 时,只替换此文件内部实现。
type EinoEmbedder struct{}
func NewEinoEmbedder() *EinoEmbedder {
return &EinoEmbedder{}
}
func (e *EinoEmbedder) Embed(_ context.Context, _ []string, _ string) ([][]float32, error) {
return nil, errors.New("eino embedder is not implemented yet")
}

View File

@@ -0,0 +1,46 @@
package embed
import (
"context"
"crypto/sha256"
"encoding/binary"
"strings"
)
const defaultDim = 16
// MockEmbedder 是本地可运行的占位向量化实现。
//
// 说明:
// 1. 该实现用于开发阶段打通链路,不代表真实语义向量质量;
// 2. 后续可替换为 Eino embedding 实现,接口保持不变。
type MockEmbedder struct {
dim int
}
func NewMockEmbedder(dim int) *MockEmbedder {
if dim <= 0 {
dim = defaultDim
}
return &MockEmbedder{dim: dim}
}
func (e *MockEmbedder) Embed(_ context.Context, texts []string, _ string) ([][]float32, error) {
result := make([][]float32, 0, len(texts))
for _, text := range texts {
result = append(result, e.embedOne(text))
}
return result, nil
}
func (e *MockEmbedder) embedOne(text string) []float32 {
normalized := strings.TrimSpace(strings.ToLower(text))
sum := sha256.Sum256([]byte(normalized))
vec := make([]float32, e.dim)
for i := 0; i < e.dim; i++ {
offset := (i * 4) % len(sum)
v := binary.BigEndian.Uint32(sum[offset : offset+4])
vec[i] = float32(v%1000) / 1000
}
return vec
}

23
backend/infra/rag/rag.go Normal file
View File

@@ -0,0 +1,23 @@
package rag
import (
"github.com/LoveLosita/smartflow/backend/infra/rag/chunk"
"github.com/LoveLosita/smartflow/backend/infra/rag/core"
"github.com/LoveLosita/smartflow/backend/infra/rag/embed"
"github.com/LoveLosita/smartflow/backend/infra/rag/rerank"
"github.com/LoveLosita/smartflow/backend/infra/rag/store"
)
// NewDefaultPipeline 构造默认可运行的 RAG Pipeline。
//
// 当前策略:
// 1. 默认使用本地 MockEmbedder + InMemoryStore保证零外部依赖可运行
// 2. 后续切 Milvus / Eino 时仅替换依赖,不改业务调用方式。
func NewDefaultPipeline() *core.Pipeline {
return core.NewPipeline(
chunk.NewTextChunker(),
embed.NewMockEmbedder(16),
store.NewInMemoryVectorStore(),
rerank.NewNoopReranker(),
)
}

View File

@@ -0,0 +1,19 @@
package rerank
import (
"context"
"errors"
"github.com/LoveLosita/smartflow/backend/infra/rag/core"
)
// EinoReranker 是 Eino 重排器占位实现。
type EinoReranker struct{}
func NewEinoReranker() *EinoReranker {
return &EinoReranker{}
}
func (r *EinoReranker) Rerank(_ context.Context, _ string, _ []core.ScoredChunk, _ int) ([]core.ScoredChunk, error) {
return nil, errors.New("eino reranker is not implemented yet")
}

View File

@@ -0,0 +1,30 @@
package rerank
import (
"context"
"sort"
"github.com/LoveLosita/smartflow/backend/infra/rag/core"
)
// NoopReranker 是默认重排器(仅按原 score 排序)。
type NoopReranker struct{}
func NewNoopReranker() *NoopReranker {
return &NoopReranker{}
}
func (r *NoopReranker) Rerank(_ context.Context, _ string, candidates []core.ScoredChunk, topK int) ([]core.ScoredChunk, error) {
if len(candidates) == 0 {
return nil, nil
}
sorted := make([]core.ScoredChunk, len(candidates))
copy(sorted, candidates)
sort.SliceStable(sorted, func(i, j int) bool {
return sorted[i].Score > sorted[j].Score
})
if topK <= 0 || topK >= len(sorted) {
return sorted, nil
}
return sorted[:topK], nil
}

View File

@@ -0,0 +1,90 @@
package retrieve
import (
"context"
"fmt"
"strings"
"github.com/LoveLosita/smartflow/backend/infra/rag/core"
)
// VectorRetriever 是通用检索器embed + vector search
type VectorRetriever struct {
embedder core.Embedder
store core.VectorStore
}
func NewVectorRetriever(embedder core.Embedder, store core.VectorStore) *VectorRetriever {
return &VectorRetriever{
embedder: embedder,
store: store,
}
}
func (r *VectorRetriever) Retrieve(ctx context.Context, req core.RetrieveRequest) ([]core.ScoredChunk, error) {
if r == nil || r.embedder == nil || r.store == nil {
return nil, core.ErrNilDependency
}
query := strings.TrimSpace(req.Query)
if query == "" {
return nil, core.ErrInvalidQuery
}
topK := req.TopK
if topK <= 0 {
topK = 8
}
action := strings.TrimSpace(req.Action)
if action == "" {
action = "search"
}
vectors, err := r.embedder.Embed(ctx, []string{query}, action)
if err != nil {
return nil, err
}
if len(vectors) != 1 {
return nil, fmt.Errorf("embedding query length mismatch: %d", len(vectors))
}
rows, err := r.store.Search(ctx, core.VectorSearchRequest{
QueryVector: vectors[0],
TopK: topK,
Filter: req.Filter,
})
if err != nil {
return nil, err
}
result := make([]core.ScoredChunk, 0, len(rows))
for _, row := range rows {
if row.Score < req.Threshold {
continue
}
result = append(result, core.ScoredChunk{
ChunkID: row.Row.ID,
DocumentID: asString(row.Row.Metadata["document_id"]),
Text: row.Row.Text,
Score: row.Score,
Metadata: cloneMap(row.Row.Metadata),
})
}
return result, nil
}
func cloneMap(src map[string]any) map[string]any {
if len(src) == 0 {
return map[string]any{}
}
dst := make(map[string]any, len(src))
for k, v := range src {
dst[k] = v
}
return dst
}
func asString(v any) string {
if v == nil {
return ""
}
return fmt.Sprintf("%v", v)
}

View File

@@ -0,0 +1,166 @@
package store
import (
"context"
"fmt"
"math"
"sort"
"strings"
"sync"
"time"
"github.com/LoveLosita/smartflow/backend/infra/rag/core"
)
// InMemoryVectorStore 是本地开发用向量存储实现。
//
// 注意:
// 1. 仅用于开发调试,不建议生产使用;
// 2. 真实环境可替换为 MilvusStore接口保持一致。
type InMemoryVectorStore struct {
mu sync.RWMutex
rows map[string]core.VectorRow
}
func NewInMemoryVectorStore() *InMemoryVectorStore {
return &InMemoryVectorStore{
rows: make(map[string]core.VectorRow),
}
}
func (s *InMemoryVectorStore) Upsert(_ context.Context, rows []core.VectorRow) error {
if len(rows) == 0 {
return nil
}
now := time.Now()
s.mu.Lock()
defer s.mu.Unlock()
for _, row := range rows {
current, exists := s.rows[row.ID]
if exists {
row.CreatedAt = current.CreatedAt
row.UpdatedAt = now
} else {
if row.CreatedAt.IsZero() {
row.CreatedAt = now
}
row.UpdatedAt = now
}
s.rows[row.ID] = row
}
return nil
}
func (s *InMemoryVectorStore) Search(_ context.Context, req core.VectorSearchRequest) ([]core.ScoredVectorRow, error) {
topK := req.TopK
if topK <= 0 {
topK = 8
}
s.mu.RLock()
defer s.mu.RUnlock()
result := make([]core.ScoredVectorRow, 0, len(s.rows))
for _, row := range s.rows {
if !matchMetadataFilter(row.Metadata, req.Filter) {
continue
}
score := cosineSimilarity(req.QueryVector, row.Vector)
result = append(result, core.ScoredVectorRow{
Row: row,
Score: score,
})
}
sort.SliceStable(result, func(i, j int) bool {
return result[i].Score > result[j].Score
})
if len(result) <= topK {
return result, nil
}
return result[:topK], nil
}
func (s *InMemoryVectorStore) Delete(_ context.Context, ids []string) error {
if len(ids) == 0 {
return nil
}
s.mu.Lock()
defer s.mu.Unlock()
for _, id := range ids {
delete(s.rows, id)
}
return nil
}
func (s *InMemoryVectorStore) Get(_ context.Context, ids []string) ([]core.VectorRow, error) {
if len(ids) == 0 {
return nil, nil
}
s.mu.RLock()
defer s.mu.RUnlock()
result := make([]core.VectorRow, 0, len(ids))
for _, id := range ids {
row, exists := s.rows[id]
if !exists {
continue
}
result = append(result, row)
}
return result, nil
}
func cosineSimilarity(a, b []float32) float64 {
if len(a) == 0 || len(b) == 0 {
return 0
}
n := len(a)
if len(b) < n {
n = len(b)
}
if n == 0 {
return 0
}
var dot, normA, normB float64
for i := 0; i < n; i++ {
av := float64(a[i])
bv := float64(b[i])
dot += av * bv
normA += av * av
normB += bv * bv
}
if normA == 0 || normB == 0 {
return 0
}
return dot / (math.Sqrt(normA) * math.Sqrt(normB))
}
func matchMetadataFilter(metadata map[string]any, filter map[string]any) bool {
if len(filter) == 0 {
return true
}
for key, wanted := range filter {
got, exists := metadata[key]
if !exists {
return false
}
if !equalAny(got, wanted) {
return false
}
}
return true
}
func equalAny(left any, right any) bool {
return toString(left) == toString(right)
}
func toString(v any) string {
if v == nil {
return ""
}
return fmtAny(v)
}
func fmtAny(v any) string {
return strings.TrimSpace(fmt.Sprintf("%v", v))
}

View File

@@ -0,0 +1,35 @@
package store
import (
"context"
"errors"
"github.com/LoveLosita/smartflow/backend/infra/rag/core"
)
// MilvusStore 是 Milvus 连接器占位实现。
//
// 说明:
// 1. 本轮先保留接口结构,便于后续平滑替换 InMemoryStore
// 2. 真实接入时需补充连接池、集合初始化、元数据过滤与错误转换。
type MilvusStore struct{}
func NewMilvusStore() *MilvusStore {
return &MilvusStore{}
}
func (s *MilvusStore) Upsert(_ context.Context, _ []core.VectorRow) error {
return errors.New("milvus store is not implemented yet")
}
func (s *MilvusStore) Search(_ context.Context, _ core.VectorSearchRequest) ([]core.ScoredVectorRow, error) {
return nil, errors.New("milvus store is not implemented yet")
}
func (s *MilvusStore) Delete(_ context.Context, _ []string) error {
return errors.New("milvus store is not implemented yet")
}
func (s *MilvusStore) Get(_ context.Context, _ []string) ([]core.VectorRow, error) {
return nil, errors.New("milvus store is not implemented yet")
}

View File

@@ -0,0 +1,8 @@
package store
import "github.com/LoveLosita/smartflow/backend/infra/rag/core"
// EnsureCompile 用于静态校验实现是否满足接口。
func EnsureCompile() {
var _ core.VectorStore = (*InMemoryVectorStore)(nil)
}

View File

@@ -23,6 +23,10 @@ func autoMigrateModels(db *gorm.DB) error {
&model.AgentOutboxMessage{},
&model.AgentScheduleState{},
&model.AgentStateSnapshotRecord{},
&model.MemoryItem{},
&model.MemoryJob{},
&model.MemoryAuditLog{},
&model.MemoryUserSetting{},
}
for _, m := range models {

View File

@@ -0,0 +1,165 @@
# HANDOFFRAG 复用后续实施计划Memory 目录留档)
## 1. 文档目的
1. 给下一位开发者一个可直接执行的后续实施蓝图。
2. 明确“已完成/未完成/切流点/回滚点”,避免重复摸索。
3. 保证 Memory 与 WebSearch 共用一套 RAG Core减少重复实现。
## 2. 当前状态(截至本次交接)
### 2.1 已完成
1. `backend/infra/rag` 共享骨架已建好,包含:
- `core`统一类型、接口、pipeline、错误码。
- `chunk/embed/retrieve/rerank/store`默认可运行实现mock/in-memory/noop
- `corpus``MemoryCorpus``WebCorpus` 适配器。
- `config``rag.*` 配置读取。
2. Memory Day1 写入链路已打通(`memory.extract.requested` 事件、`memory_jobs` 入库、worker 状态推进)。
3. Milvus 相关 `docker-compose` 服务定义已补齐(本机因网络问题尚未拉起验证)。
### 2.2 未完成
1. RAG Core 尚未接入真实业务调用点(当前仍是并行骨架状态)。
2. Milvus 真正读写实现未落地(`MilvusStore` 为占位)。
3. Eino Embedding/Reranker 未落地(当前占位实现)。
4. Memory 读取注入链路Day2尚未切到 RAG Core。
5. WebSearch 尚未切到 `WebCorpus + RAG Core`
## 3. 后续实施总原则
1. 并行迁移:新旧逻辑并存,先灰度后切流,最后删除旧实现。
2. 单轮只做一个能力域:先 Memory 读链路,再 WebSearch。
3. 保留可回滚开关:任何切流都必须有一键回退路径。
4. 失败可降级Rerank/Vector 异常不影响主链路响应。
## 4. 建议执行顺序4 轮)
## Round 2Memory 读链路接入 RAG Core优先
### 目标
1. Memory 检索由 `infra/rag` 承载,保留旧检索兜底。
2. 注入效果不劣于当前逻辑,且可观测。
### 实施项
1. 在 Memory ReadService 中接入 `core.Pipeline.Retrieve`
2. 构造 `MemoryRetrieveInput`,强制过滤:
- `user_id + assistant_id + conversation_id`
3. 配置开关:
- `memory.rag.enabled`(默认 `false`,灰度开启)
4. 降级策略:
- RAG 失败 -> 回退旧检索链路;
- Rerank 失败 -> 使用原排序pipeline 已支持 fallback 标记)。
5. 指标补齐:
- `memory_rag_hit_count`
- `memory_rag_fallback_rate`
- `memory_rag_latency_ms`
### 验收
1. 开关关闭时行为与当前一致。
2. 开关开启时可稳定召回,失败能回退。
3. 日志可追踪“为什么没注入某条记忆”。
## Round 3WebSearch 接入 WebCorpus + RAG Core
### 目标
1. WebSearch 复用同一检索流程,不再各写一套召回逻辑。
2. 严格限制跨 query/session 串召回。
### 实施项
1. 把抓取结果映射为 `WebIngestItem` 入 RAG。
2. 检索时必须带:
- `query_id``session_id`
3. 配置开关:
- `websearch.rag.enabled`(默认 `false`
4. 保留原 websearch 结果路径RAG 仅先做“补充召回层”。
### 验收
1. 开启后答案质量不下降,且无跨 query 污染。
2. 关闭后立即回到旧逻辑。
## Round 4Milvus + Eino 真实实现替换
### 目标
1.`InMemoryStore` 替换为 `MilvusStore`(生产可用)。
2.`MockEmbedder/NoopReranker` 替换为 Eino 实现。
### 实施项
1. `store/milvus_store.go`
- 实现 `Upsert/Search/Delete/Get`
- 建 collection 与 metadata filter 映射
2. `embed/eino_embedder.go`
- 完成 embedding 调用与超时控制
3. `rerank/eino_reranker.go`
- 完成重排调用与错误降级
4. 配置补齐:
- `rag.store=milvus`
- `rag.embed.provider=eino`
- `rag.reranker.provider=eino`
### 验收
1. Milvus 可稳定写入/检索。
2. 模型服务波动时主链路可降级。
3. 指标完整命中、延迟、fallback、错误码
## Round 5统一切流与旧逻辑收敛
### 目标
1. Memory + WebSearch 默认走 RAG Core。
2. 删除重复旧实现,避免双轨长期维护成本。
### 实施项
1. 提升开关默认值为 `true`
2. 观察窗口(建议 3~7 天)稳定后删除旧分支代码。
3. 更新文档:
- `backend/memory/记忆模块实施计划.md`
- `backend/agent/通用能力接入文档.md`(若新增/替换通用能力,必须同步)
### 验收
1. 代码无重复检索实现。
2. 回滚开关仍可用(应急时关闭 RAG
3. 线上指标稳定且可追踪。
## 5. 开关与回滚建议
1. 建议开关:
- `memory.rag.enabled`
- `websearch.rag.enabled`
- `rag.reranker.enabled`
2. 回滚策略:
- 先关 corpus 级开关memory/websearch
- 再关 reranker。
- 极端情况降级到旧检索链路。
## 6. 关键风险与应对
1. 风险:切流后召回漂移。
应对:双写日志对比(旧链路 vs 新链路 TopK
2. 风险Milvus 延迟波动。
应对:检索超时 + fallback + 限流。
3. 风险:跨会话串数据。
应对:过滤维度强校验,不满足则拒绝检索。
## 7. 接手即办清单(最小行动)
1. 先做 Round 2Memory 读链路接 RAG不要同时改 WebSearch。
2. 提交前必须跑:
- `go test ./...`
3. 每次本地 `go test` 后清理项目根目录 `.gocache`
4. 完成一轮后在本文件追加:
- 已落地清单
- 待办差距
- 下一轮入口

28
backend/memory/README.md Normal file
View File

@@ -0,0 +1,28 @@
# Memory 模块Day1 骨架)
## 本轮目标
1. 打通 `memory.extract.requested` 事件发布与消费。
2. 消费后把任务可靠写入 `memory_jobs`(支持幂等)。
3. 提供 `worker.RunOnce()`,可手工推进 `pending -> processing -> success/failed`
## 本轮边界(刻意不做)
1. 不接真实 LLM 抽取与冲突决策。
2. 不接 Milvus 向量召回。
3. 不做读取注入链路Day2 再接)。
## 目录说明
- `model/`:记忆领域 DTO、状态机、配置对象。
- `repo/``memory_*` 表访问。
- `service/`:任务入队门面与配置加载。
- `orchestrator/`写入链路编排Day1 为 mock 抽取)。
- `worker/`:任务执行器(支持手工触发单次运行)。
- `utils/``ExtractJSON``NormalizeFacts` 等工具函数。
## 手工验证建议
1. 发起一轮聊天后,检查 outbox 是否存在 `memory.extract.requested`
2. 等待消费后,检查 `memory_jobs` 是否新增 `pending` 记录。
3. 手工调用 `worker.RunOnce()`,确认任务推进到 `success/failed`

View File

@@ -0,0 +1,16 @@
package model
import "time"
// AuditLogDTO 是审计日志领域对象。
type AuditLogDTO struct {
ID int64
MemoryID int64
UserID int
Operation string
OperatorType string
Reason string
BeforeJSON string
AfterJSON string
CreatedAt *time.Time
}

View File

@@ -0,0 +1,25 @@
package model
import "time"
// Config 是记忆模块配置对象Day1 首版)。
//
// 职责边界:
// 1. 只承载模块运行参数,不承载业务状态;
// 2. 允许启动期统一注入,避免业务层直接依赖配置中心。
type Config struct {
Enabled bool
ExtractPrompt string
DecisionPrompt string
Threshold float64
EnableReranker bool
LLMTemperature float64
LLMTopP float64
JobMaxRetry int
WorkerPollEvery time.Duration
WorkerClaimBatch int
}

View File

@@ -0,0 +1,27 @@
package model
import "time"
// ItemDTO 是记忆条目对外读写 DTO。
//
// 职责边界:
// 1. 面向 memory 模块内部服务层使用;
// 2. 不直接绑定 GORM 标签,避免传输结构与存储结构强耦合。
type ItemDTO struct {
ID int64
UserID int
ConversationID string
AssistantID string
RunID string
MemoryType string
Title string
Content string
Confidence float64
Importance float64
SensitivityLevel int
IsExplicit bool
Status string
TTLAt *time.Time
CreatedAt *time.Time
UpdatedAt *time.Time
}

View File

@@ -0,0 +1,41 @@
package model
import "time"
// ExtractJobPayload 是 memory_jobs.payload_json 的领域视图。
//
// 职责边界:
// 1. 只描述抽取任务执行所需字段;
// 2. 与数据库模型解耦,避免后续表结构调整污染 worker 逻辑。
type ExtractJobPayload struct {
UserID int `json:"user_id"`
ConversationID string `json:"conversation_id"`
AssistantID string `json:"assistant_id,omitempty"`
RunID string `json:"run_id,omitempty"`
SourceMessageID int64 `json:"source_message_id,omitempty"`
SourceRole string `json:"source_role"`
SourceText string `json:"source_text"`
OccurredAt time.Time `json:"occurred_at"`
TraceID string `json:"trace_id,omitempty"`
IdempotencyKey string `json:"idempotency_key"`
}
// FactCandidate 表示抽取阶段得到的候选事实。
type FactCandidate struct {
MemoryType string
Title string
Content string
Confidence float64
IsExplicit bool
}
// NormalizedFact 表示通过标准化后的可入库事实。
type NormalizedFact struct {
MemoryType string
Title string
Content string
NormalizedContent string
ContentHash string
Confidence float64
IsExplicit bool
}

View File

@@ -0,0 +1,12 @@
package model
import "time"
// UserSettingDTO 是用户记忆开关领域对象。
type UserSettingDTO struct {
UserID int
MemoryEnabled bool
ImplicitMemoryEnabled bool
SensitiveMemoryEnabled bool
UpdatedAt *time.Time
}

View File

@@ -0,0 +1,59 @@
package model
import "strings"
const (
// MemoryTypePreference 表示用户偏好类记忆。
MemoryTypePreference = "preference"
// MemoryTypeConstraint 表示硬约束类记忆。
MemoryTypeConstraint = "constraint"
// MemoryTypeFact 表示一般事实类记忆。
MemoryTypeFact = "fact"
// MemoryTypeTodoHint 表示近期待办线索类记忆。
MemoryTypeTodoHint = "todo_hint"
)
const (
// DecisionActionAdd 表示新增记忆。
DecisionActionAdd = "ADD"
// DecisionActionUpdate 表示更新记忆。
DecisionActionUpdate = "UPDATE"
// DecisionActionDelete 表示删除记忆。
DecisionActionDelete = "DELETE"
// DecisionActionNone 表示不做写入动作。
DecisionActionNone = "NONE"
)
var validMemoryTypes = map[string]struct{}{
MemoryTypePreference: {},
MemoryTypeConstraint: {},
MemoryTypeFact: {},
MemoryTypeTodoHint: {},
}
var validDecisionActions = map[string]struct{}{
DecisionActionAdd: {},
DecisionActionUpdate: {},
DecisionActionDelete: {},
DecisionActionNone: {},
}
// NormalizeMemoryType 统一记忆类型字符串。
//
// 职责边界:
// 1. 只做字符串标准化,不做业务兜底;
// 2. 若调用方传入非法类型,返回空字符串供上游决定丢弃或降级。
func NormalizeMemoryType(memoryType string) string {
normalized := strings.ToLower(strings.TrimSpace(memoryType))
if _, ok := validMemoryTypes[normalized]; !ok {
return ""
}
return normalized
}
// IsValidDecisionAction 校验决策动作是否合法。
func IsValidDecisionAction(action string) bool {
normalized := strings.ToUpper(strings.TrimSpace(action))
_, ok := validDecisionActions[normalized]
return ok
}

View File

@@ -0,0 +1,43 @@
package orchestrator
import (
"context"
"strings"
memorymodel "github.com/LoveLosita/smartflow/backend/memory/model"
memoryutils "github.com/LoveLosita/smartflow/backend/memory/utils"
)
// WriteOrchestrator 是写入链路编排器Day1 首版)。
//
// 职责边界:
// 1. Day1 只做 mock 抽取 + 标准化,不接 LLM 决策;
// 2. Day2/Day3 再引入冲突消解、重排与向量召回。
type WriteOrchestrator struct{}
func NewWriteOrchestrator() *WriteOrchestrator {
return &WriteOrchestrator{}
}
// ExtractFacts 执行“候选事实抽取 -> 标准化”链路。
//
// Day1 策略:
// 1. 先用 source_text 直接构造候选事实,确保链路可跑通;
// 2. 后续再替换成 LLM 抽取与结构化决策。
func (o *WriteOrchestrator) ExtractFacts(_ context.Context, payload memorymodel.ExtractJobPayload) ([]memorymodel.NormalizedFact, error) {
sourceText := strings.TrimSpace(payload.SourceText)
if sourceText == "" {
return nil, nil
}
candidates := []memorymodel.FactCandidate{
{
MemoryType: memorymodel.MemoryTypeFact,
Title: "用户近期提及",
Content: sourceText,
Confidence: 0.6,
IsExplicit: false,
},
}
return memoryutils.NormalizeFacts(candidates), nil
}

View File

@@ -0,0 +1,25 @@
package repo
import (
"context"
"errors"
"github.com/LoveLosita/smartflow/backend/model"
"gorm.io/gorm"
)
// AuditRepo 封装 memory_audit_logs 的数据访问。
type AuditRepo struct {
db *gorm.DB
}
func NewAuditRepo(db *gorm.DB) *AuditRepo {
return &AuditRepo{db: db}
}
func (r *AuditRepo) Create(ctx context.Context, log model.MemoryAuditLog) error {
if r == nil || r.db == nil {
return errors.New("memory audit repo is nil")
}
return r.db.WithContext(ctx).Create(&log).Error
}

View File

@@ -0,0 +1,33 @@
package repo
import (
"context"
"errors"
"github.com/LoveLosita/smartflow/backend/model"
"gorm.io/gorm"
)
// ItemRepo 封装 memory_items 的数据访问Day1 先占位)。
type ItemRepo struct {
db *gorm.DB
}
func NewItemRepo(db *gorm.DB) *ItemRepo {
return &ItemRepo{db: db}
}
// UpsertItems 预留给 Day2/Day3 的写入链路。
//
// Day1 约束:
// 1. 先完成任务入队与状态机闭环;
// 2. 不在本阶段引入复杂冲突消解与向量写入。
func (r *ItemRepo) UpsertItems(ctx context.Context, items []model.MemoryItem) error {
if r == nil || r.db == nil {
return errors.New("memory item repo is nil")
}
if len(items) == 0 {
return nil
}
return r.db.WithContext(ctx).Create(&items).Error
}

View File

@@ -0,0 +1,221 @@
package repo
import (
"context"
"encoding/json"
"errors"
"time"
memorymodel "github.com/LoveLosita/smartflow/backend/memory/model"
"github.com/LoveLosita/smartflow/backend/model"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
// JobRepo 封装 memory_jobs 的数据访问。
type JobRepo struct {
db *gorm.DB
}
func NewJobRepo(db *gorm.DB) *JobRepo {
return &JobRepo{db: db}
}
func (r *JobRepo) WithTx(tx *gorm.DB) *JobRepo {
return &JobRepo{db: tx}
}
// CreatePendingExtractJob 创建“待抽取”任务(幂等写入)。
//
// 失败语义:
// 1. 参数非法直接返回 error由上游决定 dead 或重试;
// 2. 同幂等键重复写入采用 DoNothing保证无副作用。
func (r *JobRepo) CreatePendingExtractJob(
ctx context.Context,
payload memorymodel.ExtractJobPayload,
sourceEventID string,
) error {
if r == nil || r.db == nil {
return errors.New("memory job repo is nil")
}
if payload.UserID <= 0 {
return errors.New("invalid user_id")
}
if payload.IdempotencyKey == "" {
return errors.New("idempotency_key is empty")
}
rawPayload, err := json.Marshal(payload)
if err != nil {
return err
}
now := time.Now()
job := model.MemoryJob{
UserID: payload.UserID,
ConversationID: strPtrOrNil(payload.ConversationID),
SourceMessageID: int64PtrOrNil(payload.SourceMessageID),
SourceEventID: strPtrOrNil(sourceEventID),
JobType: model.MemoryJobTypeExtract,
IdempotencyKey: payload.IdempotencyKey,
PayloadJSON: string(rawPayload),
Status: model.MemoryJobStatusPending,
RetryCount: 0,
MaxRetry: 6,
NextRetryAt: &now,
}
return r.db.WithContext(ctx).
Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "idempotency_key"}},
DoNothing: true,
}).
Create(&job).Error
}
// ClaimNextRunnableExtractJob 抢占一个可执行的 extract 任务。
//
// 抢占规则:
// 1. 只从 pending/failed 中挑 next_retry_at 已到期任务;
// 2. 用行锁避免多个 worker 抢到同一条任务;
// 3. 抢占成功后立即置为 processing防止重复执行。
func (r *JobRepo) ClaimNextRunnableExtractJob(ctx context.Context, now time.Time) (*model.MemoryJob, error) {
if r == nil || r.db == nil {
return nil, errors.New("memory job repo is nil")
}
var claimed *model.MemoryJob
err := r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var job model.MemoryJob
queryErr := tx.
Clauses(clause.Locking{Strength: "UPDATE"}).
Where("job_type = ?", model.MemoryJobTypeExtract).
Where("status IN ?", []string{model.MemoryJobStatusPending, model.MemoryJobStatusFailed}).
Where("(next_retry_at IS NULL OR next_retry_at <= ?)", now).
Order("id ASC").
First(&job).Error
if queryErr != nil {
if errors.Is(queryErr, gorm.ErrRecordNotFound) {
return nil
}
return queryErr
}
updates := map[string]any{
"status": model.MemoryJobStatusProcessing,
"updated_at": now,
"last_error": nil,
}
if updateErr := tx.Model(&model.MemoryJob{}).Where("id = ?", job.ID).Updates(updates).Error; updateErr != nil {
return updateErr
}
job.Status = model.MemoryJobStatusProcessing
job.UpdatedAt = &now
claimed = &job
return nil
})
if err != nil {
return nil, err
}
return claimed, nil
}
// MarkSuccess 把任务推进为 success 最终态。
func (r *JobRepo) MarkSuccess(ctx context.Context, jobID int64) error {
if r == nil || r.db == nil {
return errors.New("memory job repo is nil")
}
now := time.Now()
updates := map[string]any{
"status": model.MemoryJobStatusSuccess,
"last_error": nil,
"next_retry_at": nil,
"updated_at": now,
}
return r.db.WithContext(ctx).Model(&model.MemoryJob{}).Where("id = ?", jobID).Updates(updates).Error
}
// MarkFailed 按重试策略推进任务到 failed/dead。
//
// 规则:
// 1. retry_count +1 后若超上限,直接 dead
// 2. 未超上限则写 failed 并设置 next_retry_at。
func (r *JobRepo) MarkFailed(ctx context.Context, jobID int64, reason string) error {
if r == nil || r.db == nil {
return errors.New("memory job repo is nil")
}
return r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var job model.MemoryJob
queryErr := tx.
Clauses(clause.Locking{Strength: "UPDATE"}).
Where("id = ?", jobID).
First(&job).Error
if queryErr != nil {
return queryErr
}
if job.Status == model.MemoryJobStatusSuccess || job.Status == model.MemoryJobStatusDead {
return nil
}
maxRetry := job.MaxRetry
if maxRetry <= 0 {
maxRetry = 6
}
nextRetryCount := job.RetryCount + 1
now := time.Now()
status := model.MemoryJobStatusFailed
var nextRetryAt *time.Time
if nextRetryCount >= maxRetry {
status = model.MemoryJobStatusDead
nextRetryAt = nil
} else {
t := now.Add(calcRetryBackoff(nextRetryCount))
nextRetryAt = &t
}
lastErr := truncateError(reason)
updates := map[string]any{
"status": status,
"retry_count": nextRetryCount,
"last_error": &lastErr,
"next_retry_at": nextRetryAt,
"updated_at": now,
}
return tx.Model(&model.MemoryJob{}).Where("id = ?", jobID).Updates(updates).Error
})
}
func calcRetryBackoff(retryCount int) time.Duration {
if retryCount <= 0 {
return time.Second
}
if retryCount > 6 {
retryCount = 6
}
return time.Second * time.Duration(1<<(retryCount-1))
}
func truncateError(reason string) string {
if len(reason) <= 2000 {
return reason
}
return reason[:2000]
}
func strPtrOrNil(v string) *string {
if v == "" {
return nil
}
value := v
return &value
}
func int64PtrOrNil(v int64) *int64 {
if v <= 0 {
return nil
}
value := v
return &value
}

View File

@@ -0,0 +1,34 @@
package repo
import (
"context"
"errors"
"github.com/LoveLosita/smartflow/backend/model"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
// SettingsRepo 封装 memory_user_settings 的读写。
type SettingsRepo struct {
db *gorm.DB
}
func NewSettingsRepo(db *gorm.DB) *SettingsRepo {
return &SettingsRepo{db: db}
}
func (r *SettingsRepo) Upsert(ctx context.Context, setting model.MemoryUserSetting) error {
if r == nil || r.db == nil {
return errors.New("memory settings repo is nil")
}
return r.db.WithContext(ctx).Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "user_id"}},
DoUpdates: clause.AssignmentColumns([]string{
"memory_enabled",
"implicit_memory_enabled",
"sensitive_memory_enabled",
"updated_at",
}),
}).Create(&setting).Error
}

View File

@@ -0,0 +1,49 @@
package service
import (
"time"
memorymodel "github.com/LoveLosita/smartflow/backend/memory/model"
"github.com/spf13/viper"
)
// LoadConfigFromViper 读取记忆模块配置并做默认值兜底。
//
// 默认策略:
// 1. temperature/top_p 使用低随机参数,提升可复现性;
// 2. Day1 先提供参数位,不强制所有参数立即生效;
// 3. 轮询与重试参数给出保守默认值,避免对主链路造成压力。
func LoadConfigFromViper() memorymodel.Config {
cfg := memorymodel.Config{
Enabled: viper.GetBool("memory.enabled"),
ExtractPrompt: viper.GetString("memory.prompt.extract"),
DecisionPrompt: viper.GetString("memory.prompt.decision"),
Threshold: viper.GetFloat64("memory.threshold"),
EnableReranker: viper.GetBool("memory.enableReranker"),
LLMTemperature: viper.GetFloat64("memory.llm.temperature"),
LLMTopP: viper.GetFloat64("memory.llm.topP"),
JobMaxRetry: viper.GetInt("memory.job.maxRetry"),
WorkerPollEvery: viper.GetDuration("memory.worker.pollEvery"),
WorkerClaimBatch: viper.GetInt("memory.worker.claimBatch"),
}
if cfg.Threshold <= 0 {
cfg.Threshold = 0.55
}
if cfg.LLMTemperature <= 0 {
cfg.LLMTemperature = 0.1
}
if cfg.LLMTopP <= 0 {
cfg.LLMTopP = 0.2
}
if cfg.JobMaxRetry <= 0 {
cfg.JobMaxRetry = 6
}
if cfg.WorkerPollEvery <= 0 {
cfg.WorkerPollEvery = 2 * time.Second
}
if cfg.WorkerClaimBatch <= 0 {
cfg.WorkerClaimBatch = 1
}
return cfg
}

View File

@@ -0,0 +1,33 @@
package service
import (
"context"
"errors"
memorymodel "github.com/LoveLosita/smartflow/backend/memory/model"
memoryrepo "github.com/LoveLosita/smartflow/backend/memory/repo"
)
// EnqueueService 是 Day1 的“任务入队门面”。
//
// 职责边界:
// 1. 只负责把抽取请求入 memory_jobs
// 2. 不负责执行抽取、不负责写 memory_items。
type EnqueueService struct {
jobRepo *memoryrepo.JobRepo
}
func NewEnqueueService(jobRepo *memoryrepo.JobRepo) *EnqueueService {
return &EnqueueService{jobRepo: jobRepo}
}
func (s *EnqueueService) EnqueueExtractJob(
ctx context.Context,
payload memorymodel.ExtractJobPayload,
sourceEventID string,
) error {
if s == nil || s.jobRepo == nil {
return errors.New("memory enqueue service is nil")
}
return s.jobRepo.CreatePendingExtractJob(ctx, payload, sourceEventID)
}

View File

@@ -0,0 +1,104 @@
package utils
import (
"encoding/json"
"errors"
"regexp"
"strings"
)
var fencedJSONPattern = regexp.MustCompile("(?s)```(?:json)?\\s*([\\[{].*[\\]}])\\s*```")
// ExtractJSON 从模型输出中提取 JSON 文本(兼容代码块包裹)。
//
// 步骤:
// 1. 先判断整段文本是否本身就是合法 JSON
// 2. 再尝试匹配 ```json ... ``` 代码块;
// 3. 最后做一次“首个 JSON 对象/数组”扫描提取。
func ExtractJSON(raw string) (string, error) {
trimmed := strings.TrimSpace(raw)
if trimmed == "" {
return "", errors.New("empty model output")
}
// 1. 直接 JSON 命中时,避免做额外启发式扫描。
if json.Valid([]byte(trimmed)) {
return trimmed, nil
}
// 2. 兼容 markdown 代码块包裹 JSON。
matches := fencedJSONPattern.FindStringSubmatch(trimmed)
if len(matches) > 1 {
candidate := strings.TrimSpace(matches[1])
if json.Valid([]byte(candidate)) {
return candidate, nil
}
}
// 3. 兜底扫描首个完整 JSON 片段,尽量提升容错能力。
if candidate, ok := findFirstJSONSegment(trimmed); ok {
return candidate, nil
}
return "", errors.New("json not found in model output")
}
func findFirstJSONSegment(raw string) (string, bool) {
start := -1
var open, close rune
for i, ch := range raw {
if ch == '{' {
start = i
open = '{'
close = '}'
break
}
if ch == '[' {
start = i
open = '['
close = ']'
break
}
}
if start < 0 {
return "", false
}
depth := 0
inString := false
escaped := false
for i, ch := range raw[start:] {
if inString {
if escaped {
escaped = false
continue
}
if ch == '\\' {
escaped = true
continue
}
if ch == '"' {
inString = false
}
continue
}
if ch == '"' {
inString = true
continue
}
if ch == open {
depth++
continue
}
if ch == close {
depth--
if depth == 0 {
candidate := strings.TrimSpace(raw[start : start+i+1])
if json.Valid([]byte(candidate)) {
return candidate, true
}
return "", false
}
}
}
return "", false
}

View File

@@ -0,0 +1,102 @@
package utils
import (
"crypto/sha256"
"encoding/hex"
"fmt"
"strings"
memorymodel "github.com/LoveLosita/smartflow/backend/memory/model"
)
const (
maxTitleLength = 64
maxContentLength = 1000
)
// NormalizeFacts 对候选事实做标准化与过滤。
//
// 步骤:
// 1. 标准化 memory_type 与文本字段,丢弃空值和非法类型;
// 2. 对超长内容截断,避免脏数据污染后续链路;
// 3. 基于“类型+标准化内容”做去重,避免同一轮重复写入。
func NormalizeFacts(candidates []memorymodel.FactCandidate) []memorymodel.NormalizedFact {
if len(candidates) == 0 {
return nil
}
result := make([]memorymodel.NormalizedFact, 0, len(candidates))
seen := make(map[string]struct{}, len(candidates))
for _, candidate := range candidates {
memoryType := memorymodel.NormalizeMemoryType(candidate.MemoryType)
if memoryType == "" {
continue
}
content := normalizeWhitespace(candidate.Content)
if content == "" {
continue
}
content = truncateByRune(content, maxContentLength)
title := normalizeWhitespace(candidate.Title)
if title == "" {
title = truncateByRune(content, maxTitleLength)
}
title = truncateByRune(title, maxTitleLength)
confidence := clamp01(candidate.Confidence)
if confidence == 0 {
confidence = 0.6
}
normalizedContent := strings.ToLower(content)
contentHash := hashContent(memoryType, normalizedContent)
dedupKey := fmt.Sprintf("%s:%s", memoryType, contentHash)
if _, exists := seen[dedupKey]; exists {
continue
}
seen[dedupKey] = struct{}{}
result = append(result, memorymodel.NormalizedFact{
MemoryType: memoryType,
Title: title,
Content: content,
NormalizedContent: normalizedContent,
ContentHash: contentHash,
Confidence: confidence,
IsExplicit: candidate.IsExplicit,
})
}
return result
}
func normalizeWhitespace(raw string) string {
return strings.Join(strings.Fields(strings.TrimSpace(raw)), " ")
}
func truncateByRune(raw string, max int) string {
if max <= 0 {
return ""
}
runes := []rune(raw)
if len(runes) <= max {
return raw
}
return string(runes[:max])
}
func clamp01(v float64) float64 {
if v < 0 {
return 0
}
if v > 1 {
return 1
}
return v
}
func hashContent(memoryType, normalizedContent string) string {
sum := sha256.Sum256([]byte(memoryType + "::" + normalizedContent))
return hex.EncodeToString(sum[:])
}

View File

@@ -0,0 +1,22 @@
package worker
import (
"context"
memorymodel "github.com/LoveLosita/smartflow/backend/memory/model"
memoryorchestrator "github.com/LoveLosita/smartflow/backend/memory/orchestrator"
)
// Extractor 是 worker 抽取依赖接口。
//
// 设计目的:
// 1. Day1 先接 mock 编排器跑通状态机;
// 2. Day2/Day3 可无缝替换为真实 LLM 抽取实现。
type Extractor interface {
ExtractFacts(ctx context.Context, payload memorymodel.ExtractJobPayload) ([]memorymodel.NormalizedFact, error)
}
// NewMockExtractor 返回 Day1 默认 mock 抽取器。
func NewMockExtractor() Extractor {
return memoryorchestrator.NewWriteOrchestrator()
}

View File

@@ -0,0 +1,95 @@
package worker
import (
"context"
"encoding/json"
"errors"
"fmt"
"log"
"time"
memorymodel "github.com/LoveLosita/smartflow/backend/memory/model"
memoryrepo "github.com/LoveLosita/smartflow/backend/memory/repo"
"github.com/LoveLosita/smartflow/backend/model"
)
// RunOnceResult 描述一次手工触发执行结果。
type RunOnceResult struct {
Claimed bool
JobID int64
Status string
Facts int
}
// Runner 是 Day1 首版任务执行器。
//
// 职责边界:
// 1. 只负责推进 memory_jobs 状态机;
// 2. Day1 不做 memory_items 真正落库,仅做 mock 抽取与状态推进。
type Runner struct {
jobRepo *memoryrepo.JobRepo
extractor Extractor
logger *log.Logger
}
func NewRunner(jobRepo *memoryrepo.JobRepo, extractor Extractor) *Runner {
return &Runner{
jobRepo: jobRepo,
extractor: extractor,
logger: log.Default(),
}
}
// RunOnce 手工执行一次任务抢占与处理。
//
// 返回语义:
// 1. Claimed=false 表示当前无可执行任务;
// 2. Claimed=true 且 Status=success/failed/dead 表示状态已推进完成。
func (r *Runner) RunOnce(ctx context.Context) (*RunOnceResult, error) {
if r == nil || r.jobRepo == nil || r.extractor == nil {
return nil, errors.New("memory worker runner is not initialized")
}
// 1. 抢占一条可执行任务,避免并发 worker 重复处理同一记录。
job, err := r.jobRepo.ClaimNextRunnableExtractJob(ctx, time.Now())
if err != nil {
return nil, err
}
if job == nil {
return &RunOnceResult{Claimed: false}, nil
}
result := &RunOnceResult{
Claimed: true,
JobID: job.ID,
Status: model.MemoryJobStatusProcessing,
Facts: 0,
}
// 2. 解析 payload_json。解析失败属于数据质量问题走失败重试并打日志。
var payload memorymodel.ExtractJobPayload
if err = json.Unmarshal([]byte(job.PayloadJSON), &payload); err != nil {
failReason := fmt.Sprintf("解析任务载荷失败: %v", err)
_ = r.jobRepo.MarkFailed(ctx, job.ID, failReason)
result.Status = model.MemoryJobStatusFailed
return result, nil
}
// 3. 调用抽取器执行 mock 抽取。Day1 先保证“能推进状态”,不引入重计算。
facts, extractErr := r.extractor.ExtractFacts(ctx, payload)
if extractErr != nil {
failReason := fmt.Sprintf("抽取执行失败: %v", extractErr)
_ = r.jobRepo.MarkFailed(ctx, job.ID, failReason)
result.Status = model.MemoryJobStatusFailed
return result, nil
}
// 4. 抽取成功后把任务置为 success。
if err = r.jobRepo.MarkSuccess(ctx, job.ID); err != nil {
return nil, err
}
result.Status = model.MemoryJobStatusSuccess
result.Facts = len(facts)
r.logger.Printf("memory worker run once success: job_id=%d extracted_facts=%d", job.ID, len(facts))
return result, nil
}

165
backend/model/memory.go Normal file
View File

@@ -0,0 +1,165 @@
package model
import "time"
const (
// MemoryItemStatusActive 表示记忆条目可参与检索与注入。
MemoryItemStatusActive = "active"
// MemoryItemStatusArchived 表示记忆条目被归档,不再默认参与注入。
MemoryItemStatusArchived = "archived"
// MemoryItemStatusDeleted 表示记忆条目已软删除。
MemoryItemStatusDeleted = "deleted"
)
const (
// MemoryJobTypeExtract 表示“候选事实抽取”任务。
MemoryJobTypeExtract = "extract"
// MemoryJobTypeEmbed 表示“向量化同步”任务Day1 仅预留)。
MemoryJobTypeEmbed = "embed"
// MemoryJobTypeReconcile 表示“冲突消解”任务Day1 仅预留)。
MemoryJobTypeReconcile = "reconcile"
)
const (
// MemoryJobStatusPending 表示任务待执行。
MemoryJobStatusPending = "pending"
// MemoryJobStatusProcessing 表示任务执行中。
MemoryJobStatusProcessing = "processing"
// MemoryJobStatusSuccess 表示任务执行成功(最终态)。
MemoryJobStatusSuccess = "success"
// MemoryJobStatusFailed 表示任务执行失败但可重试。
MemoryJobStatusFailed = "failed"
// MemoryJobStatusDead 表示任务不可恢复失败(最终态)。
MemoryJobStatusDead = "dead"
)
// MemoryItem 对应 memory_items 表,用于保存长期可注入记忆。
//
// 职责边界:
// 1. 该模型只定义存储结构,不承载抽取/决策业务逻辑;
// 2. Day1 先建表与基础字段Day2 再补读取注入链路;
// 3. 向量字段vector_status/vector_id仅做状态桥接不等于向量库真值。
type MemoryItem struct {
ID int64 `gorm:"column:id;primaryKey;autoIncrement"`
UserID int `gorm:"column:user_id;not null;index:idx_memory_items_user_status_type,priority:1;index:idx_memory_items_user_conv_status,priority:1;index:idx_memory_items_user_asst_run_status,priority:1;index:idx_memory_items_user_type_hash,priority:1;comment:用户ID"`
ConversationID *string `gorm:"column:conversation_id;type:varchar(64);index:idx_memory_items_user_conv_status,priority:2;comment:会话ID"`
AssistantID *string `gorm:"column:assistant_id;type:varchar(64);index:idx_memory_items_user_asst_run_status,priority:2;comment:助手ID"`
RunID *string `gorm:"column:run_id;type:varchar(64);index:idx_memory_items_user_asst_run_status,priority:3;comment:运行ID"`
MemoryType string `gorm:"column:memory_type;type:varchar(32);not null;index:idx_memory_items_user_status_type,priority:3;index:idx_memory_items_user_type_hash,priority:2;comment:preference/constraint/fact/todo_hint"`
Title string `gorm:"column:title;type:varchar(128);not null;comment:记忆标题"`
Content string `gorm:"column:content;type:text;not null;comment:记忆内容"`
NormalizedContent *string `gorm:"column:normalized_content;type:text;comment:标准化内容"`
ContentHash *string `gorm:"column:content_hash;type:varchar(64);index:idx_memory_items_user_type_hash,priority:3;comment:幂等去重哈希"`
Confidence float64 `gorm:"column:confidence;type:decimal(5,4);not null;default:0.6;comment:置信度"`
Importance float64 `gorm:"column:importance;type:decimal(5,4);not null;default:0.5;comment:重要度"`
SensitivityLevel int `gorm:"column:sensitivity_level;not null;default:0;comment:敏感级别"`
SourceMessageID *int64 `gorm:"column:source_message_id;index:idx_memory_items_source_message;comment:来源消息ID"`
SourceEventID *string `gorm:"column:source_event_id;type:varchar(64);comment:来源事件ID"`
IsExplicit bool `gorm:"column:is_explicit;not null;default:false;comment:是否显式记忆"`
Status string `gorm:"column:status;type:varchar(16);not null;default:active;index:idx_memory_items_user_status_type,priority:2;index:idx_memory_items_user_conv_status,priority:3;index:idx_memory_items_user_asst_run_status,priority:4;comment:active/archived/deleted"`
TTLAt *time.Time `gorm:"column:ttl_at;index:idx_memory_items_ttl;comment:过期时间"`
LastAccessAt *time.Time `gorm:"column:last_access_at;comment:最后访问时间"`
CreatedAt *time.Time `gorm:"column:created_at;autoCreateTime"`
UpdatedAt *time.Time `gorm:"column:updated_at;autoUpdateTime"`
VectorStatus string `gorm:"column:vector_status;type:varchar(16);not null;default:pending;comment:pending/synced/failed"`
VectorID *string `gorm:"column:vector_id;type:varchar(128);comment:向量库映射ID"`
}
func (MemoryItem) TableName() string {
return "memory_items"
}
// MemoryJob 对应 memory_jobs 表,用于承接异步任务。
//
// 职责边界:
// 1. 该表是“可重试状态机”,不是业务事实库;
// 2. payload_json 只存任务执行最小上下文;
// 3. status/retry_count/next_retry_at 组合定义可重试行为。
type MemoryJob struct {
ID int64 `gorm:"column:id;primaryKey;autoIncrement"`
UserID int `gorm:"column:user_id;not null;index:idx_memory_jobs_user_created,priority:1;comment:用户ID"`
ConversationID *string `gorm:"column:conversation_id;type:varchar(64);comment:会话ID"`
SourceMessageID *int64 `gorm:"column:source_message_id;comment:来源消息ID"`
SourceEventID *string `gorm:"column:source_event_id;type:varchar(64);index:idx_memory_jobs_source_event;comment:来源事件ID"`
JobType string `gorm:"column:job_type;type:varchar(32);not null;comment:extract/embed/reconcile"`
IdempotencyKey string `gorm:"column:idempotency_key;type:varchar(128);not null;uniqueIndex:uk_memory_jobs_idempotency;comment:幂等键"`
PayloadJSON string `gorm:"column:payload_json;type:longtext;not null;comment:任务载荷JSON"`
Status string `gorm:"column:status;type:varchar(16);not null;index:idx_memory_jobs_status_next,priority:1;comment:pending/processing/success/failed/dead"`
RetryCount int `gorm:"column:retry_count;not null;default:0;comment:已重试次数"`
MaxRetry int `gorm:"column:max_retry;not null;default:6;comment:最大重试次数"`
NextRetryAt *time.Time `gorm:"column:next_retry_at;index:idx_memory_jobs_status_next,priority:2;comment:下次重试时间"`
LastError *string `gorm:"column:last_error;type:varchar(2000);comment:最后错误"`
CreatedAt *time.Time `gorm:"column:created_at;autoCreateTime;index:idx_memory_jobs_user_created,priority:2"`
UpdatedAt *time.Time `gorm:"column:updated_at;autoUpdateTime"`
}
func (MemoryJob) TableName() string {
return "memory_jobs"
}
// MemoryAuditLog 对应 memory_audit_logs 表,用于记忆变更审计。
type MemoryAuditLog struct {
ID int64 `gorm:"column:id;primaryKey;autoIncrement"`
MemoryID int64 `gorm:"column:memory_id;not null;index:idx_memory_audit_memory_id;comment:记忆ID"`
UserID int `gorm:"column:user_id;not null;index:idx_memory_audit_user_id;comment:用户ID"`
Operation string `gorm:"column:operation;type:varchar(32);not null;comment:create/update/archive/delete/restore"`
OperatorType string `gorm:"column:operator_type;type:varchar(16);not null;comment:system/user"`
Reason string `gorm:"column:reason;type:varchar(255);not null;default:'';comment:操作原因"`
BeforeJSON *string `gorm:"column:before_json;type:longtext;comment:变更前快照"`
AfterJSON *string `gorm:"column:after_json;type:longtext;comment:变更后快照"`
CreatedAt *time.Time `gorm:"column:created_at;autoCreateTime"`
}
func (MemoryAuditLog) TableName() string {
return "memory_audit_logs"
}
// MemoryUserSetting 对应 memory_user_settings 表,用于用户记忆开关控制。
type MemoryUserSetting struct {
UserID int `gorm:"column:user_id;primaryKey;comment:用户ID"`
MemoryEnabled bool `gorm:"column:memory_enabled;not null;default:true;comment:总开关"`
ImplicitMemoryEnabled bool `gorm:"column:implicit_memory_enabled;not null;default:true;comment:隐式记忆开关"`
SensitiveMemoryEnabled bool `gorm:"column:sensitive_memory_enabled;not null;default:false;comment:敏感记忆开关"`
UpdatedAt *time.Time `gorm:"column:updated_at;autoUpdateTime"`
}
func (MemoryUserSetting) TableName() string {
return "memory_user_settings"
}
// MemoryExtractRequestedPayload 是 memory.extract.requested(v1) 事件载荷。
//
// 说明:
// 1. Day1 先承载最小可执行字段;
// 2. assistant_id/run_id/source_message_id/trace_id 允许为空,后续链路补齐;
// 3. idempotency_key 必填,用于 memory_jobs 去重与无副作用消费。
type MemoryExtractRequestedPayload struct {
UserID int `json:"user_id"`
ConversationID string `json:"conversation_id"`
AssistantID string `json:"assistant_id,omitempty"`
RunID string `json:"run_id,omitempty"`
SourceMessageID int64 `json:"source_message_id,omitempty"`
SourceRole string `json:"source_role"`
SourceText string `json:"source_text"`
OccurredAt time.Time `json:"occurred_at"`
TraceID string `json:"trace_id,omitempty"`
IdempotencyKey string `json:"idempotency_key"`
}

View File

@@ -0,0 +1,142 @@
# WebSearch 两阶段实施计划newAgent
## 1. 目标与范围
本文用于把 `newAgent` 的 WebSearch 能力按两阶段落地:
1. 第一阶段:先接入可用的检索与抓取能力(低风险、快交付)。
2. 第二阶段:在第一阶段基础上升级为 WebRAG 语义召回链路(提升复杂问题命中率与可解释性)。
约束:
1. 不走 `infra/smartflow-mcp-server`,直接走 `newAgent/tools` 工具注册链路。
2. 保持现有执行模式不变:读操作 `action=continue + tool_call`
3. 第一阶段只接单供应商;第二阶段再考虑 provider fallback。
---
## 2. 第一阶段V1WebSearch + 简单抓取
### 2.1 交付目标
让模型可以:
1. 通过 `web_search` 获得结构化检索结果标题、摘要、URL、来源域名、时间
2. 通过 `web_fetch` 拉取指定 URL 正文并做最小清洗。
3. 在不改主流程的前提下,把结果作为标准 `tool observation` 写回历史。
### 2.2 计划新增工具
1. `web_search`
- 输入:`query``top_k``domain_allow``recency_days` 等。
- 输出JSON 字符串(`tool``query``count``items[]`)。
2. `web_fetch`
- 输入:`url``max_chars`
- 输出JSON 字符串(`tool``url``title``content``truncated`)。
### 2.3 代码落点
新增文件:
1. `backend/newAgent/tools/web_tools.go`:工具参数解析、输出组装、错误兜底。
2. `backend/newAgent/tools/web_provider.go`:搜索供应商抽象接口与通用数据结构。
3. `backend/newAgent/tools/web_provider_tavily.go`(或 `web_provider_brave.go`):首个 provider 实现。
4. `backend/newAgent/tools/web_fetcher.go`URL 抓取与 HTML 最小清洗。
修改文件:
1. `backend/newAgent/tools/registry.go`:注册 `web_search``web_fetch` 两个读工具。
2. `backend/cmd/start.go`:初始化 provider 配置并注入 registry或通过包级配置读取
3. `backend/newAgent/prompt/execute_context.go`:补充新工具的 schema 说明与示例。
### 2.4 V1 验收标准
1. 模型能稳定调用 `web_search` 并拿到可解析 JSON 结果。
2. `web_fetch` 在正文可达时返回正文,在失败时返回明确错误码与原因。
3. 工具超时、429、5xx 均不会打断主流程,只返回可恢复 observation。
4. 日志可定位query、tool、耗时、结果数、失败原因。
---
## 3. 第二阶段V2WebRAG 语义召回
### 3.1 交付目标
新增 `web_rag_search`,把“检索 + 抓取 + 分块 + 召回 + 重排 + 证据返回”收敛为一个读工具,提升复杂问答质量。
### 3.2 链路设计
1. 查询改写:把用户问题改写为 1~3 个检索子查询。
2. WebSearch 召回:拿到候选 URL 集合。
3. 抓取清洗:抽正文,去噪。
4. 分块:按段落与 token 预算切块。
5. 召回:向量召回 + 关键词召回(混合召回)。
6. 重排:按 query 相关性重排 chunk。
7. 输出:返回答案所需证据片段、来源 URL、片段得分。
### 3.3 代码落点
新增文件:
1. `backend/newAgent/tools/web_rag_tools.go``web_rag_search` 工具入口。
2. `backend/newAgent/tools/web_rag_chunker.go`:清洗后分块。
3. `backend/newAgent/tools/web_rag_retriever.go`:混合召回。
4. `backend/newAgent/tools/web_rag_rerank.go`:重排层。
5. `backend/newAgent/tools/web_rag_store.go`:会话级索引缓存(先内存/Redis TTL
修改文件:
1. `backend/newAgent/tools/registry.go`:注册 `web_rag_search`
2. `backend/newAgent/prompt/execute_context.go`:增加 `web_rag_search` 使用规范。
### 3.4 V2 验收标准
1. 同类复杂问题下,回答引用质量和相关性明显高于 V1。
2. 返回至少包含:`answer_evidence[]`(片段+URL+score
3. 召回或重排失败时可降级到 V1`web_search + web_fetch`)路径。
4. 提供基础评估指标:命中率、延迟、成本、失败率。
---
## 4. 与记忆系统的关系
`WebRAG` 与记忆系统 RAG 高度重合,建议“共用内核、分语料适配”:
1. 共用chunk / embed / retrieve / rerank 的通用接口与实现。
2. 分开:`MemoryCorpus`(私有数据)与 `WebCorpus`(公网数据)的数据源适配层。
3. 在工具层保持两个入口:`memory_search``web_rag_search`,返回结构尽量统一。
---
## 5. 上线顺序与回滚策略
### 5.1 上线顺序
1. 先灰度 V1仅开放 `web_search``web_fetch`
2. 观察稳定性与成本后再灰度 V2`web_rag_search`
3. V2 稳定后再考虑 provider fallback 与更长周期缓存。
### 5.2 回滚策略
1. `web_rag_search` 异常时,快速降级为 V1 工具集。
2. V1 供应商异常时,返回“检索暂不可用”的结构化 observation不阻断主流程。
3. 保留 feature flag按工具级别开关支持秒级关闭。
---
## 6. 风险清单
1. 供应商配额/限流导致查询失败。
2. 页面反爬与正文抽取质量不稳定。
3. RAG 链路成本上升(抓取+embedding+重排)。
4. 引用片段与最终答案不一致(需要强制证据对齐策略)。
---
## 7. 里程碑建议
1. M11~2 天V1 工具跑通,联调 Execute 节点可调用。
2. M22~4 天V1 稳定性优化(超时/限流/日志/错误码)。
3. M34~7 天V2 WebRAG MVP混合召回+基础重排+证据输出)。
4. M4后续统一 RAG Core打通记忆系统复用。

View File

@@ -1,814 +0,0 @@
# Handoff工具研究与运行态重置
以下内容可直接交给下一位助理继续做。
本文档更新时间2026-04-08
## 0. 当前结论先说清
当前可以明确分成三段看:
1. 第 1-2 阶段已完成
粗排链路和 `abort -> deliver` 正式终止协议已打通,真实链路已验证可跑。
2. 第 3 阶段第一版已落地
`execute` 上下文已改成固定 4 消息结构,并接入当轮 ReAct 窗口压缩(按工具去重保留最新 observation
工具结果在执行链路中已改为不截断prompt 也已加硬约束JSON、参数名、读写动作、重复读约束等
3. 当前主矛盾已转移到“工具收敛能力”
最新日志显示:不是上下文失控,而是工具输出对“何时停止微调”支持不足,导致模型持续 `list_tasks/find_first_free/move` 小步循环。
一句话总结:
- 粗排和执行框架已经能跑;
- 上下文瘦身第一版已经到位;
- 下一棒应优先改工具层的“冲突表达 + 候选位质量 + 结束判据”,而不是再回头补粗排。
---
## 1. 用户已经明确确认的业务语义
这些结论已经和用户对齐,后续不要再摇摆:
1. 第一轮实现优先参考旧 agent 链路,不能闭门造车。
2. 对排程场景来说LLM 的角色是“粗排后的优化器”,不是“粗排漏排后的人工补排器”。
3. 如果粗排完成后仍有真实 `pending` 任务,这不是正常状态,而是异常状态。
正确处理是:
- 直接终止本轮
- 明确报错
- 不再引导 LLM 去一个个 `place`
4. `always_execute`、是否自动放行、是否写库,这些都是后端执行层语义,不应该写进 prompt 让 LLM 判断。
5. prompt 必须可插拔,不能写死成“排程专用 prompt”。
6. prompt 必须保留明确编号结构,例如 `1. / 1.1 / 2.1`,不能写成一坨说明文。
7. prompt 里要保留最小可用 JSON 示例,否则 LLM 很容易输出跑偏。
---
## 2. 第 1 阶段:粗排链路修复,已完成
### 2.1 已完成的核心修复
#### A. 粗排结果来源修正
位置:
- `backend/service/agentsvc/agent_newagent.go`
已做:
- `makeRoughBuildFunc()` 不再只看 `[]TaskClassItem`
- 改为使用 `HybridScheduleWithPlanMulti()` 返回的 `[]HybridScheduleEntry`
- 只提取 `status="suggested"``task_item_id>0` 的条目
意义:
- 旧问题是:只会拿到有 `EmbeddedTime` 的一部分结果,普通 suggested 条目会丢
- 现在粗排 placement 来源和旧 agent 主链路一致
#### B. 作用域修正:只按本轮 `task_class_ids` 看 pending
位置:
- `backend/newAgent/tools/status.go`
- `backend/newAgent/model/graph_run_state.go`
- `backend/newAgent/node/rough_build.go`
已做:
- 新增按 `task_class_ids` 判断 task 是否属于本轮范围
- `EnsureScheduleState()` 会对 `ScheduleState` / `OriginalScheduleState` 做域内裁剪
- `countPendingTasks()` 只统计本轮任务类范围内的真实 pending
意义:
- 修掉“用户所有任务类的 pending 被误算进本轮粗排结果”的问题
#### C. 窗口修正ScheduleState 的 DayMapping 改为优先按本轮任务类加载
位置:
- `backend/newAgent/model/state_store.go`
- `backend/newAgent/model/graph_run_state.go`
- `backend/conv/schedule_provider.go`
已做:
- 新增可选接口 `ScopedScheduleStateProvider`
- `AgentGraphState.EnsureScheduleState()` 首次加载时,若 provider 支持 scoped 加载,则优先走 `LoadScheduleStateForTaskClasses()`
- `ScheduleProvider` 实现了按本轮 `task_class_ids` 精确加载 `ScheduleState`
- `buildWindowFromTaskClasses()` 改成逐个任务类做日期转换,坏日期/超学期日期会被忽略,不再把整轮窗口拖坏
这次修的是一个真实 bug
- 用户真实日志里曾出现:
- `placements=44`
- `pending_in_scope=44`
- 这说明 placement 已经出来了,但一个都没写回 state
- 根因不是粗排没算出来,而是 `DayMapping` 窗口被全量任务类脏日期污染,导致 `WeekDayToDay()` 映射失败
现在这个问题已经修掉,用户已确认“粗排跑通了”。
#### D. 粗排落位语义改成 `suggested`
位置:
- `backend/newAgent/node/rough_build.go`
- `backend/newAgent/tools/read_tools.go`
- `backend/newAgent/tools/read_helpers.go`
- `backend/newAgent/tools/write_tools.go`
- `backend/newAgent/tools/SCHEDULE_TOOLS.md`
已做:
- 粗排成功写回后,任务状态会从真实 pending 变成 `suggested`
- 不再使用“pending + slots”的旧兼容语义作为主路径
- 工具读写层已经统一支持 `existing / suggested / pending`
意义:
- 粗排结果是“建议落位”,后续应使用 `move / swap / unplace`
- 不应该再让 LLM 把这些任务当作待补排任务来 `place`
#### E. 粗排节点增加了可观测 debug
位置:
- `backend/newAgent/node/rough_build.go`
现在会打这些日志:
- `placements`
- `applied`
- `day_mapping_miss`
- `task_item_match_miss`
- `pending_in_scope`
- `window_days`
- 少量 miss 样本
意义:
- 下次如果粗排再挂,不需要再靠猜
- 能直接区分是:
- DayMapping 没命中
- 还是 `task_item_id` 没命中
### 2.2 粗排阶段当前的真实状态
当前真实结论:
- 粗排已经能跑通
- 用户已在真实链路确认这一点
所以粗排这条线当前不是主矛盾了。
---
## 3. 第 2 阶段:正式 abort/terminal 协议,已完成
### 3.1 已完成的核心改动
#### A. CommonState 引入统一 terminal outcome
位置:
- `backend/newAgent/model/common_state.go`
已做:
- 新增:
- `FlowTerminalStatus`
- `FlowTerminalOutcome`
- 新增方法:
- `Abort(...)`
- `Exhaust(...)`
- `HasTerminalOutcome()`
- `TerminalStatus()`
- `IsCompleted()`
- `IsAborted()`
- `IsExhaustedTerminal()`
- `ClearTerminalOutcome()`
意义:
- 后续 graph / execute / deliver 不再各自猜“当前算不算结束”
- 统一围绕一份终止结果收口
#### B. execute contract 正式支持 `abort`
位置:
- `backend/newAgent/model/execute_contract.go`
- `backend/newAgent/prompt/execute.go`
已做:
- 新增 `ExecuteActionAbort = "abort"`
- `ExecuteDecision` 新增 `Abort *AbortIntent`
- `Validate()` 已补齐互斥/必填规则
- execute prompt / react prompt 都已加入 `abort` 契约和 JSON 示例
#### C. rough_build 异常会正式 abort而不是让 LLM 补排
位置:
- `backend/newAgent/node/rough_build.go`
已做:
- 若粗排后仍有真实 pending
- 发失败状态
- `flowState.Abort(...)`
- 直接交给 deliver 收口
这部分已经和用户确认过业务语义,是对的。
#### D. execute 层接入 abort / exhausted 正式语义
位置:
- `backend/newAgent/node/execute.go`
已做:
- execute 轮次耗尽时,不再伪装成 `done`
- 改为 `flowState.Exhaust(...)`
- 新增 `handleExecuteActionAbort(...)`
- `abort` 不在 execute 内直接最终答复,而是交给 deliver
#### E. graph 路由改成看正式 terminal outcome
位置:
- `backend/newAgent/graph/common_graph.go`
已做:
- `RoughBuild -> Execute` 改为 branch
- 粗排异常终止时可以直接 `RoughBuild -> Deliver`
- `branchAfterExecute()` 不再因为“刚好最后一轮”就提前 deliver
- 必须等 execute 正式写入终止结果
#### F. deliver 收口统一按 terminal outcome 输出
位置:
- `backend/newAgent/node/deliver.go`
- `backend/newAgent/node/agent_nodes.go`
已做:
- deliver 不再无脑 `Done()`
- deliver summary 会优先看 terminal outcome
- 新增:
- `buildAbortSummary`
- `buildExhaustedSummary`
- 只有 `completed` 路径才写 schedule preview
- `aborted / exhausted` 跳过 preview
### 3.2 第 2 阶段的验证情况
本地已跑通过:
```bash
go test ./conv ./newAgent/node ./newAgent/model ./newAgent/graph ./newAgent/tools ./service/agentsvc
```
注意:
- 每次 `go test` 后都已清理根目录 `.gocache`
- 过程中使用过临时 `*_test.go` 验证窗口问题和粗排写回问题,跑完后已全部删除
---
## 4. 非常重要:当前“上下文”到底改没改
这个点后续助理很容易误判,这里单独写清楚。
### 4.1 没做的事
这轮**没有**做第 3 阶段那种“系统性的上下文瘦身”。
也就是说,以下这些东西没有被系统重构:
- `ConversationContext` 的整体装配框架
- 历史恢复/回填主流程
- 历史裁剪策略
- token budget 接入
- “把 history 从流水账改成摘要流”的体系化工程
### 4.2 但确实改了会进入模型上下文的内容
为了适配新的粗排/abort 语义,这轮确实改了“上下文内容本身”,主要包括:
1. 粗排完成后的 pinned block 文案变了
位置:
- `backend/newAgent/node/rough_build.go`
2. execute 的提示词/输出契约变了
位置:
- `backend/newAgent/prompt/execute.go`
3. 工具层的状态语义从 `pending/existing` 扩成了 `pending/suggested/existing`
位置:
- `backend/newAgent/tools/read_tools.go`
- `backend/newAgent/tools/write_tools.go`
4. state summary 会展示 terminal outcome
位置:
- `backend/newAgent/prompt/base.go`
### 4.3 所以当前准确结论是
可以这样理解:
- **没有做上下文瘦身**
- **但确实改了进入模型的上下文内容**
所以如果下游 `execute` 现在表现依旧很差,不能简单说:
- “是第 1-2 阶段没做好”
更准确的说法是:
- 第 1-2 阶段把粗排和终止语义打通了
- 但下游本来就被上下文噪音淹着
- 现在必须进入第 3 阶段,先瘦身,再评价整体效果
---
## 5. 当前主矛盾:不是粗排,而是 execute 上下文过胖
用户已经明确表达:
- “下游本来就基本跑不通”
- “必须要瘦身上下文才能看出整体的效果”
这意味着明天接手时不要再把精力放在:
- 继续 patch 粗排小问题
- 继续围绕 abort 收口补语义
下一步的主任务必须切换为:
- 第 3 阶段:上下文瘦身
- 第 4 阶段prompt 结构重构
---
## 6. 第 3 阶段:上下文瘦身,明天应该怎么做
### 6.1 目标
目标不是“再写一版更长的 prompt”而是
- 让 execute 每轮看到的输入,变成“足够决策、但不淹没”的信息量
换句话说,要把现在的:
- `system + tool summary + 全量 history + pinned + runtime prompt`
改造成更可控的:
- 少量稳定规则
- 少量当前必要状态
- 少量最近真实有效观察
### 6.2 第 3 阶段必须完成的事
#### A. 历史去流水账
重点文件:
- `backend/service/agentsvc/agent_newagent.go`
- `backend/newAgent/prompt/base.go`
- `backend/newAgent/node/execute.go`
要做:
- 同工具同参数的重复查询,不保留多份原文
- 旧结果改成摘要
- 只保留最近一条原始结果
典型对象:
- `get_overview`
- `list_tasks`
- `find_free`
#### B. assistant 过程废话不再进入后续模型历史
要处理的典型文本:
- “我先查一下”
- “我接下来会……”
- “我准备先获取……”
这些文本对下一轮决策价值极低,但会稳定吃 token。
#### C. 失败模式保留“摘要”,不要保留整段原始失败链路
例如:
- `place``task_id`
- `find_free``duration`
正确保留方式应该更像:
- 最近失败模式:`place` 缺少 `task_id`
而不是整段原始 tool call / tool result 全保留。
#### D. 缩短 state summary / pinned / runtime prompt 的重叠信息
目前这三层有明显重复:
- `renderStateSummary`
- pinned blocks
- execute runtime user prompt
第 3 阶段要先做减法,减少重复指令。
#### E. 参考旧链路的 token budget
重点参考:
- `backend/service/agentsvc/agent.go`
- `backend/pkg/token_budget.go`
要求:
- 不一定照搬
- 但至少要借鉴旧链路“按预算裁历史”的思路
### 6.3 第 3 阶段推荐顺序
建议按这个顺序做:
1. 先打印/抓取 `BuildExecuteMessages()` 的真实输入样本
2. 再压 `history`
3. 再压 `pinned/state summary/runtime prompt` 的重叠
4. 最后再接 token budget
原因:
- 先把“到底胖在哪”看清楚
- 避免直接上 budget 把有用信息也一起砍掉
### 6.4 第 3 阶段完成标准
至少要满足:
1. execute 首轮 messages 显著变短
2. 同类查询不会反复堆原始结果
3. assistant 流水话大幅减少
4. 最近失败模式还能被模型感知
5. 不破坏第 1-2 阶段已经打通的粗排/abort 语义
---
## 7. 第 4 阶段prompt 结构重构,明天应该怎么做
### 7.1 目标
把当前 execute prompt 从“堆规则 + 堆领域细节 + 堆运行时说明”的混合物,重构成:
1. 通用执行内核
2. 领域模块
3. 运行时任务简报
这是用户已经认可的方向。
### 7.2 建议的三层结构
#### 第一层:通用执行内核
负责:
- agent 身份
- 通用动作协议
- 通用输出字段
- 最小 JSON 示例
这里不要写死排程语义。
#### 第二层:领域模块
对排程领域来说,建议抽成单独模块,至少包含:
- `domain_name`
- `domain_primary_responsibility`
- `domain_out_of_scope`
- `domain_goals`
- `domain_non_goals`
- `tool_catalog_brief`
- `hard_constraints`
- `soft_objectives`
- `abort_conditions`
- `done_conditions`
对“粗排后排程优化”这个领域,必须明确写清:
- 这是优化器,不是补排器
- `pending > 0` 是异常,不是待办
#### 第三层:运行时任务简报
负责:
- 用户原始目标
- 最新补充指令
- 当前阶段
- 当前轮次
- 当前实例级约束
- 最近状态变化
- 最近失败摘要
- 上一次工具结果摘要
- 本轮目标
- 推荐下一步动作
### 7.3 第 4 阶段不要做成什么样
不要做成:
- 换一版更长的 execute prompt
- 继续把排程规则硬编码进通用层
- 继续把运行时状态散落在多个 message 里重复讲
### 7.4 第 4 阶段推荐落地文件
可以考虑在以下位置拆分:
- `backend/newAgent/prompt/base.go`
- `backend/newAgent/prompt/execute.go`
如有必要,可以新增文件,例如:
- `backend/newAgent/prompt/execute_core.go`
- `backend/newAgent/prompt/execute_domain_schedule.go`
- `backend/newAgent/prompt/execute_runtime_brief.go`
### 7.5 第 4 阶段完成标准
至少要满足:
1. 通用执行协议不再写死排程业务
2. 排程语义以领域模块方式注入
3. 运行时信息有单独的结构化简报
4. prompt 保留编号结构
5. prompt 保留最小 JSON 示例
---
## 8. 明天接手时,最重要的判断标准
明天的助理接手后,不要先问:
- “为什么微调效果还是差?”
而要先问:
- “第 3 阶段有没有把上下文瘦下来?”
只有在第 3 阶段做完之后,才适合重新评估:
- execute 是否真正理解了粗排后的 suggested 语义
- prompt 重构是否真的提升了整体表现
- 端到端排程链路是否比之前更稳定
---
## 9. 关键文件清单
### 第 1-2 阶段已改文件
- `backend/conv/schedule_provider.go`
- `backend/newAgent/model/state_store.go`
- `backend/newAgent/model/graph_run_state.go`
- `backend/newAgent/model/common_state.go`
- `backend/newAgent/model/execute_contract.go`
- `backend/newAgent/graph/common_graph.go`
- `backend/newAgent/node/rough_build.go`
- `backend/newAgent/node/execute.go`
- `backend/newAgent/node/deliver.go`
- `backend/newAgent/node/agent_nodes.go`
- `backend/newAgent/prompt/base.go`
- `backend/newAgent/prompt/execute.go`
- `backend/newAgent/tools/read_helpers.go`
- `backend/newAgent/tools/read_tools.go`
- `backend/newAgent/tools/write_tools.go`
- `backend/newAgent/tools/SCHEDULE_TOOLS.md`
- `backend/service/agentsvc/agent_newagent.go`
### 第 3-4 阶段重点关注文件
- `backend/service/agentsvc/agent_newagent.go`
- `backend/newAgent/prompt/base.go`
- `backend/newAgent/prompt/execute.go`
- `backend/newAgent/node/execute.go`
- `backend/pkg/token_budget.go`
- `backend/service/agentsvc/agent.go`
---
## 10. 测试与验证注意事项
1.`go test` 后必须清理项目根目录 `.gocache`
2. 如果为了验证临时补 `*_test.go`,跑完必须删除,不要长期保留
3. 当前用户明确不希望这轮把 agent 写库逻辑接回去,仍然以“内存态运行”为主
4. 所有说明、注释、文档都继续用中文
---
## 11. 一句话交给下一位助理
第 1-2 阶段已经把“粗排接入”和“正式 abort 收口”打通了,粗排真实链路也已经跑通;现在不要再围绕粗排打补丁,直接进入第 3 阶段做 execute 上下文瘦身,再做第 4 阶段 prompt 三层重构,完成后再评估整体链路效果。
---
## 12. 2026-04-08 最新增量交接(以本节为准)
> 本节优先级高于前文历史描述。接手时请先读本节,再看上文细节。
### 12.1 本轮已完成的落地项
1. execute 上下文结构已固定为 4 条消息:
- `message[0]`:固定 prompt规则 + JSON 约束 + 工具简表)
- `message[1]`:历史上下文短摘要(聊天摘要 + 早期 ReAct 摘要)
- `message[2]`:当轮 ReAct Loop 窗口thought/reason + tool_call + observation 绑定)
- `message[3]`:当前执行状态(初始目标、结束判断、非目标)
2. 当轮 ReAct 压缩已接入:
- 窗口内同工具只保留最新 observation 原文;
- 被压缩旧结果替换为“当前工具调用结果过于久远,已经被删除。”
3. execute 输出稳态增强:
- `continue / ask_user / confirm``speak` 时会兜底回退 `reason`
- 工具结果写入 history 前的截断已删除(不再自动裁到 3000 字符)。
4. 工具能力已升级:
- `get_overview` 改为任务视角全量输出(课程仅占位统计,不展开课程明细);
- 新增 `find_first_free``find_free` 保留兼容别名;
- `move / batch_move` 限定仅允许 `suggested`
- `list_tasks` 增加输入约束(`status` 单值、`category` 不接受 task_class_ids 列表)。
5. prompt 与文档同步:
- execute prompt 已切换到 `find_first_free` 表达;
- `SCHEDULE_TOOLS.md` 已同步 `get_overview / find_first_free / move/batch_move` 新语义;
- plan prompt 中读工具示例也已从 `find_free` 更新为 `find_first_free`
### 12.2 最新日志结论(关键)
本轮问题已不是“上下文塞不下”,而是“工具不利于收敛”:
1. `find_first_free` 当前策略过于贪心
只返回最早可用位,模型会持续把任务向前挪,容易出现“局部改善但全局不收口”的微调循环。
2. `query_range` 把“可嵌入共存”和“硬冲突”混合输出
模型会把可嵌入并存也当成冲突,导致不必要移动。
3. 缺少“完成判据工具”
当前只有读写事实工具,没有明确“是否可结束”的评估口径,模型自然倾向继续优化。
4. 写工具冲突口径存在潜在不一致
`findConflict``CanEmbed` 的处理与 `findEmbedHost` 的可嵌入约束并非同一套判据,后续应统一。
### 12.3 下一棒建议优先级(按顺序做)
#### P0必须先做
1. 新增评估类只读工具(建议名:`evaluate_balance`
返回最少三项:
- `hard_conflict_count`
- `load_variance`(或等价离散指标)
- `done_suggestion`(可结束/建议继续 + 原因)
2.`find_first_free` 为“候选集”而非单点
建议支持 `top_k`(默认 3并返回每个候选的
- 位置
- 目标日负载变化
- 是否涉及可嵌入
3.`query_range` 输出结构
必须区分:
- `hard_conflict`
- `embeddable_overlap`
避免模型把可嵌入并存误判为硬冲突。
#### P1P0 后做)
4. prompt 增加收敛指引
明确要求模型在每次写操作后优先调用 `evaluate_balance`,满足条件就 `done`,避免“无限微调”。
### 12.4 接手后的最小验证清单
1. 跑一轮真实 execute确认不会长时间卡在 `list_tasks/find_first_free/move` 循环。
2. 确认 `query_range` 可区分硬冲突与可嵌入并存。
3. 确认 `evaluate_balance` 能触发 `done` 收口。
4. 每次 `go test` 后清理项目根目录 `.gocache`
### 12.5 关键文件(本轮增量相关)
- `backend/newAgent/prompt/execute_context.go`
- `backend/newAgent/prompt/execute.go`
- `backend/newAgent/prompt/plan.go`
- `backend/newAgent/node/execute.go`
- `backend/newAgent/tools/read_tools.go`
- `backend/newAgent/tools/write_tools.go`
- `backend/newAgent/tools/registry.go`
- `backend/newAgent/tools/SCHEDULE_TOOLS.md`
---
## 13. 2026-04-08 下班前交接(本节优先级最高)
> 下一棒主线已经明确:先研究工具收敛能力,再修“新一轮执行前必要参数重置”。
### 13.1 本轮新增落地(已完成)
1. 顺序约束链路已落地:
- 新增 `order_guard` 节点,并接入 graph 分支;
- 默认 `AllowReorder=false` 时,`PhaseDone(completed)` 会先走 `order_guard``deliver`
- 用户明确允许打乱顺序时才放行。
2. `min_context_switch` 工具已接入:
- 新增工具实现与注册;
- execute 层已加护栏:未授权打乱顺序时拒绝执行并返回明确 observation
- prompt / 工具文档已同步“仅用户明确授权才可用”。
3. `execute.go` 乱码坏块已修复:
- 清理了污染字符串、重复 if、结构异常
- 该段目前为单一、可读、可编译分支。
4. 当前分支编译验证通过:
- `go test ./newAgent/... ./logic/...` 已通过;
- 测试后已清理项目根目录 `.gocache`
### 13.2 已确认的两个“未完成关键点”
#### A. msg2/msg3 语义尚未改到目标形态
现状:
- execute 仍是固定 4 消息骨架(`msg2=当轮 ReAct 窗口``msg3=执行状态锚点`
- “当轮 ReAct 结束后降级为普通历史并走统一 token 裁剪”还没真正落地;
- `ConversationContext.AppendHistory` 本身不做裁剪,统一裁剪链路也尚未接入。
相关文件:
- `backend/newAgent/prompt/execute_context.go`
- `backend/newAgent/prompt/base.go`
- `backend/newAgent/model/conversation_context.go`
#### B. round 未在“新一轮开始”自动重置
现状:
- `RoundUsed` 只在 `NextRound()` 累加;
- `StartDirectExecute()` 不会重置 `RoundUsed`
- 快照恢复会带回旧 `RoundUsed`,所以“对话已结束但 round 没清零”可复现。
相关文件:
- `backend/newAgent/model/common_state.go`
- `backend/service/agentsvc/agent_newagent.go`
### 13.3 下一步实施建议(按此顺序)
#### P0运行态重置必须先做
目标:不丢运行态,不破坏连续对话;只重置执行期临时字段。
1.`CommonState` 新增 `ResetForNextRun()`(统一重置入口):
- 需要重置:`RoundUsed``ConsecutiveCorrections``PlanSteps/CurrentStep``NeedsRoughBuild``NeedsRefineAfterRoughBuild``AllowReorder``SuggestedOrderBaseline``TerminalOutcome`
- 不重置:`ConversationID``UserID`、历史对话、ScheduleState。
2.`Chat` 节点入口做主路径重置:
- 条件:`!HasPendingInteraction()` 且上一轮 `PhaseDone`
- 目的:用户发起新轮请求时自动清执行期脏状态。
3. 在冷加载恢复处做同样重置兜底:
- 位置:`loadOrCreateRuntimeState()`
- 条件同上;
- 目的:覆盖断联恢复场景,避免旧 round 污染新轮。
#### P1工具收敛能力研究与改造
延续 12.x 的结论,优先做:
1. `evaluate_balance`(完成判据工具)
2. `find_first_free` 从单点升级为候选集(`top_k`
3. `query_range` 明确区分 `hard_conflict``embeddable_overlap`
### 13.4 本节涉及的关键文件
- `backend/newAgent/model/common_state.go`
- `backend/newAgent/node/chat.go`
- `backend/service/agentsvc/agent_newagent.go`
- `backend/newAgent/prompt/execute_context.go`
- `backend/newAgent/tools/read_tools.go`
- `backend/newAgent/tools/registry.go`
- `backend/newAgent/tools/SCHEDULE_TOOLS.md`

View File

@@ -1,465 +0,0 @@
# 阶段 3上下文瘦身设计
本文档更新时间2026-04-08
## 0. 文档目的
这份文档只服务第 3 阶段“上下文瘦身”。
职责边界:
1. 记录当前已经和用户对齐的“瘦身后 execute 上下文骨架”。
2. 记录第 3 阶段的落地顺序、非目标和完成标准。
3. 作为后续上下文被裁剪后的继续施工依据。
明确不负责:
1. 不展开第 4 阶段 prompt 三层拆分的最终结构。
2. 不在这里继续讨论粗排和 abort 协议本身。
---
## 1. 当前已经确认的收敛结论
以下结论已经对齐,后续不要再回摆:
1. 第 3 阶段先解决“上下文过胖、重复、噪音多”,不是先做 prompt 三层重构。
2. 工具参数定义和 JSON 调用示例,应该放在工具块里,不应该散落在别的 message。
3. 上下文结构应尽量通用化,后端只填“已有事实”,不要引入大量需要主观概括的分类字段。
4. `msg3``msg4` 必须是一一对应的一组:
- `msg3` 是最近一次工具调用记录;
- `msg4` 是与 `msg3` 对应的工具结果;
- 如果当前没有最近工具调用,则 `msg3` / `msg4` 应一起缺省。
5. `msg4` 已经承载“最近一次工具结果”,`msg5` 不应再重复这部分内容。
6. `msg5` 只保留“有唯一来源的运行态事实”,不能放没有明确 owner 的解释性字段。
7. `当前步骤` 可以保留,唯一来源应是 `CommonState.CurrentPlanStep()`
8. `当前目标` 当前不保留,因为没有稳定 owner容易变成后端替模型总结下一步动作。
9. `最近观察` 当前不保留,因为没有稳定结构化来源;如果需要相关信息,应直接通过 `msg4` 表达最近一次工具结果。
10. `已确认语义` 这类过宽、难填、边界不清的字段不要进入第 3 阶段方案。
---
## 2. 瘦身后的目标上下文骨架
这里记录的是第 3 阶段完成后,`execute` 阶段理想的 messages 骨架。
注意:
1. 这是“上下文瘦身后的骨架”,不是第 4 阶段 prompt 三层重构的最终形态。
2. 重点是减量、去重、压缩,而不是新增更多层次和字段。
### 2.1 目标 message 列表
```text
message[0] role=system
执行规则:
- 只围绕当前步骤行动
- 只输出严格 JSON
- 不要伪造工具结果
- 读操作使用 action=continue + tool_call
- 写操作使用 action=confirm + tool_call
- 缺少关键信息时用 action=ask_user
- 当前流程应终止时用 action=abort
- next_plan / done 时 goal_check 必填
message[1] role=system
可用工具:
- 工具名
- 工具说明
- 参数定义
- 最小 JSON 调用示例
message[2] role=assistant
历史摘要:
- 用户目标
- 更早但仍有效的事实
- 最近失败摘要
- 已折叠说明(重复查询、过程话术、旧修正链已省略)
message[3] role=assistant
最近一次工具调用记录(与 message[4] 成对)
message[4] role=tool
最近一次工具结果
message[5] role=system
当前执行状态:
- 当前轮次
- 当前步骤
- 当前步骤完成判定
message[6] role=user
请继续当前任务的执行阶段,严格输出 JSON。
```
### 2.2 各 message 的职责边界
#### message[0]:执行规则
只保留 execute 的稳定规则,不在这里重复工具参数,也不在这里放运行态数据。
#### message[1]:工具块
必须包含:
1. 工具名
2. 工具说明
3. 参数定义
4. 最小 JSON 调用示例
原因:
1. 这是 LLM 真实调用工具时最直接依赖的材料。
2. 如果只给工具名不给参数,模型很容易继续出现缺参调用。
3. 第 3 阶段里,工具块比“更长的执行 prompt”更重要。
#### message[2]:历史摘要
只保留“更早但仍有效”的摘要,不保留全量流水账。
允许保留的内容:
1. 用户原始目标
2. 当前会话中仍然有效的约束
3. 最近失败摘要
4. 历史折叠说明
不应保留的内容:
1. assistant 的过程话术,例如“我先看一下”“我接下来准备……”
2. 同工具同参数的多份原始重复结果
3. 整段 correction 往返原文
#### message[3] + message[4]:最近一组工具观察
这是 execute 在当前轮之前最关键、最新鲜的一组观察。
约束:
1. `message[3]``message[4]` 必须成对出现。
2. `message[3]` 是调用记录,`message[4]` 是与之对应的工具结果。
3. 若当前没有最近工具调用,则这两条一起省略。
4. 不能凭空生成“最近结果摘要”。
#### message[5]:当前执行状态
这里只允许放“有唯一来源的运行态事实”。
当前允许的字段:
1. 当前轮次
2. 当前步骤
3. 当前步骤完成判定
当前不允许的字段:
1. 当前目标
2. 最近观察
3. 已确认语义
4. 任何需要后端主观解释才能生成的字段
#### message[6]:本轮触发指令
作用很单一:
1. 明确“现在继续 execute”
2. 强化“严格输出 JSON”
不要在这里重复整份计划、工具说明或历史摘要。
---
## 3. 一个更接近真实落地的示例
下面这个例子只用于校准方向,不要求文案逐字一致。
```text
message[0] role=system
你是 SmartFlow NewAgent 的执行器。
执行规则:
1. 只围绕当前步骤行动,不要跳到别的步骤。
2. 只输出严格 JSON不要输出 markdown不要输出 JSON 之外的解释。
3. 不要伪造工具结果。
4. 读操作用 action=continue + tool_call。
5. 写操作用 action=confirm + tool_call。
6. 缺少关键上下文且无法补齐时,输出 action=ask_user。
7. 当前流程应正式终止时,输出 action=abort。
8. 输出 action=next_plan 或 action=done 时goal_check 必填。
message[1] role=system
可用工具:
1. get_overview
说明:查看当前窗口内整体分布。
参数:
{}
调用示例:
{
"name": "get_overview",
"arguments": {}
}
2. find_free
说明:查找满足指定连续时长的空位。
参数:
{
"duration": "int, 必填",
"day": "int, 可选"
}
调用示例:
{
"name": "find_free",
"arguments": {
"duration": 2,
"day": 5
}
}
3. list_tasks
说明:列出任务,可按类别和状态过滤。
参数:
{
"category": "string, 可选",
"status": "string, 可选, 可取 all/existing/suggested/pending"
}
调用示例:
{
"name": "list_tasks",
"arguments": {
"status": "suggested"
}
}
4. move
说明:移动一个已预排任务(仅 suggested
参数:
{
"task_id": "int, 必填",
"new_day": "int, 必填",
"new_slot_start": "int, 必填"
}
调用示例:
{
"name": "move",
"arguments": {
"task_id": 128,
"new_day": 5,
"new_slot_start": 1
}
}
5. swap
说明:交换两个已落位任务。
参数:
{
"task_a": "int, 必填",
"task_b": "int, 必填"
}
调用示例:
{
"name": "swap",
"arguments": {
"task_a": 128,
"task_b": 136
}
}
6. batch_move
说明:批量原子移动多个任务。
参数:
{
"moves": [
{
"task_id": "int, 必填",
"new_day": "int, 必填",
"new_slot_start": "int, 必填"
}
]
}
调用示例:
{
"name": "batch_move",
"arguments": {
"moves": [
{
"task_id": 128,
"new_day": 5,
"new_slot_start": 1
},
{
"task_id": 129,
"new_day": 5,
"new_slot_start": 3
}
]
}
}
7. unplace
说明:取消一个已落位任务。
参数:
{
"task_id": "int, 必填"
}
调用示例:
{
"name": "unplace",
"arguments": {
"task_id": 128
}
}
补充约束:
- suggested 可以 move / swap / unplace
- existing 不能 move / batch_move仅作已安排事实层
- pending 不能 move / swap / unplace
- 如果当前任务已经是 suggested不要再把它当 pending 去 place
message[2] role=assistant
历史摘要:
- 用户目标:把任务类 [101,102] 调整到本周,尽量分布更均匀,周五不要太满。
- 当前已知事实:
1. 当前阶段是 execute
2. task_class_ids=[101,102]
3. 当前状态统计existing=9, suggested=18, pending=0
- 最近一次失败摘要find_free 缺少 duration 参数
- 更早的重复查询、过程话术、旧修正链已折叠
message[3] role=assistant
tool_call:
{
"name": "get_overview",
"arguments": {}
}
message[4] role=tool
规划窗口概览:
- existing=9
- suggested=18
- pending=0
- 周三第5-8节 suggested 偏密
- 周五第1-2节有空位
- 周五第3-4节有空位
message[5] role=system
当前执行状态:
- 当前轮次2/8
- 当前步骤:先识别最值得调整的 suggested 任务和候选空位
- 当前步骤完成判定:能明确指出哪些任务值得调整,以及候选目标时段
message[6] role=user
请继续当前任务的执行阶段,严格输出 JSON。
```
---
## 4. 第 3 阶段的具体落地计划
### 4.1 第一件事:先抓真实输入样本
目标:
1. 先拿到 `BuildExecuteMessages()` 真正送给模型的样本。
2. 不先拍脑袋改结构,先确认“胖点”究竟在 history、pinned还是 runtime prompt。
当前可复用入口:
1. `backend/newAgent/prompt/execute.go`
2. `backend/newAgent/prompt/base.go`
3. `backend/newAgent/node/execute.go` 中已有 execute 上下文调试日志
产出:
1. 至少 2 到 3 份真实样本
2. 标出每个 message 的长度、重复点和噪音来源
### 4.2 第二件事:补 execute 专用的历史压缩层
目标:
1. 不再把 `ConversationContext.History` 原样全量喂给 execute。
2. 形成“更早历史摘要 + 最近一组工具观察”的结构。
必须做的事:
1. 同工具同参数的重复查询,不保留多份原始结果。
2. 更早结果改成摘要,只保留最近一条原始结果。
3. assistant 过程话术不再进入后续模型历史。
4. correction / 工具失败链改成“最近失败摘要”,不要保留整段往返原文。
5. 保留合法的 assistant tool_call + tool result 成对消息,不能破坏 OpenAI 兼容格式。
### 4.3 第三件事:压缩 pinned / runtime 的重复信息
目标:
1. 避免 `state summary + pinned + runtime user prompt` 三处重复抄同一份信息。
2. 保留最新、必要、唯一来源的信息。
原则:
1. 当前计划/当前步骤只保留最新版本,不做历史累积。
2. `msg5` 只保留“当前轮次 + 当前步骤 + 当前步骤完成判定”。
3. 工具结果只出现在 `msg4`,不在 `msg5` 再复述。
4. 粗排语义只保留一处,不要在多条 message 重复提醒。
### 4.4 第四件事:最后再接 token budget
目标:
1. 在完成摘要化和去重后,再做按预算裁剪。
2. 避免一上来直接砍历史,把真正有价值的信息也一起砍掉。
可复用思路:
1. 参考旧链路 `agent.go` 的历史预算计算和裁剪流程。
2. 参考 `backend/pkg/token_budget.go` 中的预算估算与窗口裁剪函数。
要求:
1. 不一定照搬旧链路。
2. 但应复用“先估算、再裁剪、最后收敛会话窗口”的思路。
---
## 5. 第 3 阶段明确不做什么
为了避免和第 4 阶段混淆,这一轮明确不做:
1. 不把 execute prompt 直接拆成三层正式文件结构。
2. 不在通用执行 prompt 里重写完整排程领域模块。
3. 不额外新增没有稳定 owner 的字段,例如:
- 当前目标
- 最近观察
- 已确认语义
4. 不继续围绕粗排补边角语义。
5. 不继续围绕 abort 协议扩展描述文案。
---
## 6. 第 3 阶段完成标准
至少要满足:
1. execute 首轮 messages 明显变短。
2. 同工具同参数的重复查询不会继续堆多份原始结果。
3. assistant 过程话术不再进入后续执行历史。
4. 最近一次失败模式仍能被模型感知。
5. 最近一次工具调用与结果仍以合法配对形式保留。
6. `msg5` 不再重复 `msg4` 的内容。
7. 不破坏第 1-2 阶段已经打通的粗排 / abort 语义。
---
## 7. 供下一轮继续时快速判断的检查清单
如果下一轮接手时要快速判断是否做对,可以先问这几个问题:
1. execute 现在是否还是“全量 history + 全量 pinned + 全量 runtime prompt”直接拼接
2. 工具参数和 JSON 示例是否已经进入单独工具块?
3. `msg3` / `msg4` 是否仍保持一一对应?
4. `msg5` 是否只保留运行态事实,而没有重复工具结果?
5. assistant 的过程话术是否还在继续污染后续历史?
6. correction 失败链是否还在整段保留?
如果以上问题仍然大多回答“是”,说明第 3 阶段还没有真正完成。

View File

@@ -44,6 +44,7 @@ func RegisterChatHistoryPersistHandler(
if repoManager == nil {
return errors.New("repo manager is nil")
}
kafkaCfg := kafkabus.LoadConfig()
// 2. 定义统一处理器:
// 2.1 解析 payload
@@ -62,7 +63,7 @@ func RegisterChatHistoryPersistHandler(
// 2.2.1 基于同一个 tx 构造 RepoManager复用你现有跨包事务模型。
txM := repoManager.WithTx(tx)
// 2.2.2 在同事务内写入聊天历史与会话计数。
return txM.Agent.SaveChatHistoryInTx(
if err := txM.Agent.SaveChatHistoryInTx(
ctx,
payload.UserID,
payload.ConversationID,
@@ -75,6 +76,19 @@ func RegisterChatHistoryPersistHandler(
payload.RetryFromUserMessageID,
payload.RetryFromAssistantMessageID,
payload.TokensConsumed,
); err != nil {
return err
}
// 2.2.3 Day1 追加“记忆抽取请求”事件入队:
// 1) 仅对 user 消息投递,避免把助手回复重复喂给抽取链路;
// 2) 与聊天落库放在同一事务,保证“消息存在 -> 事件一定可追踪”;
// 3) 若入队失败,整体回滚并触发 outbox 重试,不留半成功状态。
return EnqueueMemoryExtractRequestedInTx(
ctx,
outboxRepo.WithTx(tx),
kafkaCfg,
payload,
)
})
}

View File

@@ -0,0 +1,194 @@
package events
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"strconv"
"strings"
"time"
kafkabus "github.com/LoveLosita/smartflow/backend/infra/kafka"
outboxinfra "github.com/LoveLosita/smartflow/backend/infra/outbox"
memorymodel "github.com/LoveLosita/smartflow/backend/memory/model"
memoryrepo "github.com/LoveLosita/smartflow/backend/memory/repo"
memoryservice "github.com/LoveLosita/smartflow/backend/memory/service"
"github.com/LoveLosita/smartflow/backend/model"
"github.com/spf13/viper"
"gorm.io/gorm"
)
const (
// EventTypeMemoryExtractRequested 是“记忆抽取请求”事件类型。
EventTypeMemoryExtractRequested = "memory.extract.requested"
maxMemorySourceTextLength = 1500
)
// RegisterMemoryExtractRequestedHandler 注册“记忆抽取请求”消费者。
//
// 职责边界:
// 1. 只负责把事件转为 memory_jobs 任务;
// 2. 不在消费回调里执行 LLM 重计算;
// 3. 用 outbox 通用事务保证“任务入库 + consumed 推进”原子一致。
func RegisterMemoryExtractRequestedHandler(
bus *outboxinfra.EventBus,
outboxRepo *outboxinfra.Repository,
) error {
if bus == nil {
return errors.New("event bus is nil")
}
if outboxRepo == nil {
return errors.New("outbox repository is nil")
}
handler := func(ctx context.Context, envelope kafkabus.Envelope) error {
var payload model.MemoryExtractRequestedPayload
if unmarshalErr := json.Unmarshal(envelope.Payload, &payload); unmarshalErr != nil {
_ = outboxRepo.MarkDead(ctx, envelope.OutboxID, "解析记忆抽取载荷失败: "+unmarshalErr.Error())
return nil
}
if validateErr := validateMemoryExtractPayload(payload); validateErr != nil {
_ = outboxRepo.MarkDead(ctx, envelope.OutboxID, "记忆抽取载荷非法: "+validateErr.Error())
return nil
}
return outboxRepo.ConsumeAndMarkConsumed(ctx, envelope.OutboxID, func(tx *gorm.DB) error {
enqueueService := memoryservice.NewEnqueueService(memoryrepo.NewJobRepo(tx))
jobPayload := memorymodel.ExtractJobPayload{
UserID: payload.UserID,
ConversationID: strings.TrimSpace(payload.ConversationID),
AssistantID: strings.TrimSpace(payload.AssistantID),
RunID: strings.TrimSpace(payload.RunID),
SourceMessageID: payload.SourceMessageID,
SourceRole: strings.TrimSpace(payload.SourceRole),
SourceText: strings.TrimSpace(payload.SourceText),
OccurredAt: payload.OccurredAt,
TraceID: strings.TrimSpace(payload.TraceID),
IdempotencyKey: strings.TrimSpace(payload.IdempotencyKey),
}
return enqueueService.EnqueueExtractJob(ctx, jobPayload, envelope.EventID)
})
}
return bus.RegisterEventHandler(EventTypeMemoryExtractRequested, handler)
}
// EnqueueMemoryExtractRequestedInTx 在事务内写入 memory.extract.requested outbox 消息。
//
// 设计目的:
// 1. 让“聊天消息已落库”与“记忆抽取事件已入队”同事务提交;
// 2. 任何一步失败都整体回滚,避免出现链路断点。
func EnqueueMemoryExtractRequestedInTx(
ctx context.Context,
outboxRepo *outboxinfra.Repository,
kafkaCfg kafkabus.Config,
chatPayload model.ChatHistoryPersistPayload,
) error {
if !isMemoryWriteEnabled() {
return nil
}
if outboxRepo == nil {
return errors.New("outbox repository is nil")
}
memoryPayload, shouldEnqueue := buildMemoryExtractPayloadFromChat(chatPayload)
if !shouldEnqueue {
return nil
}
payloadJSON, err := json.Marshal(memoryPayload)
if err != nil {
return err
}
outboxPayload := outboxinfra.OutboxEventPayload{
EventType: EventTypeMemoryExtractRequested,
EventVersion: outboxinfra.DefaultEventVersion,
AggregateID: strings.TrimSpace(chatPayload.ConversationID),
Payload: payloadJSON,
}
_, err = outboxRepo.CreateMessage(
ctx,
EventTypeMemoryExtractRequested,
kafkaCfg.Topic,
strings.TrimSpace(chatPayload.ConversationID),
outboxPayload,
kafkaCfg.MaxRetry,
)
return err
}
func buildMemoryExtractPayloadFromChat(chatPayload model.ChatHistoryPersistPayload) (model.MemoryExtractRequestedPayload, bool) {
role := strings.ToLower(strings.TrimSpace(chatPayload.Role))
if role != "user" {
return model.MemoryExtractRequestedPayload{}, false
}
sourceText := strings.TrimSpace(chatPayload.Message)
if sourceText == "" {
return model.MemoryExtractRequestedPayload{}, false
}
truncatedSourceText := truncateByRune(sourceText, maxMemorySourceTextLength)
now := time.Now()
return model.MemoryExtractRequestedPayload{
UserID: chatPayload.UserID,
ConversationID: strings.TrimSpace(chatPayload.ConversationID),
// Day1 先保留 assistant_id/run_id 空值,后续从主链路上下文补齐。
AssistantID: "",
RunID: "",
SourceMessageID: 0,
SourceRole: role,
SourceText: truncatedSourceText,
OccurredAt: now,
TraceID: "",
IdempotencyKey: buildMemoryExtractIdempotencyKey(chatPayload.UserID, chatPayload.ConversationID, truncatedSourceText),
}, true
}
func validateMemoryExtractPayload(payload model.MemoryExtractRequestedPayload) error {
if payload.UserID <= 0 {
return errors.New("user_id is invalid")
}
if strings.TrimSpace(payload.ConversationID) == "" {
return errors.New("conversation_id is empty")
}
if strings.TrimSpace(payload.SourceRole) == "" {
return errors.New("source_role is empty")
}
if strings.TrimSpace(payload.SourceText) == "" {
return errors.New("source_text is empty")
}
if strings.TrimSpace(payload.IdempotencyKey) == "" {
return errors.New("idempotency_key is empty")
}
return nil
}
func buildMemoryExtractIdempotencyKey(userID int, conversationID, sourceText string) string {
raw := fmt.Sprintf("%d|%s|%s", userID, strings.TrimSpace(conversationID), strings.TrimSpace(sourceText))
sum := sha256.Sum256([]byte(raw))
return "memory_extract_" + strconv.Itoa(userID) + "_" + hex.EncodeToString(sum[:8])
}
func truncateByRune(raw string, max int) string {
if max <= 0 {
return ""
}
runes := []rune(raw)
if len(runes) <= max {
return raw
}
return string(runes[:max])
}
func isMemoryWriteEnabled() bool {
if !viper.IsSet("memory.enabled") {
return true
}
return viper.GetBool("memory.enabled")
}

View File

@@ -82,7 +82,88 @@ services:
--replication-factor 1
restart: "no"
etcd:
image: quay.io/coreos/etcd:v3.5.5
container_name: smartflow-etcd
restart: unless-stopped
environment:
ETCD_AUTO_COMPACTION_MODE: revision
ETCD_AUTO_COMPACTION_RETENTION: "1000"
ETCD_QUOTA_BACKEND_BYTES: "4294967296"
ETCD_SNAPSHOT_COUNT: "50000"
command: >
etcd
-advertise-client-urls=http://etcd:2379
-listen-client-urls=http://0.0.0.0:2379
--data-dir=/etcd
volumes:
- etcd_data:/etcd
healthcheck:
test: ["CMD", "etcdctl", "endpoint", "health"]
interval: 10s
timeout: 5s
retries: 20
minio:
image: minio/minio:RELEASE.2023-03-20T20-16-18Z
container_name: smartflow-minio
restart: unless-stopped
environment:
MINIO_ROOT_USER: minioadmin
MINIO_ROOT_PASSWORD: minioadmin
command: minio server /minio_data --console-address ":9001"
ports:
- "9000:9000"
- "9001:9001"
volumes:
- minio_data:/minio_data
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
interval: 10s
timeout: 5s
retries: 20
milvus-standalone:
image: milvusdb/milvus:v2.4.4
container_name: smartflow-milvus
restart: unless-stopped
command: ["milvus", "run", "standalone"]
environment:
ETCD_USE_EMBED: "false"
ETCD_ENDPOINTS: etcd:2379
MINIO_ADDRESS: minio:9000
ports:
- "19530:19530"
- "9091:9091"
volumes:
- milvus_data:/var/lib/milvus
depends_on:
etcd:
condition: service_healthy
minio:
condition: service_healthy
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:9091/healthz"]
interval: 10s
timeout: 5s
retries: 30
attu:
image: zilliz/attu:v2.4.3
container_name: smartflow-attu
restart: unless-stopped
ports:
- "8000:3000"
environment:
MILVUS_URL: smartflow-milvus:19530
depends_on:
milvus-standalone:
condition: service_healthy
volumes:
mysql_data:
redis_data:
kafka_data:
etcd_data:
minio_data:
milvus_data: