"""按需读一次群里的聊天 —— 只在被叫到的时候。 ## 这是一个产品性质,不是一个优化 助手**不订阅**群里的每一条消息;它只处理 @ 到自己的那些。要读上下文时(有人说 「总结一下」)才去拉一次历史。一个默默记录它看得见的一切的机器人,和一个被叫到 才应答的机器人,不是同一个东西 —— 而员工能不能分辨这两者,决定了他们在群里 说不说真话。 所以这里的每一次读取都: * 由**一条明确的指令**触发(不是定时,不是「顺便」); * 落一行审计(读了哪个房间、多少条、谁让读的),**不含正文**; * 只取这个房间的消息,只取文本类,并且**跳过机器人自己发的**。 ## 为什么要把 open_id 翻成名字 模型读到的应该是「张三:我这边看不到日志」,而不是「ou_a3f2…:我这边看不到 日志」。名字先按 Helix 的人查(那是权威的),查不到再用飞书通讯录的显示名 —— 一段认不出说话人的记录,总结出来必然是含糊的。 """ from __future__ import annotations import json import logging from dataclasses import dataclass from datetime import UTC, datetime from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.identity.person_models import IdentityExternalIdentity, IdentityPerson logger = logging.getLogger(__name__) #: 一次最多读多少条。够覆盖一个工单群的全部讨论(实测这类房间几十条封顶), #: 又不至于让一次「总结一下」变成十几次分页往返。 DEFAULT_MAX_MESSAGES = 200 #: 飞书单页上限。 _PAGE_SIZE = 50 #: 一条消息进提示词的最大长度。有人往群里贴一整份日志时,截断的是那一条, #: 不是整段记录 —— 后者会把真正的对话挤出上下文窗口。 _MAX_CHARS_PER_MESSAGE = 800 @dataclass(frozen=True, slots=True) class RoomMessage: """群里的一条消息,已经翻成人话。""" who: str text: str at: datetime | None is_bot: bool @dataclass(frozen=True, slots=True) class RoomTranscript: """一次读取的结果。 ``truncated`` 说明「还有更早的没读」—— 总结里必须体现这一点,否则一份只 覆盖了后半段讨论的总结会被当成全部。 """ messages: tuple[RoomMessage, ...] truncated: bool @property def human_messages(self) -> tuple[RoomMessage, ...]: return tuple(m for m in self.messages if not m.is_bot) def render(self) -> str: """喂给模型的那段文本。""" lines = [] for message in self.messages: if message.is_bot: # 机器人自己说过的话不进总结的输入:它们是这段讨论的**产物**, # 让模型把自己的旧话当成事实来源,是最典型的自我强化。 continue stamp = message.at.strftime("%m-%d %H:%M") if message.at else "" lines.append(f"[{stamp}] {message.who}:{message.text}") return "\n".join(lines) def _extract_text(msg_type: str, raw_body: Any) -> str: """飞书把正文塞在一个 JSON **字符串**里,每种消息类型形状还不一样。""" content = "" if isinstance(raw_body, dict): content = str(raw_body.get("content") or "") elif isinstance(raw_body, str): content = raw_body if not content: return "" try: parsed = json.loads(content) except (ValueError, TypeError): return "" if msg_type == "text" and isinstance(parsed, dict): return str(parsed.get("text") or "").strip() if msg_type == "post" and isinstance(parsed, dict): # ``post`` 是嵌套的段落/元素数组,每个元素可能是 text / md / a / at。 chunks: list[str] = [] for locale_body in parsed.values(): if not isinstance(locale_body, dict): continue for paragraph in locale_body.get("content") or []: for element in paragraph or []: if not isinstance(element, dict): continue chunks.append(str( element.get("text") or element.get("content") or "", )) return " ".join(c for c in chunks if c).strip() # 图片、文件、卡片:说不出内容,但**说得出发生过** —— 一段讨论里「他发了 # 一张截图」本身是信息。 if msg_type in ("image", "file", "media"): return f"（{msg_type}）" return "" async def _name_map( db: AsyncSession, open_ids: set[str], ) -> dict[str, str]: """``open_id`` → 人名。先按 Helix 的人查,那是权威的。""" if not open_ids: return {} rows = (await db.execute( select(IdentityExternalIdentity.provider_subject, IdentityPerson.display_name) .join( IdentityPerson, IdentityPerson.id == IdentityExternalIdentity.person_id, ) .where( IdentityExternalIdentity.provider == "feishu", IdentityExternalIdentity.provider_subject.in_(open_ids), ) )).all() return {subject: name for subject, name in rows if name} async def read_room( db: AsyncSession, *, chat_id: str, since: datetime | None = None, max_messages: int = DEFAULT_MAX_MESSAGES, ) -> RoomTranscript | None: """把这个房间的聊天读成结构。读不了返回 ``None``。 ``since`` 通常是群创建时间:早于它的消息不属于这个房间(它本来也不会有)。 读失败(缺权限、飞书抖动)返回 ``None`` 而不是空记录 —— 「读不到」和「没人 说过话」是两件事,而基于后者写出来的总结会说「大家什么都没讨论」。 """ from app.assistant.channel import build_client # noqa: PLC0415 — cycle client = await build_client(db) if client is None or not chat_id: return None collected: list[dict[str, Any]] = [] page_token: str | None = None truncated = False start_time = int(since.timestamp()) if since else None while len(collected) < max_messages: try: data = await client.list_chat_messages( chat_id=chat_id, start_time=start_time, page_size=_PAGE_SIZE, page_token=page_token, ) except Exception as exc: # noqa: BLE001 — 读不到就说读不到 logger.warning("assistant: could not read room %s: %s", chat_id, exc) return None items = data.get("items") if isinstance(data, dict) else None if not isinstance(items, list): break collected.extend(i for i in items if isinstance(i, dict)) page_token = str(data.get("page_token") or "") or None if not data.get("has_more"): break if not page_token: # 飞书说还有,却没给游标 —— 我们**没有办法**再往下读。这也是截断, # 和「上限到了」是同一件事:把它报成读全了,会让一份只覆盖后半段 # 讨论的总结被当成全部。(2026-09-08 复审补的第二处;第一处是 # 「刚好取到上限」。) truncated = True break if len(collected) >= max_messages: # 上限到了而飞书说还有 —— 那就是**截断**,即使刚好取到 200 条。 # 少了这一行,一份只覆盖了后半段讨论的总结会被当成全部,而 # ``truncated`` 那句「只读了最近的一部分」正是为此存在的。 truncated = True break if len(collected) > max_messages: # 超出上限时留**最近的**那些:一个正在进行的讨论里,后半段才是现在 # 要决定的事。丢掉的部分由 ``truncated`` 说出来。 collected = collected[-max_messages:] truncated = True open_ids = { str((item.get("sender") or {}).get("id") or "") for item in collected } names = await _name_map(db, {o for o in open_ids if o}) messages: list[RoomMessage] = [] for item in collected: sender = item.get("sender") or {} sender_id = str(sender.get("id") or "") is_bot = str(sender.get("sender_type") or "") != "user" text = _extract_text( str(item.get("msg_type") or ""), item.get("body"), ) if not text: continue at = None raw_ts = item.get("create_time") if raw_ts: try: # 飞书给的是**毫秒**字符串。当秒读会把 2026 年读成 55000 年, # 而那个日期会原样出现在总结里。 at = datetime.fromtimestamp(int(raw_ts) / 1000, tz=UTC) except (ValueError, TypeError, OSError): at = None messages.append(RoomMessage( who=names.get(sender_id) or ("Helix" if is_bot else "同事"), text=text[:_MAX_CHARS_PER_MESSAGE], at=at, is_bot=is_bot, )) return RoomTranscript(messages=tuple(messages), truncated=truncated) # ── 把「读这个房间」交到模型手里 ──────────────────────────────────── #: 模型能调的那个工具名。 TOOL_READ_ROOM = "read_this_room" #: 一次工具调用最多读多少条。比总结那条路(``DEFAULT_MAX_MESSAGES``)保守: #: 这段文本会整段进模型的上下文,而房间里的闲聊密度比工单正文高得多。 TOOL_MAX_MESSAGES = 80 def read_room_tool() -> dict[str, Any]: """群里那一轮额外给模型的工具。 🔴 **为什么这必须是一个工具,而不是一个关键词。** 这个能力原来只挂在字面命令上(群里发「总结」两个字)。于是员工用**一句 自然语言**问同一件事 —— 「看一下群里的聊天 然后总结一下什么问题」—— 落到通用问答,而模型手里没有任何读群的手段,于是它如实回答「我无法查看 群聊历史,只能看到当前对话」。 那句话对**平台**是不准确的:平台能读。一个「我做不到」的回答,在平台其实 做得到的时候,比一个错误答案更贵 —— 员工据此不再问第二次,而这条能力从此 只对知道咒语的人存在。2026-09-08 生产上就是这样(而这正是这条链路最初要 解决的那句话)。 ``chat_id`` **不在参数里**:它来自这一轮的信封(闭包),和身份同一条规矩 —— 模型说不出别的房间的名字,也就读不到别的房间。 """ return { "name": TOOL_READ_ROOM, "description": ( "Read the recent chat history of THIS group room you are in — who " "said what, in order. Call it whenever the employee asks you to " "look at, catch up on, summarise or explain the conversation in " "this room ('看一下群里的聊天', 'what's the problem here', " "'总结一下'), or when answering needs context you were not shown. " "You CAN read this room: never tell the employee you are unable " "to see the group history. Scoped to this room only." ), "input_schema": { "type": "object", "properties": { "limit": { "type": "integer", "description": ( f"How many recent messages to read (1-{TOOL_MAX_MESSAGES}, " "default 40). Ask for more only if the discussion is long." ), }, }, "additionalProperties": False, }, } async def dispatch_read_room( db: AsyncSession, *, chat_id: str, tool_input: dict[str, Any], ) -> str: """执行那次读取。**读不到时如实说读不到** —— 不返回空字符串。 空字符串会被模型读成「大家什么都没说」,而它接着会据此总结出一个「没有讨论」 的结论。「读不到」和「没人说过话」是两件不同的事,这条链路上每一处都按这条 办(见 :func:`read_room` 的 ``None``)。 """ raw = tool_input.get("limit") try: limit = int(raw) if raw is not None else 40 except (TypeError, ValueError): limit = 40 limit = max(1, min(limit, TOOL_MAX_MESSAGES)) transcript = await read_room(db, chat_id=chat_id, max_messages=limit) if transcript is None: # 最常见的真实原因是权限:读会话历史要 ``im:message`` 系列 + # ``im:message.group_msg``。把它说出来,否则排查会从「模型不会用工具」 # 开始,而那是错的方向。 return ( "COULD NOT READ THIS ROOM. The Feishu API refused or was " "unreachable — most often the app is missing the group-message " "read permission (im:message.group_msg). Tell the employee you " "could not load the room history this time and ask them to " "describe the problem, and say IT can check the assistant's " "Feishu permissions. Do NOT summarise as if the room were empty." ) body = transcript.render() if not body.strip(): return ( "THIS ROOM HAS NO MESSAGES FROM PEOPLE YET (only the assistant's " "own posts). There is nothing to summarise — ask the employee " "what they need." ) head = ( "Only the most recent messages are included; earlier discussion is " "not here.\n" if transcript.truncated else "" ) return f"{head}{body}" __all__ = [ "TOOL_READ_ROOM", "dispatch_read_room", "read_room_tool", "DEFAULT_MAX_MESSAGES", "RoomMessage", "RoomTranscript", "read_room", ]