From e0f3685ad2175358a2e7f495b132e437d3653299 Mon Sep 17 00:00:00 2001 From: Manideep Malyala Date: Sun, 23 Aug 2026 00:38:39 +0530 Subject: [PATCH 1/2] feat: Add text-tags fallback mode for LLM tool calling (#1138) --- strix/agents/prompts/system_prompt.jinja | 14 +++ strix/config/models.py | 122 ++++++++++++++++++++++- strix/config/settings.py | 4 + strix/core/runner.py | 2 + strix/interface/cli_args.py | 16 +++ tests/test_runner_root_prompt.py | 2 +- 6 files changed, 158 insertions(+), 2 deletions(-) diff --git a/strix/agents/prompts/system_prompt.jinja b/strix/agents/prompts/system_prompt.jinja index 23493d2d2..f0630536d 100644 --- a/strix/agents/prompts/system_prompt.jinja +++ b/strix/agents/prompts/system_prompt.jinja @@ -505,3 +505,17 @@ On-demand specialist skills. Spawn a specialist via `create_agent(skills=[...])` {% endfor -%} {% endif %} + +{% if system_prompt_context.get("tool_mode") == "text-tags" %} + +You are operating in a text-tag tool mode because this environment lacks native structured tool calling support. +To call a tool, you MUST use the exact following format in your output. Do not output markdown code blocks for the tool call. +Instead, use the `[TOOL: ]` and `[/TOOL]` tags and provide the JSON arguments inside. + +[TOOL: tool_name] +{"argument1": "value1"} +[/TOOL] + +You must provide valid JSON inside the tags. You can only call one tool at a time. Wait for the result before making another tool call. + +{% endif %} diff --git a/strix/config/models.py b/strix/config/models.py index f6848ca4c..2f5042df6 100644 --- a/strix/config/models.py +++ b/strix/config/models.py @@ -445,6 +445,124 @@ def _response_usage(usage: Usage | None) -> ResponseUsage | None: ) +class _TextTagDispatchModel(Model): + """Fallback dispatch mode for models that fail to emit structured tool_calls. + + Extracts [TOOL: name] ... [/TOOL] text tags from the response output and + synthesizes ResponseFunctionToolCall instances so the SDK can execute them. + """ + + def __init__(self, inner: Model) -> None: + self._inner = inner + + async def close(self) -> None: + await self._inner.close() + + def get_retry_advice(self, request: ModelRetryAdviceRequest) -> ModelRetryAdvice | None: + return self._inner.get_retry_advice(request) + + async def get_response( + self, + system_instructions: str | None, + input: str | list[TResponseInputItem], # noqa: A002 + model_settings: ModelSettings, + tools: list[Tool], + output_schema: AgentOutputSchemaBase | None, + handoffs: list[Handoff], + tracing: ModelTracing, + *, + previous_response_id: str | None, + conversation_id: str | None, + prompt: ResponsePromptParam | None, + ) -> ModelResponse: + import re + import uuid + import json + from agents.items import ResponseFunctionToolCall + + # We need the inner get_response first + response = await self._inner.get_response( + system_instructions, + input, + model_settings, + tools, + output_schema, + handoffs, + tracing, + previous_response_id=previous_response_id, + conversation_id=conversation_id, + prompt=prompt, + ) + + TEXT_TAG_PATTERN = re.compile(r"\[TOOL:\s*([^\]]+)\](.*?)\[/TOOL\]", re.DOTALL | re.IGNORECASE) + new_output = [] + + for item in response.output: + if getattr(item, "type", None) == "message": + content = getattr(item, "content", "") + if content and "[TOOL:" in content: + matches = list(TEXT_TAG_PATTERN.finditer(content)) + if matches: + clean_content = TEXT_TAG_PATTERN.sub("", content).strip() + if clean_content: + try: + item.content = clean_content + except AttributeError: + if hasattr(item, "raw_item") and hasattr(item.raw_item, "content"): + item.raw_item.content = clean_content + new_output.append(item) + for match in matches: + tool_name = match.group(1).strip() + tool_args = match.group(2).strip() + # Check if the args are valid JSON, otherwise it will fail gracefully later + try: + json.loads(tool_args) + except ValueError: + pass + + tool_call = ResponseFunctionToolCall( + id=uuid.uuid4().hex[:8], + name=tool_name, + arguments=tool_args, + caller="agent", + ) + new_output.append(tool_call) + continue + new_output.append(item) + + response.output = new_output + return response + + async def stream_response( + self, + system_instructions: str | None, + input: str | list[TResponseInputItem], # noqa: A002 + model_settings: ModelSettings, + tools: list[Tool], + output_schema: AgentOutputSchemaBase | None, + handoffs: list[Handoff], + tracing: ModelTracing, + *, + previous_response_id: str | None, + conversation_id: str | None, + prompt: ResponsePromptParam | None, + ) -> AsyncIterator[TResponseStreamEvent]: + # Text-tag parsing over a stream is complex; delegate to get_response like _NonStreamingModel + response = await self.get_response( + system_instructions, + input, + model_settings, + tools, + output_schema, + handoffs, + tracing, + previous_response_id=previous_response_id, + conversation_id=conversation_id, + prompt=prompt, + ) + yield _completed_stream_event(response, getattr(self._inner, "model", None)) + + class StrixProvider(MultiProvider): """Route any non-OpenAI prefix through LiteLLM with the prefix preserved, so users type ``deepseek/deepseek-chat`` rather than @@ -483,7 +601,9 @@ def get_model(self, model_name: str | None) -> Model: ) else: model = super().get_model(model_name) - if llm.disable_streaming: + if getattr(llm, "tool_mode", "native") == "text-tags": + model = _TextTagDispatchModel(model) + if llm.disable_streaming or getattr(llm, "tool_mode", "native") == "text-tags": model = _NonStreamingModel(model) # The wrapper emits its single event only once the whole request # is done, so an idle gap is meaningless here; the request diff --git a/strix/config/settings.py b/strix/config/settings.py index 42a2c97ea..f8baa00c5 100644 --- a/strix/config/settings.py +++ b/strix/config/settings.py @@ -63,6 +63,10 @@ class LlmSettings(BaseSettings): ge=0, alias="LLM_MAX_TOOL_CALLS_PER_TURN", ) + tool_mode: Literal["native", "text-tags"] = Field( + default="native", + alias="LLM_TOOL_MODE", + ) class DedupeSettings(BaseSettings): diff --git a/strix/core/runner.py b/strix/core/runner.py index 0dfe75d09..22817dd8e 100644 --- a/strix/core/runner.py +++ b/strix/core/runner.py @@ -292,6 +292,8 @@ async def _spill_to_workspace(output_id: str, text: str) -> str | None: coordinator.set_budget_extender(hooks.extend_budget) scope_context = build_scope_context(scan_config) + if isinstance(scope_context, dict): + scope_context["tool_mode"] = getattr(settings.llm, "tool_mode", "native") root_context = _merge_root_prompt_context(scope_context, extra_system_prompt_context) root_instructions = _compose_root_instructions_override( root_instructions_override, diff --git a/strix/interface/cli_args.py b/strix/interface/cli_args.py index 42ebf18d5..e241e37ce 100644 --- a/strix/interface/cli_args.py +++ b/strix/interface/cli_args.py @@ -244,6 +244,18 @@ def parse_arguments() -> argparse.Namespace: ), ) + parser.add_argument( + "--llm-tool-mode", + dest="llm_tool_mode", + choices=["native", "text-tags"], + default="native", + help=( + "Tool dispatch mode for the LLM. Use 'text-tags' to provide a fallback " + "dispatch mode for local or small models that fail to emit structured tool_calls. " + "Default: native." + ), + ) + parser.add_argument( "--resume", type=str, @@ -267,6 +279,10 @@ def parse_arguments() -> argparse.Namespace: if args.config: apply_config_override(validate_config_file(args.config)) + if args.llm_tool_mode and args.llm_tool_mode != "native": + import os + os.environ["LLM_TOOL_MODE"] = args.llm_tool_mode + if args.update: sys.exit(0 if self_update() else 1) diff --git a/tests/test_runner_root_prompt.py b/tests/test_runner_root_prompt.py index 2c3462035..3d0f8cc5b 100644 --- a/tests/test_runner_root_prompt.py +++ b/tests/test_runner_root_prompt.py @@ -177,7 +177,7 @@ async def test_root_prompt_options_default_to_none( kwargs = captured["kwargs"] assert kwargs["instructions_override"] is None - assert kwargs["system_prompt_context"] == {"scope": "built-in"} + assert kwargs["system_prompt_context"] == {"scope": "built-in", "tool_mode": "native"} @pytest.mark.asyncio From 203f26bb946dd36a5bef064e32fa103e0d4a64fd Mon Sep 17 00:00:00 2001 From: Manideep Malyala Date: Sun, 23 Aug 2026 00:53:47 +0530 Subject: [PATCH 2/2] fix: Parse structured content parts in text-tag interceptor --- strix/config/models.py | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/strix/config/models.py b/strix/config/models.py index 2f5042df6..1c14d2b06 100644 --- a/strix/config/models.py +++ b/strix/config/models.py @@ -499,7 +499,20 @@ async def get_response( for item in response.output: if getattr(item, "type", None) == "message": - content = getattr(item, "content", "") + raw_content = getattr(item, "content", "") + + if isinstance(raw_content, list): + content = "" + for part in raw_content: + if isinstance(part, str): + content += part + elif isinstance(part, dict) and "text" in part: + content += part["text"] + elif hasattr(part, "text"): + content += part.text + else: + content = str(raw_content) if raw_content else "" + if content and "[TOOL:" in content: matches = list(TEXT_TAG_PATTERN.finditer(content)) if matches: