business-assistant/src/step3_question_answer/intent_check.py

"""Opt-in real intent model check with frozen synthetic inputs and downstream spies.

Calls DeepSeek only; never calls QCC, Milvus or the local embedding model.
"""

import json
from pathlib import Path
from unittest.mock import Mock

from .graph import build_graph
from .intent import deepseek_analyzer
from .persona import DIRECT_RESPONSES


def main():
    root = Path(__file__).resolve().parents[2]
    cases = json.loads(Path(__file__).with_name('tests').joinpath('conversation_cases.json').read_text(encoding='utf-8'))
    analyzer = deepseek_analyzer(root / '.env')
    from src.step1_knowledge_prepare.summarize import load_config
    model = load_config(root / '.env').model
    rows = []
    for case in cases['cases']:
        client, retriever, recommender = Mock(), Mock(), Mock()
        retriever.search.return_value = {'candidates': [], 'complete': True}
        recommender.plan.return_value = {'queries': [], 'error': None}
        recommender.answer.return_value = {'overview': '合成下游答复'}
        graph = build_graph(client, analyzer, retriever=retriever, recommender=recommender)
        state = graph.invoke({'question': case['question']}, {'configurable': {'thread_id': case['id']}})
        actual = state['conversation_kind']
        expected = case['kind']
        no_downstream = not client.mock_calls and not retriever.mock_calls and not recommender.mock_calls
        routing_ok = (state.get('__interrupt__') or retriever.search.called) if expected == 'specific_need' else (
            no_downstream if expected in ('smalltalk',) else retriever.search.called and not client.mock_calls)
        direct = expected in ('smalltalk',)
        reply_ok = (not state.get('errors') and bool(state.get('direct_response'))
                    and state['response'] != DIRECT_RESPONSES.get(expected)) if direct else True
        if case.get('identity'):
            reply_ok = reply_ok and '青浦区营商助手' in state['response']
        row = {'id': case['id'], 'expected': expected, 'actual': actual,
               'passed': bool(actual == expected and routing_ok and reply_ok)}
        if direct:
            # Frozen synthetic prompts only, for semantic review of model replies.
            row['response'] = state['response']
        rows.append(row)
        print(json.dumps(row, ensure_ascii=False), flush=True)
    report = {'version': cases['version'], 'model': model, 'rows': rows,
              'passed': all(row['passed'] for row in rows)}
    print(json.dumps(report, ensure_ascii=False), flush=True)
    return 0 if report['passed'] else 1


if __name__ == '__main__':
    raise SystemExit(main())