Close the EXT-8 remaining blocks: vector sharing, hostapi face, citation consistency
ci / vet-and-test (pull_request) Successful in 37s
ci / vet-and-test (pull_request) Successful in 37s
Why: EXT-8 closure needs the vector channel shared (not just fulltext), the /host/v1 knowledge face honoring the authorizer, and citation-ref-to-artifact consistency pinned as an acceptance test. Evidence: TestRetrieveSharedVectorChannel (hash embedder: authorized shared vector hit retrievable, denied yields zero), TestSharedSearchFace (hostapi switches to SharedSearch when wired), TestSharedCitationRefDownloadable (shared hit -> Ref -> artifact content download for the authorized member); go test ./... 32/32 packages ok. Impact: store.SearchKnowledgeVectorsShared; rag.Pipeline.RetrieveShared (own + shared fulltext + shared vector, RRF-merged); server searchKnowledge prefers RetrieveShared when the retriever implements it; hostapi Server.SharedSearch wired in the composition root through the same authorizer slot. Default behavior unchanged. Refs: EXT-8 proposal acceptance; slices 1-2.
This commit is contained in:
@@ -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)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,5 @@
|
||||
package hostapi
|
||||
|
||||
import "encoding/json"
|
||||
|
||||
func jsonUnmarshal(b []byte, v any) error { return json.Unmarshal(b, v) }
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user