groq: parse legacy <function=NAME{...}> tool-call tags correctly

Groq models that emit the legacy <function=NAME{...args...}></function>
text format were being forwarded with the entire NAME{...args...} blob as
the tool name, causing tool_use_failed errors from Groq's API.

Add _parse_legacy_function_tool_calls() that splits the bare tool name
off at the first '{' or whitespace, JSON-decodes the trailing args, and
raises OutputParserException for unknown tool names. Wire it into
_convert_dict_to_message() as a fallback when no structured tool_calls
are present, and cover both the parser and the end-to-end conversion
path with unit tests.
This commit is contained in:
LangSmith Issues Agent
2026-05-26 19:00:00 +00:00
parent 7bb4130c7d
commit 43f09eecf0
2 changed files with 134 additions and 0 deletions

View File

@@ -12,6 +12,7 @@ from langchain_core.callbacks import (
AsyncCallbackManagerForLLMRun,
CallbackManagerForLLMRun,
)
from langchain_core.exceptions import OutputParserException
from langchain_core.language_models import (
LanguageModelInput,
ModelProfile,
@@ -1473,6 +1474,81 @@ def _convert_chunk_to_message_chunk(
return default_class(content=content) # type: ignore[call-arg]
def _parse_legacy_function_tool_calls(
text: str,
allowed_tool_names: set[str] | None = None,
) -> list[dict]:
"""Parse legacy ``<function=NAME{...args...}>...</function>`` tool-call tags.
The tool name is taken as everything after ``<function=`` up to (but
not including) the first ``{`` or whitespace -- never the full
``NAME{...args...}`` blob. Whatever follows (up to the closing ``>``)
is parsed as JSON arguments.
"""
if not text or "<function=" not in text:
return []
tool_calls: list[dict] = []
cursor = 0
while True:
start = text.find("<function=", cursor)
if start == -1:
break
header_start = start + len("<function=")
header_end = text.find(">", header_start)
if header_end == -1:
break
header = text[header_start:header_end]
brace_idx = header.find("{")
ws_idx = -1
for i, ch in enumerate(header):
if ch.isspace():
ws_idx = i
break
split_candidates = [i for i in (brace_idx, ws_idx) if i != -1]
if split_candidates:
split_idx = min(split_candidates)
name = header[:split_idx].strip()
args_blob = header[split_idx:].strip()
else:
name = header.strip()
args_blob = ""
if not name:
cursor = header_end + 1
continue
if allowed_tool_names is not None and name not in allowed_tool_names:
msg = (
f"Model emitted legacy <function=...> tool call for "
f"'{name}', which is not in the configured tools "
f"({sorted(allowed_tool_names)})."
)
raise OutputParserException(msg)
if args_blob:
try:
args = json.loads(args_blob)
except json.JSONDecodeError as e:
msg = (
f"Failed to JSON-decode arguments for legacy "
f"<function={name}...> tool call: {e}. "
f"Arguments blob was: {args_blob!r}"
)
raise OutputParserException(msg) from e
else:
args = {}
tool_calls.append(
{
"type": "function",
"id": None,
"function": {
"name": name,
"arguments": json.dumps(args, ensure_ascii=False),
},
}
)
close = text.find("</function>", header_end)
cursor = (close + len("</function>")) if close != -1 else (header_end + 1)
return tool_calls
def _convert_dict_to_message(_dict: Mapping[str, Any]) -> BaseMessage:
"""Convert a dictionary to a LangChain message.
@@ -1515,6 +1591,23 @@ def _convert_dict_to_message(_dict: Mapping[str, Any]) -> BaseMessage:
invalid_tool_calls.append(
make_invalid_tool_call(raw_tool_call, str(e))
)
elif isinstance(content, str) and "<function=" in content:
# Some Groq models emit tool calls in the legacy
# ``<function=NAME{...args...}>...</function>`` text format
# rather than as structured ``tool_calls``. Extract them so the
# bare tool name (not ``NAME{...args...}``) is forwarded.
legacy_raw_tool_calls = _parse_legacy_function_tool_calls(content)
if legacy_raw_tool_calls:
additional_kwargs["tool_calls"] = legacy_raw_tool_calls
for raw_tool_call in legacy_raw_tool_calls:
try:
tool_calls.append(
parse_tool_call(raw_tool_call, return_id=True)
)
except Exception as e: # pylint: disable=broad-except
invalid_tool_calls.append(
make_invalid_tool_call(raw_tool_call, str(e))
)
return AIMessage(
content=content,
id=id_,

View File

@@ -7,6 +7,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
import langchain_core.load as lc_load
import pytest
from langchain_core.exceptions import OutputParserException
from langchain_core.messages import (
AIMessage,
AIMessageChunk,
@@ -25,6 +26,7 @@ from langchain_groq.chat_models import (
_convert_dict_to_message,
_create_usage_metadata,
_format_message_content,
_parse_legacy_function_tool_calls,
)
if "GROQ_API_KEY" not in os.environ:
@@ -143,6 +145,45 @@ def test__convert_dict_to_message_tool_call() -> None:
assert result == expected_output
def test__parse_legacy_function_tool_calls_strips_args_from_name() -> None:
text = '<function=my_tool {"x": 1}></function>'
result = _parse_legacy_function_tool_calls(text)
assert len(result) == 1
assert result[0]["function"]["name"] == "my_tool"
assert json.loads(result[0]["function"]["arguments"]) == {"x": 1}
def test__parse_legacy_function_tool_calls_no_whitespace() -> None:
result = _parse_legacy_function_tool_calls(
'<function=generator_function{"type": "greeting", "style": "Pirate"}>'
"</function>"
)
assert len(result) == 1
assert result[0]["function"]["name"] == "generator_function"
assert json.loads(result[0]["function"]["arguments"]) == {
"type": "greeting",
"style": "Pirate",
}
def test__parse_legacy_function_tool_calls_unknown_tool_raises() -> None:
text = '<function=bogus_tool {"x": 1}></function>'
with pytest.raises(OutputParserException, match="bogus_tool"):
_parse_legacy_function_tool_calls(text, allowed_tool_names={"my_tool"})
def test__convert_dict_to_message_legacy_function_tag() -> None:
message = {
"role": "assistant",
"content": '<function=my_tool {"x": 1}></function>',
}
result = _convert_dict_to_message(message)
assert isinstance(result, AIMessage)
assert len(result.tool_calls) == 1
assert result.tool_calls[0]["name"] == "my_tool"
assert result.tool_calls[0]["args"] == {"x": 1}
def test__convert_dict_to_message_system() -> None:
message = {"role": "system", "content": "foo"}
result = _convert_dict_to_message(message)