LLM - hadad/LFM2.5-1.2B:Q4_K_M

lmf2-tools.py

"""
Test script for LFM2.5-1.2B tool calling via Ollama.
Uses LFM2.5's native ChatML + Pythonic tool call format directly,
since the community Ollama upload lacks a tools template.

Make sure Ollama is running and the model is pulled:
  ollama pull hadad/LFM2.5-1.2B:Q4_K_M
"""

import ast
import json
import re
import requests

OLLAMA_URL = "http://localhost:11434/api/generate"
MODEL = "hadad/LFM2.5-1.2B:Q4_K_M"

# ---------------------------------------------------------------------------
# Fake tool implementations
# ---------------------------------------------------------------------------

def get_weather(location: str, unit: str = "celsius") -> dict:
    fake_data = {
        "paris":    {"temp": 18, "condition": "Partly cloudy", "humidity": 65},
        "london":   {"temp": 14, "condition": "Rainy",         "humidity": 80},
        "tokyo":    {"temp": 25, "condition": "Sunny",         "humidity": 55},
        "new york": {"temp": 22, "condition": "Clear",         "humidity": 50},
    }
    data = dict(fake_data.get(location.lower(), {"temp": 20, "condition": "Unknown", "humidity": 60}))
    if unit == "fahrenheit":
        data["temp"] = round(data["temp"] * 9 / 5 + 32)
    data["location"] = location
    data["unit"] = unit
    return data


def calculate(expression: str) -> dict:
    try:
        # Normalise natural language: "15% of 280" → "0.15 * 280"
        expr = re.sub(
            r'(\d+(?:\.\d+)?)\s*%\s*of\s*(\d+(?:\.\d+)?)',
            lambda m: f"({m.group(1)} / 100) * {m.group(2)}",
            expression,
            flags=re.IGNORECASE,
        )
        # Replace bare "X%" → "(X/100)"
        expr = re.sub(r'(\d+(?:\.\d+)?)\s*%', r'(\1/100)', expr)

        allowed = set("0123456789+-*/()., ")
        if not all(c in allowed for c in expr):
            return {"error": f"Invalid characters in expression: {expr!r}"}
        result = eval(expr, {"__builtins__": {}})  # noqa: S307
        return {"expression": expression, "result": round(result, 10)}
    except Exception as e:
        return {"error": str(e)}


def search_contacts(name: str) -> dict:
    contacts = [
        {"name": "Alice Martin",   "email": "alice@example.com",   "phone": "+33 6 12 34 56 78"},
        {"name": "Bob Dupont",     "email": "bob@example.com",     "phone": "+33 6 98 76 54 32"},
        {"name": "Charlie Durand", "email": "charlie@example.com", "phone": "+33 6 55 44 33 22"},
    ]
    matches = [c for c in contacts if name.lower() in c["name"].lower()]
    return {"query": name, "results": matches, "count": len(matches)}


TOOL_MAP = {
    "get_weather": get_weather,
    "calculate": calculate,
    "search_contacts": search_contacts,
}

# ---------------------------------------------------------------------------
# Tool definitions as JSON (injected into system prompt per LFM2.5 spec)
# ---------------------------------------------------------------------------

TOOLS_JSON = json.dumps([
    {
        "name": "get_weather",
        "description": "Get the current weather for a given city.",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {"type": "string", "description": "City name, e.g. 'Paris'"},
                "unit":     {"type": "string", "enum": ["celsius", "fahrenheit"], "description": "Temperature unit"},
            },
            "required": ["location"],
        },
    },
    {
        "name": "calculate",
        "description": (
            "Evaluate a math expression. "
            "Use standard arithmetic operators: +, -, *, /. "
            "For percentages write '15% of 280' or '0.15 * 280'. "
            "Do NOT pass natural language like 'fifteen percent of 280'."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "Math expression, e.g. '(3+5)*2' or '15% of 280' or '0.15*280'",
                },
            },
            "required": ["expression"],
        },
    },
    {
        "name": "search_contacts",
        "description": "Search for a person in the contact book by name.",
        "parameters": {
            "type": "object",
            "properties": {
                "name": {"type": "string", "description": "Full or partial name"},
            },
            "required": ["name"],
        },
    },
])

# ---------------------------------------------------------------------------
# ChatML prompt builder
# ---------------------------------------------------------------------------

SYSTEM_PROMPT = f"You are a helpful assistant. List of tools: {TOOLS_JSON}"


def build_prompt(turns: list[dict]) -> str:
    """
    Build a ChatML prompt from a list of {role, content} turns.
    Roles: system, user, assistant, tool
    """
    out = "<|startoftext|>"
    for t in turns:
        role = t["role"]
        content = t["content"]
        out += f"<|im_start|>{role}\n{content}<|im_end|>\n"
    out += "<|im_start|>assistant\n"
    return out


# ---------------------------------------------------------------------------
# Parse Pythonic tool calls from model output
# e.g.  <|tool_call_start|>[get_weather(location="Paris")]<|tool_call_end|>
# ---------------------------------------------------------------------------

TOOL_CALL_RE = re.compile(
    r"<\|tool_call_start\|>(.*?)<\|tool_call_end\|>", re.DOTALL
)
# Match function name, then grab everything up to the MATCHING closing paren
# using a depth counter (done in parse_tool_calls, not regex).
FUNC_NAME_RE = re.compile(r'(\w+)\s*\(')


