12112350604

Committed 02 Dec 2024 03:52AM UTC coverage: 89.891% (-0.04%) from 89.932%

Build # 12112350604

Build Type

Pull #1611

github

Committed by

web-flow

Commit Message

Merge 9176d03d1 into f2f296b80

Pull Request Pull Request #1611: Mail channel implementation

Run Details

383 of 434 new or added lines in 15 files covered. (88.25%)

12 existing lines in 2 files now uncovered.

24141 of 26856 relevant lines covered (89.89%)

0.9 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.85

/kairon/chat/agent_processor.py

import os
from typing import Text

from loguru import logger as logging
from rasa.core.agent import Agent
from rasa.core.channels import UserMessage

from kairon.shared.chat.cache.in_memory_agent import AgentCache
from kairon.exceptions import AppException
from kairon.shared.data.processor import MongoProcessor
from .agent.agent import KaironAgent
from kairon.shared.chat.cache.in_memory_agent import InMemoryAgentCache
from ..shared.live_agent.live_agent import LiveAgentHandler
from ..shared.utils import Utility
from kairon.shared.otel import record_custom_attributes



class AgentProcessor:
    """
    Class contains logic for loading bot agents
    """

    mongo_processor = MongoProcessor()
    cache_provider: AgentCache = InMemoryAgentCache()

    @staticmethod
    def get_agent(bot: Text) -> Agent:
        """
        fetch the bot agent from cache if exist otherwise load it into the cache

        :param bot: bot id
        :return: Agent Object
        """
        if not AgentProcessor.cache_provider.is_exists(bot) or not AgentProcessor.is_latest_version_in_mem(bot):
            AgentProcessor.reload(bot)
        record_custom_attributes(num_models=AgentProcessor.cache_provider.len())
        return AgentProcessor.cache_provider.get(bot)

    @staticmethod
    def reload(bot: Text):
        """
        reload bot agent

        :param bot: bot id
        :return: None
        """
        try:
            endpoint = AgentProcessor.mongo_processor.get_endpoints(
                bot, raise_exception=False
            )
            action_endpoint = Utility.get_action_url(endpoint)
            lock_store_endpoint = Utility.get_lock_store(bot)
            model_path = Utility.get_latest_model(bot)
            domain = AgentProcessor.mongo_processor.load_domain(bot)
            bot_settings = AgentProcessor.mongo_processor.get_bot_settings(bot, "system")
            mongo_store = Utility.get_local_mongo_store(bot, domain)
            agent = KaironAgent.load(model_path, action_endpoint=action_endpoint, tracker_store=mongo_store,
                                     lock_store=lock_store_endpoint)
            agent.model_ver = model_path.split("/")[-1]
            AgentProcessor.cache_provider.set(bot, agent, is_billed=bot_settings.is_billed)
        except Exception as e:
            logging.exception(e)
            raise AppException("Bot has not been trained yet!")

    @staticmethod
    def is_latest_version_in_mem(bot: Text):
        model_path = Utility.get_latest_model(bot)
        latest_ver = model_path.split("/")[-1]
        in_mem_model_ver = AgentProcessor.cache_provider.get(bot).model_ver
        logging.debug(f"PID:{os.getpid()} In memory model:{in_mem_model_ver}, latest trained model:{latest_ver}")
        return latest_ver == in_mem_model_ver

    @staticmethod
    async def handle_channel_message(bot: Text, userdata: UserMessage):
        is_live_agent_enabled = await LiveAgentHandler.check_live_agent_active(bot, userdata)
        logging.debug(f"Live agent enabled:{is_live_agent_enabled}")
        if not is_live_agent_enabled:
            return await AgentProcessor.get_agent(bot).handle_message(userdata)
        return await LiveAgentHandler.process_live_agent(bot, userdata)

    @staticmethod
    def get_agent_without_cache(bot: str, use_store: bool = True) -> Agent:
        endpoint = AgentProcessor.mongo_processor.get_endpoints(
            bot, raise_exception=False
        )
        action_endpoint = Utility.get_action_url(endpoint)
        model_path = Utility.get_latest_model(bot)
        domain = AgentProcessor.mongo_processor.load_domain(bot)
        if use_store:
            mongo_store = Utility.get_local_mongo_store(bot, domain)
            lock_store_endpoint = Utility.get_lock_store(bot)
            agent = KaironAgent.load(model_path, action_endpoint=action_endpoint, tracker_store=mongo_store,
                                     lock_store=lock_store_endpoint)
        else:
            agent = KaironAgent.load(model_path, action_endpoint=action_endpoint)

        agent.model_ver = model_path.split("/")[-1]
        return agent

