Add metadata filters to search, JSON metadata on upload, paragraph-aw…

README.md

@@ -226,6 +226,7 @@ LocalRecall uses environment variables to configure its behavior. These variable
 | `VECTOR_ENGINE`             | Vector database engine to use (`chromem` by default, `postgres` for PostgreSQL).                              |
 | `MAX_CHUNKING_SIZE`         | Maximum size (in characters) for breaking down documents into chunks. Affects performance and accuracy.       |
 | `CHUNK_OVERLAP`             | Overlap in characters between consecutive chunks (word-aligned). Default: 0. Use to improve context across chunk boundaries. |
+| `CHUNK_BY_PARAGRAPH`        | When not `false` (default), files are chunked per blank-line-separated paragraph so chunk boundaries never cross a paragraph separator (e.g. one passage per line in mythological texts). Set to `false` for the legacy single-block chunking. |
 | `HYBRID_SEARCH_BM25_WEIGHT` | Weight for BM25 keyword search in hybrid search (default: 0.5, PostgreSQL only).                                 |
 | `HYBRID_SEARCH_VECTOR_WEIGHT` | Weight for vector similarity search in hybrid search (default: 0.5, PostgreSQL only).                           |
 | `POSTGRES_LOCK_TIMEOUT`     | Per-connection `lock_timeout` for the PostgreSQL engine (default: `30s`). Bounds how long a statement waits to acquire a lock so a single stuck operation cannot make every other statement on the table queue indefinitely. Set to `0`/`off` to disable. |
@@ -255,13 +256,16 @@ curl -X POST $BASE_URL/collections \
   -d '{"name":"myCollection"}'
 ```
 
-- **Upload File**:
+- **Upload File** (with optional JSON metadata attached to every chunk):
 
 ```sh
 curl -X POST $BASE_URL/collections/myCollection/upload \
-  -F "file=@/path/to/file.txt"
+  -F "file=@/path/to/file.txt" \
+  -F 'metadata={"work":"Pyramid Texts","lang":"fr","form":"transliteration"}'
 ```
 
+The `metadata` form field is a JSON object of string values. Keys `source`, `type`, `file_name` and `memory_*` are reserved and rejected.
+
 - **List Collections**:
 
 ```sh
@@ -290,14 +294,16 @@ curl -X GET $BASE_URL/collections/myCollection/entries/file.pdf/raw
 
 Returns the original uploaded binary file with the appropriate Content-Type header.
 
-- **Search Collection**:
+- **Search Collection** (with optional metadata filters, AND semantics):
 
 ```sh
 curl -X POST $BASE_URL/collections/myCollection/search \
   -H "Content-Type: application/json" \
-  -d '{"query":"search term", "max_results":5}'
+  -d '{"query":"search term", "max_results":5, "filters":{"lang":"fr","form":"gardiner"}}'
 ```
 
