40778c74e2
The aggregator request deliberately ends `user(task), user(guidance)` on iteration 1 of every turn (#113175) so the whole prefix stays byte-stable for the provider prompt cache. Strict-alternation chat templates (llama.cpp / vLLM Jinja templates, Mistral, some OpenRouter routes) 400 on that adjacency ("Conversation roles must alternate ..."), and the turn failed. Merging proactively for everyone was declined because it brings back the byte divergence #113175/#113784 removed and only moves the 400. Reactive, destination-scoped recovery instead: - error_classifier: new `FailoverReason.role_alternation` for the vendor alternation wordings (checked before the request-validation table since the body also carries `invalid_request_error`); same abort+fallback hints as format_error so non-MoA consumers behave exactly as before. - moa_alternation (new sibling): `merge_same_role_messages` (reuses the loop's `_merge_user_content`), `destination_key` (base_url|provider, model), `is_role_alternation_rejection`. - moa_loop._call_prepared_aggregator: on that 400, retry ONCE with the adjacent user turns merged, remember the destination on the facade for the session so later iterations pre-merge, never touch destinations that accepted the split shape. The trace records the messages actually sent. - docs: caching section explains the reactive merge. Live loopback (real call_llm -> SDK -> HTTP stub that 400s on same-role adjacency): before, iteration 1 fails with BadRequestError after 1 request; after, 2 requests (split -> 400 -> merged -> 200), next turn pre-merged in 1 request; the accepting-stub control sends byte-identical requests. Fixes #112358
65 lines
3.1 KiB
Python
65 lines
3.1 KiB
Python
"""Reactive same-role merge for the MoA aggregator request (#112358, last atom).
|
|
|
|
The aggregator request deliberately ends ``user(task), user(guidance)`` on iteration 1 of a
|
|
turn: the guidance is its own trailing message so every earlier message stays byte-stable and
|
|
the provider prefix cache keeps growing (``moa_loop._attach_reference_guidance``). Strict-
|
|
alternation chat templates (llama.cpp / vLLM Jinja templates, Mistral, some OpenRouter routes)
|
|
400 on that adjacency. Merging proactively for everyone would re-introduce the divergence the
|
|
split shape removed and only move the 400, so the merge is reactive and destination-scoped:
|
|
|
|
* a 400 classified ``FailoverReason.role_alternation`` → retry ONCE with adjacent same-role
|
|
messages merged;
|
|
* the destination (``base_url``/provider + model) is remembered on the facade for the rest of
|
|
the session, so later iterations pre-merge for it and never pay the 400 again;
|
|
* destinations that accepted the split shape are never touched — their prefix stays byte-stable.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def destination_key(runtime: dict[str, Any]) -> tuple[str, str]:
|
|
"""``(route, model)`` identity of an aggregator destination: the base_url when the slot
|
|
resolved one (two providers can share a model id), else the provider slug."""
|
|
route = str(runtime.get("base_url") or runtime.get("provider") or "").strip().rstrip("/")
|
|
return route, str(runtime.get("model") or "").strip()
|
|
|
|
|
|
def merge_same_role_messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
"""Return ``messages`` with adjacent user turns folded into one (the only same-role adjacency
|
|
the aggregator request produces); other rows are shared, never mutated. Returns the input
|
|
object itself when nothing merged so callers can detect a no-op."""
|
|
from agent.agent_runtime_helpers import _UNMERGEABLE, _merge_user_content
|
|
|
|
merged: list[dict[str, Any]] = []
|
|
changed = False
|
|
for message in messages:
|
|
prev = merged[-1] if merged else None
|
|
content: Any = _UNMERGEABLE
|
|
if prev is not None and prev.get("role") == "user" and message.get("role") == "user":
|
|
content = _merge_user_content(prev.get("content", ""), message.get("content", ""))
|
|
if content is _UNMERGEABLE:
|
|
merged.append(message)
|
|
continue
|
|
merged[-1] = {**prev, "content": content}
|
|
changed = True
|
|
return merged if changed else messages
|
|
|
|
|
|
def is_role_alternation_rejection(exc: Exception, runtime: dict[str, Any]) -> bool:
|
|
"""True when the aggregator destination rejected the request for adjacent same-role messages."""
|
|
from agent.error_classifier import FailoverReason, classify_api_error
|
|
|
|
try:
|
|
classified = classify_api_error(
|
|
exc, provider=str(runtime.get("provider") or ""), model=str(runtime.get("model") or ""),
|
|
base_url=str(runtime.get("base_url") or ""),
|
|
)
|
|
except Exception: # pragma: no cover - classification must never mask the original error
|
|
return False
|
|
return classified.reason is FailoverReason.role_alternation
|