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 time 

11from dataclasses import dataclass, field 

12from enum import StrEnum 

13 

14 

15class ChannelType(StrEnum): 

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

17 

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

19 WECOM = "wecom" # 企业微信 

20 FEISHU = "feishu" # 飞书 

21 DINGTALK = "dingtalk" # 钉钉 

22 QQ = "qq" # QQ 机器人 

23 SLACK = "slack" # Slack 

24 DISCORD = "discord" # Discord 

25 TELEGRAM = "telegram" # Telegram 

26 WHATSAPP = "whatsapp" # WhatsApp Business 

27 LINE = "line" # LINE Messaging 

28 WEB = "web" # Web 端 

29 MOBILE = "mobile" # 移动端 

30 

31 

32class MessageType(StrEnum): 

33 """消息类型。""" 

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 

50 channel: ChannelType 

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

52 session_id: str # AgentOS 内会话 ID 

53 channel_config: dict = field(default_factory=dict) 

54 metadata: dict = field(default_factory=dict) 

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

56 

57 

58@dataclass 

59class ChannelMessage: 

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

61 

62 设计原则: 

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

64 - 渠道特有数据塞入 extra 

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

66 """ 

67 

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

69 channel: ChannelType # 来源渠道 

70 msg_type: MessageType = MessageType.TEXT 

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

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

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

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

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

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

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

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

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

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

81 

82 @property 

83 def is_from_mobile(self) -> bool: 

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

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

86 

87 @property 

88 def is_group_chat(self) -> bool: 

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

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

91 

92 @property 

93 def display_source(self) -> str: 

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

95 labels = { 

96 ChannelType.WECHAT_MP: "微信", 

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

98 ChannelType.FEISHU: "飞书", 

99 ChannelType.DINGTALK: "钉钉", 

100 ChannelType.QQ: "QQ", 

101 ChannelType.WEB: "Web", 

102 ChannelType.MOBILE: "手机", 

103 } 

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

105 

106 def to_dict(self) -> dict: 

107 return { 

108 "msg_id": self.msg_id, 

109 "channel": self.channel.value, 

110 "msg_type": self.msg_type.value, 

111 "content": self.content, 

112 "sender_id": self.sender_id, 

113 "sender_name": self.sender_name, 

114 "timestamp": self.timestamp, 

115 "conversation_id": self.conversation_id, 

116 "session_id": self.session_id, 

117 "media_url": self.media_url, 

118 "extra": self.extra, 

119 } 

120 

121 @classmethod 

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

123 return cls( 

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

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

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

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

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

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

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

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

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

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

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

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

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

137 )