teff.harness.formats

teff.harness.formats

Response parsing and message-format normalisation for LLM providers.

Functions:

Name	Description
extract_content	Extract the assistant text from a response.
extract_message	Normalise response formats to {role, content, tool_calls}.
extract_usage	Extract (prompt_tokens, completion_tokens) from an LLM response.
normalize_text_tool_calls	Turn a text-embedded tool call into the structured tool_calls list.
parse_text_tool_call	Parse a tool call embedded in plain text content.

extract_content

extract_content(data, provider_type, path='', fallback='')

Extract the assistant text from a response.

path is a dot-separated path into data; otherwise the extraction follows the wire protocol provider_type (Anthropic content blocks, Ollama root message).

Source code in teff/harness/formats.py

def extract_content(
    data: dict, provider_type: str, path: str = "", fallback: str = ""
) -> str:
    """Extract the assistant text from a response.

    *path* is a dot-separated path into *data*; otherwise the extraction
    follows the wire protocol *provider_type* (Anthropic content blocks,
    Ollama root ``message``).
    """
    if path:
        parts = path.split(".")
        val: typing.Any = data
        try:
            for p in parts:
                if p.isdigit():
                    val = val[int(p)]
                else:
                    val = val.get(p, "")
        except (AttributeError, IndexError, KeyError, TypeError, ValueError):
            return ""
        return str(val) if val else ""

    if provider_type == "anthropic_compatible":
        for block in data.get("content", []):
            if block.get("type") == "text":
                return block.get("text", "")
        return ""

    if provider_type == "ollama":
        return data.get("message", {}).get("content", "")

    return fallback

extract_message

extract_message(data)

Normalise response formats to {role, content, tool_calls}.

Handles OpenAI (data["choices"][0]["message"]) and Ollama (data["message"] at root).

Source code in teff/harness/formats.py

def extract_message(data: dict) -> dict:
    """Normalise response formats to ``{role, content, tool_calls}``.

    Handles OpenAI (``data["choices"][0]["message"]``) and
    Ollama (``data["message"]`` at root).
    """
    choice = (data.get("choices") or [{}])[0]
    msg = choice.get("message", {})
    if not msg and "message" in data:
        msg = data["message"]
    return msg

extract_usage

extract_usage(data)

Extract (prompt_tokens, completion_tokens) from an LLM response.

Handles both OpenAI-style (data["usage"]) and Ollama-style (data["prompt_eval_count"] / data["eval_count"]) formats.

Source code in teff/harness/formats.py

def extract_usage(data: dict) -> tuple[int, int]:
    """Extract ``(prompt_tokens, completion_tokens)`` from an LLM response.

    Handles both OpenAI-style (``data["usage"]``) and Ollama-style
    (``data["prompt_eval_count"]`` / ``data["eval_count"]``) formats.
    """
    usage = data.get("usage") or {}
    prompt = usage.get("prompt_tokens")
    completion = usage.get("completion_tokens")
    if prompt is None:
        prompt = data.get("prompt_eval_count")
    if completion is None:
        completion = data.get("eval_count")
    if prompt is None:
        prompt = usage.get("input_tokens")
    if completion is None:
        completion = usage.get("output_tokens")
    return int(prompt or 0), int(completion or 0)

normalize_text_tool_calls

normalize_text_tool_calls(content, msg, *, seq=0)

Turn a text-embedded tool call into the structured tool_calls list.

When content parses as a single {name, arguments|parameters} object, returns ([tool_call], msg_with_tool_calls); otherwise returns ([], msg) unchanged. The generated call_id is derived from seq + the tool name so it is unique within a run.

Returns:

Type	Description
list[dict]	A (tool_calls, message) pair. message is msg with
dict	tool_calls attached when a text call was found.

Source code in teff/harness/formats.py

def normalize_text_tool_calls(
    content: str, msg: dict, *, seq: int = 0
) -> tuple[list[dict], dict]:
    """Turn a text-embedded tool call into the structured ``tool_calls`` list.

    When ``content`` parses as a single ``{name, arguments|parameters}``
    object, returns ``([tool_call], msg_with_tool_calls)``; otherwise
    returns ``([], msg)`` unchanged.  The generated ``call_id`` is derived
    from *seq* + the tool name so it is unique within a run.

    Returns:
        A ``(tool_calls, message)`` pair.  *message* is *msg* with
        ``tool_calls`` attached when a text call was found.
    """
    parsed = parse_text_tool_call(content)
    if not parsed:
        return [], msg
    name, args = parsed
    call_id = f"call_{seq}_{name}"
    tool_calls = [
        {
            "id": call_id,
            "type": "function",
            "function": {"name": name, "arguments": json.dumps(args)},
        }
    ]
    return tool_calls, {**msg, "tool_calls": tool_calls}

parse_text_tool_call

parse_text_tool_call(content)

Parse a tool call embedded in plain text content.

Local models sometimes emit {"name": "rag", "parameters": {...}} or {"name": "rag", "arguments": {...}} as text instead of using the structured tool_calls field. Returns (name, args) if found.

Source code in teff/harness/formats.py

def parse_text_tool_call(content: str) -> tuple[str, dict] | None:
    """Parse a tool call embedded in plain text content.

    Local models sometimes emit ``{"name": "rag", "parameters": {...}}``
    or ``{"name": "rag", "arguments": {...}}`` as text instead of using
    the structured ``tool_calls`` field. Returns ``(name, args)`` if found.
    """
    m = re.search(r'"name"\s*:\s*"([^"]+)"', content)
    if not m:
        return None
    name = m.group(1)
    for key in ("parameters", "arguments"):
        idx = content.find(f'"{key}"')
        if idx == -1:
            continue
        brace = content.find("{", content.find(":", idx))
        if brace == -1:
            continue
        obj = extract_json_object(content, brace)
        if obj is None:
            continue
        try:
            args = json.loads(obj)
        except json.JSONDecodeError:
            args = {}
        return name, args
    return name, {}