Spaces:

build-small-hackathon
/

tiny-trigger

Sleeping

Javier Montalvo

MVP working. Missing: add cloud providers

f005306 5 months ago

11.3 kB

from __future__ import annotations

import json
import re
from typing import Any

from pydantic import BaseModel, ValidationError

from .automation import AutomationDocument, automation_schema


SYSTEM_PROMPT = """You compile home automation requests into Tiny Trigger rules.
Return JSON only. Never return code, markdown, explanations, or tool calls.
The root object MUST include a non-empty "rules" array.
Use video conditions in when: present, count, near, far.
Use state gates in gate: enabled, cooldown.
Use trigger.on for edge behavior: while, enter, exit, change.
Use only these action types: simulate, webhook.
Use trigger.on="enter" for state assertions like "must be on", "should be on", "keep on", "turn on when", or "notify when".
Use trigger.on="while" only when the user explicitly wants repeated actions while a condition remains true, usually with a cooldown.
When the request says one object is near, next to, beside, at, by, close to, or in front of another object, you MUST emit a near condition.
Do not replace a near relation with two present conditions.
Use max_gap_percent for near/far box-edge distance. It is the largest horizontal/vertical edge gap between boxes in normalized frame percent; touching or overlapping boxes have gap 0.
If the user mentions elapsed time since an action or limiting repeat fires, encode it as gate.cooldown.
If the user asks for an action when a condition starts and another action when it stops, use trigger.on="change" and then.enter / then.exit.
"""


class LLMCompileResult(BaseModel):
    raw_text: str
    document: AutomationDocument


def compile_automation_with_llamacpp(
    *,
    instruction: str,
    class_names: list[str],
    base_url: str = "http://127.0.0.1:8080/v1",
    model: str = "ggml-org/Qwen3-1.7B-GGUF:Q4_K_M",
    timeout: float = 60.0,
) -> LLMCompileResult:
    """Compile natural language into validated rules through llama.cpp server."""
    try:
        import requests
    except ImportError as exc:  # pragma: no cover - dependency guard
        raise RuntimeError("Install requests to use the llama.cpp compiler.") from exc

    endpoint = base_url.rstrip("/") + "/chat/completions"
    user_prompt = _build_user_prompt(instruction=instruction, class_names=class_names)
    raw_text = _post_chat_completion(
        endpoint=endpoint,
        model=model,
        user_prompt=user_prompt,
        timeout=timeout,
        requests_module=requests,
    )
    try:
        return _validate_compile_result(raw_text)
    except (json.JSONDecodeError, ValidationError, ValueError) as first_error:
        repair_prompt = _build_repair_prompt(
            original_prompt=user_prompt,
            bad_response=raw_text,
            error=str(first_error),
        )
        repaired_text = _post_chat_completion(
            endpoint=endpoint,
            model=model,
            user_prompt=repair_prompt,
            timeout=timeout,
            requests_module=requests,
        )
        try:
            return _validate_compile_result(repaired_text)
        except (json.JSONDecodeError, ValidationError, ValueError) as second_error:
            raise ValueError(
                "llama.cpp returned invalid automation JSON after a repair attempt. "
                f"Last validation error: {second_error}. Last response: {repaired_text}"
            ) from second_error


def compile_automation_with_replicate(
    *,
    instruction: str,
    class_names: list[str],
    api_token: str,
    model: str = "openai/gpt-5.2",
    reasoning_effort: str = "medium",
    timeout: float = 600.0,
) -> LLMCompileResult:
    """Compile natural language into validated rules through Replicate."""
    try:
        import replicate
    except ImportError as exc:  # pragma: no cover - dependency guard
        raise RuntimeError("Install replicate to use the Replicate compiler.") from exc

    user_prompt = _build_user_prompt(instruction=instruction, class_names=class_names)
    raw_text = _stream_replicate_completion(
        model=model,
        prompt=_provider_prompt(user_prompt),
        api_token=api_token,
        reasoning_effort=reasoning_effort,
        timeout=timeout,
        replicate_module=replicate,
    )
    try:
        return _validate_compile_result(raw_text)
    except (json.JSONDecodeError, ValidationError, ValueError) as first_error:
        repair_prompt = _build_repair_prompt(
            original_prompt=user_prompt,
            bad_response=raw_text,
            error=str(first_error),
        )
        repaired_text = _stream_replicate_completion(
            model=model,
            prompt=_provider_prompt(repair_prompt),
            api_token=api_token,
            reasoning_effort=reasoning_effort,
            timeout=timeout,
            replicate_module=replicate,
        )
        try:
            return _validate_compile_result(repaired_text)
        except (json.JSONDecodeError, ValidationError, ValueError) as second_error:
            raise ValueError(
                "Replicate returned invalid automation JSON after a repair attempt. "
                f"Last validation error: {second_error}. Last response: {repaired_text}"
            ) from second_error