+Filters are supported by the `chromem` and `postgres` engines; the `localai` engine returns an error when `filters` is non-empty.
+
 - **Reset Collection**:
 
 ```sh

pkg/chunk/chunk_test.go

@@ -127,3 +127,19 @@ var _ = Describe("Chunk", func() {
 		})
 	})
 })
+
+var _ = Describe("SplitIntoParagraphs", func() {
+	It("splits on blank lines and trims paragraphs", func() {
+		content := "first line\nsecond line\n\n\n  third paragraph  \n"
+		Expect(SplitIntoParagraphs(content)).To(Equal([]string{"first line\nsecond line", "third paragraph"}))
+	})
+
+	It("returns a single paragraph when there are no blank lines", func() {
+		Expect(SplitIntoParagraphs("one\ntwo\nthree")).To(Equal([]string{"one\ntwo\nthree"}))
+	})
+
+	It("returns no paragraphs for empty or whitespace-only content", func() {
+		Expect(SplitIntoParagraphs("")).To(BeEmpty())
+		Expect(SplitIntoParagraphs("  \n \n\t\n")).To(BeEmpty())
+	})
+})

pkg/chunk/chunking.go

@@ -177,6 +177,31 @@ func SplitParagraphIntoChunksWithOptions(paragraph string, opts Options) []strin
 	return chunks
 }
 
+// SplitIntoParagraphs splits content into paragraphs separated by one or more
+// blank lines. Each paragraph is trimmed; empty paragraphs are dropped.
+// A content without blank lines yields a single paragraph.
+func SplitIntoParagraphs(content string) []string {
+	raw := strings.Split(content, "\n")
+	var paragraphs []string
+	var current []string
+	flush := func() {
+		para := strings.TrimSpace(strings.Join(current, "\n"))
+		if para != "" {
+			paragraphs = append(paragraphs, para)
+		}
+		current = nil
+	}
+	for _, line := range raw {
+		if strings.TrimSpace(line) == "" {
+			flush()
+			continue
+		}
+		current = append(current, line)
+	}
+	flush()
+	return paragraphs
+}
+
 // SplitParagraphIntoChunks takes a paragraph and a maxChunkSize as input,
 // and returns a slice of strings where each string is a chunk of the paragraph
 // that is at most maxChunkSize long, ensuring that words are not split.

rag/engine/chromem.go

@@ -265,6 +265,10 @@ func (c *ChromemDB) GetBySource(source string) ([]types.Result, error) {
 }
 
 func (c *ChromemDB) Search(s string, similarEntries int) ([]types.Result, error) {
+	return c.SearchWithFilters(s, similarEntries, nil)
+}
+
+func (c *ChromemDB) SearchWithFilters(s string, similarEntries int, where map[string]string) ([]types.Result, error) {
 	// chromem rejects a query limit larger than the collection, including an
 	// empty collection after complete memory deletion.
 	if n := c.collection.Count(); similarEntries > n {
@@ -273,7 +277,7 @@ func (c *ChromemDB) Search(s string, similarEntries int) ([]types.Result, error)
 		}
 		similarEntries = n
 	}
-	res, err := c.query(context.Background(), s, similarEntries, nil)
+	res, err := c.query(context.Background(), s, similarEntries, where)
 	if err != nil {
 		return nil, err
 	}

rag/engine/localai.go

@@ -92,6 +92,12 @@ func (db *LocalAIRAGDB) GetBySource(source string) ([]types.Result, error) {
 	return nil, fmt.Errorf("not implemented")
 }
 
+// SearchWithFilters is unsupported: the LocalAI store find API has no metadata
+// filter, and filtering client-side would silently change the top-k semantics.
+func (db *LocalAIRAGDB) SearchWithFilters(s string, similarEntries int, where map[string]string) ([]types.Result, error) {
+	return nil, fmt.Errorf("metadata filters are not supported by the localai engine")
+}
+
 func (db *LocalAIRAGDB) Search(s string, similarEntries int) ([]types.Result, error) {
 	resp, err := db.openaiClient.CreateEmbeddings(context.TODO(),
 		openai.EmbeddingRequestStrings{

rag/engine/mock.go

@@ -58,6 +58,43 @@ func (m *MockEngine) StoreDocuments(s []string, metadata map[string]string) ([]R
 	return results, nil
 }
 
+func (m *MockEngine) SearchWithFilters(s string, similarEntries int, where map[string]string) ([]types.Result, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	var results []types.Result
+	for _, doc := range m.docs {
+		match := true
+		for k, v := range where {
+			if doc.Metadata[k] != v {
+				match = false
+				break
+			}
+		}
+		if match && strings.Contains(strings.ToLower(doc.Content), strings.ToLower(s)) {
+			results = append(results, doc)
+		}
+	}
+	if len(results) == 0 {
+		for _, doc := range m.docs {
+			match := true
+			for k, v := range where {
+				if doc.Metadata[k] != v {
+					match = false
+					break
+				}
+			}
+			if match {
+				results = append(results, doc)
+			}
+		}
+	}
+	if len(results) > similarEntries {
+		results = results[:similarEntries]
+	}
+	return results, nil
+}
+
 func (m *MockEngine) Search(s string, similarEntries int) ([]types.Result, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()

rag/engine/postgres.go

@@ -1055,7 +1055,59 @@ func buildScopedHybridSearchQuery(tableName, indexName string) string {
 	`, tableName, candidatePool, rrfK, strings.ReplaceAll(indexName, "'", "''"))
 }
 
