88 lines
3.6 KiB
Python
88 lines
3.6 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from app.services import steward_intent_bootstrap # noqa: F401 触发意图注册
|
||
|
|
from app.services.steward_intent_registry import (
|
||
|
|
all_flow_ids,
|
||
|
|
all_intents,
|
||
|
|
all_signal_keywords,
|
||
|
|
all_task_types,
|
||
|
|
field_allowlist_for,
|
||
|
|
get_intent,
|
||
|
|
resolve_intent_by_action,
|
||
|
|
resolve_task_type_for_flow,
|
||
|
|
)
|
||
|
|
from app.services.steward_constants import BUSINESS_CANONICAL_FIELDS
|
||
|
|
|
||
|
|
|
||
|
|
def test_registry_registers_application_reimbursement_and_query_intents():
|
||
|
|
task_types = all_task_types()
|
||
|
|
assert "expense_application" in task_types
|
||
|
|
assert "reimbursement" in task_types
|
||
|
|
assert "query_travel_standard" in task_types
|
||
|
|
|
||
|
|
application = get_intent("expense_application")
|
||
|
|
assert application is not None
|
||
|
|
assert application.assigned_agent == "application_assistant"
|
||
|
|
assert application.flow_id == "travel_application"
|
||
|
|
|
||
|
|
reimbursement = get_intent("reimbursement")
|
||
|
|
assert reimbursement is not None
|
||
|
|
assert reimbursement.assigned_agent == "reimbursement_assistant"
|
||
|
|
assert reimbursement.flow_id == "travel_reimbursement"
|
||
|
|
|
||
|
|
query = get_intent("query_travel_standard")
|
||
|
|
assert query is not None
|
||
|
|
assert query.assigned_agent == "policy_query_assistant"
|
||
|
|
assert query.flow_id is None # 查询意图不进入候选流程确认
|
||
|
|
|
||
|
|
|
||
|
|
def test_registry_aggregates_flow_ids_and_signal_keywords():
|
||
|
|
flow_ids = set(all_flow_ids())
|
||
|
|
assert flow_ids == {"travel_application", "travel_reimbursement"}
|
||
|
|
|
||
|
|
keywords = all_signal_keywords()
|
||
|
|
assert "出差" in keywords # 来自 expense_application
|
||
|
|
assert "报销" in keywords # 来自 reimbursement
|
||
|
|
assert "差旅标准" in keywords # 来自 query_travel_standard
|
||
|
|
|
||
|
|
|
||
|
|
def test_registry_resolves_intent_by_action_type():
|
||
|
|
assert resolve_intent_by_action("save_application_draft").task_type == "expense_application"
|
||
|
|
assert resolve_intent_by_action("submit_application").task_type == "expense_application"
|
||
|
|
assert resolve_intent_by_action("create_reimbursement_draft").task_type == "reimbursement"
|
||
|
|
assert resolve_intent_by_action("associate_attachments").task_type == "reimbursement"
|
||
|
|
assert resolve_intent_by_action("execute_travel_standard_query").task_type == "query_travel_standard"
|
||
|
|
assert resolve_intent_by_action("unknown_action") is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_registry_resolves_task_type_for_flow():
|
||
|
|
assert resolve_task_type_for_flow("travel_application") == "expense_application"
|
||
|
|
assert resolve_task_type_for_flow("travel_reimbursement") == "reimbursement"
|
||
|
|
assert resolve_task_type_for_flow("unknown_flow") is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_field_allowlist_uses_per_intent_overrides():
|
||
|
|
# 申请/报销沿用全局 BUSINESS_CANONICAL_FIELDS
|
||
|
|
application_fields = field_allowlist_for("expense_application")
|
||
|
|
assert "location" in application_fields
|
||
|
|
assert "amount" in application_fields
|
||
|
|
assert application_fields == frozenset(BUSINESS_CANONICAL_FIELDS)
|
||
|
|
|
||
|
|
# 查询意图使用专属槽位集合
|
||
|
|
query_fields = field_allowlist_for("query_travel_standard")
|
||
|
|
assert "location" in query_fields
|
||
|
|
assert "employee_grade" in query_fields
|
||
|
|
assert "standard_category" in query_fields
|
||
|
|
assert "amount" not in query_fields # 查询不需要金额
|
||
|
|
|
||
|
|
# 未注册意图回退到 fallback
|
||
|
|
fallback_fields = field_allowlist_for("unknown", fallback=frozenset({"foo"}))
|
||
|
|
assert fallback_fields == frozenset({"foo"})
|
||
|
|
|
||
|
|
|
||
|
|
def test_query_intent_prompt_fragment_includes_identification_guidance():
|
||
|
|
query = get_intent("query_travel_standard")
|
||
|
|
assert query is not None
|
||
|
|
assert "差旅" in query.prompt_fragment
|
||
|
|
assert "住宿标准" in query.prompt_fragment
|