""" Quote creation tool that generates professional quotes in Markdown format using product IDs from database. This tool queries products by ID and creates structured quotes with proper Markdown tables. """ import os from datetime import datetime from typing import Union, List, Dict, Any from collections import Counter from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage from dotenv import load_dotenv from .quote_template import ( Quote, QuoteItem, CustomerInfo, generate_markdown_quote, GREETING_PROMPT, INTRO_PROMPT, generate_product_description, generate_product_name, generate_product_id ) from .get_exchange_rates import convert_amount from .agent_tools_utils import get_data_for_agent from .tool_schemas import QuoteInput, QuoteOutput, QuoteItemSummary, ErrorOutput # Load environment variables load_dotenv() def fetch_products_by_ids(product_ids: List[int], target_currency: str = "EUR") -> List[Dict[str, Any]]: """ Fetch product details from database by product IDs and convert prices to target currency. Args: product_ids: List of product IDs to fetch target_currency: Currency to convert prices to Returns: List of product dictionaries with converted prices """ try: # Query to get product details including model numbers product_ids_str = ",".join(map(str, product_ids)) query = f""" SELECT id, manufacturer, model_name, model_number_long, model_number_short, description, msrp, currency, category, sub_category FROM streamnet.products_list WHERE id IN ({product_ids_str}) """ # Execute query df_original, df_formatted = get_data_for_agent(query, return_type="dict_records") if not df_formatted or len(df_formatted) == 0: return [] # Process and convert currencies products = [] for product in df_formatted: original_currency = product.get('currency', 'EUR') msrp_price = float(product.get('msrp', 0)) # Convert currency if needed if original_currency != target_currency and msrp_price > 0: conversion_result = convert_amount( original_currency, target_currency, msrp_price ) if hasattr(conversion_result, 'converted_amount'): converted_price = conversion_result.converted_amount else: converted_price = msrp_price # Use original price else: converted_price = msrp_price # Build product name product_name = product.get('model_name', '') or product.get('model_number_long', '') or f"Product {product.get('id')}" products.append({ 'id': product.get('id'), 'product_name': product_name, 'model_number_short': product.get('model_number_short'), 'model_number_long': product.get('model_number_long'), 'unit_price': converted_price, 'currency': target_currency, 'original_currency': original_currency, 'manufacturer': product.get('manufacturer', ''), 'category': product.get('category', ''), 'sub_category': product.get('sub_category', ''), 'original_description': product.get('description', '') }) return products except Exception as e: print(f"Error fetching products: {e}") return [] def generate_dynamic_content(prompt: str, max_retries: int = 2) -> str: """ Generate dynamic content using LLM. Args: prompt: The prompt for content generation max_retries: Number of retry attempts if generation fails Returns: Generated content string """ try: # Initialize the LLM model = ChatOpenAI( model=os.getenv("MODEL_NAME", "gpt-4o-mini"), api_key=os.getenv("OPENAI_API_KEY"), temperature=0.3 # Slightly creative but controlled ) # Generate content response = model.invoke([HumanMessage(content=prompt)]) return response.content.strip() except Exception as e: print(f"Warning: LLM content generation failed: {e}") # Fallback to generic content if "greeting" in prompt.lower(): return "Dear Valued Customer, thank you for your interest in our products." else: return "We are pleased to provide you with this detailed quotation for your consideration." def create_quote_file(quote: Quote, content: str) -> str: """ Save the quote to a Markdown file in the created_quotes directory. Args: quote: Quote object with metadata content: The formatted quote content Returns: Path to the saved file """ # Get the directory of the current file and navigate to created_quotes current_dir = os.path.dirname(os.path.abspath(__file__)) quotes_dir = os.path.join(current_dir, "..", "created_quotes") quotes_dir = os.path.abspath(quotes_dir) # Resolve to absolute path os.makedirs(quotes_dir, exist_ok=True) # Generate unique filename timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") customer_name = quote.customer.name.replace(" ", "_").replace("/", "_") filename = f"quote_{quote.quote_id}_{customer_name}_{timestamp}.md" filepath = os.path.join(quotes_dir, filename) # Write the quote to file with open(filepath, 'w', encoding='utf-8') as f: f.write(content) return filepath @tool(args_schema=QuoteInput) def create_quote( product_ids: List[int], customer_name: str, customer_email: str = None, customer_company: str = None, target_currency: str = "EUR", notes: str = None ) -> Union[QuoteOutput, ErrorOutput]: """ A quote creation tool that generates professional quotes using product IDs from the database. This tool takes a list of product IDs (can include duplicates for multiple quantities), fetches product details from the database, and generates a professional quote with proper Markdown tables and formatting. This tool only works if the products have prices set in the database. Args: product_ids (List[int]): List of product IDs from database. Duplicates indicate multiple quantities. Example: [1, 1, 1, 3, 4, 5] means 3x product ID 1, 1x product ID 3, 1x product ID 4, 1x product ID 5 customer_name (str): Customer's full name (required) customer_email (Optional[str]): Customer's email address customer_company (Optional[str]): Customer's company name target_currency (str): Currency for the final quote (EUR, USD, HUF) notes (Optional[str]): Additional notes for the quote Example usage: create_quote( product_ids=[123, 123, 456, 789], # 2x product 123, 1x product 456, 1x product 789 customer_name="John Smith", customer_email="john@company.com", customer_company="Tech Solutions Inc", target_currency="EUR" ) Returns: QuoteOutput: Quote details with file path and summary information. """ try: # Pydantic validation handled by LangGraph automatically # Count quantities for each product ID product_quantities = Counter(product_ids) unique_product_ids = list(product_quantities.keys()) # Fetch product details from database products_data = fetch_products_by_ids(unique_product_ids, target_currency) if not products_data: return ErrorOutput(error="No products found for the provided IDs") # Create customer info customer = CustomerInfo( name=customer_name, email=customer_email, company=customer_company ) # Initialize LLM for description generation llm = ChatOpenAI( model=os.getenv("MODEL_NAME", "gpt-4o-mini"), api_key=os.getenv("OPENAI_API_KEY"), temperature=0.3 ) # Create quote items with quantities and LLM-generated descriptions, names, and IDs quote_items = [] for product_data in products_data: product_id = product_data['id'] quantity = product_quantities[product_id] # Prepare product info for AI generation product_info = { 'id': product_data['id'], 'manufacturer': product_data['manufacturer'], 'model_name': product_data['product_name'], 'model_number_short': product_data['model_number_short'], 'model_number_long': product_data['model_number_long'], 'category': product_data['category'], 'sub_category': product_data['sub_category'], 'description': product_data['original_description'] } # Generate AI-enhanced product name ai_product_name = generate_product_name(product_info, llm) # Generate AI-selected product ID ai_product_id = generate_product_id(product_info, llm) # Generate concise description using LLM llm_description = generate_product_description(product_info, llm) quote_item = QuoteItem( product_id=product_id, # Keep database ID for internal reference product_name=ai_product_name, # AI-generated name model_number_short=ai_product_id, # AI-selected ID for display model_number_long=product_data['model_number_long'], description=llm_description, quantity=quantity, unit_price=product_data['unit_price'], currency=target_currency ) quote_items.append(quote_item) # Create the quote object quote = Quote( customer=customer, items=quote_items, currency=target_currency, notes=notes ) # Generate LLM content company_info = f" from {customer_company}" if customer_company else "" # Generate greeting greeting_prompt = GREETING_PROMPT.format( customer_name=customer_name, company_info=company_info, company=customer_company or "N/A" ) greeting = generate_dynamic_content(greeting_prompt) # Generate introduction detailed_product_list = [] for item in quote_items: if item.quantity > 1: detailed_product_list.append(f"{item.quantity}x {item.product_name}") # Using AI-generated name else: detailed_product_list.append(item.product_name) # Using AI-generated name product_list_str = ", ".join(detailed_product_list) intro_prompt = INTRO_PROMPT.format( customer_name=customer_name, company=customer_company or "your organization", detailed_product_list=product_list_str, unique_product_count=len(unique_product_ids), total_item_count=sum(product_quantities.values()) ) introduction = generate_dynamic_content(intro_prompt) # Generate the complete Markdown quote quote_template = generate_markdown_quote(quote) final_quote = quote_template.format( greeting=greeting, introduction=introduction ) # Save to Markdown file file_path = create_quote_file(quote, final_quote) # Create quote summary with Pydantic models quote_summary_items = [ QuoteItemSummary( product_id=item.product_id, name=item.product_name, quantity=item.quantity, unit_price=item.unit_price, total=item.total_price ).dict() for item in quote_items ] return QuoteOutput( success=True, quote_id=quote.quote_id, customer_name=customer_name, grand_total=quote.grand_total, currency=target_currency, item_count=len(quote_items), unique_products=len(unique_product_ids), total_items=sum(product_quantities.values()), file_path=file_path, file_format="markdown", created_date=quote.created_date.isoformat(), valid_until=quote.valid_until.isoformat(), quote_summary={ "grand_total": quote.grand_total, "items": quote_summary_items } ) except Exception as e: return ErrorOutput(error=f"Quote creation failed: {str(e)}") def create_sample_quote_from_ids(): """Create a sample quote using product IDs for testing purposes.""" # Sample with some repeated IDs to test quantity handling sample_product_ids = [760, 760, 763, 765, 765, 768] # 2x product 760, 1x product 763, 2x product 765, 1x product 768 result = create_quote.invoke({ "product_ids": sample_product_ids, "customer_name": "Jane Doe", "customer_email": "jane.doe@example.com", "customer_company": "Example Corp", "target_currency": "EUR", "notes": "Sample quote generated from product IDs" }) return result