加载中

🦐 「HermesAgent」飞书消息格式适配器 + 表格渲染升级方案

白木 2026年8月7日 2026年8月7日 170

> 让 markdown 表格在飞书 DM 里渲染成原生表格，同时不破坏粗体/标题/列表等其他 markdown 元素的渲染

📌 背景

Hermes 默认的飞书适配器只能发送 text 或 post 类型消息。post 类型支持 markdown 标签（粗体/斜体/代码/标题/列表/引用/链接），但不支持表格——markdown 表格会显示成原始的 |---|---| 字符。

本方案在不破坏现有 markdown 渲染的前提下，给飞书适配器增加表格 → CardKit v2 table 组件的能力，并保证：

标题/粗体/列表等 markdown 元素继续走 post 路径正常渲染
含表格的消息自动拆分成多条消息，每条表格独立成 CardKit v2 卡片
多表格消息每个表格独立卡片，中间的标题/正文不被吞

🎯 目标

输入	输出	类型
—	—	—
`## 标题` + 纯文本	1 条 post 消息	正常渲染
`粗体` + 文本	1 条 post 消息	粗体渲染
表格	1 条 CardKit v2 卡片	表格渲染
`## 标题` + 表格	2 条消息：post + 卡片	各自独立
`## 标题` + 表格 + `## 标题` + 表格	4 条消息：post + 卡片 + post + 卡片	全部独立

📚 前置条件

1. 飞书开放平台

已创建企业自建应用
已开启机器人能力
已配置事件订阅 + 消息接收
已有 app_id 和 app_secret
应用具备 im:message + im:message:send_as_bot 权限

2. Hermes 版本

Hermes ≥ 0.x（已实现 interactive msg_type 发送路径）
飞书适配器文件：/opt/hermes/gateway/platforms/feishu.py

3. Python 依赖

lark-oapi（Hermes 内置）
标准库 re / json（无需额外安装）

🔬 核心原理

三种飞书消息类型对比

类型	msg_type	支持 markdown	支持表格	适用场景
—	—	—	—	—
纯文本	`text`	❌	❌	简单文本（不带任何格式）
富文本 post	`post`	✅ 粗体/斜体/代码/标题/列表/引用/链接/删除线	❌ 表格	大部分 markdown 内容
互动卡片	`interactive`	❌（用 lark_md 组件）	✅ CardKit v2 table	含表格的消息

关键发现

飞书没有“自动识别 markdown”的开关——必须显式选 msg_type。所以：

含 markdown 但不含表格 → 走 post
含表格 → 走 interactive（CardKit v2）
两者都不能混（一条消息只能一个 msg_type）

CardKit v2 table 组件 schema

{
  "schema": "2.0",
  "config": {"wide_screen_mode": true},
  "body": {
    "elements": [
      {
        "tag": "table",
        "columns": [
          {"name": "col_0", "display_name": "列名1", "data_type": "text", "width": "auto"},
          {"name": "col_1", "display_name": "列名2", "data_type": "text", "width": "auto"}
        ],
        "rows": [
          {"col_0": "值1", "col_1": "值2"},
          {"col_0": "值3", "col_1": "值4"}
        ],
        "header_style": {
          "bold": true,
          "text_align": "left",
          "text_size": "normal",
          "background_style": "none",
          "text_color": "default",
          "lines": 1
        }
      }
    ]
  }
}

要点：

schema: "2.0" 是 CardKit v2 标记（v1 不支持 table 组件）
wide_screen_mode: true 让卡片在 PC 端铺满宽度
columns 用 name（程序内引用）+ display_name（展示）
rows 用 dict 引用 column 的 name
header_style.bold: true 让表头加粗

📁 文件改动一览

只需改动一个文件：feishu.py

类型	内容	行号参考
—	—	—
新增	`_MARKDOWN_TABLE_RE` 正则	~158
新增	`_strip_inline_markdown` 函数	~171
新增	`_parse_markdown_table` 函数	~183
新增	`_convert_table_to_cardkit_payload` 函数	~228
新增	`_extract_first_table` 函数	~266
新增	`_build_table_cardkit_payload` 函数	~293
新增	`_split_outbound_payloads` 方法	~4370
修改	`_send_message` 双层 for-loop	~1910

🔧 完整代码实现

1. 正则常量（line 158 附近）

