给 Agent 接上可回查的本地资料检索

能力检查

python3 -c 'import sqlite3; c=sqlite3.connect(":memory:"); c.execute("CREATE VIRTUAL TABLE t USING fts5(x)"); print(sqlite3.sqlite_version)'

创建、同步、查询、测试

python3 make_corpus.py
python3 retrieval.py sync
python3 retrieval.py query --query '导出 保留'
python3 -m unittest -v test_retrieval

本次实测命中行摘录；完整元数据见 query 输出

handbook.txt:2  [导出] 文件 [保留] 30 天 ， 到期 后 清理 。
meeting.txt:2   [导出] 文件 [保留] 14 天 ， 待 负责人 确认 。

回查命中行

from retrieval import connect, retrieve, verify_source
db = connect('sample-index.sqlite')
try:
    for hit in retrieve(db, '导出 保留', as_of='2026-10-07'):
        print(hit['path'], hit['line'], verify_source('corpus', hit))
finally:
    db.close()

make_corpus.py

"""Create ONLY a new synthetic corpus directory; never overwrite one."""
import json
from pathlib import Path

DOCUMENTS = [
    ('handbook-v1', 'r1', 'archived', None,
     '旧版 导出 说明\n导出 文件 保留 7 天 。\n'),
    ('handbook', 'r2', 'current', None,
     '星屿 项目 导出 手册\n导出 文件 保留 30 天 ， 到期 后 清理 。\n下载 入口 位于 设置 页面 。\n'),
    ('meeting', 'r1', 'current', None,
     '星屿 会议 纪要\n导出 文件 保留 14 天 ， 待 负责人 确认 。\n'),
    ('notice', 'r1', 'current', '2026-10-07',
     '临时 导出 通知\n导出 文件 保留 60 天 ， 仅限 本次 演练 。\n'),
    ('garden', 'r1', 'current', None,
     '窗台 养护 笔记\n薄荷 每周 检查 土壤 湿度 。\n'),
    ('compact', 'r1', 'current', None,
     '导出文件保留三十天\n'),
]


