Recall named documents alongside semantic candidates

docs/README.md

@@ -4,6 +4,7 @@
 
 | 任务 | 阅读文件 |
 | --- | --- |
+| 指名文件标题召回、向量候选合并与12条预算 | [标题召回](title-recall.md) |
 | DMS五栏目ID、名称及字段映射调整 | [DMS配置](dms-schema.md) |
 | 内存接纳拒绝、前后台共享名额、Linux硬内存隔离 | [内存保护](memory-safety.md) |
 | 最终会话回读慢、Redis取回量与选中解码量、反序列化拆分 | [Redis读取诊断](redis-read-timing.md) |

docs/policy-discovery.md

@@ -2,6 +2,8 @@
 
 ## 目标与当前边界
 
+2026-10-09 F128：新增通用标题字面召回，与向量并行，全局最多3条标题命中优先进入原12份候选预算；完整原文校验后在rerank保护。详见[标题召回](title-recall.md)。以下每库top3仍指向量通道；本次未重建向量、扩大普通分析上限或验收总体政策发现质量。
+
 检索实际覆盖6类库（policy、finance、qa、article、service_guide、service_catalog；catalog不检索），每查询每库最多3条，去重合并后最多12条，非每库必须3条。检索层仍优先每库最高分且无相似度阈值。F093在全文分析前新增可开关的逐资料四级并发rerank，依据标题/摘要及上下文过滤弱/不相关、保留强/中，必要前置、缺摘要和失败项保留；不再让全部弱关联命中默认进入全文分析，但保护项和保守分类仍可能保留。详见[rerank](rerank.md)，单一向量分数不能代表相关性或资格，总体召回与分类质量仍待评测。
 
 2026-09-21 F090：规划用`prerequisite_queries`单列前置事项，验证需求下标和原发言依据并去重，顺序为直接需求、前置、相邻方向；旧结果缺字段时兼容空数组。新开主体分别查登记指南和网上入口QA，不将行业词附到通用查询，不锁死未知企业形式；全部只是待原文核验的假设。候选仍在12条上限内先保留每库最高分，再按查询順序保留各查询最高分命中，去重后按全局分数补足，名额不足不保证全覆盖。只读完整原问题重放已开展，曾出现有召回但最终未采用入口QA，须区分召回、采用和正确展示，不能仅凭检索命中验收。

docs/rerank.md

@@ -1,5 +1,7 @@
 # 全文分析前的四级相关性判断
 
+F128补充：标题通道最多3份与向量候选合并后仍最多12份；通过完整原文及标题校验的title_match候选受到保护，低相关等级也保留进入原文分析。等级保持模型原值，保护不等于政策有效或资格符合；见[标题召回](title-recall.md)。
+
 F098补充：无附件且已召回可用同题QA时，在本阶段之前直接走原答案路径，rerank不调用；详见[同题QA提前返回](step3-agent.md)。其他请求保持下述规则。
 
 F093：已召回的最多12份完整资料在全文分块分析前，逐份由现有模型结合当前问题及上下文分类。提示154字，单份返回 `{"level":"强相关"}`，取值限强相关、中相关、弱相关、不相关。无解释、分数或重写摘要，输出预算32 tokens。

docs/title-recall.md

