refac
This commit is contained in:
1 parent
8ddf119570
commit
4a7d4ebada
4 files changed
+28
-6
No files matched your search
@@ -565,6 +565,16 @@ try:
|
|||||||
except (ValueError, TypeError):
|
except (ValueError, TypeError):
|
||||||
AIOHTTP_CLIENT_TIMEOUT = 300
|
AIOHTTP_CLIENT_TIMEOUT = 300
|
||||||
|
|
||||||
|
# Optional between-chunks idle cap for streaming aiohttp requests.
|
||||||
|
_stream_idle_timeout_raw = os.getenv('AIOHTTP_CLIENT_STREAM_IDLE_TIMEOUT', '')
|
||||||
|
try:
|
||||||
|
AIOHTTP_CLIENT_STREAM_IDLE_TIMEOUT = int(_stream_idle_timeout_raw) if _stream_idle_timeout_raw else None
|
||||||
|
except (ValueError, TypeError):
|
||||||
|
AIOHTTP_CLIENT_STREAM_IDLE_TIMEOUT = None
|
||||||
|
|
||||||
|
if AIOHTTP_CLIENT_STREAM_IDLE_TIMEOUT is not None and AIOHTTP_CLIENT_STREAM_IDLE_TIMEOUT <= 0:
|
||||||
|
AIOHTTP_CLIENT_STREAM_IDLE_TIMEOUT = None
|
||||||
|
|
||||||
|
|
||||||
# SSL verification for general outbound requests (OpenAI, OAuth, etc.).
|
# SSL verification for general outbound requests (OpenAI, OAuth, etc.).
|
||||||
# Accepts "True", "False", or a path to a CA bundle file.
|
# Accepts "True", "False", or a path to a CA bundle file.
|
||||||
|
|||||||
@@ -48,7 +48,7 @@ from open_webui.utils.payload import (
|
|||||||
apply_model_params_to_body_openai,
|
apply_model_params_to_body_openai,
|
||||||
apply_system_prompt_to_body,
|
apply_system_prompt_to_body,
|
||||||
)
|
)
|
||||||
from open_webui.utils.session_pool import cleanup_response, get_session, stream_wrapper
|
from open_webui.utils.session_pool import cleanup_response, get_client_timeout, get_session, stream_wrapper
|
||||||
|
|
||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
@@ -130,7 +130,7 @@ async def send_request(
|
|||||||
data=payload,
|
data=payload,
|
||||||
headers=headers,
|
headers=headers,
|
||||||
ssl=AIOHTTP_CLIENT_SESSION_SSL,
|
ssl=AIOHTTP_CLIENT_SESSION_SSL,
|
||||||
timeout=aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT),
|
timeout=get_client_timeout(stream=stream),
|
||||||
)
|
)
|
||||||
|
|
||||||
if not r.ok:
|
if not r.ok:
|
||||||
|
|||||||
@@ -55,6 +55,7 @@ from open_webui.utils.payload import (
|
|||||||
)
|
)
|
||||||
from open_webui.utils.session_pool import (
|
from open_webui.utils.session_pool import (
|
||||||
cleanup_response,
|
cleanup_response,
|
||||||
|
get_client_timeout,
|
||||||
get_session,
|
get_session,
|
||||||
stream_wrapper,
|
stream_wrapper,
|
||||||
)
|
)
|
||||||
@@ -1344,6 +1345,7 @@ async def generate_chat_completion(
|
|||||||
part.get('text', '') for part in message['content'] if part.get('type') in ('input_text', 'text')
|
part.get('text', '') for part in message['content'] if part.get('type') in ('input_text', 'text')
|
||||||
)
|
)
|
||||||
|
|
||||||
|
is_streaming_request = bool(payload.get('stream', False))
|
||||||
payload = json.dumps(payload)
|
payload = json.dumps(payload)
|
||||||
|
|
||||||
r = None
|
r = None
|
||||||
@@ -1360,7 +1362,7 @@ async def generate_chat_completion(
|
|||||||
headers=headers,
|
headers=headers,
|
||||||
cookies=cookies,
|
cookies=cookies,
|
||||||
ssl=AIOHTTP_CLIENT_SESSION_SSL,
|
ssl=AIOHTTP_CLIENT_SESSION_SSL,
|
||||||
timeout=aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT),
|
timeout=get_client_timeout(stream=is_streaming_request),
|
||||||
)
|
)
|
||||||
|
|
||||||
# Check if response is SSE
|
# Check if response is SSE
|
||||||
@@ -1587,6 +1589,7 @@ async def responses(
|
|||||||
Routes to the correct upstream backend based on the model field.
|
Routes to the correct upstream backend based on the model field.
|
||||||
"""
|
"""
|
||||||
payload = form_data.model_dump(exclude_none=True)
|
payload = form_data.model_dump(exclude_none=True)
|
||||||
|
is_streaming_request = bool(payload.get('stream', False))
|
||||||
|
|
||||||
idx = 0
|
idx = 0
|
||||||
model_id = form_data.model
|
model_id = form_data.model
|
||||||
@@ -1637,7 +1640,7 @@ async def responses(
|
|||||||
headers=headers,
|
headers=headers,
|
||||||
cookies=cookies,
|
cookies=cookies,
|
||||||
ssl=AIOHTTP_CLIENT_SESSION_SSL,
|
ssl=AIOHTTP_CLIENT_SESSION_SSL,
|
||||||
timeout=aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT),
|
timeout=get_client_timeout(stream=is_streaming_request),
|
||||||
)
|
)
|
||||||
|
|
||||||
# Check if response is SSE
|
# Check if response is SSE
|
||||||
@@ -1707,6 +1710,7 @@ async def proxy(path: str, request: Request, user=Depends(get_verified_user)):
|
|||||||
payload = json.loads(body)
|
payload = json.loads(body)
|
||||||
except (json.JSONDecodeError, ValueError):
|
except (json.JSONDecodeError, ValueError):
|
||||||
payload = None
|
payload = None
|
||||||
|
is_streaming_request = bool(payload.get('stream', False)) if isinstance(payload, dict) else False
|
||||||
|
|
||||||
idx = 0
|
idx = 0
|
||||||
model_id = payload.get('model') if isinstance(payload, dict) else None
|
model_id = payload.get('model') if isinstance(payload, dict) else None
|
||||||
@@ -1758,7 +1762,7 @@ async def proxy(path: str, request: Request, user=Depends(get_verified_user)):
|
|||||||
headers=headers,
|
headers=headers,
|
||||||
cookies=cookies,
|
cookies=cookies,
|
||||||
ssl=AIOHTTP_CLIENT_SESSION_SSL,
|
ssl=AIOHTTP_CLIENT_SESSION_SSL,
|
||||||
timeout=aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT),
|
timeout=get_client_timeout(stream=is_streaming_request),
|
||||||
)
|
)
|
||||||
|
|
||||||
# Check if response is SSE
|
# Check if response is SSE
|
||||||
|
|||||||
@@ -28,6 +28,7 @@ from typing import Optional
|
|||||||
|
|
||||||
import aiohttp
|
import aiohttp
|
||||||
from open_webui.env import (
|
from open_webui.env import (
|
||||||
|
AIOHTTP_CLIENT_STREAM_IDLE_TIMEOUT,
|
||||||
AIOHTTP_CLIENT_TIMEOUT,
|
AIOHTTP_CLIENT_TIMEOUT,
|
||||||
AIOHTTP_POOL_CONNECTIONS,
|
AIOHTTP_POOL_CONNECTIONS,
|
||||||
AIOHTTP_POOL_CONNECTIONS_PER_HOST,
|
AIOHTTP_POOL_CONNECTIONS_PER_HOST,
|
||||||
@@ -39,6 +40,13 @@ log = logging.getLogger(__name__)
|
|||||||
_session: Optional[aiohttp.ClientSession] = None
|
_session: Optional[aiohttp.ClientSession] = None
|
||||||
|
|
||||||
|
|
||||||
|
def get_client_timeout(stream: bool = False) -> aiohttp.ClientTimeout:
|
||||||
|
return aiohttp.ClientTimeout(
|
||||||
|
total=AIOHTTP_CLIENT_TIMEOUT,
|
||||||
|
sock_read=AIOHTTP_CLIENT_STREAM_IDLE_TIMEOUT if stream else None,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
async def get_session() -> aiohttp.ClientSession:
|
async def get_session() -> aiohttp.ClientSession:
|
||||||
"""Return the shared aiohttp ClientSession, creating it lazily."""
|
"""Return the shared aiohttp ClientSession, creating it lazily."""
|
||||||
global _session
|
global _session
|
||||||
@@ -56,7 +64,7 @@ async def get_session() -> aiohttp.ClientSession:
|
|||||||
else:
|
else:
|
||||||
connector_kwargs['limit_per_host'] = 0 # aiohttp: 0 = unlimited
|
connector_kwargs['limit_per_host'] = 0 # aiohttp: 0 = unlimited
|
||||||
connector = aiohttp.TCPConnector(**connector_kwargs)
|
connector = aiohttp.TCPConnector(**connector_kwargs)
|
||||||
timeout = aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT)
|
timeout = get_client_timeout()
|
||||||
_session = aiohttp.ClientSession(
|
_session = aiohttp.ClientSession(
|
||||||
connector=connector,
|
connector=connector,
|
||||||
timeout=timeout,
|
timeout=timeout,
|
||||||
|
|||||||
Reference in new issue
Block a user