diff --git a/src/praisonai/praisonai/_entrypoint.py b/src/praisonai/praisonai/_entrypoint.py index 1d4581e122..79c1aad764 100644 --- a/src/praisonai/praisonai/_entrypoint.py +++ b/src/praisonai/praisonai/_entrypoint.py @@ -42,16 +42,67 @@ def _resolve_run_inputs(framework: str | None) -> tuple[Any, list[dict[str, Any] return adapter, _build_config_list() +# Loose kwargs whose CLI dest differs from the intuitive Python name. Mapping +# them keeps the documented option effective instead of silently dropping it +# into ``cli_config`` under a key no downstream consumer reads. +_CLI_KWARG_ALIASES = {"session": "resume_session"} + + +def _apply_model_override(config_list: list[dict[str, Any]], extra: dict) -> None: + """Honour a loose ``model=``/``llm=`` kwarg by writing it onto the resolved + ``config_list`` the generator reads model selection from. + + Mirrors the CLI's ``--llm`` behaviour (see ``direct_prompt`` / + ``agents_generator``: model is taken from ``config_list[0]['model']``), so a + Python caller's ``model=`` is actually applied instead of being ignored. + Mutates ``extra`` in place, popping the consumed key so it is not also + forwarded through ``cli_config``. + """ + model = extra.pop("model", None) + if model is None: + model = extra.pop("llm", None) + if model and config_list: + config_list[0]["model"] = model + + +def _merge_cli_config(cli_config: dict | None, extra: dict) -> dict | None: + """Merge loose ``run()``/``arun()`` kwargs into the ``cli_config`` escape + hatch the CLI already forwards to ``AgentsGenerator``. + + This gives Python callers the same pass-through the CLI uses for advanced + options without inventing a parallel option surface. Loose names whose CLI + dest differs (e.g. ``session`` -> ``resume_session``) are normalised so they + reach the consumer that reads them. Explicit ``cli_config`` keys win over + loose kwargs. + """ + if not extra: + return cli_config + merged = dict(cli_config or {}) + for key, value in extra.items(): + merged.setdefault(_CLI_KWARG_ALIASES.get(key, key), value) + return merged + + def run(agent_file: str, framework: str | None = None, *, tools: list | None = None, agent_yaml: str | None = None, - cli_config: dict | None = None) -> str: - """One-line Python entry point. Equivalent to `praisonai `.""" + cli_config: dict | None = None, + **kwargs: Any) -> str: + """One-line Python entry point. Equivalent to `praisonai `. + + Advanced ``praisonai run`` options can be passed as loose keyword arguments. + ``model=`` (alias ``llm=``) selects the LLM like the CLI's ``--llm``; + ``session=`` resumes a session like ``--resume``; any other option is + forwarded through ``cli_config`` to the generator, mirroring the CLI's + pass-through. + """ from .agents_generator import AgentsGenerator adapter, config_list = _resolve_run_inputs(framework) + _apply_model_override(config_list, kwargs) + cli_config = _merge_cli_config(cli_config, kwargs) # Own the generator's lifecycle so its lazily-allocated tool-timeout # executor is released once the single run completes, instead of leaking @@ -73,8 +124,12 @@ async def arun(agent_file: str, *, tools: list | None = None, agent_yaml: str | None = None, - cli_config: dict | None = None) -> str: - """Async equivalent of `run()` using native async framework adapters.""" + cli_config: dict | None = None, + **kwargs: Any) -> str: + """Async equivalent of `run()` using native async framework adapters. + + Accepts the same loose keyword arguments as :func:`run`. + """ import asyncio from .agents_generator import AgentsGenerator @@ -83,6 +138,8 @@ async def arun(agent_file: str, # synchronous credential/config-file I/O does not block it. This is the # reason arun exists: a FastAPI handler awaiting arun must not stall the loop. adapter, config_list = await asyncio.to_thread(_resolve_run_inputs, framework) + _apply_model_override(config_list, kwargs) + cli_config = _merge_cli_config(cli_config, kwargs) # Own the generator's lifecycle so its lazily-allocated tool-timeout # executor is released once the single run completes, instead of leaking diff --git a/src/praisonai/praisonai/api/call.py b/src/praisonai/praisonai/api/call.py index cafa2d48e5..fa852842b0 100644 --- a/src/praisonai/praisonai/api/call.py +++ b/src/praisonai/praisonai/api/call.py @@ -41,6 +41,38 @@ def _maybe_load_dotenv() -> None: PORT = int(os.getenv('PORT', 8090)) NGROK_AUTH_TOKEN = os.getenv('NGROK_AUTH_TOKEN') PUBLIC = os.getenv('PUBLIC', 'false').lower() == 'true' + + +def _resolve_realtime_endpoint(): + """Resolve the realtime WebSocket URL + auth headers. + + Defaults to OpenAI so existing users with only ``OPENAI_API_KEY`` set keep + working unchanged. Operators running Azure / a self-hosted realtime-capable + gateway can point the voice path elsewhere without editing this module via: + - ``PRAISONAI_REALTIME_URL`` full ``wss://...`` URL (takes precedence) + - ``PRAISONAI_REALTIME_MODEL`` model for the default OpenAI URL + - ``PRAISONAI_REALTIME_API_KEY`` bearer key (falls back to OPENAI_API_KEY) + """ + explicit_url = os.getenv('PRAISONAI_REALTIME_URL') + api_key = os.getenv('PRAISONAI_REALTIME_API_KEY') or OPENAI_API_KEY + if explicit_url: + headers = {"Authorization": f"Bearer {api_key}"} if api_key else {} + if "openai.com" in explicit_url: + headers["OpenAI-Beta"] = "realtime=v1" + return explicit_url, headers + + model = os.getenv( + 'PRAISONAI_REALTIME_MODEL', 'gpt-4o-realtime-preview-2024-10-01' + ) + return ( + f"wss://api.openai.com/v1/realtime?model={model}", + { + "Authorization": f"Bearer {api_key}", + "OpenAI-Beta": "realtime=v1", + }, + ) + + SYSTEM_MESSAGE = ( "You are a helpful and bubbly AI assistant who loves to chat about " "anything the user is interested in and is prepared to offer them facts. " @@ -313,12 +345,17 @@ async def handle_media_stream(websocket: WebSocket): print("Client connected") await websocket.accept() + realtime_url, realtime_headers = _resolve_realtime_endpoint() async with websockets.connect( - 'wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01', - extra_headers={ - "Authorization": f"Bearer {OPENAI_API_KEY}", - "OpenAI-Beta": "realtime=v1" - } + realtime_url, + extra_headers=realtime_headers, + # Bounded connect / heartbeat / close so a dead upstream cannot + # hold a live Twilio media leg (and phone number) indefinitely. + open_timeout=10, + ping_interval=20, + ping_timeout=20, + close_timeout=5, + max_size=2 ** 20, # 1 MiB frame cap ) as openai_ws: await send_session_update(openai_ws) stream_sid = None diff --git a/src/praisonai/praisonai/llm/gateways.py b/src/praisonai/praisonai/llm/gateways.py index 5f030de66e..6a537d4546 100644 --- a/src/praisonai/praisonai/llm/gateways.py +++ b/src/praisonai/praisonai/llm/gateways.py @@ -78,6 +78,11 @@ def _resolve_model_and_kwargs(self, prompt: str, **kwargs): for key in ["provider", "gateway"]: completion_kwargs.pop(key, None) + # Enforce a default timeout + bounded retries so an unresponsive + # gateway cannot pin a request coroutine / worker thread forever. + from .registry import _apply_default_timeout + _apply_default_timeout(completion_kwargs) + messages = [{"role": "user", "content": prompt}] return litellm, self.model_id, messages, completion_kwargs diff --git a/src/praisonai/praisonai/llm/registry.py b/src/praisonai/praisonai/llm/registry.py index be82ea972c..20291ae6bd 100644 --- a/src/praisonai/praisonai/llm/registry.py +++ b/src/praisonai/praisonai/llm/registry.py @@ -31,6 +31,37 @@ def generate(self, prompt): ProviderType = Union[ProviderClass, ProviderFactory] +# Default ceilings so an unresponsive provider cannot pin a request coroutine +# (or a bridge worker thread) indefinitely. Both are overridable per-call via +# kwargs, and the timeout is tunable for operators via PRAISONAI_LLM_TIMEOUT. +_DEFAULT_LLM_TIMEOUT_SECONDS = 60.0 +_DEFAULT_LLM_NUM_RETRIES = 2 + + +def default_llm_timeout() -> float: + """Resolve the default LLM call timeout, tolerating a bad env value.""" + import os # lazy: keep module top-level imports to stdlib typing/threading only + raw = os.getenv("PRAISONAI_LLM_TIMEOUT") + if not raw: + return _DEFAULT_LLM_TIMEOUT_SECONDS + try: + return float(raw) + except ValueError: + import logging + logging.getLogger(__name__).warning( + "Invalid PRAISONAI_LLM_TIMEOUT=%r; falling back to %.0fs", + raw, _DEFAULT_LLM_TIMEOUT_SECONDS, + ) + return _DEFAULT_LLM_TIMEOUT_SECONDS + + +def _apply_default_timeout(completion_kwargs: Dict[str, Any]) -> Dict[str, Any]: + """Seed a default timeout + bounded retries. An explicit caller value wins.""" + completion_kwargs.setdefault("timeout", default_llm_timeout()) + completion_kwargs.setdefault("num_retries", _DEFAULT_LLM_NUM_RETRIES) + return completion_kwargs + + class LLMProviderRegistry(PluginRegistry[ProviderType]): """ Registry for LLM providers. @@ -255,6 +286,7 @@ def _resolve_model_and_kwargs(self, prompt: str, **kwargs): completion_kwargs = { k: v for k, v in {**self.config, **kwargs}.items() if k != "provider" } + _apply_default_timeout(completion_kwargs) messages = [{"role": "user", "content": prompt}] return litellm, full_model, messages, completion_kwargs