wehub-resource-sync
554 lines
20 KiB
Python
554 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
from dataclasses import dataclass
|
|
from typing import Any, Literal, cast
|
|
|
|
import httpx
|
|
import openai
|
|
from openai.types.chat import (
|
|
ChatCompletionChunk,
|
|
ChatCompletionMessageParam,
|
|
ChatCompletionPredictionContentParam,
|
|
ChatCompletionToolChoiceOptionParam,
|
|
ChatCompletionToolParam,
|
|
completion_create_params,
|
|
)
|
|
from openai.types.chat.chat_completion_chunk import Choice
|
|
from openai.types.shared.reasoning_effort import ReasoningEffort
|
|
from openai.types.shared_params import Metadata
|
|
from typing_extensions import TypedDict
|
|
|
|
from .. import llm
|
|
from .._exceptions import APIConnectionError, APIStatusError, APITimeoutError
|
|
from ..llm import ToolChoice, utils as llm_utils
|
|
from ..llm.chat_context import ChatContext
|
|
from ..llm.tool_context import Tool
|
|
from ..log import logger
|
|
from ..types import DEFAULT_API_CONNECT_OPTIONS, NOT_GIVEN, APIConnectOptions, NotGivenOr
|
|
from ..utils import is_given
|
|
from ._utils import (
|
|
HEADER_INFERENCE_PRIORITY,
|
|
HEADER_INFERENCE_PROVIDER,
|
|
create_access_token,
|
|
get_default_inference_url,
|
|
get_inference_headers,
|
|
)
|
|
|
|
lk_oai_debug = int(os.getenv("LK_OPENAI_DEBUG", 0))
|
|
|
|
# Reasoning models don't support sampling parameters.
|
|
# See: https://platform.openai.com/docs/guides/reasoning
|
|
_REASONING_UNSUPPORTED_PARAMS: set[str] = {
|
|
"temperature",
|
|
"top_p",
|
|
"presence_penalty",
|
|
"frequency_penalty",
|
|
"logit_bias",
|
|
"logprobs",
|
|
"top_logprobs",
|
|
"n",
|
|
}
|
|
|
|
# xAI reasoning models only restrict presence_penalty, frequency_penalty, stop.
|
|
# They still support temperature and top_p.
|
|
_XAI_REASONING_UNSUPPORTED_PARAMS: set[str] = {
|
|
"presence_penalty",
|
|
"frequency_penalty",
|
|
"stop",
|
|
}
|
|
|
|
# Model prefix -> set of param names that should be dropped
|
|
_UNSUPPORTED_PARAMS: dict[str, set[str]] = {
|
|
"o1": _REASONING_UNSUPPORTED_PARAMS,
|
|
"o3": _REASONING_UNSUPPORTED_PARAMS,
|
|
"o4": _REASONING_UNSUPPORTED_PARAMS,
|
|
"gpt-5": _REASONING_UNSUPPORTED_PARAMS,
|
|
"grok-4-1-fast-reasoning": _XAI_REASONING_UNSUPPORTED_PARAMS,
|
|
"grok-4.20-0309-reasoning": _XAI_REASONING_UNSUPPORTED_PARAMS,
|
|
"grok-4.20-multi-agent": _XAI_REASONING_UNSUPPORTED_PARAMS,
|
|
}
|
|
|
|
# models that don't support reasoning_effort when function tools are present
|
|
_REASONING_EFFORT_TOOL_INCOMPATIBLE_PREFIXES: set[str] = {"gpt-5.2", "gpt-5.4"}
|
|
|
|
|
|
def drop_unsupported_params(
|
|
model: str, params: dict[str, Any], tools: list[Any] | None = None
|
|
) -> dict[str, Any]:
|
|
"""Remove parameters that are not supported by the given model.
|
|
|
|
Strips any provider prefix (e.g. ``openai/o3-pro`` -> ``o3-pro``) before
|
|
matching against known model prefixes.
|
|
"""
|
|
model_name = model.split("/")[-1] if "/" in model else model
|
|
for prefix, unsupported in _UNSUPPORTED_PARAMS.items():
|
|
if model_name.startswith(prefix):
|
|
params = {k: v for k, v in params.items() if k not in unsupported}
|
|
break
|
|
if tools and any(
|
|
model_name.startswith(p) for p in _REASONING_EFFORT_TOOL_INCOMPATIBLE_PREFIXES
|
|
):
|
|
params = {k: v for k, v in params.items() if k != "reasoning_effort"}
|
|
return params
|
|
|
|
|
|
OpenAIModels = Literal[
|
|
"openai/gpt-4o",
|
|
"openai/gpt-4o-mini",
|
|
"openai/gpt-4.1",
|
|
"openai/gpt-4.1-mini",
|
|
"openai/gpt-4.1-nano",
|
|
"openai/gpt-5",
|
|
"openai/gpt-5-mini",
|
|
"openai/gpt-5-nano",
|
|
"openai/gpt-5.1",
|
|
"openai/gpt-5.1-chat-latest",
|
|
"openai/gpt-5.2",
|
|
"openai/gpt-5.2-chat-latest",
|
|
"openai/gpt-5.3-chat-latest",
|
|
"openai/gpt-5.4",
|
|
"openai/gpt-5.4-mini",
|
|
"openai/gpt-5.4-nano",
|
|
"openai/gpt-5.5",
|
|
"openai/chat-latest",
|
|
"openai/gpt-oss-120b",
|
|
]
|
|
|
|
GoogleModels = Literal[
|
|
"google/gemini-3.1-pro",
|
|
"google/gemini-3-flash",
|
|
"google/gemini-3.1-flash-lite",
|
|
"google/gemini-3.5-flash",
|
|
"google/gemini-2.5-pro",
|
|
"google/gemini-2.5-flash",
|
|
"google/gemini-2.5-flash-lite",
|
|
]
|
|
|
|
KimiModels = Literal[
|
|
"moonshotai/kimi-k2.5",
|
|
"moonshotai/kimi-k2.6",
|
|
]
|
|
|
|
DeepSeekModels = Literal[
|
|
"deepseek-ai/deepseek-v3",
|
|
"deepseek-ai/deepseek-v3.2",
|
|
]
|
|
|
|
ZAIModels = Literal["zai/glm-5.1"]
|
|
|
|
XAIModels = Literal[
|
|
"xai/grok-4-1-fast-non-reasoning",
|
|
"xai/grok-4-1-fast-reasoning",
|
|
"xai/grok-4.20-0309-non-reasoning",
|
|
"xai/grok-4.20-0309-reasoning",
|
|
"xai/grok-4.20-multi-agent-0309",
|
|
]
|
|
|
|
LLMModels = OpenAIModels | GoogleModels | KimiModels | DeepSeekModels | ZAIModels | XAIModels
|
|
|
|
InferenceClass = Literal["priority", "standard"]
|
|
|
|
|
|
class ChatCompletionOptions(TypedDict, total=False):
|
|
frequency_penalty: float | None
|
|
logit_bias: dict[str, int] | None
|
|
logprobs: bool | None
|
|
max_completion_tokens: int | None
|
|
max_tokens: int | None
|
|
metadata: Metadata | None
|
|
modalities: list[Literal["text", "audio"]] | None
|
|
n: int | None
|
|
parallel_tool_calls: bool
|
|
prediction: ChatCompletionPredictionContentParam | None
|
|
presence_penalty: float | None
|
|
prompt_cache_key: str
|
|
prompt_cache_retention: Literal["in_memory", "24h"] | None
|
|
reasoning_effort: ReasoningEffort | None
|
|
safety_identifier: str
|
|
seed: int | None
|
|
service_tier: Literal["auto", "default", "flex", "scale", "priority"] | None
|
|
stop: str | None | list[str] | None
|
|
store: bool | None
|
|
temperature: float | None
|
|
top_logprobs: int | None
|
|
top_p: float | None
|
|
user: str
|
|
verbosity: Literal["low", "medium", "high"] | None
|
|
web_search_options: completion_create_params.WebSearchOptions
|
|
|
|
# livekit-typed arguments
|
|
tool_choice: ToolChoice
|
|
# TODO(theomonnomn): support repsonse format
|
|
# response_format: completion_create_params.ResponseFormat
|
|
|
|
|
|
@dataclass
|
|
class _LLMOptions:
|
|
model: LLMModels | str
|
|
provider: str | None
|
|
base_url: str
|
|
api_key: str
|
|
api_secret: str
|
|
inference_class: InferenceClass | None
|
|
extra_kwargs: ChatCompletionOptions | dict[str, Any]
|
|
|
|
|
|
class LLM(llm.LLM):
|
|
def __init__(
|
|
self,
|
|
model: LLMModels | str,
|
|
*,
|
|
provider: str | None = None,
|
|
base_url: str | None = None,
|
|
api_key: str | None = None,
|
|
api_secret: str | None = None,
|
|
inference_class: InferenceClass | None = None,
|
|
extra_kwargs: ChatCompletionOptions | dict[str, Any] | None = None,
|
|
) -> None:
|
|
super().__init__()
|
|
|
|
lk_base_url = base_url if base_url else get_default_inference_url()
|
|
|
|
lk_api_key = (
|
|
api_key
|
|
if api_key
|
|
else os.getenv("LIVEKIT_INFERENCE_API_KEY", os.getenv("LIVEKIT_API_KEY", ""))
|
|
)
|
|
if not lk_api_key:
|
|
raise ValueError(
|
|
"api_key is required, either as argument or set LIVEKIT_API_KEY environmental variable"
|
|
)
|
|
|
|
lk_api_secret = (
|
|
api_secret
|
|
if api_secret
|
|
else os.getenv("LIVEKIT_INFERENCE_API_SECRET", os.getenv("LIVEKIT_API_SECRET", ""))
|
|
)
|
|
if not lk_api_secret:
|
|
raise ValueError(
|
|
"api_secret is required, either as argument or set LIVEKIT_API_SECRET environmental variable"
|
|
)
|
|
|
|
self._opts = _LLMOptions(
|
|
model=model,
|
|
provider=provider,
|
|
base_url=lk_base_url,
|
|
api_key=lk_api_key,
|
|
api_secret=lk_api_secret,
|
|
inference_class=inference_class,
|
|
extra_kwargs=extra_kwargs or {},
|
|
)
|
|
self._client = openai.AsyncClient(
|
|
api_key=create_access_token(self._opts.api_key, self._opts.api_secret),
|
|
base_url=self._opts.base_url,
|
|
max_retries=0,
|
|
http_client=httpx.AsyncClient(
|
|
timeout=httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0),
|
|
follow_redirects=True,
|
|
limits=httpx.Limits(
|
|
max_connections=50, max_keepalive_connections=50, keepalive_expiry=120
|
|
),
|
|
),
|
|
)
|
|
|
|
async def aclose(self) -> None:
|
|
await self._client.close()
|
|
|
|
@classmethod
|
|
def from_model_string(cls, model: str) -> LLM:
|
|
"""Create a LLM instance from a model string"""
|
|
return cls(model)
|
|
|
|
def update_options(
|
|
self,
|
|
*,
|
|
model: NotGivenOr[LLMModels | str] = NOT_GIVEN,
|
|
extra_kwargs: NotGivenOr[ChatCompletionOptions | dict[str, Any]] = NOT_GIVEN,
|
|
) -> None:
|
|
"""Update LLM configuration options.
|
|
|
|
Each option is read on the next ``chat()`` call, so a swap
|
|
takes effect on the agent's next turn without recreating the
|
|
LLM. ``extra_kwargs`` *replaces* the persistent kwargs dict
|
|
rather than merging — pass ``{}`` to clear it.
|
|
"""
|
|
if is_given(model):
|
|
self._opts.model = model
|
|
if is_given(extra_kwargs):
|
|
self._opts.extra_kwargs = dict(extra_kwargs)
|
|
|
|
@property
|
|
def model(self) -> str:
|
|
"""Get the model name for this LLM instance."""
|
|
return self._opts.model
|
|
|
|
@property
|
|
def provider(self) -> str:
|
|
return "livekit"
|
|
|
|
def chat(
|
|
self,
|
|
*,
|
|
chat_ctx: ChatContext,
|
|
tools: list[Tool] | None = None,
|
|
conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
|
|
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
|
|
tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
|
|
response_format: NotGivenOr[
|
|
completion_create_params.ResponseFormat | type[llm_utils.ResponseFormatT]
|
|
] = NOT_GIVEN,
|
|
inference_class: NotGivenOr[InferenceClass] = NOT_GIVEN,
|
|
extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
|
|
) -> LLMStream:
|
|
extra = {}
|
|
if is_given(extra_kwargs):
|
|
extra.update(extra_kwargs)
|
|
|
|
parallel_tool_calls = (
|
|
parallel_tool_calls
|
|
if is_given(parallel_tool_calls)
|
|
else self._opts.extra_kwargs.get("parallel_tool_calls", NOT_GIVEN)
|
|
)
|
|
if is_given(parallel_tool_calls):
|
|
extra["parallel_tool_calls"] = parallel_tool_calls
|
|
|
|
extra_tool_choice = self._opts.extra_kwargs.get("tool_choice", NOT_GIVEN)
|
|
tool_choice = tool_choice if is_given(tool_choice) else extra_tool_choice
|
|
if is_given(tool_choice):
|
|
oai_tool_choice: ChatCompletionToolChoiceOptionParam
|
|
if isinstance(tool_choice, dict):
|
|
oai_tool_choice = {
|
|
"type": "function",
|
|
"function": {"name": tool_choice["function"]["name"]},
|
|
}
|
|
extra["tool_choice"] = oai_tool_choice
|
|
elif tool_choice in ("auto", "required", "none"):
|
|
oai_tool_choice = tool_choice
|
|
extra["tool_choice"] = oai_tool_choice
|
|
|
|
if is_given(response_format):
|
|
extra["response_format"] = llm_utils.to_openai_response_format(response_format) # type: ignore
|
|
|
|
extra.update(self._opts.extra_kwargs)
|
|
|
|
effective_inference_class = (
|
|
inference_class if is_given(inference_class) else self._opts.inference_class
|
|
)
|
|
|
|
self._client.api_key = create_access_token(self._opts.api_key, self._opts.api_secret)
|
|
return LLMStream(
|
|
self,
|
|
model=self._opts.model,
|
|
provider=self._opts.provider,
|
|
inference_class=effective_inference_class,
|
|
strict_tool_schema=True,
|
|
client=self._client,
|
|
chat_ctx=chat_ctx,
|
|
tools=tools or [],
|
|
conn_options=conn_options,
|
|
extra_kwargs=extra,
|
|
)
|
|
|
|
|
|
class LLMStream(llm.LLMStream):
|
|
def __init__(
|
|
self,
|
|
llm_v: LLM | llm.LLM,
|
|
*,
|
|
model: LLMModels | str,
|
|
provider: str | None = None,
|
|
inference_class: InferenceClass | None = None,
|
|
strict_tool_schema: bool,
|
|
client: openai.AsyncClient,
|
|
chat_ctx: llm.ChatContext,
|
|
tools: list[Tool],
|
|
conn_options: APIConnectOptions,
|
|
extra_kwargs: dict[str, Any],
|
|
provider_fmt: str = "openai", # used internally for chat_ctx format
|
|
) -> None:
|
|
super().__init__(llm_v, chat_ctx=chat_ctx, tools=tools, conn_options=conn_options)
|
|
self._model = model
|
|
self._provider = provider
|
|
self._inference_class = inference_class
|
|
self._provider_fmt = provider_fmt
|
|
self._strict_tool_schema = strict_tool_schema
|
|
self._client = client
|
|
self._llm = llm_v
|
|
self._extra_kwargs = drop_unsupported_params(model, extra_kwargs, tools=tools)
|
|
self._tool_ctx = llm.ToolContext(tools)
|
|
|
|
async def _run(self) -> None:
|
|
# current function call that we're waiting for full completion (args are streamed)
|
|
# (defined inside the _run method to make sure the state is reset for each run/attempt)
|
|
self._oai_stream: openai.AsyncStream[ChatCompletionChunk] | None = None
|
|
self._tool_call_id: str | None = None
|
|
self._fnc_name: str | None = None
|
|
self._fnc_raw_arguments: str | None = None
|
|
self._tool_extra: dict[str, Any] | None = None
|
|
self._tool_index: int | None = None
|
|
retryable = True
|
|
|
|
try:
|
|
chat_ctx, _ = self._chat_ctx.to_provider_format(format=self._provider_fmt)
|
|
tool_schemas = cast(
|
|
list[ChatCompletionToolParam],
|
|
self._tool_ctx.parse_function_tools("openai", strict=self._strict_tool_schema),
|
|
)
|
|
if lk_oai_debug:
|
|
tool_choice = self._extra_kwargs.get("tool_choice", NOT_GIVEN)
|
|
logger.debug(
|
|
"chat.completions.create",
|
|
extra={
|
|
"fnc_ctx": tool_schemas,
|
|
"tool_choice": tool_choice,
|
|
"chat_ctx": chat_ctx,
|
|
},
|
|
)
|
|
if not self._tools:
|
|
# remove tool_choice from extra_kwargs if no tools are provided
|
|
self._extra_kwargs.pop("tool_choice", None)
|
|
|
|
extra_headers = self._extra_kwargs.setdefault("extra_headers", {})
|
|
extra_headers.update(get_inference_headers())
|
|
if self._provider:
|
|
extra_headers[HEADER_INFERENCE_PROVIDER] = self._provider
|
|
if self._inference_class:
|
|
extra_headers[HEADER_INFERENCE_PRIORITY] = self._inference_class
|
|
|
|
self._oai_stream = stream = await self._client.chat.completions.create(
|
|
messages=cast(list[ChatCompletionMessageParam], chat_ctx),
|
|
tools=tool_schemas or openai.omit,
|
|
model=self._model,
|
|
stream_options={"include_usage": True},
|
|
stream=True,
|
|
timeout=httpx.Timeout(self._conn_options.timeout),
|
|
**self._extra_kwargs,
|
|
)
|
|
|
|
thinking = asyncio.Event()
|
|
async with stream:
|
|
async for chunk in stream:
|
|
for choice in chunk.choices:
|
|
chat_chunk = self._parse_choice(chunk.id, choice, thinking)
|
|
if chat_chunk is not None:
|
|
retryable = False
|
|
self._event_ch.send_nowait(chat_chunk)
|
|
|
|
if chunk.usage is not None:
|
|
retryable = False
|
|
tokens_details = chunk.usage.prompt_tokens_details
|
|
cached_tokens = tokens_details.cached_tokens if tokens_details else 0
|
|
usage_chunk = llm.ChatChunk(
|
|
id=chunk.id,
|
|
usage=llm.CompletionUsage(
|
|
completion_tokens=chunk.usage.completion_tokens,
|
|
prompt_tokens=chunk.usage.prompt_tokens,
|
|
prompt_cached_tokens=cached_tokens or 0,
|
|
total_tokens=chunk.usage.total_tokens,
|
|
service_tier=getattr(chunk, "service_tier", None),
|
|
),
|
|
)
|
|
self._event_ch.send_nowait(usage_chunk)
|
|
|
|
except openai.APITimeoutError:
|
|
raise APITimeoutError(retryable=retryable) from None
|
|
except openai.APIStatusError as e:
|
|
raise APIStatusError(
|
|
e.message,
|
|
status_code=e.status_code,
|
|
request_id=e.request_id,
|
|
body=e.body,
|
|
retryable=retryable,
|
|
) from None
|
|
except Exception as e:
|
|
raise APIConnectionError(retryable=retryable) from e
|
|
|
|
def _parse_choice(
|
|
self, id: str, choice: Choice, thinking: asyncio.Event
|
|
) -> llm.ChatChunk | None:
|
|
delta = choice.delta
|
|
|
|
# https://github.com/livekit/agents/issues/688
|
|
# the delta can be None when using Azure OpenAI (content filtering)
|
|
if delta is None:
|
|
return None
|
|
|
|
if delta.tool_calls:
|
|
for tool in delta.tool_calls:
|
|
if not tool.function:
|
|
continue
|
|
|
|
call_chunk = None
|
|
if self._tool_call_id and tool.id and tool.index != self._tool_index:
|
|
call_chunk = llm.ChatChunk(
|
|
id=id,
|
|
delta=llm.ChoiceDelta(
|
|
role="assistant",
|
|
content=delta.content,
|
|
tool_calls=[
|
|
llm.FunctionToolCall(
|
|
arguments=self._fnc_raw_arguments or "",
|
|
name=self._fnc_name or "",
|
|
call_id=self._tool_call_id or "",
|
|
extra=self._tool_extra,
|
|
)
|
|
],
|
|
),
|
|
)
|
|
self._tool_call_id = self._fnc_name = self._fnc_raw_arguments = None
|
|
self._tool_extra = None
|
|
|
|
if tool.function.name:
|
|
self._tool_index = tool.index
|
|
self._tool_call_id = tool.id
|
|
self._fnc_name = tool.function.name
|
|
self._fnc_raw_arguments = tool.function.arguments or ""
|
|
# Extract extra from tool call (e.g., Google thought signatures)
|
|
self._tool_extra = getattr(tool, "extra_content", None)
|
|
elif tool.function.arguments:
|
|
self._fnc_raw_arguments += tool.function.arguments # type: ignore
|
|
|
|
if call_chunk is not None:
|
|
return call_chunk
|
|
|
|
if choice.finish_reason in ("tool_calls", "stop") and self._tool_call_id:
|
|
finish_extra = getattr(delta, "extra_content", None)
|
|
call_chunk = llm.ChatChunk(
|
|
id=id,
|
|
delta=llm.ChoiceDelta(
|
|
role="assistant",
|
|
content=delta.content,
|
|
extra=finish_extra,
|
|
tool_calls=[
|
|
llm.FunctionToolCall(
|
|
arguments=self._fnc_raw_arguments or "",
|
|
name=self._fnc_name or "",
|
|
call_id=self._tool_call_id or "",
|
|
extra=self._tool_extra,
|
|
)
|
|
],
|
|
),
|
|
)
|
|
self._tool_call_id = self._fnc_name = self._fnc_raw_arguments = None
|
|
self._tool_extra = None
|
|
return call_chunk
|
|
|
|
delta.content = llm_utils.strip_thinking_tokens(delta.content, thinking)
|
|
|
|
# Extract extra from delta (e.g., Google thought signatures on text parts)
|
|
delta_extra = getattr(delta, "extra_content", None)
|
|
|
|
if not delta.content and not delta_extra:
|
|
return None
|
|
|
|
return llm.ChatChunk(
|
|
id=id,
|
|
delta=llm.ChoiceDelta(
|
|
content=delta.content,
|
|
role="assistant",
|
|
extra=delta_extra,
|
|
),
|
|
)
|