我正在尝试让机器人通过单独的功能回复现有对话。 我的代码如下所示。
class TestClass():
def callback(self, message):
async def async_callback(turn_context):
await turn_context.send_activity(message)
return async_callback
async def testReply(self, message):
conversation_reference = shared_storage[self.user_id]
adapter = MyBotAdapter()
await adapter.continue_conversation(conversation_reference, self.callback(message))
print("OUTTTTTTTTTTTTTTT")
class TestBot(ActivityHandler):
async def on_message_activity(self, turn_context: TurnContext):
conversation_reference = TurnContext.get_conversation_reference(turn_context.activity)
self.user_id = turn_context.activity.from_property.id
shared_storage[self.user_id] = conversation_reference
other_class = TestClass(self.user_id)
await other_class.testReply(turn_context.activity.text)
没有错误，但 testReply 根本不回复。知道是什么原因造成的吗？
尝试这个方法：
app.py：
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
import sys
import traceback
import uuid
from datetime import datetime
from http import HTTPStatus
from typing import Dict
from aiohttp import web
from aiohttp.web import Request, Response, json_response
from botbuilder.core import (
TurnContext,
)
from botbuilder.core.integration import aiohttp_error_middleware
from botbuilder.integration.aiohttp import CloudAdapter, ConfigurationBotFrameworkAuthentication
from botbuilder.schema import Activity, ActivityTypes, ConversationReference
from bots import ProactiveBot
from config import DefaultConfig
CONFIG = DefaultConfig()
# Create adapter.
# See https://aka.ms/about-bot-adapter to learn more about how bots work.
ADAPTER = CloudAdapter(ConfigurationBotFrameworkAuthentication(CONFIG))
# 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 a shared dictionary. The Bot will add conversation references when users
# join the conversation and send messages.
CONVERSATION_REFERENCES: Dict[str, ConversationReference] = dict()
# If the channel is the Emulator, and authentication is not in use, the AppId will be null.
# We generate a random AppId for this case only. This is not required for production, since
# the AppId will have a value.
APP_ID = ...YOUR APP ID HERE
# Create the Bot
BOT = ProactiveBot(CONVERSATION_REFERENCES)
# 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=HTTPStatus.UNSUPPORTED_MEDIA_TYPE)
activity = Activity().deserialize(body)
auth_header = req.headers["Authorization"] if "Authorization" in req.headers else ""
response = await ADAPTER.process_activity(auth_header, activity, BOT.on_turn)
if response:
return json_response(data=response.body, status=response.status)
return Response(status=HTTPStatus.OK)
# Listen for requests on /api/notify, and send a messages to all conversation members.
async def notify(req: Request) -> Response: # pylint: disable=unused-argument
await _send_proactive_message()
return Response(status=HTTPStatus.OK, text="Proactive messages have been sent")
# Send a message to all conversation members.
# This uses the shared Dictionary that the Bot adds conversation references to.
async def _send_proactive_message():
for conversation_reference in CONVERSATION_REFERENCES.values():
await ADAPTER.continue_conversation(
conversation_reference,
lambda turn_context: turn_context.send_activity("proactive hello"),
APP_ID,
)
APP = web.Application(middlewares=[aiohttp_error_middleware])
APP.router.add_post("/api/messages", messages)
APP.router.add_get("/api/notify", notify)
if __name__ == "__main__":
try:
web.run_app(APP, host="localhost", port=CONFIG.PORT)
except Exception as error:
raise error
bot.py：
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
from typing import Dict
from botbuilder.core import ActivityHandler, TurnContext
from botbuilder.schema import ChannelAccount, ConversationReference, Activity
class ProactiveBot(ActivityHandler):
def __init__(self, conversation_references: Dict[str, ConversationReference]):
self.conversation_references = conversation_references
async def on_conversation_update_activity(self, turn_context: TurnContext):
self._add_conversation_reference(turn_context.activity)
return await super().on_conversation_update_activity(turn_context)
async def on_members_added_activity(
self, members_added: [ChannelAccount], turn_context: TurnContext
):
for member in members_added:
if member.id != turn_context.activity.recipient.id:
await turn_context.send_activity(
"Welcome to the Proactive Bot sample. Navigate to "
"http://localhost:3978/api/notify to proactively message everyone "
"who has previously messaged this bot."
)
async def on_message_activity(self, turn_context: TurnContext):
self._add_conversation_reference(turn_context.activity)
return await turn_context.send_activity(
f"You sent: {turn_context.activity.text}"
)
def _add_conversation_reference(self, activity: Activity):
"""
This populates the shared Dictionary that holds conversation references. In this sample,
this dictionary is used to send a message to members when /api/notify is hit.
:param activity:
:return:
"""
conversation_reference = TurnContext.get_conversation_reference(activity)
self.conversation_references[
conversation_reference.user.id
] = conversation_reference
此代码将在 http://localhost:3978/api/notify (GET) 上提供一个端点，每次请求时，它都会向之前发送过消息的用户发送主动通知。
我遇到了和你一样的问题，所以我克隆了 botbuilder-samples 存储库，并根据存储库上的示例 botbuilder-samples/samples/bot-samples/16.proactive-messages 实现了这种方法.
这是此代码所在的 github 存储库的链接：
https://github.com/microsoft/BotBuilder-Samples.git
这里有一份微软文档可供贡献：