# ============================================================ # 通用助理：拒识不再甩清单；概念题正面回答 # ============================================================ from __future__ import annotations import pytest from server.agent_core.assistant import reply as assistant_reply from server.agent_core.intent import parse_fast, recognize from server.agent_core.providers import reset_provider from server.contracts import IntentResult from server.aps_domain.workflow import _flex_compare, handle_intent from server.state.seed import seed_world class FakeStore: def __init__(self): self.data = seed_world() self._counters: dict[str, int] = {} def next_id(self, kind: str) -> int: self._counters[kind] = self._counters.get(kind, 0) + 1 return self._counters[kind] def save(self) -> None: pass @pytest.fixture(autouse=True) def _no_llm(monkeypatch): monkeypatch.delenv("LLM_API_KEY", raising=False) monkeypatch.delenv("LLM_PROVIDER", raising=False) reset_provider() yield reset_provider() @pytest.mark.asyncio async def test_recognize_falls_back_to_assistant(): world = seed_world() r = await recognize("今天天气怎么样顺便聊聊排产", world) assert r.intent == "assistant.reply" assert r.params.get("query") def test_fast_greeting_and_howto(): world = seed_world() assert parse_fast("你好", world).intent == "assistant.reply" assert parse_fast("你能做什么", world).intent == "assistant.reply" assert parse_fast("现在项目怎么样", world).intent == "assistant.reply" assert parse_fast("你给我解释什么叫排产", world).intent == "assistant.reply" assert parse_fast("你会排产吗", world).intent == "assistant.reply" def test_imperative_schedule_not_assistant_menu(): """「给我排产」必须直接开排，不能落到助理四选一。""" world = seed_world() for phrase in ( "给我排产", "我要你排产", "你给我排产", "开始排产", "排产吧", "给我模拟排产一下", "帮我排产一下", "模拟排产", "模拟排产一下", "给我模拟排产", "帮我模拟排产", "试排一版", "试排一下", "模拟排一下", ): r = parse_fast(phrase, world) assert r is not None, phrase assert r.intent in ("flex.schedule", "folder.schedule", "schedule.run"), ( phrase, r.intent ) def test_simulate_schedule_does_not_hijack_other_intents(): """「模拟排产」新增快路不得误吞向导/报工/提问句。""" world = seed_world() assert parse_fast("带我排一版", world).intent == "schedule.wizard" assert parse_fast("怎么排产", world).intent == "schedule.wizard" assert parse_fast("模拟报工", world).intent == "mes.report" r = parse_fast("帮我看看现在能不能排", world) assert r is not None and r.intent == "assistant.reply" assert parse_fast("解析项目文件夹", world).intent == "folder.analyze" assert parse_fast("帮我分析一下当前项目里面的数据", world).intent == "folder.analyze" def test_professional_starter_commands_keep_expected_intents(): world = seed_world() delivery = parse_fast("生成一版交期优先试排方案", world) assert delivery is not None and delivery.intent == "schedule.run" assert delivery.params.get("strategy") == "DELIVERY_FIRST" flex = parse_fast("生成一版柔性排产方案", world) assert flex is not None and flex.intent == "flex.schedule" def test_explicit_kitting_and_skill_strategy_phrases_keep_strategy_slots(): world = seed_world() for phrase, expected in ( ("按齐套优先试排", "KITTING_FIRST"), ("按技能优先试排", "SKILL_FIRST"), ): result = parse_fast(phrase, world) assert result is not None assert result.intent == "schedule.run" assert result.params.get("strategy") == expected def test_flex_compare_trial_only_uses_matrix_recommendation(monkeypatch): import server.aps_domain.flex as flex_module rows = [ {"sortMode": "SKILL_FIRST", "label": "技能优先", "strategyStatus": "DEGRADED", "totalTardiness": 0, "conflictCount": 0, "avgUtilization": 90, "onTimeCount": 10, "orderCount": 10, "vlCount": 10}, {"sortMode": "ASC", "label": "正排", "strategyStatus": "READY", "totalTardiness": 12, "conflictCount": 1, "avgUtilization": 80, "onTimeCount": 9, "orderCount": 10, "vlCount": 10}, ] monkeypatch.setattr(flex_module, "compare_sort_modes", lambda *_args, **_kwargs: { "rows": rows, "hint": "数据证据不足", "compressedDue": True, "dataQuality": { "productionReady": False, "warnings": [{"message": "设备能力仅覆盖 1/41 台设备"}], }, }) reply = _flex_compare(FakeStore()) table = reply.blocks[0].props["table"] assert table["recommendationId"] == "ASC" assert table["recommendationStatus"] == "TRIAL_ONLY" assert "试排较优" in reply.text assert "【正排】" in reply.text assert "数据待补" in reply.text assert "不能据此下达生产计划" in reply.text def test_flex_compare_without_ready_strategy_has_no_recommendation(monkeypatch): import server.aps_domain.flex as flex_module rows = [ {"sortMode": "KITTING_FIRST", "label": "齐套优先", "strategyStatus": "PARTIAL", "totalTardiness": 5, "conflictCount": 0, "avgUtilization": 80, "onTimeCount": 9, "orderCount": 10, "vlCount": 10}, {"sortMode": "SKILL_FIRST", "label": "技能优先", "strategyStatus": "DEGRADED", "totalTardiness": 0, "conflictCount": 0, "avgUtilization": 90, "onTimeCount": 10, "orderCount": 10, "vlCount": 10}, ] monkeypatch.setattr(flex_module, "compare_sort_modes", lambda *_args, **_kwargs: { "rows": rows, "hint": "策略输入不足", "compressedDue": True, "dataQuality": {"productionReady": False, "warnings": []}, }) reply = _flex_compare(FakeStore()) table = reply.blocks[0].props["table"] assert table["recommendationStatus"] == "UNAVAILABLE" assert table["recommendation"] is None assert table["recommendationId"] is None assert "当前没有可推荐策略" in reply.text assert "试排较优" not in reply.text @pytest.mark.asyncio async def test_what_time_is_it(): world = seed_world() reply = await assistant_reply(world, "现在几点了") assert "现在是" in reply.text assert "排产无关" not in reply.text assert "帮不上" not in reply.text @pytest.mark.asyncio async def test_unknown_handle_gives_project_brief_not_tip_spam(): store = FakeStore() intent = IntentResult(intent="unknown", params={"query": "随便说说"}, confidence=0.1, source="LLM") reply = await handle_intent(store, "s1", intent) assert "置信度不足" not in reply.text assert "我还没锁定" not in reply.text assert "排产无关" not in reply.text assert "你可以任选一句" not in reply.text @pytest.mark.asyncio async def test_explain_scheduling_not_inventory_dump(): world = seed_world() reply = await assistant_reply(world, "你给我解释什么叫排产") assert "什么是排产" in reply.text assert "可执行的生产计划" in reply.text assert "要排的单子" not in reply.text assert "你可以任选一句" not in reply.text assert not reply.blocks @pytest.mark.asyncio async def test_can_schedule_not_inventory_dump(): world = seed_world() reply = await assistant_reply(world, "你会排产吗？") assert "可以" in reply.text assert "排产阻断项" in reply.text assert "要排的单子" not in reply.text assert not reply.blocks @pytest.mark.asyncio async def test_assistant_offline_analyze_style(): world = seed_world() reply = await assistant_reply(world, "分析一下现状") assert "项目数据概览" in reply.text assert "建议操作" in reply.text @pytest.mark.asyncio async def test_assistant_professional_copy_has_no_casual_markers(): world = seed_world() replies = [ await assistant_reply(world, "你好"), await assistant_reply(world, "什么是排产"), await assistant_reply(world, "分析一下现状"), ] text = "\n".join(r.text for r in replies) for marker in ("啥", "咱们", "哟", "我在", "大白话", "冷笑话"): assert marker not in text @pytest.mark.asyncio async def test_assistant_chitchat_warm_guide(): world = seed_world() reply = await assistant_reply(world, "今天好累不想干") assert "置信度不足" not in reply.text assert "排产" in reply.text or "单子" in reply.text assert "你可以任选一句" not in reply.text