金融Agent通信：消息降级矩阵与容错设计

当你的Agent告警卡片发到只支持纯文本的SMS渠道时，是直接报错还是自动降级？大多数Agent系统的答案是前者——然后用户错过了关键告警。

一、被忽视的运维断层

Agent系统通常对接多个消息渠道：飞书卡片、企业企微图文、Web Console、SMS短信。每个渠道的消息格式能力差异巨大：

问题来了：一个含卡片+3个按钮+完整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

三级映射表：

为什么不全剥离到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是一个有序列表，记录了降级的执行顺序。这在两个场景特别有用：

1. 事后排查：用户反馈"SMS收到的告警格式乱了"，运维通过degraded_from定位是哪一层降级出问题（是card拆解错了，还是分片截断了关键信息）
2. 能力升级：当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。新增渠道只需声明能力，降级链自动适配。

Agent Skills开源生态

如果你的Agent系统对接了多个消息渠道，别让格式不兼容吃掉你的关键告警。4层降级矩阵，30行核心代码，让每条消息都安全到达。

Star agent-ops-toolkit 一起把Agent运维做到生产级。

觉得有用？给个 Star 支持一下！你的 Star 是我们持续开源的最大动力。

更多开源 Agent Skills：financial-ai-skills · teleagent-skills · regulated-rag · fintech-h5-demos —— 全部在 github.com/yuzhaopeng-up，欢迎 Star/Fork/PR！