diff --git a/CLAUDE.md b/CLAUDE.md index 2141f1e5..5dd5d7a4 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -572,7 +572,8 @@ uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 - AdaptiveRAG 场景化 Prompt 选择器必须在 `mixin_deep_guard_answer.py` 组装 Prompt 前执行，并遵循固定优先级：成员负载 > 趋势（项目关联 external_trend / 通用 industry_trend） > mixed > AI PMO 入口 > 默认 RAG - 飞书中的“想聊需求 / 先别创建 / 先收敛第一版”类探索问题必须先由 `workflow_semantic_planner` 标记为 `message_intent=requirement_plan`，再统一落到 `PROMPT_AI_PMO_REQUIREMENT_PLAN_QA` 的需求收敛场景；该场景禁止落回“五大口径”评审或直接自动创建 - `requirement_plan` 现在必须是多轮状态机，而不是单轮问一句：`ExecuteQueryProjectMixin` 要把同线程/同群同人的需求收敛状态持久化到 `mounted_topic_workspaces + mounted_draft_workspaces`，每轮只推进 1 个关键问题，信息足够后直接整理成 Feishu 友好的需求草稿 -- `requirement_plan` 不允许再把“同群同人”直接等同于“同一个需求 topic”：Mounted Runtime 必须用 `conversation_scope_key` 归并会话范围，再用显式 `topic_key` 区分具体需求；用户说“另一个需求 / 换个需求 / 不是这个需求”时必须新开 topic，并把旧 topic 置为 `suspended` +- `requirement_plan` 不允许再把“同群同人”直接等同于“同一个需求 topic”：Mounted Runtime 必须用 `conversation_scope_key` 归并会话范围，再用显式 `topic_key` 区分具体需求；用户说“另一个需求 / 换个需求 / 不是这个需求 / 聊个新需求”时必须新开 topic，并把旧 topic 置为 `suspended` +- `requirement_plan` 的 topic resolver 不能只靠显式关键词：当同会话存在 active plan workspace，且新消息是新的需求起手式、与当前 topic card/draft 的重叠度明显偏低时，必须按 `candidate_new_topic` 优先切新 topic；尤其旧 topic 已处于 `awaiting_execute_confirm / plan_ready` 时，不得继续把“关于 openclaw 批量部署”这类低重叠消息续写到“A 股打分工具”草稿上 - `workflow_semantic_planner` 必须把“先按技术指标来 / 主要给投顾看 / 先做成飞书日报 / 每天收盘后更新”这类短槽位回答继续识别为 `message_intent=requirement_plan`，禁止第二轮开始掉回普通 query 或高阶 RAG - `workflow_semantic_planner` 也必须把通用需求收敛里的短槽位回答继续识别为 `message_intent=requirement_plan`，例如“先给内部 PM 和研发负责人用 / 飞书总结就行 / 每次会后半小时内整理出来”；这类碎片化回复不得因为没有新 `semantic_bundle` 就退回普通 query - `ExecuteQueryProjectMixin` 对 `message_intent=requirement_plan` 的查询必须优先走 mounted runtime 直答快路径：默认只输出自然短话术，每轮最多只保留 1 个关键问题；当 `scoring_basis/target_user/output_form/cadence/first_scope` 收集完成时，必须直接整理成需求草稿，禁止继续空转追问 diff --git a/backend/app/services/ai_pmo/ai_pmo_typed_executor_execute/mixin_execute_query_project.py b/backend/app/services/ai_pmo/ai_pmo_typed_executor_execute/mixin_execute_query_project.py index 88a6c160..533d7d48 100644 --- a/backend/app/services/ai_pmo/ai_pmo_typed_executor_execute/mixin_execute_query_project.py +++ b/backend/app/services/ai_pmo/ai_pmo_typed_executor_execute/mixin_execute_query_project.py @@ -610,10 +610,193 @@ class ExecuteQueryProjectMixin: return False if WorkflowSemanticPlanner._looks_like_requirement_plan_slot_answer(message_text): return True - if cls._extract_requirement_plan_updates(message_text=message_text, card={}): + material_updates = { + key: value + for key, value in cls._extract_requirement_plan_updates(message_text=message_text, card={}).items() + if key not in {"plan_profile", "title", "goal"} + } + if material_updates: return True return any(token in normalized for token in ("整理成草稿", "出草稿", "汇总一下", "按这版", "就这样先收")) + @classmethod + def _collect_requirement_topic_text_fragments(cls, payload: Any) -> list[str]: + fragments: list[str] = [] + if isinstance(payload, dict): + for value in payload.values(): + fragments.extend(cls._collect_requirement_topic_text_fragments(value)) + return fragments + if isinstance(payload, (list, tuple, set)): + for item in payload: + fragments.extend(cls._collect_requirement_topic_text_fragments(item)) + return fragments + text = str(payload or "").strip() + if text: + fragments.append(text) + return fragments + + @classmethod + def _build_requirement_topic_overlap_tokens(cls, text: str) -> set[str]: + normalized = WorkflowSemanticPlanner._normalize_text(text) + if not normalized: + return set() + for stopword in ( + "想跟你聊", + "跟你聊", + "聊聊", + "聊个", + "聊需求", + "先不创建", + "先别创建", + "先不要创建", + "先收第一版", + "先收一版", + "第一版", + "关于", + "需求", + "项目", + "工具", + "系统", + "平台", + "方案", + "一个", + "这个", + "那个", + "我们", + "这次", + "现在", + ): + normalized = normalized.replace(stopword, "") + tokens: set[str] = { + token + for token in re.findall(r"[a-z][a-z0-9_-]{1,}", normalized) + if len(token) >= 2 + } + chinese_text = re.sub(r"[a-z0-9_-]+", "", normalized) + for index in range(len(chinese_text) - 1): + token = chinese_text[index : index + 2] + if len(token.strip()) == 2: + tokens.add(token) + return tokens + + @classmethod + def _compute_requirement_topic_overlap_score( + cls, + *, + topic: Optional[MountedTopicWorkspace], + draft: Optional[MountedDraftWorkspace], + message_text: str, + ) -> float: + context_parts: list[str] = [] + if topic is not None: + context_parts.extend(cls._collect_requirement_topic_text_fragments(topic.facts_json)) + context_parts.extend(cls._collect_requirement_topic_text_fragments(topic.goals_json)) + context_parts.extend(cls._collect_requirement_topic_text_fragments(topic.constraints_json)) + if draft is not None: + context_parts.extend(cls._collect_requirement_topic_text_fragments(draft.requirement_card)) + context_parts.extend(cls._collect_requirement_topic_text_fragments(draft.draft_payload)) + topic_tokens = cls._build_requirement_topic_overlap_tokens(" ".join(context_parts)) + message_tokens = cls._build_requirement_topic_overlap_tokens(message_text) + if not topic_tokens or not message_tokens: + return 0.0 + overlap = len(topic_tokens & message_tokens) + denominator = max(1, min(len(topic_tokens), len(message_tokens))) + return round(overlap / denominator, 4) + + @classmethod + def _looks_like_requirement_new_topic_candidate(cls, *, message_text: str) -> bool: + normalized = WorkflowSemanticPlanner._normalize_text(message_text) + if not normalized: + return False + if WorkflowSemanticPlanner.looks_like_explicit_new_requirement_topic(message_text): + return True + if cls._looks_like_requirement_plan_continuation(message_text=message_text): + return False + if not cls._looks_like_requirement_plan_seed(message_text=message_text): + return False + return ( + WorkflowSemanticPlanner._looks_like_requirement_plan_chat(message_text) + or WorkflowSemanticPlanner.looks_like_project_create_request(message_text) + or any(token in normalized for token in ("想做", "做个", "做一个", "有个需求", "关于")) + ) + + @classmethod + def _should_switch_requirement_topic_by_overlap( + cls, + *, + topic: MountedTopicWorkspace, + draft: Optional[MountedDraftWorkspace], + message_text: str, + ) -> bool: + if not cls._looks_like_requirement_new_topic_candidate(message_text=message_text): + return False + overlap_score = cls._compute_requirement_topic_overlap_score( + topic=topic, + draft=draft, + message_text=message_text, + ) + plan_status = str(getattr(topic, "plan_status", "") or "").strip().lower() + if plan_status in {"awaiting_execute_confirm", "plan_ready"}: + return overlap_score <= 0.12 + if plan_status == "options_ready": + return overlap_score <= 0.08 + return overlap_score == 0.0 + + async def _find_active_requirement_plan_workspace( + self, + *, + source_system: str, + source_message_id: str, + source_chat_id: Optional[str], + sender: Optional[str], + source_sender_open_id: Optional[str], + thread_context: Optional[dict[str, Any]], + conversation_key: Optional[str], + ) -> tuple[Optional[MountedTopicWorkspace], Optional[MountedDraftWorkspace]]: + if not self._supports_requirement_plan_workspace(self.db): + return None, None + topic_key, conditions, root_message_id, thread_id, scope_key = self._build_requirement_plan_anchor_candidates( + source_system=source_system, + source_message_id=source_message_id, + source_chat_id=source_chat_id, + sender=sender, + source_sender_open_id=source_sender_open_id, + thread_context=thread_context, + conversation_key=conversation_key, + force_new_topic=False, + ) + topic_rows = ( + await self.db.execute( + select(MountedTopicWorkspace).where( + MountedTopicWorkspace.tenant_id == int(self.tenant_id), + MountedTopicWorkspace.source_system == str(source_system or "").strip().lower(), + MountedTopicWorkspace.mode == "plan", + MountedTopicWorkspace.status == "active", + or_(*conditions), + ) + .order_by(MountedTopicWorkspace.updated_at.desc(), MountedTopicWorkspace.id.desc()) + ) + ).scalars().all() + topic = self._select_requirement_plan_topic( + candidates=list(topic_rows), + topic_key=topic_key, + root_message_id=root_message_id, + thread_id=thread_id, + conversation_scope_key=scope_key or conversation_key, + ) + if topic is None: + return None, None + draft = ( + await self.db.execute( + select(MountedDraftWorkspace).where( + MountedDraftWorkspace.tenant_id == int(self.tenant_id), + MountedDraftWorkspace.topic_workspace_id == int(topic.id), + MountedDraftWorkspace.state.in_(("collecting", "ready")), + ) + ) + ).scalar_one_or_none() + return topic, draft + async def _load_requirement_plan_workspace( self, *, @@ -723,9 +906,9 @@ class ExecuteQueryProjectMixin: conversation_key: Optional[str], force_new_topic: bool = False, ) -> bool: - if not self._supports_requirement_plan_workspace(self.db): + if force_new_topic: return False - topic_key, conditions, root_message_id, thread_id, scope_key = self._build_requirement_plan_anchor_candidates( + topic, draft = await self._find_active_requirement_plan_workspace( source_system=source_system, source_message_id=source_message_id, source_chat_id=source_chat_id, @@ -733,41 +916,42 @@ class ExecuteQueryProjectMixin: source_sender_open_id=source_sender_open_id, thread_context=thread_context, conversation_key=conversation_key, - force_new_topic=force_new_topic, ) - if force_new_topic: - return False - topic_rows = ( - await self.db.execute( - select(MountedTopicWorkspace).where( - MountedTopicWorkspace.tenant_id == int(self.tenant_id), - MountedTopicWorkspace.source_system == str(source_system or "").strip().lower(), - MountedTopicWorkspace.mode == "plan", - MountedTopicWorkspace.status == "active", - or_(*conditions), - ) - .order_by(MountedTopicWorkspace.updated_at.desc(), MountedTopicWorkspace.id.desc()) - ) - ).scalars().all() - topic = self._select_requirement_plan_topic( - candidates=list(topic_rows), - topic_key=topic_key, - root_message_id=root_message_id, - thread_id=thread_id, - conversation_scope_key=scope_key or conversation_key, + return topic is not None and draft is not None + + async def _infer_requirement_plan_topic_switch( + self, + *, + message_text: str, + source_system: str, + source_message_id: str, + source_chat_id: Optional[str], + sender: Optional[str], + source_sender_open_id: Optional[str], + thread_context: Optional[dict[str, Any]], + conversation_key: Optional[str], + message_intent: Optional[str], + ) -> Optional[str]: + if str(message_intent or "").strip().lower() != "requirement_plan": + return None + topic, draft = await self._find_active_requirement_plan_workspace( + source_system=source_system, + source_message_id=source_message_id, + source_chat_id=source_chat_id, + sender=sender, + source_sender_open_id=source_sender_open_id, + thread_context=thread_context, + conversation_key=conversation_key, ) if topic is None: - return False - draft = ( - await self.db.execute( - select(MountedDraftWorkspace).where( - MountedDraftWorkspace.tenant_id == int(self.tenant_id), - MountedDraftWorkspace.topic_workspace_id == int(topic.id), - MountedDraftWorkspace.state.in_(("collecting", "ready")), - ) - ) - ).scalar_one_or_none() - return draft is not None + return None + if self._should_switch_requirement_topic_by_overlap( + topic=topic, + draft=draft, + message_text=message_text, + ): + return "new" + return None async def _build_requirement_planning_direct_result( self, @@ -1189,6 +1373,18 @@ class ExecuteQueryProjectMixin: topic_switch = "new" message_intent = message_intent or "requirement_plan" consultant_mode_preference = consultant_mode_preference or "clarify" + if not topic_switch: + topic_switch = await self._infer_requirement_plan_topic_switch( + message_text=message_text, + source_system=source_system, + source_message_id=source_message_id, + source_chat_id=source_chat_id, + sender=None, + source_sender_open_id=source_sender_open_id, + thread_context=thread_context, + conversation_key=conversation_key, + message_intent=message_intent, + ) if not message_intent and self._looks_like_requirement_plan_continuation(message_text=message_text): has_active_plan_workspace = await self._has_active_requirement_plan_workspace( source_system=source_system, diff --git a/backend/app/services/ai_pmo/workflow_semantic_planner/constants.py b/backend/app/services/ai_pmo/workflow_semantic_planner/constants.py index d4e4d03f..c2f14b1c 100644 --- a/backend/app/services/ai_pmo/workflow_semantic_planner/constants.py +++ b/backend/app/services/ai_pmo/workflow_semantic_planner/constants.py @@ -176,9 +176,22 @@ REQUIREMENT_PLAN_CHAT_HINTS = ( "先不要入库", ) REQUIREMENT_TOPIC_SWITCH_NEW_HINTS = ( + "聊个新需求", + "聊一个新需求", + "聊聊新需求", + "有个新需求", + "这是个新需求", + "这是新需求", + "这次是新需求", + "现在是新需求", + "换个新需求", + "换一个新需求", + "切到新需求", + "切个新需求", "另一个需求", "换个需求", "换一个需求", + "新的需求", "不是这个需求", "不是刚才那个需求", "别沿用刚才那版", @@ -186,6 +199,17 @@ REQUIREMENT_TOPIC_SWITCH_NEW_HINTS = ( "另起一个需求", "重新开一个需求", "单开一个需求", + "跟上一个无关", + "和上一个无关", + "不是延续刚才那个", + "不是接着刚才那个", +) +REQUIREMENT_TOPIC_SWITCH_NEW_PATTERNS = ( + re.compile(r"(?:聊|说|讨论|讲).{0,6}(?:个|一个)?新需求", re.IGNORECASE), + re.compile(r"(?:这是|这次是|现在是|我有|这边有|还有|另外有).{0,8}(?:个|一个)?新需求", re.IGNORECASE), + re.compile(r"(?:切|换|转).{0,4}(?:到|个|一个)?(?:新需求|新的需求|新话题|newtopic|topic)", re.IGNORECASE), + re.compile(r"(?:另起|单开|重新开|新开|开个新|拉个新|起个新).{0,6}(?:需求|话题|topic)", re.IGNORECASE), + re.compile(r"(?:跟上一个无关|和上一个无关|不是延续刚才那个|不是接着刚才那个)", re.IGNORECASE), ) CORRECTION_ESCALATION_HINTS = ( "重新回答", diff --git a/backend/app/services/ai_pmo/workflow_semantic_planner/service.py b/backend/app/services/ai_pmo/workflow_semantic_planner/service.py index 1849fb40..052c7604 100644 --- a/backend/app/services/ai_pmo/workflow_semantic_planner/service.py +++ b/backend/app/services/ai_pmo/workflow_semantic_planner/service.py @@ -65,6 +65,7 @@ from .constants import ( QUESTION_HINTS, REQUIREMENT_PLAN_CHAT_HINTS, REQUIREMENT_TOPIC_SWITCH_NEW_HINTS, + REQUIREMENT_TOPIC_SWITCH_NEW_PATTERNS, REPLY_MODE_CARD, REPLY_MODE_TEXT, ROUTE_FOLLOWUP_CONFIRM, @@ -254,6 +255,8 @@ class WorkflowSemanticPlanner: return False if any(token in normalized for token in REQUIREMENT_TOPIC_SWITCH_NEW_HINTS): return True + if any(pattern.search(str(text or "")) for pattern in REQUIREMENT_TOPIC_SWITCH_NEW_PATTERNS): + return True if "现在是" in normalized and any(token in normalized for token in ("另一个需求", "新的需求", "新话题")): return True if "不对" in normalized and any(token in normalized for token in ("另一个需求", "不是这个需求", "别沿用", "不要沿用")): @@ -272,7 +275,7 @@ class WorkflowSemanticPlanner: ): return False slot_hits = 0 - slot_hits += int(any(token in normalized for token in ("技术指标", "均线", "成交量", "涨跌比", "macd", "rsi", "规则", "自动化"))) + slot_hits += int(any(token in normalized for token in ("技术指标", "均线", "成交量", "涨跌比", "macd", "rsi", "规则"))) slot_hits += int(any(token in normalized for token in ("情绪", "涨停家数", "北向", "北向资金", "舆情", "热度"))) slot_hits += int( any( @@ -299,7 +302,7 @@ class WorkflowSemanticPlanner: slot_hits += int(any(token in normalized for token in ("第一版", "先保留", "先只做", "别做太重", "先做成", "先按"))) has_planning_prefix = any( token in normalized - for token in ("先按", "主要给", "先给", "先做成", "第一版", "先保留") + for token in ("先按", "主要给", "先给", "先做成", "先保留") ) has_generic_target_user_shape = any( token in normalized @@ -310,6 +313,8 @@ class WorkflowSemanticPlanner: for token in ("会后", "半小时", "当天内", "整理出来", "整理出", "产出") ) compact_length = len(normalized) + if cls._looks_like_requirement_plan_chat(text) and slot_hits <= 1 and not has_generic_target_user_shape and not has_generic_cadence_shape: + return False return ( slot_hits >= 2 or (has_planning_prefix and slot_hits >= 1) @@ -465,6 +470,8 @@ class WorkflowSemanticPlanner: return True if any(token in normalized for token in PROJECT_CREATE_DEFERRED_CONFIRM_HINTS): return True + if cls._looks_like_requirement_plan_chat(text): + return True if cls._looks_like_requirement_plan_slot_answer(text): return True return cls._looks_like_assignee_recommendation(text) diff --git a/backend/tests/test_ai_pmo_ingress_mainline.py b/backend/tests/test_ai_pmo_ingress_mainline.py index 64865abe..801e9685 100644 --- a/backend/tests/test_ai_pmo_ingress_mainline.py +++ b/backend/tests/test_ai_pmo_ingress_mainline.py @@ -1949,6 +1949,114 @@ async def test_execute_query_intent_explicit_new_requirement_topic_suspends_prev assert rows[1].plan_status == "collecting" +@pytest.mark.asyncio +async def test_execute_query_intent_low_overlap_requirement_seed_switches_new_topic( + monkeypatch, + requirement_plan_db_session, +) -> None: + class FakeAdaptiveRAGService: + def __init__(self, db, tenant_id: int): + _ = (db, tenant_id) + + async def answer_question(self, **kwargs): + _ = kwargs + raise AssertionError("low-overlap requirement switching should not enter heavy RAG path") + + class FakeExecutor(mixin_execute_query_project.ExecuteQueryProjectMixin): + def __init__(self) -> None: + self.db = requirement_plan_db_session + self.tenant_id = 3 + + monkeypatch.setattr(mixin_execute_query_project, "AdaptiveRAGService", FakeAdaptiveRAGService) + executor = FakeExecutor() + + current_topic = MountedTopicWorkspace( + tenant_id=3, + topic_key="feishu:chat-plan-overlap:root:msg-plan-overlap-root", + source_system="feishu", + channel_ref="chat-plan-overlap", + conversation_scope_key=build_feishu_conversation_scope_key( + source_chat_id="chat-plan-overlap", + source_sender_open_id="ou-plan-overlap", + sender=None, + ), + root_message_id="msg-plan-overlap-root", + mode="plan", + status="active", + plan_status="awaiting_execute_confirm", + facts_json={ + "title": "每日 A 股趋势打分工具", + "target_user": "投顾", + "scoring_basis": "技术指标（均线、成交量、涨跌比等）", + "output_form": "飞书日报", + }, + goals_json=["每天产出一个可解释的 A 股趋势分数，辅助判断和记录。"], + constraints_json={"first_scope": "先保留一个分数、方向判断和一句解释"}, + pending_actions=["confirm_requirement_draft"], + metadata_json={"plan_mode": True}, + ) + requirement_plan_db_session.add(current_topic) + await requirement_plan_db_session.flush() + requirement_plan_db_session.add( + MountedDraftWorkspace( + tenant_id=3, + topic_workspace_id=int(current_topic.id), + state="ready", + draft_version=1, + requirement_card={ + "title": "每日 A 股趋势打分工具", + "goal": "每天产出一个可解释的 A 股趋势分数，辅助判断和记录。", + "target_user": "投顾", + "scoring_basis": "技术指标（均线、成交量、涨跌比等）", + "output_form": "飞书日报", + "cadence": "每日", + "first_scope": "先保留一个分数、方向判断和一句解释", + "plan_profile": "scoring", + }, + draft_payload={"title": "每日 A 股趋势打分工具"}, + required_confirmations=["需求草稿需要人工确认后才允许正式创建"], + ) + ) + await requirement_plan_db_session.flush() + + implicit_switch_intent = SimpleNamespace( + project_id=None, + metadata={ + "source_chat_id": "chat-plan-overlap", + "source_sender_open_id": "ou-plan-overlap", + "semantic_bundle": { + "diagnostics": { + "message_intent": "requirement_plan", + "consultant_mode_preference": "clarify", + } + }, + }, + ) + + switch_result = await executor._execute_query_intent( + intent=implicit_switch_intent, + message_text="想跟你聊需求，关于批量自动化部署openclaw，先不创建，先收第一版。", + operator_user_id=999, + source_system="feishu", + source_message_id="msg-plan-overlap-new", + ) + + rows = ( + await requirement_plan_db_session.execute( + select(MountedTopicWorkspace) + .where(MountedTopicWorkspace.tenant_id == 3) + .order_by(MountedTopicWorkspace.id.asc()) + ) + ).scalars().all() + + assert len(rows) == 2 + assert "每日 A 股趋势打分工具" not in str(switch_result["answer"] or "") + assert rows[0].status == "suspended" + assert rows[0].plan_status == "suspended" + assert rows[1].status == "active" + assert rows[1].plan_status == "collecting" + + @pytest.mark.asyncio async def test_execute_query_intent_requirement_plan_accepts_fragmented_short_answers_and_only_drafts_when_scope_arrives( monkeypatch, diff --git a/backend/tests/test_workflow_semantic_planner.py b/backend/tests/test_workflow_semantic_planner.py index 76ad2ad9..9ae72ac6 100644 --- a/backend/tests/test_workflow_semantic_planner.py +++ b/backend/tests/test_workflow_semantic_planner.py @@ -129,6 +129,18 @@ def test_workflow_semantic_planner_marks_explicit_new_requirement_topic_switch() 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="那你觉得分配给谁做比较合适？",