mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-26 17:38:36 +00:00
feat(aux): force streaming for providers that reject non-stream requests
Some OpenAI-compatible endpoints — notably Tencent Copilot (copilot.tencent.com) — only accept streaming chat requests; any non-streaming call returns HTTP 400 (code 11101, 'Non-stream chat request is currently not supported'). The main conversation loop already streams, so interactive chat works, but every auxiliary task (title generation, compression, web extraction) used the non-streaming path and failed on each call. _provider_requires_stream() detects stream-only endpoints (copilot.tencent.com built in, plus user-configurable auxiliary.stream_only_base_urls substring markers in config.yaml). Matching sync auxiliary calls route through _create_with_progress (force_stream=True) and async calls through the new _acreate_with_stream, aggregating the chunk stream — including tool-call deltas and reasoning deltas — into a complete response via the shared _ChatStreamAccumulator. Salvaged from PR #60686 by @kudi88 onto the progress-aware streaming machinery from #71508, addressing both sweeper-review gaps: the async path now consumes the stream with 'async for' (awaiting create() and iterating synchronously raised on AsyncOpenAI streams), and tool-call deltas are reassembled instead of dropped (MCP passes tools= through call_llm). Under force_stream there is no silent non-streaming retry — a stream-only provider rejects those by definition, so the original error surfaces to the normal recovery chains.
This commit is contained in:
kudi88
Teknium
parent
fffa661223
commit
5121a2a20e
5 changed files with 404 additions and 94 deletions
|
|
@ -7234,20 +7234,67 @@ def _is_streaming_rejected_error(exc: Exception) -> bool:
|
|||
)
|
||||
|
||||
|
||||
def _create_with_progress(client: Any, kwargs: Dict[str, Any], task: Optional[str] = None) -> Any:
|
||||
"""chat.completions.create() that streams when a progress hook is active.
|
||||
def _provider_requires_stream(provider: str, base_url: Optional[str]) -> bool:
|
||||
"""Detect providers that only accept streaming (non-stream = HTTP 400).
|
||||
|
||||
Some OpenAI-compatible endpoints reject non-streaming chat requests
|
||||
outright — e.g. Tencent Copilot returns
|
||||
``{"code": 11101, "msg": "Non-stream chat request is currently not
|
||||
supported"}``. The main conversation loop already streams, so interactive
|
||||
chat works; auxiliary tasks (title generation, compression, web extract)
|
||||
used the non-streaming path and failed on every call. When this returns
|
||||
True the auxiliary client sends ``stream=True`` and aggregates the chunks
|
||||
itself (see :func:`_aggregate_chat_stream`). Credit @kudi88 (PR #60686).
|
||||
|
||||
Beyond the known-host list, users can mark ANY custom endpoint as
|
||||
stream-only via ``auxiliary.stream_only_base_urls`` in config.yaml
|
||||
(list of substrings matched against the endpoint URL).
|
||||
"""
|
||||
_url = str(base_url or "").lower()
|
||||
if not _url:
|
||||
return False
|
||||
# Tencent Copilot — "Non-stream chat request is currently not supported"
|
||||
if base_url_host_matches(_url, "copilot.tencent.com"):
|
||||
return True
|
||||
try:
|
||||
from hermes_cli.config import load_config
|
||||
aux_cfg = (load_config() or {}).get("auxiliary", {})
|
||||
markers = aux_cfg.get("stream_only_base_urls") or []
|
||||
if isinstance(markers, (list, tuple)):
|
||||
for marker in markers:
|
||||
if isinstance(marker, str) and marker.strip() and marker.strip().lower() in _url:
|
||||
return True
|
||||
except Exception:
|
||||
# Config read is best-effort; never break an aux call over it.
|
||||
pass
|
||||
return False
|
||||
|
||||
|
||||
def _create_with_progress(
|
||||
client: Any,
|
||||
kwargs: Dict[str, Any],
|
||||
task: Optional[str] = None,
|
||||
*,
|
||||
force_stream: bool = False,
|
||||
) -> Any:
|
||||
"""chat.completions.create() that streams when a progress hook is active
|
||||
or the provider only accepts streamed requests.
|
||||
|
||||
Behavior is byte-for-byte identical to a plain ``create(**kwargs)`` when
|
||||
no hook is installed (every existing caller/task) or when the client's
|
||||
neither trigger applies (every existing caller/task) or when the client's
|
||||
wire adapter streams internally. With a hook + a chunk-capable client,
|
||||
the request is sent with ``stream=True`` and aggregated, ticking the hook
|
||||
per chunk — so the configured ``timeout`` acts per stream read (idle)
|
||||
rather than as a total budget, and outer liveness watchdogs see tokens
|
||||
moving. Providers that reject the streamed request fall back to the
|
||||
plain non-streaming call.
|
||||
moving. ``force_stream=True`` (stream-only providers such as Tencent
|
||||
Copilot — credit @kudi88, PR #60686) takes the same streamed path even
|
||||
without a hook. Providers that reject the streamed request fall back to
|
||||
the plain non-streaming call — except under ``force_stream``, where a
|
||||
stream-only provider rejects the plain call by definition, so the
|
||||
original error is surfaced to the normal recovery chains instead.
|
||||
"""
|
||||
_notify_aux_progress() # request dispatched counts as progress
|
||||
if not _aux_progress_active() or _client_streams_internally(client):
|
||||
if (not _aux_progress_active() and not force_stream) or _client_streams_internally(client):
|
||||
return client.chat.completions.create(**kwargs)
|
||||
|
||||
total_ceiling = _aux_stream_total_ceiling(kwargs.get("timeout"))
|
||||
|
|
@ -7262,7 +7309,8 @@ def _create_with_progress(client: Any, kwargs: Dict[str, Any], task: Optional[st
|
|||
# recovery chains (credential refresh, pool rotation, provider
|
||||
# fallback) see the same error they would on a plain call.
|
||||
if (
|
||||
_is_transient_transport_error(exc)
|
||||
force_stream
|
||||
or _is_transient_transport_error(exc)
|
||||
or _is_auth_error(exc)
|
||||
or _is_payment_error(exc)
|
||||
or _is_rate_limit_error(exc)
|
||||
|
|
@ -7300,62 +7348,13 @@ def _aggregate_chat_stream(
|
|||
TimeoutError when *total_ceiling* seconds elapse before the stream
|
||||
finishes — phrased with "timed out" so existing timeout classification
|
||||
(``_is_timeout_error``) treats it exactly like a request timeout.
|
||||
Accumulation is shared with the async mirror via
|
||||
:class:`_ChatStreamAccumulator`.
|
||||
"""
|
||||
started = time.monotonic()
|
||||
content_parts: List[str] = []
|
||||
reasoning_parts: List[str] = []
|
||||
tool_calls_acc: Dict[int, Dict[str, Any]] = {}
|
||||
finish_reason = None
|
||||
usage = None
|
||||
resp_id = ""
|
||||
resp_model = model or ""
|
||||
acc = _ChatStreamAccumulator(model=model, total_ceiling=total_ceiling)
|
||||
try:
|
||||
for chunk in chunks:
|
||||
_notify_aux_progress()
|
||||
if total_ceiling is not None and (time.monotonic() - started) >= total_ceiling:
|
||||
raise TimeoutError(
|
||||
f"Auxiliary streamed call timed out after {total_ceiling:.0f}s "
|
||||
"total ceiling (stream still open but over budget)"
|
||||
)
|
||||
resp_id = getattr(chunk, "id", None) or resp_id
|
||||
resp_model = getattr(chunk, "model", None) or resp_model
|
||||
chunk_usage = getattr(chunk, "usage", None)
|
||||
if chunk_usage:
|
||||
usage = chunk_usage
|
||||
choices = getattr(chunk, "choices", None) or []
|
||||
if not choices:
|
||||
continue
|
||||
choice = choices[0]
|
||||
finish_reason = getattr(choice, "finish_reason", None) or finish_reason
|
||||
delta = getattr(choice, "delta", None)
|
||||
if delta is None:
|
||||
continue
|
||||
piece = getattr(delta, "content", None)
|
||||
if piece:
|
||||
content_parts.append(piece)
|
||||
# Thinking models stream reasoning deltas before any content —
|
||||
# they count as forward progress (the tick above) and are kept
|
||||
# out of .content, mirroring non-streaming responses where
|
||||
# reasoning arrives in a separate field.
|
||||
reasoning_piece = (
|
||||
getattr(delta, "reasoning", None)
|
||||
or getattr(delta, "reasoning_content", None)
|
||||
)
|
||||
if reasoning_piece and isinstance(reasoning_piece, str):
|
||||
reasoning_parts.append(reasoning_piece)
|
||||
for tc in (getattr(delta, "tool_calls", None) or []):
|
||||
idx = getattr(tc, "index", 0) or 0
|
||||
acc = tool_calls_acc.setdefault(
|
||||
idx, {"id": "", "name": "", "arguments": []}
|
||||
)
|
||||
if getattr(tc, "id", None):
|
||||
acc["id"] = tc.id
|
||||
fn = getattr(tc, "function", None)
|
||||
if fn is not None:
|
||||
if getattr(fn, "name", None):
|
||||
acc["name"] = fn.name
|
||||
if getattr(fn, "arguments", None):
|
||||
acc["arguments"].append(fn.arguments)
|
||||
acc.feed(chunk)
|
||||
finally:
|
||||
close_fn = getattr(chunks, "close", None)
|
||||
if callable(close_fn):
|
||||
|
|
@ -7363,38 +7362,157 @@ def _aggregate_chat_stream(
|
|||
close_fn()
|
||||
except Exception:
|
||||
pass
|
||||
return acc.finish()
|
||||
|
||||
tool_calls = None
|
||||
if tool_calls_acc:
|
||||
tool_calls = [
|
||||
SimpleNamespace(
|
||||
id=acc["id"],
|
||||
type="function",
|
||||
function=SimpleNamespace(
|
||||
name=acc["name"],
|
||||
arguments="".join(acc["arguments"]),
|
||||
),
|
||||
|
||||
class _ChatStreamAccumulator:
|
||||
"""Shared per-chunk accumulation for sync and async stream aggregation.
|
||||
|
||||
Mirrors :func:`_aggregate_chat_stream`'s chunk handling so the async
|
||||
consumer below cannot drift from the sync one (same content/reasoning/
|
||||
tool-call delta reassembly, same "timed out" ceiling phrasing).
|
||||
"""
|
||||
|
||||
def __init__(self, model: str = "", total_ceiling: Optional[float] = None):
|
||||
self._started = time.monotonic()
|
||||
self._total_ceiling = total_ceiling
|
||||
self.content_parts: List[str] = []
|
||||
self.reasoning_parts: List[str] = []
|
||||
self.tool_calls_acc: Dict[int, Dict[str, Any]] = {}
|
||||
self.finish_reason = None
|
||||
self.usage = None
|
||||
self.resp_id = ""
|
||||
self.resp_model = model or ""
|
||||
|
||||
def feed(self, chunk: Any) -> None:
|
||||
_notify_aux_progress()
|
||||
if (
|
||||
self._total_ceiling is not None
|
||||
and (time.monotonic() - self._started) >= self._total_ceiling
|
||||
):
|
||||
raise TimeoutError(
|
||||
f"Auxiliary streamed call timed out after {self._total_ceiling:.0f}s "
|
||||
"total ceiling (stream still open but over budget)"
|
||||
)
|
||||
for _idx, acc in sorted(tool_calls_acc.items())
|
||||
]
|
||||
self.resp_id = getattr(chunk, "id", None) or self.resp_id
|
||||
self.resp_model = getattr(chunk, "model", None) or self.resp_model
|
||||
chunk_usage = getattr(chunk, "usage", None)
|
||||
if chunk_usage:
|
||||
self.usage = chunk_usage
|
||||
choices = getattr(chunk, "choices", None) or []
|
||||
if not choices:
|
||||
return
|
||||
choice = choices[0]
|
||||
self.finish_reason = getattr(choice, "finish_reason", None) or self.finish_reason
|
||||
delta = getattr(choice, "delta", None)
|
||||
if delta is None:
|
||||
return
|
||||
piece = getattr(delta, "content", None)
|
||||
if piece:
|
||||
self.content_parts.append(piece)
|
||||
reasoning_piece = (
|
||||
getattr(delta, "reasoning", None)
|
||||
or getattr(delta, "reasoning_content", None)
|
||||
)
|
||||
if reasoning_piece and isinstance(reasoning_piece, str):
|
||||
self.reasoning_parts.append(reasoning_piece)
|
||||
for tc in (getattr(delta, "tool_calls", None) or []):
|
||||
idx = getattr(tc, "index", 0) or 0
|
||||
acc = self.tool_calls_acc.setdefault(
|
||||
idx, {"id": "", "name": "", "arguments": []}
|
||||
)
|
||||
if getattr(tc, "id", None):
|
||||
acc["id"] = tc.id
|
||||
fn = getattr(tc, "function", None)
|
||||
if fn is not None:
|
||||
if getattr(fn, "name", None):
|
||||
acc["name"] = fn.name
|
||||
if getattr(fn, "arguments", None):
|
||||
acc["arguments"].append(fn.arguments)
|
||||
|
||||
message = SimpleNamespace(
|
||||
role="assistant",
|
||||
content="".join(content_parts),
|
||||
tool_calls=tool_calls,
|
||||
reasoning="".join(reasoning_parts) or None,
|
||||
)
|
||||
choice = SimpleNamespace(
|
||||
index=0,
|
||||
message=message,
|
||||
finish_reason=finish_reason or "stop",
|
||||
)
|
||||
return SimpleNamespace(
|
||||
id=resp_id,
|
||||
model=resp_model,
|
||||
object="chat.completion",
|
||||
choices=[choice],
|
||||
usage=usage,
|
||||
def finish(self) -> Any:
|
||||
tool_calls = None
|
||||
if self.tool_calls_acc:
|
||||
tool_calls = [
|
||||
SimpleNamespace(
|
||||
id=acc["id"],
|
||||
type="function",
|
||||
function=SimpleNamespace(
|
||||
name=acc["name"],
|
||||
arguments="".join(acc["arguments"]),
|
||||
),
|
||||
)
|
||||
for _idx, acc in sorted(self.tool_calls_acc.items())
|
||||
]
|
||||
message = SimpleNamespace(
|
||||
role="assistant",
|
||||
content="".join(self.content_parts),
|
||||
tool_calls=tool_calls,
|
||||
reasoning="".join(self.reasoning_parts) or None,
|
||||
)
|
||||
choice = SimpleNamespace(
|
||||
index=0,
|
||||
message=message,
|
||||
finish_reason=self.finish_reason or "stop",
|
||||
)
|
||||
return SimpleNamespace(
|
||||
id=self.resp_id,
|
||||
model=self.resp_model,
|
||||
object="chat.completion",
|
||||
choices=[choice],
|
||||
usage=self.usage,
|
||||
)
|
||||
|
||||
|
||||
async def _aggregate_chat_stream_async(
|
||||
chunks: Any,
|
||||
*,
|
||||
model: str = "",
|
||||
total_ceiling: Optional[float] = None,
|
||||
) -> Any:
|
||||
"""Async mirror of :func:`_aggregate_chat_stream` (``async for`` consumer).
|
||||
|
||||
The AsyncOpenAI stream contract is an async iterator — consuming it with
|
||||
the sync helper raises. Same accumulation and ceiling semantics via
|
||||
:class:`_ChatStreamAccumulator`.
|
||||
"""
|
||||
acc = _ChatStreamAccumulator(model=model, total_ceiling=total_ceiling)
|
||||
try:
|
||||
async for chunk in chunks:
|
||||
acc.feed(chunk)
|
||||
finally:
|
||||
close_fn = getattr(chunks, "close", None) or getattr(chunks, "aclose", None)
|
||||
if callable(close_fn):
|
||||
try:
|
||||
result = close_fn()
|
||||
if inspect.isawaitable(result):
|
||||
await result
|
||||
except Exception:
|
||||
pass
|
||||
return acc.finish()
|
||||
|
||||
|
||||
async def _acreate_with_stream(
|
||||
client: Any,
|
||||
kwargs: Dict[str, Any],
|
||||
task: Optional[str] = None,
|
||||
) -> Any:
|
||||
"""Async chat.completions.create() for stream-only providers.
|
||||
|
||||
Sends ``stream=True`` and aggregates the async chunk stream into a
|
||||
complete response (credit @kudi88, PR #60686 — async contract fixed to
|
||||
``async for`` and tool-call deltas preserved per sweeper review).
|
||||
"""
|
||||
total_ceiling = _aux_stream_total_ceiling(kwargs.get("timeout"))
|
||||
stream_kwargs = dict(kwargs)
|
||||
stream_kwargs["stream"] = True
|
||||
stream_kwargs["stream_options"] = {"include_usage": True}
|
||||
chunks = await client.chat.completions.create(**stream_kwargs)
|
||||
# Defensive: shims may hand back a complete response despite stream=True.
|
||||
if hasattr(chunks, "choices"):
|
||||
return chunks
|
||||
return await _aggregate_chat_stream_async(
|
||||
chunks, model=str(kwargs.get("model") or ""), total_ceiling=total_ceiling,
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -7598,7 +7716,13 @@ def call_llm(
|
|||
# for the transient retry every auxiliary task shares. (PR #16587)
|
||||
try:
|
||||
return _validate_llm_response(
|
||||
_create_with_progress(client, kwargs, task), task,
|
||||
_create_with_progress(
|
||||
client, kwargs, task,
|
||||
force_stream=_provider_requires_stream(
|
||||
resolved_provider, _base_info or resolved_base_url,
|
||||
),
|
||||
),
|
||||
task,
|
||||
provider=resolved_provider, base_url=_base_info)
|
||||
except Exception as transient_err:
|
||||
if not _is_transient_transport_error(transient_err):
|
||||
|
|
@ -7631,7 +7755,13 @@ def call_llm(
|
|||
time.sleep(_backoff)
|
||||
try:
|
||||
return _validate_llm_response(
|
||||
_create_with_progress(client, kwargs, task), task)
|
||||
_create_with_progress(
|
||||
client, kwargs, task,
|
||||
force_stream=_provider_requires_stream(
|
||||
resolved_provider, _base_info or resolved_base_url,
|
||||
),
|
||||
),
|
||||
task)
|
||||
except Exception as retry_transient:
|
||||
if not _is_transient_transport_error(retry_transient):
|
||||
raise
|
||||
|
|
@ -8184,9 +8314,25 @@ async def async_call_llm(
|
|||
# Retry ONCE on the same provider for a transient transport blip
|
||||
# before the except-chain escalates to fallback — see call_llm()
|
||||
# for the rationale. (PR #16587)
|
||||
_force_stream_async = (
|
||||
_provider_requires_stream(
|
||||
resolved_provider, _client_base or resolved_base_url,
|
||||
)
|
||||
and not isinstance(client, (
|
||||
AsyncCodexAuxiliaryClient,
|
||||
AsyncAnthropicAuxiliaryClient,
|
||||
AsyncBedrockAuxiliaryClient,
|
||||
))
|
||||
)
|
||||
|
||||
async def _acreate(_kwargs: Dict[str, Any]) -> Any:
|
||||
if _force_stream_async:
|
||||
return await _acreate_with_stream(client, _kwargs, task)
|
||||
return await client.chat.completions.create(**_kwargs)
|
||||
|
||||
try:
|
||||
return _validate_llm_response(
|
||||
await client.chat.completions.create(**kwargs), task,
|
||||
await _acreate(kwargs), task,
|
||||
provider=resolved_provider, base_url=_client_base)
|
||||
except Exception as transient_err:
|
||||
if not _is_transient_transport_error(transient_err):
|
||||
|
|
@ -8207,7 +8353,7 @@ async def async_call_llm(
|
|||
task or "call", transient_err,
|
||||
)
|
||||
return _validate_llm_response(
|
||||
await client.chat.completions.create(**kwargs), task)
|
||||
await _acreate(kwargs), task)
|
||||
except Exception as first_err:
|
||||
if "temperature" in kwargs and _is_unsupported_temperature_error(first_err):
|
||||
retry_kwargs = dict(kwargs)
|
||||
|
|
|
|||
1
contributors/emails/kudi3699@gmail.com
Normal file
1
contributors/emails/kudi3699@gmail.com
Normal file
|
|
@ -0,0 +1 @@
|
|||
kudi88
|
||||
|
|
@ -1658,6 +1658,13 @@ DEFAULT_CONFIG = {
|
|||
# not a meaningful recovery, so an unretried blip silently loses the
|
||||
# call.
|
||||
"transient_retries": 2,
|
||||
# Endpoints that reject NON-streaming chat requests outright (e.g.
|
||||
# Tencent Copilot returns HTTP 400 "Non-stream chat request is
|
||||
# currently not supported"). Auxiliary calls to a matching endpoint
|
||||
# are sent with stream=True and aggregated client-side. Entries are
|
||||
# case-insensitive substrings matched against the endpoint URL;
|
||||
# copilot.tencent.com is always treated as stream-only.
|
||||
"stream_only_base_urls": [],
|
||||
"vision": {
|
||||
"provider": "auto", # auto | openrouter | nous | codex | custom
|
||||
"model": "", # e.g. "google/gemini-2.5-flash", "gpt-4o"
|
||||
|
|
|
|||
|
|
@ -10,15 +10,18 @@ hook, behavior is byte-for-byte the old non-streaming call.
|
|||
import threading
|
||||
import time
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from agent.auxiliary_client import (
|
||||
_acreate_with_stream,
|
||||
_aggregate_chat_stream,
|
||||
_aggregate_chat_stream_async,
|
||||
_aux_stream_total_ceiling,
|
||||
_create_with_progress,
|
||||
_notify_aux_progress,
|
||||
_provider_requires_stream,
|
||||
aux_progress_hook,
|
||||
)
|
||||
from agent.conversation_compression import CompressionCommitFence
|
||||
|
|
@ -271,3 +274,144 @@ class TestFenceProgress:
|
|||
with aux_progress_hook(fence.touch_progress):
|
||||
_notify_aux_progress()
|
||||
assert fence.seconds_since_progress() < 0.05
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Stream-only providers (credit @kudi88, PR #60686)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestProviderRequiresStream:
|
||||
def test_tencent_copilot_is_stream_only(self):
|
||||
assert _provider_requires_stream(
|
||||
"custom", "https://copilot.tencent.com/v1"
|
||||
) is True
|
||||
|
||||
def test_normal_endpoints_are_not(self):
|
||||
assert _provider_requires_stream(
|
||||
"openrouter", "https://openrouter.ai/api/v1"
|
||||
) is False
|
||||
assert _provider_requires_stream("auto", None) is False
|
||||
assert _provider_requires_stream("auto", "") is False
|
||||
|
||||
def test_config_marker_matches_custom_endpoint(self):
|
||||
with patch(
|
||||
"hermes_cli.config.load_config",
|
||||
return_value={"auxiliary": {"stream_only_base_urls": ["my-proxy.example.com"]}},
|
||||
):
|
||||
assert _provider_requires_stream(
|
||||
"custom", "https://my-proxy.example.com/v1"
|
||||
) is True
|
||||
assert _provider_requires_stream(
|
||||
"custom", "https://other.example.com/v1"
|
||||
) is False
|
||||
|
||||
def test_config_read_failure_fails_open_to_non_streaming(self):
|
||||
with patch(
|
||||
"hermes_cli.config.load_config",
|
||||
side_effect=RuntimeError("config broken"),
|
||||
):
|
||||
assert _provider_requires_stream(
|
||||
"custom", "https://other.example.com/v1"
|
||||
) is False
|
||||
|
||||
|
||||
class TestForceStream:
|
||||
def test_force_stream_streams_without_a_hook(self):
|
||||
chunks = [_chunk(content="hi", finish_reason="stop")]
|
||||
client = _FakeClient(stream_chunks=chunks)
|
||||
# NO aux_progress_hook installed — force_stream alone must stream.
|
||||
result = _create_with_progress(
|
||||
client, {"model": "m1", "messages": []}, force_stream=True,
|
||||
)
|
||||
assert client.calls[0]["stream"] is True
|
||||
assert result.choices[0].message.content == "hi"
|
||||
|
||||
def test_force_stream_does_not_retry_nonstreaming_on_failure(self):
|
||||
client = _FakeClient(
|
||||
response=_COMPLETE,
|
||||
stream_error=RuntimeError("HTTP 400 bad request"),
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="bad request"):
|
||||
_create_with_progress(
|
||||
client, {"model": "m1", "messages": []}, force_stream=True,
|
||||
)
|
||||
# No silent non-streaming retry — the provider rejects those anyway.
|
||||
assert len(client.calls) == 1
|
||||
|
||||
|
||||
class TestAsyncStreamAggregation:
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_stream_is_consumed_with_async_for(self):
|
||||
# The sweeper review of PR #60686 flagged that awaiting create() and
|
||||
# then iterating synchronously raises — the async contract is
|
||||
# ``async for``. Verify the async aggregator consumes a real async
|
||||
# iterator and preserves tool-call deltas.
|
||||
tc0 = SimpleNamespace(
|
||||
index=0, id="call_9",
|
||||
function=SimpleNamespace(name="lookup", arguments='{"q":'),
|
||||
)
|
||||
tc1 = SimpleNamespace(
|
||||
index=0, id=None,
|
||||
function=SimpleNamespace(name=None, arguments='"x"}'),
|
||||
)
|
||||
raw_chunks = [
|
||||
_chunk(content="part1 "),
|
||||
_chunk(tool_calls=[tc0]),
|
||||
_chunk(tool_calls=[tc1], content="part2", finish_reason="tool_calls"),
|
||||
]
|
||||
|
||||
class _AsyncStream:
|
||||
def __init__(self, items):
|
||||
self._items = list(items)
|
||||
self.closed = False
|
||||
|
||||
def __aiter__(self):
|
||||
return self
|
||||
|
||||
async def __anext__(self):
|
||||
if not self._items:
|
||||
raise StopAsyncIteration
|
||||
return self._items.pop(0)
|
||||
|
||||
async def close(self):
|
||||
self.closed = True
|
||||
|
||||
stream = _AsyncStream(raw_chunks)
|
||||
result = await _aggregate_chat_stream_async(stream)
|
||||
msg = result.choices[0].message
|
||||
assert msg.content == "part1 part2"
|
||||
assert msg.tool_calls[0].function.name == "lookup"
|
||||
assert msg.tool_calls[0].function.arguments == '{"q":"x"}'
|
||||
assert result.choices[0].finish_reason == "tool_calls"
|
||||
assert stream.closed is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_acreate_with_stream_passes_stream_kwargs(self):
|
||||
calls = []
|
||||
|
||||
class _AsyncStream:
|
||||
def __init__(self, items):
|
||||
self._items = list(items)
|
||||
|
||||
def __aiter__(self):
|
||||
return self
|
||||
|
||||
async def __anext__(self):
|
||||
if not self._items:
|
||||
raise StopAsyncIteration
|
||||
return self._items.pop(0)
|
||||
|
||||
class _AsyncClient:
|
||||
def __init__(self):
|
||||
completions = SimpleNamespace(create=self._create)
|
||||
self.chat = SimpleNamespace(completions=completions)
|
||||
|
||||
async def _create(self, **kwargs):
|
||||
calls.append(kwargs)
|
||||
return _AsyncStream([_chunk(content="ok", finish_reason="stop")])
|
||||
|
||||
result = await _acreate_with_stream(
|
||||
_AsyncClient(), {"model": "m1", "messages": [], "timeout": 30},
|
||||
)
|
||||
assert calls[0]["stream"] is True
|
||||
assert result.choices[0].message.content == "ok"
|
||||
|
|
|
|||
|
|
@ -985,6 +985,18 @@ If you do not want Hermes to auto-generate titles after the first exchange, set
|
|||
`auxiliary.title_generation.enabled: false`. Manual titles still work through
|
||||
`/title` and `hermes sessions rename`.
|
||||
|
||||
### Stream-only endpoints
|
||||
|
||||
Some OpenAI-compatible endpoints reject non-streaming chat requests outright (e.g. Tencent Copilot returns HTTP 400 `"Non-stream chat request is currently not supported"`). Interactive chat already streams, but auxiliary tasks (title generation, compression, web extraction) use non-streaming calls and would fail on every attempt. Hermes always treats `copilot.tencent.com` as stream-only; for any other such endpoint, list a URL substring under `auxiliary.stream_only_base_urls`:
|
||||
|
||||
```yaml
|
||||
auxiliary:
|
||||
stream_only_base_urls:
|
||||
- "my-stream-only-proxy.example.com"
|
||||
```
|
||||
|
||||
Matching auxiliary calls are sent with `stream=True` and the chunks (including tool-call deltas) are aggregated client-side — no behavior change for any other endpoint.
|
||||
|
||||
### Video Tutorial
|
||||
|
||||
<div style={{position: 'relative', width: '100%', aspectRatio: '16 / 9', marginBottom: '1.5rem'}}>
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue