# Created by aaronkueh on 12/10/2025 # aom/agents/rca.py from typing import Dict, Any import pandas as pd from jinja2 import Template from aom.utils.logger import get_logger from aom.utils.agent_helpers import load_agent_prompts, call_tool_context, rca_sanitize from aom.mcp_server.server import call_tool from aom.utils.contracts import RCARequest, ISARequest from aom.tools.rag.rag_helper import load_rag_profile from datetime import datetime from dateutil.relativedelta import relativedelta from aom.agents.isa import ISAAgent class RCAAgent: """ Root Cause Analysis (RCA) Agent Module This module provides the RCAAgent class that performs automated root cause analysis and generates recommendations for asset alerts. The agent retrieves relevant technical documentation and historical patterns using RAG (Retrieval-Augmented Generation), then synthesizes natural language recommendations using LLM summarization. """ def __init__(self): self.log = get_logger(self.__class__.__name__) self.agent_profile = load_agent_prompts(agent='RCA', config='agents.yaml') self.extract_asset_meta = call_tool("mssql_metadata_query", {}) def run(self, req: RCARequest) -> Dict[str, Any]: """ Execute root cause analysis for an alert and generate recommendations. This method retrieves asset metadata, searches relevant documentation using RAG, and generates natural language recommendations via LLM summarization. Args: req (RCARequest): Request object containing alert details: - alert_id: Alert unique identifier - asset_id: Asset unique identifier - title: Alert title - description: Alert description - status: Optional recommendation status (0=Open, 1=Closed) - timestamp: Optional alert trigger timestamp - cause: Optional alert root cause - resolution: Optional alert resolution - db_profile: Optional database profile to use Returns: Dict[str, Any]: Dictionary containing 'summary' key with HTML-formatted recommendation text, or 'Asset ID not found' message Workflow: 1. Query asset metadata from the database 2. Build RAG query from alert details and asset type 3. Retrieve relevant context from the knowledge base 4. Generate recommendation using LLM with RAG context 5. Sanitize and return HTML response """ self.log.info('RCA Agent activated') prompts, defaults = self.agent_profile rag = load_rag_profile(config='rag_config.yaml') _retrieve = rag.get("retrieve", {}) if req.workflow_flag: asset_meta = call_tool('mssql_metadata_query', {'profile': None, 'asset_id': req.asset_id}) asset_meta = pd.DataFrame(asset_meta['rows'], columns=asset_meta['columns']) if asset_meta.empty: return {'summary':'Asset ID not found'} else: maker = [asset_meta.iloc[0]['maker']] asset_type = asset_meta.iloc[0]['asset_type'] rag_query = (f"Show me the common root causes and solutions for {asset_type} according to:\n" f"alert title: {req.title}\n" f"alert description: {req.description}\n " ) ctx_resp = call_tool('retrieve_context', { 'prompt': rag_query, 'top_k': int(_retrieve.get('top_k')), 'min_score': float(_retrieve.get('min_score')), 'maker': maker, }) rag_context = (ctx_resp or {}).get('context', '').strip() template_str = str(prompts) rca_prompt = Template(template_str).render(rag_context=rag_context, alert_message=rag_query) final_prompt = rag_query + "\n" + rca_prompt response = call_tool("llm_summarize", {"prompt": final_prompt, "max_new_tokens": defaults.get("max_new_tokens")}) return {'summary': rca_sanitize(response)} else: if not prompts: prompts = "You are an equipment asset specialist which provide root cause and solution based on failure or alert.\n" if req.asset_id is not None: asset_meta = call_tool("mssql_metadata_query", {"profile": None, "asset_id": req.asset_id}) or {} asset_meta = pd.DataFrame(asset_meta['rows'], columns=asset_meta['columns']) if asset_meta.empty: error_msg = f'Asset: {req.asset_id} not found in the system. Please specify one of following assets:\n\n' return {'summary': error_msg, 'Available assets': call_tool_context(self.extract_asset_meta)} else: maker = [asset_meta.iloc[0]['maker']] asset_type = asset_meta.iloc[0]['asset_type'] table_name = asset_meta.iloc[0]['ts_data_table'] rag_query = (f"Show me the common root causes and solutions for {asset_type} according to:\n" f"user prompt: {req.description}\n") ctx_resp = call_tool('retrieve_context', { 'prompt': rag_query, 'top_k': int(_retrieve.get('top_k')), 'min_score': float(_retrieve.get('min_score')), 'maker': maker, }) rag_context = (ctx_resp or {}).get('context', '').strip() if req.timestamp is not None: # allow both datetime and string timestamp if isinstance(req.timestamp, datetime): end_dt = req.timestamp else: end_dt = datetime.fromisoformat(str(req.timestamp).replace("Z", "").replace(" ", "T")) else: dr = call_tool("mssql_ts_date_range", {"asset_id": req.asset_id, 'table_name': table_name}) or {} latest_end = dr.get("last_date", datetime.now()) end_dt = datetime.fromisoformat(str(latest_end).replace("Z", "").replace(" ", "T")) isa_context = "" if end_dt: start_dt = end_dt - relativedelta(days=30) isa_req = ISARequest( asset_id=req.asset_id, start=start_dt, end=end_dt, db_profile=req.db_profile, ) isa_result = ISAAgent().run(isa_req) or {} isa_context = (isa_result.get("summary") or "").strip() if rag_context: fallback_template = ( "You are answering a user using BOTH the general chat prompt and the DOCUMENT CONTEXT below. " "Prioritize the DOCUMENT CONTEXT. Do not invent facts.\n\n" "DOCUMENT CONTEXT:\n" f"{rag_context}\n\n" "IMPORTANT:\n" "- If the context contradicts general knowledge, prefer the context.\n" "- If the context is insufficient, say so briefly or ask user to provide more context.\n" f"- Must provide all and complete CITE: (doc_id, page, title and image) from {rag_context}. " f"- Provide number for each CITE, if more than one CITE found in {rag_context}.\n\n" "GENERAL CHAT PROMPT:\n" f"{req.description}" f"Output template in 4 sections: Diagnosis from {isa_context}, Root Cause, Solution and Citations." "End your reply with the exact token ." ) template_str = prompts.get('summarize_template', fallback_template) else: meta_context = call_tool_context(self.extract_asset_meta) fallback_template = ( "No information found from knowledge database, you can provide your own Root Causes, Solution and Notes" "You are answering a user using BOTH the general chat prompt and the asset information below. " "Prioritize the asset information, as this is the source of truth for all assets in the system." "asset information:\n" f"{meta_context}\n\n" "IMPORTANT:\n" "Do not invent facts that are not available from asset information.\n" "If the context is insufficient, say so briefly or ask user to provide more context.\n" f"If information cannot be found from {meta_context}, inform user and provide information that are available.\n\n" "GENERAL CHAT PROMPT:\n" f"{req.description}" "Notes: The following insights are generated by the language model and should be validated before use" "Output template in 3 sections: Root Cause, Solution and Notes." "End your reply with the exact token ." ) template_str = fallback_template rca_isa_prompt = Template(template_str).render(rag_context=rag_context, alert_message=rag_query, isa_context=isa_context) final_prompt = (rag_query + "\n" + rca_isa_prompt) response = call_tool("llm_summarize", {"prompt": final_prompt, "max_new_tokens": defaults.get("max_new_tokens")}) else: try: print(f"[RAG] Enhanced query: {req.description[:200]}...") # Debug log ctx_resp = call_tool("retrieve_context", { "prompt": req.description, "top_k": int(_retrieve.get("top_k")), "min_score": float(_retrieve.get("min_score")), "maker": [] }) rag_context = (ctx_resp or {}).get('context', '').strip() except Exception as _e: # Fail-safe: if retrieval fails, we just continue with regular chat print(f'[RAG] No context retrieved: {_e}') rag_context = "" if rag_context: fallback_template = ( "You are answering a user using BOTH the general chat prompt and the DOCUMENT CONTEXT below. " "Prioritize the DOCUMENT CONTEXT. Do not invent facts.\n\n" "DOCUMENT CONTEXT:\n" f"{rag_context}\n\n" "IMPORTANT:\n" "- If the context contradicts general knowledge, prefer the context.\n" "- If the context is insufficient, say so briefly or ask user to provide more context.\n" f"- Must provide all and complete CITE: (doc_id, page, title and image) from {rag_context}. " f"- Provide number for each CITE, if more than one CITE found in {rag_context}.\n\n" "GENERAL CHAT PROMPT:\n" f"{req.description}" "Output template in 3 sections: Root Cause, Solution and Citations." ) template_str = prompts.get('summarize_template', fallback_template) rca_prompt = Template(template_str).render(rag_context=rag_context, alert_message=req.description) final_prompt = rca_prompt else: meta_context = call_tool_context(self.extract_asset_meta) final_prompt = ( "No information found from knowledge database, you can provide your own Root Causes, Solution and Notes" "You are answering a user using BOTH the general chat prompt and the asset information below. " "Prioritize the asset information, as this is the source of truth for all assets in the system." "asset information:\n" f"{meta_context}\n\n" "IMPORTANT:\n" "Do not invent facts that are not available from asset information.\n" "If the context is insufficient, say so briefly or ask user to provide more context.\n" f"If information cannot be found from {meta_context}, inform user and provide information that are available.\n\n" "GENERAL CHAT PROMPT:\n" f"{req.description}" "Notes: The following insights are generated by the language model and should be validated before use" "Output template in 3 sections: Root Cause, Solution and Notes." ) response = call_tool("llm_summarize", {"prompt": final_prompt, "max_new_tokens": defaults.get("max_new_tokens")}) text = response.get('text', '') return {'summary': text} if __name__ == "__main__": agent = RCAAgent() chatbot_test = agent.run(RCARequest(workflow_flag=False, alert_id=0, asset_id='PMP09', title="RCA Request", timestamp=None, description="Show me the root cause and solution recommendation for PMP09 pump frequent trip after start for awhile", db_profile="DatabaseLocal")) print('prompt result\n', chatbot_test)