@@ -0,0 +1,15 @@
+# 标题与向量联合召回
+
+F128：保留每查询每库最多3条向量候选；新增全库最多3条标题候选。两通道并行，按库和记录ID去重，明确标题命中先占名额，再按原跨库/查询覆盖和相似度补足，普通分析候选仍最多12份。原多情形指南有界补查（最多额外8份）保持。
+
+标题目录从当前manifest固定的六库索引加载，进程内按库缓存ID、索引首行标题和记录哈希。首次标题查询需读取目录，后续复用；重启后重新读取，不修改Milvus结构或要求重入库。混合快照按每库清单读取。失败的目录不缓存，下一次可重试；已成功的向量结果保留，完整性状态为false，不把故障说成无依据。
+
+书名号中的名称直接匹配；无书名号时仅从较明确的长主题中去掉常见问句后缀（如“有哪些措施”）再匹配。统一全半角、空白和标点，不删除地区、年份或版本词，不用模型扩展词编造文件名。多个名称先各保留一项再补足，全局上限3条；超出时诊断标记overflow，不承诺覆盖所有同名版本。此通道是保守的字面包含匹配，不保证错字、同义改写或短泛词命中；现有跨行标题仅首行进入目录，完整标题仍在原文中核验。
+
+选中标题候选后读取真实向量计算余弦，score保持向量相似度，不伪造为1或获批概率。随后复用完整分段原文回读与哈希校验，再核对原文标题，成功才附加title_match.verified。rerank会保留这些候选进入分析，同时保留模型实际相关性等级；后续仍须按地区、版本、原文内容判断是否采用，不强迫生成答案或认可资格。
+
+指名文件的分析和综合提示要求依据所问文件原文，不用相近新闻或其他法规替代。普通主题问答仍允许多来源补充。最终回答采用多少来源由核验结果决定，12份是候选上限，不是引用配额。
+
+诊断：retrieval.title_recall含status、limit、matched_count、selected_count、overflow和failed_databases；候选及candidate_diagnostics的title_match记录匹配信息。缺少标题查询时status为skipped。目录/向量/原文失败按不同阶段保留诊断。
+
+验证：2026-10-09，328项问答、96项API回归通过；真实Milvus两句原话和另一政策标题进入候选，不存在文件无标题命中。两句原话经真实意图识别、检索、rerank及DeepSeek分析/总结，均采用目标文件且无处理错误；指名总结只采用目标文件。合成覆盖多标题/相似名称/地区和版本差异、去重、12条预算、失败及保护。未重启线上服务，未验收所有标题、长期稳定性或生产容量。

src/step3_question_answer/knowledge_answers.py

@@ -72,10 +72,13 @@ def input_size(prompt, payload):
 
 
 def synthesis_prompts(payload):
+    answer = ANSWER_PROMPT
+    if '《' in payload.get('question', ''):
+        answer += '\n总结指定文件须以该文件原文为主体，不用相近新闻或其他法规替代。核对地区和版本，差异分别说明；缺少目标原文证据则明确说明。'
     if any(entry.get('images') for entry in payload['evidence']):
-        return (ANSWER_PROMPT + IMAGE_RULE,
+        return (answer + IMAGE_RULE,
                 REDUCE_PROMPT + '\n保留相关images的marker、图片说明及source_id，不丢成只有“扫码/见图”的描述。')
-    return ANSWER_PROMPT, REDUCE_PROMPT
+    return answer, REDUCE_PROMPT
 
 
 def budgeted_summary(model, payload, diagnostics, reduction_model=None):

src/step3_question_answer/recommendation.py

@@ -466,6 +466,8 @@ class Recommender:
                    'context_query': handoff.get('context_query', ''),
                    'subject': {} if overview_mode else profile_facts(handoff),
                    'subject_facts': [] if overview_mode else subject_facts(handoff)}
+        if candidate.get('title_match', {}).get('verified') is True:
+            context['title_match'] = candidate['title_match']
         context['subject_facts'] = catalog(context['subject_facts'], record['id'], 'subject')
         cancelled = Event()
 
@@ -496,6 +498,7 @@ class Recommender:
                     instruction += '\nJSON顺序：relevant、relevance_level、policy_category、reason、answer、citations、conditions、match_basis、card。match_basis不嵌入card；闭合完整对象。'
                     if overview_mode:
                         instruction += '\n本轮是知识咨询，不是用户资格匹配。直接依据原文回答所问事项；不要强行推荐政策或要求用户描述营商困难。不能把服务提供方当作助手自身。不要推测用户个人或企业情况，不向用户追问资格，不输出针对用户的申请建议或资格结论；conditions和match_basis返回空数组。名称、内容与引用必须来自本批原文。'
