BotFramework v4 Python Teams机器人主动消息仅部分送达问题咨询

• 超时阈值设置过严：当日问题推送需要发送2条带自适应卡片的消息，消息负载远大于单条答案推送消息，5秒的超时阈值会直接截断大量还未完成的发送请求，这是最可能的核心诱因。
• 错误捕获范围过窄：当前代码仅捕获TimeoutError，未捕获Bot Framework抛出的限流错误、权限错误、消息格式错误等其他异常，大量发送失败的请求没有被日志记录，也无法针对性处理。
• 串行发送触发限流：所有发送请求串行执行，当订阅用户量较大时，高频调用Teams主动消息接口很容易触发平台的429限流规则，未处理限流的情况下请求会直接失败。
• 自适应卡片格式兼容性问题：生成的当日问题卡片可能存在特殊字符、资源链接失效等边界问题，导致部分用户的会话无法正常接收该类消息。

1. 调整超时阈值和错误捕获逻辑
   将单条请求的超时阈值调整为10~15秒，同时扩大异常捕获范围，添加日志记录所有发送失败的用户ID和错误原因，方便后续定位具体问题。
2. 增加限流处理逻辑
   调用主动消息接口时添加并发控制，同时捕获429限流错误，根据响应头的Retry-After字段等待对应时间后重试，避免批量发送时被平台拦截。
3. 优化发送效率
   将串行发送改为异步并发发送，通过信号量控制同时发送的请求数量，既提升发送效率，也避免请求过于集中触发限流。
4. 提前信任服务URL
   无需在循环内重复调用AppCredentials.trust_service_url，提前提取所有会话的服务URL批量信任即可，减少不必要的重复调用开销。

优化后代码示例 #

import logging
import asyncio
from datetime import datetime

async def _send_question_of_the_day(conv_dict):
    cards = await utils._create_adaptive_card_attachment()
    now = datetime.now()
    data = now.strftime('%d/%m/%Y')
    # 提前批量信任所有服务URL
    service_urls = {ref.service_url for ref in conv_dict.values()}
    for url in service_urls:
        AppCredentials.trust_service_url(url)
    # 控制并发数为5，可根据实际情况调整
    semaphore = asyncio.Semaphore(5)
    tasks = []

    async def safe_send(ref):
        async with semaphore:
            try:
                # 超时调整为15秒
                await asyncio.wait_for(_send_qptd_sep(cards, data, ref), timeout=15)
                logging.info(f"用户{ref.user.id}当日问题推送成功")
            except Exception as e:
                logging.error(f"用户{ref.user.id}推送失败，错误：{str(e)}")
                # 处理429限流错误，重试一次
                if hasattr(e, "response") and getattr(e.response, "status", 0) == 429:
                    retry_after = int(e.response.headers.get("Retry-After", 10))
                    await asyncio.sleep(retry_after)
                    try:
                        await asyncio.wait_for(_send_qptd_sep(cards, data, ref), timeout=15)
                        logging.info(f"用户{ref.user.id}重试推送成功")
                    except Exception as retry_e:
                        logging.error(f"用户{ref.user.id}重试推送失败，错误：{str(retry_e)}")

    for ref in conv_dict.values():
        tasks.append(safe_send(ref))
    # 等待所有发送任务执行完成
    await asyncio.gather(*tasks, return_exceptions=True)

async def _send_qptd_sep(cards, data, conversation_reference):
    await ADAPTER.continue_conversation(
        conversation_reference,
        lambda turn_context: turn_context.send_activities([
        Activity(
            text="bla bla bla",
            type=ActivityTypes.message,
            attachments=[CardFactory.adaptive_card(cards[0])]
        ),
        Activity(
            type=ActivityTypes.message,
            attachments=[CardFactory.adaptive_card(cards[-1])]
        )
    ]),
        APP_ID,
    )

内容的提问来源于stack exchange，提问作者Scaramel