diff --git a/aibox/README.md b/aibox/README.md index 361d764..236348a 100644 --- a/aibox/README.md +++ b/aibox/README.md @@ -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)）： diff --git a/aibox/llm_client.py b/aibox/llm_client.py index 2d2a790..9c8a5b1 100644 --- a/aibox/llm_client.py +++ b/aibox/llm_client.py @@ -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, diff --git a/aibox/tests/test_models_profile.py b/aibox/tests/test_models_profile.py index 6441e82..70f430e 100644 --- a/aibox/tests/test_models_profile.py +++ b/aibox/tests/test_models_profile.py @@ -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: