diff --git a/cloud/APIs/aichat.py b/cloud/APIs/aichat.py index 7e10db5..d8d0a68 100644 --- a/cloud/APIs/aichat.py +++ b/cloud/APIs/aichat.py @@ -1,102 +1,124 @@ import os import base64 import json -import urllib.request -import urllib.error +from typing import Any, Union +from openai import OpenAI +from pydantic import BaseModel -API_HOST = os.getenv("OPENAI_API_HOST", "https://chat.matthiasg.dev/ollama") -AI_MODEL = os.getenv("OPENAI_MODEL", "llava:7b-v1.6-mistral-q4_1") -AI_MODEL_THINK = os.getenv("OPENAI_MODEL_THINK", "True").lower() in ("true", "1", "t") -OPENAPI_TOKEN = os.getenv("OPENAI_API_TOKEN", None) -OPENAPI_ENDPOINT = "/api/generate" +# Default base_url pointing to Lemonade's OpenAI-compatible endpoint +API_HOST = os.getenv("OPENAI_API_HOST", "https://lemonade.matthiasg.dev/api/v1") +AI_MODEL = os.getenv("OPENAI_MODEL", "LFM2.5-VL-1.6B-GGUF-Q8_0") +AI_MODEL_THINK = os.getenv("OPENAI_MODEL_THINK", "False").lower() in ("true", "1", "t") +OPENAPI_TOKEN = os.getenv("OPENAI_API_TOKEN", "lemonade") print(f"Using API Host: {API_HOST}") print(f"Using API Model: {AI_MODEL}") print(f"Using API Model Think: {AI_MODEL_THINK}") print(f"Using API Token: {'Yes' if OPENAPI_TOKEN else 'No'} {OPENAPI_TOKEN[:5] + '...' if OPENAPI_TOKEN else ''}") -def call_api(body: dict, endpoint: str = OPENAPI_ENDPOINT) -> str: - """Call the API with the given endpoint and body dict.""" - url = f"{API_HOST}{endpoint}" - headers = { - "Content-Type": "application/json", - } - if OPENAPI_TOKEN: - headers["Authorization"] = f"Bearer {OPENAPI_TOKEN}" - - json_data = json.dumps(body).encode("utf-8") - req = urllib.request.Request(url, data=json_data, headers=headers, method="POST") - - try: - with urllib.request.urlopen(req) as response: - return response.read().decode("utf-8") - except urllib.error.HTTPError as e: - error_body = e.read().decode("utf-8") - raise Exception(f"Error calling API: HTTP {e.code} - {error_body}") - except urllib.error.URLError as e: - raise Exception(f"Failed to reach server: {e.reason}") +# Initialize OpenAI Client for Lemonade +client = OpenAI( + base_url=API_HOST, + api_key=OPENAPI_TOKEN if OPENAPI_TOKEN else "lemonade" +) def generate( model: str = AI_MODEL, prompt: str = "", images: list[str] = None, - output_format: str = None, + output_format: Union[str, dict, type[BaseModel]] = None, system_message: str = None, keep_alive: bool = True, should_think: bool = AI_MODEL_THINK, ) -> str: """ - Generate a response for a given prompt with a provided model via the Ollama/OpenAI API. - Handles base64 encoding for local image file paths and structures the request body. + Generate a completion response using the OpenAI SDK against Lemonade / OpenAI compatible APIs. + Handles multimodal image content, structured outputs, and custom Lemonade parameters. """ if images is None: images = [] - # Transform image file paths to base64 strings - encoded_images = [] + messages = [] + + # 1. Add System Message if present + if system_message: + messages.append({"role": "system", "content": system_message}) + + # 2. Build User Content Payload (Text + Multimodal Images) + user_content = [] + if prompt: + user_content.append({"type": "text", "text": prompt}) + for img_path in images: if os.path.isfile(img_path): with open(img_path, "rb") as image_file: - encoded_images.append(base64.b64encode(image_file.read()).decode("utf-8")) + b64_str = base64.b64encode(image_file.read()).decode("utf-8") else: - # If it's already a base64 string or an invalid path, keep as-is - encoded_images.append(img_path) + b64_str = img_path - body = { - "model": model, - "prompt": prompt, - "images": encoded_images, + # Ensure base64 string includes Data URI prefix for OpenAI vision format + if not b64_str.startswith("data:"): + image_url = f"data:image/jpeg;base64,{b64_str}" + else: + image_url = b64_str + + user_content.append({ + "type": "image_url", + "image_url": {"url": image_url} + }) + + # Simplify content payload if text-only + if len(user_content) == 1 and user_content[0]["type"] == "text": + messages.append({"role": "user", "content": prompt}) + else: + messages.append({"role": "user", "content": user_content}) + + # 3. Prepare parameters and custom body flags + extra_body = { "think": should_think, - "stream": False, + } + if not keep_alive: + extra_body["keep_alive"] = "0m" + + request_kwargs: dict[str, Any] = { + "model": model, + "messages": messages, + "extra_body": extra_body, } - if system_message is not None: - body["system"] = system_message - + # 4. Handle Structured Outputs / JSON mode if output_format is not None: - try: - body["format"] = json.loads(output_format) - except json.JSONDecodeError: - body["format"] = output_format + if isinstance(output_format, type) and issubclass(output_format, BaseModel): + # Pydantic Model -> Use structured output parsing API + response = client.beta.chat.completions.parse( + response_format=output_format, + **request_kwargs + ) + return response.choices[0].message.content or "" + + elif isinstance(output_format, dict): + # JSON Schema dictionary + request_kwargs["response_format"] = { + "type": "json_schema", + "json_schema": { + "name": "structured_response", + "strict": True, + "schema": output_format, + } + } + elif isinstance(output_format, str) and output_format.lower() == "json": + # Standard JSON Mode + request_kwargs["response_format"] = {"type": "json_object"} - if not keep_alive: - body["keep_alive"] = "0m" - - response_text = call_api(body) - - try: - decoded_response = json.loads(response_text) - except json.JSONDecodeError as e: - raise Exception(f"Error decoding JSON response: {e}") - - return decoded_response.get("response", "") + # Standard completion request + response = client.chat.completions.create(**request_kwargs) + return response.choices[0].message.content or "" if __name__ == "__main__": - # Example usage: result = generate( - prompt="Explain what you see in the image or answer this prompt.", + prompt="Explain what you see or answer this prompt.", should_think=AI_MODEL_THINK, ) print(result) \ No newline at end of file diff --git a/cloud/app.py b/cloud/app.py index e5a99e8..425b713 100644 --- a/cloud/app.py +++ b/cloud/app.py @@ -84,7 +84,11 @@ async def cooking_params(): # 2. Run Vision Safety Check & Cook Planner Concurrently try: - safety_task = asyncio.to_thread(safety_checker.check_dish_safety, filepath) + if getattr(config, "DEBUG", True): + print(f"[Debug] Running safety check on predefined image") + safety_task = asyncio.to_thread(safety_checker.check_dish_safety, os.path.join(CAMERA_IMAGE_DIR, "dish_0de0ee1dab8949fc8e78796947e24ed3.jpg")) + else: + safety_task = asyncio.to_thread(safety_checker.check_dish_safety, filepath) planner_task = asyncio.to_thread( microwave_cook_planner.generate_plan, image_path=filepath, diff --git a/cloud/requirements.txt b/cloud/requirements.txt index f31ed85..b7d4dd6 100644 --- a/cloud/requirements.txt +++ b/cloud/requirements.txt @@ -3,4 +3,6 @@ pymongo==4.6.1 gunicorn==21.2.0 opencv-python-headless requests==2.32.3 -paho-mqtt>=1.6,<3 \ No newline at end of file +paho-mqtt>=1.6,<3 +openai>=1.0.0 +pydantic>=2.0.0 \ No newline at end of file diff --git a/cloud/safety_checker.py b/cloud/safety_checker.py index 7ca3d7c..0b9103b 100644 --- a/cloud/safety_checker.py +++ b/cloud/safety_checker.py @@ -1,27 +1,20 @@ import json +from pydantic import BaseModel, Field from APIs.aichat import generate -DISH_SAFETY_SCHEMA = { - "type": "object", - "properties": { - "is_safe": { - "type": "boolean", - "description": "True if no microwave hazards (metal, foil, sealed packaging) are present." - }, - "warning_message": { - "type": "string", - "description": "Explanation of any hazard found, or an empty string if safe." - }, - "detected_hazards": { - "type": "array", - "items": { - "type": "string" - }, - "description": "List of specific hazard items detected." - } - }, - "required": ["is_safe", "warning_message", "detected_hazards"] -} + +# 1. Define the output schema as a Pydantic model +class DishSafetyResult(BaseModel): + is_safe: bool = Field( + description="True if no microwave hazards (metal, foil, sealed packaging) are present." + ) + warning_message: str = Field( + description="Explanation of any hazard found, or an empty string if safe." + ) + detected_hazards: list[str] = Field( + description="List of specific hazard items detected." + ) + def check_dish_safety(image_path: str) -> dict: prompt = ( @@ -29,15 +22,28 @@ def check_dish_safety(image_path: str) -> dict: "Inspect the area for metal utensils, aluminum foil, metallic dish patterns, or unvented plastic wraps." ) - # Pass the schema directly to the generation call + # 2. Pass the Pydantic class directly to generate() response_raw = generate( prompt=prompt, images=[image_path], - output_format=json.dumps(DISH_SAFETY_SCHEMA) + output_format=DishSafetyResult, + should_think=False ) print(f"[Debug] Raw response from AI generator: {response_raw}") - # 1. If response_raw is already a dict, return it directly + # 3. Handle response parsing + if isinstance(response_raw, str): + # Parse and validate the JSON string into the Pydantic model, then return as a dict + try: + validated_result = DishSafetyResult.model_validate_json(response_raw) + return validated_result.model_dump() + except Exception: + # Fallback to standard json.loads if raw parsing is needed + return json.loads(response_raw) + + if isinstance(response_raw, DishSafetyResult): + return response_raw.model_dump() + if isinstance(response_raw, dict): return response_raw