""" Tools Service for LLM Function Calling HuggingFace-compatible với prompt engineering """ import httpx from typing import List, Dict, Any, Optional import json import asyncio class ToolsService: """ Manages external API tools that LLM can call via prompt engineering """ def __init__(self, base_url: str = "https://www.festavenue.site"): self.base_url = base_url self.client = httpx.AsyncClient(timeout=10.0) def get_tools_prompt(self) -> str: """ Return prompt instruction for HuggingFace LLM về available tools """ return """ AVAILABLE TOOLS: Bạn có thể sử dụng các công cụ sau để lấy thông tin chi tiết: 1. get_event_details(event_code: str) - Mô tả: Lấy thông tin đầy đủ về một sự kiện từ hệ thống - Khi nào dùng: Khi user hỏi về ngày giờ chính xác, địa điểm cụ thể, thông tin liên hệ, hoặc chi tiết khác về một sự kiện - Tham số: event_code = ID sự kiện (LẤY TỪ metadata.id_use TRONG CONTEXT, KHÔNG PHẢI tên sự kiện!) VÍ DỤ QUAN TRỌNG: Context có: ``` metadata: { "id_use": "69194cf61c0eda56688806f7", ← DÙNG CÁI NÀY! "texts": ["Y-CONCERT - Festival âm nhạc..."] } ``` → Dùng event_code = "69194cf61c0eda56688806f7" (NOT "Y-CONCERT") CÚ PHÁP GỌI TOOL: Khi bạn cần gọi tool, hãy trả lời CHÍNH XÁC theo format JSON này: ```json { "tool_call": true, "function_name": "get_event_details", "arguments": { "event_code": "69194cf61c0eda56688806f7" }, "reason": "Cần lấy thông tin chính xác về ngày giờ tổ chức" } ``` QUAN TRỌNG: - event_code PHẢI LÀ metadata.id_use từ context (dạng MongoDB ObjectId) - KHÔNG dùng tên sự kiện như "Y-CONCERT" làm event_code - CHỈ trả JSON khi BẮT BUỘC cần gọi tool - Nếu có thể trả lời từ context sẵn có, đừng gọi tool - Sau khi nhận kết quả từ tool, hãy trả lời user bằng ngôn ngữ tự nhiên """ async def parse_and_execute(self, llm_response: str) -> Optional[Dict[str, Any]]: """ Parse LLM response và execute tool nếu có Returns: None nếu không có tool call Dict với tool result nếu có tool call """ # Try to extract JSON from response try: # Tìm JSON block trong response if "```json" in llm_response: json_start = llm_response.find("```json") + 7 json_end = llm_response.find("```", json_start) json_str = llm_response[json_start:json_end].strip() elif "{" in llm_response and "}" in llm_response: # Fallback: tìm JSON object đầu tiên json_start = llm_response.find("{") json_end = llm_response.rfind("}") + 1 json_str = llm_response[json_start:json_end] else: return None tool_call = json.loads(json_str) # Handle multiple JSON formats from LLM # Format 1: HF API nested wrapper # {"name": "tool_call", "arguments": {"tool_call": true, ...}} if "name" in tool_call and "arguments" in tool_call and isinstance(tool_call["arguments"], dict): if "tool_call" in tool_call["arguments"]: tool_call = tool_call["arguments"] # Unwrap # Format 2: Direct tool name format # {"name": "tool.get_event_details", "arguments": {"event_code": "..."}} if "name" in tool_call and "arguments" in tool_call: function_name = tool_call["name"] # Remove "tool." prefix if exists if function_name.startswith("tool."): function_name = function_name.replace("tool.", "") # Convert to standard format tool_call = { "tool_call": True, "function_name": function_name, "arguments": tool_call["arguments"], "reason": "Converted from alternate format" } # Validate tool call structure if not tool_call.get("tool_call"): return None function_name = tool_call.get("function_name") arguments = tool_call.get("arguments", {}) # Execute tool if function_name == "get_event_details": result = await self._get_event_details(arguments.get("event_code")) return { "function": function_name, "arguments": arguments, "result": result } else: return { "function": function_name, "arguments": arguments, "result": {"success": False, "error": f"Unknown function: {function_name}"} } except (json.JSONDecodeError, KeyError, ValueError) as e: # Không phải tool call, response bình thường return None async def _get_event_details(self, event_code: str) -> Dict[str, Any]: """ Call getEventByEventCode API """ print(f"\n=== CALLING API get_event_details ===") print(f"Event Code: {event_code}") try: url = f"https://hoalacrent.io.vn/api/v0/event/get-event-by-event-code" params = {"eventCode": event_code} print(f"URL: {url}") print(f"Params: {params}") response = await self.client.get(url, params=params) print(f"Status Code: {response.status_code}") # Log raw response for debugging raw_text = response.text print(f"Raw Response Length: {len(raw_text)} chars") print(f"Raw Response Preview (first 200 chars): {raw_text[:200]}") response.raise_for_status() # Try to parse JSON try: data = response.json() except json.JSONDecodeError as e: print(f"JSON Decode Error: {e}") print(f"Full Raw Response: {raw_text}") return { "success": False, "error": f"Invalid JSON response from API", "message": "API trả về dữ liệu không hợp lệ (không phải JSON)", "raw_response_preview": raw_text[:500] } print(f"Response Data Keys: {list(data.keys()) if data else 'None'}") print(f"Has 'data' field: {'data' in data}") # Extract relevant fields event = data.get("data", {}) if not event: return { "success": False, "error": "Event not found", "message": f"Không tìm thấy sự kiện với mã {event_code}" } # Extract location với nested address structure location_data = event.get("location", {}) location = { "address": { "street": location_data.get("address", {}).get("street", ""), "city": location_data.get("address", {}).get("city", ""), "state": location_data.get("address", {}).get("state", ""), "postalCode": location_data.get("address", {}).get("postalCode", ""), "country": location_data.get("address", {}).get("country", "") }, "coordinates": { "latitude": location_data.get("coordinates", {}).get("latitude"), "longitude": location_data.get("coordinates", {}).get("longitude") } } # Build event URL event_code = event.get("eventCode") event_url = f"https://www.festavenue.site/user/event/{event_code}" if event_code else None return { "success": True, "event_code": event_code, "event_name": event.get("eventName"), "event_url": event_url, # NEW: Direct link to event page "description": event.get("description"), "short_description": event.get("shortDescription"), "start_time": event.get("startTimeEventTime"), "end_time": event.get("endTimeEventTime"), "start_sale": event.get("startTicketSaleTime"), "end_sale": event.get("endTicketSaleTime"), "location": location, # Full nested structure "contact": { "email": event.get("publicContactEmail"), "phone": event.get("publicContactPhone"), "website": event.get("website") }, "capacity": event.get("capacity"), "hashtags": event.get("hashtags", []) } print(f"Successfully extracted event data for: {event.get('eventName')}") print(f"=== API CALL COMPLETE ===") return result except httpx.HTTPStatusError as e: return { "success": False, "error": f"HTTP {e.response.status_code}", "message": f"API trả về lỗi khi truy vấn sự kiện {event_code}" } except Exception as e: return { "success": False, "error": str(e), "message": "Không thể kết nối đến API để lấy thông tin sự kiện" } async def close(self): """Close HTTP client""" await self.client.aclose()