def make(folder):
    root = Path(folder)
    root.mkdir(parents=True, exist_ok=False)
    manifest = []
    for doc, revision, status, expires, text in DOCUMENTS:
        filename = doc + '.txt'
        (root / filename).write_text(text, encoding='utf-8')
        manifest.append(dict(doc_id=doc, revision=revision, status=status,
                             expires=expires, path=filename))
    (root / 'manifest.json').write_text(
        json.dumps(manifest, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')


if __name__ == '__main__':
    make('corpus')
    print('created corpus: 6 synthetic files')

retrieval.py

"""Offline, single-owner teaching index. Corpus is trusted local sample data."""
import argparse
import hashlib
import json
from pathlib import Path
import re
import sqlite3
from datetime import date

SCHEMA = '''
CREATE TABLE IF NOT EXISTS chunks (
 id INTEGER PRIMARY KEY, source_id TEXT UNIQUE NOT NULL,
 doc_id TEXT NOT NULL, revision TEXT NOT NULL, path TEXT NOT NULL,
 sha256 TEXT NOT NULL, line INTEGER NOT NULL, text TEXT NOT NULL,
 status TEXT NOT NULL, expires TEXT
);
CREATE VIRTUAL TABLE IF NOT EXISTS search USING fts5(text, tokenize='unicode61');
'''


def iso(value):
    if not isinstance(value, str) or date.fromisoformat(value).isoformat() != value:
        raise ValueError('use YYYY-MM-DD')
    return value


def connect(path):
    db = sqlite3.connect(path)
    db.row_factory = sqlite3.Row
    try:
        db.executescript(SCHEMA)
    except sqlite3.OperationalError:
        db.close()
        raise
    return db


def sync(db, folder):
    """Validate the whole manifest first; replace both tables in one transaction."""
    root = Path(folder).resolve()
    manifest = json.loads((root / 'manifest.json').read_text(encoding='utf-8'))
    if not isinstance(manifest, list):
        raise ValueError('manifest must be a list')
    prepared, seen = [], set()
    for item in manifest:
        doc, rev = item['doc_id'], item['revision']
        if not all(isinstance(x, str) and re.fullmatch(r'[a-z0-9-]{1,40}', x)
                   for x in (doc, rev)) or doc in seen:
            raise ValueError('invalid or duplicate document ID')
        seen.add(doc)
        rel = item['path']
        if not isinstance(rel, str) or Path(rel).is_absolute():
            raise ValueError('relative path required')
        path = (root / rel).resolve()
        if not path.is_relative_to(root) or path.suffix != '.txt':
            raise ValueError('only corpus .txt files allowed')
        raw = path.read_bytes()
        if len(raw) > 100000:
            raise ValueError('sample file too large')
        body = raw.decode('utf-8')
        digest = hashlib.sha256(raw).hexdigest()
        status = item['status']
        if status not in ('current', 'archived'):
            raise ValueError('unknown status')
        expires = item.get('expires')
        if expires is not None:
            iso(expires)
        for line, text in enumerate(body.splitlines(), 1):
            if text.strip():
                source = f'{doc}@{rev}:{digest}#L{line}'
                prepared.append((source, doc, rev, rel, digest, line,
                                 text, status, expires))
    with db:
        db.execute('DELETE FROM search')
        db.execute('DELETE FROM chunks')
        for values in prepared:
            cur = db.execute('INSERT INTO chunks(source_id,doc_id,revision,path,'
                             'sha256,line,text,status,expires) VALUES(?,?,?,?,?,?,?,?,?)', values)
            db.execute('INSERT INTO search(rowid,text) VALUES(?,?)',
                       (cur.lastrowid, values[6]))
    return len(prepared)


def expression(query):
    # Only the application adds AND. Quotation/operator input stays phrase data.
    if not isinstance(query, str) or len(query) > 200 or '\x00' in query:
        raise ValueError('query must be at most 200 characters, without NUL')
    parts = query.split()
    if len(parts) > 12:
        raise ValueError('at most 12 terms')
    if not parts or not any(c.isalnum() for c in query):
        return None
    return ' AND '.join('"' + p.replace('"', '""') + '"' for p in parts)


def retrieve(db, query, *, as_of, limit=10):
    iso(as_of)
    if type(limit) is not int or not 1 <= limit <= 20:
        raise ValueError('limit must be 1..20')
    expr = expression(query)
    if expr is None:
        return []
    sql = '''SELECT c.*, snippet(search,0,'[',']',' … ',18) AS snippet,
                    bm25(search) AS lexical_score
             FROM search JOIN chunks c ON c.id=search.rowid
             WHERE search MATCH ? AND c.status='current'
               AND (c.expires IS NULL OR c.expires > ?)
             ORDER BY lexical_score,c.source_id LIMIT ?'''
    return [dict(r) for r in db.execute(sql, (expr, as_of, limit))]


def verify_source(folder, hit):
    root = Path(folder).resolve()
    path = (root / hit['path']).resolve()
    if not path.is_relative_to(root):
        return False
    try:
        raw = path.read_bytes()
        lines = raw.decode('utf-8').splitlines()
        return (hashlib.sha256(raw).hexdigest() == hit['sha256']
                and 1 <= hit['line'] <= len(lines)
                and lines[hit['line'] - 1] == hit['text'])
    except (OSError, UnicodeError):
        return False


def search_tool(db, query, *, as_of):
    # as_of is a host policy value, not a date chosen by the model.
    return {'kind': 'document_evidence', 'query': query,
            'as_of': as_of, 'hits': retrieve(db, query, as_of=as_of),
            'action_authority': False}


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('action', choices=['sync', 'query'])
    parser.add_argument('--db', default='sample-index.sqlite')
    parser.add_argument('--corpus', default='corpus')
    parser.add_argument('--as-of', default='2026-10-07')
    parser.add_argument('--query', default='导出 保留')
    args = parser.parse_args()
    db = connect(args.db)
    try:
        if args.action == 'sync':
            print(json.dumps({'indexed_lines': sync(db, args.corpus)}))
        else:
            print(json.dumps(search_tool(db, args.query, as_of=args.as_of),
                             ensure_ascii=False, indent=2))
    finally:
        db.close()


if __name__ == '__main__':
    main()

test_retrieval.py

import json
from pathlib import Path
import sqlite3
import tempfile
import unittest
from make_corpus import make
from retrieval import connect, sync, retrieve, verify_source, expression, search_tool

TODAY = '2026-10-07'


class RetrievalTests(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.root = Path(self.tmp.name) / 'corpus'
        make(self.root)
        self.db = connect(Path(self.tmp.name) / 'index.sqlite')
        sync(self.db, self.root)

    def tearDown(self):
        self.db.close()
        self.tmp.cleanup()

    def search(self, q, **kwargs):
        return retrieve(self.db, q, as_of=kwargs.pop('as_of', TODAY), **kwargs)

    def test_conflicting_documents_are_both_returned(self):
        hits = self.search('导出 保留')
        self.assertEqual({r['doc_id'] for r in hits}, {'handbook', 'meeting'})
        self.assertEqual({r['line'] for r in hits}, {2})
        self.assertTrue(all(verify_source(self.root, r) for r in hits))
        self.assertTrue(all('[导出]' in r['snippet'] for r in hits))

    def test_archived_and_expired_excluded(self):
        self.assertEqual(self.search('7 天'), [])
        self.assertEqual(self.search('60 天'), [])
        self.assertEqual(len(self.search('60 天', as_of='2026-10-06')), 1)

    def test_empty_unrelated_and_synonym(self):
        for q in ('', ' ', '"', '***', '发票', '输出 保存'):
            with self.subTest(q=q):
                self.assertEqual(self.search(q), [])

    def test_unicode61_is_not_chinese_word_segmentation(self):
        hits = self.search('导出文件保留三十天')
        self.assertEqual([r['doc_id'] for r in hits], ['compact'])
        self.assertNotIn('compact', {r['doc_id'] for r in self.search('导出')})

    def test_quotes_and_operators_are_not_fts_programs(self):
        self.assertEqual(len(self.search('"导出" 保留')), 2)
        for q in ('导出 OR 薄荷', "x'; DROP TABLE chunks; --", 'NEAR(',
                  '导出 : (', '" OR "', '\"导出'):
            with self.subTest(q=q):
                self.search(q)  # must not raise SQL or MATCH syntax error
        self.assertEqual(self.db.execute('SELECT count(*) FROM chunks').fetchone()[0], 12)
        self.assertEqual(self.search('导出 OR 薄荷'), [])

    def test_direct_bound_match_still_has_syntax_errors(self):
        with self.assertRaises(sqlite3.OperationalError):
            self.db.execute('SELECT rowid FROM search WHERE search MATCH ?', ('"',)).fetchall()

    def test_reindex_is_stable_and_rename_keeps_id(self):
        before = {r['source_id'] for r in self.search('导出 保留')}
        sync(self.db, self.root)
        self.assertEqual(before, {r['source_id'] for r in self.search('导出 保留')})
        mf = self.root / 'manifest.json'
        items = json.loads(mf.read_text())
        (self.root / 'handbook.txt').rename(self.root / 'renamed.txt')
        items[1]['path'] = 'renamed.txt'
        mf.write_text(json.dumps(items), encoding='utf-8')
        sync(self.db, self.root)
        self.assertEqual(before, {r['source_id'] for r in self.search('导出 保留')})

    def test_update_replaces_old_terms_and_detects_stale_hit(self):
        old = self.search('30 天')[0]
        path = self.root / 'handbook.txt'
        path.write_text(path.read_text().replace('30', '45'), encoding='utf-8')
        self.assertFalse(verify_source(self.root, old))
        # An unsynchronized index still returns its old snapshot.
        self.assertEqual(len(self.search('30 天')), 1)
        sync(self.db, self.root)
        self.assertEqual(self.search('30 天'), [])
        new = self.search('45 天')[0]
        self.assertNotEqual(old['source_id'], new['source_id'])
        self.assertTrue(verify_source(self.root, new))

    def test_delete_requires_manifest_and_index_sync(self):
        mf = self.root / 'manifest.json'
        items = json.loads(mf.read_text())
        (self.root / 'meeting.txt').unlink()
        with self.assertRaises(FileNotFoundError):
            sync(self.db, self.root)
        self.assertEqual(len(self.search('14 天')), 1)  # old index intact
        mf.write_text(json.dumps([x for x in items if x['doc_id'] != 'meeting']))
        sync(self.db, self.root)
        self.assertEqual(self.search('14 天'), [])

    def test_invalid_manifest_cannot_wipe_index(self):
        mf = self.root / 'manifest.json'
        items = json.loads(mf.read_text())
        items[0]['path'] = '../outside.txt'
        mf.write_text(json.dumps(items))
        with self.assertRaises(ValueError):
            sync(self.db, self.root)
        self.assertEqual(len(self.search('导出 保留')), 2)

    def test_invalid_query_contract(self):
        for q in ('x' * 201, 'a\x00b', ' '.join(['a'] * 13), None):
            with self.subTest(q=q), self.assertRaises(ValueError):
                expression(q)
        with self.assertRaises(ValueError):
            self.search('导出', limit=True)
        with self.assertRaises(ValueError):
            self.search('导出', as_of='2026-1-1')

    def test_tool_returns_no_answer_on_no_hit(self):
        result = search_tool(self.db, '发票', as_of=TODAY)
        self.assertEqual(result['hits'], [])
        self.assertFalse(result['action_authority'])
        self.assertNotIn('answer', result)

    def test_snapshot_persists_on_reopen(self):
        path = Path(self.tmp.name) / 'index.sqlite'
        another = connect(path)
        try:
            self.assertEqual(len(retrieve(another, '导出 保留', as_of=TODAY)), 2)
        finally:
            another.close()


if __name__ == '__main__':
    unittest.main()