模型调用的延迟预算：默认无超时、退避、Retry-After 与重试叠加实测

"""A local OpenAI-compatible server with scriptable latency, error statuses and streaming stalls.
 
It exists to measure how long one model call can take when the provider misbehaves. Only the
standard library is used; the server binds to 127.0.0.1 and never sees a real credential.
"""
import json
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
 
class LatencyServer:
    """Behavior knobs (all optional):
 
    delay            seconds to wait before answering every request
    fail_first       list of (status, headers) used for the first N requests, then 200
    always_status    answer every request with this status (headers from always_headers)
    stream_chunks    number of content chunks in a streamed reply
    stream_interval  seconds between streamed chunks
    stall_after      stop sending after this many chunks (None = never stall)
    stall_seconds    how long the stall lasts before the stream resumes
    """
 
    def __init__(self, delay=0.0, fail_first=(), always_status=None, always_headers=None,
                 stream_chunks=5, stream_interval=0.0, stall_after=None, stall_seconds=0.0):
        self.delay = delay
        self.fail_first = list(fail_first)
        self.always_status = always_status
        self.always_headers = dict(always_headers or {})
        self.stream_chunks = stream_chunks
        self.stream_interval = stream_interval
        self.stall_after = stall_after
        self.stall_seconds = stall_seconds
        self.count = 0
        self._lock = threading.Lock()
        outer = self
 
        class Handler(BaseHTTPRequestHandler):
            def log_message(self, *args):  # keep stdout clean
                pass
 
            def do_POST(self):
                try:
                    self._handle()
                except (BrokenPipeError, ConnectionResetError):
                    pass  # the client gave up (timeout or cancellation)
                except Exception as exc:  # a bug in this fake must surface as a readable 500
                    self._json(500, {"error": {"message": f"latency server bug: {exc!r}"}})
 
            def _handle(self):
                length = int(self.headers.get("Content-Length", 0))
                body = json.loads(self.rfile.read(length) or b"{}")
                with outer._lock:
                    outer.count += 1
                    seen = outer.count
                if outer.delay:
                    time.sleep(outer.delay)
                if outer.always_status is not None:
                    return self._error(outer.always_status, outer.always_headers)
                if seen <= len(outer.fail_first):
                    status, headers = outer.fail_first[seen - 1]
                    return self._error(status, headers)
                if body.get("stream"):
                    return self._stream()
                return self._json(200, {
                    "id": "c1", "object": "chat.completion", "created": 0, "model": "fake-model",
                    "choices": [{"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}],
                    "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
                })
 
            def _error(self, status, headers):
                self._json(status, {"error": {"message": f"scripted status {status}", "type": "scripted"}}, headers)
 
            def _json(self, status, payload, headers=None):
                data = json.dumps(payload).encode()
                self.send_response(status)
                self.send_header("Content-Type", "application/json")
                self.send_header("Content-Length", str(len(data)))
                for key, value in (headers or {}).items():
                    self.send_header(key, value)
                self.end_headers()
                self.wfile.write(data)
 
            def _stream(self):
                self.send_response(200)
                self.send_header("Content-Type", "text/event-stream")
                self.end_headers()
                for index in range(outer.stream_chunks):
                    if outer.stall_after is not None and index == outer.stall_after:
                        time.sleep(outer.stall_seconds)
                    chunk = {"id": "c1", "object": "chat.completion.chunk", "created": 0, "model": "fake-model",
                             "choices": [{"index": 0, "delta": {"content": f"w{index} "}, "finish_reason": None}]}
                    self.wfile.write(f"data: {json.dumps(chunk)}\n\n".encode())
                    time.sleep(outer.stream_interval)
                done = {"id": "c1", "object": "chat.completion.chunk", "created": 0, "model": "fake-model",
                        "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]}
                self.wfile.write(f"data: {json.dumps(done)}\n\n".encode())
                self.wfile.write(b"data: [DONE]\n\n")
 
        self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
 
    @property
    def base_url(self) -> str:
        return f"http://127.0.0.1:{self.server.server_address[1]}/v1"
 
    def __enter__(self):
        self.thread.start()
        return self
 
    def __exit__(self, *exc):
        self.server.shutdown()
        self.server.server_close()

"""Shared helpers: build a client against the local server and time one call."""
import os
import time
 
os.environ.setdefault("NO_PROXY", "127.0.0.1,localhost")
 
from langchain_openai import ChatOpenAI
 
KEY = "placeholder-api-key-0001"
 
 
def make_llm(server, **kwargs) -> ChatOpenAI:
    return ChatOpenAI(model="fake-model", base_url=server.base_url, api_key=KEY, **kwargs)
 
 
def timed(call):
    """Run `call()` and return (elapsed_seconds, 'ok' or the exception class name)."""
    start = time.perf_counter()
    try:
        call()
        outcome = "ok"
    except Exception as exc:
        outcome = type(exc).__name__
    return time.perf_counter() - start, outcome
 
 
def half(seconds: float) -> float:
    """Round to the nearest half second so jitter does not make recorded output unstable."""
    return round(seconds * 2) / 2

"""What timeout and retry settings does a freshly built client really carry?"""
import openai
from langchain_openai import ChatOpenAI
 
KEY = "placeholder-api-key-0001"
 
 
def describe(label: str, sdk_client, lc_timeout="-", lc_retries="-") -> None:
    http_timeout = sdk_client._client.timeout
    print(f"{label}")
    print(f"  LangChain request_timeout / max_retries : {lc_timeout} / {lc_retries}")
    print(f"  SDK client.timeout / max_retries        : {sdk_client.timeout} / {sdk_client.max_retries}")
    print(f"  HTTP layer timeout                      : {http_timeout}")
 
 
describe("openai.OpenAI() with defaults", openai.OpenAI(api_key=KEY))
for label, kwargs in [
    ("ChatOpenAI() with defaults", {}),
    ("ChatOpenAI(timeout=30)", {"timeout": 30}),
    ("ChatOpenAI(max_retries=0)", {"max_retries": 0}),
    ("ChatOpenAI(timeout=(3, 30))", {"timeout": (3, 30)}),
]:
    llm = ChatOpenAI(model="fake-model", api_key=KEY, **kwargs)
    describe(label, llm.root_client, llm.request_timeout, llm.max_retries)

openai.OpenAI() with defaults
  LangChain request_timeout / max_retries : - / -
  SDK client.timeout / max_retries        : Timeout(connect=5.0, read=600, write=600, pool=600) / 2
  HTTP layer timeout                      : Timeout(connect=5.0, read=600, write=600, pool=600)
ChatOpenAI() with defaults
  LangChain request_timeout / max_retries : None / None
  SDK client.timeout / max_retries        : None / 2
  HTTP layer timeout                      : Timeout(timeout=None)
ChatOpenAI(timeout=30)
  LangChain request_timeout / max_retries : 30.0 / None
  SDK client.timeout / max_retries        : 30.0 / 2
  HTTP layer timeout                      : Timeout(timeout=30.0)
ChatOpenAI(max_retries=0)
  LangChain request_timeout / max_retries : None / 0
  SDK client.timeout / max_retries        : None / 0
  HTTP layer timeout                      : Timeout(timeout=None)
ChatOpenAI(timeout=(3, 30))
  LangChain request_timeout / max_retries : (3.0, 30.0) / None
  SDK client.timeout / max_retries        : (3.0, 30.0) / 2
  HTTP layer timeout                      : Timeout(connect=3.0, read=30.0, write=None, pool=None)

"""Worst-case latency of one failing model call, and a release gate built on it.
 
The model mirrors the OpenAI Python SDK retry loop observed in this lab (see retry_budget.py):
attempt i is cut off after `timeout`; between attempts the SDK sleeps min(0.5 * 2**k, 8) seconds
reduced by a random 0-25% jitter. A server-sent Retry-After is deliberately not modelled: it can
only replace the backoff, never shorten the timeout (see retry_after.py).
 
Usage:
    python budget.py                                   # print the reference table
    python budget.py --timeout 30 --retries 2 --limit 120   # gate: exit 1 when the worst case exceeds the limit
"""
import argparse
import math
import sys
 
INITIAL_DELAY = 0.5
MAX_DELAY = 8.0
JITTER = 0.25
 
 
def backoff_total(retries: int) -> float:
    """Nominal (no jitter) sleep time spent between the attempts of one SDK call."""
    return sum(min(INITIAL_DELAY * 2**k, MAX_DELAY) for k in range(retries))
 
 
def latency_range(timeout: float, retries: int, wrapper_retries: int = 0, wrapper_delay: float = 0.0):
    """Return (low, high) seconds for a call that fails on every attempt.
 
    `wrapper_retries` is an outer retry layer (for example ModelRetryMiddleware) that re-runs the
    whole SDK call, so the two layers multiply.
    """
    calls = 1 + wrapper_retries
    attempts = calls * (1 + retries)
    nominal = calls * backoff_total(retries)
    waiting = wrapper_retries * wrapper_delay
    low = attempts * timeout + nominal * (1 - JITTER) + waiting
    high = attempts * timeout + nominal + waiting
    return low, high
 
 
def fmt(seconds: float) -> str:
    return "unbounded" if math.isinf(seconds) else f"{seconds:.1f}s"
 
 
def parse_timeout(text: str) -> float:
    return math.inf if text.lower() == "none" else float(text)
 
 
def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--timeout", type=parse_timeout, help="per-attempt timeout in seconds, or 'none'")
    parser.add_argument("--retries", type=int, default=2, help="SDK max_retries (default 2)")
    parser.add_argument("--wrapper-retries", type=int, default=0, help="outer retry layer attempts")
    parser.add_argument("--wrapper-delay", type=float, default=0.0, help="sleep between outer attempts; use the largest one for a conservative bound")
    parser.add_argument("--limit", type=float, help="latency budget in seconds; enables gate mode")
    args = parser.parse_args()
 
    if args.limit is None:
        print(f"{'timeout':>9} {'max_retries':>12} {'attempts':>9} {'worst case (no Retry-After)':>30}")
        for timeout, retries in [(math.inf, 2), (600.0, 2), (30.0, 2), (30.0, 0), (10.0, 1), (5.0, 2), (5.0, 0)]:
            low, high = latency_range(timeout, retries)
            span = fmt(high) if math.isinf(high) else f"{fmt(low)} .. {fmt(high)}"
            shown = "none" if math.isinf(timeout) else f"{timeout:g}s"
            print(f"{shown:>9} {retries:>12} {1 + retries:>9} {span:>30}")
        return 0
 
    if args.timeout is None:
        parser.error("--timeout is required in gate mode")
    low, high = latency_range(args.timeout, args.retries, args.wrapper_retries, args.wrapper_delay)
    attempts = (1 + args.wrapper_retries) * (1 + args.retries)
    verdict = "PASS" if high <= args.limit else "FAIL"
    print(f"requests per failed call: {attempts}")
    print(f"worst case: {fmt(high)} (best of the worst: {fmt(low)})  limit: {fmt(args.limit)}  -> {verdict}")
    return 0 if verdict == "PASS" else 1
 
 
if __name__ == "__main__":
    sys.exit(main())

  timeout  max_retries  attempts    worst case (no Retry-After)
     none            2         3                      unbounded
     600s            2         3             1801.1s .. 1801.5s
      30s            2         3                 91.1s .. 91.5s
      30s            0         1                 30.0s .. 30.0s
      10s            1         2                 20.4s .. 20.5s
       5s            2         3                 16.1s .. 16.5s
       5s            0         1                   5.0s .. 5.0s

"""Measure how long a call that times out on every attempt really takes, and compare with budget.py."""
from budget import latency_range
from common import make_llm, timed
from latency_server import LatencyServer
 
TIMEOUT = 0.3
SERVER_DELAY = 1.0  # always longer than TIMEOUT, so every attempt times out
SLACK = 0.5  # scheduling overhead allowed above the model's upper bound
 
print(f"every attempt times out: timeout={TIMEOUT}s, server delay={SERVER_DELAY}s")
for retries in (0, 1, 2, 3):
    with LatencyServer(delay=SERVER_DELAY) as server:
        elapsed, outcome = timed(lambda: make_llm(server, timeout=TIMEOUT, max_retries=retries).invoke("hi"))
        sent = server.count
    low, high = latency_range(TIMEOUT, retries)
    naive = (1 + retries) * TIMEOUT
    print(
        f"max_retries={retries}  requests={sent}  outcome={outcome}  "
        f"naive (1+retries)*timeout={naive:.1f}s  model={low:.1f}..{high:.1f}s  "
        f"measured within model: {low <= elapsed <= high + SLACK}  "
        f"measured above naive by more than 0.3s: {elapsed - naive > 0.3}"
    )

every attempt times out: timeout=0.3s, server delay=1.0s
max_retries=0  requests=1  outcome=OpenAITimeoutError  naive (1+retries)*timeout=0.3s  model=0.3..0.3s  measured within model: True  measured above naive by more than 0.3s: False
max_retries=1  requests=2  outcome=OpenAITimeoutError  naive (1+retries)*timeout=0.6s  model=1.0..1.1s  measured within model: True  measured above naive by more than 0.3s: True
max_retries=2  requests=3  outcome=OpenAITimeoutError  naive (1+retries)*timeout=0.9s  model=2.0..2.4s  measured within model: True  measured above naive by more than 0.3s: True
max_retries=3  requests=4  outcome=OpenAITimeoutError  naive (1+retries)*timeout=1.2s  model=3.8..4.7s  measured within model: True  measured above naive by more than 0.3s: True

"""How do server headers change retry timing? One scripted failure, then success."""
from common import half, make_llm, timed
from latency_server import LatencyServer
 
CASES = [
    ("429, no header", 429, {}),
    ("429, Retry-After: 0", 429, {"Retry-After": "0"}),
    ("429, Retry-After: 2", 429, {"Retry-After": "2"}),
    ("429, retry-after-ms: 1500", 429, {"retry-after-ms": "1500"}),
    ("429, Retry-After: 130", 429, {"Retry-After": "130"}),
    ("503, x-should-retry: false", 503, {"x-should-retry": "false"}),
    ("400, x-should-retry: true", 400, {"x-should-retry": "true"}),
]
 
print("one scripted failure, then a normal 200 reply; client uses timeout=5, default max_retries")
print(f"{'scripted failure':<30} {'requests':>8} {'outcome':<22} {'elapsed (to 0.5s)':>17}")
for label, status, headers in CASES:
    with LatencyServer(fail_first=[(status, headers)]) as server:
        elapsed, outcome = timed(lambda: make_llm(server, timeout=5).invoke("hi"))
        sent = server.count
    print(f"{label:<30} {sent:>8} {outcome:<22} {half(elapsed):>17.1f}")

one scripted failure, then a normal 200 reply; client uses timeout=5, default max_retries
scripted failure               requests outcome                elapsed (to 0.5s)
429, no header                        2 ok                                   0.5
429, Retry-After: 0                   2 ok                                   0.5
429, Retry-After: 2                   2 ok                                   2.0
429, retry-after-ms: 1500             2 ok                                   1.5
429, Retry-After: 130                 1 OpenAIRateLimitError                 0.0
503, x-should-retry: false            1 OpenAIAPIError                       0.0
400, x-should-retry: true             2 ok                                   0.5

"""Is `timeout` a cap on total time? Compare a plain call with streamed replies."""
import time
 
from common import half, make_llm, timed
from latency_server import LatencyServer
 
TIMEOUT = 0.5
 
 
def drain(llm, sink: list) -> None:
    for chunk in llm.stream("hi"):
        sink.append(chunk.content)
 
 
print(f"client timeout={TIMEOUT}s; max_retries=0 for the plain call, 2 for the streams")
 
with LatencyServer(delay=1.0) as server:
    _, outcome = timed(lambda: make_llm(server, timeout=TIMEOUT, max_retries=0).invoke("hi"))
    print(f"plain call, server needs 1.0s before replying   -> {outcome}, requests={server.count}")
 
with LatencyServer(stream_chunks=10, stream_interval=0.2) as server:
    received: list = []
    elapsed, outcome = timed(lambda: drain(make_llm(server, timeout=TIMEOUT, max_retries=2), received))
    print(
        f"stream, 10 chunks 0.2s apart (2.0s in total)    -> {outcome}, chunks={len([c for c in received if c])}, "
        f"requests={server.count}, elapsed~{half(elapsed):.1f}s, longer than timeout: {elapsed > TIMEOUT}"
    )
 
with LatencyServer(stream_chunks=10, stream_interval=0.05, stall_after=3, stall_seconds=2.0) as server:
    received = []
    _, outcome = timed(lambda: drain(make_llm(server, timeout=TIMEOUT, max_retries=2), received))
    time.sleep(0.5)  # give a hypothetical retry the chance to show up
    print(
        f"stream, stalls for 2.0s after 3 chunks          -> {outcome}, chunks kept={len([c for c in received if c])}, "
        f"requests={server.count}"
    )

client timeout=0.5s; max_retries=0 for the plain call, 2 for the streams
plain call, server needs 1.0s before replying   -> OpenAITimeoutError, requests=1
stream, 10 chunks 0.2s apart (2.0s in total)    -> ok, chunks=10, requests=1, elapsed~2.0s, longer than timeout: True
stream, stalls for 2.0s after 3 chunks          -> OpenAITimeoutError, chunks kept=3, requests=1

"""timeout does not bound the whole call. Does an outer deadline stop the retries too?"""
import asyncio
import time
 
from common import half, make_llm
from latency_server import LatencyServer
 
TIMEOUT = 0.3
RETRIES = 4
DEADLINE = 1.5
SERVER_DELAY = 1.0  # every attempt times out
 
 
async def main() -> None:
    with LatencyServer(delay=SERVER_DELAY) as server:
        llm = make_llm(server, timeout=TIMEOUT, max_retries=RETRIES)
        start = time.perf_counter()
        try:
            await asyncio.wait_for(llm.ainvoke("hi"), timeout=DEADLINE)
            outcome = "ok"
        except asyncio.TimeoutError:
            outcome = "deadline hit"
        elapsed = time.perf_counter() - start
        at_deadline = server.count
        await asyncio.sleep(2.5)  # long enough for several more attempts if the retry loop were still alive
        after = server.count
    print(f"timeout={TIMEOUT}s, max_retries={RETRIES}, outer deadline={DEADLINE}s")
    print(f"outcome: {outcome}, stopped at ~{half(elapsed):.1f}s")
    print(f"requests sent by the deadline: {at_deadline}")
    print(f"requests sent 2.5s after cancellation: {after} (retry loop stopped: {after == at_deadline})")
 
 
asyncio.run(main())

timeout=0.3s, max_retries=4, outer deadline=1.5s
outcome: deadline hit, stopped at ~1.5s
requests sent by the deadline: 2
requests sent 2.5s after cancellation: 2 (retry loop stopped: True)

"""Two retry layers multiply: the SDK inside ChatOpenAI and ModelRetryMiddleware around it."""
from budget import latency_range
from common import make_llm, timed
from langchain.agents import create_agent
from langchain.agents.middleware import ModelRetryMiddleware
from langchain_core.messages import HumanMessage
from latency_server import LatencyServer
 
ASK = {"messages": [HumanMessage("hi")]}
SLACK = 0.5  # scheduling overhead allowed above the model's upper bound
 
print("the provider answers 503 to every request at once; one agent call, no tools; the middleware adds no delay")
print(f"{'SDK retries':>11} {'middleware retries':>19} {'requests':>9} {'(1+sdk)*(1+mw)':>15} {'backoff model':>15} {'within model':>13}")
for sdk_retries, mw_retries in [(0, 0), (2, 0), (0, 2), (2, 2)]:
    with LatencyServer(always_status=503) as server:
        llm = make_llm(server, timeout=5, max_retries=sdk_retries)
        middleware = [ModelRetryMiddleware(max_retries=mw_retries, on_failure="error", initial_delay=0.0, jitter=False)]
        agent = create_agent(llm, tools=[], middleware=middleware)
        elapsed, outcome = timed(lambda: agent.invoke(ASK))
        sent = server.count
    expected = (1 + sdk_retries) * (1 + mw_retries)
    low, high = latency_range(0.0, sdk_retries, mw_retries)  # failures are instant, so only backoff remains
    print(
        f"{sdk_retries:>11} {mw_retries:>19} {sent:>9} {expected:>15} "
        f"{low:>6.1f}..{high:.1f}s {str(low <= elapsed <= high + SLACK):>13}"
    )

the provider answers 503 to every request at once; one agent call, no tools; the middleware adds no delay
SDK retries  middleware retries  requests  (1+sdk)*(1+mw)   backoff model  within model
          0                   0         1               1    0.0..0.0s          True
          2                   0         3               3    1.1..1.5s          True
          0                   2         3               3    0.0..0.0s          True
          2                   2         9               9    3.4..4.5s          True

requests per failed call: 3
worst case: unbounded (best of the worst: unbounded)  limit: 60.0s  -> FAIL

requests per failed call: 9
worst case: 94.5s (best of the worst: 93.4s)  limit: 60.0s  -> FAIL