def _extract_call_body(text: str, start: int) -> str:
    """
    Given text and the index of the opening '(' of a function call,
    return the raw content inside the outermost parens (handles nesting).
    """
    depth = 0
    for i in range(start, len(text)):
        if text[i] == '(':
            depth += 1
        elif text[i] == ')':
            depth -= 1
            if depth == 0:
                return text[start + 1:i]
    return text[start + 1:]  # unclosed — return remainder


def _parse_value(node: ast.expr):
    """Recursively convert an AST node to a Python value."""
    if isinstance(node, ast.Constant):
        return node.value
    if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.USub):
        return -_parse_value(node.operand)
    if isinstance(node, ast.List):
        return [_parse_value(e) for e in node.elts]
    if isinstance(node, ast.Dict):
        return {_parse_value(k): _parse_value(v) for k, v in zip(node.keys, node.values)}
    # Fallback: return the source text
    return ast.unparse(node)


def parse_tool_calls(text: str) -> list[dict]:
    """
    Extract tool calls from LFM2.5's Pythonic format.
    Handles nested parentheses, quoted strings with special chars, etc.
    """
    calls = []
    for block in TOOL_CALL_RE.findall(text):
        pos = 0
        while pos < len(block):
            m = FUNC_NAME_RE.search(block, pos)
            if not m:
                break
            name = m.group(1)
            # Skip built-ins / list wrapper that wraps all calls
            if name in ("print", "list", "dict", "str", "int", "float"):
                pos = m.end()
                continue
            paren_start = m.end() - 1  # index of '('
            body = _extract_call_body(block, paren_start)
            pos = paren_start + len(body) + 2  # skip past closing ')'

            # Parse keyword arguments with ast
            args = {}
            try:
                # Wrap in a dummy call so ast can parse it
                tree = ast.parse(f"_f({body})", mode="eval")
                call_node = tree.body
                for kw in call_node.keywords:
                    args[kw.arg] = _parse_value(kw.value)
                # Positional args (unusual but possible)
                for i, arg in enumerate(call_node.args):
                    args[f"_arg{i}"] = _parse_value(arg)
            except SyntaxError:
                # Last resort: treat whole body as the first positional arg
                args = {"_arg0": body.strip().strip('"\'') }

            if name in TOOL_MAP:
                calls.append({"name": name, "args": args})

    return calls


# ---------------------------------------------------------------------------
# Ollama generate call
# ---------------------------------------------------------------------------

def generate(prompt: str) -> str:
    payload = {
        "model": MODEL,
        "prompt": prompt,
        "stream": False,
        "options": {
            "temperature": 0.1,
            "top_k": 50,
            "top_p": 0.1,
            "repeat_penalty": 1.05,
            "stop": ["<|im_end|>", "<|im_start|>"],
        },
    }
    resp = requests.post(OLLAMA_URL, json=payload, timeout=120)
    resp.raise_for_status()
    return resp.json()["response"]


# ---------------------------------------------------------------------------
# Agentic loop
# ---------------------------------------------------------------------------

def run(user_prompt: str) -> str:
    print(f"\n{'='*60}")
    print(f"USER: {user_prompt}")

    turns = [
        {"role": "system",  "content": SYSTEM_PROMPT},
        {"role": "user",    "content": user_prompt},
    ]

    for _ in range(5):  # max iterations
        prompt = build_prompt(turns)
        raw = generate(prompt).strip()

        tool_calls = parse_tool_calls(raw)

        if not tool_calls:
            # Final answer — strip any leftover special tokens
            answer = re.sub(r"<\|.*?\|>", "", raw).strip()
            print(f"ASSISTANT: {answer}")
            return answer

        # Add assistant turn (with tool call markup)
        turns.append({"role": "assistant", "content": raw})

        # Execute tools and feed results back
        for tc in tool_calls:
            name, args = tc["name"], tc["args"]
            print(f"  → TOOL CALL: {name}({args})")

            fn = TOOL_MAP.get(name)
            if fn:
                result = fn(**args)
            else:
                result = {"error": f"Unknown tool: {name}"}

            result_str = json.dumps(result)
            print(f"  ← RESULT:    {result_str}")

            turns.append({"role": "tool", "content": result_str})

    return "(max iterations reached)"


# ---------------------------------------------------------------------------
# Test prompts
# ---------------------------------------------------------------------------

TEST_PROMPTS = [
    "What's the weather like in Paris right now?",
    "What is (123 * 456) + 789?",
    "Find contact info for Alice.",
    "What's the weather in Tokyo in Fahrenheit, and also calculate 15% of 280.",
]


def main():
    print(f"Model: {MODEL}")
    print("Ensure Ollama is running:  ollama serve")

    try:
        requests.get("http://localhost:11434", timeout=5)
    except requests.ConnectionError:
        print("\nERROR: Cannot reach Ollama at localhost:11434.")
        print("Start it with:  ollama serve")
        return

    passed = 0
    for prompt in TEST_PROMPTS:
        try:
            answer = run(prompt)
            if answer:
                passed += 1
        except Exception as e:
            print(f"  ERROR: {e}")

    print(f"\n{'='*60}")
    print(f"Done: {passed}/{len(TEST_PROMPTS)} prompts completed.")


if __name__ == "__main__":
    main()