840 lines
35 KiB
Python
840 lines
35 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import inspect
|
|
import json
|
|
import logging
|
|
import os
|
|
import time
|
|
from collections.abc import AsyncGenerator
|
|
from typing import Any
|
|
|
|
from realtime_voice_service.providers.base import BaseLLM
|
|
from realtime_voice_service.providers.base import LLMStreamEvent
|
|
|
|
|
|
LOGGER = logging.getLogger("uvicorn.error")
|
|
|
|
|
|
def _preview_text(text: str, *, limit: int = 160) -> str:
|
|
normalized = " ".join(str(text or "").split())
|
|
if len(normalized) <= limit:
|
|
return normalized
|
|
return f"{normalized[:limit]}..."
|
|
|
|
|
|
def _timeout_seconds() -> float:
|
|
raw = os.getenv("OPENAI_TIMEOUT_SECONDS")
|
|
if raw is None:
|
|
return 30.0
|
|
try:
|
|
return max(float(raw.strip()), 1.0)
|
|
except ValueError:
|
|
return 30.0
|
|
|
|
|
|
def _max_context_messages() -> int:
|
|
raw = os.getenv("OPENAI_LLM_MAX_CONTEXT_MESSAGES")
|
|
if raw is None:
|
|
return 8
|
|
try:
|
|
return max(int(raw.strip()), 0)
|
|
except ValueError:
|
|
return 8
|
|
|
|
|
|
def _read_bool_env(name: str, default: bool) -> bool:
|
|
raw = os.getenv(name)
|
|
if raw is None:
|
|
return default
|
|
return str(raw).strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def _read_int_env(name: str, default: int) -> int:
|
|
raw = os.getenv(name)
|
|
if raw is None:
|
|
return default
|
|
try:
|
|
return int(str(raw).strip())
|
|
except ValueError:
|
|
return default
|
|
|
|
|
|
class OpenAILLM(BaseLLM):
|
|
def __init__(
|
|
self,
|
|
*,
|
|
api_key: str | None = None,
|
|
model: str | None = None,
|
|
base_url: str | None = None,
|
|
system_prompt: str | None = None,
|
|
timeout_seconds: float | None = None,
|
|
temperature: float = 0.7,
|
|
max_retries: int = 2,
|
|
max_context_messages: int | None = None,
|
|
reasoning_effort: str | None = None,
|
|
enable_tools: bool | None = None,
|
|
max_tool_roundtrips: int | None = None,
|
|
serper_api_key: str | None = None,
|
|
serper_api_base: str | None = None,
|
|
) -> None:
|
|
self._api_key = str(api_key if api_key is not None else os.getenv("OPENAI_API_KEY", "")).strip()
|
|
self._model = str(model or os.getenv("OPENAI_LLM_MODEL", "gpt-4o-mini")).strip() or "gpt-4o-mini"
|
|
self._base_url = str(base_url or os.getenv("OPENAI_BASE_URL", "")).strip() or None
|
|
self._system_prompt = str(
|
|
system_prompt
|
|
if system_prompt is not None
|
|
else os.getenv(
|
|
"OPENAI_LLM_SYSTEM_PROMPT",
|
|
"Ты — дружелюбный, живой и эмпатичный голосовой ИИ-ассистент. Отвечай кратко, как в реальном диалоге. Используй разговорный стиль. Чтобы синтезатор речи (TTS) читал аббревиатуры и английские термины без акцента, пиши их русскими буквами так, как они произносятся (например, 'ай-ти' вместо 'IT', 'би-ту-би' вместо 'B2B', 'си-эр-эм' вместо 'CRM').",
|
|
)
|
|
).strip()
|
|
self._timeout_seconds = max(float(timeout_seconds if timeout_seconds is not None else _timeout_seconds()), 1.0)
|
|
self._temperature = max(min(float(temperature), 2.0), 0.0)
|
|
self._reasoning_effort = str(
|
|
reasoning_effort if reasoning_effort is not None else os.getenv("OPENAI_LLM_REASONING_EFFORT", "")
|
|
).strip().lower()
|
|
self._max_retries = max(int(max_retries), 0)
|
|
self._max_context_messages = max(
|
|
int(max_context_messages if max_context_messages is not None else _max_context_messages()),
|
|
0,
|
|
)
|
|
self._enable_tools = (
|
|
_read_bool_env("OPENAI_LLM_ENABLE_TOOLS", True)
|
|
if enable_tools is None
|
|
else bool(enable_tools)
|
|
)
|
|
self._max_tool_roundtrips = max(
|
|
int(max_tool_roundtrips if max_tool_roundtrips is not None else _read_int_env("OPENAI_LLM_MAX_TOOL_ROUNDTRIPS", 2)),
|
|
0,
|
|
)
|
|
self._serper_api_key = str(
|
|
serper_api_key if serper_api_key is not None else os.getenv("SERPER_API_KEY", "")
|
|
).strip()
|
|
self._serper_api_base = (
|
|
str(serper_api_base or os.getenv("SERPER_API_BASE", "https://google.serper.dev")).strip().rstrip("/")
|
|
or "https://google.serper.dev"
|
|
)
|
|
self._client: Any | None = None
|
|
self._openai_module: Any | None = None
|
|
self._serper_session = None
|
|
self._serper_session_lock = asyncio.Lock()
|
|
LOGGER.info(
|
|
"OpenAI LLM config: model=%s base_url=%s timeout=%s max_context_messages=%s "
|
|
"reasoning_effort=%s tools_enabled=%s serper_configured=%s max_tool_roundtrips=%s",
|
|
self._model,
|
|
self._base_url or "default",
|
|
self._timeout_seconds,
|
|
self._max_context_messages,
|
|
self._reasoning_effort or "default",
|
|
self._enable_tools,
|
|
bool(self._serper_api_key),
|
|
self._max_tool_roundtrips,
|
|
)
|
|
|
|
async def generate_stream(self, text: str, context: list) -> AsyncGenerator[LLMStreamEvent, None]:
|
|
if not self._api_key:
|
|
raise RuntimeError("OPENAI_API_KEY is required for OpenAI LLM")
|
|
|
|
messages = self._build_messages(text, context)
|
|
turn_started_monotonic = time.perf_counter()
|
|
LOGGER.info(
|
|
"OpenAI LLM turn start: model=%s input_chars=%s context_entries=%s messages=%s "
|
|
"tools_available=%s input_preview=%r",
|
|
self._model,
|
|
len(text),
|
|
len(context),
|
|
len(messages),
|
|
bool(self._build_tools()),
|
|
_preview_text(text),
|
|
)
|
|
for round_index in range(self._max_tool_roundtrips + 1):
|
|
tool_buffers: dict[int, dict[str, str]] = {}
|
|
announced_tool_indexes: set[int] = set()
|
|
text_event_count = 0
|
|
text_char_count = 0
|
|
round_started_monotonic = time.perf_counter()
|
|
async for event in self._stream_completion(
|
|
messages=messages,
|
|
tool_buffers=tool_buffers,
|
|
announced_tool_indexes=announced_tool_indexes,
|
|
):
|
|
if event.type == "text":
|
|
content = str(event.content or "")
|
|
text_event_count += 1
|
|
text_char_count += len(content)
|
|
if text_event_count == 1 or text_event_count % 20 == 0:
|
|
LOGGER.info(
|
|
"OpenAI LLM text stream: round=%s events=%s chars=%s latest=%r",
|
|
round_index,
|
|
text_event_count,
|
|
text_char_count,
|
|
_preview_text(content, limit=80),
|
|
)
|
|
elif event.type == "tool_call_start":
|
|
LOGGER.info(
|
|
"OpenAI LLM tool_call_start: round=%s name=%s tool_call_id=%s",
|
|
round_index,
|
|
event.name,
|
|
event.tool_call_id,
|
|
)
|
|
yield event
|
|
LOGGER.info(
|
|
"OpenAI LLM stream round completed: round=%s text_events=%s text_chars=%s "
|
|
"tool_calls=%s latency_ms=%s",
|
|
round_index,
|
|
text_event_count,
|
|
text_char_count,
|
|
len(tool_buffers),
|
|
int((time.perf_counter() - round_started_monotonic) * 1000.0),
|
|
)
|
|
if not tool_buffers:
|
|
LOGGER.info(
|
|
"OpenAI LLM turn completed: total_latency_ms=%s",
|
|
int((time.perf_counter() - turn_started_monotonic) * 1000.0),
|
|
)
|
|
return
|
|
|
|
assistant_tool_calls = self._finalize_tool_calls(tool_buffers)
|
|
if not assistant_tool_calls:
|
|
LOGGER.warning("OpenAI LLM produced tool buffer without finalized tool calls")
|
|
return
|
|
messages.append(
|
|
{
|
|
"role": "assistant",
|
|
"content": None,
|
|
"tool_calls": assistant_tool_calls,
|
|
}
|
|
)
|
|
tool_messages = await self._execute_tool_calls(assistant_tool_calls)
|
|
messages.extend(tool_messages)
|
|
LOGGER.warning(
|
|
"OpenAI LLM max tool roundtrips reached: max_tool_roundtrips=%s total_latency_ms=%s",
|
|
self._max_tool_roundtrips,
|
|
int((time.perf_counter() - turn_started_monotonic) * 1000.0),
|
|
)
|
|
final_text_event_count = 0
|
|
final_text_char_count = 0
|
|
final_round_started_monotonic = time.perf_counter()
|
|
async for event in self._stream_completion(
|
|
messages=messages,
|
|
tool_buffers={},
|
|
announced_tool_indexes=set(),
|
|
enable_tools=False,
|
|
):
|
|
if event.type == "text":
|
|
content = str(event.content or "")
|
|
final_text_event_count += 1
|
|
final_text_char_count += len(content)
|
|
if final_text_event_count == 1 or final_text_event_count % 20 == 0:
|
|
LOGGER.info(
|
|
"OpenAI LLM final no-tool stream: events=%s chars=%s latest=%r",
|
|
final_text_event_count,
|
|
final_text_char_count,
|
|
_preview_text(content, limit=80),
|
|
)
|
|
yield event
|
|
LOGGER.info(
|
|
"OpenAI LLM final no-tool round completed: text_events=%s text_chars=%s latency_ms=%s total_latency_ms=%s",
|
|
final_text_event_count,
|
|
final_text_char_count,
|
|
int((time.perf_counter() - final_round_started_monotonic) * 1000.0),
|
|
int((time.perf_counter() - turn_started_monotonic) * 1000.0),
|
|
)
|
|
|
|
async def close(self) -> None:
|
|
client = self._client
|
|
self._client = None
|
|
if client is not None and hasattr(client, "close"):
|
|
result = client.close()
|
|
if inspect.isawaitable(result):
|
|
await result
|
|
session = self._serper_session
|
|
self._serper_session = None
|
|
if session is not None and not session.closed:
|
|
await session.close()
|
|
|
|
async def _stream_completion(
|
|
self,
|
|
*,
|
|
messages: list[dict[str, Any]],
|
|
tool_buffers: dict[int, dict[str, str]],
|
|
announced_tool_indexes: set[int],
|
|
enable_tools: bool = True,
|
|
) -> AsyncGenerator[LLMStreamEvent, None]:
|
|
client = self._get_client()
|
|
request_kwargs: dict[str, Any] = {
|
|
"model": self._model,
|
|
"messages": messages,
|
|
"stream": True,
|
|
}
|
|
if self._supports_custom_temperature():
|
|
request_kwargs["temperature"] = self._temperature
|
|
tools = self._build_tools() if enable_tools else None
|
|
reasoning_effort = self._reasoning_effort
|
|
if tools and self._model.lower().startswith("gpt-5.5"):
|
|
reasoning_effort = ""
|
|
if reasoning_effort:
|
|
request_kwargs["reasoning_effort"] = reasoning_effort
|
|
if tools:
|
|
request_kwargs["tools"] = tools
|
|
request_kwargs["tool_choice"] = "auto"
|
|
LOGGER.info(
|
|
"OpenAI LLM stream request: model=%s messages=%s tools=%s temperature=%s reasoning_effort=%s enable_tools=%s",
|
|
self._model,
|
|
len(messages),
|
|
len(tools or []),
|
|
self._temperature if self._supports_custom_temperature() else "default",
|
|
reasoning_effort or "default",
|
|
enable_tools,
|
|
)
|
|
|
|
try:
|
|
stream = await client.chat.completions.create(**request_kwargs)
|
|
async for chunk in stream:
|
|
choices = getattr(chunk, "choices", None) or []
|
|
if not choices:
|
|
continue
|
|
choice = choices[0]
|
|
delta = getattr(choice, "delta", None)
|
|
if delta is None:
|
|
continue
|
|
content = getattr(delta, "content", None)
|
|
if content:
|
|
yield LLMStreamEvent(type="text", content=str(content))
|
|
tool_calls = getattr(delta, "tool_calls", None) or []
|
|
for tool_delta in tool_calls:
|
|
for event in self._consume_tool_delta(
|
|
tool_delta=tool_delta,
|
|
tool_buffers=tool_buffers,
|
|
announced_tool_indexes=announced_tool_indexes,
|
|
):
|
|
yield event
|
|
except Exception as exc: # noqa: BLE001
|
|
openai_module = self._openai_module
|
|
if openai_module is not None and isinstance(exc, getattr(openai_module, "APITimeoutError", ())):
|
|
raise RuntimeError("OpenAI LLM request timed out") from exc
|
|
if openai_module is not None and isinstance(exc, getattr(openai_module, "RateLimitError", ())):
|
|
raise RuntimeError("OpenAI LLM rate limit exceeded") from exc
|
|
if openai_module is not None and isinstance(exc, getattr(openai_module, "APIStatusError", ())):
|
|
status_code = getattr(exc, "status_code", "unknown")
|
|
raise RuntimeError(f"OpenAI LLM returned HTTP {status_code}") from exc
|
|
if openai_module is not None and isinstance(exc, getattr(openai_module, "APIConnectionError", ())):
|
|
raise RuntimeError("OpenAI LLM connection failed") from exc
|
|
raise RuntimeError("OpenAI LLM streaming failed") from exc
|
|
|
|
def _consume_tool_delta(
|
|
self,
|
|
*,
|
|
tool_delta: Any,
|
|
tool_buffers: dict[int, dict[str, str]],
|
|
announced_tool_indexes: set[int],
|
|
) -> list[LLMStreamEvent]:
|
|
index = int(getattr(tool_delta, "index", 0) or 0)
|
|
state = tool_buffers.setdefault(index, {"id": "", "name": "", "arguments": ""})
|
|
tool_id = getattr(tool_delta, "id", None)
|
|
if tool_id:
|
|
state["id"] = str(tool_id)
|
|
function = getattr(tool_delta, "function", None)
|
|
if function is not None:
|
|
function_name = getattr(function, "name", None)
|
|
if function_name:
|
|
state["name"] = str(function_name)
|
|
function_arguments = getattr(function, "arguments", None)
|
|
if function_arguments:
|
|
state["arguments"] += str(function_arguments)
|
|
if state["name"] and index not in announced_tool_indexes:
|
|
announced_tool_indexes.add(index)
|
|
return [
|
|
LLMStreamEvent(
|
|
type="tool_call_start",
|
|
name=state["name"],
|
|
tool_call_id=state["id"] or f"tool-call-{index}",
|
|
)
|
|
]
|
|
return []
|
|
|
|
async def _execute_tool_calls(self, assistant_tool_calls: list[dict[str, Any]]) -> list[dict[str, str]]:
|
|
LOGGER.info("OpenAI LLM executing tool calls: count=%s", len(assistant_tool_calls))
|
|
results = await asyncio.gather(
|
|
*(self._execute_tool_call(tool_call) for tool_call in assistant_tool_calls),
|
|
return_exceptions=False,
|
|
)
|
|
tool_messages: list[dict[str, str]] = []
|
|
for tool_call, tool_result in zip(assistant_tool_calls, results, strict=False):
|
|
tool_messages.append(
|
|
{
|
|
"role": "tool",
|
|
"tool_call_id": str(tool_call["id"]),
|
|
"content": tool_result,
|
|
}
|
|
)
|
|
return tool_messages
|
|
|
|
async def _execute_tool_call(self, tool_call: dict[str, Any]) -> str:
|
|
function = tool_call.get("function") or {}
|
|
name = str(function.get("name") or "").strip().lower()
|
|
raw_arguments = str(function.get("arguments") or "{}")
|
|
try:
|
|
arguments = json.loads(raw_arguments)
|
|
except json.JSONDecodeError:
|
|
arguments = {}
|
|
|
|
started_monotonic = time.perf_counter()
|
|
LOGGER.info(
|
|
"OpenAI LLM tool execution start: name=%s tool_call_id=%s args=%s",
|
|
name,
|
|
tool_call.get("id"),
|
|
raw_arguments[:500],
|
|
)
|
|
if name == "serper":
|
|
result = await self._run_serper_tool(arguments)
|
|
else:
|
|
result = f"Tool `{name}` is not supported by this runtime."
|
|
LOGGER.info(
|
|
"OpenAI LLM tool execution done: name=%s tool_call_id=%s latency_ms=%s result_chars=%s",
|
|
name,
|
|
tool_call.get("id"),
|
|
int((time.perf_counter() - started_monotonic) * 1000.0),
|
|
len(result),
|
|
)
|
|
return result
|
|
|
|
async def _run_serper_tool(self, arguments: dict[str, Any]) -> str:
|
|
if not self._serper_api_key:
|
|
return "Serper API is unavailable: SERPER_API_KEY is not configured."
|
|
query = str(arguments.get("query") or arguments.get("q") or "").strip()
|
|
if not query:
|
|
return "Serper API error: empty search query."
|
|
|
|
try:
|
|
import aiohttp
|
|
except Exception as exc: # noqa: BLE001
|
|
raise RuntimeError("The `aiohttp` package is required for Serper tool execution") from exc
|
|
|
|
session = await self._get_serper_session()
|
|
request_payload = {
|
|
"q": query,
|
|
"gl": str(arguments.get("gl") or os.getenv("SERPER_SEARCH_GL", "kz")).strip(),
|
|
"hl": str(arguments.get("hl") or os.getenv("SERPER_SEARCH_HL", "ru")).strip(),
|
|
"num": max(int(arguments.get("num") or os.getenv("SERPER_SEARCH_NUM", 5)), 1),
|
|
}
|
|
started_monotonic = time.perf_counter()
|
|
LOGGER.info(
|
|
"Serper request start: query=%r gl=%s hl=%s num=%s",
|
|
_preview_text(query),
|
|
request_payload["gl"],
|
|
request_payload["hl"],
|
|
request_payload["num"],
|
|
)
|
|
try:
|
|
async with session.post(
|
|
f"{self._serper_api_base}/search",
|
|
headers={
|
|
"X-API-KEY": self._serper_api_key,
|
|
"Content-Type": "application/json",
|
|
},
|
|
json=request_payload,
|
|
) as response:
|
|
payload_text = await response.text()
|
|
LOGGER.info(
|
|
"Serper response: status=%s latency_ms=%s response_bytes=%s",
|
|
response.status,
|
|
int((time.perf_counter() - started_monotonic) * 1000.0),
|
|
len(payload_text),
|
|
)
|
|
if response.status >= 400:
|
|
return f"Serper API returned HTTP {response.status}: {payload_text[:300]}"
|
|
except (TimeoutError, asyncio.TimeoutError):
|
|
return "Serper API timed out while searching."
|
|
except aiohttp.ClientError as exc:
|
|
return f"Serper API request failed: {exc}"
|
|
|
|
try:
|
|
payload = json.loads(payload_text)
|
|
except json.JSONDecodeError:
|
|
return "Serper API returned invalid JSON."
|
|
summary = self._summarize_serper_payload(query=query, payload=payload)
|
|
LOGGER.info("Serper summary built: chars=%s preview=%r", len(summary), _preview_text(summary))
|
|
return summary
|
|
|
|
async def _get_serper_session(self):
|
|
try:
|
|
import aiohttp
|
|
except Exception as exc: # noqa: BLE001
|
|
raise RuntimeError("The `aiohttp` package is required for Serper tool execution") from exc
|
|
|
|
if self._serper_session is not None and not self._serper_session.closed:
|
|
return self._serper_session
|
|
async with self._serper_session_lock:
|
|
if self._serper_session is not None and not self._serper_session.closed:
|
|
return self._serper_session
|
|
timeout = aiohttp.ClientTimeout(total=self._timeout_seconds)
|
|
connector = aiohttp.TCPConnector(limit=16, ttl_dns_cache=300)
|
|
self._serper_session = aiohttp.ClientSession(timeout=timeout, connector=connector)
|
|
return self._serper_session
|
|
|
|
def _build_tools(self) -> list[dict[str, Any]] | None:
|
|
if not self._enable_tools or not self._serper_api_key:
|
|
return None
|
|
return [
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "serper",
|
|
"description": (
|
|
"Search the public web for recent or external information when the user asks "
|
|
"about current facts, websites, company data, schedules, or anything requiring live search."
|
|
),
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"query": {
|
|
"type": "string",
|
|
"description": "Precise search query to send to Serper.",
|
|
},
|
|
"num": {
|
|
"type": "integer",
|
|
"description": "How many results to fetch, usually 3 to 5.",
|
|
"minimum": 1,
|
|
"maximum": 10,
|
|
},
|
|
"hl": {
|
|
"type": "string",
|
|
"description": "UI language code, for example ru or en.",
|
|
},
|
|
"gl": {
|
|
"type": "string",
|
|
"description": "Country code for result localization, for example kz or us.",
|
|
},
|
|
},
|
|
"required": ["query"],
|
|
"additionalProperties": False,
|
|
},
|
|
},
|
|
}
|
|
]
|
|
|
|
def _get_client(self):
|
|
if self._client is not None:
|
|
return self._client
|
|
try:
|
|
import httpx
|
|
import openai
|
|
from openai import AsyncOpenAI
|
|
except Exception as exc: # noqa: BLE001
|
|
raise RuntimeError("The `openai` package is required for OpenAI LLM") from exc
|
|
|
|
timeout = httpx.Timeout(
|
|
self._timeout_seconds,
|
|
connect=min(self._timeout_seconds, 5.0),
|
|
write=min(self._timeout_seconds, 15.0),
|
|
read=self._timeout_seconds,
|
|
)
|
|
self._openai_module = openai
|
|
self._client = AsyncOpenAI(
|
|
api_key=self._api_key,
|
|
base_url=self._base_url,
|
|
timeout=timeout,
|
|
max_retries=self._max_retries,
|
|
)
|
|
return self._client
|
|
|
|
def _supports_custom_temperature(self) -> bool:
|
|
return not self._model.lower().startswith("gpt-5")
|
|
|
|
def _build_messages(self, text: str, context: list) -> list[dict[str, Any]]:
|
|
messages: list[dict[str, Any]] = []
|
|
if self._system_prompt:
|
|
messages.append({"role": "system", "content": self._system_prompt})
|
|
|
|
context_messages: list[dict[str, Any]] = []
|
|
for entry in context:
|
|
role: str | None = None
|
|
content: str | None = None
|
|
if isinstance(entry, dict):
|
|
role = str(entry.get("role") or "").strip().lower() or None
|
|
content = str(entry.get("content") or "").strip() or None
|
|
elif isinstance(entry, (tuple, list)) and len(entry) >= 2:
|
|
speaker = str(entry[0]).strip().lower()
|
|
role = "assistant" if speaker == "assistant" else "user"
|
|
content = str(entry[1]).strip() or None
|
|
if role and content:
|
|
context_messages.append({"role": role, "content": content})
|
|
|
|
original_context_count = len(context_messages)
|
|
if self._max_context_messages > 0 and len(context_messages) > self._max_context_messages:
|
|
context_messages = context_messages[-self._max_context_messages :]
|
|
LOGGER.info(
|
|
"OpenAI LLM context trimmed: original=%s retained=%s max_context_messages=%s",
|
|
original_context_count,
|
|
len(context_messages),
|
|
self._max_context_messages,
|
|
)
|
|
|
|
messages.extend(context_messages)
|
|
if text.strip():
|
|
if not messages or messages[-1].get("role") != "user" or messages[-1].get("content") != text:
|
|
messages.append({"role": "user", "content": text})
|
|
return messages
|
|
|
|
@staticmethod
|
|
def _finalize_tool_calls(tool_buffers: dict[int, dict[str, str]]) -> list[dict[str, Any]]:
|
|
tool_calls: list[dict[str, Any]] = []
|
|
for index in sorted(tool_buffers):
|
|
state = tool_buffers[index]
|
|
name = str(state.get("name") or "").strip()
|
|
arguments = str(state.get("arguments") or "{}").strip() or "{}"
|
|
if not name:
|
|
continue
|
|
tool_calls.append(
|
|
{
|
|
"id": str(state.get("id") or f"tool-call-{index}"),
|
|
"type": "function",
|
|
"function": {
|
|
"name": name,
|
|
"arguments": arguments,
|
|
},
|
|
}
|
|
)
|
|
return tool_calls
|
|
|
|
@staticmethod
|
|
def _summarize_serper_payload(*, query: str, payload: dict[str, Any]) -> str:
|
|
lines = [f"Search query: {query}"]
|
|
answer_box = payload.get("answerBox")
|
|
if isinstance(answer_box, dict):
|
|
answer_text = str(answer_box.get("answer") or answer_box.get("snippet") or "").strip()
|
|
if answer_text:
|
|
lines.append(f"Answer box: {answer_text}")
|
|
|
|
knowledge_graph = payload.get("knowledgeGraph")
|
|
if isinstance(knowledge_graph, dict):
|
|
title = str(knowledge_graph.get("title") or "").strip()
|
|
description = str(knowledge_graph.get("description") or "").strip()
|
|
if title or description:
|
|
lines.append(f"Knowledge graph: {title} {description}".strip())
|
|
|
|
organic = payload.get("organic")
|
|
if isinstance(organic, list):
|
|
for index, item in enumerate(organic[:5], start=1):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
title = str(item.get("title") or "").strip()
|
|
snippet = str(item.get("snippet") or "").strip()
|
|
link = str(item.get("link") or "").strip()
|
|
if title or snippet or link:
|
|
lines.append(f"{index}. {title} | {snippet} | {link}".strip())
|
|
if len(lines) == 1:
|
|
lines.append("No useful search results were returned.")
|
|
return "\n".join(lines)
|
|
|
|
|
|
|
|
class OllamaLLM(BaseLLM):
|
|
def __init__(
|
|
self,
|
|
*,
|
|
model: str | None = None,
|
|
base_url: str | None = None,
|
|
system_prompt: str | None = None,
|
|
timeout_seconds: float | None = None,
|
|
temperature: float | None = None,
|
|
max_context_messages: int | None = None,
|
|
) -> None:
|
|
self._model = str(model or os.getenv("OLLAMA_LLM_MODEL", "qwen2.5:1.5b")).strip() or "qwen2.5:1.5b"
|
|
self._base_url = (
|
|
str(base_url or os.getenv("OLLAMA_BASE_URL", "http://host.docker.internal:11434")).strip().rstrip("/")
|
|
or "http://host.docker.internal:11434"
|
|
)
|
|
self._system_prompt = str(
|
|
system_prompt
|
|
if system_prompt is not None
|
|
else os.getenv(
|
|
"OLLAMA_LLM_SYSTEM_PROMPT",
|
|
os.getenv(
|
|
"OPENAI_LLM_SYSTEM_PROMPT",
|
|
"Ты — дружелюбный, живой и эмпатичный голосовой ИИ-ассистент. Отвечай кратко, как в реальном диалоге. Используй разговорный стиль. Чтобы синтезатор речи (TTS) читал аббревиатуры и английские термины без акцента, пиши их русскими буквами так, как они произносятся (например, 'ай-ти' вместо 'IT', 'би-ту-би' вместо 'B2B', 'си-эр-эм' вместо 'CRM').",
|
|
),
|
|
)
|
|
).strip()
|
|
self._timeout_seconds = max(
|
|
float(timeout_seconds if timeout_seconds is not None else self._read_float_env("OLLAMA_TIMEOUT_SECONDS", _timeout_seconds())),
|
|
1.0,
|
|
)
|
|
self._temperature = max(
|
|
min(float(temperature if temperature is not None else self._read_float_env("OLLAMA_LLM_TEMPERATURE", 0.7)), 2.0),
|
|
0.0,
|
|
)
|
|
self._max_context_messages = max(
|
|
int(
|
|
max_context_messages
|
|
if max_context_messages is not None
|
|
else self._read_int_env("OLLAMA_LLM_MAX_CONTEXT_MESSAGES", _max_context_messages())
|
|
),
|
|
0,
|
|
)
|
|
self._num_predict = max(self._read_int_env("OLLAMA_LLM_NUM_PREDICT", 64), 0)
|
|
self._num_ctx = max(self._read_int_env("OLLAMA_LLM_NUM_CTX", 1024), 0)
|
|
self._client: Any | None = None
|
|
LOGGER.info(
|
|
"Ollama LLM config: model=%s base_url=%s timeout=%s max_context_messages=%s "
|
|
"temperature=%s num_predict=%s num_ctx=%s",
|
|
self._model,
|
|
self._base_url,
|
|
self._timeout_seconds,
|
|
self._max_context_messages,
|
|
self._temperature,
|
|
self._num_predict or "default",
|
|
self._num_ctx or "default",
|
|
)
|
|
|
|
async def generate_stream(self, text: str, context: list) -> AsyncGenerator[LLMStreamEvent, None]:
|
|
messages = self._build_messages(text, context)
|
|
options: dict[str, Any] = {"temperature": self._temperature}
|
|
if self._num_predict > 0:
|
|
options["num_predict"] = self._num_predict
|
|
if self._num_ctx > 0:
|
|
options["num_ctx"] = self._num_ctx
|
|
payload: dict[str, Any] = {
|
|
"model": self._model,
|
|
"messages": messages,
|
|
"stream": True,
|
|
"options": options,
|
|
}
|
|
started_monotonic = time.perf_counter()
|
|
text_event_count = 0
|
|
text_char_count = 0
|
|
LOGGER.info(
|
|
"Ollama LLM turn start: model=%s input_chars=%s context_entries=%s messages=%s input_preview=%r",
|
|
self._model,
|
|
len(text),
|
|
len(context),
|
|
len(messages),
|
|
_preview_text(text),
|
|
)
|
|
try:
|
|
client = self._get_client()
|
|
async with client.stream("POST", f"{self._base_url}/api/chat", json=payload) as response:
|
|
if response.status_code >= 400:
|
|
body = (await response.aread()).decode("utf-8", "replace")
|
|
raise RuntimeError(f"Ollama LLM returned HTTP {response.status_code}: {body[:300]}")
|
|
async for line in response.aiter_lines():
|
|
if not line:
|
|
continue
|
|
try:
|
|
chunk = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
LOGGER.warning("Ollama LLM ignored invalid stream line: %r", line[:200])
|
|
continue
|
|
error = str(chunk.get("error") or "").strip()
|
|
if error:
|
|
raise RuntimeError(f"Ollama LLM error: {error}")
|
|
message = chunk.get("message")
|
|
content = ""
|
|
if isinstance(message, dict):
|
|
content = str(message.get("content") or "")
|
|
if content:
|
|
text_event_count += 1
|
|
text_char_count += len(content)
|
|
if text_event_count == 1 or text_event_count % 20 == 0:
|
|
LOGGER.info(
|
|
"Ollama LLM text stream: events=%s chars=%s latest=%r",
|
|
text_event_count,
|
|
text_char_count,
|
|
_preview_text(content, limit=80),
|
|
)
|
|
yield LLMStreamEvent(type="text", content=content)
|
|
if bool(chunk.get("done")):
|
|
break
|
|
except (TimeoutError, asyncio.TimeoutError) as exc:
|
|
raise RuntimeError("Ollama LLM request timed out") from exc
|
|
except RuntimeError:
|
|
raise
|
|
except Exception as exc: # noqa: BLE001
|
|
raise RuntimeError("Ollama LLM streaming failed") from exc
|
|
finally:
|
|
LOGGER.info(
|
|
"Ollama LLM turn completed: text_events=%s text_chars=%s total_latency_ms=%s",
|
|
text_event_count,
|
|
text_char_count,
|
|
int((time.perf_counter() - started_monotonic) * 1000.0),
|
|
)
|
|
|
|
async def close(self) -> None:
|
|
client = self._client
|
|
self._client = None
|
|
if client is not None:
|
|
await client.aclose()
|
|
|
|
def _get_client(self):
|
|
if self._client is not None:
|
|
return self._client
|
|
try:
|
|
import httpx
|
|
except Exception as exc: # noqa: BLE001
|
|
raise RuntimeError("The `httpx` package is required for Ollama LLM") from exc
|
|
timeout = httpx.Timeout(
|
|
self._timeout_seconds,
|
|
connect=min(self._timeout_seconds, 3.0),
|
|
write=min(self._timeout_seconds, 10.0),
|
|
read=self._timeout_seconds,
|
|
)
|
|
self._client = httpx.AsyncClient(timeout=timeout)
|
|
return self._client
|
|
|
|
def _build_messages(self, text: str, context: list) -> list[dict[str, Any]]:
|
|
messages: list[dict[str, Any]] = []
|
|
if self._system_prompt:
|
|
messages.append({"role": "system", "content": self._system_prompt})
|
|
|
|
context_messages: list[dict[str, Any]] = []
|
|
for entry in context:
|
|
role: str | None = None
|
|
content: str | None = None
|
|
if isinstance(entry, dict):
|
|
role = str(entry.get("role") or "").strip().lower() or None
|
|
content = str(entry.get("content") or "").strip() or None
|
|
elif isinstance(entry, (tuple, list)) and len(entry) >= 2:
|
|
speaker = str(entry[0]).strip().lower()
|
|
role = "assistant" if speaker == "assistant" else "user"
|
|
content = str(entry[1]).strip() or None
|
|
if role and content:
|
|
context_messages.append({"role": role, "content": content})
|
|
|
|
original_context_count = len(context_messages)
|
|
if self._max_context_messages > 0 and len(context_messages) > self._max_context_messages:
|
|
context_messages = context_messages[-self._max_context_messages :]
|
|
LOGGER.info(
|
|
"Ollama LLM context trimmed: original=%s retained=%s max_context_messages=%s",
|
|
original_context_count,
|
|
len(context_messages),
|
|
self._max_context_messages,
|
|
)
|
|
|
|
messages.extend(context_messages)
|
|
if text.strip():
|
|
if not messages or messages[-1].get("role") != "user" or messages[-1].get("content") != text:
|
|
messages.append({"role": "user", "content": text})
|
|
return messages
|
|
|
|
@staticmethod
|
|
def _read_float_env(name: str, default: float) -> float:
|
|
raw = os.getenv(name)
|
|
if raw is None:
|
|
return default
|
|
try:
|
|
return float(str(raw).strip())
|
|
except ValueError:
|
|
return default
|
|
|
|
@staticmethod
|
|
def _read_int_env(name: str, default: int) -> int:
|
|
raw = os.getenv(name)
|
|
if raw is None:
|
|
return default
|
|
try:
|
|
return int(str(raw).strip())
|
|
except ValueError:
|
|
return default
|