feat(aibox): stream_chat first-event failover + cancel-flag guard (L4b)
FailoverClient.stream_chat now walks the chain but only fails over while no event has been produced yet — replay before the first reasoning/content/tool_call event is invisible to the user. Once any event is out, failures propagate directly (replay would duplicate frontend output and corrupt tool-call accumulation). DeepSeekClient.stream_chat classifies non-200 responses: 429/5xx raise TransientLLMError (failover-eligible), auth/param 4xx keep raising RuntimeError. Chain exhaustion re-raises the last exception so the existing rollback/error handling in think_and_respond catches it; an all-cooling chain raises RuntimeError. Cancel-flag guard: LLMClient.cancel() sets _cancelled because the RequestException raised by force-closing the connection is indistinguishable from a real network fault. FailoverClient.cancel() is now defined explicitly (overriding __getattr__ delegation), broadcasts to every client in the chain and sets the wrapper flag; stream_chat/chat_completion reset it per call and suppress failover when it is set — otherwise interrupting would make the system hop to another vendor and keep generating against the user's intent. Tests: new TestL4bStreamFailover (12 cases: failover before first event on TransientLLMError/ConnectionError, no failover after first event, cancel never fails over on both paths, cancel broadcast, chain exhaustion, all-cooling, cooldown skip, success isolation, transient/ auth status classification). Full suite 342 passed.
This commit is contained in:
1 parent
3d7f5ca3cb
commit
d53eba77f0
3 files changed
+231
-9
No files matched your search
+4
-3
@@ -92,7 +92,7 @@ aibox/
|
||||
│ ├── prompt_tools.py # B 批:UpdateIdentity/UpdateSoul/SaveMemoryAndEndConversation
|
||||
│ ├── proxy_tools.py # B 批:SetProxy/GetProxy
|
||||
│ └── misc_tools.py # B 批:AskUserTool
|
||||
├── tests/ # 单元测试套件(330 项全过)
|
||||
├── tests/ # 单元测试套件(342 项全过)
|
||||
│ ├── conftest.py # sys.path 配置 + websockets 桩
|
||||
│ ├── test_basic_tools.py # A 批工具 43 项
|
||||
│ ├── test_b_tools.py # B 批工具 58 项
|
||||
@@ -553,7 +553,7 @@ AI 可以通过 `update_soul` 工具动态更新 `99_SOUL.md`,通过 `update_i
|
||||
| `timeout` | 请求超时秒数 |
|
||||
| `temperature` | 档案级默认采样温度(可选)。优先级:CLI/调用方显式传入 > 本字段 > 全局默认;reasoning 模型建议按厂商文档设置 |
|
||||
| `display_name` | 展示名(可选,缺省用档案名)。`client.name` 用于对话上下文中的 assistant 名,不再恒显示 "DeepSeek" |
|
||||
| `fallback` | 失败转移档案名列表(可选,如 `["kimi", "local"]`,一层不递归)。主档案瞬时故障(HTTP 429/5xx、连接失败/超时)时按序转移;认证/参数类 4xx **不转移**(配置问题响亮暴露);失败档案进入 60s 冷却期。当前仅 `chat_completion`(对话压缩/轻量子代理)生效,流式路径 L4b 支持 |
|
||||
| `fallback` | 失败转移档案名列表(可选,如 `["kimi", "local"]`,一层不递归)。主档案瞬时故障(HTTP 429/5xx、连接失败/超时)时按序转移;认证/参数类 4xx **不转移**(配置问题响亮暴露);失败档案进入 60s 冷却期。`chat_completion`(对话压缩/轻量子代理)无条件转移;`stream_chat`(主对话流式)仅**首事件前**转移(已输出内容/用户中断绝不转移) |
|
||||
|
||||
**选择优先级**:`--model <档案名>`(命中 `models` 节)→ `api.default_model` 指向档案 → 回退旧 `api` 扁平节。`--model` 为**进程级覆盖**:WebSocket 服务模式(主对话、每个会话实例、定时任务实例)与 `--once` 一次性模式均生效。
|
||||
|
||||
@@ -903,7 +903,8 @@ elif command == "/mycommand":
|
||||
- **L3(档案级温度)**:档案 `temperature` 字段作为默认采样温度,优先级:CLI/调用方显式传入 > 档案字段 > 全局默认(0.7/0.3),主对话/轻量子代理/compressor 全链路生效
|
||||
- **L5(思考流字段别名)**:`_parse_delta` 识别 `reasoning_content`/`reasoning`/`thinking`/`thinking_content` 四种常见思考字段(首个命中生效),其他厂商思考流不再静默丢失
|
||||
- ✅ **L4a(chat_completion 跨档案失败转移)**:档案新字段 `fallback`(档案名列表,一层不递归);`create_llm_client` 构造 `FailoverClient` 包装链(主档案 + fallback)。瞬时故障(HTTP 429/5xx 抛 `TransientLLMError`、连接失败/超时)按链转移,认证/参数类 4xx 不转移(配置问题响亮暴露);失败档案 60s 冷却;fallback 档案缺 Key 等环境问题跳过 + warning 不阻断主档案;配置结构错误(类型/不存在/自引用)启动期硬性拦截。当前覆盖对话压缩与轻量子代理链路,流式(`stream_chat`)转移属 L4b
|
||||
- ✅ **测试**:330 项全过(H1 隔离 4 + H2 校验 6 + H3 传导 6 + M2 提示 5 + M1 替换 2 + M3 契约 4 + L 批 16 + L4a 转移 15 项新增)
|
||||
- ✅ **L4b(stream_chat 首事件前转移 + 中断联动)**:流式仅在**未产出任何事件**(reasoning/content/tool_call)时转移——重放对用户不可见;已产出事件后失败直接上抛(防前端重复输出/工具增量错乱)。`LLMClient.cancel()` 置 `_cancelled` 标志,`FailoverClient.cancel()` 广播整条链并置包装器标志——用户中断引发的连接异常与真实故障不可区分,凭标志跳过转移(否则用户越中断、系统越换厂商重试);`chat_completion` 同受标志保护。链耗尽抛最后异常(既有回滚/报错处理接住),全链冷却抛 RuntimeError
|
||||
- ✅ **测试**:342 项全过(H1 隔离 4 + H2 校验 6 + H3 传导 6 + M2 提示 5 + M1 替换 2 + M3 契约 4 + L 批 16 + L4a 转移 15 + L4b 流式 12 项新增)
|
||||
|
||||
### v3.3
|
||||
- ✅ **安全审计与修复**(两轮审计报告见 [docs/reports/aibox_audit_20260817.md](../docs/reports/aibox_audit_20260817.md)):
|
||||
|
||||
+57
-6
@@ -47,6 +47,9 @@ class LLMClient(ABC):
|
||||
self.default_temperature = default_temperature
|
||||
# 当前活动的 HTTP 响应引用,供 cancel() 强制关闭底层连接以中断阻塞读取
|
||||
self._current_response = None
|
||||
# L4b:用户中断标志——cancel() 强关连接引发的异常与真实网络故障不可区分,
|
||||
# FailoverClient 据此标志跳过转移(否则用户越中断、系统越换厂商重试)
|
||||
self._cancelled = False
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
@@ -70,6 +73,7 @@ class LLMClient(ABC):
|
||||
|
||||
def cancel(self):
|
||||
"""中断当前流式请求。线程安全,可从任意线程调用。"""
|
||||
self._cancelled = True
|
||||
resp = self._current_response
|
||||
self._current_response = None
|
||||
if resp is not None:
|
||||
@@ -202,6 +206,10 @@ class DeepSeekClient(LLMClient):
|
||||
self._current_response = response
|
||||
|
||||
if response.status_code != 200:
|
||||
if response.status_code == 429 or response.status_code >= 500:
|
||||
# L4b:瞬时故障显式分类(此时尚未产出任何事件,可安全转移重放)
|
||||
raise TransientLLMError(
|
||||
f"LLM 流式请求瞬时失败: HTTP {response.status_code}: {response.text[:200]}")
|
||||
raise RuntimeError(
|
||||
f"HTTP {response.status_code}: {response.text}")
|
||||
|
||||
@@ -275,7 +283,10 @@ class FailoverClient:
|
||||
失败档案进入冷却期(cooldown_seconds,默认 60s),期间直接跳过,
|
||||
避免每次请求都撞刚失败的端点。链耗尽返回 None(与 chat_completion 失败契约一致)。
|
||||
|
||||
stream_chat 暂直接委托主 client(L4b 实现首 token 前转移)。
|
||||
L4b 范围——stream_chat 仅**首事件前**故障转移(连接失败/HTTP 429/5xx,
|
||||
尚未产出任何 reasoning/content/tool_call 事件,重放对用户不可见);
|
||||
已产出事件后失败直接上抛(重放会导致前端重复输出/工具增量错乱);
|
||||
用户中断(cancel() 置位)引发的连接异常绝不转移。
|
||||
"""
|
||||
|
||||
def __init__(self, clients: List, logger=None, cooldown_seconds: float = 60.0):
|
||||
@@ -283,6 +294,7 @@ class FailoverClient:
|
||||
self.logger = logger
|
||||
self.cooldown_seconds = cooldown_seconds
|
||||
self._fail_at = {} # client.name → 失败时刻(time.monotonic)
|
||||
self._cancelled = False
|
||||
|
||||
def __getattr__(self, item):
|
||||
# 未显式定义的属性/方法委托主 client(name/model/api_key/cancel 等);
|
||||
@@ -305,8 +317,21 @@ class FailoverClient:
|
||||
def _mark_fail(self, client):
|
||||
self._fail_at[client.name] = time.monotonic()
|
||||
|
||||
def cancel(self):
|
||||
"""L4b:中断链上全部 client(任一可能正在流式中)并置自身中断标志。
|
||||
|
||||
显式定义以覆盖 __getattr__ 委托——包装器必须同时通知整条链。
|
||||
"""
|
||||
self._cancelled = True
|
||||
for client in self._clients:
|
||||
try:
|
||||
client.cancel()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def chat_completion(self, messages: List[Dict], temperature: float = None,
|
||||
max_tokens: int = 1024, **kwargs):
|
||||
self._cancelled = False # 新一轮请求,清除上一轮中断标志
|
||||
last_exc = None
|
||||
tried = 0
|
||||
for client in self._clients:
|
||||
@@ -322,6 +347,8 @@ class FailoverClient:
|
||||
# None:认证/参数类 4xx 或响应异常——不可转移,直接透传
|
||||
return None
|
||||
except (TransientLLMError, requests.exceptions.RequestException) as e:
|
||||
if self._cancelled:
|
||||
return None # 用户中断引发的异常:不转移
|
||||
last_exc = e
|
||||
self._mark_fail(client)
|
||||
self._log("warning", f"模型档案 {client.name} 瞬时故障({e}),转移下一档案")
|
||||
@@ -332,11 +359,35 @@ class FailoverClient:
|
||||
max_tokens: int = 8192, tools: Optional[List[Dict]] = None,
|
||||
tool_choice: Optional[str] = None,
|
||||
reasoning_enabled: bool = False, **extra) -> Iterator[Dict]:
|
||||
# L4a:暂不转移(L4b 实现首 token 前转移),直接委托主 client
|
||||
yield from self._clients[0].stream_chat(
|
||||
messages, temperature=temperature, max_tokens=max_tokens,
|
||||
tools=tools, tool_choice=tool_choice,
|
||||
reasoning_enabled=reasoning_enabled, **extra)
|
||||
"""L4b:流式转移——仅首事件前故障可安全重放(对用户不可见)。"""
|
||||
self._cancelled = False # 新一轮请求,清除上一轮中断标志
|
||||
last_exc = None
|
||||
for client in self._clients:
|
||||
if self._cooling(client):
|
||||
self._log("info", f"模型档案 {client.name} 处于失败冷却期,跳过")
|
||||
continue
|
||||
produced_any = False
|
||||
try:
|
||||
for event in client.stream_chat(
|
||||
messages, temperature=temperature, max_tokens=max_tokens,
|
||||
tools=tools, tool_choice=tool_choice,
|
||||
reasoning_enabled=reasoning_enabled, **extra):
|
||||
produced_any = True
|
||||
yield event
|
||||
return # 自然结束(含 done 事件)视为成功
|
||||
except (TransientLLMError, requests.exceptions.RequestException) as e:
|
||||
if self._cancelled:
|
||||
raise # 用户中断(cancel 强关连接):绝不转移
|
||||
if produced_any:
|
||||
raise # 已向消费方产出事件,重放会重复输出/工具增量错乱
|
||||
last_exc = e
|
||||
self._mark_fail(client)
|
||||
self._log("warning",
|
||||
f"模型档案 {client.name} 流式瞬时故障(首事件前,{e}),转移下一档案")
|
||||
if last_exc is not None:
|
||||
raise last_exc # 链耗尽:抛最后异常,由既有错误处理(回滚/报错)接住
|
||||
raise RuntimeError(
|
||||
f"全部 {len(self._clients)} 个模型档案处于失败冷却期,暂不可用")
|
||||
|
||||
|
||||
def resolve_api_key(api_key: Optional[str], config=None,
|
||||
|
||||
@@ -869,6 +869,176 @@ class TestL4aChatCompletionFailover:
|
||||
assert kwargs["temperature"] == 0.5 and kwargs["max_tokens"] == 99
|
||||
|
||||
|
||||
# ==================== L4b:stream_chat 首事件前转移 + 中断联动 ====================
|
||||
|
||||
def _fake_stream(events=None, exc=None, before=None):
|
||||
"""构造伪 stream_chat 可调用:先 before(),逐个 yield events,最后抛 exc。"""
|
||||
def call(*args, **kwargs):
|
||||
if before is not None:
|
||||
before()
|
||||
for e in (events or []):
|
||||
yield e
|
||||
if exc is not None:
|
||||
raise exc
|
||||
return call
|
||||
|
||||
|
||||
class TestL4bStreamFailover:
|
||||
"""L4b:流式仅在首事件前转移;已产出事件/用户中断绝不转移。
|
||||
|
||||
- 转移窗口:连接建立失败、HTTP 429/5xx(TransientLLMError)——此时未产出任何事件
|
||||
- 不转移:产出首个事件后的任何失败(重放会前端重复/工具增量错乱)、cancel 引发的异常
|
||||
- cancel() 广播整条链并置标志;chat_completion 同样受标志保护
|
||||
"""
|
||||
|
||||
EVENTS = [{"type": "reasoning", "text": "think"},
|
||||
{"type": "content", "text": "hello"},
|
||||
{"type": "done"}]
|
||||
|
||||
def _client(self, name, stream=None, completion=None):
|
||||
c = mock.MagicMock()
|
||||
c.name = name
|
||||
if stream is not None:
|
||||
c.stream_chat = mock.MagicMock(side_effect=stream)
|
||||
if completion is not None:
|
||||
c.chat_completion = mock.MagicMock(side_effect=completion)
|
||||
return c
|
||||
|
||||
def _fc(self, primary, fallback):
|
||||
from llm_client import FailoverClient
|
||||
return FailoverClient([primary, fallback])
|
||||
|
||||
def test_failover_before_first_event(self):
|
||||
"""主档案未产出任何事件即瞬时失败 → 转移 fallback,事件完整来自 fallback"""
|
||||
from llm_client import TransientLLMError
|
||||
primary = self._client("deepseek",
|
||||
stream=_fake_stream(exc=TransientLLMError("HTTP 429")))
|
||||
fallback = self._client("kimi", stream=_fake_stream(events=self.EVENTS))
|
||||
fc = self._fc(primary, fallback)
|
||||
assert list(fc.stream_chat([{"role": "user", "content": "q"}])) == self.EVENTS
|
||||
assert fallback.stream_chat.called
|
||||
|
||||
def test_failover_on_connection_error_before_events(self):
|
||||
"""主档案连接失败(未产出事件)→ 转移成功"""
|
||||
primary = self._client("deepseek",
|
||||
stream=_fake_stream(exc=requests.exceptions.ConnectionError("boom")))
|
||||
fallback = self._client("kimi", stream=_fake_stream(events=self.EVENTS))
|
||||
fc = self._fc(primary, fallback)
|
||||
assert list(fc.stream_chat([{"role": "user", "content": "q"}])) == self.EVENTS
|
||||
|
||||
def test_no_failover_after_first_event(self):
|
||||
"""已产出 1 个事件后失败 → 直接上抛,不调 fallback(防前端重复输出)"""
|
||||
primary = self._client("deepseek",
|
||||
stream=_fake_stream(events=self.EVENTS[:1],
|
||||
exc=requests.exceptions.ConnectionError("mid")))
|
||||
fallback = self._client("kimi", stream=_fake_stream(events=self.EVENTS))
|
||||
fc = self._fc(primary, fallback)
|
||||
with pytest.raises(requests.exceptions.ConnectionError):
|
||||
list(fc.stream_chat([{"role": "user", "content": "q"}]))
|
||||
fallback.stream_chat.assert_not_called()
|
||||
|
||||
def test_cancel_never_fails_over_stream(self):
|
||||
"""用户中断(cancel 引发连接异常)→ 上抛原异常,绝不转移"""
|
||||
holder = {}
|
||||
|
||||
def do_cancel():
|
||||
holder["fc"].cancel()
|
||||
|
||||
primary = self._client("deepseek",
|
||||
stream=_fake_stream(before=do_cancel,
|
||||
exc=requests.exceptions.ConnectionError("cancelled")))
|
||||
fallback = self._client("kimi", stream=_fake_stream(events=self.EVENTS))
|
||||
fc = self._fc(primary, fallback)
|
||||
holder["fc"] = fc
|
||||
with pytest.raises(requests.exceptions.ConnectionError):
|
||||
list(fc.stream_chat([{"role": "user", "content": "q"}]))
|
||||
fallback.stream_chat.assert_not_called()
|
||||
assert fc._cancelled is True
|
||||
primary.cancel.assert_called_once() # 广播到主 client
|
||||
|
||||
def test_cancel_never_fails_over_chat_completion(self):
|
||||
"""chat_completion 期间的 cancel 引发异常 → 返回 None,不转移"""
|
||||
holder = {}
|
||||
|
||||
def raise_after_cancel(*a, **k):
|
||||
holder["fc"].cancel()
|
||||
raise requests.exceptions.ConnectionError("cancelled")
|
||||
|
||||
primary = self._client("deepseek", completion=raise_after_cancel)
|
||||
fallback = self._client("kimi", stream=_fake_stream(events=self.EVENTS))
|
||||
fc = self._fc(primary, fallback)
|
||||
holder["fc"] = fc
|
||||
assert fc.chat_completion([{"role": "user", "content": "q"}]) is None
|
||||
fallback.chat_completion.assert_not_called()
|
||||
|
||||
def test_cancel_broadcasts_to_whole_chain(self):
|
||||
"""fc.cancel() → 链上全部 client.cancel() 均被调用 + 自身置标志"""
|
||||
primary = self._client("deepseek")
|
||||
fallback = self._client("kimi")
|
||||
fc = self._fc(primary, fallback)
|
||||
fc.cancel()
|
||||
assert fc._cancelled is True
|
||||
primary.cancel.assert_called_once()
|
||||
fallback.cancel.assert_called_once()
|
||||
|
||||
def test_chain_exhausted_raises_last(self):
|
||||
"""全链首事件前瞬时失败 → 抛最后一个异常(既有错误处理接住)"""
|
||||
from llm_client import TransientLLMError
|
||||
primary = self._client("deepseek", stream=_fake_stream(exc=requests.exceptions.Timeout("t1")))
|
||||
fallback = self._client("kimi", stream=_fake_stream(exc=TransientLLMError("HTTP 500")))
|
||||
fc = self._fc(primary, fallback)
|
||||
with pytest.raises(TransientLLMError):
|
||||
list(fc.stream_chat([{"role": "user", "content": "q"}]))
|
||||
|
||||
def test_all_cooling_raises_runtime_error(self):
|
||||
"""全链处于冷却期 → RuntimeError(不发任何请求)"""
|
||||
import time as _time
|
||||
primary = self._client("deepseek", stream=_fake_stream(events=self.EVENTS))
|
||||
fallback = self._client("kimi", stream=_fake_stream(events=self.EVENTS))
|
||||
fc = self._fc(primary, fallback)
|
||||
fc._fail_at = {"deepseek": _time.monotonic(), "kimi": _time.monotonic()}
|
||||
with pytest.raises(RuntimeError, match="冷却期"):
|
||||
list(fc.stream_chat([{"role": "user", "content": "q"}]))
|
||||
primary.stream_chat.assert_not_called()
|
||||
|
||||
def test_cooldown_skips_failed_primary_stream(self):
|
||||
"""主档案冷却中 → 流式直接走 fallback"""
|
||||
import time as _time
|
||||
primary = self._client("deepseek", stream=_fake_stream(events=self.EVENTS))
|
||||
fallback = self._client("kimi", stream=_fake_stream(events=self.EVENTS))
|
||||
fc = self._fc(primary, fallback)
|
||||
fc._fail_at["deepseek"] = _time.monotonic()
|
||||
assert list(fc.stream_chat([{"role": "user", "content": "q"}])) == self.EVENTS
|
||||
primary.stream_chat.assert_not_called()
|
||||
|
||||
def test_success_does_not_touch_fallback(self):
|
||||
"""主档案成功(含 done 自然结束)→ fallback 从未被调"""
|
||||
primary = self._client("deepseek", stream=_fake_stream(events=self.EVENTS))
|
||||
fallback = self._client("kimi", stream=_fake_stream(events=self.EVENTS))
|
||||
fc = self._fc(primary, fallback)
|
||||
assert list(fc.stream_chat([{"role": "user", "content": "q"}])) == self.EVENTS
|
||||
fallback.stream_chat.assert_not_called()
|
||||
|
||||
def test_http_transient_status_raises_transient(self):
|
||||
"""裸 DeepSeekClient.stream_chat:429/503 → TransientLLMError(可转移信号)"""
|
||||
from llm_client import DeepSeekClient, TransientLLMError
|
||||
c = DeepSeekClient(api_key="k", model="m", base_url="https://x")
|
||||
for status in (429, 503):
|
||||
resp = mock.MagicMock(status_code=status, text="err")
|
||||
with mock.patch("llm_client.requests.post", return_value=resp):
|
||||
with pytest.raises(TransientLLMError):
|
||||
list(c.stream_chat([{"role": "user", "content": "hi"}]))
|
||||
|
||||
def test_http_auth_status_raises_runtime(self):
|
||||
"""裸 DeepSeekClient.stream_chat:401 → RuntimeError(非瞬时,不触发转移类型)"""
|
||||
from llm_client import DeepSeekClient
|
||||
c = DeepSeekClient(api_key="k", model="m", base_url="https://x")
|
||||
resp = mock.MagicMock(status_code=401, text="unauthorized")
|
||||
with mock.patch("llm_client.requests.post", return_value=resp):
|
||||
with pytest.raises(RuntimeError, match="401"):
|
||||
list(c.stream_chat([{"role": "user", "content": "hi"}]))
|
||||
|
||||
|
||||
# ==================== DeepSeekChat 档案级 reasoning 读取 ====================
|
||||
|
||||
class TestChatReasoningFlag:
|
||||
|
||||
Reference in new issue
Block a user