# Detect markdown tables: a line starting with | followed by a separator line.
# Feishu post-type 'md' elements do not render tables, so we force text mode.
_MARKDOWN_TABLE_RE = re.compile(r"^\|.*\|\n\|[-|: ]+\|", re.MULTILINE)

说明：

^\|.*\| — 表格起始行（以 | 开头和结尾）
\n\|[-|: ]+\| — 分隔行（|---|---|、|:---:| 等）
re.MULTILINE — 让 ^ 匹配每行开头

2. `_strip_inline_markdown`（line 171）

def _strip_inline_markdown(text: str) -> str:
    """Remove inline markdown markers so cell text renders cleanly in CardKit v2."""
    # Strip **bold**, *italic*, `code`, ~~strike~~, <u>...</u>, [link](url)
    text = re.sub(r"\*\*(.+?)\*\*", r"\1", text)
    text = re.sub(r"\*(.+?)\*", r"\1", text)
    text = re.sub(r"~~(.+?)~~", r"\1", text)
    text = re.sub(r"<u>(.+?)</u>", r"\1", text)
    text = re.sub(r"`([^`\n]+)`", r"\1", text)
    text = re.sub(r"\[([^\]]+)\]\(([^)]+)\)", r"\1", text)
    return text.strip()

说明：

CardKit v2 table 的 cell 内容是纯文本，不支持 markdown 标签
所以表格里的粗体、斜体、代码块、链接都要先剥掉
顺序很重要：避免被星符号规则先吃掉

3. _parse_markdown_table（line 183）

def _parse_markdown_table(table_text: str) -> dict:
    """Parse a markdown table block into columns + rows dicts.

    Input format (assumed already trimmed of surrounding text):
        | col1 | col2 | col3 |
        | --- | --- | --- |
        | a | b | c |
        | d | e | f |
    """
    lines = [l.strip() for l in table_text.strip().split("\n") if l.strip()]
    if len(lines) < 2:
        return {"columns": [], "rows": []}

    def split_row(line: str) -> list:
        # Strip leading/trailing pipes, split by |
        inner = line.strip()
        if inner.startswith("|"):
            inner = inner[1:]
        if inner.endswith("|"):
            inner = inner[:-1]
        return [c.strip() for c in inner.split("|")]

    header_cells = split_row(lines[0])
    # Skip separator line (lines[1])
    body_rows = [split_row(l) for l in lines[2:]]

    columns = [
        {
            "name": f"col_{i}",
            "display_name": _strip_inline_markdown(h),
            "data_type": "text",
            "width": "auto",
        }
        for i, h in enumerate(header_cells)
    ]
    rows = []
    for r in body_rows:
        row_dict = {}
        for i, val in enumerate(r):
            if i < len(columns):
                row_dict[columns[i]["name"]] = _strip_inline_markdown(val)
        rows.append(row_dict)
    return {"columns": columns, "rows": rows}

说明：

输入：原始 markdown 表格块（含表头、分隔行、数据行）
输出：{"columns": [...], "rows": [...]} 供 CardKit v2 使用
lines[1] 永远是分隔行（|---|---|），跳过不解析
列名是 col_0、col_1… 程序内部引用
display_name 是用户看到的表头

4. `_convert_table_to_cardkit_payload`（line 228）

def _convert_table_to_cardkit_payload(table_text: str, surrounding_text: str = "") -> tuple[str, str]:
    """Build a CardKit v2 interactive payload from a markdown table.

    Returns (msg_type, payload_json_str).
    """
    parsed = _parse_markdown_table(table_text)
    card: dict = {
        "schema": "2.0",
        "config": {"wide_screen_mode": True},
        "body": {"elements": []},
    }
    # Optional surrounding text as a div first
    if surrounding_text.strip():
        card["body"]["elements"].append(
            {
                "tag": "div",
                "text": {"tag": "plain_text", "content": surrounding_text.strip()},
            }
        )
    # Table element
    card["body"]["elements"].append(
        {
            "tag": "table",
            "columns": parsed["columns"],
            "rows": parsed["rows"],
            "header_style": {
                "bold": True,
                "text_align": "left",
                "text_size": "normal",
                "background_style": "none",
                "text_color": "default",
                "lines": 1,
            },
        }
    )
    return "interactive", json.dumps(card, ensure_ascii=False)

说明：

