Close the EXT-8 remaining blocks: vector sharing, hostapi face, citation consistency #109

Merged
weichen merged 1 commits from feature/ext8-remaining-shared-reads into main 2026-10-08 17:22:46 +00:00
9 changed files with 313 additions and 7 deletions
+15
View File
@@ -403,6 +403,21 @@ func Run(opt Options) error {
hostSrv.Search = func(ctx context.Context, owner, text string) ([]core.RetrievalHit, error) {
return caps.searchDefault(ctx, owner, text)
}
// EXT-8 余项:hostapi 检索面同样吃适配器共享(本人 + 放行的他人项目知识)。
hostSrv.SharedSearch = func(ctx context.Context, owner, text string) ([]core.RetrievalHit, error) {
own, err := caps.searchDefault(ctx, owner, text)
if err != nil {
return nil, err
}
shared, err := st.SearchKnowledgeShared(owner, core.RetrievalQuery{Text: text, TopK: 5}, func(docOwner, projectID core.ID) bool {
ok, aerr := authzSlot.Authorize(owner, projectID)
return aerr == nil && ok
})
if err != nil {
return nil, err
}
return append(own, shared...), nil
}
hostSrv.Complete = func(ctx context.Context, owner, prompt string) (string, error) {
return caps.complete(ctx, owner, prompt)
}
+5
View File
@@ -0,0 +1,5 @@
package hostapi
import "encoding/json"
func jsonUnmarshal(b []byte, v any) error { return json.Unmarshal(b, v) }
+10 -2
View File
@@ -18,7 +18,10 @@ type Server struct {
Store *store.Store
Dispatch func(ctx context.Context, pluginID, owner, commandID string, params map[string]any) error
Search func(ctx context.Context, owner, text string) ([]core.RetrievalHit, error)
Complete func(ctx context.Context, owner, prompt string) (string, error)
// SharedSearch EXT-8 余项:非 nil 时 search 端点改用之(组合根传入
// 「本人结果 + 适配器放行的共享追加」组合;nil=现状仅本人)。
SharedSearch func(ctx context.Context, owner, text string) ([]core.RetrievalHit, error)
Complete func(ctx context.Context, owner, prompt string) (string, error)
}
func (s *Server) Handler() http.Handler {
@@ -143,7 +146,12 @@ func (s *Server) search(w http.ResponseWriter, r *http.Request) {
http.Error(w, "retriever unavailable", http.StatusServiceUnavailable)
return
}
hits, err := s.Search(r.Context(), s.owner(r), r.URL.Query().Get("q"))
// EXT-8 余项:共享版可用时整体切换(本人+共享追加)。
fn := s.Search
if s.SharedSearch != nil {
fn = s.SharedSearch
}
hits, err := fn(r.Context(), s.owner(r), r.URL.Query().Get("q"))
if err != nil {
http.Error(w, err.Error(), http.StatusBadGateway)
return
+49
View File
@@ -0,0 +1,49 @@
package hostapi
import (
"context"
"net/http"
"net/http/httptest"
"testing"
core "agent-system-core"
)
// EXT-8 余项:hostapi 检索面——SharedSearch 可用时整体切换(本人+适配器放行的
// 共享追加);不可用时维持现状(仅本人回调)。
func TestSharedSearchFace(t *testing.T) {
own := []core.RetrievalHit{{Doc: core.KnowledgeDoc{Owner: "bob", Chunk: "本人命中"}}}
sharedHit := core.RetrievalHit{Doc: core.KnowledgeDoc{Owner: "alice", ProjectID: "prj-s", Chunk: "共享命中"}}
srv := &Server{Token: "t", Search: func(ctx context.Context, owner, text string) ([]core.RetrievalHit, error) {
return own, nil
}}
run := func() int {
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/host/v1/knowledge/search?q=x", nil)
req.Header.Set("Authorization", "Bearer "+srv.Token)
srv.Handler().ServeHTTP(rec, req)
return rec.Code
}
if code := run(); code != http.StatusOK {
t.Fatalf("plain search: %d", code)
}
srv.SharedSearch = func(ctx context.Context, owner, text string) ([]core.RetrievalHit, error) {
return append(own, sharedHit), nil
}
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/host/v1/knowledge/search?q=x", nil)
req.Header.Set("Authorization", "Bearer "+srv.Token)
srv.Handler().ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("shared search: %d %s", rec.Code, rec.Body.String())
}
var hits []core.RetrievalHit
if err := jsonUnmarshal(rec.Body.Bytes(), &hits); err != nil {
t.Fatal(err)
}
if len(hits) != 2 || hits[1].Doc.Owner != "alice" {
t.Fatalf("hits = %+v, want 本人+共享", hits)
}
}
+32
View File
@@ -124,6 +124,38 @@ func firstRef(docs []core.KnowledgeDoc) string {
return ""
}
// RetrieveShared EXT-8 余项:本人两路如常 + 追加共享两路(向量 Shared + 全文
// Shared,allow=适配器判定),RRF 融合后裁剪 topK。embedder 不可用时共享仅全文
// (与切片 2 行为一致)。
func (p *Pipeline) RetrieveShared(ctx context.Context, owner core.ID, q core.RetrievalQuery, allow func(docOwner, projectID core.ID) bool) ([]core.RetrievalHit, error) {
own, err := p.Retrieve(ctx, owner, q)
if err != nil {
return nil, err
}
shared, err := p.st.SearchKnowledgeShared(string(owner), q, allow)
if err != nil {
return nil, err
}
// 向量共享路:embedder 可用时并入。
if p.embedder != nil && p.embeddingRoute(ctx).OK() {
if qvecs, e := p.embedder.Embed(ctx, []string{q.Text}); e == nil && len(qvecs) > 0 && len(qvecs[0]) > 0 {
topK := q.TopK
if topK <= 0 {
topK = 5
}
vs, e2 := p.st.SearchKnowledgeVectorsShared(ctx, string(owner), qvecs[0], topK, q, allow)
if e2 == nil {
shared = rrfMerge(vs, shared)
}
}
}
merged := rrfMerge(own, shared)
if limit := q.TopK; limit > 0 && len(merged) > limit {
merged = merged[:limit]
}
return merged, nil
}
// Retrieve 统一检索(core.Retriever)。
func (p *Pipeline) Retrieve(ctx context.Context, owner core.ID, q core.RetrievalQuery) ([]core.RetrievalHit, error) {
topK := q.TopK
+53
View File
@@ -0,0 +1,53 @@
package rag
import (
"context"
"path/filepath"
"testing"
"agent-system-backend/internal/settings"
"agent-system-backend/internal/store"
core "agent-system-core"
)
// EXT-8 余项红/绿:向量通道共享——他人 owner 的向量命中经 allow 放行可检。
// 用 hash 档 embedder(确定性向量),owner=alice 建项目知识并嵌入;bob 经
// RetrieveShared 放行后命中(含 vector Kind),拒绝时零命中。
func TestRetrieveSharedVectorChannel(t *testing.T) {
st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "t.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { st.Close() })
set := settings.NewService(st)
p := NewPipeline(st, set, nil, nil)
p.SetEmbedder(SwitchingEmbedder{Provider: func() string { return "hash" }})
docs := []core.KnowledgeDoc{{Owner: "alice", ProjectID: "prj-s", Source: "plugin:rdm",
Ref: "kb-v", Chunk: "向量共享独占词"}}
// 只走管线写入(分块行+向量一体;不再手工 AddKnowledge 避免主键冲突)。
if err := indexForTest(t, st, p, docs); err != nil {
t.Fatal(err)
}
allow := func(docOwner, projectID core.ID) bool {
return docOwner == "alice" && projectID == "prj-s"
}
hits, err := p.RetrieveShared(context.Background(), "bob", core.RetrievalQuery{Text: "向量共享独占词", TopK: 5}, allow)
if err != nil {
t.Fatal(err)
}
if len(hits) == 0 {
t.Fatal("authorized shared vector hit must be retrievable(向量通道共享)")
}
deny := func(docOwner, projectID core.ID) bool { return false }
if hits2, _ := p.RetrieveShared(context.Background(), "bob", core.RetrievalQuery{Text: "向量共享独占词", TopK: 5}, deny); len(hits2) != 0 {
t.Fatalf("denied shared search = %d hits, want 0", len(hits2))
}
}
// indexForTest:把 docs 直接嵌入并落向量(复用管线的公开写入路径)。
func indexForTest(t *testing.T, st *store.Store, p *Pipeline, docs []core.KnowledgeDoc) error {
t.Helper()
return p.IndexChunks(context.Background(), "alice", docs)
}
+13 -5
View File
@@ -1474,6 +1474,18 @@ func (s *Server) searchKnowledge(w http.ResponseWriter, r *http.Request) {
topK, _ := strconv.Atoi(r.URL.Query().Get("topK"))
query := core.RetrievalQuery{ProjectID: r.URL.Query().Get("project"), Text: q, TopK: topK}
owner := s.owner(r)
// EXT-8 余项:Retriever 支持共享版(向量+全文两路追加)时整体走共享版;
// 否则维持切片 2 行为(本人检索 + 全文共享追加)。
allow := func(docOwner, projectID core.ID) bool {
return s.authorizeSharedProject(owner, projectID)
}
if rs, ok := s.Retriever.(interface {
RetrieveShared(ctx context.Context, owner core.ID, q core.RetrievalQuery, allow func(docOwner, projectID core.ID) bool) ([]core.RetrievalHit, error)
}); ok && s.Retriever != nil {
hits, err := rs.RetrieveShared(r.Context(), core.ID(owner), query, allow)
writeJSONOr(w, hits, err)
return
}
var hits []core.RetrievalHit
var err error
if s.Retriever != nil {
@@ -1485,11 +1497,7 @@ func (s *Server) searchKnowledge(w http.ResponseWriter, r *http.Request) {
writeStoreErr(w, err)
return
}
// EXT-8 切片 2:追加「他人 owner 且其 ProjectID 经适配器放行」的共享命中
//(全文通道;本人命中在前,合计截断到 topK)。
shared, serr := s.st.SearchKnowledgeShared(owner, query, func(docOwner, projectID core.ID) bool {
return s.authorizeSharedProject(owner, projectID)
})
shared, serr := s.st.SearchKnowledgeShared(owner, query, allow)
if serr != nil {
writeStoreErr(w, serr)
return
+81
View File
@@ -5,6 +5,7 @@ import (
"context"
"encoding/json"
"fmt"
"mime/multipart"
"net/http"
"net/http/httptest"
"testing"
@@ -140,3 +141,83 @@ func TestProjectSSEScopeGate(t *testing.T) {
t.Fatalf("alice own project: %d, want 200(现状不受影响)", c)
}
}
// EXT-8 余项(引用一致性):适配器放行后,共享检索命中的 Ref(工件 ID)可经
// /api/artifacts/{id}/content 取回——原件/搜索/引用/下载四口径一致。
func TestSharedCitationRefDownloadable(t *testing.T) {
srv, st := newAuthTestServer(t)
h := srv.Handler()
alice, err := st.CreateAPIKey("alice", "e2e")
if err != nil {
t.Fatal(err)
}
bob, err := st.CreateAPIKey("bob", "e2e")
if err != nil {
t.Fatal(err)
}
req := httptest.NewRequest(http.MethodPost, "/api/projects", bytes.NewReader(mustJSON2(t, map[string]string{"name": "引用链"})))
req.Header.Set("Authorization", "Bearer "+alice)
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req)
var pj map[string]any
readJSON(t, rec.Body.Bytes(), &pj)
pid := fmt.Sprint(pj["id"])
// alice 上传原件(走真实 upload,含索引),并以其 artifactId 为共享知识的 Ref。
up := uploadMultipart(t, h, alice, pid, "引用源.txt", "引用链独占词 原文内容")
// 上传端点已把可解析文本入索引(Retriever=nil 走 AddKnowledge,Source=upload,
// owner=alice、project=pid);bob 未授权时零命中。
if n := searchCount(t, h, bob, "引用链独占词"); n != 0 {
t.Fatalf("bob before grant = %d, want 0", n)
}
srv.SetProjectAuthorizer(func(owner string, projectID core.ID) (bool, error) {
return owner == "bob" && string(projectID) == pid, nil
})
if n := searchCount(t, h, bob, "引用链独占词"); n != 1 {
t.Fatalf("bob after grant = %d, want 1(共享命中,Ref=工件)", n)
}
// 共享命中的 Ref(=上传工件 ID)可下载(引用→原件一致)。
reqC := httptest.NewRequest(http.MethodGet, "/api/artifacts/"+up+"/content", nil)
reqC.Header.Set("Authorization", "Bearer "+bob)
recC := httptest.NewRecorder()
h.ServeHTTP(recC, reqC)
if recC.Code != http.StatusOK || !bytes.Contains(recC.Body.Bytes(), []byte("原文内容")) {
t.Fatalf("shared citation ref download: %d %q", recC.Code, recC.Body.String())
}
}
func uploadMultipart(t *testing.T, h http.Handler, key, pid, name, text string) string {
t.Helper()
var buf bytes.Buffer
mw := multipart.NewWriter(&buf)
fw, _ := mw.CreateFormFile("file", name)
fw.Write([]byte(text))
mw.Close()
req := httptest.NewRequest(http.MethodPost, "/api/projects/"+pid+"/uploads", &buf)
req.Header.Set("Authorization", "Bearer "+key)
req.Header.Set("Content-Type", mw.FormDataContentType())
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("upload: %d %s", rec.Code, rec.Body.String())
}
var up struct {
ArtifactID string `json:"artifactId"`
}
readJSON(t, rec.Body.Bytes(), &up)
return up.ArtifactID
}
func searchCount(t *testing.T, h http.Handler, key, q string) int {
t.Helper()
req := httptest.NewRequest(http.MethodGet, "/api/knowledge/search?q=%E5%BC%95%E7%94%A8%E9%93%BE%E7%8B%AC%E5%8D%A0%E8%AF%8D", nil)
req.Header.Set("Authorization", "Bearer "+key)
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("search: %d %s", rec.Code, rec.Body.String())
}
var hits []any
readJSON(t, rec.Body.Bytes(), &hits)
return len(hits)
}
+55
View File
@@ -4,6 +4,7 @@ import (
"context"
"encoding/binary"
"math"
"sort"
core "agent-system-core"
)
@@ -61,6 +62,60 @@ func (s *Store) DeleteKnowledgeVectors(ctx context.Context, owner, ref string) e
// SearchKnowledgeVectors 余弦扫描:候选 = 租户本人 + 共享(owner=”),dim 匹配,
// 再按 RetrievalQuery 的项目/来源过滤。
// SearchKnowledgeVectorsShared EXT-8 余项:他人 owner 的向量行 + 谓词后置判定
// (allow 由上层注入适配器;无 ProjectID 行拒绝)。结构同自检索(余弦扫描)。
func (s *Store) SearchKnowledgeVectorsShared(ctx context.Context, requester string, vec []float32, topK int, filter core.RetrievalQuery, allow func(docOwner, projectID core.ID) bool) ([]core.RetrievalHit, error) {
if topK <= 0 {
topK = 5
}
rows, err := s.db.QueryContext(ctx, s.q(`
SELECT k.id,k.owner,k.project_id,k.source,k.title,k.ref,k.chunk,k.created_at, v.vec
FROM knowledge_vectors v JOIN knowledge k ON k.id = v.doc_id
WHERE (v.owner<>? AND v.owner<>'') AND v.dim=?`), requester, len(vec))
if err != nil {
return nil, err
}
defer rows.Close()
type cand struct {
hit core.RetrievalHit
sim float64
}
var cands []cand
for rows.Next() {
var d core.KnowledgeDoc
var own, ca string
var blob []byte
if err := rows.Scan(&d.ID, &own, &d.ProjectID, &d.Source, &d.Title, &d.Ref, &d.Chunk, &ca, &blob); err != nil {
return nil, err
}
d.Owner = own
d.CreatedAt, _ = parseTime(ca)
if d.ProjectID == "" || !allow(d.Owner, d.ProjectID) {
continue
}
if filter.ProjectID != "" && d.ProjectID != filter.ProjectID {
continue
}
if len(filter.Sources) > 0 && !containsStr(filter.Sources, d.Source) {
continue
}
sim := cosine(vec, decodeVector(blob))
cands = append(cands, cand{hit: core.RetrievalHit{Doc: d, Score: sim, Kind: "vector"}, sim: sim})
}
if err := rows.Err(); err != nil {
return nil, err
}
sort.SliceStable(cands, func(i, j int) bool { return cands[i].sim > cands[j].sim })
out := []core.RetrievalHit{}
for _, c := range cands {
out = append(out, c.hit)
if len(out) >= topK {
break
}
}
return out, nil
}
func (s *Store) SearchKnowledgeVectors(ctx context.Context, owner string, vec []float32, topK int, filter core.RetrievalQuery) ([]core.RetrievalHit, error) {
if topK <= 0 {
topK = 5