228 lines
8.7 KiB
Python
228 lines
8.7 KiB
Python
# ============================================================
|
|
# 通用助理：拒识不再甩清单；概念题正面回答
|
|
# ============================================================
|
|
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
|