"""群里的助手 —— 它什么时候该说话,说给谁听,以及它在群里知道什么。 ## 单聊和群聊差在哪 单聊里「这条消息是给我的」是自明的:那个会话里只有员工和机器人。群里不是。一个 群里有别人、有别的机器人、有和 IT 毫不相关的话题,所以群里的第一个判断永远是 **「这句话是在跟我说吗」** —— 答案只有一个可靠来源:员工 @ 了这个机器人。飞书把 提及作为结构化字段(``event.message.mentions``)送来,每一项带一个 ``open_id``; 把它和本应用自己的 bot ``open_id`` 比对,就是这个判断的全部。 ⚠️ **不能靠「飞书把这条消息投给了我们」来推断「我被 @ 了」。** 那取决于应用当前 持有哪些权限(``im:message`` 与 ``im:message.group_at_msg`` 的投递范围不同),而 权限会变 —— 一个今天成立的推断,明天多开一个 scope 就变成「机器人开始回答它偷听 到的每一句话」。所以判断留在代码里,不留在权限配置里。 ## 两种群,两条闸 * **工单群** —— Helix 自己为「转人工」建的房间,``assistant_handoffs.feishu_chat_id`` 上有它。这类群里的助手是**服务台**:它知道这一单是什么、卡在哪、谁在处理。 * **任意群** —— 有人把机器人拉进了自己的群。这里的助手就是单聊里的那个助手, 只是换了个地方说话。 两者各有开关:工单群默认开(平台自己建的房间里不说话,是 2026-09-08 生产上被 用户当面撞见的缺陷);任意群默认关 —— 那是一个触达面决定,应该由管理员显式做出, 而不是由一次发布顺带做出。 ## 隐私姿态:被叫到才读,不是一直在听 助手**不订阅**群里的每一条消息;它只处理 @ 到自己的那些。需要读上下文时(总结 一段讨论)才按需去拉历史。区别不是性能,是产品性质:一个默默记录它看得见的一切 的机器人,和一个被叫到才应答的机器人,不是同一个东西。 """ from __future__ import annotations import logging from dataclasses import dataclass from typing import Any from sqlalchemy.ext.asyncio import AsyncSession from app.assistant.models import CHAT_STATE_OPEN from app.core.locale import DEFAULT_LANG, Lang from app.core.redis_client import get_redis logger = logging.getLogger(__name__) # ── 开关 ──────────────────────────────────────────────────────────── #: 工单群(Helix 自己建的转人工群)里助手回不回话。默认**开** —— 平台把员工和 #: 负责人拉进一个房间,然后自己在里面装哑巴,是这条链路最难解释的行为。 SETTING_GROUP_ENABLED = "assistant.group_chat_enabled" #: 任意群(别人把机器人拉进去的房间)里助手回不回话。默认**关**:多一个群能 #: @ 它,就是多一片触达面,该由管理员显式打开。 SETTING_GROUP_ANY_ENABLED = "assistant.group_chat_any_enabled" #: 每个群每小时的回合上限。防的是「一个群把整个模型预算吃光」——它和每人上限 #: 是两个独立的闸:十个人在一个群里各问六句,每人都没超,群超了。 SETTING_GROUP_TURNS_PER_HOUR = "assistant.group_turns_per_hour" DEFAULT_GROUP_ENABLED = True DEFAULT_GROUP_ANY_ENABLED = False DEFAULT_GROUP_TURNS_PER_HOUR = 120 async def group_enabled(db: AsyncSession, *, is_handoff: bool) -> bool: """这一类群此刻通不通。两个开关分开读,因为它们回答两个不同的问题。""" from app.integrations.service import get_platform_setting # noqa: PLC0415 key = SETTING_GROUP_ENABLED if is_handoff else SETTING_GROUP_ANY_ENABLED default = DEFAULT_GROUP_ENABLED if is_handoff else DEFAULT_GROUP_ANY_ENABLED value = await get_platform_setting(key, default, session=db) # 与 ``assistant_enabled`` 同一条规矩:必须是 JSON 布尔。字符串 "false" 在 # Python 里为真,而一个管理员在设置页输入 "false" 想关掉它时,他表达的意思 # 毫无歧义。 return value is True # ── 本应用自己的 bot 身份 ────────────────────────────────────────── _BOT_OPEN_ID_KEY = "assistant:bot_open_id" #: 一小时。bot 的 open_id 实际上永不改变,缓存只是为了不给每条群消息加一次 #: 外部往返;TTL 存在是为了让一次错误的缓存值有自愈的机会。 _BOT_OPEN_ID_TTL = 3600 async def bot_open_id(db: AsyncSession) -> str | None: """这个机器人自己的 ``open_id``,用于判断「他们 @ 的是我吗」。 取不到时返回 ``None`` —— 调用方据此走一条**保守**的降级(见 :func:`addressed_to_bot`),而不是把这条消息当作没发生。 """ try: cached = await get_redis().get(_BOT_OPEN_ID_KEY) if cached: return cached.decode() if isinstance(cached, bytes) else str(cached) except Exception as exc: # noqa: BLE001 — 缓存不可用不该挡住这条链路 logger.info("assistant: bot open_id cache unavailable: %s", exc) from app.assistant.channel import build_client # noqa: PLC0415 — cycle client = await build_client(db) if client is None: return None try: info = await client.get_bot_info() except Exception as exc: # noqa: BLE001 — 身份取不到是降级,不是故障 logger.warning("assistant: could not resolve bot identity: %s", exc) return None open_id = str(info.get("open_id") or "") if not open_id: logger.warning("assistant: bot info carried no open_id: %r", sorted(info)) return None try: await get_redis().set(_BOT_OPEN_ID_KEY, open_id, ex=_BOT_OPEN_ID_TTL) except Exception as exc: # noqa: BLE001 logger.info("assistant: bot open_id not cached: %s", exc) return open_id @dataclass(frozen=True, slots=True) class Mention: """一条消息里的一次 @。``key`` 是正文里的占位符(``@_user_1``)。""" key: str open_id: str name: str def parse_mentions(message: dict[str, Any]) -> tuple[Mention, ...]: """把 ``event.message.mentions`` 读成结构。形状不对的项静默跳过。""" raw = message.get("mentions") if not isinstance(raw, list): return () out: list[Mention] = [] for item in raw: if not isinstance(item, dict): continue ids = item.get("id") ids = ids if isinstance(ids, dict) else {} open_id = str(ids.get("open_id") or "") if not open_id: continue out.append(Mention( key=str(item.get("key") or ""), open_id=open_id, name=str(item.get("name") or ""), )) return tuple(out) def addressed_to_bot( mentions: tuple[Mention, ...], *, bot_id: str | None, ) -> bool: """群里这句话是不是在跟助手说。 ``bot_id`` 为 ``None`` 表示身份没解析出来(飞书接口抖动、凭据不可用)。这时 **有任何一次 @ 就算数**:飞书之所以把这条群消息投给我们,最常见的原因就是 机器人被 @ 了,而在一个已经拿到消息的房间里偶尔多答一句,好过在整个降级 期间一句不答。没有任何 @ 的群消息**永远**不处理 —— 那是「偷听」和「应答」 的分界线,不因降级而移动。 """ if not mentions: return False if bot_id is None: return True return any(m.open_id == bot_id for m in mentions) def strip_bot_mention( text: str, mentions: tuple[Mention, ...], *, bot_id: str | None, ) -> str: """把 @机器人 从正文里拿掉,其余的 @ 换成人名。 模型读到的应该是「看一下群里的聊天」,不是「@_user_1 看一下群里的聊天」—— 占位符对它毫无意义,而且它会**照抄**:生产上模型学舌 ``@_user_1`` 出现在 回答里,员工看到的是一串内部记号。 其余的 @ 保留成 ``@名字``,因为它们是内容的一部分:「转给 @张三」里的那个 人正是这句话的宾语。 """ cleaned = text or "" for mention in mentions: if not mention.key: continue if bot_id is not None and mention.open_id == bot_id: cleaned = cleaned.replace(mention.key, " ") elif bot_id is None and not mention.name: # 身份未知且这次 @ 连名字都没有:留着占位符只会污染提问。 cleaned = cleaned.replace(mention.key, " ") else: cleaned = cleaned.replace(mention.key, f"@{mention.name}") return " ".join(cleaned.split()) # ── 群的上下文 ────────────────────────────────────────────────────── @dataclass(frozen=True, slots=True) class GroupContext: """助手在这个房间里知道的事。 ``ticket_chat`` 有值表示这是 Helix 自己为一张工单开的房间 —— 助手在里面不只是 一个问答机器人,它是这张工单的**同一个面**:它知道这单是什么、卡在哪、谁在 处理,并且能在这里推进它。 """ chat_id: str ticket_chat: Any = None @property def is_handoff(self) -> bool: """这是不是一个工单群。 名字保留 ``handoff``:这个概念一开始只有「转人工」一个来源,而闸门、提示 词、审计口径都按这个词写过一遍。它现在的含义是「Helix 为一张工单开的房间」 —— 来源可以是转人工、求助人自己开、或处理人开的。 """ return self.ticket_chat is not None @property def ticket_id(self) -> Any: return self.ticket_chat.ticket_id if self.ticket_chat is not None else None async def resolve_group_context( db: AsyncSession, *, chat_id: str, ) -> GroupContext: """这个 ``chat_id`` 是不是 Helix 为某张工单开的房间。 只认台账 ``assistant_ticket_chats``(工单 ↔ 群的唯一注册表)。**不**去问飞书 这个群是谁建的 —— 平台对一个房间的权限,来自它自己的记录,不来自一次外部 查询的回答。 ⚠️ 刻意**不**读 ``assistant_handoffs.feishu_chat_id``:那一列记的是「那次 转人工当时拉起来了吗」,是一条写完不再变的事件记录;群的生命周期(加人、 卡片、解散)在这张表上。两个来源会漂,而漂的表现是「群还在,助手却当它不 存在」。历史行由迁移 633 搬了过来。 """ from app.assistant import ticket_chat as ticket_chat_mod # noqa: PLC0415 if not chat_id: return GroupContext(chat_id=chat_id) row = await ticket_chat_mod.get_by_chat(db, chat_id) if row is None or row.state != CHAT_STATE_OPEN: return GroupContext(chat_id=chat_id) return GroupContext(chat_id=chat_id, ticket_chat=row) # ── 提示词 ────────────────────────────────────────────────────────── _GROUP_PROMPT = """ YOU ARE IN A GROUP CHAT, NOT A ONE-TO-ONE CHAT Several people can read every word you write here. The person who just spoke is the one you are answering, and everything you looked up is scoped to THEM. - Never read out somebody's personal records to the room. If the answer is about one person's device, salary-adjacent data, account or request, keep the visible reply to what that person already said in public and tell them you have sent the detail privately. - Anything that needs a confirmation card is delivered to that person's own chat with you, never posted here. Say so in one short clause. - Be shorter than you would be in private. A room reads a wall of text as noise. - Address people by name when it disambiguates who you mean. YOU CAN READ THIS ROOM Call `read_this_room` to see who said what here. Do it whenever someone asks you to look at, catch up on, summarise or explain the conversation in this room, or when the question needs context you were not shown. NEVER tell people you cannot see the group history — you can, and saying otherwise sends them away from something the platform does. """ _HANDOFF_PROMPT = """ THIS ROOM IS A HELIX HANDOFF GROUP It was opened by Helix because the assistant could not resolve an employee's question, and it holds the employee who asked plus the IT owner for that category. There is a support ticket behind it. Treat yourself as the service desk in the room: keep the thread moving, be precise about state, and never invent where a ticket stands. """ def group_prompt_block(context: GroupContext, lang: Lang = DEFAULT_LANG) -> str: """群里的行为守则,拼在通道提示词后面。 与 ``card_context.as_prompt_block()`` 同一种形状:一段追加的系统上下文, 而不是第二套提示词 —— 助手在群里和在单聊里必须是同一个助手。 """ del lang # 守则是给模型读的,语言由通道提示词自己那一段决定。 block = _GROUP_PROMPT if context.is_handoff: block += _HANDOFF_PROMPT return block.strip() __all__ = [ "DEFAULT_GROUP_ANY_ENABLED", "DEFAULT_GROUP_ENABLED", "DEFAULT_GROUP_TURNS_PER_HOUR", "SETTING_GROUP_ANY_ENABLED", "SETTING_GROUP_ENABLED", "SETTING_GROUP_TURNS_PER_HOUR", "GroupContext", "Mention", "addressed_to_bot", "bot_open_id", "group_enabled", "group_prompt_block", "parse_mentions", "resolve_group_context", "strip_bot_mention", ]