279 lines
12 KiB
Python
279 lines
12 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 == "query_flow"
|
||
assert [item.workflow_type for item in bundle.workflows] == ["query_flow"]
|
||
assert bundle.workflows[0].metadata["message_intent"] == "requirement_plan"
|
||
assert bundle.workflows[0].metadata["consultant_mode_preference"] == "clarify"
|
||
|
||
|
||
def test_workflow_semantic_planner_keeps_scope_framing_discussion_in_query_flow() -> None:
|
||
bundle = WorkflowSemanticPlanner.plan_message(
|
||
message_text="我们想做一个面向医疗器械 BD 的政策雷达工具，先聚焦华东三省，先别创建项目，我还在补范围。",
|
||
source_system="feishu",
|
||
)
|
||
|
||
assert bundle.primary_workflow == "query_flow"
|
||
assert [item.workflow_type for item in bundle.workflows] == ["query_flow"]
|
||
assert bundle.workflows[0].metadata["message_intent"] == "requirement_plan"
|
||
assert bundle.workflows[0].metadata["consultant_mode_preference"] == "clarify"
|
||
|
||
|
||
def test_workflow_semantic_planner_marks_explicit_new_requirement_topic_switch() -> None:
|
||
bundle = WorkflowSemanticPlanner.plan_message(
|
||
message_text="不对啊，现在是另一个需求，先不要沿用刚才那版。",
|
||
source_system="feishu",
|
||
)
|
||
|
||
assert bundle.primary_workflow == "query_flow"
|
||
assert bundle.workflows[0].metadata["message_intent"] == "requirement_plan"
|
||
assert bundle.workflows[0].metadata["topic_switch"] == "new"
|
||
assert bundle.diagnostics["topic_switch"] == "new"
|
||
|
||
|
||
def test_workflow_semantic_planner_marks_chatty_new_requirement_phrase_as_topic_switch() -> None:
|
||
bundle = WorkflowSemanticPlanner.plan_message(
|
||
message_text="想跟你聊个新需求关于批量自动化部署openclaw的",
|
||
source_system="feishu",
|
||
)
|
||
|
||
assert bundle.primary_workflow == "query_flow"
|
||
assert bundle.workflows[0].metadata["message_intent"] == "requirement_plan"
|
||
assert bundle.workflows[0].metadata["topic_switch"] == "new"
|
||
assert bundle.diagnostics["topic_switch"] == "new"
|
||
|
||
|
||
def test_workflow_semantic_planner_routes_assignee_recommendation_back_to_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"]
|
||
assert bundle.workflows[0].metadata["message_intent"] == "business_review"
|
||
assert bundle.workflows[0].metadata["consultant_mode_preference"] == "advise"
|
||
|
||
|
||
def test_workflow_semantic_planner_routes_plain_assignee_recommendation_without_question_mark() -> None:
|
||
bundle = WorkflowSemanticPlanner.plan_message(
|
||
message_text="那你觉得这件事分配给谁做比较合适",
|
||
source_system="feishu",
|
||
)
|
||
|
||
assert bundle.primary_workflow == "query_flow"
|
||
assert bundle.workflows[0].metadata["message_intent"] == "business_review"
|
||
assert bundle.workflows[0].metadata["consultant_mode_preference"] == "advise"
|
||
|
||
|
||
def test_workflow_semantic_planner_marks_casual_chat_as_chat_query() -> None:
|
||
bundle = WorkflowSemanticPlanner.plan_message(
|
||
message_text="最近有点累，今天先不推进需求了，陪我简单聊两句。",
|
||
source_system="feishu",
|
||
)
|
||
|
||
assert bundle.primary_workflow == "query_flow"
|
||
assert bundle.workflows[0].metadata["message_intent"] == "casual_chat"
|
||
assert bundle.workflows[0].metadata["consultant_mode_preference"] == "brainstorm"
|
||
|
||
|
||
def test_workflow_semantic_planner_marks_requirement_followup_clarify_as_business_review() -> None:
|
||
bundle = WorkflowSemanticPlanner.plan_message(
|
||
message_text="那就先按上面的方向继续，但是别太重",
|
||
source_system="feishu",
|
||
)
|
||
|
||
assert bundle.primary_workflow == "query_flow"
|
||
assert bundle.workflows[0].metadata["message_intent"] == "business_review"
|
||
assert bundle.workflows[0].metadata["consultant_mode_preference"] == "clarify"
|
||
|
||
|
||
def test_workflow_semantic_planner_marks_requirement_scope_advise_as_business_review() -> None:
|
||
bundle = WorkflowSemanticPlanner.plan_message(
|
||
message_text="如果第一版先做 MVP，这个需求范围怎么收才比较稳？",
|
||
source_system="feishu",
|
||
)
|
||
|
||
assert bundle.primary_workflow == "query_flow"
|
||
assert bundle.workflows[0].metadata["message_intent"] == "business_review"
|
||
assert bundle.workflows[0].metadata["consultant_mode_preference"] == "advise"
|
||
|
||
|
||
def test_workflow_semantic_planner_routes_requirement_chat_into_requirement_plan() -> None:
|
||
bundle = WorkflowSemanticPlanner.plan_message(
|
||
message_text="想跟你聊聊需求，关于自动给每日的大A趋势打分",
|
||
source_system="feishu",
|
||
)
|
||
|
||
assert bundle.primary_workflow == "query_flow"
|
||
assert [item.workflow_type for item in bundle.workflows] == ["query_flow"]
|
||
assert bundle.workflows[0].metadata["message_intent"] == "requirement_plan"
|
||
assert bundle.workflows[0].metadata["consultant_mode_preference"] == "clarify"
|
||
assert bundle.diagnostics["message_intent"] == "requirement_plan"
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"message_text",
|
||
[
|
||
"技术指标吧",
|
||
"主要给投顾看",
|
||
"飞书日报就行",
|
||
"每天收盘后发",
|
||
"先给内部 PM 和研发负责人用",
|
||
"飞书总结就行",
|
||
"每次会后半小时内整理出来",
|
||
],
|
||
)
|
||
def test_workflow_semantic_planner_treats_short_requirement_slot_answers_as_requirement_plan(
|
||
message_text: str,
|
||
) -> None:
|
||
bundle = WorkflowSemanticPlanner.plan_message(
|
||
message_text=message_text,
|
||
source_system="feishu",
|
||
)
|
||
|
||
assert bundle.primary_workflow == "query_flow"
|
||
assert bundle.workflows[0].metadata["message_intent"] == "requirement_plan"
|
||
assert bundle.workflows[0].metadata["consultant_mode_preference"] == "clarify"
|
||
|
||
|
||
def test_workflow_semantic_planner_routes_generic_requirement_chat_into_requirement_plan() -> None:
|
||
bundle = WorkflowSemanticPlanner.plan_message(
|
||
message_text="想跟你聊个需求，我们开完会以后待办经常散掉，想做个小工具自动从会议纪要里提待办，先别创建，先收第一版。",
|
||
source_system="feishu",
|
||
)
|
||
|
||
assert bundle.primary_workflow == "query_flow"
|
||
assert bundle.workflows[0].metadata["message_intent"] == "requirement_plan"
|
||
assert bundle.workflows[0].metadata["consultant_mode_preference"] == "clarify"
|
||
|
||
|
||
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 == ""
|