+                    instruction += '\n若用户指名文件，先核对原文标题、发文地区和版本；省略发文机关或试行后缀不当然代表不同文件。title_match仅表示程序核对了标题文字，不证明现行有效或用户资格。总结指定文件须依据该文件原文；仅主题相近的新闻或其他法规不能替代所问文件。'
                     if attempt:
                         instruction += '\n上次校验码：' + rejection + '。按validation_feedback逐项修正引用；仍须完整JSON、正确字段枚举和原文逐字证据。'
                     if rejection in ('model_invalid_content_json', 'model_invalid_content_schema'):
@@ -836,6 +839,7 @@ class Recommender:
             candidate_diagnostics.append({
                 'source_id': source_id, 'title': record['title'],
                 'knowledge_type': candidate['type'], 'snapshot': candidate['snapshot'],
+                'title_match': candidate.get('title_match'),
                 'policy_card_allowed': policy_card_allowed(record),
                 'analysis_complete': result['complete'], 'error': result['error'],
                 'chunk_count': result.get('chunk_count'), 'failed_chunks': result.get('failed_chunks', []),

src/step3_question_answer/rerank.py

@@ -79,6 +79,7 @@ class Reranker:
             if not isinstance(summary, str):
                 summary = ''
             protected = (not summary.strip() or bool(protected_queries.intersection(candidate.get('best_query_indices', [])))
+                         or candidate.get('title_match', {}).get('verified') is True
                          or is_qa and question_key(summary) == question_key(handoff['original_question']))
             payload = {**context, 'title': record.get('title', ''), 'summary': summary}
             if candidate['type'] == 'service_guide':

src/step3_question_answer/retrieval.py

@@ -8,6 +8,7 @@ import os
 from concurrent.futures import ThreadPoolExecutor
 from threading import RLock
 from .timing import slot, measured, mapped
+from .title_recall import TITLE_LIMIT, title_queries, title_match, select_titles
 
 from src.step2_knowledge_ingest.ingest import (
     CONTRACT, DIM, TYPES, Embedder, canonical, config, model_matches_manifest, restore, sha,
@@ -46,7 +47,7 @@ def related_guide_queries(question, records):
     return [key for key in groups if key not in present][:MAX_RELATED_GUIDES]
 
 
-def choose_candidates(hits_by_type, query_hits, limit=MAX_CANDIDATES):
+def choose_candidates(hits_by_type, query_hits, limit=MAX_CANDIDATES, title_hits=()):
     """Keep bank coverage, then each query's best hit before score-only filling."""
     chosen, seen = [], set()
     def add(kind, hit):
@@ -54,6 +55,8 @@ def choose_candidates(hits_by_type, query_hits, limit=MAX_CANDIDATES):
         if len(chosen) < limit and key not in seen:
             chosen.append((kind, hit))
             seen.add(key)
+    for kind, hit in title_hits:
+        add(kind, hit)
     for kind, hits in hits_by_type.items():
         if hits:
             add(kind, hits[0])
@@ -101,6 +104,32 @@ class ReadStore:
     def get(self, name, ids):
         return self.client.get(name, ids=ids, output_fields=['*'], consistency_level='Strong', timeout=self.timeout)
 
+    def titles(self, name):
+        # Existing snapshots index title + newline + summary (QA indexes its question).
+        # Cache only title/hash/identity, never all summaries or document bodies.
+        iterator = self.client.query_iterator(collection_name=name, batch_size=256,
+            output_fields=['id', 'index_text', 'record_hash'],
+            consistency_level='Strong', timeout=self.timeout)
+        rows, seen = [], set()
+        try:
+            while batch := iterator.next():
+                for row in batch:
+                    identity = row['id']
+                    if (not isinstance(identity, str) or not re.fullmatch('k_[a-f0-9]{24}', identity)
+                            or identity in seen or not isinstance(row['index_text'], str)
+                            or not re.fullmatch('[a-f0-9]{64}', row['record_hash'])):
+                        raise RetrievalError('invalid_title_index')
+                    seen.add(identity)
+                    rows.append({'id': identity, 'title': row['index_text'].split('\n', 1)[0],
+                                 'record_hash': row['record_hash']})
+        finally:
+            iterator.close()
+        return rows
+
+    def index_vectors(self, name, ids):
+        return self.client.get(name, ids=ids, output_fields=['vector', 'record_hash'],
+                               consistency_level='Strong', timeout=self.timeout)
+
     def close(self):
         self.client.close()
 
@@ -132,6 +161,8 @@ class Retriever:
         self._stores = {}
         self._warmed = False
         self.lock = RLock()
+        self._title_locks = {kind: RLock() for kind in SEARCH_TYPES}
+        self._title_cache = {}
 
     def _initialize_model(self):
         if self.embedder is None:
@@ -169,6 +200,62 @@ class Retriever:
             for store in self._stores.values():
                 store.close()
             self._stores.clear()
+            self._title_cache.clear()
+
+    def _titles_for(self, kind):
+        with self._title_locks[kind]:
+            if kind not in self._title_cache:
+                rows = self._store_for(kind).titles(self.manifest['databases'][kind]['index_collection'])
+                self._title_cache[kind] = tuple({**row, 'type': kind} for row in rows)
+            return self._title_cache[kind]
+
+    def _recall_titles(self, queries, vectors):
+        if not queries:
+            return [], {'status': 'skipped', 'limit': TITLE_LIMIT, 'matched_count': 0, 'failed_databases': []}
+        failures, entries = [], []
+        def read(kind):
+            try:
+                return kind, self._titles_for(kind), False
+            except Exception:
+                return kind, (), True
+        with ThreadPoolExecutor(max_workers=self.bank_workers) as pool:
+            for kind, rows, failed in mapped(pool, read, SEARCH_TYPES):
+                entries.extend(rows)
+                if failed:
+                    failures.append(kind)
+        selected, count = select_titles(entries, queries)
+        hits = []
+        for kind in SEARCH_TYPES:
+            group = [row for row in selected if row['type'] == kind]
+            if not group:
+                continue
+            try:
+                values = self._store_for(kind).index_vectors(
+                    self.manifest['databases'][kind]['index_collection'], [row['id'] for row in group])
+                by_id = {row['id']: row for row in values}
+                if len(by_id) != len(group):
+                    raise RetrievalError('missing_title_vector')
+                bank_hits = []
+                for row in group:
+                    value = by_id[row['id']]
+                    vector = value['vector']
+                    if (value['record_hash'] != row['record_hash'] or len(vector) != len(vectors[0])
+                            or not all(math.isfinite(v) for v in vector) or not any(vector)):
+                        raise RetrievalError('invalid_title_vector')
+                    scores = [sum(a * b for a, b in zip(vector, q)) /
+                              (math.sqrt(sum(a*a for a in vector)) * math.sqrt(sum(b*b for b in q)))
+                              for q in vectors]
+                    bank_hits.append((kind, {'id': row['id'], 'distance': max(scores),
+                        'entity': {'record_hash': row['record_hash']}, 'title_match': row['title_match']}))
+                hits.extend(bank_hits)
+            except Exception:
+                if kind not in failures:
+                    failures.append(kind)
+        order = {(row['type'], row['id']): index for index, row in enumerate(selected)}
+        hits.sort(key=lambda pair: order[(pair[0], pair[1]['id'])])
+        return hits, {'status': 'partial' if failures else 'complete', 'limit': TITLE_LIMIT,
+                      'matched_count': count, 'selected_count': len(hits),
+                      'overflow': count > TITLE_LIMIT, 'failed_databases': failures}
 
     def search(self, question, expansions=()):
         # GPU inference is serialized, but network retrieval can overlap across users.
@@ -184,6 +271,7 @@ class Retriever:
         if not isinstance(question, str) or not question.strip() or len(question) > 8000:
             raise RetrievalError('invalid_question')
         queries = [question]
+        named_queries = title_queries(question)
         for text in expansions:
             if isinstance(text, str) and text.strip() and text not in queries and len(text) <= 500:
                 queries.append(text)
@@ -229,7 +317,9 @@ class Retriever:
         failed_queries = []
         query_hits = [[] for _ in queries]
         successful_searches = 0
-        with ThreadPoolExecutor(max_workers=min(self.query_workers, len(queries))) as pool:
+        with ThreadPoolExecutor(max_workers=1) as title_pool, ThreadPoolExecutor(
+                max_workers=min(self.query_workers, len(queries))) as pool:
+            title_future = title_pool.submit(self._recall_titles, named_queries, vectors)
             # Every query is scheduled; map preserves query order for deterministic merging.
             for index, rows in mapped(pool, recall_query, range(len(queries))):
                 for kind, hits, failed in rows:
@@ -242,13 +332,14 @@ class Retriever:
                         previous = merged[kind].get(hit['id'])
                         if previous is None or hit['distance'] > previous['distance']:
                             merged[kind][hit['id']] = hit
+            title_hits, title_diagnostics = title_future.result()
         failures = [kind for kind in SEARCH_TYPES if any(row['type'] == kind for row in failed_queries)]
         hits_by_type = {kind: sorted(hits.values(), key=lambda hit: hit['distance'], reverse=True)
                         for kind, hits in merged.items()}
         if not successful_searches:
             raise RetrievalError('all_databases_failed')
         # Preserve cross-bank coverage before filling the original 12-candidate budget.
-        chosen = choose_candidates(hits_by_type, query_hits)
+        chosen = choose_candidates(hits_by_type, query_hits, title_hits=title_hits)
         query_best = [max(hits, key=lambda pair: pair[1]['distance']) if hits else None for hits in query_hits]
         selected_by_type = {kind: [hit for selected_kind, hit in chosen if selected_kind == kind]
                             for kind in SEARCH_TYPES}
@@ -293,10 +384,14 @@ class Retriever:
                 record = restore(parts)
                 if record['id'] != hit['id'] or sha(canonical(record)) != hit['entity']['record_hash']:
                     raise RetrievalError('index_content_mismatch')
+                match = title_match(record['title'], named_queries) if hit.get('title_match') else None
+                if hit.get('title_match') and not match:
+                    raise RetrievalError('title_content_mismatch')
                 records.append({'record': record, 'type': kind, 'score': hit['distance'],
                                 'best_query_indices': [i for i, best in enumerate(query_best)
                                                        if best and best[0] == kind and best[1]['id'] == hit['id']],
-                                'snapshot': self.manifest['snapshot']})
+                                'snapshot': self.manifest['snapshot'],
+                                **({'title_match': {**match, 'verified': True}} if match else {})})
             except Exception:
                 content_failures.append(hit['id'])
         related = related_guide_queries(question, records)
@@ -348,7 +443,9 @@ class Retriever:
         return {'candidates': records, 'queries': queries, 'failed_databases': failures,
                 'failed_queries': failed_queries,
                 'content_failures': content_failures, 'related_guide_failures': related_guide_failures,
-                'complete': not failures and not content_failures and not related_guide_failures,
+                'complete': not failures and not content_failures and not related_guide_failures
+                            and not title_diagnostics['failed_databases'],
+                'title_recall': title_diagnostics,
                 'snapshot': self.manifest['snapshot'], 'searched_types': list(SEARCH_TYPES),
                 'candidate_limit': MAX_CANDIDATES + (MAX_RELATED_GUIDES if related else 0),
                 'base_candidate_limit': MAX_CANDIDATES,

src/step3_question_answer/tests/test_recommendation.py

@@ -166,6 +166,8 @@ class FakeStore:
         if self.tamper:
             parts[0]['payload'] += 'tampered'
         return [p for p in parts if p['id'] in ids]
+    def titles(self, name):
+        return []
 
 
 class RetrievalTests(unittest.TestCase):
@@ -227,6 +229,8 @@ class RetrievalTests(unittest.TestCase):
                 return [[1.0, 0.0] for _ in texts]
         obj.embedder = Encoder()
         obj.lock = RLock()
+        obj._title_locks = {kind: RLock() for kind in SEARCH_TYPES}
+        obj._title_cache = {}
         obj.manifest = {'snapshot': 'synthetic', 'databases': {
             kind: {'index_collection': 'index', 'content_collection': 'content'} for kind in SEARCH_TYPES}}
         return obj

src/step3_question_answer/tests/test_title_recall.py

@@ -0,0 +1,163 @@
+import copy
+import unittest
+from unittest.mock import Mock
+
+from src.step2_knowledge_ingest.ingest import canonical, content_parts, sha
+from src.step3_question_answer.title_recall import title_queries, title_match, select_titles
+from src.step3_question_answer.retrieval import ReadStore
+from src.step3_question_answer.rerank import Reranker
+import test_recommendation as fixtures
+
+
+class TitleMatchTests(unittest.TestCase):
+    def test_original_questions_and_omitted_publisher(self):
+        title = '上海市青浦区数据局关于印发《青浦区关于加快推进数字经济和数智化建设的若干措施（试行）》的通知'
+        for question in ('《关于加快推进数字经济和数智化建设的若干措施》有那些内容啊，帮我总结一下',
+                         '加快推进数字经济和数智化建设有哪些措施', title):
+            self.assertIsNotNone(title_match(title, title_queries(question)))
+
+    def test_unrelated_topics_regions_years_and_versions_do_not_match(self):
+        for question, title in (
+            ('《合成职业培训补贴办法》请总结', '合成设备更新支持办法'),
+            ('《甲市技能培训办法》', '乙市技能培训办法'),
+            ('《2025年合成研发支持办法》', '2026年合成研发支持办法'),
+            ('《合成研发支持办法（修订）》', '合成研发支持办法（试行）'),
+            ('《根本不存在的政策文件》', '某市关于支持合成数字产业发展的措施'),
+        ):
+            self.assertIsNone(title_match(title, title_queries(question)))
+
+    def test_formatting_and_multiple_titles(self):
+        queries = title_queries('对比《合成创业支持办法》和《合成技能培训规定》')
+        self.assertEqual(len(queries), 2)
+        self.assertIsNotNone(title_match('合成 创业 支持办法', queries))
+        self.assertEqual(title_match('合成技能培训规定', queries)['query_index'], 1)
+        self.assertEqual(title_queries('《！！！？？？》'), [])
+        self.assertEqual(title_queries('你好'), [])
+        self.assertEqual(title_queries('有哪些措施'), [])
+
+    def test_global_limit_preserves_distinct_queries_and_versions(self):
+        entries = [{'id': str(i), 'type': 'article', 'title': title}
+                   for i, title in enumerate(['甲市合成创业支持办法', '乙市合成创业支持办法',
+                                             '丙市合成创业支持办法', '合成技能培训规定'])]
+        selected, count = select_titles(entries, title_queries('《合成创业支持办法》和《合成技能培训规定》'))
+        self.assertEqual(count, 4)
+        self.assertEqual(len(selected), 3)
+        self.assertIn('3', [row['id'] for row in selected])
+        self.assertEqual(len({row['id'] for row in selected}), 3)
+
+
+class TitleRetrievalTests(unittest.TestCase):
+    question = '《合成职业技能培训办法》帮我总结一下'
+
+    def make(self, count=1):
+        obj = fixtures.RetrievalTests().make()
+        records = []
+        for i in range(count):
+            record = copy.deepcopy(fixtures.RECORD)
+            record.update(id='k_' + f'{i + 10:024x}', title=f'合成职业技能培训办法（第{i + 1}版）')
+            record['summary'] = '合成技能服务。'
+            record['data']['support'] = '完整支持原文。' * (2000 if i == 0 else 1)
+            records.append(record)
+        store = obj._stores['article']
+        store.titles = Mock(return_value=[{'id': r['id'], 'title': r['title'],
+                                          'record_hash': sha(canonical(r))} for r in records])
+        store.index_vectors = Mock(side_effect=lambda name, ids: [
+            {'id': r['id'], 'vector': [0.1, 0.99], 'record_hash': sha(canonical(r))}
+            for r in records if r['id'] in ids])
+        parts = {p['id']: p for r in records for p in content_parts(r)}
+        store.get = lambda name, ids: [parts[i] for i in ids if i in parts]
+        return obj, records
+
+    def test_title_outside_vector_hits_is_restored_and_protected(self):
+        obj, records = self.make()
+        result = obj.search(self.question)
+        row = result['candidates'][0]
+        self.assertEqual(row['record'], records[0])
+        self.assertTrue(row['title_match']['verified'])
+        self.assertLess(row['score'], .2)  # Keep genuine vector score, never invent 1.0.
+        self.assertTrue(result['complete'])
+        reranked = Reranker(lambda *_: {'level': '不相关'}).select(fixtures.handoff(), result)
+        self.assertIn(records[0]['id'], [r['record']['id'] for r in reranked['candidates']])
+        self.assertTrue(next(r for r in reranked['rerank']['judgments']
+                             if r['source_id'] == records[0]['id'])['protected'])
+
+    def test_global_three_budget_cache_and_vector_deduplication(self):
+        obj, records = self.make(5)
+        store = obj._stores['article']
+        store.search = lambda *a, **k: [[{'id': records[0]['id'], 'distance': .8,
+                                         'entity': {'record_hash': sha(canonical(records[0]))}}]]
+        result = obj.search(self.question)
+        obj.search(self.question)
+        store.titles.assert_called_once()
+        self.assertEqual(result['title_recall']['selected_count'], 3)
+        self.assertTrue(result['title_recall']['overflow'])
+        self.assertLessEqual(len(result['candidates']), 12)
+        ids = [r['record']['id'] for r in result['candidates']]
+        self.assertEqual(ids.count(records[0]['id']), 1)
+
+    def test_title_failure_preserves_vector_hits_and_is_not_empty_success(self):
+        obj, _ = self.make()
+        store = obj._stores['article']
+        store.titles.side_effect = RuntimeError('synthetic directory failure')
+        result = obj.search(self.question)
+        self.assertFalse(result['complete'])
+        self.assertEqual(result['title_recall']['failed_databases'], ['article'])
+        self.assertTrue(result['candidates'])
+        self.assertNotIn('article', obj._title_cache)
+        store.titles.side_effect = None
+        self.assertTrue(obj.search(self.question)['complete'])
+
+    def test_missing_vectors_and_tampered_full_record_do_not_become_evidence(self):
+        obj, records = self.make()
+        obj._stores['article'].index_vectors.return_value = []
+        obj._stores['article'].index_vectors.side_effect = None
+        result = obj.search(self.question)
+        self.assertFalse(result['complete'])
+        self.assertFalse(any(r.get('title_match') for r in result['candidates']))
+        obj, records = self.make()
+        obj._stores['article'].get = lambda *a: []
+        result = obj.search(self.question)
+        self.assertIn(records[0]['id'], result['content_failures'])
+        self.assertFalse(result['complete'])
+
+    def test_plain_question_no_title_query_does_not_scan_catalogue(self):
+        obj, _ = self.make()
+        result = obj.search('我想开店')
+        obj._stores['article'].titles.assert_not_called()
+        self.assertEqual(result['title_recall']['status'], 'skipped')
+
+    def test_title_priority_with_eighteen_vector_hits_stays_within_twelve(self):
+        obj, target = self.make()
+        for bank, store in enumerate(obj._stores.values()):
+            records = []
+            for i in range(3):
+                r = copy.deepcopy(fixtures.RECORD)
+                r['id'] = 'k_' + f'{100 + bank * 10 + i:024x}'
+                records.append(r)
+            store.search = lambda *a, _records=records, **k: [[
+                {'id': r['id'], 'distance': .9, 'entity': {'record_hash': sha(canonical(r))}} for r in _records]]
+            parts = {p['id']: p for r in [*records, *target] for p in content_parts(r)}
+            store.get = lambda name, ids, _parts=parts: [_parts[i] for i in ids if i in _parts]
+        result = obj.search(self.question)
+        self.assertEqual(len(result['candidates']), 12)
+        self.assertEqual(result['candidates'][0]['record']['id'], target[0]['id'])
+        self.assertEqual(len({r['type'] for r in result['candidates']}), 6)
+
+
+class TitleDirectoryTests(unittest.TestCase):
+    def test_iterator_exhaustion_and_error_close(self):
+        store = ReadStore.__new__(ReadStore)
+        store.timeout = 1
+        store.client = Mock()
+        iterator = store.client.query_iterator.return_value
+        row = {'id': 'k_' + '1' * 24, 'index_text': '合成标题\n很长摘要', 'record_hash': 'a' * 64}
+        iterator.next.side_effect = [[row], []]
+        result = store.titles('pinned_index')
+        self.assertEqual(result[0]['title'], '合成标题')
+        self.assertNotIn('index_text', result[0])
+        iterator.close.assert_called_once()
+        iterator.reset_mock()
+        iterator.next.side_effect = [[row], [row]]
+        with self.assertRaisesRegex(Exception, 'invalid_title_index'):
+            store.titles('pinned_index')
+        iterator.close.assert_called_once()

src/step3_question_answer/title_recall.py

@@ -0,0 +1,72 @@
+"""Conservative literal title recall; similarity and document identity stay separate."""
+
+import re
+import unicodedata
+
+TITLE_LIMIT = 3
+
+
+def normalized(text):
+    return ''.join(c for c in unicodedata.normalize('NFKC', text).lower() if c.isalnum())
+
+
+def title_queries(question):
+    """Use explicit names, or a sufficiently specific topic before a question suffix.
+
+    Never invent names from model expansions, remove locations, years or versions.
+    """
+    quoted = re.findall(r'《([^《》]{4,200})》', question)
+    if quoted:
+        return list(dict.fromkeys(value for q in quoted if 4 <= len(value := normalized(q)) <= 200))[:12]
+    text = re.sub(r'^(?:请问|请|帮我|介绍一下|总结一下|我想了解|想了解)+', '', question.strip())
+    text = re.split(r'(?:都?有)?(?:哪些|那些|什么)(?:内容|措施|规定|政策|要求)?|[，,。？！?!]', text, maxsplit=1)[0]
+    text = re.sub(r'(?:帮我|请)?(?:介绍|总结|解读)(?:一下)?$', '', text).strip()
+    value = normalized(text)
+    return [value] if 8 <= len(value) <= 200 else []
+
+
+def title_match(title, queries):
+    """Substring equality is lexical evidence, never proof of eligibility/effectiveness."""
+    full = normalized(title)
+    inner = [normalized(t) for t in re.findall(r'《([^《》]+)》', title)]
+    matches = []
+    for index, query in enumerate(queries):
+        if query == full:
+            strength = 4
+        elif query in inner:
+            strength = 3
+        elif any(query in part for part in inner):
+            strength = 2
+        elif query in full:
+            strength = 1
+        else:
+            continue
+        matches.append((strength, len(query), -index))
+    if not matches:
+        return None
+    strength, length, negative_index = max(matches)
+    return {'query_index': -negative_index, 'strength': strength, 'matched_characters': length}
+
+
+def select_titles(entries, queries, limit=TITLE_LIMIT):
+    """Reserve one result per named document, then fill deterministically."""
+    matches = []
+    for entry in entries:
+        match = title_match(entry['title'], queries)
+        if match:
+            matches.append({**entry, 'title_match': match})
+    matches.sort(key=lambda row: (-row['title_match']['strength'],
+                                 -row['title_match']['matched_characters'], row['type'], row['id']))
+    selected, seen = [], set()
+    def add(row):
+        key = (row['type'], row['id'])
+        if key not in seen and len(selected) < limit:
+            selected.append(row)
+            seen.add(key)
+    for index in range(len(queries)):
+        first = next((row for row in matches if row['title_match']['query_index'] == index), None)
+        if first:
+            add(first)
+    for row in matches:
+        add(row)
+    return selected, len(matches)