import os from pathlib import Path import gradio as gr from dotenv import load_dotenv import google.generativeai as genai from google.generativeai import types from google.protobuf.json_format import MessageToDict from google.protobuf.struct_pb2 import ListValue as ProtoListValue, Struct as ProtoStruct, Value as ProtoValue from tools import ( estimate_repair, lookup_product, record_customer_interest, record_feedback, record_service_feedback, ) load_dotenv() API_KEY = os.getenv("GEMINI_API_KEY") if not API_KEY: raise RuntimeError("GEMINI_API_KEY is not set. Create a .env file based on .env.example.") genai.configure(api_key=API_KEY) MODEL_NAME = os.getenv("GEMINI_MODEL", "gemini-2.5-flash") SYSTEM_PROMPT = Path("me/system_prompt.txt").read_text(encoding="utf-8") SUMMARY = Path("me/business_summary.txt").read_text(encoding="utf-8") FUNCTION_DECLARATIONS = [ { "name": "lookup_product", "description": ( "Search the Fix&Furn curated catalog and IKEA Saudi Arabia reference dataset. " "Return relevant catalog_match/catalog_results and ikea_results including item_id, name, " "category, price_sar, dimensions_cm, availability, and link when available." ), "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Keyword, color, category, SKU, or IKEA item ID to search for.", } }, "required": ["query"], }, }, { "name": "estimate_repair", "description": "Estimate repair price and turnaround tiers based on issue, material, and size_category.", "parameters": { "type": "object", "properties": { "issue": { "type": "string", "description": "Issue such as scratch, broken_glass, wobble, loose_joint, hinge_alignment, drawer_stick, upholstery_tear, refinish, repaint.", }, "material": { "type": "string", "description": "Primary material (wood, glass, metal, fabric, or any).", }, "size_category": { "type": "string", "description": "Furniture size bucket: small, medium, or large.", }, }, "required": ["issue"], }, }, { "name": "record_customer_interest", "description": "Capture customer details when they are ready to buy or book a repair.", "parameters": { "type": "object", "properties": { "email": {"type": "string", "description": "Customer email address."}, "name": {"type": "string", "description": "Customer full name."}, "message": {"type": "string", "description": "Short note about the product or repair request."}, }, "required": ["email", "name", "message"], }, }, { "name": "record_feedback", "description": "Log customer questions that the assistant could not resolve.", "parameters": { "type": "object", "properties": { "question": {"type": "string", "description": "Unanswered or unclear customer request."} }, "required": ["question"], }, }, { "name": "record_service_feedback", "description": "Capture post-service feedback about the overall experience, product satisfaction, or repair quality.", "parameters": { "type": "object", "properties": { "email": {"type": "string", "description": "Customer email to match the service record."}, "name": {"type": "string", "description": "Customer full name."}, "service_type": { "type": "string", "description": "What we delivered (e.g., purchase, repair, delivery, install).", }, "satisfaction": { "type": "string", "description": "Quick sentiment summary (e.g., happy, neutral, unhappy, 1-5).", }, "comments": { "type": "string", "description": "Optional free-text feedback on the experience.", }, }, "required": ["email", "name", "service_type", "satisfaction"], }, }, ] MODEL = genai.GenerativeModel( model_name=MODEL_NAME, system_instruction=f"{SYSTEM_PROMPT}\n\n{SUMMARY}", tools=[{"function_declarations": FUNCTION_DECLARATIONS}], ) GENERATION_CONFIG = types.GenerationConfig(temperature=0.2) TOOL_CONFIG = {"function_calling_config": {"mode": "AUTO"}} def _content(role: str, text: str): if not text: return None return {"role": role, "parts": [{"text": text}]} def _convert_history(history): converted = [] for user, assistant in history: if user: converted.append(_content("user", user)) if assistant: converted.append(_content("model", assistant)) return [msg for msg in converted if msg is not None] def _call_tool(name: str, args: dict): try: if name == "lookup_product": return lookup_product(**args) if name == "estimate_repair": return estimate_repair(**args) if name == "record_customer_interest": return record_customer_interest(**args) if name == "record_feedback": return record_feedback(**args) if name == "record_service_feedback": return record_service_feedback(**args) return {"ok": False, "msg": f"Unknown tool '{name}'."} except TypeError as exc: return {"ok": False, "msg": f"Invalid arguments for {name}: {exc}"} def _first_function_call(response): for candidate in response.candidates or []: if not candidate or not candidate.content: continue for part in candidate.content.parts: if part.function_call: return part.function_call return None def _proto_to_python(value): if value is None: return None if isinstance(value, ProtoValue): kind = value.WhichOneof("kind") if kind == "struct_value": return {k: _proto_to_python(v) for k, v in value.struct_value.fields.items()} if kind == "list_value": return [_proto_to_python(v) for v in value.list_value.values] if kind == "string_value": return value.string_value if kind == "number_value": return value.number_value if kind == "bool_value": return value.bool_value if kind == "null_value": return None return MessageToDict(value, preserving_proto_field_name=True) if isinstance(value, ProtoStruct): return {k: _proto_to_python(v) for k, v in value.fields.items()} if isinstance(value, ProtoListValue): return [_proto_to_python(v) for v in value.values] if isinstance(value, dict): return {k: _proto_to_python(v) for k, v in value.items()} if hasattr(value, "items"): return {k: _proto_to_python(v) for k, v in value.items()} if isinstance(value, (list, tuple)): return [_proto_to_python(v) for v in value] return value def _function_args_to_dict(function_call): args = getattr(function_call, "args", None) if args is None: return {} if isinstance(args, dict): return {k: _proto_to_python(v) for k, v in args.items()} if hasattr(args, "_pb"): try: return _proto_to_python(args._pb) except Exception: pass if hasattr(args, "ListFields"): try: return _proto_to_python(args) except Exception: pass if hasattr(args, "items"): return {k: _proto_to_python(v) for k, v in args.items()} converted = _proto_to_python(args) return converted if isinstance(converted, dict) else {} def _send_function_response(chat, name: str, payload: dict): message = { "role": "tool", "parts": [ { "function_response": { "name": name, "response": payload, } } ], } return chat.send_message(message, generation_config=GENERATION_CONFIG, tool_config=TOOL_CONFIG) def chat_fn(message, history): history_msgs = _convert_history(history) chat = MODEL.start_chat(history=history_msgs) response = chat.send_message( message, generation_config=GENERATION_CONFIG, tool_config=TOOL_CONFIG, ) while True: function_call = _first_function_call(response) if not function_call: break args = _function_args_to_dict(function_call) tool_result = _call_tool(function_call.name, args) response = _send_function_response(chat, function_call.name, tool_result) text = response.text or "" return text.strip() TITLE = "Fix&Furn Mini - Furniture Sales & Repair Concierge" demo = gr.ChatInterface(chat_fn, title=TITLE) if __name__ == "__main__": demo.launch()