Coverage for agentos/channels/message.py: 0%

1""" 

2AgentOS Channels — 统一消息模型与渠道协议。 

3 

4所有渠道（微信/企业微信/飞书/钉钉/QQ）的消息 

5均归一化为 ChannelMessage，确保 Agent Engine 零差异处理。 

6""" 

7 

8from __future__ import annotations 

9 

10import json 

11import time 

12from dataclasses import dataclass, field 

13from enum import Enum 

14from typing import Optional 

15 

16 

17class ChannelType(str, Enum): 

18 """渠道类型标识。""" 

19 WECHAT_MP = "wechat-mp" # 微信公众号 

20 WECOM = "wecom" # 企业微信 

21 FEISHU = "feishu" # 飞书 

22 DINGTALK = "dingtalk" # 钉钉 

23 QQ = "qq" # QQ 机器人 

24 SLACK = "slack" # Slack 

25 DISCORD = "discord" # Discord 

26 TELEGRAM = "telegram" # Telegram 

27 WHATSAPP = "whatsapp" # WhatsApp Business 

28 LINE = "line" # LINE Messaging 

29 WEB = "web" # Web 端 

30 MOBILE = "mobile" # 移动端 

31 

32 

33class MessageType(str, Enum): 

34 """消息类型。""" 

35 TEXT = "text" # 文本 

36 IMAGE = "image" # 图片 

37 VOICE = "voice" # 语音 

38 VIDEO = "video" # 视频 

39 FILE = "file" # 文件 

40 LOCATION = "location" # 位置 

41 LINK = "link" # 链接 

42 EVENT = "event" # 事件（关注/点击菜单等） 

43 MINIPROGRAM = "miniprogram" # 小程序卡片 

44 

45 

46@dataclass 

47class ConversationContext: 

48 """会话上下文 — 跨渠道统一。""" 

49 channel: ChannelType 

50 user_id: str # 渠道内用户唯一标识 

51 session_id: str # AgentOS 内会话 ID 

52 channel_config: dict = field(default_factory=dict) 

53 metadata: dict = field(default_factory=dict) 

54 history: list[dict] = field(default_factory=list) # 最近 N 轮对话 

55 

56 

57@dataclass 

58class ChannelMessage: 

59 """统一消息模型 — 所有渠道归一化为此格式。 

60 

61 设计原则: 

62 - 字段命名中性，不偏袒任何渠道 

63 - 渠道特有数据塞入 extra 

64 - Agent Engine 只看本模型，不感知渠道差异 

65 """ 

66 

67 msg_id: str # 渠道内消息唯一 ID 

68 channel: ChannelType # 来源渠道 

69 msg_type: MessageType = MessageType.TEXT 

70 content: str = "" # 文本内容（image/voice 等为 media_url） 

71 sender_id: str = "" # 发送者 ID 

72 sender_name: str = "" # 发送者昵称 

73 timestamp: float = field(default_factory=time.time) 

74 conversation_id: str = "" # 渠道内会话/群 ID 

75 reply_token: str = "" # 渠道回复令牌（用于被动回复） 

76 session_id: str = "" # AgentOS 会话 ID 

77 media_url: str = "" # 多媒体 URL 

78 media_id: str = "" # 渠道媒体 ID（用于下载） 

79 extra: dict = field(default_factory=dict) # 渠道特有字段 

80 

81 @property 

82 def is_from_mobile(self) -> bool: 

83 """是否来自移动端渠道。""" 

84 return self.channel in (ChannelType.MOBILE, ChannelType.WECHAT_MP, ChannelType.QQ) 

85 

86 @property 

87 def is_group_chat(self) -> bool: 

88 """是否群聊消息。""" 

89 return self.extra.get("is_group", False) 

90 

91 @property 

92 def display_source(self) -> str: 

93 """人类可读的来源标识。""" 

94 labels = { 

95 ChannelType.WECHAT_MP: "微信", 

96 ChannelType.WECOM: "企业微信", 

97 ChannelType.FEISHU: "飞书", 

98 ChannelType.DINGTALK: "钉钉", 

99 ChannelType.QQ: "QQ", 

100 ChannelType.WEB: "Web", 

101 ChannelType.MOBILE: "手机", 

102 } 

103 return labels.get(self.channel, self.channel.value) 

104 

105 def to_dict(self) -> dict: 

106 return { 

107 "msg_id": self.msg_id, 

108 "channel": self.channel.value, 

109 "msg_type": self.msg_type.value, 

110 "content": self.content, 

111 "sender_id": self.sender_id, 

112 "sender_name": self.sender_name, 

113 "timestamp": self.timestamp, 

114 "conversation_id": self.conversation_id, 

115 "session_id": self.session_id, 

116 "media_url": self.media_url, 

117 "extra": self.extra, 

118 } 

119 

120 @classmethod 

121 def from_dict(cls, d: dict) -> "ChannelMessage": 

122 return cls( 

123 msg_id=d.get("msg_id", ""), 

124 channel=ChannelType(d.get("channel", "web")), 

125 msg_type=MessageType(d.get("msg_type", "text")), 

126 content=d.get("content", ""), 

127 sender_id=d.get("sender_id", ""), 

128 sender_name=d.get("sender_name", ""), 

129 timestamp=d.get("timestamp", time.time()), 

130 conversation_id=d.get("conversation_id", ""), 

131 reply_token=d.get("reply_token", ""), 

132 session_id=d.get("session_id", ""), 

133 media_url=d.get("media_url", ""), 

134 media_id=d.get("media_id", ""), 

135 extra=d.get("extra", {}), 

136 )