I tried to set slots in my own channel, but it was stuck in the HTTP connection.
However, I can got correct result through CURL POST in shell:
Here is my code:
class MyIO(InputChannel):
def name(cls) -> Text:
"""Name of your custom channel."""
return "myio"
# To set slots in the tracker, send the tracker id, actual slot name and value to be set in the slot
async def set_slot(self, tracker_id, slot_name, slot_value):
api_url_base = 'http://localhost:1234/conversations/' + tracker_id + '/tracker/events'
print(api_url_base)
data = json.dumps({
"event": "slot",
"name": slot_name,
"value": slot_value
})
headers = {'Content-Type': 'application/json'}
# It always stuck in the post connection, Why?
set_slot = await requests.post(api_url_base, data=data, headers=headers)
if set_slot.status_code != 200:
set_slot = await requests.post(api_url_base, data=data, headers=headers)
return "Done"
def blueprint(
self, on_new_message: Callable[[UserMessage], Awaitable[None]]
) -> Blueprint:
custom_webhook = Blueprint(
"custom_webhook_{}".format(type(self).__name__),
inspect.getmodule(self).__name__,
)
@custom_webhook.route("/", methods=["GET"])
async def health(request: Request) -> HTTPResponse:
return response.json({"status": "ok"})
@custom_webhook.route("/webhook", methods=["POST"])
async def receive(request: Request) -> HTTPResponse:
sender_id = request.json.get("sender") # method to get sender_id
input_channel = self.name() # method to fetch input channel
metadata = self.get_metadata(request) # method to get metadata
message = request.json.get("message")
wav = request.json.get("wav")
print(wav)
try:
if wav is not None:
await self.set_slot(sender_id,"base64", wav)
print('base64 saved success~')
except Exception as e:
print(e)
print('base64 slot failed!!!')
collector = CollectingOutputChannel()
# include exception handling
await on_new_message(
UserMessage(
message, # message,
collector,
sender_id,
input_channel=input_channel,
metadata=metadata,
)
)
return response.json(collector.messages)
return custom_webhook
def get_metadata(self, request: Request) -> Optional[Dict[Text, Any]]:
return request.json.get("metadata", None)
I wanna know whether it’s correct to sent post to tracker API inside the InputChannel. Or is there any other way to set slots in custom channel (like rasa sdk)? Will the SlotSet(‘base64’, ‘abcd’) work in some way?
Thanks a lot. My Rasa Version is 2.6.0