顶层入口，构造 CardKit v2 卡片
surrounding_text 是表格前后的文本（默认传空字符串，因为我们要拆分发送）
返回 (msg_type, payload) 元组，只能返回一条消息

5. `_extract_first_table`（line 266）

def _extract_first_table(content: str) -> tuple[str, str, str] | None:
    """Find first markdown table block; return (pre_text, table_text, post_text) or None."""
    lines = content.split("\n")
    table_start = None
    for i, line in enumerate(lines):
        stripped = line.strip()
        if stripped.startswith("|") and i + 1 < len(lines):
            next_line = lines[i + 1].strip()
            # Separator line: |---|---| or |:---:| etc.
            if re.match(r"^\|[-:| ]+\|$", next_line):
                table_start = i
                break
    if table_start is None:
        return None
    # Find end of table (consecutive lines starting with |)
    table_end = table_start
    for j in range(table_start, len(lines)):
        if lines[j].strip().startswith("|"):
            table_end = j
        else:
            break
    pre_text = "\n".join(lines[:table_start]).rstrip()
    table_text = "\n".join(lines[table_start:table_end + 1])
    post_text = "\n".join(lines[table_end + 1:]).strip()
    return pre_text, table_text, post_text

说明：

输入：完整 markdown 文本
输出：(前文本, 表格块, 后文本) 或 None
表格块包括：表头行 + 分隔行 + 所有数据行（连续的 | 开头行）
表格前后文本独立保留，供后续拆分发送

6. `_build_table_cardkit_payload`（line 293）

def _build_table_cardkit_payload(content: str) -> tuple[str, str]:
    """Top-level: detect markdown table → return CardKit v2 payload."""
    extracted = _extract_first_table(content)
    if not extracted:
        text_payload = {"text": content}
        return "text", json.dumps(text_payload, ensure_ascii=False)
    pre_text, table_text, _post_text = extracted
    return _convert_table_to_cardkit_payload(table_text, pre_text)

说明：

_build_outbound_payload 检测到表格时调用
把 pre_text 当 surrounding_text 塞进卡片（这个会被 _split_outbound_payloads 替代）
如果没找到表格，回退到 text 路径

7. `_split_outbound_payloads`（line 4370，新增方法）

def _split_outbound_payloads(self, content: str) -> list[tuple[str, str]]:
    """Split content into a list of outbound (msg_type, payload) tuples.

    When content contains one or more markdown tables interleaved with
    markdown elements (heading/bold/list/...), split it so each table
    is sent as its own CardKit v2 interactive card and the surrounding
    segments are sent separately as post/text. This avoids mixing
    markdown and CardKit table in a single message (which would break
    rendering). Recurses for multi-table messages so every table gets
    its own card and no markdown segment is swallowed by a card.
    Returns a single-element list when no split is needed.
    """
    if not _MARKDOWN_TABLE_RE.search(content):
        return [self._build_outbound_payload(content)]
    out: list[tuple[str, str]] = []
    remaining = content
    while True:
        if not _MARKDOWN_TABLE_RE.search(remaining):
            # No more tables — emit the rest as a single payload (if non-empty).
            stripped = remaining.strip()
            if stripped:
                out.append(self._build_outbound_payload(remaining))
            break
        extracted = _extract_first_table(remaining)
        if not extracted:
            # Shouldn't happen given the regex check above, but bail safely.
            stripped = remaining.strip()
            if stripped:
                out.append(self._build_outbound_payload(remaining))
            break
        pre_text, table_text, post_text = extracted
        if pre_text.strip():
            out.append(self._build_outbound_payload(pre_text))
        # Always send table as its own CardKit v2 card.
        out.append(_build_table_cardkit_payload(table_text))
        # Continue with whatever follows the table (may contain more tables).
        remaining = post_text
    if not out:
        return [self._build_outbound_payload(content)]
    return out

说明：

这是本方案最核心的方法
无表格 → 直接返回 [单条消息]（保持原行为）
有表格 → while 循环处理：

1. 提取第一个表格 + pre_text + post_text 2. pre_text → post 消息 3. table → CardKit v2 卡片 4. post_text 当作 remaining 继续循环

多表格消息自动递归拆分

8. `_send_message` 修改（line 1910）

把原来的：

