OpenAI 兼容模型接入检查：工具调用、结构化输出、超时与重试

"""A local fake OpenAI-compatible server whose behavior is selected by a named profile.

The profiles are *hypothetical* provider quirks used to exercise an acceptance suite. They do not
describe any real provider. Only the standard library is used; the server binds to 127.0.0.1.
"""
import json
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

PROFILES = {
    "standard": {},
    "no_tools": {"reject_tools": True},
    "bad_tool_args": {"tool_args_style": "python_dict"},
    "no_json_schema": {"reject_response_format": True},
    "slow": {"delay_seconds": 1.5},
    "flaky_429": {"rate_limit_first": 1},
}


class FakeProvider:
    def __init__(self, profile: str):
        self.cfg = PROFILES[profile]
        self.requests: list[dict] = []
        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 (for example after a timeout)
                except Exception as exc:  # a bug in this fake must surface as a readable 500
                    self._json(500, {"error": {"message": f"fake provider 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.requests.append(body)
                    seen = len(outer.requests)
                if self.path != "/v1/chat/completions":
                    return self._json(404, {"error": {"message": f"unknown path {self.path}"}})
                if outer.cfg.get("delay_seconds"):
                    time.sleep(outer.cfg["delay_seconds"])
                if seen <= outer.cfg.get("rate_limit_first", 0):
                    return self._json(429, {"error": {"message": "rate limited", "type": "rate_limit"}}, {"Retry-After": "0"})
                if outer.cfg.get("reject_tools") and body.get("tools"):
                    return self._json(400, {"error": {"message": "tools are not supported by this model", "type": "invalid_request_error"}})
                if outer.cfg.get("reject_response_format") and body.get("response_format"):
                    return self._json(400, {"error": {"message": "response_format is not supported", "type": "invalid_request_error"}})
                message = outer.build_message(body)
                if body.get("stream"):
                    return self._stream(message)
                return self._json(200, outer.completion(message))

            def _json(self, status: int, payload: dict, headers: dict | None = 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 k, v in (headers or {}).items():
                    self.send_header(k, v)
                self.end_headers()
                self.wfile.write(data)

            def _stream(self, message: dict):
                self.send_response(200)
                self.send_header("Content-Type", "text/event-stream")
                self.end_headers()
                text = message.get("content") or ""
                pieces = [text[i : i + 4] for i in range(0, len(text), 4)] or [""]
                for piece in pieces:
                    chunk = {"id": "c1", "object": "chat.completion.chunk", "created": 0, "model": "fake-model",
                             "choices": [{"index": 0, "delta": {"content": piece}, "finish_reason": None}]}
                    self.wfile.write(f"data: {json.dumps(chunk)}\n\n".encode())
                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 build_message(self, body: dict) -> dict:
        if body.get("tools"):
            function = body["tools"][0]["function"]
            name = function["name"]
            sample = {"city": "Shenzhen", "temperature_c": 26}
            args = {key: sample.get(key, "x") for key in function["parameters"].get("properties", {})}
            raw = json.dumps(args) if self.cfg.get("tool_args_style") != "python_dict" else str(args)
            return {"role": "assistant", "content": None,
                    "tool_calls": [{"id": "call_1", "type": "function", "function": {"name": name, "arguments": raw}}]}
        if body.get("response_format"):
            return {"role": "assistant", "content": json.dumps({"city": "Shenzhen", "temperature_c": 26})}
        return {"role": "assistant", "content": "Hello from the fake provider."}

    def completion(self, message: dict) -> dict:
        finish = "tool_calls" if message.get("tool_calls") else "stop"
        return {"id": "c1", "object": "chat.completion", "created": 0, "model": "fake-model",
                "choices": [{"index": 0, "message": message, "finish_reason": finish}],
                "usage": {"prompt_tokens": 5, "completion_tokens": 7, "total_tokens": 12}}

    def __enter__(self):
        self.thread.start()
        return self

    def __exit__(self, *exc):
        self.server.shutdown()
        self.server.server_close()

"""Acceptance suite for an OpenAI-compatible endpoint, run here against the fake provider.

usage: acceptance.py <profile>
Each check prints PASS or FAIL with the failure class. Exit code 1 if any check fails.
Point `base_url` at a real endpoint (and use a real key from your own environment) to run the
same suite against a provider you are evaluating.
"""
import os
import sys

os.environ.setdefault("NO_PROXY", "127.0.0.1,localhost")  # never route the local server through a proxy

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from pydantic import BaseModel

from fake_provider import FakeProvider


@tool
def get_weather(city: str) -> str:
    """Get the weather for a city."""
    return "sunny"


class Weather(BaseModel):
    city: str
    temperature_c: int


def check_invoke(llm):
    reply = llm.invoke("hi")
    assert reply.content, "empty content"


def check_stream(llm):
    text = "".join(chunk.content for chunk in llm.stream("hi"))
    assert text, "stream produced no text"


def check_tool_calling(llm):
    reply = llm.bind_tools([get_weather]).invoke("weather in Shenzhen?")
    if reply.invalid_tool_calls:
        bad = reply.invalid_tool_calls[0]
        raise AssertionError(f"invalid_tool_calls: {bad['name']} args={bad['args']!r}")
    assert reply.tool_calls, "no tool_calls returned"
    assert reply.tool_calls[0]["name"] == "get_weather"
    assert reply.tool_calls[0]["args"] == {"city": "Shenzhen"}


def check_structured_output(llm):
    result = llm.with_structured_output(Weather).invoke("weather in Shenzhen?")
    assert isinstance(result, Weather), f"unexpected result {result!r}"


def check_structured_output_tools(llm):
    result = llm.with_structured_output(Weather, method="function_calling").invoke("weather in Shenzhen?")
    assert isinstance(result, Weather), f"unexpected result {result!r}"


def check_usage(llm):
    reply = llm.invoke("hi")
    assert reply.usage_metadata and reply.usage_metadata["total_tokens"] > 0, "usage_metadata missing"


CHECKS = [
    ("invoke", check_invoke),
    ("stream", check_stream),
    ("tool_calling", check_tool_calling),
    ("structured_output", check_structured_output),
    ("structured_output_tools", check_structured_output_tools),
    ("usage_metadata", check_usage),
]


def describe(exc: Exception) -> str:
    """Prefer the provider's own error message over the SDK's full repr."""
    body = getattr(exc, "body", None)
    if isinstance(body, dict) and body.get("message"):
        return f"HTTP {getattr(exc, 'status_code', '?')} {body['message']}"
    return (str(exc).splitlines() or [""])[0]


def run_suite(base_url: str, api_key: str) -> int:
    llm = ChatOpenAI(model="fake-model", base_url=base_url, api_key=api_key, max_retries=0, timeout=0.5)
    failed = 0
    for name, check in CHECKS:
        try:
            check(llm)
            print(f"  PASS {name}")
        except Exception as exc:
            failed += 1
            print(f"  FAIL {name}: {type(exc).__name__}: {describe(exc)}")
    return failed


if __name__ == "__main__":
    profile = sys.argv[1] if len(sys.argv) > 1 else "standard"
    print(f"profile: {profile}")
    with FakeProvider(profile) as server:
        sys.exit(1 if run_suite(server.base_url, api_key="test-key-not-real") else 0)

profile: standard
  PASS invoke
  PASS stream
  PASS tool_calling
  PASS structured_output
  PASS structured_output_tools
  PASS usage_metadata

profile: no_tools
  PASS invoke
  PASS stream
  FAIL tool_calling: OpenAIInvalidRequestError: HTTP 400 tools are not supported by this model
  PASS structured_output
  FAIL structured_output_tools: OpenAIInvalidRequestError: HTTP 400 tools are not supported by this model
  PASS usage_metadata

profile: no_json_schema
  PASS invoke
  PASS stream
  PASS tool_calling
  FAIL structured_output: OpenAIInvalidRequestError: HTTP 400 response_format is not supported
  PASS structured_output_tools
  PASS usage_metadata

profile: bad_tool_args
  PASS invoke
  PASS stream
  FAIL tool_calling: AssertionError: invalid_tool_calls: get_weather args="{'city': 'Shenzhen'}"
  PASS structured_output
  FAIL structured_output_tools: AssertionError: unexpected result None
  PASS usage_metadata

profile: slow
  FAIL invoke: OpenAITimeoutError: Request timed out.
  FAIL stream: OpenAITimeoutError: Request timed out.
  FAIL tool_calling: OpenAITimeoutError: Request timed out.
  FAIL structured_output: OpenAITimeoutError: Request timed out.
  FAIL structured_output_tools: OpenAITimeoutError: Request timed out.
  FAIL usage_metadata: OpenAITimeoutError: Request timed out.

profile: flaky_429
  FAIL invoke: OpenAIRateLimitError: HTTP 429 rate limited
  PASS stream
  PASS tool_calling
  PASS structured_output
  PASS structured_output_tools
  PASS usage_metadata

"""When a provider returns tool arguments that are not valid JSON, an agent can end silently."""
import os

os.environ.setdefault("NO_PROXY", "127.0.0.1,localhost")

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

from fake_provider import FakeProvider

EXECUTED: list = []


@tool
def get_weather(city: str) -> str:
    """Get the weather for a city."""
    EXECUTED.append(city)
    return "sunny"


with FakeProvider("bad_tool_args") as server:
    llm = ChatOpenAI(model="fake-model", base_url=server.base_url, api_key="test-key-not-real", max_retries=0, timeout=2)
    agent = create_agent(llm, tools=[get_weather])
    result = agent.invoke({"messages": [("user", "What is the weather in Shenzhen?")]})

last = result["messages"][-1]
print(f"messages in the run: {len(result['messages'])}")
print(f"tool executed: {EXECUTED}")
print(f"final content: {last.content!r}")
print(f"tool_calls: {last.tool_calls}")
print(f"invalid_tool_calls: {[(c['name'], c['args']) for c in last.invalid_tool_calls]}")

messages in the run: 2
tool executed: []
final content: ''
tool_calls: []
invalid_tool_calls: [('get_weather', "{'city': 'Shenzhen'}")]

"""How many HTTP requests does one failed call cost? Compare max_retries=0 with the default."""
import os

os.environ.setdefault("NO_PROXY", "127.0.0.1,localhost")

from langchain_openai import ChatOpenAI

from acceptance import get_weather
from fake_provider import FakeProvider


def attempt(profile: str, label: str, with_tools: bool = False, **kwargs) -> None:
    with FakeProvider(profile) as server:
        llm = ChatOpenAI(model="fake-model", base_url=server.base_url, api_key="test-key-not-real", **kwargs)
        try:
            (llm.bind_tools([get_weather]) if with_tools else llm).invoke("hi")
            outcome = "succeeded"
        except Exception as exc:
            outcome = f"failed with {type(exc).__name__}"
        print(f"{label:<34} requests sent: {len(server.requests)}, {outcome}")


attempt("flaky_429", "429 once, max_retries=0", max_retries=0)
attempt("flaky_429", "429 once, default retries")
attempt("no_tools", "400 on tools, default retries", with_tools=True)
attempt("slow", "timeout 0.3s, max_retries=0", max_retries=0, timeout=0.3)
attempt("slow", "timeout 0.3s, max_retries=2", max_retries=2, timeout=0.3)

429 once, max_retries=0            requests sent: 1, failed with OpenAIRateLimitError
429 once, default retries          requests sent: 2, succeeded
400 on tools, default retries      requests sent: 1, failed with OpenAIInvalidRequestError
timeout 0.3s, max_retries=0        requests sent: 1, failed with OpenAITimeoutError
timeout 0.3s, max_retries=2        requests sent: 3, failed with OpenAITimeoutError