""" Agent tools for the Karishye AI agent. This module contains all tool functions that the karishye_agent can call: - search_structured_knowledge_base: Search Karishye knowledge base - get_puja_information_and_pricing: Get puja pricing, samagri, and detailed info from API - generate_booking_confirmation: Generate final booking confirmation with URL """ import logging import re from pydantic_ai import RunContext from app.agents.agents import karishye_agent, general_agent, ChatContext from app.constants import needs_muhurtam_question from app.utils.puja_api_cache import get_puja_info logger = logging.getLogger(__name__) @karishye_agent.tool_plain async def search_structured_knowledge_base(query: str) -> str: """ NEW: Search the structured Karishye knowledge base using intelligent file navigation. This tool: 1. First reads the index.md to understand the knowledge base structure 2. Uses AI to determine which specific MD files are most relevant to the query 3. Reads those specific files and returns their content 4. Provides comprehensive, structured information from official documentation Use this tool for ANY question about: - Karishye's vision, mission, goals, values - Pujaris, their role, selection, and benefits - Specific homams/pujas (Ganapati, Saraswati, Maha Mrityunjaya, etc.) - Services, approach, quality standards - Sanatana Dharma concepts - My Dharma platform Args: query: User's question or topic to search for Returns: Relevant content from knowledge base files """ try: import re from pathlib import Path # Dynamically resolve knowledge base path relative to project root # This works regardless of where the project is deployed current_file = Path(__file__).resolve() # Path to this file project_root = current_file.parent.parent.parent # Go up: agents -> app -> project_root kb_base_path = project_root / "knowledge-base" index_path = kb_base_path / "index.md" # Step 1: Read the index to understand structure if not index_path.exists(): return "Knowledge base index not found. Please ensure the knowledge base is properly set up." with index_path.open('r', encoding='utf-8') as f: index_content = f.read() # Step 2: Use AI to determine relevant files selection_prompt = f"""Based on this knowledge base index structure and the user's query, identify the MOST relevant markdown files to read. KNOWLEDGE BASE INDEX: {index_content} USER QUERY: {query} Analyze the query and return ONLY a JSON array of file paths relative to the knowledge base root. Consider: - Direct topic matches (e.g., "Ganapati Homam" → 04-services/03-ganapati-homam.md) - Related topics (e.g., "health puja" → health-related homam files) - Conceptual matches (e.g., "why Karishye" → vision/mission files) - Multiple relevant files if the query is broad Return 1-4 most relevant files. Format: ["path/file.md", "path/file.md"] If query is about multiple homams or general services, include the overview file. RESPOND WITH ONLY THE JSON ARRAY, NO OTHER TEXT.""" # Get file suggestions from AI file_selection_result = await general_agent.run(selection_prompt) file_paths_text = file_selection_result.output.strip() # Parse the JSON array file_paths = [] try: # Extract JSON array from the response json_match = re.search(r'\[(.*?)\]', file_paths_text, re.DOTALL) if json_match: import json as json_lib file_paths = json_lib.loads(f"[{json_match.group(1)}]") except Exception as parse_error: logger.warning(f"Failed to parse file paths: {parse_error}") # Fallback: try to extract paths manually file_paths = re.findall(r'["\']([^"\']+\.md)["\']', file_paths_text) if not file_paths: logger.debug(f"No files identified for query") return "I couldn't identify the specific files to answer your question. Let me search the general knowledge base." logger.debug(f"Identified {len(file_paths)} relevant files") # Step 3: Read the identified files retrieved_content = [] for file_path in file_paths[:4]: # Limit to 4 files max full_path = kb_base_path / file_path.strip() if full_path.exists(): try: with full_path.open('r', encoding='utf-8') as f: content = f.read() retrieved_content.append(f"=== FROM: {file_path} ===\n{content}\n") except Exception as read_error: logger.error(f"Error reading {file_path}: {read_error}") else: logger.warning(f"File not found: {full_path}") if not retrieved_content: return "I found relevant topics but couldn't read the files. Please try rephrasing your question." # Step 4: Combine and return content combined_content = "\n\n".join(retrieved_content) # Add helpful context about the source result = f"""[Knowledge Base Information] {combined_content} [End of Knowledge Base Information] This information is from Karishye's official knowledge base. Use it to provide a comprehensive, accurate answer to the user's question.""" return result except Exception as e: logger.error(f"[KB TOOL] Error: {type(e).__name__}") return "I'm having trouble accessing the knowledge base right now. Please try again in a moment." @karishye_agent.tool_plain async def get_puja_information_and_pricing(query: str, include_pricing: bool = False) -> str: """ Get puja information from Karishye's backend API - with SPIRITUAL-FIRST approach. **CRITICAL: Use include_pricing parameter correctly:** - include_pricing=False (DEFAULT): Returns ONLY spiritual/ritual details (NO pricing) - include_pricing=True: Returns spiritual details + pricing breakdown **TOKEN-OPTIMIZED APPROACH:** This tool uses an intelligent multi-tier search strategy to minimize token usage: TIER 1 - Simple Keyword Matching (FREE, INSTANT): - Extracts lightweight index (just names, IDs, categories) from cache - Uses fast keyword matching on index - Works for 90% of queries like "Ganapati Homam", "Naamakaranam", etc. - Zero LLM tokens used TIER 2 - LLM Semantic Matching (MINIMAL TOKENS): - Triggered only if keyword search finds nothing - Sends only lightweight index to LLM (not full data) - LLM intelligently matches query to puja IDs - Example: "puja for new baby" → finds "Naamakaranam" TIER 3 - Detailed Fetch (OPTIMIZED): - Fetches full details only for matched pujas - Returns rich information: ritual details, duration, samagri (and pricing if requested) This saves 80-90% of tokens compared to sending all puja data to LLM! **When to use include_pricing=False (DEFAULT):** - "Tell me about [puja]" - "What is [puja]?" - "Explain [puja] procedure" - "Benefits of [puja]" - "What happens in [puja]?" **When to use include_pricing=True:** - "How much is [puja]?" - "What's the cost of [puja]?" - "Price for [puja]" - "I want to book [puja]" Args: query: User's question about puja Example: "Ganapati Homam", "puja for new baby", "Rudrabhishekam procedure" include_pricing: If False (default), returns only spiritual/ritual details. If True, includes pricing breakdown. Returns: Formatted puja information (spiritual details always; pricing only if requested) """ try: # Enable LLM for semantic matching when keyword search fails # This helps with queries like "puja for new baby" → Naamakaranam result = await get_puja_info(query, use_llm=True, include_pricing=include_pricing) return result except Exception as e: # Log without stack trace to avoid exposing query content logger.error(f"[PUJA API TOOL] Error: {type(e).__name__}") return f"I apologize, but I'm having trouble accessing the puja information right now. Please try again in a moment, or contact our support team for details." @karishye_agent.tool_plain async def get_puja_advance_payment_url(puja_key: str) -> str: """ Get the advance payment URL for a specific puja. **WHEN TO USE:** - When user asks "where do I pay?" or "send me the payment link" - When you need to share the payment URL at any point in the conversation - When user wants to proceed but you're still collecting some details - When user asks about the booking/payment process - Can be used BEFORE collecting all details if user specifically requests the link **URL LOGIC:** - Returns puja-specific URL from PUJA_BOOKING_URLS if available - Falls back to general collection page if puja-specific URL not found - Always returns a valid URL (never empty) Args: puja_key: The internal puja key (e.g., "ganapathi_homam", "gruhapravesham") Returns: The payment URL for the specified puja Example: url = await get_puja_advance_payment_url("ganapathi_homam") # Returns: "https://karishye.co/products/ganapati-homam-telugu" """ from app.constants import get_puja_booking_url, DEFAULT_BOOKING_URL # Get puja-specific booking URL from constants booking_url = get_puja_booking_url(puja_key) # Validate URL (should never be empty due to DEFAULT_BOOKING_URL fallback) if not booking_url or not booking_url.strip(): logger.error(f"[TOOL:get_payment_url] Empty URL for puja_key: '{puja_key}', using default") booking_url = DEFAULT_BOOKING_URL logger.info(f"[TOOL:get_payment_url] Retrieved URL for '{puja_key}': {booking_url}") return booking_url @karishye_agent.tool async def generate_booking_confirmation(ctx: RunContext[ChatContext], puja_name: str, puja_key: str) -> str: """ Generate a professional confirmation message when ALL details are collected. **CRITICAL: CALL THIS TOOL ONLY ONCE PER PUJA** - Once you've sent the booking confirmation for a specific puja, DO NOT call this tool again for that SAME puja - If user asks about customization/requirements for a puja AFTER its confirmation is sent, reassure them and reference the payment link already shared for that specific puja - DO NOT regenerate confirmations for the same puja - **Multiple pujas OK**: User can book multiple different pujas - each puja gets ONE confirmation **WHEN TO USE:** - When you have collected ALL required information for a specific puja: * Location (city/area in AP/TG) * Date (preferred date/timeframe) * Muhurtam (if needed for this puja type) - User has completed providing all required information for this puja - You're ready to send the final confirmation with payment link - **AND you haven't sent a confirmation for THIS specific puja yet** **DO NOT USE:** - When still collecting information for this puja - When missing any required fields (location, date, or muhurtam if needed) - Just to get the payment URL (use get_puja_advance_payment_url instead) - **When you've already sent a booking confirmation for THIS specific puja** **NOTE:** This tool automatically calls get_puja_advance_payment_url to fetch the URL. Args: puja_name: The display name of the puja (e.g., "Ganapati Homam", "Gruhapravesha Puja") puja_key: The internal puja key (e.g., "ganapathi_homam", "gruhapravesham") Returns: A warm, professional confirmation message with booking URL and next steps """ # Get the payment URL using the dedicated tool booking_url = await get_puja_advance_payment_url(puja_key) logger.info(f"[TOOL:generate_confirmation] Generating confirmation for '{puja_name}' with URL: {booking_url}") response_parts = [ "🙏 Thank you for providing all the details for your sacred {puja_name} with such reverence. May this divine ritual bring abundant blessings to you and your loved ones.".format(puja_name=puja_name), "", "**Next Sacred Steps:**", "", "To **secure your preferred slot/date** and start the coordination, kindly proceed with the confirmation:", "", "**Confirmation Link:**", "[Complete Your Arrangement]({url})".format(url=booking_url), "", "**What Happens Next:**", "", "• **Minimal Advance Payment:** A small token amount to reserve your sacred date and pujari (fully refundable, no questions asked if you change your mind)", "", "• **Personal Connect:** Our dedicated team will reach out **at the earliest** to confirm all arrangements and help with muhurtam if needed", "", "• **Final Preparations:** We'll guide you through any preparatory requirements and answer all your questions", "", "", "**Our Sacred Promise to You:**", "", "At Karishye, every puja is performed with the highest spiritual integrity. Our Pujaris are chosen not just for their knowledge, but for their:", "", "• **Deep Reverence** – Profound understanding of Vedic scriptures and sacred traditions", "", "• **Pure Conduct** – Adherence to _tapas_, _brahmacharya_, and righteous living", "", "• **Compassionate Hearts** – Gentle, selfless nature that radiates divine grace", "", "• **Spiritual Authority** – Years of training in proper ritual execution and mantra pronunciation", "", "Each ritual is conducted with complete _shraddha_ (devotion) to invoke the blessings of the Divine for your family's spiritual upliftment.", "", "", "**We Are Here for You:**", "", "Should you have any questions about ritual preparations, samagri requirements, or the sacred significance of any aspect, please don't hesitate to ask. We're honored to guide you on this spiritual journey.", "", "**May the Divine bless your endeavor with success and prosperity.**", "", "ఓం శాంతిః శాంతిః శాంతిః 🙏" ] final_message = "\n".join(response_parts) return final_message