try:
    for chunk in chunks:
        msg_type, payload = self._build_outbound_payload(chunk)
        try:
            response = await self._feishu_send_with_retry(
                chat_id=chat_id,
                msg_type=msg_type,
                payload=payload,
                reply_to=reply_to,
                metadata=metadata,
            )
        except Exception as exc:
            if msg_type != "post" or not _POST_CONTENT_INVALID_RE.search(str(exc)):
                raise
            logger.warning("[Feishu] Invalid post payload rejected by API; falling back to plain text")
            response = await self._feishu_send_with_retry(
                chat_id=chat_id,
                msg_type="text",
                payload=json.dumps({"text": _strip_markdown_to_plain_text(chunk)}, ensure_ascii=False),
                reply_to=reply_to,
                metadata=metadata,
            )
        if (
            msg_type == "post"
            and not self._response_succeeded(response)
            and _POST_CONTENT_INVALID_RE.search(str(getattr(response, "msg", "") or ""))
        ):
            logger.warning("[Feishu] Post payload rejected by API response; falling back to plain text")
            response = await self._feishu_send_with_retry(
                chat_id=chat_id,
                msg_type="text",
                payload=json.dumps({"text": _strip_markdown_to_plain_text(chunk)}, ensure_ascii=False),
                reply_to=reply_to,
                metadata=metadata,
            )
        last_response = response

改成：

try:
    for chunk in chunks:
        for msg_type, payload in self._split_outbound_payloads(chunk):
            try:
                response = await self._feishu_send_with_retry(
                    chat_id=chat_id,
                    msg_type=msg_type,
                    payload=payload,
                    reply_to=reply_to,
                    metadata=metadata,
                )
            except Exception as exc:
                if msg_type != "post" or not _POST_CONTENT_INVALID_RE.search(str(exc)):
                    raise
                logger.warning("[Feishu] Invalid post payload rejected by API; falling back to plain text")
                response = await self._feishu_send_with_retry(
                    chat_id=chat_id,
                    msg_type="text",
                    payload=json.dumps({"text": _strip_markdown_to_plain_text(chunk)}, ensure_ascii=False),
                    reply_to=reply_to,
                    metadata=metadata,
                )
            if (
                msg_type == "post"
                and not self._response_succeeded(response)
                and _POST_CONTENT_INVALID_RE.search(str(getattr(response, "msg", "") or ""))
            ):
                logger.warning("[Feishu] Post payload rejected by API response; falling back to plain text")
                response = await self._feishu_send_with_retry(
                    chat_id=chat_id,
                    msg_type="text",
                    payload=json.dumps({"text": _strip_markdown_to_plain_text(chunk)}, ensure_ascii=False),
                    reply_to=reply_to,
                    metadata=metadata,
                )
            last_response = response

说明：

内层 for 循环由拆分函数产生的多条消息顺序发送
异常处理块移到内层 try，保证单条消息失败不影响其他消息
last_response 记录最后一条消息的响应，供最终结果判断

🚀 集成步骤

Step 1：备份原文件

cp /opt/hermes/gateway/platforms/feishu.py /opt/hermes/gateway/platforms/feishu.py.bak.$(date +%Y%m%d)

Step 2：新增代码

按行号顺序插入：

line 158 附近新增 _MARKDOWN_TABLE_RE（如果已有就跳过）
line 170 附近新增 CardKit v2 注释 + 6 个新函数（_strip_inline_markdown ~ _build_table_cardkit_payload）
line 4370 附近新增 _split_outbound_payloads 方法

Step 3：修改 `_send_message`

把 line 1910 的 for chunk in chunks: 内层改成双层 for-loop（参考上面”修改”段落）。

Step 4：语法检查

python3 -m py_compile /opt/hermes/gateway/platforms/feishu.py

无输出 = 通过，有 SyntaxError = 检查缩进。

Step 5：重启 gateway

# Docker/Podman 环境下
kill <gateway_pid>
sleep 3
ps aux | grep "hermes gateway" | grep -v grep  # 确认 PID 重启
curl -s -o /dev/null -w "Dashboard HTTP %{http_code}\n" http://127.0.0.1:9119/

Step 6：验证 dashboard

返回 200 = gateway 重启成功，飞书适配器已加载新代码。

✅ 测试用例

用例 1：纯文本（无 markdown）

输入：

你好，世界

期望：1 条 post 消息，正常显示「你好，世界」

用例 2：粗体 + 斜体

输入：

这是 **粗体** 和 *斜体* 测试

期望：1 条 post 消息，粗体/斜体正常渲染

