From 7070880b4efb4e77de2e910cf26f1db476c19357 Mon Sep 17 00:00:00 2001 From: Liang Jiaqing Date: Tue, 28 Apr 2026 17:54:22 +0800 Subject: [PATCH] refactor: extract _stream_with_retry, MixinSession failover on stream abort --- agentmain.py | 2 +- frontends/qtapp.py | 2 +- llmcore.py | 128 ++++++++++++++++++++++----------------------- 3 files changed, 64 insertions(+), 68 deletions(-) diff --git a/agentmain.py b/agentmain.py index 6cb731c4..1a7eb25f 100644 --- a/agentmain.py +++ b/agentmain.py @@ -72,7 +72,7 @@ class GeneraticAgent: mixin = MixinSession(llm_sessions, s['mixin_cfg']) if isinstance(mixin._sessions[0], (NativeClaudeSession, NativeOAISession)): llm_sessions[i] = NativeToolClient(mixin) else: llm_sessions[i] = ToolClient(mixin) - except Exception as e: print(f'[WARN] Failed to init MixinSession with cfg {s["mixin_cfg"]}: {e}') + except Exception as e: print(f'\n\n\n[ERROR] Failed to init MixinSession with cfg {s["mixin_cfg"]}: {e}!!!\n\n') self.llmclients = llm_sessions self.llmclient = self.llmclients[self.llm_no%len(self.llmclients)] if oldhistory: self.llmclient.backend.history = oldhistory diff --git a/frontends/qtapp.py b/frontends/qtapp.py index 3d5de073..a4d695bd 100644 --- a/frontends/qtapp.py +++ b/frontends/qtapp.py @@ -1564,7 +1564,7 @@ class ChatPanel(QWidget): def _check_backend(self, idx: int, backend): ok = False try: - reply = backend.ask("你好", stream=False) + reply = backend.ask("你好") # 兼容生成器函数（NativeClaudeSession.ask是生成器） if hasattr(reply, '__iter__') and not isinstance(reply, str): reply = ''.join(str(b) for b in reply if isinstance(b, str)) diff --git a/llmcore.py b/llmcore.py index 6711c638..982954c0 100644 --- a/llmcore.py +++ b/llmcore.py @@ -107,6 +107,14 @@ def auto_make_url(base, path): if b.endswith(p): return b return f"{b}/{p}" if re.search(r'/v\d+(/|$)', b) else f"{b}/v1/{p}" +def _parse_claude_json(data): + content_blocks = data.get("content", []) + _record_usage(data.get("usage", {}), "messages") + for b in content_blocks: + if b.get("type") == "text": yield b.get("text", "") + elif b.get("type") == "thinking": yield "" + return content_blocks + def _parse_claude_sse(resp_lines): """Parse Anthropic SSE stream. Yields text chunks, returns list[content_block].""" content_blocks = []; current_block = None; tool_json_buf = "" @@ -175,7 +183,6 @@ def _parse_claude_sse(resp_lines): content_blocks.append({"type": "text", "text": warn}); yield warn return content_blocks - def _try_parse_tool_args(raw): """Parse tool args string; split concatenated JSON objects like {..}{..} if needed. Returns list of parsed dicts.""" @@ -342,9 +349,43 @@ def _stamp_oai_cache_markers(messages, model): c = list(c); c[-1] = dict(c[-1], cache_control={'type': 'ephemeral'}) messages[idx] = {**messages[idx], 'content': c} +def _stream_with_retry(sess, url, headers, payload, parse_fn): + _RETRYABLE = {408, 409, 425, 429, 500, 502, 503, 504, 529} + def _delay(resp, attempt): + try: ra = float((resp.headers or {}).get("retry-after")) + except: ra = None + return max(0.5, ra if ra is not None else min(30.0, 1.5 * (2 ** attempt))) + for attempt in range(sess.max_retries + 1): + streamed = False + try: + with requests.post(url, headers=headers, json=payload, stream=sess.stream, + timeout=(sess.connect_timeout, sess.read_timeout), proxies=sess.proxies, verify=sess.verify) as r: + if r.status_code >= 400: + if r.status_code in _RETRYABLE and attempt < sess.max_retries: + d = _delay(r, attempt) + print(f"[LLM Retry] HTTP {r.status_code}, retry in {d:.1f}s ({attempt+1}/{sess.max_retries+1})") + time.sleep(d); continue + try: body = r.text.strip()[:500] + except: body = "" + err = f"!!!Error: HTTP {r.status_code}" + (f": {body}" if body else "") + yield err; return [{"type": "text", "text": err}] + gen = parse_fn(r) + try: + while True: streamed = True; yield next(gen) + except StopIteration as e: return e.value or [] + except (requests.Timeout, requests.ConnectionError) as e: + if attempt < sess.max_retries and not streamed: + d = _delay(None, attempt) + print(f"[LLM Retry] {type(e).__name__}, retry in {d:.1f}s ({attempt+1}/{sess.max_retries+1})") + time.sleep(d); continue + err = f"!!!Error: {type(e).__name__}" + yield err; return [{"type": "text", "text": err}] + except Exception as e: + err = f"!!!Error: {type(e).__name__}: {e}" + yield err; return [{"type": "text", "text": err}] + def _openai_stream(sess, messages): - """Shared OpenAI-compatible streaming request with retry. Yields text chunks, returns list[content_block].""" - model, api_mode, max_retries = sess.model, sess.api_mode, sess.max_retries + model, api_mode = sess.model, sess.api_mode ml = model.lower() temperature = sess.temperature if 'kimi' in ml or 'moonshot' in ml: temperature = 1 @@ -368,41 +409,8 @@ def _openai_stream(sess, messages): tools = getattr(sess, 'tools', None) if tools: payload["tools"] = _prepare_oai_tools(tools, api_mode) if sess.service_tier: payload["service_tier"] = sess.service_tier - RETRYABLE = {408, 409, 425, 429, 500, 502, 503, 504, 529} - def _delay(resp, attempt): - try: ra = float((resp.headers or {}).get("retry-after")) - except: ra = None - return max(0.5, ra if ra is not None else min(30.0, 1.5 * (2 ** attempt))) - for attempt in range(max_retries + 1): - streamed = False - try: - with requests.post(url, headers=headers, json=payload, stream=sess.stream, - timeout=(sess.connect_timeout, sess.read_timeout), proxies=sess.proxies, verify=sess.verify) as r: - if r.status_code >= 400: - if r.status_code in RETRYABLE and attempt < max_retries: - d = _delay(r, attempt) - print(f"[LLM Retry] HTTP {r.status_code}, retry in {d:.1f}s ({attempt+1}/{max_retries+1})") - time.sleep(d); continue - body = "" - try: body = r.text.strip()[:500] - except: pass - err = f"!!!Error: HTTP {r.status_code}" + (f": {body}" if body else "") - yield err; return [{"type": "text", "text": err}] - gen = _parse_openai_sse(r.iter_lines(), api_mode) if sess.stream else _parse_openai_json(r.json(), api_mode) - try: - while True: streamed = True; yield next(gen) - except StopIteration as e: - return e.value or [] - except (requests.Timeout, requests.ConnectionError) as e: - if attempt < max_retries and not streamed: - d = _delay(None, attempt) - print(f"[LLM Retry] {type(e).__name__}, retry in {d:.1f}s ({attempt+1}/{max_retries+1})") - time.sleep(d); continue - err = f"!!!Error: {type(e).__name__}" - yield err; return [{"type": "text", "text": err}] - except Exception as e: - err = f"!!!Error: {type(e).__name__}: {e}" - yield err; return [{"type": "text", "text": err}] + parse_fn = (lambda r: _parse_openai_sse(r.iter_lines(), api_mode)) if sess.stream else (lambda r: _parse_openai_json(r.json(), api_mode)) + return (yield from _stream_with_retry(sess, url, headers, payload, parse_fn)) def _prepare_oai_tools(tools, api_mode="chat_completions"): if api_mode == "responses": @@ -510,7 +518,7 @@ class BaseSession: proxy = cfg.get('proxy') self.proxies = {"http": proxy, "https": proxy} if proxy else None self.verify = cfg.get('verify', True) - self.max_retries = max(0, int(cfg.get('max_retries', 1))) + self.max_retries = max(0, int(cfg.get('max_retries', 4))) self.stream = cfg.get('stream', True) default_ct, default_rt = (5, 30) if self.stream else (10, 240) self.connect_timeout = max(1, int(cfg.get('timeout', default_ct))) @@ -538,7 +546,7 @@ class BaseSession: effort = {'low': 'low', 'medium': 'medium', 'high': 'high', 'xhigh': 'max'}.get(self.reasoning_effort) if effort: payload["output_config"] = {"effort": effort} else: print(f"[WARN] reasoning_effort {self.reasoning_effort!r} is unsupported for Claude output_config.effort, ignored.") - def ask(self, prompt, stream=False): + def ask(self, prompt): def _ask_gen(): with self.lock: self.history.append({"role": "user", "content": [{"type": "text", "text": prompt}]}) @@ -555,7 +563,7 @@ class BaseSession: tu = {'name': block.get('name', ''), 'arguments': block.get('input', {})} yield f'{json.dumps(tu, ensure_ascii=False)}' if not content.startswith("!!!Error:"): self.history.append({"role": "assistant", "content": [{"type": "text", "text": content}]}) - return _ask_gen() if stream else ''.join(list(_ask_gen())) + return _ask_gen() if self.stream else ''.join(list(_ask_gen())) def _keep_claude_block(b): return not isinstance(b, dict) or b.get("type") != "thinking" or b.get("signature") def _drop_unsigned_thinking(messages): @@ -579,17 +587,13 @@ class ClaudeSession(BaseSession): def raw_ask(self, messages): if self.max_tokens is None: self.max_tokens = 8192 headers = {"x-api-key": self.api_key, "Content-Type": "application/json", "anthropic-version": "2023-06-01", "anthropic-beta": "prompt-caching-2024-07-31"} - payload = {"model": self.model, "messages": messages, "max_tokens": self.max_tokens, "stream": True} + payload = {"model": self.model, "messages": messages, "max_tokens": self.max_tokens, "stream": self.stream} if self.temperature != 1: payload["temperature"] = self.temperature self._apply_claude_thinking(payload) if self.system: payload["system"] = [{"type": "text", "text": self.system, "cache_control": {"type": "persistent"}}] - try: - with requests.post(auto_make_url(self.api_base, "messages"), headers=headers, json=payload, stream=True, timeout=(self.connect_timeout, self.read_timeout), verify=self.verify) as r: - if r.status_code != 200: raise Exception(f"HTTP {r.status_code} {r.content.decode('utf-8', errors='replace')[:500]}") - return (yield from _parse_claude_sse(r.iter_lines())) or [] - except Exception as e: - yield (err := f"!!!Error: {e}") - return [{"type": "text", "text": err}] + url = auto_make_url(self.api_base, "messages") + parse_fn = (lambda r: _parse_claude_sse(r.iter_lines())) if self.stream else (lambda r: _parse_claude_json(r.json())) + return (yield from _stream_with_retry(self, url, headers, payload, parse_fn)) def make_messages(self, raw_list): msgs = _drop_unsigned_thinking([{"role": m['role'], "content": list(m['content'])} for m in raw_list]) user_idxs = [i for i, m in enumerate(msgs) if m['role'] == 'user'] @@ -656,20 +660,9 @@ class NativeClaudeSession(BaseSession): for idx in user_idxs[-2:]: messages[idx] = {**messages[idx], "content": list(messages[idx]["content"])} messages[idx]["content"][-1] = dict(messages[idx]["content"][-1], cache_control={"type": "ephemeral"}) - try: - with requests.post(auto_make_url(self.api_base, "messages")+'?beta=true', headers=headers, json=payload, stream=self.stream, timeout=(self.connect_timeout, self.read_timeout), verify=self.verify) as resp: - if resp.status_code != 200: raise Exception(f"HTTP {resp.status_code} {resp.content.decode('utf-8', errors='replace')[:500]}") - if self.stream: return (yield from _parse_claude_sse(resp.iter_lines())) or [] - else: - data = resp.json(); content_blocks = data.get("content", []) - _record_usage(data.get("usage", {}), "messages") - for b in content_blocks: - if b.get("type") == "text": yield b.get("text", "") - elif b.get("type") == "thinking": yield "" - return content_blocks - except Exception as e: - yield (err := f"!!!Error: {e}") - return [{"type": "text", "text": err}] + url = auto_make_url(self.api_base, "messages") + '?beta=true' + parse_fn = (lambda r: _parse_claude_sse(r.iter_lines())) if self.stream else (lambda r: _parse_claude_json(r.json())) + return (yield from _stream_with_retry(self, url, headers, payload, parse_fn)) def ask(self, msg): assert type(msg) is dict @@ -714,7 +707,6 @@ def openai_tools_to_claude(tools): 'input_schema': fn.get('parameters', {'type': 'object', 'properties': {}})}) return result - class MockFunction: def __init__(self, name, arguments): self.name, self.arguments = name, arguments @@ -743,7 +735,7 @@ class ToolClient: full_prompt = self._build_protocol_prompt(messages, tools) print("Full prompt length:", len(full_prompt), 'chars') prompt_log = full_prompt - gen = self.backend.ask(full_prompt, stream=True) + gen = self.backend.ask(full_prompt) _write_llm_log('Prompt', prompt_log) raw_text = ''; summarytag = '[NextWillSummary]' for chunk in gen: @@ -915,7 +907,8 @@ class MixinSession: groups = {is_native(s) for s in self._sessions} assert len(groups) == 1, f"MixinSession: sessions must be in same group (Native or non-Native), got {[type(s).__name__ for s in self._sessions]}" self.name = '|'.join(s.name for s in self._sessions) - import copy; self._sessions[0] = copy.copy(self._sessions[0]) + import copy; self._sessions = [copy.copy(s) for s in self._sessions] + for s in self._sessions: s.max_retries = 0 self._orig_raw_asks = [s.raw_ask for s in self._sessions] self._sessions[0].raw_ask = self._raw_ask self.model = getattr(self._sessions[0], 'model', None) @@ -950,6 +943,9 @@ class MixinSession: is_err = test_error(last_chunk) if not is_err: if attempt > 0: self._cur_idx = idx; self._switched_at = time.time() + elif isinstance(last_chunk, str) and '[!!! 流异常中断' in last_chunk and n > 1: + self._cur_idx = (idx + 1) % n; self._switched_at = time.time() + print(f'[MixinSession] Partial failure, next call → s{self._cur_idx} ({self._sessions[self._cur_idx].name})') return return_val if attempt >= self._retries: yield last_chunk; return return_val