def _provider_prompt(user_prompt: str) -> str:
    return f"{SYSTEM_PROMPT}\n\n{user_prompt}\n\nReturn only the JSON object."


def _stream_replicate_completion(
    *,
    model: str,
    prompt: str,
    api_token: str,
    reasoning_effort: str,
    timeout: float,
    replicate_module: Any,
) -> str:
    _split_replicate_model(model)
    payload = {
        "prompt": prompt,
        "messages": [],
        "verbosity": "medium",
        "reasoning_effort": reasoning_effort,
    }
    client = replicate_module.Client(api_token=api_token)
    chunks: list[str] = []
    for event in client.stream(model, input=payload):
        chunks.append(str(event))
    text = "".join(chunks).strip()
    if not text:
        raise ValueError("Replicate stream returned no output.")
    return text


def _split_replicate_model(model: str) -> tuple[str, str]:
    parts = model.strip().split("/", 1)
    if len(parts) != 2 or not all(parts):
        raise ValueError("Replicate model must be in owner/model format, for example openai/gpt-5.2.")
    return parts[0], parts[1]


def _post_chat_completion(
    *,
    endpoint: str,
    model: str,
    user_prompt: str,
    timeout: float,
    requests_module: Any,
) -> str:
    payload = _chat_payload(model=model, user_prompt=user_prompt, response_format="json_object")
    response = requests_module.post(endpoint, json=payload, timeout=timeout)
    response.raise_for_status()
    body = response.json()
    return body["choices"][0]["message"]["content"]


def _validate_compile_result(raw_text: str) -> LLMCompileResult:
    data = json.loads(extract_json_object(raw_text))
    document = AutomationDocument.model_validate(data)
    if not document.rules:
        raise ValueError("LLM response must include a non-empty rules array.")
    return LLMCompileResult(raw_text=raw_text, document=document)


def extract_json_object(text: str) -> str:
    stripped = text.strip()
    if stripped.startswith("```"):
        stripped = re.sub(r"^```(?:json)?", "", stripped, flags=re.IGNORECASE).strip()
        stripped = re.sub(r"```$", "", stripped).strip()
    if stripped.startswith("{") and stripped.endswith("}"):
        return stripped

    match = re.search(r"\{.*\}", stripped, flags=re.DOTALL)
    if not match:
        raise ValueError("llama.cpp response did not contain a JSON object.")
    return match.group(0)


def _build_user_prompt(*, instruction: str, class_names: list[str]) -> str:
    class_hint = ", ".join(class_names) if class_names else "Use labels from the request."
    return f"""Available detection labels: {class_hint}
Automation request:
{instruction}
Return a JSON object matching this high-level shape:
{{
  "rules": [
    {{
      "name": "short-kebab-case-name",
      "when": {{"all": [{{"present": {{"label": "cat", "min_count": 1}}}}]}},
      "trigger": {{"on": "while"}},
      "gate": {{"enabled": true}},
      "then": [{{"type": "simulate", "name": "action name"}}]
    }}
  ]
}}
Examples:
User: If person near steering wheel then you have to turn on pc.
JSON:
{{
  "rules": [
    {{
      "name": "person-near-steering-wheel",
      "when": {{
        "all": [
          {{"present": {{"label": "person", "min_count": 1}}}},
          {{"near": {{"a": "person", "b": "steering wheel", "max_gap_percent": 8}}}}
        ]
      }},
      "trigger": {{"on": "enter"}},
      "gate": {{"enabled": true}},
      "then": [{{"type": "simulate", "name": "turn on pc"}}]
    }}
  ]
}}
User: If package is at door and 15 minutes since last notification, notify me.
JSON:
{{
  "rules": [
    {{
      "name": "package-at-door",
      "when": {{
        "all": [
          {{"present": {{"label": "package", "min_count": 1}}}},
          {{"near": {{"a": "package", "b": "door", "max_gap_percent": 8}}}}
        ]
      }},
      "trigger": {{"on": "while"}},
      "gate": {{"enabled": true, "cooldown": {{"key": "package-at-door", "minutes": 15}}}},
      "then": [{{"type": "simulate", "name": "notify me"}}]
    }}
  ]
}}
User: While there is a guitar in the scene, amplifier must be on.
JSON:
{{
  "rules": [
    {{
      "name": "guitar-amplifier-on",
      "when": {{
        "all": [
          {{"present": {{"label": "guitar", "min_count": 1}}}}
        ]
      }},
      "trigger": {{"on": "enter"}},
      "gate": {{"enabled": true}},
      "then": [{{"type": "simulate", "name": "turn on amplifier"}}]
    }}
  ]
}}
User: If person is near monitor turn on lights. When they leave, turn off lights.
JSON:
{{
  "rules": [
    {{
      "name": "monitor-presence-lights",
      "when": {{
        "all": [
          {{"near": {{"a": "person", "b": "monitor", "max_gap_percent": 8}}}}
        ]
      }},
      "trigger": {{"on": "change"}},
      "gate": {{"enabled": true}},
      "then": {{
        "enter": [{{"type": "simulate", "name": "turn on lights"}}],
        "exit": [{{"type": "simulate", "name": "turn off lights"}}]
      }}
    }}
  ]
}}
Full validation schema:
{json.dumps(automation_schema(), indent=2)}
"""