+// buildScopedHybridSearchQueryFiltered is buildScopedHybridSearchQuery with an
+// additional metadata containment filter ($6::jsonb) on the final join, so only
+// documents whose metadata contains every filter entry are returned.
+func buildScopedHybridSearchQueryFiltered(tableName, indexName string) string {
+	candidatePool := fmt.Sprintf("GREATEST($5 * %d, %d)", hybridCandidateMultiplier, hybridCandidateFloor)
+	return fmt.Sprintf(`
+		WITH bm25_results AS (
+			SELECT id, ROW_NUMBER() OVER (ORDER BY full_text <@> to_bm25query($1, '%[4]s')) AS rank
+			FROM %[1]s
+			ORDER BY full_text <@> to_bm25query($1, '%[4]s')
+			LIMIT %[2]s
+		),
+		vector_results AS (
+			SELECT id, ROW_NUMBER() OVER (ORDER BY embedding <=> $3::vector) AS rank
+			FROM %[1]s
+			WHERE embedding IS NOT NULL
+			ORDER BY embedding <=> $3::vector
+			LIMIT %[2]s
+		),
+		fused AS (
+			SELECT
+				COALESCE(b.id, v.id) AS id,
+				COALESCE($2::float8 / (%[3]d + b.rank), 0) + COALESCE($4::float8 / (%[3]d + v.rank), 0) AS similarity
+			FROM bm25_results b
+			FULL OUTER JOIN vector_results v ON b.id = v.id
+		)
+		SELECT
+			d.id::text,
+			COALESCE(d.title, '') as title,
+			d.content,
+			d.metadata,
+			f.similarity
+		FROM fused f
+		JOIN %[1]s d ON d.id = f.id AND d.metadata @> $6::jsonb
+		ORDER BY f.similarity DESC
+		LIMIT $5
+	`, tableName, candidatePool, rrfK, strings.ReplaceAll(indexName, "'", "''"))
+}
+
 func (p *PostgresDB) Search(s string, similarEntries int) ([]types.Result, error) {
+	return p.search(s, similarEntries, nil)
+}
+
+// SearchWithFilters restricts the hybrid search to rows whose JSONB metadata
+// contains every entry of where (AND semantics, via the @> containment operator).
+func (p *PostgresDB) SearchWithFilters(s string, similarEntries int, where map[string]string) ([]types.Result, error) {
+	if len(where) == 0 {
+		return p.Search(s, similarEntries)
+	}
+	return p.search(s, similarEntries, where)
+}
+
+func (p *PostgresDB) search(s string, similarEntries int, where map[string]string) ([]types.Result, error) {
 	ctx := context.Background()
 
 	// Get query embedding
@@ -1065,15 +1117,28 @@ func (p *PostgresDB) Search(s string, similarEntries int) ([]types.Result, error
 	}
 	queryEmbeddingStr := formatVector(queryEmbedding)
 
+	var filterJSON []byte
+	if len(where) > 0 {
+		filterJSON, err = json.Marshal(where)
+		if err != nil {
+			return nil, fmt.Errorf("failed to marshal metadata filters: %w", err)
+		}
+	}
+
 	// Build hybrid search query (BM25 + vector similarity)
 	query := buildScopedHybridSearchQuery(p.tableName, p.scope.relation(p.indexBase("bm25")))
+	args := []interface{}{s, p.bm25Weight, queryEmbeddingStr, p.vectorWeight, similarEntries}
+	if filterJSON != nil {
+		query = buildScopedHybridSearchQueryFiltered(p.tableName, p.scope.relation(p.indexBase("bm25")))
+		args = append(args, string(filterJSON))
+	}
 
-	rows, err := p.pool.Query(ctx, query, s, p.bm25Weight, queryEmbeddingStr, p.vectorWeight, similarEntries)
+	rows, err := p.pool.Query(ctx, query, args...)
 	if err != nil {
 		// If BM25 query fails, fallback to vector-only search
 		xlog.Warn("BM25 search failed, falling back to vector search", "error", err)
 		query = fmt.Sprintf(`
-			SELECT 
+			SELECT
 				id::text,
 				COALESCE(title, '') as title,
 				content,
@@ -1084,7 +1149,23 @@ func (p *PostgresDB) Search(s string, similarEntries int) ([]types.Result, error
 			ORDER BY embedding <=> $1::vector
 			LIMIT $2
 		`, p.tableName)
-		rows, err = p.pool.Query(ctx, query, queryEmbeddingStr, similarEntries)
+		fallbackArgs := []interface{}{queryEmbeddingStr, similarEntries}
+		if filterJSON != nil {
+			query = fmt.Sprintf(`
+				SELECT
+					id::text,
+					COALESCE(title, '') as title,
+					content,
+					metadata,
+					(1 - (embedding <=> $1::vector)) as similarity
+				FROM %s
+				WHERE embedding IS NOT NULL AND metadata @> $3::jsonb
+				ORDER BY embedding <=> $1::vector
+				LIMIT $2
+			`, p.tableName)
+			fallbackArgs = append(fallbackArgs, string(filterJSON))
+		}
+		rows, err = p.pool.Query(ctx, query, fallbackArgs...)
 		if err != nil {
 			return nil, fmt.Errorf("failed to execute search: %w", err)
 		}

rag/engine/types.go

@@ -1,5 +1,14 @@
 package engine
 
+import "github.com/mudler/localrecall/rag/types"
+
 type Result struct {
 	ID string `json:"id"`
 }
+
+// FilteredSearcher is an optional extension of the storage engines: when an
+// engine implements it, callers can restrict search results to documents whose
+// metadata matches all entries of the where map (AND semantics).
+type FilteredSearcher interface {
+	SearchWithFilters(s string, similarEntries int, where map[string]string) ([]types.Result, error)
+}

rag/persistency.go

@@ -209,6 +209,23 @@ func (db *PersistentKB) Search(s string, similarEntries int) ([]types.Result, er
 	return db.Engine.Search(s, similarEntries)
 }
 
+// SearchWithFilters behaves like Search but restricts results to chunks whose
+// metadata matches all entries of filters (AND semantics). Engines that do not
+// implement engine.FilteredSearcher return an error when filters are non-empty.
+func (db *PersistentKB) SearchWithFilters(s string, similarEntries int, filters map[string]string) ([]types.Result, error) {
+	db.Lock()
+	defer db.Unlock()
+
+	if len(filters) == 0 {
+		return db.Engine.Search(s, similarEntries)
+	}
+	fs, ok := db.Engine.(engine.FilteredSearcher)
+	if !ok {
+		return nil, fmt.Errorf("metadata filters are not supported by engine %T", db.Engine)
+	}
+	return fs.SearchWithFilters(s, similarEntries, filters)
+}
+
 func (db *PersistentKB) Reset() error {
 	db.Lock()
 	defer db.Unlock()
@@ -767,11 +784,27 @@ func chunkFile(fpath string, maxchunksize, chunkOverlap int) ([]string, error) {
 	}
 
 	opts := chunk.Options{MaxSize: maxchunksize, Overlap: chunkOverlap, SplitLongWords: true}
-	chunks := chunk.SplitParagraphIntoChunksWithOptions(content, opts)
+	var chunks []string
+	if paragraphChunkingEnabled() {
+		// Chunk each paragraph separately so chunk boundaries never cross a
+		// blank-line separator (e.g. one line per passage in mythological texts).
+		for _, para := range chunk.SplitIntoParagraphs(content) {
+			chunks = append(chunks, chunk.SplitParagraphIntoChunksWithOptions(para, opts)...)
+		}
+	} else {
+		chunks = chunk.SplitParagraphIntoChunksWithOptions(content, opts)
+	}
 	xlog.Info("Chunked file", "file", fpath, "content_length", len(content), "max_chunk_size", maxchunksize, "chunk_overlap", chunkOverlap, "chunk_count", len(chunks))
 	return chunks, nil
 }
 
+// paragraphChunkingEnabled reports whether files are chunked per paragraph
+// (blank-line separated) instead of as a single block. Enabled by default;
+// set CHUNK_BY_PARAGRAPH=false to restore the legacy single-block behavior.
+func paragraphChunkingEnabled() bool {
+	return os.Getenv("CHUNK_BY_PARAGRAPH") != "false"
+}
+
 // migrateToUUIDLayout migrates flat files in assetDir (files not in UUID
 // subdirectories) to UUID subdirectory layout. This is a one-time migration.
 func (db *PersistentKB) migrateToUUIDLayout() error {