Agent消息降级矩阵:从卡片到纯文本的自适应

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

## 一、被忽视的运维断层

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

| 渠道 | 卡片 | 按钮 | Markdown | 最大长度 |

|------|------|------|----------|----------|

| 飞书 | 支持 | 支持 | 完整 | 30KB |

| 企微 | 部分支持 | 不支持 | 基础 | 4KB |

| Console | 不支持 | 不支持 | 完整 | 无限 |

| SMS | 不支持 | 不支持 | 不支持 | 500字符 |

问题来了:一个含卡片+3个按钮+完整Markdown的告警消息,要发到只支持500字符纯文本的SMS,怎么办?

大多数Agent系统的做法是**直接报错或截断**。结果是:关键告警在跨渠道时丢失,运维人员毫不知情。

agent-ops-toolkit(github.com/yuzhaopeng-up/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四个部分。降级到纯文本时,不是简单拼接,而是**按语义结构化拆解**:

```python

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降级为内联编号文本:

```python

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的映射表:

```json

{

  "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字符),执行分片:

```python

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都不支持时,执行三级剥离:

```python

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中都会记录降级轨迹:

```json

{

  "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`历史记录统计哪些降级可以取消,量化渠道升级的收益

## 五、跨渠道路由器的降级触发逻辑

降级不是手动触发的,而是**路由器自动检测目标渠道能力后触发**:

```python

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`列表),消息发送时自动对比能力差异,按需触发降级。整个降级链对业务代码透明——业务层只管发飞书卡片,路由器负责适配所有渠道。

## 六、快速上手

```bash

git clone agent-ops-toolkit(github.com/yuzhaopeng-up/agent-ops-toolkit)

cd agent-ops-toolkit

```

```python

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开源生态

| 仓库 | 定位 | GitHub |

|------|------|--------|

| financial-ai-skills | 金融AI技能库:104个场景纯Python实现 | github.com/yuzhaopeng-up/financial-ai-skills |

| teleagent-skills | 5个通用业务Skill:4-Phase编排+规则参数化 | github.com/yuzhaopeng-up/teleagent-skills |

| agent-cluster-comm | 多Agent集群5层通信架构 | github.com/yuzhaopeng-up/agent-cluster-comm |

| skill-framework | Skill治理框架:L0-L4分类+YAML模板 | github.com/yuzhaopeng-up/skill-framework |

| fintech-h5-demos | 57个零依赖金融H5演示 | github.com/yuzhaopeng-up/fintech-h5-demos |

| soe-compliant-office | 17个央国企合规办公Skill | github.com/yuzhaopeng-up/soe-compliant-office |

| regulated-rag | 零依赖RAG工具包:BM25+TF-IDF+RRF | github.com/yuzhaopeng-up/regulated-rag |

| **agent-ops-toolkit** | **企业级Agent运维基础设施:降级+告警+工作流** | **https://github.com/yuzhaopeng-up/agent-ops-toolkit** |

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

Star agent-ops-toolkit(github.com/yuzhaopeng-up/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!

©著作权归作者所有,转载或内容合作请联系作者
【社区内容提示】社区部分内容疑似由AI辅助生成,浏览时请结合常识与多方信息审慎甄别。
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

友情链接更多精彩内容