136 lines
6.2 KiB
Python
136 lines
6.2 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"
|
|
|
|
|
|
def test_workflow_semantic_planner_treats_natural_problem_statement_as_new_project_candidate() -> None:
|
|
bundle = WorkflowSemanticPlanner.plan_message(
|
|
message_text="最近直播复盘太慢了，我想做个给运营用的小工具，能自动切片、总结亮点掉点，再给复盘清单。",
|
|
source_system="feishu",
|
|
)
|
|
|
|
assert bundle.primary_workflow == "new_project_flow"
|
|
assert [item.workflow_type for item in bundle.workflows] == ["new_project_flow"]
|
|
assert bundle.workflows[0].workflow_action == "create"
|
|
|
|
|
|
def test_workflow_semantic_planner_keeps_natural_solution_consulting_in_query_flow() -> None:
|
|
bundle = WorkflowSemanticPlanner.plan_message(
|
|
message_text="我们最近直播复盘很慢，先帮我分析一下第一版应该怎么做，不要直接创建。",
|
|
source_system="feishu",
|
|
)
|
|
|
|
assert bundle.primary_workflow == "query_flow"
|
|
assert [item.workflow_type for item in bundle.workflows] == ["query_flow"]
|
|
|
|
|
|
def test_workflow_semantic_planner_keeps_deferred_creation_request_in_new_project_flow() -> None:
|
|
bundle = WorkflowSemanticPlanner.plan_message(
|
|
message_text="最近直播复盘太慢了，想做个给运营团队用的小工具，先不要直接创建，我再补充几轮，最后确认再入库。",
|
|
source_system="feishu",
|
|
)
|
|
|
|
assert bundle.primary_workflow == "new_project_flow"
|
|
assert [item.workflow_type for item in bundle.workflows] == ["new_project_flow"]
|
|
|
|
|
|
def test_workflow_semantic_planner_treats_natural_draft_details_as_revision() -> None:
|
|
action, note = WorkflowSemanticPlanner.parse_draft_action(
|
|
message_text="第一版先做直播录屏和评论回放切片，再加亮点和掉点总结，最后给一份可执行复盘清单。"
|
|
)
|
|
|
|
assert action == "revise"
|
|
assert note == "第一版先做直播录屏和评论回放切片，再加亮点和掉点总结，最后给一份可执行复盘清单。"
|
|
|
|
|
|
def test_workflow_semantic_planner_does_not_treat_smalltalk_as_revision() -> None:
|
|
action, note = WorkflowSemanticPlanner.parse_draft_action(message_text="收到，辛苦了")
|
|
|
|
assert action == "unknown"
|
|
assert note == ""
|
|
|
|
|
|
def test_workflow_semantic_planner_treats_scope_narrowing_as_revision_not_cancel() -> None:
|
|
message = "第一版先别做太大，先把三件事做好就行：直播录屏和评论回放切片、亮点和掉点自动总结、给团队一份能直接执行的复盘清单。"
|
|
|
|
action, note = WorkflowSemanticPlanner.parse_draft_action(message_text=message)
|
|
|
|
assert action == "revise"
|
|
assert note == message
|
|
|
|
|
|
def test_workflow_semantic_planner_keeps_terminal_stop_phrase_as_cancel() -> None:
|
|
action, note = WorkflowSemanticPlanner.parse_draft_action(message_text="这个需求先别做了，取消吧。")
|
|
|
|
assert action == "cancel"
|
|
assert note == ""
|