diff --git a/CLAUDE.md b/CLAUDE.md index 3de4f19c..d0415c61 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -39,6 +39,8 @@ backend/ │ ├── config/ # 配置与 prompt │ └── utils/ # 工具函数 ├── alembic/ # 数据库迁移 +│ └── versions/ +│ └── 20260310_add_feishu_outbox_reply_hash.py # 飞书 outbox 回执哈希幂等字段 ├── requirements.txt # Python 依赖 ├── Dockerfile └── tests/ @@ -129,6 +131,7 @@ uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 - 禁止在 API 层写业务逻辑，业务逻辑归属 `services/` - 禁止在前端组件内直接调用 `fetch`，必须通过 `frontend/src/api/` 封装层 - 前端状态管理：服务端数据用 React Query，本地 UI 状态用 `useState` +- 飞书回执发送仅允许 `main.py` 的 drain_loop 作为消费者，禁止在业务路径即时派发 outbox 发送任务 ## 9. 新增代码的参考模式 - 新增后端接口：参考 `backend/app/api/sstc.py` 的结构 @@ -139,6 +142,7 @@ uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 ## 10. 幂等与事务约束 - 所有入站事件必须经过 `ingest_events` 幂等去重 - 飞书回复必须经过 `feishu_reply_outbox`，禁止同步直接发送 +- 飞书回复发送幂等：`reply_hash = MD5(source_message_id + reply_text)`，同租户下 `status=sent` 的相同哈希不得重复发送 - 跨表写操作必须在同一个数据库事务内完成 ## 11. 已知 BOUNDARY（不要幻构行为） diff --git a/backend/alembic/versions/20260310_add_feishu_outbox_reply_hash.py b/backend/alembic/versions/20260310_add_feishu_outbox_reply_hash.py new file mode 100644 index 00000000..3d131875 --- /dev/null +++ b/backend/alembic/versions/20260310_add_feishu_outbox_reply_hash.py @@ -0,0 +1,111 @@ +"""add feishu outbox reply hash + +Revision ID: add_feishu_outbox_reply_hash +Revises: add_ai_pmo_simulation_field +Create Date: 2026-03-10 +""" + +from __future__ import annotations + +from typing import Any + +from alembic import op +import sqlalchemy as sa + + +revision = "add_feishu_outbox_reply_hash" +down_revision = "add_ai_pmo_simulation_field" +branch_labels = None +depends_on = None + + +def _table_exists(bind, table_name: str) -> bool: + return sa.inspect(bind).has_table(table_name) + + +def _column_exists(bind, table_name: str, column_name: str) -> bool: + inspector = sa.inspect(bind) + if not inspector.has_table(table_name): + return False + return any(col.get("name") == column_name for col in inspector.get_columns(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_not_exists( + bind, + *, + index_name: str, + table_name: str, + columns: list[str], + unique: bool = False, + postgresql_where: Any = None, + sqlite_where: Any = None, +) -> None: + if _index_exists(bind, table_name, index_name): + return + kwargs: dict[str, Any] = {} + if postgresql_where is not None: + kwargs["postgresql_where"] = postgresql_where + if sqlite_where is not None: + kwargs["sqlite_where"] = sqlite_where + op.create_index(index_name, table_name, columns, unique=unique, **kwargs) + + +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, "feishu_reply_outbox"): + return + + if not _column_exists(bind, "feishu_reply_outbox", "source_message_id"): + op.add_column("feishu_reply_outbox", sa.Column("source_message_id", sa.String(length=128), nullable=True)) + if not _column_exists(bind, "feishu_reply_outbox", "reply_hash"): + op.add_column("feishu_reply_outbox", sa.Column("reply_hash", sa.String(length=32), nullable=True)) + + _create_index_if_not_exists( + bind, + index_name="ix_feishu_reply_outbox_source_message_id", + table_name="feishu_reply_outbox", + columns=["source_message_id"], + ) + _create_index_if_not_exists( + bind, + index_name="ix_feishu_reply_outbox_reply_hash", + table_name="feishu_reply_outbox", + columns=["reply_hash"], + ) + sent_where = sa.text("status = 'sent' AND reply_hash IS NOT NULL") + _create_index_if_not_exists( + bind, + index_name="uq_feishu_reply_outbox_sent_reply_hash", + table_name="feishu_reply_outbox", + columns=["tenant_id", "reply_hash"], + unique=True, + postgresql_where=sent_where, + sqlite_where=sent_where, + ) + + +def downgrade() -> None: + bind = op.get_bind() + if not _table_exists(bind, "feishu_reply_outbox"): + return + + _drop_index_if_exists(bind, "feishu_reply_outbox", "uq_feishu_reply_outbox_sent_reply_hash") + _drop_index_if_exists(bind, "feishu_reply_outbox", "ix_feishu_reply_outbox_reply_hash") + _drop_index_if_exists(bind, "feishu_reply_outbox", "ix_feishu_reply_outbox_source_message_id") + + if _column_exists(bind, "feishu_reply_outbox", "reply_hash"): + op.drop_column("feishu_reply_outbox", "reply_hash") + if _column_exists(bind, "feishu_reply_outbox", "source_message_id"): + op.drop_column("feishu_reply_outbox", "source_message_id") diff --git a/backend/app/config/ai_prompts.py b/backend/app/config/ai_prompts.py index 11f6cc7e..a2847e8b 100644 --- a/backend/app/config/ai_prompts.py +++ b/backend/app/config/ai_prompts.py @@ -205,6 +205,31 @@ PROMPT_FEISHU_REQUIREMENT_ANALYSIS = """ """ +PROMPT_FEISHU_HUMANIZED_REPLY = """ +直接回答用户问题，不要做自我介绍。 + +你将基于以下上下文生成一条 35-140 字中文回复： +- 当前系统时间（北京时间）：{now_cn_text} +- 提及用户：{sender_name} +- 用户消息：{user_message} +- 系统处理状态：{status} +- 分类结果：{classification} +- 处理原因：{reason} +- 处理追踪ID：{ai_trace_id} +- 关联项目ID：{linked_projects_text} + +硬性规则： +1. 首句必须直接回答“用户消息”的核心问题，禁止先讲系统身份或接入方式。 +2. 禁止默认输出“我是...”“我可以...”“我现在通过...接入...”这类自我介绍句式。 +3. 仅当用户明确询问“你是谁/你能做什么/系统能力”时，才允许用一句话说明能力边界。 +4. 若信息不足，请明确缺失信息并给出 1-3 条可执行下一步。 +5. 如果用户询问日期/时间/星期，必须基于“当前系统时间（北京时间）”直接回答，不得编造。 +6. 不要输出 Markdown，不要输出 JSON。 + +只输出最终回复正文。 +""".strip() + + PROMPT_FEISHU_PROJECT_SIGNAL_CLASSIFICATION = """ 你是 PMFlow 的项目需求信号分流器。 请基于“项目列表上下文”和“飞书消息”，判断该消息是否为有效需求信号，并识别它属于哪个已有项目。 diff --git a/backend/app/main.py b/backend/app/main.py index 02c84cc1..bb3fb20a 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -4,23 +4,26 @@ from __future__ import annotations import asyncio import os -from contextlib import asynccontextmanager +from contextlib import asynccontextmanager, suppress +from datetime import datetime, timedelta, timezone from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware import sqlalchemy as sa +from sqlalchemy.exc import OperationalError, ProgrammingError from .api.router import api_router from .core.config import settings from .core.logging import configure_logging, get_logger from .core.response import success_response from .database import AsyncSessionLocal, engine -from .models import Base +from .models import Base, Project, WorkItem from .services.scheduler_center import WorkloadRecalcScheduler from .services.scheduler_center import KnowledgeUrlRefreshScheduler from .services.scheduler_center import KnowledgeGraphIncrementalScheduler from .services.scheduler_center import AIPMOCadenceScheduler from .services.ai_orchestration_run_service import AIOrchestrationRunService +from .services.ai_pmo_kernel_service import AIPMOKernelService from .services.skill_service import init_system_skill_tags from .services.feishu_long_connection import init_feishu_long_connection_manager from .services.feishu_integration import dispatch_pending_feishu_replies @@ -30,6 +33,17 @@ configure_logging(settings.debug) logger = get_logger(__name__) register_context_cache_invalidation_hooks() +ACTIVE_PROJECT_STATUSES = ("active", "planning", "todo", "doing") + + +def _is_missing_table_error(exc: Exception) -> bool: + text = str(exc).lower() + return ( + "no such table" in text + or "undefined table" in text + or "relation" in text and "does not exist" in text + ) + def _sqlite_backfill_columns(sync_conn) -> None: """为历史 SQLite 库补齐新增列，避免运行时 500。""" @@ -78,6 +92,8 @@ def _sqlite_backfill_columns(sync_conn) -> None: _add_column("ai_processing_logs", "operation_code", "VARCHAR(80)") _add_column("ai_processing_logs", "operation_result", "JSON") _add_column("ai_processing_logs", "reply_source", "VARCHAR(40)") + _add_column("feishu_reply_outbox", "source_message_id", "VARCHAR(128)") + _add_column("feishu_reply_outbox", "reply_hash", "VARCHAR(32)") def _normalize_dialect(value: str) -> str: @@ -165,8 +181,10 @@ def _ensure_feishu_integrity_guards(sync_conn) -> None: 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.Column("trace_id", sa.String(length=64), nullable=True), + sa.Column("source_message_id", sa.String(length=128), nullable=True), sa.Column("chat_id", sa.String(length=128), nullable=False), sa.Column("message_text", sa.Text(), nullable=False), + sa.Column("reply_hash", sa.String(length=32), nullable=True), sa.Column("status", sa.String(length=20), nullable=False, server_default=sa.text("'pending'")), sa.Column("attempts", sa.Integer(), nullable=False, server_default=sa.text("0")), sa.Column("next_retry_at", sa.TIMESTAMP(), nullable=True), @@ -174,21 +192,37 @@ def _ensure_feishu_integrity_guards(sync_conn) -> None: sa.Column("last_error", sa.Text(), nullable=True), ) outbox_table.create(sync_conn, checkfirst=True) + outbox_columns = {col["name"] for col in sa.inspect(sync_conn).get_columns("feishu_reply_outbox")} + if "source_message_id" not in outbox_columns: + sync_conn.exec_driver_sql("ALTER TABLE feishu_reply_outbox ADD COLUMN source_message_id VARCHAR(128)") + if "reply_hash" not in outbox_columns: + sync_conn.exec_driver_sql("ALTER TABLE feishu_reply_outbox ADD COLUMN reply_hash VARCHAR(32)") sync_conn.exec_driver_sql( "CREATE INDEX IF NOT EXISTS ix_feishu_reply_outbox_tenant_id ON feishu_reply_outbox (tenant_id)" ) sync_conn.exec_driver_sql( "CREATE INDEX IF NOT EXISTS ix_feishu_reply_outbox_trace_id ON feishu_reply_outbox (trace_id)" ) + sync_conn.exec_driver_sql( + "CREATE INDEX IF NOT EXISTS ix_feishu_reply_outbox_source_message_id ON feishu_reply_outbox (source_message_id)" + ) sync_conn.exec_driver_sql( "CREATE INDEX IF NOT EXISTS ix_feishu_reply_outbox_chat_id ON feishu_reply_outbox (chat_id)" ) + sync_conn.exec_driver_sql( + "CREATE INDEX IF NOT EXISTS ix_feishu_reply_outbox_reply_hash ON feishu_reply_outbox (reply_hash)" + ) sync_conn.exec_driver_sql( "CREATE INDEX IF NOT EXISTS ix_feishu_reply_outbox_status ON feishu_reply_outbox (status)" ) sync_conn.exec_driver_sql( "CREATE INDEX IF NOT EXISTS ix_feishu_reply_outbox_next_retry_at ON feishu_reply_outbox (next_retry_at)" ) + sync_conn.exec_driver_sql( + "CREATE UNIQUE INDEX IF NOT EXISTS uq_feishu_reply_outbox_sent_reply_hash " + "ON feishu_reply_outbox (tenant_id, reply_hash) " + "WHERE status = 'sent' AND reply_hash IS NOT NULL" + ) inspector = sa.inspect(sync_conn) if not inspector.has_table("feishu_items"): @@ -311,7 +345,11 @@ def _ensure_feishu_integrity_guards(sync_conn) -> None: ) -async def _feishu_outbox_drain_loop() -> None: +def _is_postgresql_runtime() -> bool: + return _normalize_dialect(engine.dialect.name) == "postgresql" + + +async def _feishu_outbox_polling_loop() -> None: while True: try: dispatched = await dispatch_pending_feishu_replies(limit=50) @@ -323,6 +361,161 @@ async def _feishu_outbox_drain_loop() -> None: await asyncio.sleep(2.0) +async def _feishu_outbox_listen_loop() -> None: + while True: + listener_callback = None + driver_connection = None + try: + async with engine.connect() as conn: + await conn.execute(sa.text("LISTEN feishu_outbox_ready")) + await conn.commit() + raw_conn = await conn.get_raw_connection() + driver_connection = getattr(raw_conn, "driver_connection", None) + if driver_connection is None or not hasattr(driver_connection, "add_listener"): + logger.warning("feishu_outbox_listen_unsupported_driver", driver=str(type(driver_connection))) + await _feishu_outbox_polling_loop() + return + + notify_queue: asyncio.Queue[str] = asyncio.Queue(maxsize=1024) + + def _on_notify(_connection, _pid, _channel, payload) -> None: + try: + notify_queue.put_nowait(str(payload or "")) + except asyncio.QueueFull: + pass + + listener_callback = _on_notify + await driver_connection.add_listener("feishu_outbox_ready", listener_callback) + logger.info("feishu_outbox_listen_ready") + + while True: + dispatched = await dispatch_pending_feishu_replies(limit=50) + if dispatched > 0: + continue + try: + await asyncio.wait_for(notify_queue.get(), timeout=2.0) + while not notify_queue.empty(): + notify_queue.get_nowait() + except asyncio.TimeoutError: + continue + except asyncio.CancelledError: + break + except Exception as exc: + logger.warning("feishu_outbox_listen_failed", error=str(exc)) + await asyncio.sleep(2.0) + finally: + if driver_connection is not None and listener_callback is not None: + with suppress(Exception): + await driver_connection.remove_listener("feishu_outbox_ready", listener_callback) + + +async def _feishu_outbox_drain_loop() -> None: + if _is_postgresql_runtime(): + await _feishu_outbox_listen_loop() + return + await _feishu_outbox_polling_loop() + + +async def _get_recent_active_projects( + *, + days: int, + limit_per_tenant: int, + max_projects: int, +) -> list[tuple[int, int]]: + since = (datetime.now(timezone.utc) - timedelta(days=max(1, int(days)))).replace(tzinfo=None) + per_tenant_limit = max(1, int(limit_per_tenant)) + total_limit = max(per_tenant_limit, int(max_projects)) + + async with AsyncSessionLocal() as db: + try: + rows = ( + await db.execute( + sa.select( + WorkItem.tenant_id.label("tenant_id"), + WorkItem.project_id.label("project_id"), + sa.func.max(WorkItem.created_at).label("last_seen_at"), + ) + .join( + Project, + sa.and_( + Project.tenant_id == WorkItem.tenant_id, + Project.id == WorkItem.project_id, + ), + ) + .where( + sa.and_( + WorkItem.project_id.is_not(None), + WorkItem.created_at >= since, + Project.status.in_(ACTIVE_PROJECT_STATUSES), + ) + ) + .group_by(WorkItem.tenant_id, WorkItem.project_id) + ) + ).all() + except (OperationalError, ProgrammingError) as exc: + if not _is_missing_table_error(exc): + raise + logger.warning("[WM_WARMUP] 活跃项目查询跳过", reason="missing_table", error=str(exc)) + return [] + + bucket: dict[int, list[tuple[datetime, int]]] = {} + for row in rows: + tenant_id = int(getattr(row, "tenant_id")) + project_id = int(getattr(row, "project_id")) + last_seen = getattr(row, "last_seen_at") + last_seen_at = last_seen if isinstance(last_seen, datetime) else datetime.min + bucket.setdefault(tenant_id, []).append((last_seen_at, project_id)) + + picked: list[tuple[datetime, int, int]] = [] + for tenant_id, entries in bucket.items(): + entries.sort(key=lambda item: item[0], reverse=True) + for last_seen_at, project_id in entries[:per_tenant_limit]: + picked.append((last_seen_at, tenant_id, project_id)) + + picked.sort(key=lambda item: item[0], reverse=True) + return [(tenant_id, project_id) for _, tenant_id, project_id in picked[:total_limit]] + + +async def _warmup_world_models() -> None: + days = max(1, int(os.getenv("WORLD_MODEL_WARMUP_DAYS", "7"))) + limit_per_tenant = max(1, int(os.getenv("WORLD_MODEL_WARMUP_LIMIT_PER_TENANT", "5"))) + max_projects = max(limit_per_tenant, int(os.getenv("WORLD_MODEL_WARMUP_MAX_PROJECTS", "50"))) + try: + active_projects = await _get_recent_active_projects( + days=days, + limit_per_tenant=limit_per_tenant, + max_projects=max_projects, + ) + warmed = 0 + for tenant_id, project_id in active_projects: + try: + async with AsyncSessionLocal() as db: + kernel = AIPMOKernelService(db, tenant_id=int(tenant_id)) + await kernel.build_world_model( + tenant_id=int(tenant_id), + project_id=int(project_id), + sender="wm_warmup", + ) + warmed += 1 + except Exception as exc: + logger.warning( + "[WM_WARMUP] 项目预热失败", + tenant_id=tenant_id, + project_id=project_id, + error=str(exc), + ) + logger.info( + "[WM_WARMUP] 预热完成", + count=warmed, + total=len(active_projects), + days=days, + limit_per_tenant=limit_per_tenant, + max_projects=max_projects, + ) + except Exception as exc: + logger.warning("[WM_WARMUP] 预热失败", error=str(exc)) + + @asynccontextmanager async def lifespan(_: FastAPI): """应用生命周期：初始化表与系统技能标签。""" @@ -367,6 +560,10 @@ async def lifespan(_: FastAPI): _feishu_outbox_drain_loop(), name="feishu-outbox-drain-loop", ) + wm_warmup_task = asyncio.create_task( + _warmup_world_models(), + name="wm-world-model-warmup", + ) yield outbox_task.cancel() @@ -374,6 +571,10 @@ async def lifespan(_: FastAPI): await outbox_task except asyncio.CancelledError: pass + if not wm_warmup_task.done(): + wm_warmup_task.cancel() + with suppress(asyncio.CancelledError): + await wm_warmup_task await feishu_lc.stop() await ai_pmo_cadence_scheduler.stop() await kg_scheduler.stop() diff --git a/backend/app/models/feishu_reply_outbox.py b/backend/app/models/feishu_reply_outbox.py index 1ec38717..064b5ec1 100644 --- a/backend/app/models/feishu_reply_outbox.py +++ b/backend/app/models/feishu_reply_outbox.py @@ -15,8 +15,10 @@ class FeishuReplyOutbox(Base): id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) trace_id: Mapped[Optional[str]] = mapped_column(VARCHAR(64), nullable=True, index=True) + source_message_id: Mapped[Optional[str]] = mapped_column(VARCHAR(128), nullable=True, index=True) chat_id: Mapped[str] = mapped_column(VARCHAR(128), nullable=False, index=True) message_text: Mapped[str] = mapped_column(TEXT, nullable=False) + reply_hash: Mapped[Optional[str]] = mapped_column(VARCHAR(32), nullable=True, index=True) status: Mapped[str] = mapped_column(VARCHAR(20), nullable=False, default="pending", index=True) attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0) next_retry_at: Mapped[Optional[datetime]] = mapped_column(TIMESTAMP, nullable=True, index=True) diff --git a/backend/app/services/ai_pmo_kernel_service.py b/backend/app/services/ai_pmo_kernel_service.py index 8b46b543..18eb24ff 100644 --- a/backend/app/services/ai_pmo_kernel_service.py +++ b/backend/app/services/ai_pmo_kernel_service.py @@ -92,7 +92,30 @@ RECEIPT_PHASE_ACTION = "action_applied" RECEIPT_PHASE_PENDING = "pending_manual" RECEIPT_PHASE_ERROR = "error" -QUESTION_HINTS = ("?", "？", "怎么", "为何", "为什么", "什么", "多少", "是否", "帮我看", "状态") +QUESTION_HINTS = ( + "?", + "？", + "怎么", + "如何", + "为何", + "为什么", + "什么", + "是什么", + "哪些", + "有哪些", + "特点", + "技术特点", + "能力", + "介绍", + "说明", + "多少", + "是否", + "帮我看", + "状态", + "风险", + "能做什么", + "可以做什么", +) SMALLTALK_EXACT = { "收到", "收到啦", @@ -161,6 +184,27 @@ class CanonicalIntent: class IntentResolver: """把输入收口为 canonical intent。""" + OPERATION_VERBS = ( + "新增", + "增加", + "加一个", + "加个", + "新建", + "创建", + "建立", + "立项", + "接入", + "修复", + "优化", + "改成", + "改为", + "更新", + "分配", + "指派", + "归档", + "执行", + ) + @staticmethod def resolve(*, text: str, project_id: Optional[int]) -> CanonicalIntent: normalized = re.sub(r"\s+", "", str(text or "")).lower() @@ -191,7 +235,7 @@ class IntentResolver: pretty[:300], {}, ) - if any(token in pretty for token in QUESTION_HINTS): + if IntentResolver._looks_like_query(text=pretty): return CanonicalIntent( INTENT_QUERY, 0.88, @@ -210,14 +254,31 @@ class IntentResolver: {}, ) return CanonicalIntent( - INTENT_SIGNAL_CREATE, - 0.65, + INTENT_QUERY, + 0.6, project_id, - IntentResolver._build_signal_title(pretty, default="新信号"), + "", pretty[:300], - {"fallback": True}, + {"fallback": True, "reason": "default_query_fallback"}, ) + @staticmethod + def _looks_like_query(*, text: str) -> bool: + normalized = re.sub(r"\s+", "", str(text or "")).lower() + if not normalized: + return False + if any(token in normalized for token in QUESTION_HINTS): + return True + has_question_structure = ( + "?" in normalized + or "？" in normalized + or any(token in normalized for token in ("吗", "呢", "吧", "是否", "是不是")) + ) + if not has_question_structure: + return False + has_operation = any(token in normalized for token in IntentResolver.OPERATION_VERBS) + return not has_operation + @staticmethod def _build_signal_title(text: str, *, default: str) -> str: cleaned = re.sub(r"\s+", " ", str(text or "")).strip(" ，,:：") diff --git a/backend/app/services/feishu_integration.py b/backend/app/services/feishu_integration.py index df2a01ed..f63c834a 100644 --- a/backend/app/services/feishu_integration.py +++ b/backend/app/services/feishu_integration.py @@ -21,7 +21,8 @@ from uuid import uuid4 from datetime import datetime, timedelta, timezone from typing import Any, Dict, List, Optional import httpx -from sqlalchemy import and_, or_ +from sqlalchemy import and_, or_, text, update +from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.future import select @@ -35,10 +36,10 @@ from ..models import ( AIProcessingLog, ) from ..database import AsyncSessionLocal +from ..config.ai_prompts import PROMPT_FEISHU_HUMANIZED_REPLY from .notification_service import NotificationService from .signal_capture import ProactiveSignalCaptureService from .llm import LLMService -from .prompt_profile_service import prompt_profile_service CN_TZ = timezone(timedelta(hours=8)) WEEKDAY_CN = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"] @@ -64,6 +65,29 @@ class FeishuIntegrationService: def _normalize_event_mode(_: Optional[str] = None) -> str: return "long_connection" + @staticmethod + def _is_postgresql_session(db: AsyncSession) -> bool: + try: + bind = db.get_bind() + except Exception: + return False + dialect_name = str(getattr(getattr(bind, "dialect", None), "name", "")).lower() + return dialect_name.startswith("postgres") + + @staticmethod + async def _notify_outbox_ready(db: AsyncSession, *, outbox_id: Optional[int]) -> None: + """仅在 PostgreSQL 下发送 outbox ready 事件。""" + if not FeishuIntegrationService._is_postgresql_session(db): + return + try: + payload = str(int(outbox_id))[:64] if outbox_id is not None else "" + await db.execute( + text("SELECT pg_notify('feishu_outbox_ready', :payload)"), + {"payload": payload}, + ) + except Exception as exc: + print(f"[Feishu] outbox notify failed: outbox_id={outbox_id}, error={exc}") + async def load_config(self, tenant_id: int = 1): """从数据库加载飞书配置""" self.tenant_id = tenant_id @@ -229,6 +253,7 @@ class FeishuIntegrationService: mention_name=(sender.get('sender_name') or sender.get('name') or '你'), text="我已收到你的消息，但目前无法直接解析纯图片内容。请补充 1-2 句文字需求，我会立即给出分派建议。", trace_id=state_trace_id, + source_message_id=message_id, ) self._emit_state_transition( trace_id=state_trace_id, @@ -292,6 +317,7 @@ class FeishuIntegrationService: mention_name=sender_name, text=self._build_error_reply_text(), trace_id=state_trace_id, + source_message_id=message_id, ) self._emit_state_transition( trace_id=state_trace_id, @@ -374,6 +400,7 @@ class FeishuIntegrationService: mention_name=sender_name, text=deterministic_reply, trace_id=state_trace_id, + source_message_id=message_id, ) reply_route = "execute_template" elif scene == "qa": @@ -389,6 +416,7 @@ class FeishuIntegrationService: mention_name=sender_name, text=str(result.get("direct_reply_text") or "已收到，正在处理。"), trace_id=state_trace_id, + source_message_id=message_id, ) reply_route = "qa_direct" elif ( @@ -405,6 +433,7 @@ class FeishuIntegrationService: message_content=message_content, capture_result=result, trace_id=state_trace_id, + source_message_id=message_id, ) reply_route = "qa_llm" else: @@ -421,6 +450,7 @@ class FeishuIntegrationService: mention_name=sender_name, text=self._build_error_reply_text(), trace_id=state_trace_id, + source_message_id=message_id, ) reply_route = "error_template" @@ -558,7 +588,8 @@ class FeishuIntegrationService: message_content: str, capture_result: Dict[str, Any], trace_id: Optional[str] = None, - ) -> None: + source_message_id: Optional[str] = None, + ) -> None: """被 @ 时回复一条有人味的执行反馈。""" deterministic_time_reply = self._build_time_truth_reply(message_content) if deterministic_time_reply: @@ -568,6 +599,7 @@ class FeishuIntegrationService: mention_name=sender_name, text=deterministic_time_reply, trace_id=trace_id, + source_message_id=source_message_id, ) return @@ -586,6 +618,7 @@ class FeishuIntegrationService: mention_name=sender_name, text=execution_receipt, trace_id=trace_id, + source_message_id=source_message_id, ) return @@ -597,20 +630,11 @@ class FeishuIntegrationService: mention_name=sender_name, text=butler_reply, trace_id=trace_id, + source_message_id=source_message_id, ) return try: - active_role = await prompt_profile_service.get_active_role(self.db, tenant_id=self.tenant_id) - role = active_role.get("role") if isinstance(active_role, dict) else {} - role_name = str((role or {}).get("role_name") or "产品经理（任务分派）").strip() - system_prompt = str((role or {}).get("system_prompt") or "").strip() - if len(system_prompt) < 20: - system_prompt = ( - "你是 PMFlow 的任务分派产品经理。" - "请用简洁、专业、有温度的中文回复用户，先确认需求，再给可执行下一步。" - ) - status = str(capture_result.get("status") or "unknown") classification = str(capture_result.get("classification") or "unknown") reason = str(capture_result.get("reason") or "").strip() @@ -623,27 +647,16 @@ class FeishuIntegrationService: f"{WEEKDAY_CN[now_cn.weekday()]} {now_cn.strftime('%H:%M:%S')}" ) - prompt = f""" -{system_prompt} - -你正在飞书群里以「{role_name}」身份回复用户。请基于以下上下文，生成一条 35-140 字中文回复： -- 当前系统时间（北京时间）：{now_cn_text} -- 提及用户：{sender_name} -- 用户消息：{(message_content or '')[:600]} -- 系统处理状态：{status} -- 分类结果：{classification} -- 处理原因：{reason or "无"} -- 处理追踪ID：{ai_trace_id or "无"} -- 关联项目ID：{linked_projects_text or '无'} - -回复要求： -1. 语气自然、专业，避免机器口吻。 -2. 先确认你已处理，再给 1-3 条下一步建议。 -3. 若信息不足，请明确要补充什么。 -4. 如果用户询问“日期/今天/时间/几点/星期”，必须基于“当前系统时间（北京时间）”直接回答，不得编造历史日期。 -5. 不要输出 Markdown，不要输出 JSON。 -只输出最终回复正文。 -""".strip() + prompt = PROMPT_FEISHU_HUMANIZED_REPLY.format( + now_cn_text=now_cn_text, + sender_name=sender_name, + user_message=(message_content or "")[:600], + status=status, + classification=classification, + reason=(reason or "无"), + ai_trace_id=(ai_trace_id or "无"), + linked_projects_text=(linked_projects_text or "无"), + ) llm = LLMService() raw_reply = await llm.generate( @@ -655,6 +668,8 @@ class FeishuIntegrationService: max_tokens=180, ) reply_text = self._clean_reply_text(raw_reply) + if not self._is_capability_question(message_content): + reply_text = self._strip_identity_lead(reply_text) if (not reply_text) or self._looks_like_provider_error(reply_text): reply_text = self._fallback_reply_text(capture_result) except Exception as exc: @@ -667,6 +682,7 @@ class FeishuIntegrationService: mention_name=sender_name, text=reply_text, trace_id=trace_id, + source_message_id=source_message_id, ) async def _safe_reply_text( @@ -677,6 +693,7 @@ class FeishuIntegrationService: mention_open_id: Optional[str] = None, mention_name: Optional[str] = None, trace_id: Optional[str] = None, + source_message_id: Optional[str] = None, ) -> None: try: plain_text = self._normalize_outgoing_reply_text(text) @@ -685,23 +702,16 @@ class FeishuIntegrationService: display_name=mention_name, ) final_text = f"{mention_prefix}{plain_text}".strip() - outbox_id = await self._enqueue_reply_outbox( + await self._enqueue_reply_outbox( chat_id=chat_id, message_text=final_text, trace_id=trace_id, + source_message_id=source_message_id, ) - asyncio.create_task(self._dispatch_outbox_entry(outbox_id=outbox_id)) except Exception as exc: print(f"[Feishu] enqueue reply outbox failed: chat_id={chat_id}, error={exc}") - try: - plain_text = self._normalize_outgoing_reply_text(text) - mention_prefix = self._build_mention_prefix( - open_id=mention_open_id, - display_name=mention_name, - ) - await self.send_message_to_group(chat_id=chat_id, message=f"{mention_prefix}{plain_text}".strip()) - except Exception as send_exc: - print(f"[Feishu] send fallback failed: chat_id={chat_id}, error={send_exc}") + # 统一由 drain_loop 消费 outbox，禁止业务路径直发，避免双发送。 + return async def _enqueue_reply_outbox( self, @@ -709,130 +719,90 @@ class FeishuIntegrationService: chat_id: str, message_text: str, trace_id: Optional[str], + source_message_id: Optional[str], ) -> int: async with AsyncSessionLocal() as db: + normalized_source_message_id = (source_message_id or "")[:128] or None + normalized_message_text = (message_text or "")[:2000] or "已收到，正在处理。" + reply_hash = self._build_reply_hash( + source_message_id=normalized_source_message_id, + message_text=normalized_message_text, + ) + if reply_hash: + sent_exists = ( + await db.execute( + select(FeishuReplyOutbox.id) + .where( + FeishuReplyOutbox.tenant_id == int(self.tenant_id or 1), + FeishuReplyOutbox.reply_hash == reply_hash, + FeishuReplyOutbox.status == "sent", + ) + .limit(1) + ) + ).scalar_one_or_none() + if sent_exists is not None: + return int(sent_exists) row = FeishuReplyOutbox( tenant_id=int(self.tenant_id or 1), trace_id=(trace_id or "")[:64] or None, + source_message_id=normalized_source_message_id, chat_id=(chat_id or "")[:128], - message_text=(message_text or "")[:2000] or "已收到，正在处理。", + message_text=normalized_message_text, + reply_hash=reply_hash, status="pending", attempts=0, next_retry_at=None, ) db.add(row) - await db.commit() + try: + await db.flush() + await self._notify_outbox_ready(db, outbox_id=int(row.id) if row.id is not None else None) + await db.commit() + except IntegrityError: + await db.rollback() + if reply_hash: + sent_exists = ( + await db.execute( + select(FeishuReplyOutbox.id) + .where( + FeishuReplyOutbox.tenant_id == int(self.tenant_id or 1), + FeishuReplyOutbox.reply_hash == reply_hash, + FeishuReplyOutbox.status == "sent", + ) + .limit(1) + ) + ).scalar_one_or_none() + if sent_exists is not None: + return int(sent_exists) + raise await db.refresh(row) return int(row.id) - async def _dispatch_outbox_entry(self, *, outbox_id: int) -> None: - await self.__class__._dispatch_outbox_entry_by_id(outbox_id=outbox_id) - - @classmethod - async def _dispatch_outbox_entry_by_id(cls, *, outbox_id: int) -> None: - for _ in range(OUTBOX_MAX_ATTEMPTS + 1): - now = datetime.utcnow() - should_wait_seconds = 0.0 - async with AsyncSessionLocal() as db: - row = ( - await db.execute( - select(FeishuReplyOutbox).where( - FeishuReplyOutbox.id == int(outbox_id) - ) - ) - ).scalar_one_or_none() - if row is None: - return - status = str(row.status or "").lower() - if status == "sent": - return - if status == "failed": - return - if status == "sending": - stale_seconds = 0.0 - marker = row.updated_at or row.created_at - if isinstance(marker, datetime): - try: - marker_naive = marker.replace(tzinfo=None) if marker.tzinfo else marker - stale_seconds = max(0.0, (now - marker_naive).total_seconds()) - except Exception: - stale_seconds = 0.0 - if stale_seconds >= OUTBOX_STALE_SENDING_SECONDS: - row.status = "retrying" - row.next_retry_at = now - await db.commit() - else: - return - if row.next_retry_at and row.next_retry_at > now: - should_wait_seconds = max(0.2, min(30.0, (row.next_retry_at - now).total_seconds())) - else: - row.status = "sending" - row.attempts = int(row.attempts or 0) + 1 - row.last_error = None - await db.commit() - - if should_wait_seconds > 0: - await asyncio.sleep(should_wait_seconds) - continue - - send_error: Optional[str] = None - async with AsyncSessionLocal() as db_send: - row = ( - await db_send.execute( - select(FeishuReplyOutbox).where( - FeishuReplyOutbox.id == int(outbox_id) - ) - ) - ).scalar_one_or_none() - if row is None: - return - try: - service = FeishuIntegrationService(db_send) - await service.load_config(tenant_id=int(row.tenant_id)) - await service.send_message_to_group( - chat_id=str(row.chat_id or ""), - message=str(row.message_text or ""), - ) - row.status = "sent" - row.sent_at = datetime.utcnow() - row.next_retry_at = None - row.last_error = None - await db_send.commit() - return - except Exception as exc: - send_error = str(exc) - - async with AsyncSessionLocal() as db_retry: - row = ( - await db_retry.execute( - select(FeishuReplyOutbox).where( - FeishuReplyOutbox.id == int(outbox_id) - ) - ) - ).scalar_one_or_none() - if row is None: - return - attempts = int(row.attempts or 0) - if attempts >= OUTBOX_MAX_ATTEMPTS: - row.status = "failed" - row.next_retry_at = None - else: - row.status = "retrying" - delay_seconds = min(30, 2 * (2 ** max(0, attempts - 1))) - row.next_retry_at = datetime.utcnow() + timedelta(seconds=delay_seconds) - row.last_error = (send_error or "unknown_send_error")[:1000] - await db_retry.commit() - - if attempts >= OUTBOX_MAX_ATTEMPTS: - return - await asyncio.sleep(min(5, 2 * max(1, attempts))) - @classmethod async def dispatch_pending_outbox_entries(cls, *, limit: int = 30) -> int: - now = datetime.utcnow() fetch_limit = max(1, min(200, int(limit))) + dispatched = 0 + for _ in range(fetch_limit): + claimed = await cls._claim_next_outbox_entry(now=datetime.utcnow()) + if claimed is None: + break + await cls._dispatch_claimed_outbox_entry(claimed=claimed) + dispatched += 1 + return dispatched + + @classmethod + async def _claim_next_outbox_entry(cls, *, now: datetime) -> Optional[dict[str, Any]]: + stale_before = now - timedelta(seconds=OUTBOX_STALE_SENDING_SECONDS) async with AsyncSessionLocal() as db: - due_rows = ( + await db.execute( + update(FeishuReplyOutbox) + .where( + FeishuReplyOutbox.status == "sending", + FeishuReplyOutbox.updated_at <= stale_before, + ) + .values(status="retrying", next_retry_at=now) + ) + candidate_id = ( await db.execute( select(FeishuReplyOutbox.id) .where( @@ -843,18 +813,121 @@ class FeishuIntegrationService: ), ) .order_by(FeishuReplyOutbox.id.asc()) - .limit(fetch_limit) + .with_for_update(skip_locked=True) + .limit(1) ) - ).scalars().all() + ).scalar_one_or_none() + if candidate_id is None: + await db.commit() + return None - if not due_rows: - return 0 + claimed = ( + await db.execute( + update(FeishuReplyOutbox) + .where( + FeishuReplyOutbox.id == int(candidate_id), + FeishuReplyOutbox.status.in_(("pending", "retrying")), + or_( + FeishuReplyOutbox.next_retry_at.is_(None), + FeishuReplyOutbox.next_retry_at <= now, + ), + ) + .values( + status="sending", + attempts=FeishuReplyOutbox.attempts + 1, + last_error=None, + next_retry_at=None, + ) + .returning( + FeishuReplyOutbox.id, + FeishuReplyOutbox.tenant_id, + FeishuReplyOutbox.chat_id, + FeishuReplyOutbox.message_text, + FeishuReplyOutbox.reply_hash, + FeishuReplyOutbox.attempts, + ) + ) + ).mappings().first() + await db.commit() + if claimed is None: + return None + return dict(claimed) - await asyncio.gather( - *(cls._dispatch_outbox_entry_by_id(outbox_id=int(outbox_id)) for outbox_id in due_rows), - return_exceptions=True, - ) - return len(due_rows) + @classmethod + async def _dispatch_claimed_outbox_entry(cls, *, claimed: dict[str, Any]) -> None: + outbox_id = int(claimed.get("id") or 0) + if outbox_id <= 0: + return + tenant_id = int(claimed.get("tenant_id") or 1) + reply_hash = str(claimed.get("reply_hash") or "").strip() + if reply_hash: + async with AsyncSessionLocal() as db_check: + sent_duplicate = ( + await db_check.execute( + select(FeishuReplyOutbox.id) + .where( + FeishuReplyOutbox.id != outbox_id, + FeishuReplyOutbox.tenant_id == tenant_id, + FeishuReplyOutbox.reply_hash == reply_hash, + FeishuReplyOutbox.status == "sent", + ) + .limit(1) + ) + ).scalar_one_or_none() + if sent_duplicate is not None: + async with AsyncSessionLocal() as db_skip: + row = ( + await db_skip.execute( + select(FeishuReplyOutbox).where(FeishuReplyOutbox.id == outbox_id) + ) + ).scalar_one_or_none() + if row is not None and str(row.status or "").lower() == "sending": + row.status = "sent" + row.sent_at = datetime.utcnow() + row.next_retry_at = None + row.last_error = f"dedupe_skip_sent:{int(sent_duplicate)}" + await db_skip.commit() + return + + send_error: Optional[str] = None + try: + async with AsyncSessionLocal() as db_send: + service = FeishuIntegrationService(db_send) + await service.load_config(tenant_id=int(claimed.get("tenant_id") or 1)) + await service.send_message_to_group( + chat_id=str(claimed.get("chat_id") or ""), + message=str(claimed.get("message_text") or ""), + ) + except Exception as exc: + send_error = str(exc) + + async with AsyncSessionLocal() as db_update: + row = ( + await db_update.execute( + select(FeishuReplyOutbox).where(FeishuReplyOutbox.id == outbox_id) + ) + ).scalar_one_or_none() + if row is None: + return + if str(row.status or "").lower() != "sending": + return + attempts = int(row.attempts or claimed.get("attempts") or 0) + if send_error is None: + row.status = "sent" + row.sent_at = datetime.utcnow() + row.next_retry_at = None + row.last_error = None + await db_update.commit() + return + if attempts >= OUTBOX_MAX_ATTEMPTS: + row.status = "failed" + row.next_retry_at = None + else: + row.status = "retrying" + delay_seconds = min(30, 2 * (2 ** max(0, attempts - 1))) + row.next_retry_at = datetime.utcnow() + timedelta(seconds=delay_seconds) + row.last_error = (send_error or "unknown_send_error")[:1000] + await db_update.commit() @staticmethod def _clean_reply_text(value: Optional[str]) -> str: @@ -872,6 +945,53 @@ class FeishuIntegrationService: return cls._build_error_reply_text() return text[:500] + @staticmethod + def _build_reply_hash(*, source_message_id: Optional[str], message_text: str) -> Optional[str]: + source = str(source_message_id or "").strip() + body = str(message_text or "").strip() + if not source or not body: + return None + raw = f"{source}{body}".encode("utf-8") + return hashlib.md5(raw, usedforsecurity=False).hexdigest() + + @staticmethod + def _is_capability_question(message_content: Optional[str]) -> bool: + normalized = re.sub(r"\s+", "", str(message_content or "").lower()) + if not normalized: + return False + capability_tokens = ( + "你是谁", + "你是什么", + "你能做什么", + "你可以做什么", + "系统能力", + "能力介绍", + "介绍下你", + ) + return any(token in normalized for token in capability_tokens) + + @staticmethod + def _strip_identity_lead(reply_text: str) -> str: + text = str(reply_text or "").strip() + if not text: + return text + lowered = text.lower() + forbidden_prefixes = ( + "我是", + "我可以", + "我现在", + "是的，我现在", + "可以在这里和我", + ) + forbidden_contains = ( + "通过 feishu", + "群聊接入 openclaw", + "你可以直接在这里和我对话", + ) + if any(lowered.startswith(prefix) for prefix in forbidden_prefixes) or any(token in lowered for token in forbidden_contains): + return "" + return text + @staticmethod def _looks_like_provider_error(text: str) -> bool: lowered = str(text or "").strip().lower() diff --git a/backend/app/services/feishu_signal_pipeline.py b/backend/app/services/feishu_signal_pipeline.py index 407608eb..1759756f 100644 --- a/backend/app/services/feishu_signal_pipeline.py +++ b/backend/app/services/feishu_signal_pipeline.py @@ -106,7 +106,11 @@ FOLLOWUP_REFERENCE_RE = re.compile( re.IGNORECASE, ) OBVIOUS_QUESTION_RE = re.compile( - r"([?？]$|吗$|呢$|呀$|吧$|怎么|如何|为什么|多少|几号|几点|周几|有没有|是否|是不是|是什么|在哪|哪一个|哪几个)", + r"([?？]$|吗$|呢$|呀$|吧$|怎么|如何|为什么|多少|几号|几点|周几|有没有|是否|是不是|是什么|什么|哪些|有哪些|在哪|哪一个|哪几个|特点|能力|介绍|说明|风险)", + re.IGNORECASE, +) +QUERY_INTENT_RE = re.compile( + r"(哪些|有哪些|特点|技术特点|能力|介绍|说明|怎么|如何|是什么|什么|能做什么|可以做什么|项目有哪些风险)", re.IGNORECASE, ) FEISHU_BOT_MENTION_RE = re.compile(r"(?:]*>[^<]*|@pmflow\b|@_user_1\b)", re.IGNORECASE) @@ -669,6 +673,16 @@ class FeishuSignalPipelineService: "correction_match": correction_match, } + if self._is_query_like_question(text): + return { + "intent": INTENT_QUERY, + "canonical_action": None, + "clarification_template": "", + "risk_level": "low", + "fields": {}, + "correction_match": correction_match, + } + if self._is_obvious_question(text): return { "intent": INTENT_QUERY, @@ -2944,12 +2958,28 @@ class FeishuSignalPipelineService: text = str(message or "").strip() if not text: return False + if cls._is_query_like_question(message): + return True if cls._is_openclaw_business_command(text): return False if cls._looks_like_project_lookup_question(text): return True return bool(OBVIOUS_QUESTION_RE.search(text)) + @classmethod + def _is_query_like_question(cls, message: str) -> bool: + text = str(message or "").strip() + if not text: + return False + normalized = re.sub(r"\s+", "", text.lower()) + has_question_structure = bool(OBVIOUS_QUESTION_RE.search(text) or QUERY_INTENT_RE.search(text)) + if not has_question_structure: + return False + has_operation_verb = bool(OPENCLAW_COMMAND_VERB_RE.search(text)) + imperative_tokens = ("帮我", "请", "麻烦", "直接", "立刻", "马上", "执行", "处理", "安排") + has_imperative = any(token in normalized for token in imperative_tokens) + return not has_operation_verb and not has_imperative + @staticmethod def _decorate_capture_result( result: dict[str, Any], @@ -3450,6 +3480,8 @@ class FeishuSignalPipelineService: text = str(message or "").strip() if not text: return False + if cls._is_query_like_question(message): + return False if cls._looks_like_project_create_command(text): return True return bool(OPENCLAW_COMMAND_VERB_RE.search(text) and OPENCLAW_COMMAND_OBJECT_RE.search(text))