今天我們要來實作一個 Bing Search Chatbot。

建立一些檔案

在 requirements 輸入

aiohttp
jsonpickle
botbuilder-core>=4.10.0

開始撰寫程式碼

複製貼上以下程式碼至 config.py

import os

class DefaultConfig:
    """ Bot Configuration """
    PORT = 3978
    APP_ID = os.environ.get("MicrosoftAppId", "")
    APP_PASSWORD = os.environ.get("MicrosoftAppPassword", "")
    
    '''Bing Search API 的基本設定'''
    # (必需修改) 此KEY為Azure Portal上Bing Image API的KEY
    BING_SEARCH_API_KEY = '<貼上你的Bin Search KEY>'
    # (可修改) 使用Bing Image API所打的endpoint，可修改後面query的內容
    BING_IMAGE_SEARCH_API_ENDPOINT = "https://api.cognitive.microsoft.com/bing/v7.0/images/search"

複製貼上以下程式碼至 app.py

from aiohttp import web
from aiohttp.web import Request, Response, json_response
from botbuilder.core import (
    BotFrameworkAdapterSettings,
    TurnContext,
    BotFrameworkAdapter
)
from botbuilder.core.integration import aiohttp_error_middleware
from botbuilder.schema import Activity, ActivityTypes

from bots import BingSearchBot
from config import DefaultConfig

import sys
import traceback
from datetime import datetime

CONFIG = DefaultConfig()
# Create adapter.
# See https://aka.ms/about-bot-adapter to learn more about how bots work.
SETTINGS = BotFrameworkAdapterSettings(CONFIG.APP_ID, CONFIG.APP_PASSWORD)
ADAPTER = BotFrameworkAdapter(SETTINGS)


# Catch-all for errors.
async def on_error(context: TurnContext, error: Exception):
    # This check writes out errors to console log .vs. app insights.
    # NOTE: In production environment, you should consider logging this to Azure
    #       application insights.
    print(f"\n [on_turn_error] unhandled error: {error}", file=sys.stderr)
    traceback.print_exc()

    # Send a message to the user
    await context.send_activity("The bot encountered an error or bug.")
    await context.send_activity(
        "To continue to run this bot, please fix the bot source code."
    )
    # Send a trace activity if we're talking to the Bot Framework Emulator
    if context.activity.channel_id == "emulator":
        # Create a trace activity that contains the error object
        trace_activity = Activity(
            label="TurnError",
            name="on_turn_error Trace",
            timestamp=datetime.utcnow(),
            type=ActivityTypes.trace,
            value=f"{error}",
            value_type="https://www.botframework.com/schemas/error",
        )
        # Send a trace activity, which will be displayed in Bot Framework Emulator
        await context.send_activity(trace_activity)

ADAPTER.on_turn_error = on_error

# create bot
BOT = BingSearchBot()


# Listen for incoming requests on /api/messages
async def messages(req: Request) -> Response:
    # Main bot message handler.
    if "application/json" in req.headers["Content-Type"]:
        body = await req.json()
    else:
        return Response(status=415)

    activity = Activity().deserialize(body)
    auth_header = req.headers["Authorization"] if "Authorization" in req.headers else ""

    response = await ADAPTER.process_activity(activity, auth_header, BOT.on_turn)
    if response:
        return json_response(data=response.body, status=response.status)
    return Response(status=201)


APP = web.Application(middlewares=[aiohttp_error_middleware])
APP.router.add_post("/api/messages", messages)

if __name__ == "__main__":
    try:
        web.run_app(APP, host="localhost", port=CONFIG.PORT)
    except Exception as error:
        raise error

在 bots 的資料夾新增兩個 .py 檔

在 __init__.py 複製貼上以下程式碼

from .bing_search_bot import BingSearchBot

__all__ = ["BingSearchBot"]

在 bing_search_bot.py 複製貼上以下程式碼

from botbuilder.core import ActivityHandler, TurnContext, MessageFactory
from botbuilder.schema import ChannelAccount, Attachment

from config import DefaultConfig

import json
import requests
import urllib.request
import os

class BingSearchBot(ActivityHandler):
    # 首次加入的問候的訊息
    async def on_members_added_activity(self, members_added: list, turn_context: TurnContext):
        for member in members_added:
            if member.id != turn_context.activity.recipient.id:
                await turn_context.send_activity(
                    f"歡迎來到Bing API的機器人，傳圖片以圖搜尋，傳文字以文字搜尋 !")

    # 每一次對話都是由這裏來處理
    async def on_message_activity(self, turn_context: TurnContext):
        image_url_list = []
        host_page_url_list = []
        host_page_name_list = []
        
        if turn_context.activity.text : 
            image_url_list, host_page_url_list, host_page_name_list = self.bing_image_api(turn_context.activity.text)
            for i in range(len(host_page_url_list)):
                await turn_context.send_activity(MessageFactory.content_url(url= image_url_list[i], content_type='image/jpg', text=host_page_name_list[i]))
                await turn_context.send_activity(MessageFactory.text(text= host_page_url_list[i]))
    
    def bing_image_api(self, query) : 
        image_url = []
        host_page_url = []
        host_page_name = []

        # Construct a request
        mkt = 'zh-tw'
        params = {'q': query, 'mkt': mkt}
        headers = {'Ocp-Apim-Subscription-Key': DefaultConfig.BING_SEARCH_API_KEY}
        
        try:
            response = requests.get(DefaultConfig.BING_IMAGE_SEARCH_API_ENDPOINT, headers=headers, params=params)
            response.raise_for_status()

            for i in range(3):
                image_url.append(response.json()['value'][i]['contentUrl'])       # 若要更改回傳內容，可修改後面的索引
                host_page_url.append(response.json()['value'][i]['hostPageUrl'])
                host_page_name.append(response.json()['value'][i]['name'])
        except Exception as ex:
            raise ex
        
        return image_url, host_page_url, host_page_name

測試機器人

• Option 1: 在 Bot Framework Emulator 上測試
• Option 2: 在 Azure Portal 上測試