73 lines
3.4 KiB
Python
73 lines
3.4 KiB
Python
import pytest
|
|
|
|
from app.services.ai_pmo.workflow_semantic_planner import (
|
|
WorkflowSemanticPlanner,
|
|
)
|
|
|
|
|
|
def test_workflow_semantic_planner_builds_multi_workflow_bundle() -> None:
|
|
bundle = WorkflowSemanticPlanner.plan_message(
|
|
message_text="帮我创建一个项目，并把这个需求接入信号源",
|
|
source_system="feishu",
|
|
)
|
|
|
|
workflow_types = [item.workflow_type for item in bundle.workflows]
|
|
|
|
assert bundle.source_system == "feishu"
|
|
assert bundle.primary_workflow == "new_project_flow"
|
|
assert workflow_types == ["new_project_flow", "signal_source_flow"]
|
|
assert bundle.reply_contract.mode == "aggregate_card"
|
|
|
|
|
|
def test_workflow_semantic_planner_derives_query_bundle() -> None:
|
|
bundle = WorkflowSemanticPlanner.plan_message(
|
|
message_text="现在项目进展怎么样？",
|
|
source_system="pmflow_manual",
|
|
)
|
|
|
|
assert bundle.primary_workflow == "query_flow"
|
|
assert [item.workflow_type for item in bundle.workflows] == ["query_flow"]
|
|
assert bundle.message_gate == "auto_execute"
|
|
|
|
|
|
def test_workflow_semantic_planner_helpers_cover_legacy_entry_semantics() -> None:
|
|
assert WorkflowSemanticPlanner.looks_like_project_lookup_question("现在有哪些项目？") is True
|
|
assert WorkflowSemanticPlanner.extract_project_lookup_keyword("请问项目 智能风控平台 有吗？") == "智能风控平台"
|
|
assert WorkflowSemanticPlanner.is_legacy_business_command("把项目状态改成已完成") is True
|
|
assert WorkflowSemanticPlanner.is_legacy_business_command("项目现在状态怎么样？") is False
|
|
assert WorkflowSemanticPlanner.wants_followup_continue("继续处理刚才那个") is True
|
|
assert WorkflowSemanticPlanner.wants_followup_status("它现在怎么样了？") is True
|
|
|
|
|
|
def test_workflow_semantic_planner_contextual_write_confirmation_is_stateful() -> None:
|
|
assert WorkflowSemanticPlanner.looks_like_contextual_write_confirmation("那就按这个继续执行") is True
|
|
assert WorkflowSemanticPlanner.looks_like_contextual_write_confirmation("这个项目现在怎么样？") is False
|
|
|
|
|
|
def test_workflow_semantic_planner_routes_new_business_review_with_five_lenses() -> None:
|
|
bundle = WorkflowSemanticPlanner.plan_message(
|
|
message_text="我要做一个大模型 API 聚合平台，帮我分析下要怎么做",
|
|
source_system="feishu",
|
|
)
|
|
|
|
assert bundle.primary_workflow == "query_flow"
|
|
assert [item.workflow_type for item in bundle.workflows] == ["query_flow"]
|
|
assert bundle.message_gate == "auto_execute"
|
|
assert bundle.workflows[0].metadata["message_intent"] == "business_review"
|
|
assert bundle.workflows[0].metadata["review_framework"] == "five_lenses"
|
|
assert bundle.diagnostics["message_intent"] == "business_review"
|
|
assert bundle.diagnostics["consultant_mode_preference"] == "clarify"
|
|
|
|
|
|
def test_workflow_semantic_planner_marks_correction_escalation_queries() -> None:
|
|
bundle = WorkflowSemanticPlanner.plan_message(
|
|
message_text="不对，调用高级模型重新回答我的问题",
|
|
source_system="feishu",
|
|
)
|
|
|
|
assert bundle.primary_workflow == "query_flow"
|
|
assert bundle.workflows[0].metadata["message_intent"] == "correction_escalation"
|
|
assert bundle.workflows[0].metadata["force_thinking_mode"] == "deep"
|
|
assert bundle.diagnostics["message_intent"] == "correction_escalation"
|
|
assert bundle.diagnostics["consultant_mode_preference"] == "challenge"
|