diff --git a/CLAUDE.md b/CLAUDE.md index f780d103..dddd2ecb 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -538,9 +538,11 @@ uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 - SSTC 发布验证清单必须优先消费 `derived_artifacts.verification_metrics`，其次才允许读取宪章 KPI；禁止再回退 3 条通用默认验收项 - AdaptiveRAG `adaptive_rag_answer_normal` 空响应必须显式抛错并交由 high failover 处理；禁止在生成层直接回落为“补充问题细节后重试”之类的兜底回复 - 用户可见问答链路（`function_name` 含 `answer/reply/qa`）禁止使用 `reasoning_content` 回退，避免思考过程外漏 +- `integrations(type='model_routing')` 的 runtime tier override 不得把 `adaptive_rag_answer_normal` / `adaptive_rag_answer_deep` 这类用户可见 QA 主函数降级到 `low`；这两个函数必须优先服从 OpenClaw 的 high reasoning 规则，防止网页/飞书问答被后台 override 锁死在低档 - AdaptiveRAG 场景化 Prompt 选择器必须在 `mixin_deep_guard_answer.py` 组装 Prompt 前执行，并遵循固定优先级：成员负载 > 趋势（项目关联 external_trend / 通用 industry_trend） > mixed > AI PMO 入口 > 默认 RAG - 飞书私聊中的“新业务/需求/MVP/方案/怎么做”类探索问题必须先由 `workflow_semantic_planner` 标记为 `message_intent=business_review`，再统一落到 `PROMPT_AI_PMO_QA` 的五大口径评审场景；禁止继续复用 `PROMPT_FEISHU_DIRECT_QA` 生成类似普通 RAG 直答的结果 - 飞书私聊中的“答错了/不对/调用高级模型/用更强模型/重新回答”必须被识别为 `correction_escalation` 指令：优先复用上一轮上下文主题并强制深度回答，禁止把这类话术当成新的系统配置查询 +- `POST /rag/ask` 在 `rag_request_context` 或统一语义规划层命中 `business_review/correction_escalation` 时，必须强制留在 `query.read_only` 主链并进入对应 QA Prompt；禁止因为 LLM 意图分类把“分析/MVP/重答”误送到 `signal.create -> draft_pending_confirm` - “高级模型是否启用/模型路由是否启用/当前是否高配”这类运行态问题，只有命中运行态配置、路由快照、引擎配置或真实调用记录时才允许下结论；若证据只有页面目录、代码图谱或功能清单，回复必须明确为“无法仅凭静态资料确认” - 通用行业趋势问答（如“宠物行业最新动态”）不得注入数据库、项目历史、用户历史或外部实时搜索结果；只有显式带项目上下文时才允许走 `external_trend`，对“不要提项目/别提项目”这类否定语句不能误判为项目上下文 - mixed 场景回复必须强制两段格式输出：`[回答]` 与 `[已收到需求] ...，已入库处理`，禁止把需求段落并入问答段落 diff --git a/backend/app/api/ai_pmo_api/helpers_inbox_filters.py b/backend/app/api/ai_pmo_api/helpers_inbox_filters.py index 97135f74..eb5f6421 100644 --- a/backend/app/api/ai_pmo_api/helpers_inbox_filters.py +++ b/backend/app/api/ai_pmo_api/helpers_inbox_filters.py @@ -1,5 +1,7 @@ from __future__ import annotations +import re + from .shared import * def _normalize_noise_text(text: str) -> str: diff --git a/backend/app/api/ai_pmo_api/helpers_inbox_payload.py b/backend/app/api/ai_pmo_api/helpers_inbox_payload.py index 20ca6a61..2e6ea77f 100644 --- a/backend/app/api/ai_pmo_api/helpers_inbox_payload.py +++ b/backend/app/api/ai_pmo_api/helpers_inbox_payload.py @@ -1,5 +1,7 @@ from __future__ import annotations +import re + from .shared import * from .helpers_inbox_filters import ( _blocking_reason, diff --git a/backend/app/services/ai_pmo/ai_pmo_kernel_ingress_flow/mixin_entry.py b/backend/app/services/ai_pmo/ai_pmo_kernel_ingress_flow/mixin_entry.py index 28362300..dbd81cee 100644 --- a/backend/app/services/ai_pmo/ai_pmo_kernel_ingress_flow/mixin_entry.py +++ b/backend/app/services/ai_pmo/ai_pmo_kernel_ingress_flow/mixin_entry.py @@ -2,11 +2,26 @@ from __future__ import annotations from .shared import * from app.utils.thread_context import build_feishu_conversation_scope_key +from app.services.ai_pmo.workflow_semantic_planner import WorkflowSemanticPlanner +from app.services.ai_pmo.workflow_semantic_planner.constants import ( + QUERY_SCENE_BUSINESS_REVIEW, + QUERY_SCENE_CORRECTION_ESCALATION, +) class KernelIngressFlowEntryMixin: _MIXED_HINT_CONNECTORS = ("顺便", "另外", "同时", "并且", "还有", "再问") _MIXED_HINT_QUESTION = ("?", "？", "问下", "多少", "几个", "进展", "状态", "吗", "呢", "如何", "怎么", "有没有") _MIXED_HINT_REQUIREMENT = ("创建", "新建", "新增", "安排", "执行", "处理", "接入", "录入", "添加") + _RAG_QUERY_SCENE_KEYS = ( + "message_intent", + "review_framework", + "consultant_mode_preference", + "force_thinking_mode", + ) + _RAG_QUERY_SCENES = frozenset({ + QUERY_SCENE_BUSINESS_REVIEW, + QUERY_SCENE_CORRECTION_ESCALATION, + }) @classmethod def _looks_like_mixed_candidate(cls, *, text: str) -> bool: @@ -25,6 +40,129 @@ class KernelIngressFlowEntryMixin: def _supports_mixed_sidecar(*, source_system: str) -> bool: return str(source_system or "").strip().lower() in {"feishu", "pmflow_manual", "ai_pmo_ui"} + @classmethod + def _build_forced_query_scene_metadata( + cls, + *, + source_system: str, + message_text: str, + rag_request_context: Optional[dict[str, Any]], + semantic_bundle: Optional[dict[str, Any]], + ) -> dict[str, Any]: + context = rag_request_context if isinstance(rag_request_context, dict) else {} + semantic_diagnostics = ( + semantic_bundle.get("diagnostics") + if isinstance(semantic_bundle, dict) and isinstance(semantic_bundle.get("diagnostics"), dict) + else {} + ) + semantic_primary_workflow = ( + str(semantic_bundle.get("primary_workflow") or "").strip().lower() + if isinstance(semantic_bundle, dict) + else "" + ) + derived = WorkflowSemanticPlanner._derive_query_scene_metadata(message_text) + scene_metadata: dict[str, Any] = { + key: value + for key, value in derived.items() + if key in cls._RAG_QUERY_SCENE_KEYS and value not in (None, "") + } + if str(source_system or "").strip().lower() == "feishu" and semantic_primary_workflow == "query_flow": + for key in cls._RAG_QUERY_SCENE_KEYS: + value = semantic_diagnostics.get(key) + if value in (None, ""): + continue + text = str(value).strip() + if text: + scene_metadata[key] = text.lower() if key != "review_framework" else text + for key in cls._RAG_QUERY_SCENE_KEYS: + value = context.get(key) + if value in (None, ""): + continue + text = str(value).strip() + if text: + scene_metadata[key] = text.lower() if key != "review_framework" else text + return scene_metadata + + @classmethod + def _should_force_query_scene( + cls, + *, + source_system: str, + rag_request_context: Optional[dict[str, Any]], + semantic_bundle: Optional[dict[str, Any]], + scene_metadata: dict[str, Any], + ) -> bool: + message_intent = str(scene_metadata.get("message_intent") or "").strip().lower() + if message_intent not in cls._RAG_QUERY_SCENES: + return False + if isinstance(rag_request_context, dict): + action_mode = str(rag_request_context.get("action_mode") or "auto").strip().lower() or "auto" + return action_mode not in {"write", "read_write"} + if str(source_system or "").strip().lower() == "feishu": + primary_workflow = ( + str(semantic_bundle.get("primary_workflow") or "").strip().lower() + if isinstance(semantic_bundle, dict) + else "" + ) + return primary_workflow == "query_flow" + return False + + def _coerce_intent_for_query_scene( + self, + *, + intent: CanonicalIntent, + source_system: str, + message_text: str, + rag_request_context: Optional[dict[str, Any]], + semantic_bundle: Optional[dict[str, Any]], + world_model: dict[str, Any], + ) -> CanonicalIntent: + scene_metadata = self._build_forced_query_scene_metadata( + source_system=source_system, + message_text=message_text, + rag_request_context=rag_request_context, + semantic_bundle=semantic_bundle, + ) + if not self._should_force_query_scene( + source_system=source_system, + rag_request_context=rag_request_context, + semantic_bundle=semantic_bundle, + scene_metadata=scene_metadata, + ): + return intent + + project_id = _safe_int(intent.project_id) or _safe_int(world_model.get("resolved_project_id")) + semantic_bundle = WorkflowSemanticPlanner.plan_from_intent( + intent_name=INTENT_QUERY, + message_text=message_text, + source_system="ai_pmo_ingress", + workflow_route=ROUTE_QUERY_ANSWER, + project_id=project_id, + metadata={ + "confidence": max(float(intent.confidence or 0.0), 0.86), + **scene_metadata, + }, + ).to_dict() + metadata = { + **(intent.metadata or {}), + **scene_metadata, + "context_source": "rag_request_context", + "workflow_route": ROUTE_QUERY_ANSWER, + "workflow_family": ROUTE_FAMILY_QUERY, + "workflow_action": ROUTE_ACTION_ANSWER, + "requires_confirmation": False, + "semantic_bundle": semantic_bundle, + } + pretty = re.sub(r"\s+", " ", str(message_text or "")).strip() + return CanonicalIntent( + name=INTENT_QUERY, + confidence=max(float(intent.confidence or 0.0), 0.86), + project_id=project_id, + signal_title="", + signal_summary=pretty[:300], + metadata=metadata, + ) + async def _prepare_mixed_sidecar_context( self, *, @@ -112,7 +250,13 @@ class KernelIngressFlowEntryMixin: message_sent_at: Optional[Any] = None, session_snapshot: Optional[dict[str, Any]] = None, thread_context: Optional[dict[str, Any]] = None, + semantic_bundle: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: + ingress_options: dict[str, Any] = {} + if isinstance(thread_context, dict) and thread_context: + ingress_options["thread_context"] = dict(thread_context) + if isinstance(semantic_bundle, dict) and semantic_bundle: + ingress_options["semantic_bundle"] = dict(semantic_bundle) return await self.process_ingress( source_system="feishu", source_message_id=source_message_id, @@ -124,11 +268,7 @@ class KernelIngressFlowEntryMixin: chat_type=chat_type, message_sent_at=message_sent_at, session_snapshot=session_snapshot, - ingress_options=( - {"thread_context": dict(thread_context)} - if isinstance(thread_context, dict) and thread_context - else None - ), + ingress_options=ingress_options or None, ) async def process_ingress( @@ -233,6 +373,16 @@ class KernelIngressFlowEntryMixin: _log_perf("_build_world_model", stage_started_at) self._schedule_portfolio_snapshot_persist(world_model=world_model) work_item.state = STATE_MODELED + rag_request_context = ( + ingress_options.get("rag_request_context") + if isinstance(ingress_options, dict) + else None + ) + semantic_bundle_override = ( + ingress_options.get("semantic_bundle") + if isinstance(ingress_options, dict) and isinstance(ingress_options.get("semantic_bundle"), dict) + else None + ) stage_started_at = time.perf_counter() intent, intent_context_source = await self._resolve_intent( message_text=clean_text, @@ -242,6 +392,14 @@ class KernelIngressFlowEntryMixin: session_snapshot=session_snapshot, thread_context=thread_context, ) + intent = self._coerce_intent_for_query_scene( + intent=intent, + source_system=source_system, + message_text=clean_text, + rag_request_context=rag_request_context, + semantic_bundle=semantic_bundle_override, + world_model=world_model, + ) intent = self._enrich_signal_source_intent_from_closure( intent=intent, message_text=clean_text, @@ -252,11 +410,6 @@ class KernelIngressFlowEntryMixin: **(intent.metadata or {}), **mixed_meta, } - rag_request_context = ( - ingress_options.get("rag_request_context") - if isinstance(ingress_options, dict) - else None - ) if isinstance(rag_request_context, dict): intent.metadata = { **(intent.metadata or {}), diff --git a/backend/app/services/llm_service/mixin_engine_resolution_core.py b/backend/app/services/llm_service/mixin_engine_resolution_core.py index 5e7d7341..9ac36509 100644 --- a/backend/app/services/llm_service/mixin_engine_resolution_core.py +++ b/backend/app/services/llm_service/mixin_engine_resolution_core.py @@ -3,6 +3,11 @@ from __future__ import annotations from .shared import * class LLMEngineResolutionCoreMixin: + _PROTECTED_QUERY_HIGH_FUNCTIONS = frozenset({ + "adaptive_rag_answer_deep", + "adaptive_rag_answer_normal", + }) + def _stub_response(self, prompt: str) -> str: _ = prompt # 保留签名，避免调用方改动 return ( @@ -262,6 +267,12 @@ class LLMEngineResolutionCoreMixin: if not override or not bool(override.get("enabled", True)): return None tier_alias = _optional_tier_alias(override.get("tier_alias") or override.get("tier")) or "high" + if tier_alias == "low" and fn in self._PROTECTED_QUERY_HIGH_FUNCTIONS: + _runtime_log( + "LLM路由", + f"函数覆盖低档降级已忽略：function={fn}, protected_reason=query_high_guard", + ) + return None config_bundle = load_model_config() matched_rule = find_matching_rule( fn, diff --git a/backend/app/services/signal/signal_capture_service/mixin_capture_entry.py b/backend/app/services/signal/signal_capture_service/mixin_capture_entry.py index ddd1b88a..65a43439 100644 --- a/backend/app/services/signal/signal_capture_service/mixin_capture_entry.py +++ b/backend/app/services/signal/signal_capture_service/mixin_capture_entry.py @@ -115,6 +115,12 @@ class SignalCaptureEntryMixin: sync_reason = "short_confirmation" if should_sync_process: + semantic_bundle = None + if str(sync_reason or "").strip().lower() == "semantic:query_flow": + semantic_bundle = WorkflowSemanticPlanner.plan_message( + message_text=cleaned_message, + source_system="feishu", + ).to_dict() self._emit_ai_think( "capture_sync_high_accuracy", trace_id=trace_id, @@ -140,6 +146,7 @@ class SignalCaptureEntryMixin: message_sent_at=effective_message_sent_at, session_snapshot=effective_session_snapshot, thread_context=thread_context, + semantic_bundle=semantic_bundle, ) if self._is_non_requirement_message(cleaned_message) and not self._is_question_like(cleaned_message): @@ -270,6 +277,7 @@ class SignalCaptureEntryMixin: message_sent_at: Optional[Any] = None, session_snapshot: Optional[Dict[str, Any]] = None, thread_context: Optional[Dict[str, Any]] = None, + semantic_bundle: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: pipeline_result = await kernel.process_feishu_message( group_chat_id=group_chat_id, @@ -282,6 +290,7 @@ class SignalCaptureEntryMixin: message_sent_at=message_sent_at, session_snapshot=session_snapshot, thread_context=thread_context, + semantic_bundle=semantic_bundle, ) if not bool(pipeline_result.get("handled")): self._emit_ai_think( diff --git a/backend/tests/test_ai_pmo_ingress_mainline.py b/backend/tests/test_ai_pmo_ingress_mainline.py index 37a06870..70fa0bdd 100644 --- a/backend/tests/test_ai_pmo_ingress_mainline.py +++ b/backend/tests/test_ai_pmo_ingress_mainline.py @@ -5,6 +5,8 @@ import pytest from fastapi import HTTPException from app.api.ai_pmo_api import routes_core +from app.api.ai_pmo_api import helpers_inbox_filters +from app.api.ai_pmo_api import helpers_inbox_payload from app.api.ai_pmo_api.schemas import PMOProcessTextRequest from app.services.ai_pmo.consultant_dispatch import build_consultant_dispatch_contract from app.services.ai_pmo.ai_pmo_kernel.mixins.intent.ai_pmo_kernel_mixins_intent_resolve import ( @@ -290,6 +292,116 @@ def test_build_feishu_thread_key_prefers_actor_scope_without_anchor() -> None: ) == "feishu:chat-1:ou_dm_1" +def test_low_value_closure_payload_helper_handles_summary_text() -> None: + assert helpers_inbox_filters._is_low_value_closure_payload( + {"closure_summary": "待补充核心痛点"} + ) is False + + +def test_serialize_work_item_summary_handles_message_text_without_nameerror() -> None: + row = SimpleNamespace( + id=9, + source_system="pmflow_manual", + source_message_id="msg-9", + draft_content={}, + state="received", + review_summary={}, + action_bundle={}, + intent="query.read_only", + project_id=None, + signal_id=None, + reply_template_key=None, + execution_result={}, + draft_expires_at=None, + draft_version=0, + confirm_message_id=None, + created_at=None, + updated_at=None, + decision_level="portfolio", + ) + ingest = SimpleNamespace(payload={"message_text": " 最近 状态 怎么样 "}) + + payload = helpers_inbox_payload._serialize_work_item_summary( + row=row, + ingest=ingest, + project=None, + ) + + assert payload["source_message_text"] == "最近 状态 怎么样" + + +def test_coerce_rag_business_review_scene_keeps_rag_ask_on_query_chain() -> None: + class FakeKernel(mixin_entry.KernelIngressFlowEntryMixin): + def __init__(self) -> None: + self.tenant_id = 1 + + kernel = FakeKernel() + original_intent = mixin_entry.CanonicalIntent( + name="signal.create", + confidence=0.91, + project_id=None, + signal_title="大模型 API 聚合平台", + signal_summary="我现在要做一个大模型api聚合平台，你帮我分析下要怎么做，给出mvp需求", + metadata={"llm_function": "classify_intent_ai_pmo_kernel"}, + ) + + coerced = kernel._coerce_intent_for_query_scene( + intent=original_intent, + source_system="pmflow_manual", + message_text="我现在要做一个大模型api聚合平台，你帮我分析下要怎么做，给出mvp需求", + rag_request_context={ + "asked_via": "pmflow_manual", + "action_mode": "auto", + }, + semantic_bundle=None, + world_model={}, + ) + + assert coerced.name == "query.read_only" + assert coerced.metadata["message_intent"] == "business_review" + assert coerced.metadata["consultant_mode_preference"] == "clarify" + assert coerced.metadata["workflow_route"] == "query.answer" + semantic_bundle = coerced.metadata["semantic_bundle"] + assert semantic_bundle["diagnostics"]["message_intent"] == "business_review" + assert semantic_bundle["diagnostics"]["workflow_route"] == "query.answer" + + +def test_coerce_feishu_semantic_business_review_bundle_keeps_query_chain() -> None: + class FakeKernel(mixin_entry.KernelIngressFlowEntryMixin): + def __init__(self) -> None: + self.tenant_id = 1 + + kernel = FakeKernel() + original_intent = mixin_entry.CanonicalIntent( + name="signal.create", + confidence=0.93, + project_id=None, + signal_title="大模型 API 聚合平台", + signal_summary="我现在要创建需求为一个大模型api聚合平台，你帮我分析下要怎么做，", + metadata={"llm_function": "classify_intent_ai_pmo_kernel"}, + ) + + coerced = kernel._coerce_intent_for_query_scene( + intent=original_intent, + source_system="feishu", + message_text="我现在要创建需求为一个大模型api聚合平台，你帮我分析下要怎么做，", + rag_request_context=None, + semantic_bundle={ + "primary_workflow": "query_flow", + "diagnostics": { + "message_intent": "business_review", + "review_framework": "five_lenses", + "consultant_mode_preference": "clarify", + }, + }, + world_model={}, + ) + + assert coerced.name == "query.read_only" + assert coerced.metadata["message_intent"] == "business_review" + assert coerced.metadata["workflow_route"] == "query.answer" + + @pytest.mark.asyncio async def test_kernel_intent_resolve_inherits_recent_feishu_session_context_without_reply_anchor() -> None: captured: dict[str, object] = {} diff --git a/backend/tests/test_feishu_pipeline_queue_contract.py b/backend/tests/test_feishu_pipeline_queue_contract.py index fa02ac5b..42eec39e 100644 --- a/backend/tests/test_feishu_pipeline_queue_contract.py +++ b/backend/tests/test_feishu_pipeline_queue_contract.py @@ -407,6 +407,66 @@ async def test_feishu_question_like_messages_use_sync_high_accuracy_path(monkeyp assert calls["cache"] == 1 +@pytest.mark.asyncio +async def test_feishu_business_review_query_forwards_semantic_bundle_into_kernel(monkeypatch) -> None: + captured: dict[str, object] = {} + + class _FakeKernel: + @staticmethod + def _build_source_message_id(*, source_system: str, source_message_id, channel_ref: str, sender: str, message_text: str): + _ = (source_system, source_message_id, channel_ref, sender, message_text) + return "msg-business-review" + + def __init__(self, db, tenant_id: int): + _ = (db, tenant_id) + + async def _get_or_create_session(self, **kwargs): + _ = kwargs + raise AssertionError("business review sync path should not create async queue session") + + def _is_short_confirmation_message(self, **kwargs): + _ = kwargs + return False + + async def process_feishu_message(self, **kwargs): + captured["kwargs"] = kwargs + return { + "handled": True, + "status": "answered", + "classification": "question", + "linked_project_ids": [], + "signal_id": None, + "created_project_id": None, + "auto_tasks": [], + "ai_trace_id": "trace-br", + "direct_reply_required": True, + "direct_reply_text": "五大口径评审中。", + } + + async def _noop_sync_recent_cache(self, **kwargs): + _ = (self, kwargs) + + monkeypatch.setattr(mixin_capture_entry, "AIPMOKernelService", _FakeKernel) + monkeypatch.setattr(ProactiveSignalCaptureService, "_sync_feishu_recent_cache", _noop_sync_recent_cache) + + service = ProactiveSignalCaptureService(object()) + result = await service.capture_from_feishu( + group_chat_id="chat-br-1", + message_content="我现在要创建需求为一个大模型api聚合平台，你帮我分析下要怎么做，", + sender="测试用户", + tenant_id=1, + feishu_message_id="msg-business-review", + chat_type="group", + ) + + assert result["status"] == "answered" + kwargs = captured["kwargs"] + assert isinstance(kwargs, dict) + semantic_bundle = kwargs["semantic_bundle"] + assert semantic_bundle["primary_workflow"] == "query_flow" + assert semantic_bundle["diagnostics"]["message_intent"] == "business_review" + + def test_capture_from_feishu_no_longer_references_legacy_fallback() -> None: source = inspect.getsource(mixin_capture_entry.SignalCaptureEntryMixin.capture_from_feishu) diff --git a/backend/tests/test_model_routing_service.py b/backend/tests/test_model_routing_service.py index 34a39dd6..15637645 100644 --- a/backend/tests/test_model_routing_service.py +++ b/backend/tests/test_model_routing_service.py @@ -166,6 +166,44 @@ async def test_model_routing_override_is_tier_only_and_reads_last_execution(db_e assert "api_env" not in mapping +@pytest.mark.asyncio +async def test_runtime_low_override_does_not_downgrade_protected_adaptive_rag_functions(db_env) -> None: + db_session, _maker = db_env + + await ModelRoutingService.upsert_override( + db_session, + tenant_id=1, + function_name="adaptive_rag_answer_deep", + tier_alias="low", + updated_by=9, + ) + + low_engine = { + "id": "low-1", + "provider": "glm", + "model": "qwen-plus", + "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1", + "api_key": "low-key", + } + high_engine = { + "id": "high-1", + "provider": "openai", + "model": "GLM-5", + "base_url": "https://api.openai.com/v1", + "api_key": "high-key", + } + + routed = await llm_service._resolve_function_override_engine( + db_session, + tenant_id=1, + function_name="adaptive_rag_answer_deep", + engines=[low_engine, high_engine], + active_engine=low_engine, + ) + + assert routed is None + + @pytest.mark.asyncio async def test_model_routing_list_includes_current_business_entry_points(db_env) -> None: db_session, _maker = db_env