1	import os	1✔
2	from typing import Text	1✔
3
4	from loguru import logger as logging	1✔
5	from rasa.core.agent import Agent	1✔
6	from rasa.core.channels import UserMessage	1✔
7
8	from kairon.shared.chat.cache.in_memory_agent import AgentCache	1✔
9	from kairon.exceptions import AppException	1✔
10	from kairon.shared.data.processor import MongoProcessor	1✔
11	from .agent.agent import KaironAgent	1✔
12	from kairon.shared.chat.cache.in_memory_agent import InMemoryAgentCache	1✔
13	from ..shared.live_agent.live_agent import LiveAgentHandler	1✔
14	from ..shared.utils import Utility	1✔
15	from kairon.shared.otel import record_custom_attributes	1✔
16
17
18
19	class AgentProcessor:	1✔
20	"""
21	Class contains logic for loading bot agents
22	"""
23
24	mongo_processor = MongoProcessor()	1✔
25	cache_provider: AgentCache = InMemoryAgentCache()	1✔
26
27	@staticmethod	1✔
28	def get_agent(bot: Text) -> Agent:	1✔
29	"""
30	fetch the bot agent from cache if exist otherwise load it into the cache
31
32	:param bot: bot id
33	:return: Agent Object
34	"""
35	if not AgentProcessor.cache_provider.is_exists(bot) or not AgentProcessor.is_latest_version_in_mem(bot):	1✔
36	AgentProcessor.reload(bot)	1✔
37	record_custom_attributes(num_models=AgentProcessor.cache_provider.len())	1✔
38	return AgentProcessor.cache_provider.get(bot)	1✔
39
40	@staticmethod	1✔
41	def reload(bot: Text):	1✔
42	"""
43	reload bot agent
44
45	:param bot: bot id
46	:return: None
47	"""
48	try:	1✔
49	endpoint = AgentProcessor.mongo_processor.get_endpoints(	1✔
50	bot, raise_exception=False
51	)
52	action_endpoint = Utility.get_action_url(endpoint)	1✔
53	lock_store_endpoint = Utility.get_lock_store(bot)	1✔
54	model_path = Utility.get_latest_model(bot)	1✔
55	domain = AgentProcessor.mongo_processor.load_domain(bot)	1✔
56	bot_settings = AgentProcessor.mongo_processor.get_bot_settings(bot, "system")	1✔
57	mongo_store = Utility.get_local_mongo_store(bot, domain)	1✔
58	agent = KaironAgent.load(model_path, action_endpoint=action_endpoint, tracker_store=mongo_store,	1✔
59	lock_store=lock_store_endpoint)
60	agent.model_ver = model_path.split("/")[-1]	1✔
61	AgentProcessor.cache_provider.set(bot, agent, is_billed=bot_settings.is_billed)	1✔
62	except Exception as e:	1✔
63	logging.exception(e)	1✔
64	raise AppException("Bot has not been trained yet!")	1✔
65
66	@staticmethod	1✔
67	def is_latest_version_in_mem(bot: Text):	1✔
68	model_path = Utility.get_latest_model(bot)	1✔
69	latest_ver = model_path.split("/")[-1]	1✔
70	in_mem_model_ver = AgentProcessor.cache_provider.get(bot).model_ver	1✔
71	logging.debug(f"PID:{os.getpid()} In memory model:{in_mem_model_ver}, latest trained model:{latest_ver}")	1✔
72	return latest_ver == in_mem_model_ver	1✔
73
74	@staticmethod	1✔
75	async def handle_channel_message(bot: Text, userdata: UserMessage):	1✔
76	is_live_agent_enabled = await LiveAgentHandler.check_live_agent_active(bot, userdata)	1✔
77	logging.debug(f"Live agent enabled:{is_live_agent_enabled}")	1✔
78	if not is_live_agent_enabled:	1✔
79	return await AgentProcessor.get_agent(bot).handle_message(userdata)	1✔
80	return await LiveAgentHandler.process_live_agent(bot, userdata)	×
81
82	@staticmethod	1✔
83	def get_agent_without_cache(bot: str, use_store: bool = True) -> Agent:	1✔
84	endpoint = AgentProcessor.mongo_processor.get_endpoints(	1✔
85	bot, raise_exception=False
86	)
87	action_endpoint = Utility.get_action_url(endpoint)	1✔
88	model_path = Utility.get_latest_model(bot)	1✔
89	domain = AgentProcessor.mongo_processor.load_domain(bot)	1✔
90	if use_store:	1✔
NEW 91	mongo_store = Utility.get_local_mongo_store(bot, domain)	×
NEW 92	lock_store_endpoint = Utility.get_lock_store(bot)	×
NEW 93	agent = KaironAgent.load(model_path, action_endpoint=action_endpoint, tracker_store=mongo_store,	×
94	lock_store=lock_store_endpoint)
95	else:
96	agent = KaironAgent.load(model_path, action_endpoint=action_endpoint)	1✔
97
98	agent.model_ver = model_path.split("/")[-1]	1✔
99	return agent	1✔

digiteinfotech / kairon / 12112350604

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous