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 Enum 

13 

14 

15class ChannelType(str, Enum): 

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

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

18 WECOM = "wecom" # 企业微信 

19 FEISHU = "feishu" # 飞书 

20 DINGTALK = "dingtalk" # 钉钉 

21 QQ = "qq" # QQ 机器人 

22 SLACK = "slack" # Slack 

23 DISCORD = "discord" # Discord 

24 TELEGRAM = "telegram" # Telegram 

25 WHATSAPP = "whatsapp" # WhatsApp Business 

26 LINE = "line" # LINE Messaging 

27 WEB = "web" # Web 端 

28 MOBILE = "mobile" # 移动端 

29 

30 

31class MessageType(str, Enum): 

32 """消息类型。""" 

33 TEXT = "text" # 文本 

34 IMAGE = "image" # 图片 

35 VOICE = "voice" # 语音 

36 VIDEO = "video" # 视频 

37 FILE = "file" # 文件 

38 LOCATION = "location" # 位置 

39 LINK = "link" # 链接 

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

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

42 

43 

44@dataclass 

45class ConversationContext: 

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

47 channel: ChannelType 

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

49 session_id: str # AgentOS 内会话 ID 

50 channel_config: dict = field(default_factory=dict) 

51 metadata: dict = field(default_factory=dict) 

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

53 

54 

55@dataclass 

56class ChannelMessage: 

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

58 

59 设计原则: 

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

61 - 渠道特有数据塞入 extra 

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

63 """ 

64 

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

66 channel: ChannelType # 来源渠道 

67 msg_type: MessageType = MessageType.TEXT 

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

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

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

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

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

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

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

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

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

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

78 

79 @property 

80 def is_from_mobile(self) -> bool: 

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

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

83 

84 @property 

85 def is_group_chat(self) -> bool: 

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

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

88 

89 @property 

90 def display_source(self) -> str: 

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

92 labels = { 

93 ChannelType.WECHAT_MP: "微信", 

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

95 ChannelType.FEISHU: "飞书", 

96 ChannelType.DINGTALK: "钉钉", 

97 ChannelType.QQ: "QQ", 

98 ChannelType.WEB: "Web", 

99 ChannelType.MOBILE: "手机", 

100 } 

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

102 

103 def to_dict(self) -> dict: 

104 return { 

105 "msg_id": self.msg_id, 

106 "channel": self.channel.value, 

107 "msg_type": self.msg_type.value, 

108 "content": self.content, 

109 "sender_id": self.sender_id, 

110 "sender_name": self.sender_name, 

111 "timestamp": self.timestamp, 

112 "conversation_id": self.conversation_id, 

113 "session_id": self.session_id, 

114 "media_url": self.media_url, 

115 "extra": self.extra, 

116 } 

117 

118 @classmethod 

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

120 return cls( 

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

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

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

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

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

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

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

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

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

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

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

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

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

134 )