fix(fireworks_ai): keep non-text system and developer parts on the folded leading system message

This commit is contained in:
mateo-berri 2026-09-08 11:48:01 -07:00
parent deaadc21d3
commit 4807630c2a
2 changed files with 62 additions and 8 deletions

View file

@ -4,7 +4,12 @@ from typing import TYPE_CHECKING, Final
from urllib.parse import unquote
import httpx
from openai.types.responses import EasyInputMessageParam, ResponseInputItemParam
from openai.types.responses import (
EasyInputMessageParam,
ResponseInputContentParam,
ResponseInputItemParam,
ResponseInputTextParam,
)
from pydantic import TypeAdapter
from litellm.llms.fireworks_ai.common_utils import (
@ -35,27 +40,42 @@ def _session_params(litellm_params: GenericLiteLLMParams) -> Mapping[str, object
_instructions_adapter: Final = TypeAdapter[str | None](str | None)
def _instruction_text(item: ResponseInputItemParam) -> str | None:
def _instruction_parts(item: ResponseInputItemParam) -> tuple[ResponseInputContentParam, ...] | None:
if "role" not in item or (item["role"] != "system" and item["role"] != "developer"):
return None
content: Final = item["content"]
if isinstance(content, str):
return content
return "\n\n".join(part["text"] for part in content if part["type"] == "input_text")
return (ResponseInputTextParam(type="input_text", text=content),)
return tuple(content)
def _leading_system_content(
instructions: str | None, parts: tuple[ResponseInputContentParam, ...]
) -> str | list[ResponseInputContentParam]:
text: Final = "\n\n".join(
chunk for chunk in (instructions, *(part["text"] for part in parts if part["type"] == "input_text")) if chunk
)
non_text: Final = tuple(part for part in parts if part["type"] != "input_text")
if not non_text:
return text
return [ResponseInputTextParam(type="input_text", text=text), *non_text] if text else list(non_text)
def _with_single_leading_system_item(
input: str | ResponseInputParam, instructions: str | None
) -> str | ResponseInputParam:
items: Final = () if isinstance(input, str) else tuple(input)
instruction_texts: Final = tuple(text for text in (instructions, *map(_instruction_text, items)) if text)
if not instruction_texts:
instruction_parts: Final = tuple(
part for item_parts in map(_instruction_parts, items) if item_parts is not None for part in item_parts
)
content: Final = _leading_system_content(instructions, instruction_parts)
if not content:
return input
leading: Final = EasyInputMessageParam(role="system", content="\n\n".join(instruction_texts), type="message")
leading: Final = EasyInputMessageParam(role="system", content=content, type="message")
rest: Final = (
(EasyInputMessageParam(role="user", content=input),)
if isinstance(input, str)
else tuple(item for item in items if _instruction_text(item) is None)
else tuple(item for item in items if _instruction_parts(item) is None)
)
return [leading, *rest]

View file

@ -231,6 +231,40 @@ def test_responses_call_folds_instructions_and_developer_item_into_one_leading_s
)
def test_responses_call_keeps_non_text_developer_parts_on_the_leading_system_message() -> None:
client: Final = _mock_http_client(_fireworks_response("accounts/fireworks/models/qwen3p8-2p4t-a95b"))
with patch(HTTPX_CLIENT_FACTORY, return_value=client):
litellm.responses(
model="fireworks_ai/accounts/fireworks/models/qwen3p8-2p4t-a95b",
instructions="Answer with one word.",
input=[ # mutable-ok: the Responses API takes input as a JSON list
{
"role": "developer",
"content": [
{"type": "input_text", "text": "Match the style of this reference image."},
{"type": "input_image", "image_url": "data:image/png;base64,iVBORw0KGgo=", "detail": "auto"},
],
},
{"role": "user", "content": "What is the capital of France?"},
],
store=False,
api_key="fw-test-key",
)
_, _, body = _sent_request(client)
assert "instructions" not in body
assert tuple(body["input"]) == (
{
"role": "system",
"content": [
{"type": "input_text", "text": "Answer with one word.\n\nMatch the style of this reference image."},
{"type": "input_image", "image_url": "data:image/png;base64,iVBORw0KGgo=", "detail": "auto"},
],
"type": "message",
},
{"role": "user", "content": "What is the capital of France?"},
)
def test_responses_call_turns_string_input_with_instructions_into_system_then_user_messages() -> None:
client: Final = _mock_http_client(_fireworks_response("accounts/fireworks/models/kimi-k3"))
with patch(HTTPX_CLIENT_FACTORY, return_value=client):