From 331e9109e281964cf3fbc54a5749e52f1bc5ea7e Mon Sep 17 00:00:00 2001 From: Yuuuu <1569752928@qq.com> Date: Wed, 11 Mar 2026 10:07:51 +0800 Subject: [PATCH] =?UTF-8?q?=E6=89=A7=E8=A1=8C=E5=85=A8=E9=A1=B9=E7=9B=AE?= =?UTF-8?q?=E9=85=8D=E7=BD=AE=E9=9A=90=E6=82=A3=E4=B8=8E=E6=A8=A1=E5=9E=8B?= =?UTF-8?q?=E5=91=BD=E5=90=8D=E6=89=AB=E6=8F=8F=E6=B8=85=E5=8D=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CLAUDE.md | 6 +- backend/.openclaw/pmflow-routing-rules.json | 8 +- ...0260311_add_conversation_sessions_table.py | 82 ++++ backend/app/config/ai_prompts.py | 29 ++ backend/app/models/__init__.py | 2 + backend/app/models/conversation_session.py | 33 ++ backend/app/schemas/__init__.py | 3 + backend/app/schemas/conversation_session.py | 24 ++ backend/app/services/ai_pmo_kernel_service.py | 385 +++++++++++++++++- backend/app/services/llm.py | 13 + .../signal_source_orchestration_service.py | 9 +- .../scripts/backfill_orphan_signal_links.py | 99 +++++ backend/tests/test_ai_pmo_kernel_service.py | 204 +++++++++- ...est_signal_source_orchestration_service.py | 5 + frontend/src/components/layout/Layout.tsx | 28 +- frontend/src/components/layout/Sidebar.tsx | 28 +- frontend/src/components/layout/TopBar.tsx | 30 +- .../src/components/sstc/AIVisibilityPanel.tsx | 2 +- .../components/sstc/SSTCCharterStagePanel.tsx | 191 +++++---- .../sstc/SSTCExecutionBoardPanel.tsx | 111 ++--- .../src/components/sstc/SSTCGlobalHeader.tsx | 12 +- .../components/sstc/SSTCReleaseStagePanel.tsx | 2 +- .../components/sstc/SSTCReviewStagePanel.tsx | 2 +- .../components/sstc/SSTCRiskFloatPanel.tsx | 6 +- .../components/sstc/SSTCSignalStagePanel.tsx | 8 +- frontend/src/components/sstc/SSTCStageNav.tsx | 6 +- .../components/sstc/SSTCTaskCreateDialog.tsx | 4 +- .../components/sstc/SSTCTaskStagePanel.tsx | 14 +- .../src/components/sstc/SSTCWorkspace.tsx | 9 +- .../sstc/SignalContentHoverCard.tsx | 2 +- frontend/src/pages/AIConfigPage.tsx | 22 +- frontend/src/pages/AIStewardPage.tsx | 36 +- frontend/src/pages/CharterDetailPage.tsx | 4 +- frontend/src/pages/DashboardPage.tsx | 16 +- frontend/src/pages/FeishuConfigPage.tsx | 16 +- frontend/src/pages/InboxPage.tsx | 10 +- frontend/src/pages/ProjectBoardPage.tsx | 17 +- frontend/src/pages/ProjectsPage.tsx | 8 +- frontend/src/pages/RecipesPage.tsx | 20 +- frontend/src/pages/SkillTasksPage.tsx | 16 +- frontend/src/pages/SkillsPage.tsx | 22 +- frontend/src/pages/UserProfilePage.tsx | 8 +- frontend/src/pages/UserSkillsPage.tsx | 14 +- frontend/src/pages/UsersPage.tsx | 20 +- frontend/src/pages/WorkloadDetailPage.tsx | 26 +- frontend/src/pages/WorkloadsPage.tsx | 26 +- 46 files changed, 1321 insertions(+), 317 deletions(-) create mode 100644 backend/alembic/versions/20260311_add_conversation_sessions_table.py create mode 100644 backend/app/models/conversation_session.py create mode 100644 backend/app/schemas/conversation_session.py create mode 100644 backend/scripts/backfill_orphan_signal_links.py diff --git a/CLAUDE.md b/CLAUDE.md index fcf1019e..806e7f0f 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -42,7 +42,10 @@ backend/ │ └── versions/ │ ├── 20260310_add_feishu_outbox_reply_hash.py # 飞书 outbox 回执哈希幂等字段 │ ├── 20260311_add_pg_trgm_text_indexes.py # pg_trgm 扩展与文本检索索引 -│ └── 20260311_add_kb_chunk_ivfflat_index.py # knowledge_base_chunks 向量 IVFFlat 索引 +│ ├── 20260311_add_kb_chunk_ivfflat_index.py # knowledge_base_chunks 向量 IVFFlat 索引 +│ └── 20260311_add_conversation_sessions_table.py # 飞书多轮会话上下文持久化表 +├── scripts/ # 后端运维/数据修复脚本 +│ └── backfill_orphan_signal_links.py # 回填孤立 signal 的 project_signal_links ├── requirements.txt # Python 依赖 ├── Dockerfile └── tests/ @@ -152,3 +155,4 @@ uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 - `agents.defaults.timeoutSeconds` 在 OpenClaw 网关行为未验证，当前以 `LLM_REQUEST_TIMEOUT` 环境变量为准 - World Model 静态层已接入 Redis 缓存（key: `wm:static:{tenant_id}:{project_id}`，TTL 600s） - 写操作后主动 invalidate，动态层不缓存 +- AI PMO 意图识别已启用 LLM 主路径（`AI_PMO_INTENT_LLM_ENABLED`，默认开启）；当模型超时/不可用/结构化解析失败时自动降级到规则链路，不中断主流程 diff --git a/backend/.openclaw/pmflow-routing-rules.json b/backend/.openclaw/pmflow-routing-rules.json index c509e59f..55b9b700 100644 --- a/backend/.openclaw/pmflow-routing-rules.json +++ b/backend/.openclaw/pmflow-routing-rules.json @@ -9,8 +9,8 @@ "enabled": true, "priority": 300, "tier": "low", - "provider": "openai", - "model": "zai/glm-4-flash", + "provider": "glm", + "model": "GLM-4.7", "match_any": [ "feishu_level_classification", "feishu_signal_extraction", @@ -26,7 +26,7 @@ "priority": 220, "tier": "high", "provider": "openai", - "model": "minimax/MiniMax-M2.5", + "model": "MiniMax-M2.5", "match_any": [ "adaptive_rag_answer", "feishu_project_prefill", @@ -49,7 +49,7 @@ "priority": 200, "tier": "high", "provider": "openai", - "model": "minimax/MiniMax-M2.5", + "model": "MiniMax-M2.5", "match_any": [ "analyze_food_calories", "feishu_new_project_brief", diff --git a/backend/alembic/versions/20260311_add_conversation_sessions_table.py b/backend/alembic/versions/20260311_add_conversation_sessions_table.py new file mode 100644 index 00000000..7fdd829f --- /dev/null +++ b/backend/alembic/versions/20260311_add_conversation_sessions_table.py @@ -0,0 +1,82 @@ +"""add conversation sessions table + +Revision ID: add_conversation_sessions_table +Revises: add_kb_chunk_ivfflat_index +Create Date: 2026-03-11 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "add_conversation_sessions_table" +down_revision = "add_kb_chunk_ivfflat_index" +branch_labels = None +depends_on = None + + +def _table_exists(bind, table_name: str) -> bool: + return sa.inspect(bind).has_table(table_name) + + +def _index_exists(bind, table_name: str, index_name: str) -> bool: + inspector = sa.inspect(bind) + if not inspector.has_table(table_name): + return False + return any(idx.get("name") == index_name for idx in inspector.get_indexes(table_name)) + + +def _create_index_if_missing(bind, table_name: str, index_name: str, columns: list[str], *, unique: bool = False) -> None: + if not _index_exists(bind, table_name, index_name): + op.create_index(index_name, table_name, columns, unique=unique) + + +def _drop_index_if_exists(bind, table_name: str, index_name: str) -> None: + if _index_exists(bind, table_name, index_name): + op.drop_index(index_name, table_name=table_name) + + +def upgrade() -> None: + bind = op.get_bind() + if not _table_exists(bind, "conversation_sessions"): + op.create_table( + "conversation_sessions", + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column("tenant_id", sa.BigInteger(), nullable=False, server_default="1"), + sa.Column("sender", sa.String(length=128), nullable=False), + sa.Column("current_project_id", sa.BigInteger(), sa.ForeignKey("projects.id"), nullable=True), + sa.Column("last_intent", sa.String(length=64), nullable=True), + sa.Column("last_message_text", sa.String(length=200), nullable=True), + sa.Column("last_active_at", sa.TIMESTAMP(), nullable=False, server_default=sa.func.now()), + sa.Column("created_at", sa.TIMESTAMP(), nullable=False, server_default=sa.func.now()), + sa.Column("updated_at", sa.TIMESTAMP(), nullable=False, server_default=sa.func.now(), onupdate=sa.func.now()), + sa.UniqueConstraint("tenant_id", "sender", name="uq_conversation_sessions_tenant_sender"), + ) + + _create_index_if_missing(bind, "conversation_sessions", "ix_conversation_sessions_tenant_id", ["tenant_id"]) + _create_index_if_missing(bind, "conversation_sessions", "ix_conversation_sessions_sender", ["sender"]) + _create_index_if_missing(bind, "conversation_sessions", "ix_conversation_sessions_current_project_id", ["current_project_id"]) + _create_index_if_missing(bind, "conversation_sessions", "ix_conversation_sessions_last_intent", ["last_intent"]) + _create_index_if_missing(bind, "conversation_sessions", "ix_conversation_sessions_last_active_at", ["last_active_at"]) + _create_index_if_missing( + bind, + "conversation_sessions", + "ix_conversation_sessions_tenant_sender_last_active", + ["tenant_id", "sender", "last_active_at"], + ) + + +def downgrade() -> None: + bind = op.get_bind() + if not _table_exists(bind, "conversation_sessions"): + return + + _drop_index_if_exists(bind, "conversation_sessions", "ix_conversation_sessions_tenant_sender_last_active") + _drop_index_if_exists(bind, "conversation_sessions", "ix_conversation_sessions_last_active_at") + _drop_index_if_exists(bind, "conversation_sessions", "ix_conversation_sessions_last_intent") + _drop_index_if_exists(bind, "conversation_sessions", "ix_conversation_sessions_current_project_id") + _drop_index_if_exists(bind, "conversation_sessions", "ix_conversation_sessions_sender") + _drop_index_if_exists(bind, "conversation_sessions", "ix_conversation_sessions_tenant_id") + op.drop_table("conversation_sessions") diff --git a/backend/app/config/ai_prompts.py b/backend/app/config/ai_prompts.py index a2847e8b..287f8270 100644 --- a/backend/app/config/ai_prompts.py +++ b/backend/app/config/ai_prompts.py @@ -500,6 +500,35 @@ PROMPT_AI_WORKER_SIGNAL_SOURCE_EXECUTION = """ """ +PROMPT_AI_PMO_INTENT_CLASSIFICATION = """ +首字符必须是 {{，末字符必须是 }}。 +你是 PMFlow 的 AI PMO 意图识别模块。你只做分类，不做执行。 + +输入： +- message_text: {message_text} +- session_last_intent: {session_last_intent} +- session_project_id: {session_project_id} +- world_model_project_id: {world_model_project_id} +- candidate_intents_json: {candidate_intents_json} + +输出 JSON： +{{ + "intent": "signal_source.onboard|signal.create|query.read_only|project.create|followup.confirm|followup.continue|ignore", + "confidence": 0.0, + "project_id": 0, + "reasoning": "一句话简述判定原因" +}} + +规则： +1. intent 必须从 candidate_intents_json 内选择，不能自造。 +2. 若消息是短确认语气，优先结合 session_last_intent 判定。 +3. project_id 无法确定时返回 null，不要猜测无依据项目。 +4. 噪音/寒暄/无关消息返回 ignore。 +5. 禁止输出 thinking、analysis、解释句、Markdown、代码块。 +6. 只输出 JSON，不要额外文本。 +""" + + # ============================================================================ # Full Adaptive RAG # ============================================================================ diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 50e5d76b..dc876a0b 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -61,6 +61,7 @@ from .decision_record import DecisionRecord from .plan_bundle import PlanBundle from .reply_receipt import ReplyReceipt from .portfolio_snapshot import PortfolioSnapshot +from .conversation_session import ConversationSession __all__ = [ "Base", @@ -126,4 +127,5 @@ __all__ = [ "PlanBundle", "ReplyReceipt", "PortfolioSnapshot", + "ConversationSession", ] diff --git a/backend/app/models/conversation_session.py b/backend/app/models/conversation_session.py new file mode 100644 index 00000000..824ec09b --- /dev/null +++ b/backend/app/models/conversation_session.py @@ -0,0 +1,33 @@ +"""飞书/消息对话会话状态。""" + +from __future__ import annotations + +from datetime import datetime +from typing import Optional + +from sqlalchemy import BigInteger, ForeignKey, Index, Integer, TIMESTAMP, UniqueConstraint, VARCHAR, func +from sqlalchemy.orm import Mapped, mapped_column + +from .base import BaseNoTenant + + +class ConversationSession(BaseNoTenant): + __tablename__ = "conversation_sessions" + __table_args__ = ( + UniqueConstraint("tenant_id", "sender", name="uq_conversation_sessions_tenant_sender"), + Index("ix_conversation_sessions_tenant_sender_last_active", "tenant_id", "sender", "last_active_at"), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + tenant_id: Mapped[int] = mapped_column(BigInteger, nullable=False, default=1, index=True) + sender: Mapped[str] = mapped_column(VARCHAR(128), nullable=False, index=True) + current_project_id: Mapped[Optional[int]] = mapped_column( + BigInteger, + ForeignKey("projects.id"), + nullable=True, + index=True, + ) + last_intent: Mapped[Optional[str]] = mapped_column(VARCHAR(64), nullable=True, index=True) + last_message_text: Mapped[Optional[str]] = mapped_column(VARCHAR(200), nullable=True) + last_active_at: Mapped[datetime] = mapped_column(TIMESTAMP, nullable=False, default=func.now(), index=True) + created_at: Mapped[datetime] = mapped_column(TIMESTAMP, nullable=False, default=func.now(), server_default=func.now()) diff --git a/backend/app/schemas/__init__.py b/backend/app/schemas/__init__.py index ec0b0a5e..796a64b3 100644 --- a/backend/app/schemas/__init__.py +++ b/backend/app/schemas/__init__.py @@ -49,6 +49,7 @@ from .recipe import ( RecipeImageUploadRequest, RecipeAnalyzeCaloriesResponse, ) +from .conversation_session import ConversationSessionBase, ConversationSessionRead from . import ai as ai_schemas __all__ = [ @@ -99,4 +100,6 @@ __all__ = [ "RecipeResponse", "RecipeImageUploadRequest", "RecipeAnalyzeCaloriesResponse", + "ConversationSessionBase", + "ConversationSessionRead", ] diff --git a/backend/app/schemas/conversation_session.py b/backend/app/schemas/conversation_session.py new file mode 100644 index 00000000..0339b5a6 --- /dev/null +++ b/backend/app/schemas/conversation_session.py @@ -0,0 +1,24 @@ +"""会话状态 Schema。""" + +from __future__ import annotations + +from datetime import datetime +from typing import Optional + +from pydantic import BaseModel, ConfigDict, Field + + +class ConversationSessionBase(BaseModel): + tenant_id: int + sender: str = Field(min_length=1, max_length=128) + current_project_id: Optional[int] = None + last_intent: Optional[str] = Field(default=None, max_length=64) + last_message_text: Optional[str] = Field(default=None, max_length=200) + last_active_at: datetime + created_at: datetime + + +class ConversationSessionRead(ConversationSessionBase): + id: int + + model_config = ConfigDict(from_attributes=True) diff --git a/backend/app/services/ai_pmo_kernel_service.py b/backend/app/services/ai_pmo_kernel_service.py index 18eb24ff..83425729 100644 --- a/backend/app/services/ai_pmo_kernel_service.py +++ b/backend/app/services/ai_pmo_kernel_service.py @@ -5,6 +5,7 @@ from __future__ import annotations import asyncio import hashlib import json +import os import re import time from dataclasses import dataclass @@ -15,11 +16,13 @@ from sqlalchemy import and_, func, or_, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession +from ..config.ai_prompts import PROMPT_AI_PMO_INTENT_CLASSIFICATION from ..core.logging import get_logger from ..database import AsyncSessionLocal from ..models import ( AIProcessingLog, Budget, + ConversationSession, DecisionRecord, IngestEvent, Notification, @@ -61,6 +64,7 @@ from .context_cache import ContextCacheService from .notification_service import NotificationService from .openclaw_commander_service import OpenClawCommanderService from .ai_pmo_domain_workflow import SignalSourceDomainWorkflow +from .llm import llm_service from .signal_source_orchestration_service import SignalSourceOrchestrationService logger = get_logger(__name__) @@ -143,6 +147,37 @@ POLICY_GATE_RULES: tuple[tuple[str, tuple[str, ...]], ...] = ( ("charter_rewrite", ("改写charter", "success criteria", "成功标准改", "范围重大变更")), ("critical_path_change", ("关键路径", "里程碑改动", "影响多个项目", "跨项目依赖调整")), ) +BOOL_TRUE_VALUES = {"1", "true", "yes", "y", "on", "是"} +SESSION_PROJECT_INHERIT_WINDOW_SECONDS = 30 * 60 +SESSION_SHORT_CONFIRM_WINDOW_SECONDS = 5 * 60 +SHORT_CONFIRM_MAX_MESSAGE_LENGTH = 10 +SHORT_CONFIRM_HINTS = ( + "是", + "好", + "对", + "确认", + "新增", + "加", + "嗯", + "yes", + "ok", + "好的", + "是的", + "行", + "可以", + "没问题", +) +INTENT_LLM_TIMEOUT_SECONDS = 8.0 +INTENT_LLM_FUNCTION_NAME = "classify_intent_ai_pmo_kernel" +INTENT_CANDIDATES = ( + INTENT_SIGNAL_SOURCE_ONBOARD, + INTENT_SIGNAL_CREATE, + INTENT_QUERY, + INTENT_PROJECT_CREATE, + INTENT_FOLLOWUP_CONFIRM, + INTENT_FOLLOWUP_CONTINUE, + INTENT_IGNORE, +) WORLD_MODEL_STATIC_CACHE_TTL_SECONDS = 10 * 60 WORLD_MODEL_STATIC_CACHE_KEY_PREFIX = "wm:static" @@ -152,6 +187,13 @@ def _utc_now_naive() -> datetime: return datetime.now(timezone.utc).replace(tzinfo=None) +def _env_bool(name: str, default: bool = False) -> bool: + raw = os.getenv(name) + if raw is None: + return default + return str(raw).strip().lower() in BOOL_TRUE_VALUES + + def _safe_int(value: Any) -> Optional[int]: try: if value is None or value == "": @@ -1004,6 +1046,7 @@ class AIPMOKernelService: source_message_id=synthetic_message_id, owner_user_id=_safe_int(getattr(operator, "id", None)), ) + session = await self._get_or_create_session(tenant_id=self.tenant_id, sender=sender) try: world_model_query_tables: set[str] = set() @@ -1011,6 +1054,7 @@ class AIPMOKernelService: message_text=clean_text, project_hint_id=project_hint_id, query_tables=world_model_query_tables, + session=session, ) world_model = await self.build_world_model( tenant_id=self.tenant_id, @@ -1021,8 +1065,16 @@ class AIPMOKernelService: ) self._schedule_portfolio_snapshot_persist(world_model=world_model) work_item.state = STATE_MODELED + intent, intent_context_source = await self._resolve_intent( + message_text=clean_text, + session=session, + world_model=world_model, + ) + if intent_context_source: + world_model["context_source"] = intent_context_source work_item.context_snapshot = world_model - intent = self.intent_resolver.resolve( + resolved_project_for_session = _safe_int(intent.project_id) or _safe_int(world_model.get("resolved_project_id")) + fallback_intent = self.intent_resolver.resolve( text=clean_text, project_id=_safe_int(world_model.get("resolved_project_id")), ) @@ -1041,6 +1093,13 @@ class AIPMOKernelService: processing_status="success", duration_ms=int((time.perf_counter() - started) * 1000), ) + await self._update_session( + tenant_id=self.tenant_id, + sender=sender, + current_project_id=resolved_project_for_session, + last_intent=intent.name, + last_message_text=clean_text[:200], + ) await self.db.commit() return { "handled": True, @@ -1138,6 +1197,13 @@ class AIPMOKernelService: processing_status="success", duration_ms=int((time.perf_counter() - started) * 1000), ) + await self._update_session( + tenant_id=self.tenant_id, + sender=sender, + current_project_id=_safe_int(work_item.project_id), + last_intent=intent.name, + last_message_text=clean_text[:200], + ) await self.db.commit() return self._build_capture_result( work_item=work_item, @@ -1235,6 +1301,15 @@ class AIPMOKernelService: processing_status="success" if work_item.state in {STATE_SUCCEEDED, STATE_WAITING_MANUAL} else "failed", duration_ms=int((time.perf_counter() - started) * 1000), ) + final_project_id = _safe_int(work_item.project_id) or resolved_project_for_session + final_intent_name = str(work_item.intent or fallback_intent.name or "").strip() or intent.name + await self._update_session( + tenant_id=self.tenant_id, + sender=sender, + current_project_id=final_project_id, + last_intent=final_intent_name, + last_message_text=clean_text[:200], + ) await self.db.commit() return self._build_capture_result( work_item=work_item, @@ -1558,12 +1633,14 @@ class AIPMOKernelService: message_text: str, sender: str, project_hint_id: Optional[int], + session: Optional[ConversationSession] = None, ) -> WorldModel: query_tables: set[str] = set() project = await self._resolve_project( message_text=message_text, project_hint_id=project_hint_id, query_tables=query_tables, + session=session, ) return await self.build_world_model( tenant_id=self.tenant_id, @@ -1584,6 +1661,7 @@ class AIPMOKernelService: message_text: str, project_hint_id: Optional[int], query_tables: Optional[set[str]] = None, + session: Optional[ConversationSession] = None, ) -> Optional[Project]: if project_hint_id is not None: self._record_world_model_query_table(query_tables, "projects") @@ -1625,8 +1703,313 @@ class AIPMOKernelService: normalized_candidate = re.sub(r"\s+", "", str(candidate or "")).lower() if normalized_candidate and normalized_candidate in normalized_message: return project + inherited_project_id = self._extract_inherited_project_id(session=session) + if inherited_project_id is not None: + self._record_world_model_query_table(query_tables, "projects") + return ( + await self.db.execute( + select(Project).where( + and_( + Project.tenant_id == self.tenant_id, + Project.id == int(inherited_project_id), + ) + ).limit(1) + ) + ).scalar_one_or_none() return None + @staticmethod + def _within_recent_window(last_active_at: Optional[datetime], *, window_seconds: int) -> bool: + if not isinstance(last_active_at, datetime): + return False + now = _utc_now_naive() + delta = now - last_active_at + return 0 <= delta.total_seconds() <= float(window_seconds) + + def _extract_inherited_project_id(self, *, session: Optional[ConversationSession]) -> Optional[int]: + if session is None: + return None + session_project_id = _safe_int(getattr(session, "current_project_id", None)) + if session_project_id is None or int(session_project_id) <= 0: + return None + if not self._within_recent_window( + getattr(session, "last_active_at", None), + window_seconds=SESSION_PROJECT_INHERIT_WINDOW_SECONDS, + ): + return None + return session_project_id + + async def _get_or_create_session(self, *, tenant_id: int, sender: str) -> ConversationSession: + sender_key = str(sender or "").strip()[:128] or "unknown" + existing = ( + await self.db.execute( + select(ConversationSession).where( + and_( + ConversationSession.tenant_id == int(tenant_id), + ConversationSession.sender == sender_key, + ) + ).limit(1) + ) + ).scalar_one_or_none() + if existing is not None: + return existing + + created = ConversationSession( + tenant_id=int(tenant_id), + sender=sender_key, + current_project_id=None, + last_intent=None, + last_message_text=None, + last_active_at=_utc_now_naive(), + ) + nested_tx = await self.db.begin_nested() + try: + self.db.add(created) + await self.db.flush() + await nested_tx.commit() + return created + except IntegrityError: + await nested_tx.rollback() + existing = ( + await self.db.execute( + select(ConversationSession).where( + and_( + ConversationSession.tenant_id == int(tenant_id), + ConversationSession.sender == sender_key, + ) + ).limit(1) + ) + ).scalar_one_or_none() + if existing is not None: + return existing + raise + + async def _update_session( + self, + *, + tenant_id: int, + sender: str, + current_project_id: Optional[int], + last_intent: Optional[str], + last_message_text: Optional[str], + ) -> None: + row = await self._get_or_create_session(tenant_id=tenant_id, sender=sender) + if current_project_id is not None: + row.current_project_id = int(current_project_id) + row.last_intent = (str(last_intent or "").strip()[:64] or None) + trimmed_message = str(last_message_text or "").strip() + row.last_message_text = trimmed_message[:200] if trimmed_message else None + row.last_active_at = _utc_now_naive() + await self.db.flush() + + @staticmethod + def _intent_llm_enabled() -> bool: + return _env_bool("AI_PMO_INTENT_LLM_ENABLED", True) + + @classmethod + def _is_short_confirmation_message( + cls, + *, + message_text: str, + session: Optional[ConversationSession], + ) -> bool: + if session is None: + return False + if not cls._within_recent_window( + getattr(session, "last_active_at", None), + window_seconds=SESSION_SHORT_CONFIRM_WINDOW_SECONDS, + ): + return False + last_intent = str(getattr(session, "last_intent", "") or "").strip() + if not last_intent: + return False + normalized = re.sub(r"\s+", "", str(message_text or "")).lower() + if not normalized: + return False + if len(normalized) > SHORT_CONFIRM_MAX_MESSAGE_LENGTH: + return False + return any(token in normalized for token in SHORT_CONFIRM_HINTS) + + @staticmethod + def _build_canonical_intent( + *, + intent_name: str, + message_text: str, + project_id: Optional[int], + confidence: float, + metadata: Optional[dict[str, Any]] = None, + ) -> CanonicalIntent: + pretty = re.sub(r"\s+", " ", str(message_text or "")).strip() + extras = dict(metadata or {}) + safe_confidence = max(0.0, min(1.0, float(confidence or 0.0))) + if intent_name == INTENT_SIGNAL_SOURCE_ONBOARD: + extras.setdefault("source_request", True) + return CanonicalIntent( + INTENT_SIGNAL_SOURCE_ONBOARD, + safe_confidence, + project_id, + IntentResolver._build_signal_title(pretty, default="新增信号源接入请求"), + pretty[:300], + extras, + ) + if intent_name == INTENT_SIGNAL_CREATE: + return CanonicalIntent( + INTENT_SIGNAL_CREATE, + safe_confidence, + project_id, + IntentResolver._build_signal_title(pretty, default="新信号"), + pretty[:300], + extras, + ) + if intent_name == INTENT_PROJECT_CREATE: + return CanonicalIntent( + INTENT_PROJECT_CREATE, + safe_confidence, + project_id, + IntentResolver._build_signal_title(pretty, default="新项目创建请求"), + pretty[:300], + extras, + ) + if intent_name == INTENT_QUERY: + return CanonicalIntent(INTENT_QUERY, safe_confidence, project_id, "", pretty[:300], extras) + if intent_name == INTENT_FOLLOWUP_CONFIRM: + return CanonicalIntent(INTENT_FOLLOWUP_CONFIRM, safe_confidence, project_id, "", "", extras) + if intent_name == INTENT_FOLLOWUP_CONTINUE: + return CanonicalIntent(INTENT_FOLLOWUP_CONTINUE, safe_confidence, project_id, "", "", extras) + return CanonicalIntent(INTENT_IGNORE, safe_confidence, project_id, "", "", extras or {"reason": "unsupported_intent"}) + + def _build_short_confirm_intent( + self, + *, + message_text: str, + session: ConversationSession, + world_model: dict[str, Any], + ) -> CanonicalIntent: + inherited_name = str(getattr(session, "last_intent", "") or "").strip() + if inherited_name not in INTENT_CANDIDATES: + inherited_name = INTENT_QUERY + inherited_project_id = _safe_int(getattr(session, "current_project_id", None)) or _safe_int(world_model.get("resolved_project_id")) + if inherited_project_id is not None and int(inherited_project_id) <= 0: + inherited_project_id = None + inherited_text = str(getattr(session, "last_message_text", "") or "").strip() or message_text + return self._build_canonical_intent( + intent_name=inherited_name, + message_text=inherited_text, + project_id=inherited_project_id, + confidence=0.86, + metadata={ + "context_source": "session_inherited", + "inherited_from_intent": str(getattr(session, "last_intent", "") or "").strip(), + }, + ) + + async def _resolve_intent_with_llm( + self, + *, + message_text: str, + session: Optional[ConversationSession], + world_model: dict[str, Any], + ) -> Optional[CanonicalIntent]: + prompt = PROMPT_AI_PMO_INTENT_CLASSIFICATION.format( + message_text=str(message_text or "")[:1200], + session_last_intent=str(getattr(session, "last_intent", "") or "").strip() if session is not None else "", + session_project_id=_safe_int(getattr(session, "current_project_id", None)) if session is not None else None, + world_model_project_id=_safe_int(world_model.get("resolved_project_id")), + candidate_intents_json=json.dumps(list(INTENT_CANDIDATES), ensure_ascii=False), + ).strip() + try: + extraction = await asyncio.wait_for( + llm_service.generate_structured_object( + prompt, + db=self.db, + tenant_id=self.tenant_id, + function_name=INTENT_LLM_FUNCTION_NAME, + use_prompt_profile=False, + temperature=0.0, + max_tokens=260, + response_mode="fast", + ), + timeout=INTENT_LLM_TIMEOUT_SECONDS, + ) + except TimeoutError: + logger.warning( + "ai_pmo_intent_llm_timeout", + tenant_id=self.tenant_id, + timeout_seconds=INTENT_LLM_TIMEOUT_SECONDS, + ) + return None + except Exception as exc: + logger.warning( + "ai_pmo_intent_llm_failed", + tenant_id=self.tenant_id, + error=str(exc), + ) + return None + + payload = extraction.payload if extraction and isinstance(extraction.payload, dict) else None + if payload is None: + logger.warning( + "ai_pmo_intent_llm_empty_payload", + tenant_id=self.tenant_id, + error_code=str(getattr(extraction, "error_code", "") or ""), + provider=str(getattr(extraction, "provider", "") or ""), + model=str(getattr(extraction, "model", "") or ""), + ) + return None + + intent_name = str(payload.get("intent") or "").strip() + if intent_name not in INTENT_CANDIDATES: + return None + confidence = float(payload.get("confidence") or 0.0) + project_id = _safe_int(payload.get("project_id")) + if project_id is not None and int(project_id) <= 0: + project_id = None + if project_id is None: + project_id = _safe_int(world_model.get("resolved_project_id")) + reasoning = str(payload.get("reasoning") or "").strip() + return self._build_canonical_intent( + intent_name=intent_name, + message_text=message_text, + project_id=project_id, + confidence=confidence, + metadata={ + "context_source": "llm", + "reasoning": reasoning[:300], + "llm_function": INTENT_LLM_FUNCTION_NAME, + }, + ) + + async def _resolve_intent( + self, + *, + message_text: str, + session: Optional[ConversationSession], + world_model: dict[str, Any], + ) -> tuple[CanonicalIntent, str]: + project_id = _safe_int(world_model.get("resolved_project_id")) + if self._is_short_confirmation_message(message_text=message_text, session=session): + if session is None: + return self.intent_resolver.resolve(text=message_text, project_id=project_id), "rule_fallback" + return ( + self._build_short_confirm_intent( + message_text=message_text, + session=session, + world_model=world_model, + ), + "session_inherited", + ) + + if self._intent_llm_enabled(): + llm_intent = await self._resolve_intent_with_llm( + message_text=message_text, + session=session, + world_model=world_model, + ) + if llm_intent is not None: + return llm_intent, "llm" + + return self.intent_resolver.resolve(text=message_text, project_id=project_id), "rule_fallback" + async def _build_portfolio_summary( self, *, diff --git a/backend/app/services/llm.py b/backend/app/services/llm.py index 5c384601..53eda2b4 100644 --- a/backend/app/services/llm.py +++ b/backend/app/services/llm.py @@ -215,6 +215,17 @@ STRUCTURED_OBJECT_SCHEMAS: dict[str, dict[str, Any]] = { "next_actions": "list", }, }, + "classify_intent_ai_pmo_kernel": { + "required_keys": ("intent", "confidence", "reasoning"), + "type_checks": { + "intent": "str", + "confidence": "number", + "reasoning": "str", + }, + "optional_type_checks": { + "project_id": ("number_or_none",), + }, + }, } @@ -1934,6 +1945,8 @@ class LLMService: return isinstance(value, list) if expected == "number": return isinstance(value, (int, float)) and not isinstance(value, bool) + if expected == "number_or_none": + return value is None or (isinstance(value, (int, float)) and not isinstance(value, bool)) if expected == "dict_or_none": return value is None or isinstance(value, dict) return True diff --git a/backend/app/services/signal_source_orchestration_service.py b/backend/app/services/signal_source_orchestration_service.py index 5b926d2d..9c8a71c7 100644 --- a/backend/app/services/signal_source_orchestration_service.py +++ b/backend/app/services/signal_source_orchestration_service.py @@ -33,7 +33,7 @@ from .prompt_profile_service import prompt_profile_service from .llm import llm_service SOURCE_ONBOARDING_ACTION_RE = re.compile( - r"(新增|增加|添加|接入|配置|注册|建立).{0,8}(信号源|来源|监控源|channel|source|webhook|飞书|github|url|rss)", + r"(新增|增加|添加|接入|配置|注册|建立|加个|加一个|加).{0,8}(信号源|来源|监控源|channel|source|webhook|飞书|github|url|rss)", re.IGNORECASE, ) SOURCE_KIND_HINTS: tuple[tuple[str, tuple[str, ...]], ...] = ( @@ -66,9 +66,12 @@ class SignalSourceOrchestrationService: @classmethod def looks_like_signal_source_request(cls, text: str) -> bool: normalized = re.sub(r"\s+", "", str(text or "")) - if len(normalized) < 6: + if len(normalized) < 3: return False - if "信号源" in normalized and any(token in normalized for token in ("新增", "增加", "添加", "接入", "配置", "注册", "建立")): + if "信号源" in normalized and any( + token in normalized + for token in ("新增", "增加", "添加", "接入", "配置", "注册", "建立", "加个", "加一个", "加") + ): return True return bool(SOURCE_ONBOARDING_ACTION_RE.search(normalized)) diff --git a/backend/scripts/backfill_orphan_signal_links.py b/backend/scripts/backfill_orphan_signal_links.py new file mode 100644 index 00000000..f8dc4386 --- /dev/null +++ b/backend/scripts/backfill_orphan_signal_links.py @@ -0,0 +1,99 @@ +#!/usr/bin/env python3 +"""回填 signals.matched_project_id 对应的 project_signal_links。""" + +from __future__ import annotations + +import argparse +import asyncio + +from sqlalchemy import and_, select + +from app.database import AsyncSessionLocal +from app.models import ProjectSignalLink, Signal + + +async def _query_orphans(*, tenant_id: int, signal_id: int | None) -> list[tuple[int, int, int, str]]: + async with AsyncSessionLocal() as db: + stmt = ( + select( + Signal.id, + Signal.tenant_id, + Signal.matched_project_id, + Signal.title, + ) + .outerjoin( + ProjectSignalLink, + and_( + ProjectSignalLink.tenant_id == Signal.tenant_id, + ProjectSignalLink.signal_id == Signal.id, + ), + ) + .where( + and_( + Signal.tenant_id == int(tenant_id), + Signal.matched_project_id.is_not(None), + ProjectSignalLink.id.is_(None), + ) + ) + .order_by(Signal.id.asc()) + ) + if signal_id is not None: + stmt = stmt.where(Signal.id == int(signal_id)) + rows = (await db.execute(stmt)).all() + return [ + ( + int(row[0]), + int(row[1]), + int(row[2]), + str(row[3] or ""), + ) + for row in rows + if row[0] is not None and row[1] is not None and row[2] is not None + ] + + +async def _apply_backfill(*, tenant_id: int, signal_id: int | None) -> int: + rows = await _query_orphans(tenant_id=tenant_id, signal_id=signal_id) + if not rows: + return 0 + + async with AsyncSessionLocal() as db: + for sid, tid, pid, _ in rows: + db.add( + ProjectSignalLink( + tenant_id=tid, + signal_id=sid, + project_id=pid, + link_type="matched_project", + relevance_score=1.0, + ) + ) + await db.commit() + return len(rows) + + +async def _main() -> None: + parser = argparse.ArgumentParser(description="Backfill orphan project_signal_links from signals.matched_project_id") + parser.add_argument("--tenant-id", type=int, default=1, help="目标租户 ID，默认 1") + parser.add_argument("--signal-id", type=int, default=None, help="仅修复指定 signal_id") + parser.add_argument("--apply", action="store_true", help="执行写入；不传时仅预览") + args = parser.parse_args() + + rows = await _query_orphans(tenant_id=args.tenant_id, signal_id=args.signal_id) + print(f"[preview] tenant_id={args.tenant_id} orphan_count={len(rows)}") + for sid, _, pid, title in rows[:20]: + print(f"- signal_id={sid} -> project_id={pid} title={title[:80]}") + if len(rows) > 20: + print(f"... ({len(rows) - 20} more)") + + if not args.apply: + print("[dry-run] 未执行写入。加 --apply 以回填。") + return + + inserted = await _apply_backfill(tenant_id=args.tenant_id, signal_id=args.signal_id) + remaining = await _query_orphans(tenant_id=args.tenant_id, signal_id=args.signal_id) + print(f"[apply] inserted={inserted} remaining_orphans={len(remaining)}") + + +if __name__ == "__main__": + asyncio.run(_main()) diff --git a/backend/tests/test_ai_pmo_kernel_service.py b/backend/tests/test_ai_pmo_kernel_service.py index 60363902..9cb87d5c 100644 --- a/backend/tests/test_ai_pmo_kernel_service.py +++ b/backend/tests/test_ai_pmo_kernel_service.py @@ -7,7 +7,21 @@ from unittest.mock import AsyncMock, patch from sqlalchemy import and_, select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine -from app.models import Base, Budget, PortfolioSnapshot, Project, ProjectDomain, Task, TruthNode, User, UserProject, UserSkillBinding, UserWorkload, WorkItem +from app.models import ( + Base, + Budget, + ConversationSession, + PortfolioSnapshot, + Project, + ProjectDomain, + Task, + TruthNode, + User, + UserProject, + UserSkillBinding, + UserWorkload, + WorkItem, +) from app.services.ai_pmo_cadence_service import AIPMOCadenceService from app.services.ai_pmo_kernel_service import ( AIPMOKernelService, @@ -15,11 +29,14 @@ from app.services.ai_pmo_kernel_service import ( IntentResolver, ) from app.services.ai_pmo_domain_workflow import SignalSourceDomainWorkflow +from app.services.llm import StructuredExtractionResult from app.services.signal_source_orchestration_service import SignalSourceOrchestrationService class AIPMOKernelServiceTest(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self) -> None: + self._old_intent_llm_enabled = os.environ.get("AI_PMO_INTENT_LLM_ENABLED") + os.environ["AI_PMO_INTENT_LLM_ENABLED"] = "0" tmp = NamedTemporaryFile(prefix="pmflow-ai-pmo-test-", suffix=".db", delete=False) tmp.close() self.db_path = tmp.name @@ -149,6 +166,10 @@ class AIPMOKernelServiceTest(unittest.IsolatedAsyncioTestCase): async def asyncTearDown(self) -> None: await self.engine.dispose() + if self._old_intent_llm_enabled is None: + os.environ.pop("AI_PMO_INTENT_LLM_ENABLED", None) + else: + os.environ["AI_PMO_INTENT_LLM_ENABLED"] = self._old_intent_llm_enabled if getattr(self, "db_path", None): try: os.remove(self.db_path) @@ -254,6 +275,187 @@ class AIPMOKernelServiceTest(unittest.IsolatedAsyncioTestCase): self.assertEqual(second["status"], "ignored") self.assertFalse(second["direct_reply_required"]) + async def test_short_confirmation_inherits_last_intent_and_project(self) -> None: + async with self.session_factory() as db: + service = AIPMOKernelService(db, tenant_id=1) + await service.process_ingress( + source_system="feishu", + source_message_id="ctx-source-1", + message_text="给项目ID: 1 新增一个 GitHub 信号源，监控 openai/codex。", + sender="老板", + channel_ref="oc_ctx", + source_sender_open_id="ou_boss", + project_hint_id=1, + ) + second = await service.process_ingress( + source_system="feishu", + source_message_id="ctx-source-2", + message_text="是的新增吧", + sender="老板", + channel_ref="oc_ctx", + source_sender_open_id="ou_boss", + ) + second_item = ( + await db.execute( + select(WorkItem).where( + and_( + WorkItem.tenant_id == 1, + WorkItem.source_message_id == "ctx-source-2", + ) + ) + ) + ).scalar_one() + session_row = ( + await db.execute( + select(ConversationSession).where( + and_( + ConversationSession.tenant_id == 1, + ConversationSession.sender == "老板", + ) + ) + ) + ).scalar_one() + + self.assertEqual(second["status"], "captured") + self.assertEqual(second_item.intent, INTENT_SIGNAL_SOURCE_ONBOARD) + self.assertEqual(int(second_item.project_id or 0), 1) + self.assertEqual((second_item.context_snapshot or {}).get("context_source"), "session_inherited") + self.assertEqual(session_row.last_intent, INTENT_SIGNAL_SOURCE_ONBOARD) + self.assertEqual(int(session_row.current_project_id or 0), 1) + + async def test_project_id_inherits_from_recent_session_when_message_has_no_project_text(self) -> None: + async with self.session_factory() as db: + service = AIPMOKernelService(db, tenant_id=1) + await service.process_ingress( + source_system="pmflow_manual", + source_message_id="project-inherit-1", + message_text="给项目ID: 1 新增一条信号：跟踪江苏医保支付改革。", + sender="老板", + channel_ref="pmflow", + project_hint_id=1, + ) + second = await service.process_ingress( + source_system="pmflow_manual", + source_message_id="project-inherit-2", + message_text="新增一条信号：跟踪医药招采节奏变化。", + sender="老板", + channel_ref="pmflow", + ) + second_item = ( + await db.execute( + select(WorkItem).where( + and_( + WorkItem.tenant_id == 1, + WorkItem.source_message_id == "project-inherit-2", + ) + ) + ) + ).scalar_one() + + self.assertTrue(second["handled"]) + self.assertEqual(second_item.intent, "signal.create") + self.assertEqual(int(second_item.project_id or 0), 1) + self.assertEqual(second["linked_project_ids"], [1]) + + async def test_intent_llm_primary_path_marks_context_source(self) -> None: + prev = os.environ.get("AI_PMO_INTENT_LLM_ENABLED") + os.environ["AI_PMO_INTENT_LLM_ENABLED"] = "1" + try: + async with self.session_factory() as db: + service = AIPMOKernelService(db, tenant_id=1) + with patch( + "app.services.ai_pmo_kernel_service.llm_service.generate_structured_object", + AsyncMock( + return_value=StructuredExtractionResult( + payload={ + "intent": "signal_source.onboard", + "confidence": 0.95, + "project_id": 1, + "reasoning": "用户要新增信号源", + }, + source_field="content", + provider="openai", + model="zai/glm-4.7", + raw_text='{"intent":"signal_source.onboard"}', + validation="ok", + ) + ), + ): + result = await service.process_ingress( + source_system="pmflow_manual", + source_message_id="llm-intent-1", + message_text="这个项目每天看竞品动态，帮我接入一下。", + sender="老板", + channel_ref="pmflow", + project_hint_id=1, + ) + item = ( + await db.execute( + select(WorkItem).where( + and_( + WorkItem.tenant_id == 1, + WorkItem.source_message_id == "llm-intent-1", + ) + ) + ) + ).scalar_one() + + self.assertTrue(result["handled"]) + self.assertEqual(item.intent, INTENT_SIGNAL_SOURCE_ONBOARD) + self.assertEqual((item.context_snapshot or {}).get("context_source"), "llm") + finally: + if prev is None: + os.environ.pop("AI_PMO_INTENT_LLM_ENABLED", None) + else: + os.environ["AI_PMO_INTENT_LLM_ENABLED"] = prev + + async def test_intent_llm_failure_falls_back_to_rule_resolver(self) -> None: + prev = os.environ.get("AI_PMO_INTENT_LLM_ENABLED") + os.environ["AI_PMO_INTENT_LLM_ENABLED"] = "1" + try: + async with self.session_factory() as db: + service = AIPMOKernelService(db, tenant_id=1) + with patch( + "app.services.ai_pmo_kernel_service.llm_service.generate_structured_object", + AsyncMock( + return_value=StructuredExtractionResult( + payload=None, + source_field="", + provider="openai", + model="zai/glm-4.7", + error_code="model_not_allowed", + raw_text="403 ModelNotAllowed", + validation="invalid", + ) + ), + ): + await service.process_ingress( + source_system="pmflow_manual", + source_message_id="llm-intent-fallback-1", + message_text="给项目ID: 1 新增一个 GitHub 信号源，监控 openai/codex。", + sender="老板", + channel_ref="pmflow", + project_hint_id=1, + ) + item = ( + await db.execute( + select(WorkItem).where( + and_( + WorkItem.tenant_id == 1, + WorkItem.source_message_id == "llm-intent-fallback-1", + ) + ) + ) + ).scalar_one() + + self.assertEqual(item.intent, INTENT_SIGNAL_SOURCE_ONBOARD) + self.assertEqual((item.context_snapshot or {}).get("context_source"), "rule_fallback") + finally: + if prev is None: + os.environ.pop("AI_PMO_INTENT_LLM_ENABLED", None) + else: + os.environ["AI_PMO_INTENT_LLM_ENABLED"] = prev + async def test_policy_gate_blocks_archive_like_request(self) -> None: async with self.session_factory() as db: service = AIPMOKernelService(db, tenant_id=1) diff --git a/backend/tests/test_signal_source_orchestration_service.py b/backend/tests/test_signal_source_orchestration_service.py index 266118f3..c679affc 100644 --- a/backend/tests/test_signal_source_orchestration_service.py +++ b/backend/tests/test_signal_source_orchestration_service.py @@ -118,6 +118,11 @@ class SignalSourceOrchestrationServiceTest(unittest.TestCase): "请新增一个 GitHub 信号源，监控核心仓库 issue。" ) ) + self.assertTrue( + SignalSourceOrchestrationService.looks_like_signal_source_request( + "给这个项目加个信号源。" + ) + ) self.assertFalse( SignalSourceOrchestrationService.looks_like_signal_source_request( "把任务 12 分给张三。" diff --git a/frontend/src/components/layout/Layout.tsx b/frontend/src/components/layout/Layout.tsx index 0f0b231f..a6518728 100644 --- a/frontend/src/components/layout/Layout.tsx +++ b/frontend/src/components/layout/Layout.tsx @@ -1,14 +1,30 @@ -import { Outlet } from "react-router-dom"; +import { useEffect, useState } from "react"; +import { Outlet, useLocation } from "react-router-dom"; import { Sidebar } from "./Sidebar"; import { TopBar } from "./TopBar"; export default function Layout() { + const location = useLocation(); + const [mobileSidebarOpen, setMobileSidebarOpen] = useState(false); + + useEffect(() => { + setMobileSidebarOpen(false); + }, [location.pathname, location.search]); + return ( -
- -
- -
+
+ setMobileSidebarOpen(false)} /> + {mobileSidebarOpen && ( +