def _build_repair_prompt(*, original_prompt: str, bad_response: str, error: str) -> str:
    return f"""{original_prompt}
Your previous response failed validation.
Validation error:
{error}
Previous response:
{bad_response}
Return corrected JSON only. The root object MUST include a non-empty "rules" array.
"""


def _chat_payload(*, model: str, user_prompt: str, response_format: str) -> dict[str, Any]:
    payload: dict[str, Any] = {
        "model": model,
        "messages": [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_prompt},
        ],
        "max_tokens": 512,
        "temperature": 0,
        "stream": False,
    }
    if response_format == "json_schema":
        payload["response_format"] = {
            "type": "json_schema",
            "json_schema": {
                "name": "tiny_trigger_automation",
                "strict": True,
                "schema": automation_schema(),
            },
        }
    else:
        payload["response_format"] = {"type": "json_object"}
    return payload

	from __future__ import annotations

	import json
	import re
	from typing import Any

	from pydantic import BaseModel, ValidationError

	from .automation import AutomationDocument, automation_schema


	SYSTEM_PROMPT = """You compile home automation requests into Tiny Trigger rules.
	Return JSON only. Never return code, markdown, explanations, or tool calls.
	The root object MUST include a non-empty "rules" array.
	Use video conditions in when: present, count, near, far.
	Use state gates in gate: enabled, cooldown.
	Use trigger.on for edge behavior: while, enter, exit, change.
	Use only these action types: simulate, webhook.
	Use trigger.on="enter" for state assertions like "must be on", "should be on", "keep on", "turn on when", or "notify when".
	Use trigger.on="while" only when the user explicitly wants repeated actions while a condition remains true, usually with a cooldown.
	When the request says one object is near, next to, beside, at, by, close to, or in front of another object, you MUST emit a near condition.
	Do not replace a near relation with two present conditions.
	Use max_gap_percent for near/far box-edge distance. It is the largest horizontal/vertical edge gap between boxes in normalized frame percent; touching or overlapping boxes have gap 0.
	If the user mentions elapsed time since an action or limiting repeat fires, encode it as gate.cooldown.
	If the user asks for an action when a condition starts and another action when it stops, use trigger.on="change" and then.enter / then.exit.
	"""


	class LLMCompileResult(BaseModel):
	raw_text: str
	document: AutomationDocument


	def compile_automation_with_llamacpp(
	*,
	instruction: str,
	class_names: list[str],
	base_url: str = "http://127.0.0.1:8080/v1",
	model: str = "ggml-org/Qwen3-1.7B-GGUF:Q4_K_M",
	timeout: float = 60.0,
	) -> LLMCompileResult:
	"""Compile natural language into validated rules through llama.cpp server."""
	try:
	import requests
	except ImportError as exc: # pragma: no cover - dependency guard
	raise RuntimeError("Install requests to use the llama.cpp compiler.") from exc

	endpoint = base_url.rstrip("/") + "/chat/completions"
	user_prompt = _build_user_prompt(instruction=instruction, class_names=class_names)
	raw_text = _post_chat_completion(
	endpoint=endpoint,
	model=model,
	user_prompt=user_prompt,
	timeout=timeout,
	requests_module=requests,
	)
	try:
	return _validate_compile_result(raw_text)
	except (json.JSONDecodeError, ValidationError, ValueError) as first_error:
	repair_prompt = _build_repair_prompt(
	original_prompt=user_prompt,
	bad_response=raw_text,
	error=str(first_error),
	)
	repaired_text = _post_chat_completion(
	endpoint=endpoint,
	model=model,
	user_prompt=repair_prompt,
	timeout=timeout,
	requests_module=requests,
	)
	try:
	return _validate_compile_result(repaired_text)
	except (json.JSONDecodeError, ValidationError, ValueError) as second_error:
	raise ValueError(
	"llama.cpp returned invalid automation JSON after a repair attempt. "
	f"Last validation error: {second_error}. Last response: {repaired_text}"
	) from second_error


	def compile_automation_with_replicate(
	*,
	instruction: str,
	class_names: list[str],
	api_token: str,
	model: str = "openai/gpt-5.2",
	reasoning_effort: str = "medium",
	timeout: float = 600.0,
	) -> LLMCompileResult:
	"""Compile natural language into validated rules through Replicate."""
	try:
	import replicate
	except ImportError as exc: # pragma: no cover - dependency guard
	raise RuntimeError("Install replicate to use the Replicate compiler.") from exc

	user_prompt = _build_user_prompt(instruction=instruction, class_names=class_names)
	raw_text = _stream_replicate_completion(
	model=model,
	prompt=_provider_prompt(user_prompt),
	api_token=api_token,
	reasoning_effort=reasoning_effort,
	timeout=timeout,
	replicate_module=replicate,
	)
	try:
	return _validate_compile_result(raw_text)
	except (json.JSONDecodeError, ValidationError, ValueError) as first_error:
	repair_prompt = _build_repair_prompt(
	original_prompt=user_prompt,
	bad_response=raw_text,
	error=str(first_error),
	)
	repaired_text = _stream_replicate_completion(
	model=model,
	prompt=_provider_prompt(repair_prompt),
	api_token=api_token,
	reasoning_effort=reasoning_effort,
	timeout=timeout,
	replicate_module=replicate,
	)
	try:
	return _validate_compile_result(repaired_text)
	except (json.JSONDecodeError, ValidationError, ValueError) as second_error:
	raise ValueError(
	"Replicate returned invalid automation JSON after a repair attempt. "
	f"Last validation error: {second_error}. Last response: {repaired_text}"
	) from second_error


	def _provider_prompt(user_prompt: str) -> str:
	return f"{SYSTEM_PROMPT}\n\n{user_prompt}\n\nReturn only the JSON object."


	def _stream_replicate_completion(
	*,
	model: str,
	prompt: str,
	api_token: str,
	reasoning_effort: str,
	timeout: float,
	replicate_module: Any,
	) -> str:
	_split_replicate_model(model)
	payload = {
	"prompt": prompt,
	"messages": [],
	"verbosity": "medium",
	"reasoning_effort": reasoning_effort,
	}
	client = replicate_module.Client(api_token=api_token)
	chunks: list[str] = []
	for event in client.stream(model, input=payload):
	chunks.append(str(event))
	text = "".join(chunks).strip()
	if not text:
	raise ValueError("Replicate stream returned no output.")
	return text


	def _split_replicate_model(model: str) -> tuple[str, str]:
	parts = model.strip().split("/", 1)
	if len(parts) != 2 or not all(parts):
	raise ValueError("Replicate model must be in owner/model format, for example openai/gpt-5.2.")
	return parts[0], parts[1]


	def _post_chat_completion(
	*,
	endpoint: str,
	model: str,
	user_prompt: str,
	timeout: float,
	requests_module: Any,
	) -> str:
	payload = _chat_payload(model=model, user_prompt=user_prompt, response_format="json_object")
	response = requests_module.post(endpoint, json=payload, timeout=timeout)
	response.raise_for_status()
	body = response.json()
	return body["choices"][0]["message"]["content"]


	def _validate_compile_result(raw_text: str) -> LLMCompileResult:
	data = json.loads(extract_json_object(raw_text))
	document = AutomationDocument.model_validate(data)
	if not document.rules:
	raise ValueError("LLM response must include a non-empty rules array.")
	return LLMCompileResult(raw_text=raw_text, document=document)


	def extract_json_object(text: str) -> str:
	stripped = text.strip()
	if stripped.startswith("```"):
	stripped = re.sub(r"^```(?:json)?", "", stripped, flags=re.IGNORECASE).strip()
	stripped = re.sub(r"```$", "", stripped).strip()
	if stripped.startswith("{") and stripped.endswith("}"):
	return stripped

	match = re.search(r"\{.*\}", stripped, flags=re.DOTALL)
	if not match:
	raise ValueError("llama.cpp response did not contain a JSON object.")
	return match.group(0)


	def _build_user_prompt(*, instruction: str, class_names: list[str]) -> str:
	class_hint = ", ".join(class_names) if class_names else "Use labels from the request."
	return f"""Available detection labels: {class_hint}

	Automation request:
	{instruction}

	Return a JSON object matching this high-level shape:
	{{
	"rules": [
	{{
	"name": "short-kebab-case-name",
	"when": {{"all": [{{"present": {{"label": "cat", "min_count": 1}}}}]}},
	"trigger": {{"on": "while"}},
	"gate": {{"enabled": true}},
	"then": [{{"type": "simulate", "name": "action name"}}]
	}}
	]
	}}

	Examples:

	User: If person near steering wheel then you have to turn on pc.
	JSON:
	{{
	"rules": [
	{{
	"name": "person-near-steering-wheel",
	"when": {{
	"all": [
	{{"present": {{"label": "person", "min_count": 1}}}},
	{{"near": {{"a": "person", "b": "steering wheel", "max_gap_percent": 8}}}}
	]
	}},
	"trigger": {{"on": "enter"}},
	"gate": {{"enabled": true}},
	"then": [{{"type": "simulate", "name": "turn on pc"}}]
	}}
	]
	}}

	User: If package is at door and 15 minutes since last notification, notify me.
	JSON:
	{{
	"rules": [
	{{
	"name": "package-at-door",
	"when": {{
	"all": [
	{{"present": {{"label": "package", "min_count": 1}}}},
	{{"near": {{"a": "package", "b": "door", "max_gap_percent": 8}}}}
	]
	}},
	"trigger": {{"on": "while"}},
	"gate": {{"enabled": true, "cooldown": {{"key": "package-at-door", "minutes": 15}}}},
	"then": [{{"type": "simulate", "name": "notify me"}}]
	}}
	]
	}}

	User: While there is a guitar in the scene, amplifier must be on.
	JSON:
	{{
	"rules": [
	{{
	"name": "guitar-amplifier-on",
	"when": {{
	"all": [
	{{"present": {{"label": "guitar", "min_count": 1}}}}
	]
	}},
	"trigger": {{"on": "enter"}},
	"gate": {{"enabled": true}},
	"then": [{{"type": "simulate", "name": "turn on amplifier"}}]
	}}
	]
	}}

	User: If person is near monitor turn on lights. When they leave, turn off lights.
	JSON:
	{{
	"rules": [
	{{
	"name": "monitor-presence-lights",
	"when": {{
	"all": [
	{{"near": {{"a": "person", "b": "monitor", "max_gap_percent": 8}}}}
	]
	}},
	"trigger": {{"on": "change"}},
	"gate": {{"enabled": true}},
	"then": {{
	"enter": [{{"type": "simulate", "name": "turn on lights"}}],
	"exit": [{{"type": "simulate", "name": "turn off lights"}}]
	}}
	}}
	]
	}}

	Full validation schema:
	{json.dumps(automation_schema(), indent=2)}
	"""


	def _build_repair_prompt(*, original_prompt: str, bad_response: str, error: str) -> str:
	return f"""{original_prompt}

	Your previous response failed validation.

	Validation error:
	{error}

	Previous response:
	{bad_response}

	Return corrected JSON only. The root object MUST include a non-empty "rules" array.
	"""


	def _chat_payload(*, model: str, user_prompt: str, response_format: str) -> dict[str, Any]:
	payload: dict[str, Any] = {
	"model": model,
	"messages": [
	{"role": "system", "content": SYSTEM_PROMPT},
	{"role": "user", "content": user_prompt},
	],
	"max_tokens": 512,
	"temperature": 0,
	"stream": False,
	}
	if response_format == "json_schema":
	payload["response_format"] = {
	"type": "json_schema",
	"json_schema": {
	"name": "tiny_trigger_automation",
	"strict": True,
	"schema": automation_schema(),
	},
	}
	else:
	payload["response_format"] = {"type": "json_object"}
	return payload