基于Teams Toolkit的Bot实现特定消息绕过Azure OpenAI

核心问题分析 #

你遇到的问题本质是没有正确中断后续的AI调用流程：

• 中间件仅修改请求但未终止后续管道，导致AI调用逻辑仍会执行
• before_turn钩子回复后未明确终止当前turn处理，流程混乱导致else分支无法触发

解决方案1：直接在消息处理逻辑中判断（最简洁） #

修改bot.py的消息处理方法，优先匹配特定消息，直接回复后跳过AI调用，其余情况走GPT-4o流程。

修改后的bot.py关键代码： #

from botbuilder.core import ActivityHandler, TurnContext
from botbuilder.ai.openai import AzureOpenAIClient, AzureOpenAIOptions

class TeamsAssistantBot(ActivityHandler):
    def __init__(self, azure_openai_options: AzureOpenAIOptions):
        self.azure_openai_client = AzureOpenAIClient(azure_openai_options)

    async def on_message_activity(self, turn_context: TurnContext):
        # 定义需要直接回复的特定消息关键词/内容
        trigger_phrases = ["你好", "帮助", "关于"]
        user_message = turn_context.activity.text.strip().lower()

        # 匹配到特定消息，直接回复并终止流程
        if any(phrase in user_message for phrase in trigger_phrases):
            await turn_context.send_activity("这是特定消息的直接回复，无需调用AI")
            return  # 终止当前方法，跳过后续AI调用

        # 其余消息走GPT-4o处理
        response = await self.azure_openai_client.complete(
            prompt=user_message,
            max_tokens=500
        )
        await turn_context.send_activity(response.choices[0].text)

解决方案2：正确使用BeforeTurn钩子终止流程 #

若坚持使用before_turn钩子，需在钩子中返回False终止后续turn处理，避免进入默认AI调用逻辑。

修改后的bot.py钩子实现： #

from botbuilder.core import TurnContext, BotAdapter
from botbuilder.core.adapters import BeforeTurn
from botbuilder.ai.openai import AzureOpenAIClient, AzureOpenAIOptions

class TeamsAssistantBot(ActivityHandler):
    def __init__(self, azure_openai_options: AzureOpenAIOptions, adapter: BotAdapter):
        self.azure_openai_client = AzureOpenAIClient(azure_openai_options)
        # 注册before_turn钩子
        adapter.add_before_turn(self.handle_before_turn)

    async def handle_before_turn(self, turn_context: TurnContext) -> bool:
        user_message = turn_context.activity.text.strip().lower()
        if "特定关键词" in user_message:
            await turn_context.send_activity("直接回复内容")
            return False  # 返回False终止后续流程，不再执行on_message_activity

        return True  # 返回True继续执行AI调用流程

    async def on_message_activity(self, turn_context: TurnContext):
        # 仅当钩子返回True时执行AI调用
        response = await self.azure_openai_client.complete(
            prompt=turn_context.activity.text,
            max_tokens=500
        )
        await turn_context.send_activity(response.choices[0].text)

解决方案3：使用中间件拦截并终止请求 #

在中间件中判断消息，回复后调用turn_context.skip_remaining_middleware()跳过后续所有中间件和消息处理逻辑。

修改后的app.py中间件实现： #

from botbuilder.core import Middleware, TurnContext, BotFrameworkAdapter
from botbuilder.schema import ActivityTypes

class DirectResponseMiddleware(Middleware):
    async def on_turn(self, turn_context: TurnContext, next):
        if turn_context.activity.type == ActivityTypes.message:
            user_message = turn_context.activity.text.strip().lower()
            if "特定关键词" in user_message:
                await turn_context.send_activity("直接回复内容")
                turn_context.skip_remaining_middleware()  # 跳过后续所有流程
                return
        await next()

# 注册中间件到适配器
adapter = BotFrameworkAdapter(settings.APP_ID, settings.APP_PASSWORD)
adapter.use(DirectResponseMiddleware())

关键注意事项 #

• 所有方案的核心是终止后续流程：要么在消息处理中return，要么在钩子返回False，要么在中间件调用skip_remaining_middleware()
• 调整特定消息的匹配逻辑（关键词/精确匹配）以符合业务需求，避免误拦截
• 测试时分别发送特定消息和普通消息，验证两种场景的处理是否符合预期

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