Coverage for askbob\action\responder.py : 100%
Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from typing import Any, Dict, Text
2from rasa.utils.endpoints import EndpointConfig
3from rasa.core.agent import Agent
4import requests
7class ResponseService:
8 """An interface for a response service, which produces responses for text queries."""
10 def is_ready(self) -> bool:
11 """Returns whether the response service is ready to accept queries.
13 Returns:
14 bool: Whether or not the response server is ready to accept queries.
15 """
16 return True
18 async def handle(self, query: str, sender: str = "askbob"):
19 """Handles a queries and produces a response.
21 Args:
22 query (str): The user's query.
23 sender (str, optional): A unique sender identifier. Defaults to "askbob".
25 Raises:
26 NotImplementedError: This is just an interface - please use a concrete implementation!
27 """
29 raise NotImplementedError(
30 "The response service has not been implemented.")
33def yielder(m: Dict[Text, Any]) -> Dict[Text, Any]:
34 """A helper function to generate correct message responses shared between ResponseService implementations.
36 Args:
37 m (Dict[Text, Any]): The message returned by Rasa.
39 Returns:
40 Dict[Text, Any]: An AskBob-formatted response.
41 """
43 if "text" in m:
44 return {"text": m["text"]}
45 elif "image" in m:
46 return {"image": m["image"]}
47 elif "custom" in m:
48 return {"custom": m["custom"]}
49 else:
50 return {"other": m}
51 # NOTE: buttons are not supported
54class RasaResponseService(ResponseService):
55 """A ResponseService that uses Rasa running locally to respond to queries."""
57 agent: Agent
59 def __init__(self, model_dir: str = "data/rasa/models", plugins_location="plugins", plugins_port=5055) -> None:
60 """Initialises the RasaResponseService.
62 The Rasa actions server required for custom actions to function is automatically started when this class is initialised.
64 Args:
65 model_dir (str, optional): The location of the Rasa model to load. Defaults to "data/rasa/models".
66 """
68 # Action Server
69 import sys
70 import subprocess
71 self.action_server = subprocess.Popen(
72 [sys.executable, "-m", "askbob.action.server", plugins_location, str(plugins_port)])
74 # Main Rasa Model
75 from rasa.model import get_latest_model
76 model_path = get_latest_model(model_dir)
77 endpoint = EndpointConfig("http://localhost:5055/webhook")
78 self.agent = Agent.load(model_path, action_endpoint=endpoint)
80 def __del__(self):
81 """Terminates the Rasa action service when this object is destroyed as it is no longer needed."""
82 self.action_server.terminate()
84 def is_ready(self) -> bool:
85 return self.agent.is_ready()
87 async def handle(self, query: str, sender: str = "askbob"):
88 for m in await self.agent.handle_text(text_message=query, sender_id=sender):
89 yield yielder(m)
92# class HTTPResponseService(ResponseService):
93# """A ResponseService that uses Rasa's HTTP webhook to respond to queries."""
95# endpoint: str
97# def __init__(self, endpoint_base) -> None:
98# self.endpoint = endpoint_base + "/webhooks/rest/webhook"
100# async def handle(self, query, sender="askbob"):
101# r = requests.post(self.endpoint, json={
102# "sender": sender,
103# "message": query
104# }).json()
106# for m in r:
107# yield yielder(m)