mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-03 02:22:24 +00:00
test(e2e): add conversational matrix across chat, messages and responses (#42359)
* test(e2e): add conversational matrix across chat, messages and responses Parameterizes one behavioral contract (reply, stream, cost log, tool call, tool round trip) across /v1/chat/completions, /v1/messages and /v1/responses, OpenAI and Anthropic models, and env-ref vs stored-credential auth, with record/replay fixtures. Adds general_settings.disable_model_info_refresh so the proxy fronting a replay fixture does not poll every OpenAI-compatible deployment's /v1/models in the background, which otherwise leaves unconsumed interactions in the recorded bundle. Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(e2e): force the weather tool on the first turn and rename Provider to Deployment Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: mateo <mateo@berri.ai> Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot]
GitHub
parent
4936427094
commit
30d8b12512
6 changed files with 767 additions and 12 deletions
|
|
@ -1494,23 +1494,29 @@ async def proxy_startup_event(app: FastAPI) -> AsyncGenerator[None, None]:
|
|||
## Initialize shared aiohttp session for connection reuse
|
||||
shared_aiohttp_session = await _initialize_shared_aiohttp_session()
|
||||
|
||||
model_info_scheduler: Final = scheduler if scheduler is not None else AsyncIOScheduler()
|
||||
model_info_scheduler.add_job(
|
||||
ProxyStartupEvent.refresh_model_info,
|
||||
"interval",
|
||||
seconds=MODEL_INFO_REFRESH_SECONDS,
|
||||
id="refresh_model_info",
|
||||
next_run_time=datetime.now(timezone.utc),
|
||||
max_instances=1,
|
||||
replace_existing=True,
|
||||
model_info_refresh_disabled: Final = (
|
||||
"disable_model_info_refresh" in general_settings and general_settings["disable_model_info_refresh"] is True
|
||||
)
|
||||
if not model_info_scheduler.running:
|
||||
model_info_scheduler.start()
|
||||
model_info_scheduler: Final = (
|
||||
None if model_info_refresh_disabled else scheduler if scheduler is not None else AsyncIOScheduler()
|
||||
)
|
||||
if model_info_scheduler is not None:
|
||||
model_info_scheduler.add_job(
|
||||
ProxyStartupEvent.refresh_model_info,
|
||||
"interval",
|
||||
seconds=MODEL_INFO_REFRESH_SECONDS,
|
||||
id="refresh_model_info",
|
||||
next_run_time=datetime.now(timezone.utc),
|
||||
max_instances=1,
|
||||
replace_existing=True,
|
||||
)
|
||||
if not model_info_scheduler.running:
|
||||
model_info_scheduler.start()
|
||||
|
||||
# End of startup event
|
||||
yield
|
||||
|
||||
if model_info_scheduler.running:
|
||||
if model_info_scheduler is not None and model_info_scheduler.running:
|
||||
model_info_scheduler.remove_job("refresh_model_info")
|
||||
if model_info_scheduler is not scheduler:
|
||||
model_info_scheduler.shutdown(wait=False)
|
||||
|
|
|
|||
|
|
@ -94,3 +94,15 @@
|
|||
- {id: llm.messages.together_ai.basic.stream.works, module: llm, tier: P1, subject_endpoint: messages, route: together_ai, capability: basic, streaming: stream, assertions: [works], source: "llm_translation/test_together_ai_e2e.py", rationale: "Together over /v1/messages streaming"}
|
||||
- {id: llm.messages.together_ai.tool_use.nonstream.works, module: llm, tier: P1, subject_endpoint: messages, route: together_ai, capability: tool_use, streaming: nonstream, assertions: [works], source: "llm_translation/test_together_ai_e2e.py", rationale: "Together tool calls over /v1/messages"}
|
||||
- {id: llm.messages.together_ai.multi_turn.nonstream.works, module: llm, tier: P1, subject_endpoint: messages, route: together_ai, capability: multi_turn, streaming: nonstream, assertions: [works], source: "llm_translation/test_together_ai_e2e.py", rationale: "Together tool result round trip over /v1/messages"}
|
||||
- {id: llm.chat_completions.anthropic.basic.nonstream.cost_logged, module: llm, tier: P0, subject_endpoint: chat_completions, route: anthropic, capability: basic, streaming: nonstream, assertions: [works, cost_logged], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "Anthropic over /chat/completions: cost header and spend row agree"}
|
||||
- {id: llm.chat_completions.anthropic.multi_turn.nonstream.works, module: llm, tier: P0, subject_endpoint: chat_completions, route: anthropic, capability: multi_turn, streaming: nonstream, assertions: [works], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "Anthropic tool result round trip over /chat/completions"}
|
||||
- {id: llm.messages.anthropic.multi_turn.nonstream.works, module: llm, tier: P0, subject_endpoint: messages, route: anthropic, capability: multi_turn, streaming: nonstream, assertions: [works], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "Anthropic tool result round trip over /v1/messages"}
|
||||
- {id: llm.messages.openai.basic.nonstream.works, module: llm, tier: P0, subject_endpoint: messages, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "OpenAI models served on the Anthropic Messages contract"}
|
||||
- {id: llm.messages.openai.basic.stream.works, module: llm, tier: P0, subject_endpoint: messages, route: openai, capability: basic, streaming: stream, assertions: [works], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "OpenAI over /v1/messages streams the Anthropic event grammar"}
|
||||
- {id: llm.messages.openai.basic.nonstream.cost_logged, module: llm, tier: P0, subject_endpoint: messages, route: openai, capability: basic, streaming: nonstream, assertions: [works, cost_logged], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "OpenAI over /v1/messages: cost header and spend row agree"}
|
||||
- {id: llm.messages.openai.tool_use.nonstream.works, module: llm, tier: P0, subject_endpoint: messages, route: openai, capability: tool_use, streaming: nonstream, assertions: [works], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "OpenAI tool calls translated to Anthropic tool_use blocks"}
|
||||
- {id: llm.messages.openai.multi_turn.nonstream.works, module: llm, tier: P0, subject_endpoint: messages, route: openai, capability: multi_turn, streaming: nonstream, assertions: [works], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "OpenAI tool result round trip over /v1/messages"}
|
||||
- {id: llm.responses.openai.multi_turn.nonstream.works, module: llm, tier: P0, subject_endpoint: responses, route: openai, capability: multi_turn, streaming: nonstream, assertions: [works], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "OpenAI function_call_output round trip over /v1/responses"}
|
||||
- {id: llm.responses.anthropic.basic.stream.works, module: llm, tier: P1, subject_endpoint: responses, route: anthropic, capability: basic, streaming: stream, assertions: [works], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "Anthropic over /v1/responses streams Responses events"}
|
||||
- {id: llm.responses.anthropic.basic.nonstream.cost_logged, module: llm, tier: P1, subject_endpoint: responses, route: anthropic, capability: basic, streaming: nonstream, assertions: [works, cost_logged], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "Anthropic over /v1/responses: cost header and spend row agree"}
|
||||
- {id: llm.responses.anthropic.multi_turn.nonstream.works, module: llm, tier: P1, subject_endpoint: responses, route: anthropic, capability: multi_turn, streaming: nonstream, assertions: [works], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "Anthropic function_call_output round trip over /v1/responses"}
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
general_settings:
|
||||
master_key: os.environ/LITELLM_MASTER_KEY
|
||||
store_model_in_db: true
|
||||
disable_model_info_refresh: true
|
||||
|
|
|
|||
544
tests/e2e/llm_translation/conversational_matrix.py
Normal file
544
tests/e2e/llm_translation/conversational_matrix.py
Normal file
|
|
@ -0,0 +1,544 @@
|
|||
"""The endpoint x deployment x auth matrix behind test_conversational_matrix_e2e.py.
|
||||
|
||||
One conversation, three wire formats. Each `Surface` speaks its own API through
|
||||
the customer SDK (chat completions and Responses through the OpenAI SDK, Messages
|
||||
through the Anthropic SDK) and folds what came back into the surface-neutral
|
||||
`Reply` / `StreamedReply`, so a single behavior test asserts the same contract on
|
||||
every cell. A new model, from an existing or a new provider, is one `Deployment` row
|
||||
in DEPLOYMENTS; a new way of handing the proxy a provider credential is one
|
||||
`AuthMethod`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from collections.abc import Iterator, Mapping, Sequence
|
||||
from dataclasses import dataclass
|
||||
from types import MappingProxyType
|
||||
from typing import Final, Literal, Protocol
|
||||
|
||||
import anthropic
|
||||
import openai
|
||||
import pytest
|
||||
from _pytest.mark.structures import ParameterSet
|
||||
from anthropic.types import (
|
||||
MessageParam,
|
||||
RawMessageStreamEvent,
|
||||
TextBlock,
|
||||
ToolChoiceToolParam,
|
||||
ToolParam,
|
||||
ToolResultBlockParam,
|
||||
ToolUseBlock,
|
||||
ToolUseBlockParam,
|
||||
)
|
||||
from e2e_config import provider_edge_base, unique_marker
|
||||
from lifecycle import ResourceManager
|
||||
from llm_translation.sdk_clients import NO_PROXY_CACHE, SdkClients, response_header
|
||||
from models import CredentialCreateBody, LiteLLMParamsBody
|
||||
from openai.types.chat import (
|
||||
ChatCompletionAssistantMessageParam,
|
||||
ChatCompletionChunk,
|
||||
ChatCompletionMessageFunctionToolCallParam,
|
||||
ChatCompletionMessageParam,
|
||||
ChatCompletionNamedToolChoiceParam,
|
||||
ChatCompletionToolMessageParam,
|
||||
ChatCompletionToolParam,
|
||||
)
|
||||
from openai.types.chat.chat_completion_message_function_tool_call import ChatCompletionMessageFunctionToolCall
|
||||
from openai.types.responses import (
|
||||
FunctionToolParam,
|
||||
ResponseFunctionToolCall,
|
||||
ResponseFunctionToolCallParam,
|
||||
ResponseInputParam,
|
||||
ResponseStreamEvent,
|
||||
ToolChoiceFunctionParam,
|
||||
)
|
||||
from openai.types.responses.response_input_param import FunctionCallOutput
|
||||
from proxy_client import ProxyClient
|
||||
from pydantic import BaseModel
|
||||
|
||||
SurfaceName = Literal["chat_completions", "messages", "responses"]
|
||||
AuthMethod = Literal["env_ref", "stored_credential"]
|
||||
Capability = Literal["basic", "tool_use", "multi_turn"]
|
||||
Streaming = Literal["stream", "nonstream"]
|
||||
Assertion = Literal["works", "cost_logged"]
|
||||
ToolMode = Literal["none", "forced", "offered"]
|
||||
|
||||
SURFACES: Final[tuple[SurfaceName, ...]] = ("chat_completions", "messages", "responses")
|
||||
AUTH_METHODS: Final[tuple[AuthMethod, ...]] = ("env_ref", "stored_credential")
|
||||
|
||||
MAX_OUTPUT_TOKENS: Final = 512
|
||||
INSTRUCTIONS: Final = "You are a terse assistant. Answer in one short sentence."
|
||||
GREETING_PROMPT: Final = "Say hello."
|
||||
WEATHER_PROMPT: Final = "What is the weather in Paris right now? Use the get_weather tool."
|
||||
WEATHER_REPORT: Final = "Paris: 22 degrees Celsius, clear skies"
|
||||
WEATHER_TOOL_NAME: Final = "get_weather"
|
||||
WEATHER_TOOL_DESCRIPTION: Final = "Current weather for a city"
|
||||
WEATHER_TOOL_SCHEMA: Final[Mapping[str, object]] = MappingProxyType(
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {"location": {"type": "string", "description": "City name"}},
|
||||
"required": ["location"],
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Deployment:
|
||||
"""One deployment target: the litellm backend string plus how to wire it."""
|
||||
|
||||
route: Literal["openai", "anthropic"]
|
||||
label: str
|
||||
backend: str
|
||||
api_key_env: str
|
||||
edge_mount: str
|
||||
edge_suffix: str
|
||||
|
||||
def api_base(self) -> str | None:
|
||||
base: Final = provider_edge_base(self.edge_mount)
|
||||
return None if base is None else f"{base}{self.edge_suffix}"
|
||||
|
||||
def api_key(self) -> str:
|
||||
key: Final = os.environ.get(self.api_key_env, "")
|
||||
assert key, f"{self.api_key_env} is not set in the test process environment"
|
||||
return key
|
||||
|
||||
|
||||
DEPLOYMENTS: Final[tuple[Deployment, ...]] = (
|
||||
Deployment(
|
||||
route="openai",
|
||||
label="gpt-4o-mini",
|
||||
backend="openai/gpt-4o-mini",
|
||||
api_key_env="OPENAI_API_KEY",
|
||||
edge_mount="openai",
|
||||
edge_suffix="/v1",
|
||||
),
|
||||
Deployment(
|
||||
route="openai",
|
||||
label="gpt-5.4-mini",
|
||||
backend="openai/gpt-5.4-mini",
|
||||
api_key_env="OPENAI_API_KEY",
|
||||
edge_mount="openai",
|
||||
edge_suffix="/v1",
|
||||
),
|
||||
Deployment(
|
||||
route="anthropic",
|
||||
label="claude-haiku-4-5",
|
||||
backend="anthropic/claude-haiku-4-5",
|
||||
api_key_env="ANTHROPIC_API_KEY",
|
||||
edge_mount="anthropic",
|
||||
edge_suffix="",
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Cell:
|
||||
surface: SurfaceName
|
||||
deployment: Deployment
|
||||
auth: AuthMethod
|
||||
|
||||
@property
|
||||
def id(self) -> str:
|
||||
return f"{self.surface}-{self.deployment.label}-{self.auth}"
|
||||
|
||||
def registry_id(self, capability: Capability, streaming: Streaming, assertion: Assertion) -> str:
|
||||
return f"llm.{self.surface}.{self.deployment.route}.{capability}.{streaming}.{assertion}"
|
||||
|
||||
|
||||
CELLS: Final[tuple[Cell, ...]] = tuple(
|
||||
Cell(surface=surface, deployment=deployment, auth=auth)
|
||||
for surface in SURFACES
|
||||
for deployment in DEPLOYMENTS
|
||||
for auth in AUTH_METHODS
|
||||
)
|
||||
|
||||
|
||||
def cells_covering(capability: Capability, streaming: Streaming, assertion: Assertion) -> tuple[ParameterSet, ...]:
|
||||
"""Every cell as a pytest param carrying the registry id its test proves."""
|
||||
return tuple(
|
||||
pytest.param(cell, id=cell.id, marks=pytest.mark.covers(cell.registry_id(capability, streaming, assertion)))
|
||||
for cell in CELLS
|
||||
)
|
||||
|
||||
|
||||
DeploymentKey = tuple[str, AuthMethod]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Deployments:
|
||||
"""Model aliases registered on the proxy, one per (deployment, auth)."""
|
||||
|
||||
aliases: Mapping[DeploymentKey, str]
|
||||
|
||||
def alias(self, cell: Cell) -> str:
|
||||
return self.aliases[(cell.deployment.label, cell.auth)]
|
||||
|
||||
|
||||
def _litellm_params(deployment: Deployment, auth: AuthMethod, credential_name: str) -> LiteLLMParamsBody:
|
||||
match auth:
|
||||
case "env_ref":
|
||||
return LiteLLMParamsBody(
|
||||
model=deployment.backend, api_key=f"os.environ/{deployment.api_key_env}", api_base=deployment.api_base()
|
||||
)
|
||||
case "stored_credential":
|
||||
return LiteLLMParamsBody(
|
||||
model=deployment.backend, litellm_credential_name=credential_name, api_base=deployment.api_base()
|
||||
)
|
||||
|
||||
|
||||
def _register(proxy: ProxyClient, resources: ResourceManager, deployment: Deployment, auth: AuthMethod) -> str:
|
||||
marker: Final = unique_marker()
|
||||
credential_name: Final = f"e2e-matrix-{deployment.label}-{marker}"
|
||||
if auth == "stored_credential":
|
||||
proxy.create_credential(
|
||||
CredentialCreateBody(credential_name=credential_name, credential_values={"api_key": deployment.api_key()})
|
||||
)
|
||||
resources.defer(lambda: proxy.delete_credential(credential_name))
|
||||
alias: Final = f"e2e-matrix-{deployment.label}-{auth}-{marker}"
|
||||
model_id: Final = proxy.create_model(alias, _litellm_params(deployment, auth, credential_name))
|
||||
resources.defer(lambda: proxy.delete_model(model_id))
|
||||
return alias
|
||||
|
||||
|
||||
def register_deployments(proxy: ProxyClient) -> Iterator[Deployments]:
|
||||
resources: Final = ResourceManager(client=proxy)
|
||||
try:
|
||||
yield Deployments(
|
||||
aliases=MappingProxyType(
|
||||
{
|
||||
(deployment.label, auth): _register(proxy, resources, deployment, auth)
|
||||
for deployment in DEPLOYMENTS
|
||||
for auth in AUTH_METHODS
|
||||
}
|
||||
)
|
||||
)
|
||||
finally:
|
||||
resources.teardown()
|
||||
|
||||
|
||||
class WeatherArgs(BaseModel):
|
||||
location: str
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ToolCall:
|
||||
call_id: str
|
||||
name: str
|
||||
arguments: str
|
||||
|
||||
def parsed(self) -> WeatherArgs:
|
||||
return WeatherArgs.model_validate_json(self.arguments)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Usage:
|
||||
input_tokens: int
|
||||
output_tokens: int
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Reply:
|
||||
"""What every surface owes the caller for one non-streamed turn."""
|
||||
|
||||
response_id: str
|
||||
text: str
|
||||
tool_calls: tuple[ToolCall, ...]
|
||||
usage: Usage | None
|
||||
call_id_header: str | None
|
||||
cost_header: str | None
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class StreamedReply:
|
||||
"""The reassembled stream: its text, whether the surface's own terminal event
|
||||
arrived, and whether usage was reported anywhere in the stream."""
|
||||
|
||||
text: str
|
||||
finished: bool
|
||||
usage_reported: bool
|
||||
event_count: int
|
||||
|
||||
|
||||
class Surface(Protocol):
|
||||
@property
|
||||
def name(self) -> SurfaceName: ...
|
||||
|
||||
def reply(self, key: str, model: str, prompt: str, *, with_tool: bool = False) -> Reply: ...
|
||||
|
||||
def stream(self, key: str, model: str, prompt: str) -> StreamedReply: ...
|
||||
|
||||
def reply_to_tool_result(self, key: str, model: str, prompt: str, call: ToolCall, result: str) -> Reply: ...
|
||||
|
||||
|
||||
def _chat_tool() -> ChatCompletionToolParam:
|
||||
return {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": WEATHER_TOOL_NAME,
|
||||
"description": WEATHER_TOOL_DESCRIPTION,
|
||||
"parameters": dict(WEATHER_TOOL_SCHEMA),
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _messages_tool() -> ToolParam:
|
||||
return {
|
||||
"name": WEATHER_TOOL_NAME,
|
||||
"description": WEATHER_TOOL_DESCRIPTION,
|
||||
"input_schema": dict(WEATHER_TOOL_SCHEMA),
|
||||
}
|
||||
|
||||
|
||||
def _responses_tool() -> FunctionToolParam:
|
||||
return {
|
||||
"type": "function",
|
||||
"name": WEATHER_TOOL_NAME,
|
||||
"description": WEATHER_TOOL_DESCRIPTION,
|
||||
"parameters": dict(WEATHER_TOOL_SCHEMA),
|
||||
"strict": False,
|
||||
}
|
||||
|
||||
|
||||
def _chat_tool_choice() -> ChatCompletionNamedToolChoiceParam:
|
||||
return {"type": "function", "function": {"name": WEATHER_TOOL_NAME}}
|
||||
|
||||
|
||||
def _messages_tool_choice() -> ToolChoiceToolParam:
|
||||
return {"type": "tool", "name": WEATHER_TOOL_NAME, "disable_parallel_tool_use": True}
|
||||
|
||||
|
||||
def _responses_tool_choice() -> ToolChoiceFunctionParam:
|
||||
return {"type": "function", "name": WEATHER_TOOL_NAME}
|
||||
|
||||
|
||||
def _usage(input_tokens: int | None, output_tokens: int | None) -> Usage | None:
|
||||
if input_tokens is None or output_tokens is None:
|
||||
return None
|
||||
return Usage(input_tokens=input_tokens, output_tokens=output_tokens)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ChatCompletionsSurface:
|
||||
sdk: SdkClients
|
||||
name: SurfaceName = "chat_completions"
|
||||
|
||||
def _turn(self, key: str, model: str, messages: Sequence[ChatCompletionMessageParam], tool: ToolMode) -> Reply:
|
||||
raw: Final = self.sdk.openai(key).chat.completions.with_raw_response.create(
|
||||
model=model,
|
||||
messages=list(messages),
|
||||
max_completion_tokens=MAX_OUTPUT_TOKENS,
|
||||
tools=openai.omit if tool == "none" else [_chat_tool()],
|
||||
tool_choice=_chat_tool_choice() if tool == "forced" else openai.omit,
|
||||
parallel_tool_calls=False if tool == "forced" else openai.omit,
|
||||
extra_body=NO_PROXY_CACHE,
|
||||
)
|
||||
completion: Final = raw.parse()
|
||||
message: Final = completion.choices[0].message
|
||||
calls: Final = tuple(
|
||||
ToolCall(call_id=call.id, name=call.function.name, arguments=call.function.arguments)
|
||||
for call in message.tool_calls or ()
|
||||
if isinstance(call, ChatCompletionMessageFunctionToolCall)
|
||||
)
|
||||
return Reply(
|
||||
response_id=completion.id,
|
||||
text=message.content or "",
|
||||
tool_calls=calls,
|
||||
usage=None
|
||||
if completion.usage is None
|
||||
else _usage(completion.usage.prompt_tokens, completion.usage.completion_tokens),
|
||||
call_id_header=response_header(raw.headers, "x-litellm-call-id"),
|
||||
cost_header=response_header(raw.headers, "x-litellm-response-cost"),
|
||||
)
|
||||
|
||||
def reply(self, key: str, model: str, prompt: str, *, with_tool: bool = False) -> Reply:
|
||||
return self._turn(key, model, _chat_history(prompt), "forced" if with_tool else "none")
|
||||
|
||||
def stream(self, key: str, model: str, prompt: str) -> StreamedReply:
|
||||
chunks: Final[tuple[ChatCompletionChunk, ...]] = tuple(
|
||||
self.sdk.openai(key).chat.completions.create(
|
||||
model=model,
|
||||
messages=_chat_history(prompt),
|
||||
max_completion_tokens=MAX_OUTPUT_TOKENS,
|
||||
stream=True,
|
||||
stream_options={"include_usage": True},
|
||||
extra_body=NO_PROXY_CACHE,
|
||||
)
|
||||
)
|
||||
return StreamedReply(
|
||||
text="".join(chunk.choices[0].delta.content or "" for chunk in chunks if chunk.choices),
|
||||
finished=any(chunk.choices[0].finish_reason is not None for chunk in chunks if chunk.choices),
|
||||
usage_reported=any(chunk.usage is not None for chunk in chunks),
|
||||
event_count=len(chunks),
|
||||
)
|
||||
|
||||
def reply_to_tool_result(self, key: str, model: str, prompt: str, call: ToolCall, result: str) -> Reply:
|
||||
tool_call: Final[ChatCompletionMessageFunctionToolCallParam] = {
|
||||
"id": call.call_id,
|
||||
"type": "function",
|
||||
"function": {"name": call.name, "arguments": call.arguments},
|
||||
}
|
||||
assistant: Final[ChatCompletionAssistantMessageParam] = {"role": "assistant", "tool_calls": [tool_call]}
|
||||
tool_result: Final[ChatCompletionToolMessageParam] = {
|
||||
"role": "tool",
|
||||
"tool_call_id": call.call_id,
|
||||
"content": result,
|
||||
}
|
||||
return self._turn(key, model, (*_chat_history(prompt), assistant, tool_result), "offered")
|
||||
|
||||
|
||||
def _chat_history(prompt: str) -> tuple[ChatCompletionMessageParam, ...]:
|
||||
return ({"role": "system", "content": INSTRUCTIONS}, {"role": "user", "content": prompt})
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class MessagesSurface:
|
||||
sdk: SdkClients
|
||||
name: SurfaceName = "messages"
|
||||
|
||||
def _turn(self, key: str, model: str, messages: Sequence[MessageParam], tool: ToolMode) -> Reply:
|
||||
raw: Final = self.sdk.anthropic(key).messages.with_raw_response.create(
|
||||
model=model,
|
||||
max_tokens=MAX_OUTPUT_TOKENS,
|
||||
system=INSTRUCTIONS,
|
||||
messages=list(messages),
|
||||
tools=anthropic.omit if tool == "none" else [_messages_tool()],
|
||||
tool_choice=_messages_tool_choice() if tool == "forced" else anthropic.omit,
|
||||
extra_body=NO_PROXY_CACHE,
|
||||
)
|
||||
message: Final = raw.parse()
|
||||
return Reply(
|
||||
response_id=message.id,
|
||||
text="".join(block.text for block in message.content if isinstance(block, TextBlock)),
|
||||
tool_calls=tuple(
|
||||
ToolCall(call_id=block.id, name=block.name, arguments=json.dumps(block.input))
|
||||
for block in message.content
|
||||
if isinstance(block, ToolUseBlock)
|
||||
),
|
||||
usage=_usage(message.usage.input_tokens, message.usage.output_tokens),
|
||||
call_id_header=response_header(raw.headers, "x-litellm-call-id"),
|
||||
cost_header=response_header(raw.headers, "x-litellm-response-cost"),
|
||||
)
|
||||
|
||||
def reply(self, key: str, model: str, prompt: str, *, with_tool: bool = False) -> Reply:
|
||||
return self._turn(key, model, ({"role": "user", "content": prompt},), "forced" if with_tool else "none")
|
||||
|
||||
def stream(self, key: str, model: str, prompt: str) -> StreamedReply:
|
||||
events: Final[tuple[RawMessageStreamEvent, ...]] = tuple(
|
||||
self.sdk.anthropic(key).messages.create(
|
||||
model=model,
|
||||
max_tokens=MAX_OUTPUT_TOKENS,
|
||||
system=INSTRUCTIONS,
|
||||
messages=[{"role": "user", "content": prompt}],
|
||||
stream=True,
|
||||
extra_body=NO_PROXY_CACHE,
|
||||
)
|
||||
)
|
||||
return StreamedReply(
|
||||
text="".join(
|
||||
event.delta.text
|
||||
for event in events
|
||||
if event.type == "content_block_delta" and event.delta.type == "text_delta"
|
||||
),
|
||||
finished=any(event.type == "message_stop" for event in events),
|
||||
usage_reported=any(event.type == "message_delta" and event.usage.output_tokens > 0 for event in events),
|
||||
event_count=len(events),
|
||||
)
|
||||
|
||||
def reply_to_tool_result(self, key: str, model: str, prompt: str, call: ToolCall, result: str) -> Reply:
|
||||
tool_use: Final[ToolUseBlockParam] = {
|
||||
"type": "tool_use",
|
||||
"id": call.call_id,
|
||||
"name": call.name,
|
||||
"input": call.parsed().model_dump(),
|
||||
}
|
||||
tool_result: Final[ToolResultBlockParam] = {
|
||||
"type": "tool_result",
|
||||
"tool_use_id": call.call_id,
|
||||
"content": result,
|
||||
}
|
||||
history: Final[tuple[MessageParam, ...]] = (
|
||||
{"role": "user", "content": prompt},
|
||||
{"role": "assistant", "content": [tool_use]},
|
||||
{"role": "user", "content": [tool_result]},
|
||||
)
|
||||
return self._turn(key, model, history, "offered")
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ResponsesSurface:
|
||||
sdk: SdkClients
|
||||
name: SurfaceName = "responses"
|
||||
|
||||
def _turn(self, key: str, model: str, history: ResponseInputParam, tool: ToolMode) -> Reply:
|
||||
raw: Final = self.sdk.openai(key).responses.with_raw_response.create(
|
||||
model=model,
|
||||
input=history,
|
||||
instructions=INSTRUCTIONS,
|
||||
max_output_tokens=MAX_OUTPUT_TOKENS,
|
||||
tools=openai.omit if tool == "none" else [_responses_tool()],
|
||||
tool_choice=_responses_tool_choice() if tool == "forced" else openai.omit,
|
||||
parallel_tool_calls=False if tool == "forced" else openai.omit,
|
||||
extra_body=NO_PROXY_CACHE,
|
||||
)
|
||||
response: Final = raw.parse()
|
||||
return Reply(
|
||||
response_id=response.id,
|
||||
text=response.output_text,
|
||||
tool_calls=tuple(
|
||||
ToolCall(call_id=item.call_id, name=item.name, arguments=item.arguments)
|
||||
for item in response.output
|
||||
if isinstance(item, ResponseFunctionToolCall)
|
||||
),
|
||||
usage=None if response.usage is None else _usage(response.usage.input_tokens, response.usage.output_tokens),
|
||||
call_id_header=response_header(raw.headers, "x-litellm-call-id"),
|
||||
cost_header=response_header(raw.headers, "x-litellm-response-cost"),
|
||||
)
|
||||
|
||||
def reply(self, key: str, model: str, prompt: str, *, with_tool: bool = False) -> Reply:
|
||||
return self._turn(key, model, [{"role": "user", "content": prompt}], "forced" if with_tool else "none")
|
||||
|
||||
def stream(self, key: str, model: str, prompt: str) -> StreamedReply:
|
||||
events: Final[tuple[ResponseStreamEvent, ...]] = tuple(
|
||||
self.sdk.openai(key).responses.create(
|
||||
model=model,
|
||||
input=prompt,
|
||||
instructions=INSTRUCTIONS,
|
||||
max_output_tokens=MAX_OUTPUT_TOKENS,
|
||||
stream=True,
|
||||
extra_body=NO_PROXY_CACHE,
|
||||
)
|
||||
)
|
||||
return StreamedReply(
|
||||
text="".join(event.delta for event in events if event.type == "response.output_text.delta"),
|
||||
finished=bool(events) and events[-1].type == "response.completed",
|
||||
usage_reported=any(
|
||||
event.type == "response.completed" and event.response.usage is not None for event in events
|
||||
),
|
||||
event_count=len(events),
|
||||
)
|
||||
|
||||
def reply_to_tool_result(self, key: str, model: str, prompt: str, call: ToolCall, result: str) -> Reply:
|
||||
function_call: Final[ResponseFunctionToolCallParam] = {
|
||||
"type": "function_call",
|
||||
"call_id": call.call_id,
|
||||
"name": call.name,
|
||||
"arguments": call.arguments,
|
||||
}
|
||||
output: Final[FunctionCallOutput] = {
|
||||
"type": "function_call_output",
|
||||
"call_id": call.call_id,
|
||||
"output": result,
|
||||
}
|
||||
return self._turn(key, model, [{"role": "user", "content": prompt}, function_call, output], "offered")
|
||||
|
||||
|
||||
def build_surfaces(sdk: SdkClients) -> Mapping[SurfaceName, Surface]:
|
||||
return MappingProxyType[SurfaceName, Surface](
|
||||
{
|
||||
"chat_completions": ChatCompletionsSurface(sdk),
|
||||
"messages": MessagesSurface(sdk),
|
||||
"responses": ResponsesSurface(sdk),
|
||||
}
|
||||
)
|
||||
163
tests/e2e/llm_translation/test_conversational_matrix_e2e.py
Normal file
163
tests/e2e/llm_translation/test_conversational_matrix_e2e.py
Normal file
|
|
@ -0,0 +1,163 @@
|
|||
"""The same conversation contract on every (endpoint, deployment, auth) cell.
|
||||
A deployment is one provider model (openai/gpt-4o-mini, anthropic/claude-haiku-4-5, ...).
|
||||
|
||||
/chat/completions, /v1/messages and /v1/responses each have their own
|
||||
translation code in the proxy, so a bug fixed on one surface tends to survive
|
||||
on the others. Every test here runs once per cell in `CELLS`
|
||||
(conversational_matrix.py), so a change to a shared helper is proven against all
|
||||
surfaces and providers at once, and a new model or provider is one row in `DEPLOYMENTS`.
|
||||
|
||||
Edge-wired: OpenAI and Anthropic traffic goes through the provider edge in
|
||||
record and replay, so the whole matrix replays with zero provider calls.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Iterator, Mapping
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
from e2e_config import unique_marker
|
||||
from lifecycle import ResourceManager
|
||||
from llm_translation.conversational_matrix import (
|
||||
GREETING_PROMPT,
|
||||
WEATHER_PROMPT,
|
||||
WEATHER_REPORT,
|
||||
WEATHER_TOOL_NAME,
|
||||
Cell,
|
||||
Deployments,
|
||||
Surface,
|
||||
SurfaceName,
|
||||
ToolCall,
|
||||
build_surfaces,
|
||||
cells_covering,
|
||||
register_deployments,
|
||||
)
|
||||
from llm_translation.sdk_clients import SdkClients
|
||||
from models import SpendLogRow
|
||||
from proxy_client import ProxyClient
|
||||
|
||||
pytestmark = [pytest.mark.e2e, pytest.mark.replayable]
|
||||
|
||||
|
||||
def _approx_equal(actual: float, expected: float) -> bool:
|
||||
return abs(actual - expected) <= max(1e-9, abs(expected) * 1e-2)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def deployments(proxy: ProxyClient) -> Iterator[Deployments]:
|
||||
yield from register_deployments(proxy)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def surfaces(sdk: SdkClients) -> Mapping[SurfaceName, Surface]:
|
||||
return build_surfaces(sdk)
|
||||
|
||||
|
||||
def _weather_call(surface: Surface, key: str, model: str) -> ToolCall:
|
||||
first: Final = surface.reply(key, model, WEATHER_PROMPT, with_tool=True)
|
||||
assert len(first.tool_calls) == 1, (
|
||||
f"{surface.name} forced tool_choice={WEATHER_TOOL_NAME} with parallel calls off, "
|
||||
f"got {len(first.tool_calls)} tool call(s): {first.tool_calls} text={first.text!r}"
|
||||
)
|
||||
call: Final = first.tool_calls[0]
|
||||
assert call.name == WEATHER_TOOL_NAME, f"{surface.name} called {call.name!r}, not the forced {WEATHER_TOOL_NAME!r}"
|
||||
assert call.call_id, f"{surface.name} tool call has no id, so the caller cannot answer it: {call}"
|
||||
assert "paris" in call.parsed().location.lower(), f"{surface.name} tool arguments lost the location: {call}"
|
||||
return call
|
||||
|
||||
|
||||
class TestConversationalMatrix:
|
||||
@pytest.mark.parametrize("cell", cells_covering("basic", "nonstream", "works"))
|
||||
def test_reply_carries_assistant_text_and_usage(
|
||||
self,
|
||||
cell: Cell,
|
||||
deployments: Deployments,
|
||||
surfaces: Mapping[SurfaceName, Surface],
|
||||
resources: ResourceManager,
|
||||
) -> None:
|
||||
surface: Final = surfaces[cell.surface]
|
||||
reply: Final = surface.reply(resources.key(), deployments.alias(cell), GREETING_PROMPT)
|
||||
|
||||
assert reply.response_id, f"{cell.id}: response has no id"
|
||||
assert reply.text.strip(), f"{cell.id}: response carried no assistant text"
|
||||
assert reply.usage is not None and reply.usage.input_tokens > 0 and reply.usage.output_tokens > 0, (
|
||||
f"{cell.id}: usage missing or zero, so the caller cannot account for this call: {reply.usage}"
|
||||
)
|
||||
assert reply.call_id_header, f"{cell.id}: x-litellm-call-id header missing"
|
||||
|
||||
@pytest.mark.parametrize("cell", cells_covering("basic", "stream", "works"))
|
||||
def test_stream_delivers_text_usage_and_a_terminal_event(
|
||||
self,
|
||||
cell: Cell,
|
||||
deployments: Deployments,
|
||||
surfaces: Mapping[SurfaceName, Surface],
|
||||
resources: ResourceManager,
|
||||
) -> None:
|
||||
surface: Final = surfaces[cell.surface]
|
||||
streamed: Final = surface.stream(resources.key(), deployments.alias(cell), GREETING_PROMPT)
|
||||
|
||||
assert streamed.event_count > 1, f"{cell.id}: stream arrived as {streamed.event_count} event(s), not a stream"
|
||||
assert streamed.text.strip(), f"{cell.id}: stream carried no text deltas"
|
||||
assert streamed.finished, f"{cell.id}: stream never sent its terminal event"
|
||||
assert streamed.usage_reported, f"{cell.id}: stream never reported usage"
|
||||
|
||||
@pytest.mark.parametrize("cell", cells_covering("basic", "nonstream", "cost_logged"))
|
||||
def test_cost_header_matches_the_spend_log(
|
||||
self,
|
||||
cell: Cell,
|
||||
deployments: Deployments,
|
||||
surfaces: Mapping[SurfaceName, Surface],
|
||||
resources: ResourceManager,
|
||||
proxy: ProxyClient,
|
||||
) -> None:
|
||||
key: Final = resources.key()
|
||||
surface: Final = surfaces[cell.surface]
|
||||
reply: Final = surface.reply(key, deployments.alias(cell), f"{GREETING_PROMPT} {unique_marker()}")
|
||||
|
||||
assert reply.cost_header is not None, f"{cell.id}: x-litellm-response-cost header missing"
|
||||
header_cost: Final = float(reply.cost_header)
|
||||
assert header_cost > 0, f"{cell.id}: x-litellm-response-cost is not positive: {header_cost}"
|
||||
|
||||
def _priced(rows: list[SpendLogRow]) -> bool:
|
||||
return any(row.spend is not None and row.spend > 0 for row in rows)
|
||||
|
||||
rows: Final = proxy.poll_logs_for_key(key, predicate=_priced)
|
||||
priced: Final = tuple(row for row in rows if row.spend is not None and row.spend > 0)
|
||||
assert len(priced) == 1, f"{cell.id}: expected exactly one priced spend row for a fresh key, got {rows}"
|
||||
row: Final = priced[0]
|
||||
assert (row.prompt_tokens or 0) > 0 and (row.completion_tokens or 0) > 0, (
|
||||
f"{cell.id}: spend row has no token counts, so the cost is not real usage: {row}"
|
||||
)
|
||||
assert row.spend is not None and _approx_equal(row.spend, header_cost), (
|
||||
f"{cell.id}: logged spend {row.spend} disagrees with x-litellm-response-cost {header_cost}"
|
||||
)
|
||||
assert row.model and cell.deployment.backend.endswith(row.model), (
|
||||
f"{cell.id}: spend row logged model {row.model!r}, not the deployment's {cell.deployment.backend!r}"
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize("cell", cells_covering("tool_use", "nonstream", "works"))
|
||||
def test_tool_call_is_returned_named_and_addressable(
|
||||
self,
|
||||
cell: Cell,
|
||||
deployments: Deployments,
|
||||
surfaces: Mapping[SurfaceName, Surface],
|
||||
resources: ResourceManager,
|
||||
) -> None:
|
||||
_weather_call(surfaces[cell.surface], resources.key(), deployments.alias(cell))
|
||||
|
||||
@pytest.mark.parametrize("cell", cells_covering("multi_turn", "nonstream", "works"))
|
||||
def test_tool_result_round_trip_reaches_the_model(
|
||||
self,
|
||||
cell: Cell,
|
||||
deployments: Deployments,
|
||||
surfaces: Mapping[SurfaceName, Surface],
|
||||
resources: ResourceManager,
|
||||
) -> None:
|
||||
key: Final = resources.key()
|
||||
model: Final = deployments.alias(cell)
|
||||
surface: Final = surfaces[cell.surface]
|
||||
call: Final = _weather_call(surface, key, model)
|
||||
|
||||
answer: Final = surface.reply_to_tool_result(key, model, WEATHER_PROMPT, call, WEATHER_REPORT)
|
||||
assert "22" in answer.text, f"{cell.id}: the model never saw the tool result: {answer.text!r}"
|
||||
|
|
@ -818,6 +818,35 @@ async def test_proxy_startup_event_prunes_dead_workers_live_gauges(tmp_path):
|
|||
assert counter.exists()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(("disable_model_info_refresh", "job_scheduled"), [(True, False), (False, True)])
|
||||
async def test_proxy_startup_event_honors_disable_model_info_refresh(
|
||||
disable_model_info_refresh: bool, job_scheduled: bool
|
||||
) -> None:
|
||||
"""``general_settings.disable_model_info_refresh: true`` keeps the proxy from polling every
|
||||
OpenAI-compatible deployment's ``/v1/models`` in the background, so a proxy fronting a replay
|
||||
fixture (or a metered upstream) makes only the calls its clients asked for."""
|
||||
scheduler = AsyncIOScheduler()
|
||||
clean_env = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")} | {
|
||||
"LITELLM_DANGEROUSLY_PERMIT_WEAK_OR_UNSET_MASTER_KEY": "true"
|
||||
}
|
||||
with (
|
||||
patch.dict(os.environ, clean_env, clear=True),
|
||||
patch.object(ps, "scheduler", scheduler),
|
||||
patch.dict(ps.general_settings, {"disable_model_info_refresh": disable_model_info_refresh}),
|
||||
):
|
||||
try:
|
||||
async with proxy_startup_event(app=None):
|
||||
job = scheduler.get_job("refresh_model_info")
|
||||
finally:
|
||||
if scheduler.running:
|
||||
scheduler.shutdown(wait=False)
|
||||
|
||||
assert (job is not None) is job_scheduled, (
|
||||
f"disable_model_info_refresh={disable_model_info_refresh} but refresh_model_info job is {job}"
|
||||
)
|
||||
|
||||
|
||||
def test_otel_global_provider_published_after_callback_init():
|
||||
"""The OTel V2 global-provider publish must run after callback
|
||||
initialization in ``proxy_startup_event``.
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue