diff --git a/agentmain.py b/agentmain.py index 139bd0f4..6551b995 100644 --- a/agentmain.py +++ b/agentmain.py @@ -116,6 +116,17 @@ class GenericAgent: if self.handler is not None: self.handler.code_stop_signal.append(1) for sess in getattr(self.llmclient.backend, '_sessions', [self.llmclient.backend]): sess.should_stop = lambda: self.stop_sig # live read; cleared by run()'s finally + try: # wake a recv() blocked in another thread. Verified on Windows: shutdown()/close() do NOT + # wake it (makefile refcount defers real closesocket); _real_close() does -> ChunkedEncodingError + import socket as _socket + raw = sess.active_response.raw + fp = getattr(getattr(raw, '_fp', None), 'fp', None) # http.client response -> buffered socket file + sock = fp.raw._sock if fp else raw.connection.sock # SocketIO._sock (SSL-wrapped OK); fallback urllib3 conn + try: sock.shutdown(_socket.SHUT_RDWR) # for non-Windows semantics + except OSError: pass + try: sock._real_close() # CPython internal; bypasses refcount -> actual closesocket + except AttributeError: sock.close() + except Exception: pass try: sess.active_response.close() except Exception: pass diff --git a/llmcore.py b/llmcore.py index 01d6505f..136659ae 100644 --- a/llmcore.py +++ b/llmcore.py @@ -446,7 +446,15 @@ def _stream_with_retry(sess, url, headers, payload, parse_fn): try: ra = float((resp.headers or {}).get("retry-after")) except: ra = None return None if ra and ra > cap else max(0.5, ra or min(30.0, 3.0 * (2 ** attempt))) + def _stopped(): return getattr(sess, 'should_stop', None) and sess.should_stop() + def _sleep(d): # interruptible sleep; True if aborted + end = time.time() + d + while time.time() < end: + if _stopped(): return True + time.sleep(0.2) + return _stopped() for attempt in range(sess.max_retries + 1): + if _stopped(): return [] streamed = False STATS.update(t_start=time.time(), t_ttft=None) if not sess.stream: STATS['t_ttft'] = STATS['t_start'] @@ -459,7 +467,8 @@ def _stream_with_retry(sess, url, headers, payload, parse_fn): d = _delay(r, attempt) if r.status_code in _RETRYABLE and attempt < sess.max_retries else None if d is not None: print(f"[LLM Retry] HTTP {r.status_code}, retry in {d:.1f}s ({attempt+1}/{sess.max_retries+1})") - time.sleep(d); continue + if _sleep(d): return [] + continue try: body = r.text.strip()[:500] except: body = "" err = f"!!!Error: HTTP {r.status_code}" + (f" (retry-after > {cap:.0f}s)" if d is None and r.status_code in _RETRYABLE and attempt < sess.max_retries else "") + (f": {body}" if body else "") @@ -483,7 +492,8 @@ def _stream_with_retry(sess, url, headers, payload, parse_fn): if attempt < sess.max_retries: 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 + if _sleep(d): return [] + continue yield err; return [{"type": "text", "text": err}] except Exception as e: err = f"\n\n[!!! 流异常中断 {type(e).__name__}: {e} !!!]" if streamed else f"!!!Error: {type(e).__name__}: {e}"