Tool calling from LFM2 and LFM2.5?

[alfiedotwtf]

Your model looks impressive and I wanted to fall in love with it...

Your docs say that LFM2 does native tool calling, but the amount of debugging I've had to churn through to get ollama up and running with your models was 😮‍💨

Is there any smoother way of getting LFM2 and LFM2.5 working with Ollama + tooling?

The main problem is that even though your docs say tooling is pythonic, there's a mix of native, json and even xml! So to use opencode with lfm2, I've had to write a proxy to intercept messages, strip out the inconsistent tooling conventions, and then reluctantly run unix commands bare when they should be going through the bash tool (because it's ignoring my prompt).

Maybe it's me and I've been doing things wrong, but I cant find anyone else who's managed to get your models working agentically. If this was a lot smoother, lfm2* could be the talk of the town!

[Paulescu]

Hi @alfiedotwtf ,

Thanks a lot for your feedback. It really helps us understand developer pain points and make your dev experience easier and more pleasant.

In general, our models generate Pythonic function calls, which do not play well with Ollama I suspect (we are working on fixing this in our next generation of models).

In the meantime, I would first suggest you modify your system prompt to include the message "Output function calls as JSON"

See the full documentation on tool calling here.

If the issue persists, please share your system prompt/Modelfile you are using, so I can help further

Wish you a great day

Pau

[Paulescu]

Hi @alfiedotwtf

Here's some work in progress that can help you
https://github.com/Liquid4All/cookbook/tree/pau/examples/home-assistant/examples/home-assistant

[alfiedotwtf]

Sorry for the late response. I'm standing in line at the Email Bankruptcy Department today...

Here's the proxy that I've iteratively prompted, which uses the /chat/completitions/. It's a hack job just to get up and running so that I could try out your model. So yeah, with this, the responses I'm getting back and JSONic Python and sometimes wrapped in XML 🫠

Your docs say it's agentic and works with tools. Is this with one of the usual backends, or are you all running internally with a hand-rolled wrapper?

One other note - testing via Ollama, output that I'm getting (although not agentic and being written to my filesystem so only printed on screen) is pretty impressive. Hats off to the team.

#!/usr/bin/env python3
"""
LFM2 Tool Call Proxy
====================
Sits between your client and Ollama (or llama-server).
Handles two things:
  1. Injects tool definitions into the system prompt using LFM2's native tokens
  2. Parses the model's Pythonic tool call output back into OpenAI-format tool_calls

Supports both streaming and non-streaming responses. It's still using `/v1/chat/responses/` rather than 

Usage:
    pip install fastapi uvicorn httpx
    python lfm2_proxy.py

Then point your client at http://localhost:8090/v1
instead of Ollama's http://localhost:11434/v1

Config (env vars or edit the constants below):
    UPSTREAM_URL   - where Ollama/llama-server is running (default: http://localhost:11434)
    PROXY_PORT     - port this proxy listens on (default: 8090)
"""

import re
import json
import ast
import uuid
import time
import os
import httpx
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse, StreamingResponse
import uvicorn

UPSTREAM_URL = os.environ.get("UPSTREAM_URL", "http://localhost:11434")
PROXY_PORT   = int(os.environ.get("PROXY_PORT", 8090))

app = FastAPI()


# ---------------------------------------------------------------------------
# Prompt building
# ---------------------------------------------------------------------------

def inject_tools_into_messages(messages: list, tools: list) -> list:
    """
    Injects tool definitions into the system message using LFM2's native tokens:
      <|tool_list_start|>
      { tool json }
      <|tool_list_end|>
    """
    tool_block = "<|tool_list_start|>\n"
    for tool in tools:
        tool_block += json.dumps(tool, indent=2) + "\n"
    tool_block += "<|tool_list_end|>"

    messages = list(messages)  # shallow copy

    if messages and messages[0]["role"] == "system":
        messages[0] = {
            **messages[0],
            "content": messages[0]["content"] + "\n" + tool_block
        }
    else:
        messages.insert(0, {"role": "system", "content": "You are a helpful assistant.\n" + tool_block})

    return messages


# ---------------------------------------------------------------------------
# Output parsing
# ---------------------------------------------------------------------------

def parse_tool_calls(content: str):
    """
    Attempts to extract tool calls from the model's output.

    Handles these formats the model may produce:
      - <|tool_call_start|>[func(arg="val")]<|tool_call_end|>
      - [func(arg="val")]
      - [func(arg="val")]some trailing text

    Returns (tool_calls, remaining_content) where:
      - tool_calls is a list of OpenAI-format tool call dicts, or None if no call found
      - remaining_content is any text outside the tool call block
    """
    # Try to find content between special tokens first
    token_match = re.search(
        r'<\|tool_call_start\|>(.*?)<\|tool_call_end\|>',
        content,
        re.DOTALL
    )
    if token_match:
        call_str = token_match.group(1).strip()
        remaining = content[:token_match.start()].strip()
        trailing = content[token_match.end():].strip()
        if trailing:
            remaining = (remaining + "\n" + trailing).strip()
    else:
        # Fall back: look for [...] at start of content that looks like a call
        bracket_match = re.match(r'^\s*(\[.*?\])(.*)', content, re.DOTALL)
        if bracket_match:
            call_str = bracket_match.group(1).strip()
            remaining = bracket_match.group(2).strip()
        else:
            return None, content

    tool_calls = parse_pythonic_calls(call_str)
    if tool_calls is None:
        # Couldn't parse - return original content untouched
        return None, content

    return tool_calls, remaining or None


def parse_pythonic_calls(call_str: str):
    """
    Parses a Pythonic tool call string like:
      [write_file(path="/tmp/test.php", content="<?php echo 'Hello'; ?>")]

    Returns a list of OpenAI-format tool call dicts, or None on failure.
    """
    inner = call_str.strip()
    if inner.startswith('[') and inner.endswith(']'):
        inner = inner[1:-1].strip()

    tool_calls = []

    for call in split_calls(inner):
        call = call.strip()
        if not call:
            continue

        fn_match = re.match(r'^([a-zA-Z_][a-zA-Z0-9_]*)\s*\((.*)\)$', call, re.DOTALL)
        if not fn_match:
            return None

        fn_name = fn_match.group(1)
        args_str = fn_match.group(2).strip()

        arguments = parse_kwargs(args_str)
        if arguments is None:
            return None

        tool_calls.append({
            "id": f"call_{uuid.uuid4().hex[:8]}",
            "type": "function",
            "function": {
                "name": fn_name,
                "arguments": json.dumps(arguments)
            }
        })

    return tool_calls if tool_calls else None


def split_calls(s: str):
    """
    Splits a string of comma-separated function calls at the top level.
    e.g. "foo(a=1), bar(b=2)" -> ["foo(a=1)", "bar(b=2)"]
    """
    calls = []
    depth = 0
    current = []
    in_string = False
    string_char = None

    for ch in s:
        if in_string:
            current.append(ch)
            if ch == string_char:
                in_string = False
        elif ch in ('"', "'"):
            in_string = True
            string_char = ch
            current.append(ch)
        elif ch == '(':
            depth += 1
            current.append(ch)
        elif ch == ')':
            depth -= 1
            current.append(ch)
        elif ch == ',' and depth == 0:
            calls.append(''.join(current).strip())
            current = []
        else:
            current.append(ch)

    if current:
        calls.append(''.join(current).strip())

    return calls


def parse_kwargs(args_str: str):
    """
    Parses keyword arguments string into a dict.
    """
    if not args_str:
        return {}

    arguments = {}
    pairs = split_kwargs(args_str)

    for pair in pairs:
        pair = pair.strip()
        if not pair:
            continue
        eq_idx = pair.index('=')
        key = pair[:eq_idx].strip()
        val_str = pair[eq_idx+1:].strip()

        try:
            value = ast.literal_eval(val_str)
        except Exception:
            value = val_str.strip('"\'')

        arguments[key] = value

    return arguments


def split_kwargs(s: str):
    """
    Splits key=value pairs at top-level commas (not inside strings/parens).
    """
    parts = []
    depth = 0
    in_string = False
    string_char = None
    current = []

    for ch in s:
        if in_string:
            current.append(ch)
            if ch == string_char:
                in_string = False
        elif ch in ('"', "'"):
            in_string = True
            string_char = ch
            current.append(ch)
        elif ch in ('(', '[', '{'):
            depth += 1
            current.append(ch)
        elif ch in (')', ']', '}'):
            depth -= 1
            current.append(ch)
        elif ch == ',' and depth == 0:
            parts.append(''.join(current))
            current = []
        else:
            current.append(ch)

    if current:
        parts.append(''.join(current))

    return parts


def process_tool_calls_in_result(result: dict) -> dict:
    """Mutates result in-place to parse tool calls from content."""
    for choice in result.get("choices", []):
        msg = choice.get("message", {})
        content = msg.get("content", "")
        if content:
            tool_calls, remaining = parse_tool_calls(content)
            if tool_calls:
                msg["tool_calls"] = tool_calls
                msg["content"] = remaining
                choice["finish_reason"] = "tool_calls"
    return result


# ---------------------------------------------------------------------------
# Proxy routes
# ---------------------------------------------------------------------------

@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    body = await request.json()

    tools = body.get("tools")
    is_streaming = body.get("stream", False)
    original_messages = body.get("messages", [])

    # If tools provided, inject them into the system prompt
    if tools:
        body = {**body, "messages": inject_tools_into_messages(original_messages, tools)}
        body.pop("tools", None)
        body.pop("tool_choice", None)

    # --- STREAMING ---
    if is_streaming:
        async def stream_generator():
            # Buffer all SSE chunks so we can post-process for tool calls
            chunks = []
            async with httpx.AsyncClient(timeout=300) as client:
                async with client.stream(
                    "POST",
                    f"{UPSTREAM_URL}/v1/chat/completions",
                    json=body,
                    headers={"Content-Type": "application/json"}
                ) as upstream_resp:
                    async for line in upstream_resp.aiter_lines():
                        if line:
                            chunks.append(line)

            if not tools:
                # No tool processing needed - pass chunks straight through
                for chunk in chunks:
                    yield chunk + "\n\n"
                return

            # Reassemble full content from delta chunks
            full_content = ""
            last_chunk_data = None

            for line in chunks:
                if not line.startswith("data: "):
                    continue
                data_str = line[6:]
                if data_str.strip() == "[DONE]":
                    continue
                try:
                    chunk_data = json.loads(data_str)
                    delta = chunk_data.get("choices", [{}])[0].get("delta", {})
                    content_piece = delta.get("content", "")
                    if content_piece:
                        full_content += content_piece
                    last_chunk_data = chunk_data
                except Exception:
                    pass

            tool_calls, remaining = parse_tool_calls(full_content)

            if tool_calls and last_chunk_data:
                # Synthesise a proper tool_calls chunk
                response_id = last_chunk_data.get("id", f"chatcmpl-{uuid.uuid4().hex[:8]}")
                model = last_chunk_data.get("model", "unknown")
                synthesised = {
                    "id": response_id,
                    "object": "chat.completion.chunk",
                    "created": int(time.time()),
                    "model": model,
                    "choices": [{
                        "index": 0,
                        "delta": {
                            "role": "assistant",
                            "content": remaining,
                            "tool_calls": [
                                {
                                    "index": i,
                                    "id": tc["id"],
                                    "type": "function",
                                    "function": {
                                        "name": tc["function"]["name"],
                                        "arguments": tc["function"]["arguments"]
                                    }
                                }
                                for i, tc in enumerate(tool_calls)
                            ]
                        },
                        "finish_reason": "tool_calls"
                    }]
                }
                yield f"data: {json.dumps(synthesised)}\n\n"
                yield "data: [DONE]\n\n"
            else:
                # No tool call found - pass original chunks through unchanged
                for chunk in chunks:
                    yield chunk + "\n\n"

        return StreamingResponse(stream_generator(), media_type="text/event-stream")

    # --- NON-STREAMING ---
    async with httpx.AsyncClient(timeout=300) as client:
        upstream_resp = await client.post(
            f"{UPSTREAM_URL}/v1/chat/completions",
            json=body,
            headers={"Content-Type": "application/json"}
        )

    if upstream_resp.status_code != 200:
        return Response(
            content=upstream_resp.content,
            status_code=upstream_resp.status_code,
            media_type="application/json"
        )

    result = upstream_resp.json()

    if tools:
        process_tool_calls_in_result(result)

    return JSONResponse(content=result)


# Pass through any other endpoints (models list, health, etc.)
@app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
async def passthrough(request: Request, path: str):
    async with httpx.AsyncClient(timeout=60) as client:
        upstream_resp = await client.request(
            method=request.method,
            url=f"{UPSTREAM_URL}/{path}",
            headers=dict(request.headers),
            content=await request.body()
        )
    return Response(
        content=upstream_resp.content,
        status_code=upstream_resp.status_code,
        media_type=upstream_resp.headers.get("content-type", "application/json")
    )