用例 3：二级标题

输入：

## 测试标题

期望：1 条 post 消息，标题正常渲染

用例 4：纯表格

输入：

| 名称 | 评分 |
|---|---|
| 赫尔墨斯 5.0 | 95 |
| 赫尔墨斯 | 93 |

期望：1 条 CardKit v2 卡片，表格正常渲染

用例 5：标题 + 表格

输入：

## 清酒评分
下面是表格：

| 名称 | 评分 |
|---|---|
| 赫尔墨斯 5.0 | 95 |

期望：3 条独立消息

post：## 清酒评分（标题渲染）
post：下面是表格：（纯文本）
CardKit v2 卡片：表格渲染

用例 6：多表格（关键测试）

输入：

## 第一组
| 名称 | 评分 |
|---|---|
| 赫尔墨斯 | 95 |

## 第二组
| 名称 | 评分 |
|---|---|
| 赫尔墨斯 | 98 |

期望：4 条独立消息

post：## 第一组（标题渲染）
CardKit v2 卡片：表格 1
post：## 第二组（标题渲染）
CardKit v2 卡片：表格 2

关键：第二个表格前的 ## 第二组 必须独立成 post，不能和表格 2 合并。

🆘 回滚预案

如果新代码导致 gateway 启动失败 / 飞书 DM 断连：

# 1. 停止 gateway
kill <gateway_pid>

# 2. 回滚代码
cp /opt/hermes/gateway/platforms/feishu.py.bak.$(date +%Y%m%d) /opt/hermes/gateway/platforms/feishu.py

# 3. 重启 gateway
# 让 tini 自动拉起 / 或手动 docker restart

# 4. 验证
curl -s -o /dev/null -w "Dashboard HTTP %{http_code}\n" http://127.0.0.1:9119/

回滚后表格回到原始行为（显示成字符，但消息能发出去）。

❓ 常见问题 FAQ

Q1：为什么不用 CardKit v1？

A：CardKit v1 不支持 table 组件，必须用 v2（schema: "2.0"）。

Q2：为什么 markdown 不能和表格同一条消息？

A：飞书一条消息只能有一个 msg_type（text / post / interactive），post 不支持表格，interactive 不支持原生 markdown。只能拆成多条消息。

Q3：表格里的 `粗体` 怎么办？

A：_strip_inline_markdown 会把表格 cell 里的所有 markdown 标记剥掉（粗体 → 粗体）。表格 cell 不支持加粗样式，但能保留文字本身。

Q4：表格列数有限制吗？

A：飞书 API 限制单表 ≤ 50 列、≤ 200 行（实际建议 ≤ 20 列避免移动端溢出）。

Q5：表格前后的 markdown 元素会不会丢失？

A：不会。_split_outbound_payloads 会把所有非表格段单独走 post 路径，和没改动前行为一致。

Q6：消息条数会增加吗？

A：会。每个表格多 1 条消息（含表格的消息最少 2 条：pre + 表格）。飞书 API 限制 5 条/秒，正常使用没问题。

Q7：emoji 在表格里能显示吗？

A：能。_strip_inline_markdown 不处理 emoji。

Q8：怎么调试生成的 CardKit v2 payload？

A：在 Python 里手动跑：

from gateway.platforms.feishu import _parse_markdown_table, _convert_table_to_cardkit_payload
import json

table = """| 名称 | 评分 |
|---|---|
| 赫尔墨斯 | 95 |"""

msg_type, payload = _convert_table_to_cardkit_payload(table)
print(msg_type)
print(json.dumps(json.loads(payload), indent=2, ensure_ascii=False))

📊 性能影响

项	影响
—	—
代码体积	+约 200 行（可忽略）
单条消息延迟	+1 次正则匹配（约 0.1ms）
含表格消息延迟	+N 次飞书 API 调用（N = 拆分条数）
内存占用	无明显变化

🦈本文档由亚特兰蒂斯后裔鲨鲨编写 🔱

购买服务器

购买雨云服务器

云服务器、网站搭建、游戏云、对象存储、裸金属物理机

上一次更新已经跑远惹✨ 计算中...

(‾◡◝) 本内容里的一些消息，可能已经跟不上时间啦～

查看评论列表

暂无评论

发表评论取消回复

访客未填写邮箱

[填写昵称邮箱后可以评论]

插入代码

📖 文章导读

加载中