business-assistant/src/step3_question_answer/tests/test_exact_qa_fast_path.py

import copy
import threading
import unittest
from unittest.mock import Mock

from src.step3_question_answer.graph import build_graph
from src.step3_question_answer.recommendation import Recommender
from src.step3_question_answer.timing import Timings, bind
from src.step3_question_answer.tests.test_knowledge_answers import RECORD, QUESTION, ANSWER, fixture, context
from src.step3_question_answer.tests.test_recommendation import candidate
from src.step4_api.runtime import GraphRuntime


class ExactQaFastPathTests(unittest.TestCase):
    def mixed(self, records=None, complete=True):
        data = fixture(records, complete)
        data['candidates'].insert(0, candidate())
        return data

    def test_exact_answer_skips_all_other_candidates_and_rerank(self):
        model, reranker = Mock(), Mock()
        engine = Recommender(model, reranker=reranker)
        engine._analyze_candidates = Mock(side_effect=AssertionError('must not start analysis'))
        original = self.mixed()
        before = copy.deepcopy(original)
        timing = Timings()
        with bind(timing):
            result = engine.answer(context(), original)
        model.assert_not_called()
        reranker.select.assert_not_called()
        engine._analyze_candidates.assert_not_called()
        self.assertEqual(original, before)
        self.assertEqual(result['summary']['text'], '> ' + ANSWER)
        self.assertEqual(result['summary']['source_ids'], [RECORD['id']])
        self.assertEqual(result['sources'][0]['full_record'], RECORD)
        self.assertEqual(len(result['sources']), 1)
        self.assertEqual(len(result['candidate_diagnostics']), 1)
        self.assertEqual(result['items'], [])
        self.assertEqual(result['next_actions'], [])
        self.assertEqual(result['follow_up_questions'], [])
        self.assertEqual(result['retrieval']['fast_path'], {
            'kind': 'exact_qa', 'matched_count': 1, 'skipped_candidate_count': 1})
        self.assertTrue(result['complete'])
        self.assertEqual(set(timing.snapshot()['spans']), {'exact_qa'})

    def test_all_conflicting_answers_preserved_without_model(self):
        second = copy.deepcopy(RECORD)
        second['id'] = 'k_' + 'c' * 24
        second['data']['answer'] = '需要提交复印件。'
        model = Mock()
        result = Recommender(model).answer(context(), self.mixed([RECORD, second]))
        model.assert_not_called()
        self.assertEqual(result['summary']['status'], 'qa_conflict')
        self.assertIn('不同答案', result['summary']['text'])
        self.assertIn(ANSWER, result['summary']['text'])
        self.assertIn(second['data']['answer'], result['summary']['text'])
        self.assertEqual(len(result['summary']['source_ids']), 2)
        self.assertEqual(len(result['sources']), 2)

    def test_partial_retrieval_stays_partial(self):
        model = Mock()
        result = Recommender(model).answer(context(), self.mixed(complete=False))
        model.assert_not_called()
        self.assertFalse(result['complete'])
        self.assertFalse(result['retrieval_analysis_complete'])
        self.assertTrue(result['summary']['notices'])
        self.assertEqual(result['retrieval']['failed_databases'], ['qa'])
        self.assertIn(ANSWER, result['summary']['text'])

    def test_nonexact_blank_and_expired_answers_keep_normal_flow(self):
        variants = []
        for field, value in [('question', QUESTION + '和原件'), ('answer', ''),
                             ('answer', '本实施意见自2020年1月1日起执行，至2020年12月31日止。')]:
            record = copy.deepcopy(RECORD)
            record['data'][field] = value
            variants.append(record)
        for record in variants:
            with self.subTest(record=record['data']):
                model = Mock(return_value={'relevant': False})
                reranker = Mock()
                reranker.select.side_effect = lambda handoff, data: data
                result = Recommender(model, reranker=reranker).answer(context(), self.mixed([record]))
                reranker.select.assert_called_once()
                self.assertTrue(model.called)
                self.assertNotIn('fast_path', result['retrieval'])
                self.assertNotEqual(result['summary']['status'], 'exact_qa')

    def test_attachment_failure_is_not_hidden_by_shortcut(self):
        model = Mock(return_value={'relevant': False})
        reranker = Mock()
        reranker.select.side_effect = lambda handoff, data: data
        result = Recommender(model, reranker=reranker).answer({**context(), 'attachments': [
            {'id': 'attachment_test', 'title': '合成附件', 'status': 'failed', 'error': 'download_failed'}]}, self.mixed())
        reranker.select.assert_called_once()
        self.assertNotIn('fast_path', result['retrieval'])
        self.assertFalse(result['complete'])
        self.assertIn('未能读取', result['summary']['text'])

    def test_api_graph_delivers_literal_answer_and_sources_without_generation(self):
        retriever, model, reranker = Mock(), Mock(), Mock()
        retriever.search.return_value = self.mixed()
        engine = Recommender(model, reranker=reranker)
        graph = build_graph(None, lambda _: {'conversation_kind': 'policy_overview', 'needs_company': False},
                            retriever=retriever, recommender=engine)
        events = []
        GraphRuntime(graph).run('synthetic-qa', QUESTION, lambda e, d: events.append((e, d)), threading.Event())
        model.assert_not_called()
        reranker.select.assert_not_called()
        self.assertEqual([d['text'] for e, d in events if e == 'summary'], ['> ' + ANSWER])
        self.assertEqual([d['id'] for e, d in events if e == 'source'], [RECORD['id']])
        self.assertFalse(any(e in ('item', 'answer_delta') for e, _ in events))
        self.assertEqual(events[-1], ('done', {'status': 'completed'}))
        result = next(d for e, d in events if e == 'result')
        self.assertEqual(result['recommendation']['summary']['citations'][0]['quote'], ANSWER)