当Agent告警卡片发到只支持纯文本的SMS渠道时,是直接报错还是自动降级?大多数Agent系统的选择是直接报错,结果就是用户错过了关键告警。这个场景听起来很熟悉,但问题远比想象中更普遍。
一、被忽视的运维断层
Agent系统通常对接多个消息渠道:飞书卡片、企业微信图文、Web Console、SMS信息。每个渠道的消息格式能力差异巨大:

| 渠道 | 卡片 | 按钮 | Markdown | 最大长度 |
|---|---|---|---|---|
| 飞书 | 支持 | 支持 | 完整 | 30KB |
| 企微 | 部分支持 | 不支持 | 基础 | 4KB |
| Console | 不支持 | 不支持 | 完整 | 无限 |
| SMS | 不支持 | 不支持 | 不支持 | 500字符 |
问题来了:一个含卡片、三个按钮、完整Markdown的告警消息,要发到只支持500字符纯文本的SMS,怎么办?
大多数Agent系统的做法是直接报错或截断。结果是:关键告警在跨渠道时丢失,运维人员毫不知情。
agent-ops-toolkit给出了一个系统化答案——4层自动降级矩阵。
二、4层降级策略总览
原始消息(卡片+按钮+长文本+完整Markdown)
│
▼
┌─────────────────────────────────────────────┐
│ Layer 1: card → text │
│ 卡片拆解为 标题 + 正文 + 链接 + 编号操作 │
└────────────────────┬────────────────────────┘
▼
┌─────────────────────────────────────────────┐
│ Layer 2: buttons → 内联编号列表 │
│ "确认"/"忽略"/"升级" → [1]确认 [2]忽略 [3]升级│
└────────────────────┬────────────────────────┘
▼
┌─────────────────────────────────────────────┐
│ Layer 3: 长文本分片 │
│ 超过max_text_length → 拆为多条 [1/3] [2/3]... │
└────────────────────┬────────────────────────┘
▼
┌─────────────────────────────────────────────┐
│ Layer 4: markdown逐级剥离 │
│ full → basic(保留加粗标题) → none(纯文本) │
└─────────────────────────────────────────────┘
每一层降级都是可选且可追溯的:降级后的消息metadata中记录了degraded_from字段,标注这条消息经历了哪些降级操作,方便事后审计和渠道能力升级时回溯优化。
三、逐层解析
3.1 Layer 1:card转text——结构化拆解
飞书卡片的JSON结构包含header、content、link、actions四个部分。降级到纯文本时,不是简单拼接,而是按语义结构化拆解:
def _card_to_text(card_msg):
parts = []
# 1. 标题(从card.header提取)
if card_msg.get('header'):
parts.append(f"【{card_msg['header']['title']}】")
# 2. 正文(从card.elements提取文本块)
for element in card_msg.get('elements', []):
if element['type'] == 'text':
parts.append(element['content'])
# 3. 链接(从card.elements提取链接块)
for element in card_msg.get('elements', []):
if element['type'] == 'link':
parts.append(f"详情: {element['url']}")
# 4. 编号操作(从card.actions提取按钮)
for i, action in enumerate(card_msg.get('actions', [])):
parts.append(f"[{i+1}]{action['text']}")
return 'n'.join(parts)
降级前后对比:
原始飞书卡片:
┌─────────────────────────────┐
│ ⚠ 告警:服务器CPU超过90% │ ← header
│ 主机: prod-web-01 │ ← element:text
│ CPU: 92% (阈值90%) │ ← element:text
│ 持续: 15分钟 │ ← element:text
│ [确认] [忽略] [升级] │ ← actions (3个按钮)
└─────────────────────────────┘
降级后纯文本:
【告警:服务器CPU超过90%】
主机: prod-web-01
CPU: 92% (阈值90%)
持续: 15分钟
[1]确认 [2]忽略 [3]升级
关键设计:按钮被转换为编号列表,而非丢弃。这样即使用户通过SMS接收,也能回复"1"来触发"确认"操作——降级不降功能。
3.2 Layer 2:buttons转内联编号列表
当目标渠道完全不支持交互按钮时(如SMS、邮件),buttons降级为内联编号文本:
def _buttons_to_inline(actions, max_per_line=5):
items = []
for i, action in enumerate(actions):
items.append(f"[{i+1}]{action['text']}")
# 每5个按钮一行,避免超长
lines = []
for i in range(0, len(items), max_per_line):
lines.append(' '.join(items[i:i+max_per_line]))
return 'n'.join(lines)
编号与回调的映射:每个编号对应原始action的value字段。用户通过SMS回复"1"时,Agent系统的消息回执解析器识别编号,映射回原始action并触发回调。这要求降级器在metadata中保留编号→action的映射表:
{
"degraded_from": "buttons_to_inline",
"action_map": {
"1": {"text": "确认", "value": "acknowledge"},
"2": {"text": "忽略", "value": "dismiss"},
"3": {"text": "升级", "value": "escalate"}
}
}
3.3 Layer 3:长文本分片
当降级后的文本仍超过渠道的max_text_length(如SMS的500字符),执行分片:
def _split_long_text(text, max_length=500):
# 预留20字符给 [i/N] 标记
effective_max = max_length - 20
chunks = []
for i in range(0, len(text), effective_max):
chunk = text[i:i+effective_max]
chunks.append(chunk)
# 给每个分片追加 [i/N] 标记
total = len(chunks)
result = []
for i, chunk in enumerate(chunks):
result.append(f"{chunk} [{i+1}/{total}]")
return result
预留20字符的设计:[99/99]最坏情况占6字符,但考虑到换行符和空格,预留20字符是安全边界。这避免了"分片后加上标记又超长"的递归问题。
幂等键追溯:每个分片消息的幂等键自动追加:p{i}后缀。原始消息幂等键为alert-001,分片后变为alert-001:p1、alert-001:p2、alert-001:p3。这样在消息回执和审计日志中,可以追溯到完整的分片链路,避免重复发送或遗漏。
3.4 Layer 4:markdown逐级剥离
当渠道连基础Markdown都不支持时,执行三级剥离:
def _strip_markdown(text, level='basic'):
if level == 'full':
return text # 完整Markdown,不处理
if level == 'basic':
# 保留加粗和标题,剥离其他语法
text = re.sub(r'`([^`]+)`', r'1', text) # 去行内代码
text = re.sub(r'!\[.*?\]\(.*?\)', '[图片]', text) # 图片占位
text = re.sub(r'\[([^]]+)\]\([^)]+\)', r'1', text) # 链接保留文字
return text
if level == 'none':
# 完全剥离为纯文本
text = re.sub(r'[#*`>_~-]', '', text) # 去所有Markdown符号
text = re.sub(r'\[([^]]+)\]\([^)]+\)', r'1', text) # 链接保留文字
return text
三级映射表:
| 级别 | 保留 | 剥离 |
|---|---|---|
| full | 完整Markdown | 无 |
| basic | 加粗、标题、列表结构 | 代码块、图片、链接URL |
| none | 纯文本 | 所有Markdown语法 |
为什么不全剥离到none:因为加粗和标题对告警可读性影响最大。**严重告警**在basic级别保留,用户能快速识别严重程度;如果在none级别变成严重告警,严重程度就模糊了。
四、审计轨迹:degraded_from设计
每条降级后的消息,metadata中都会记录降级轨迹:
{
"message_id": "alert-001:p1",
"original_format": "feishu_card",
"target_channel": "sms",
"degraded_from": ["card_to_text", "buttons_to_inline", "split_long_text", "strip_markdown:basic"],
"original_length": 1850,
"final_length": 480,
"split_count": 4,
"timestamp": "2026-07-20T14:30:00Z"
}
degraded_from是一个有序列表,记录了降级的执行顺序。这在两个场景特别有用:
- 事后排查:用户反馈"SMS收到的告警格式乱了",运维通过
degraded_from定位是哪一层降级出问题(是card拆解错了,还是分片截断了关键信息) - 能力升级:当SMS渠道升级支持Markdown时,可以通过
degraded_from历史记录统计哪些降级可以取消,量化渠道升级的收益
五、跨渠道路由器的降级触发逻辑
降级不是手动触发的,而是路由器自动检测目标渠道能力后触发:
class CrossChannelRouter:
def __init__(self):
self.adapters = {
'feishu': FeishuAdapter(supports=['card', 'buttons', 'markdown:full']),
'wecom': WeComAdapter(supports=['markdown:basic']),
'sms': SMSAdapter(supports=[]), # 纯文本,无格式
}
def route(self, message, target_channel):
adapter = self.adapters[target_channel]
capabilities = adapter.supports
# 按需降级
if 'card' not in capabilities and message.type == 'card':
message = self._degrade(message, 'card_to_text')
if 'buttons' not in capabilities and message.actions:
message = self._degrade(message, 'buttons_to_inline')
if len(message.text) > adapter.max_length:
message = self._degrade(message, 'split_long_text',
max_length=adapter.max_length)
md_level = self._get_md_level(capabilities)
if md_level < message.markdown_level:
message = self._degrade(message, 'strip_markdown', level=md_level)
return adapter.send(message)
路由器维护每个渠道的能力声明(supports列表),消息发送时自动对比能力差异,按需触发降级。整个降级链对业务代码透明——业务层只管发飞书卡片,路由器负责适配所有渠道。
六、快速上手
git clone https://github.com/yuzhaopeng-up/agent-ops-toolkit.git
cd agent-ops-toolkit
from agent_ops import CrossChannelRouter
router = CrossChannelRouter()
# 业务层只管发标准卡片
alert_card = {
"header": {"title": "CPU告警"},
"elements": [
{"type": "text", "content": "主机: prod-web-01"},
{"type": "text", "content": "CPU: 92%"},
],
"actions": [
{"text": "确认", "value": "ack"},
{"text": "升级", "value": "escalate"},
]
}
# 路由器自动降级适配各渠道
router.route(alert_card, 'feishu') # 原样发送卡片
router.route(alert_card, 'wecom') # 降级为基础Markdown+编号操作
router.route(alert_card, 'sms') # 降级为纯文本+分片
七、设计启示
消息降级看似是工程细节,实际反映了Agent系统的成熟度。几个值得借鉴的设计思想:
降级是保底而非降级:宁可给用户一个格式简化但信息完整的消息,也不要因为格式不兼容就丢弃消息。告警丢失的代价远大于告警格式不好看。
审计轨迹先行:每一步降级都记录degraded_from,不是为了好看,而是为了事后能定位问题和量化渠道升级收益。
幂等键贯穿分片链路::p{i}后缀设计看似简单,但解决了分布式消息系统中最难的消息追溯问题——哪条分片对应原始消息的哪一部分。
能力声明驱动:路由器通过渠道的supports声明自动决策降级,而非硬编码if-else。新增渠道只需声明能力,降级链自动适配。
