迁移到 GPT-5.x 而不搞坏 GPT-4：一套实用的向后兼容迁移手册

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261

"""
Model compatibility helper for GPT-5.x with GPT-4 backward compatibility.

This module centralises the parameter translation needed to talk to the
"reasoning" generation of OpenAI / Azure OpenAI models (GPT-5, GPT-5.1,
o1, o3, o4) while keeping older deployments (gpt-4, gpt-4o, gpt-4-32k,
gpt-3.5-turbo, etc.) working unchanged.
"""

from __future__ import annotations

import logging
import os
import re
from typing import Any, Dict, Iterable, Mapping, Optional

# ---------------------------------------------------------------------------
# Family detection
# ---------------------------------------------------------------------------

_REASONING_PATTERNS = (
    # gpt-5, gpt5, gpt-5.1, gpt_5, GPT 5, gpt5mini-prod-eu, ...
    re.compile(r"(?i)(^|[^a-z0-9])gpt[-_ ]?5(\.\d+)?([^0-9]|$)"),
    # o1, o3, o4, o1-mini, o3-preview ...
    re.compile(r"(?i)(^|[^a-z0-9])o[134](-mini|-preview)?([^a-z0-9]|$)"),
)

_LEGACY_PATTERNS = (
    re.compile(r"(?i)gpt[-_ ]?4o"),
    re.compile(r"(?i)gpt[-_ ]?4(?!\d)"),
    re.compile(r"(?i)gpt[-_ ]?4[-_ ]?32k"),
    re.compile(r"(?i)gpt[-_ ]?3\.?5"),
    re.compile(r"(?i)gpt[-_ ]?35"),
)

def get_model_family(model_or_deployment: Optional[str]) -> str:
    """Return ``"reasoning"`` for GPT-5.x / o-series, ``"legacy"`` otherwise.

    Honours an ``OPENAI_MODEL_FAMILY`` env-var override for deployments whose
    user-defined name does not embed the model family (e.g. ``prod-default``).
    """
    override = (os.getenv("OPENAI_MODEL_FAMILY") or "").strip().lower()
    if override in {"reasoning", "gpt-5", "gpt5", "gpt-5.1", "o-series", "o1", "o3"}:
        return "reasoning"
    if override in {"legacy", "gpt-4", "gpt4", "gpt-3.5", "gpt35", "chat"}:
        return "legacy"

    name = (model_or_deployment or "").strip()
    if not name:
        # Fail closed: when we don't know, assume legacy so old code keeps
        # working. Misclassifying a reasoning deployment as legacy fails fast
        # with a clear "Unsupported parameter" 400; the reverse silently
        # drops parameters the caller expected.
        return "legacy"

    for pat in _REASONING_PATTERNS:
        if pat.search(name):
            return "reasoning"
    for pat in _LEGACY_PATTERNS:
        if pat.search(name):
            return "legacy"
    return "legacy"

def is_reasoning_model(model_or_deployment: Optional[str]) -> bool:
    return get_model_family(model_or_deployment) == "reasoning"

# ---------------------------------------------------------------------------
# Reasoning controls
# ---------------------------------------------------------------------------

_VALID_REASONING_EFFORT = {"minimal", "low", "medium", "high"}
_VALID_VERBOSITY = {"low", "medium", "high"}

def _coerce_choice(raw: Optional[str], valid: Iterable[str]) -> Optional[str]:
    if raw is None:
        return None
    value = str(raw).strip().lower()
    if not value:
        return None
    if value not in set(valid):
        logging.warning(
            "Ignoring unsupported value '%s'; expected one of %s",
            raw, sorted(valid),
        )
        return None
    return value

def get_reasoning_effort(override: Optional[str] = None) -> Optional[str]:
    return _coerce_choice(
        override if override is not None else os.getenv("OPENAI_REASONING_EFFORT"),
        _VALID_REASONING_EFFORT,
    )

def get_verbosity(override: Optional[str] = None) -> Optional[str]:
    return _coerce_choice(
        override if override is not None else os.getenv("OPENAI_VERBOSITY"),
        _VALID_VERBOSITY,
    )

# ---------------------------------------------------------------------------
# max_completion_tokens scaling
# ---------------------------------------------------------------------------

def _reasoning_token_scale() -> float:
    """Multiplier applied to legacy ``max_tokens`` when targeting a reasoning model."""
    try:
        scale = float(os.getenv("OPENAI_REASONING_TOKEN_SCALE", "2.5"))
    except (TypeError, ValueError):
        scale = 2.5
    return scale if scale > 0 else 1.0

def _reasoning_token_floor() -> int:
    try:
        floor = int(os.getenv("OPENAI_REASONING_TOKEN_FLOOR", "4096"))
    except (TypeError, ValueError):
        floor = 4096
    return floor if floor > 0 else 4096

def scale_max_tokens_for_reasoning(max_tokens: Optional[int]) -> Optional[int]:
    """Scale a legacy ``max_tokens`` budget up for reasoning models.

    ``None`` and ``-1`` ("no explicit cap") are passed through.
    """
    if max_tokens is None:
        return None
    if max_tokens == -1:
        return -1
    return max(int(round(max_tokens * _reasoning_token_scale())), _reasoning_token_floor())

# ---------------------------------------------------------------------------
# Kwargs builders
# ---------------------------------------------------------------------------

_SAMPLING_KEYS = ("temperature", "top_p", "presence_penalty", "frequency_penalty")

def _drop_none(mapping: Mapping[str, Any]) -> Dict[str, Any]:
    return {k: v for k, v in mapping.items() if v is not None}

def build_openai_chat_kwargs(
    model: str,
    *,
    max_tokens: Optional[int] = None,
    temperature: Optional[float] = None,
    top_p: Optional[float] = None,
    presence_penalty: Optional[float] = None,
    frequency_penalty: Optional[float] = None,
    reasoning_effort: Optional[str] = None,
    verbosity: Optional[str] = None,
    extra: Optional[Mapping[str, Any]] = None,
) -> Dict[str, Any]:
    """Build kwargs for ``openai.OpenAI / AzureOpenAI .chat.completions.create``.

    Splat the result directly: ``client.chat.completions.create(**kwargs)``.
    Unsupported parameters are silently omitted for reasoning models; legacy
    deployments retain the historical behaviour.
    """
    family = get_model_family(model)
    kwargs: Dict[str, Any] = {"model": model}

    # ---- output budget ----
    if max_tokens is not None and max_tokens != -1:
        if family == "reasoning":
            kwargs["max_completion_tokens"] = scale_max_tokens_for_reasoning(int(max_tokens))
        else:
            kwargs["max_tokens"] = int(max_tokens)

    # ---- sampling ----
    if family == "legacy":
        kwargs.update(_drop_none({
            "temperature": temperature,
            "top_p": top_p,
            "presence_penalty": presence_penalty,
            "frequency_penalty": frequency_penalty,
        }))
    else:
        for key, value in (
            ("temperature", temperature), ("top_p", top_p),
            ("presence_penalty", presence_penalty), ("frequency_penalty", frequency_penalty),
        ):
            if value is not None:
                logging.debug(
                    "Dropping unsupported parameter '%s' for reasoning model '%s'",
                    key, model,
                )

    # ---- reasoning controls ----
    if family == "reasoning":
        effort = get_reasoning_effort(reasoning_effort)
        if effort is not None:
            kwargs["reasoning_effort"] = effort
        verb = get_verbosity(verbosity)
        if verb is not None:
            # ``verbosity`` is not a top-level kwarg in openai-python <= 1.65.x;
            # route it via ``extra_body`` so it lands in the JSON without a
            # TypeError from the SDK.
            kwargs.setdefault("extra_body", {})["verbosity"] = verb

    # ---- caller-supplied extras (already filtered) ----
    if extra:
        for key, value in extra.items():
            if value is None:
                continue
            if family == "reasoning" and key in _SAMPLING_KEYS:
                continue
            kwargs[key] = value

    return kwargs

def build_langchain_chat_kwargs(
    deployment_name: str,
    *,
    max_tokens: Optional[int] = None,
    temperature: Optional[float] = None,
    top_p: Optional[float] = None,
    reasoning_effort: Optional[str] = None,
    verbosity: Optional[str] = None,
) -> Dict[str, Any]:
    """Build kwargs for ``langchain_openai.AzureChatOpenAI`` / ``ChatOpenAI``.

    Older ``langchain-openai`` releases don't expose ``max_completion_tokens``
    as a top-level kwarg, so we forward it through ``model_kwargs`` (which
    langchain passes straight to the SDK).
    """
    family = get_model_family(deployment_name)
    kwargs: Dict[str, Any] = {}
    model_kwargs: Dict[str, Any] = {}

    if max_tokens is not None and max_tokens != -1:
        if family == "reasoning":
            model_kwargs["max_completion_tokens"] = scale_max_tokens_for_reasoning(int(max_tokens))
        else:
            kwargs["max_tokens"] = int(max_tokens)

    if family == "reasoning":
        effort = get_reasoning_effort(reasoning_effort)
        if effort is not None:
            model_kwargs["reasoning_effort"] = effort
        verb = get_verbosity(verbosity)
        if verb is not None:
            model_kwargs.setdefault("extra_body", {})["verbosity"] = verb
    else:
        if temperature is not None:
            kwargs["temperature"] = temperature
        if top_p is not None:
            kwargs["top_p"] = top_p

    if model_kwargs:
        kwargs["model_kwargs"] = model_kwargs
    return kwargs

def get_system_role(model_or_deployment: Optional[str] = None) -> str:
    """Return ``"developer"`` for reasoning models when opted in, ``"system"`` otherwise.

    Defaulting to ``"system"`` preserves compatibility with LangChain prompt
    templates and SDK helpers that don't yet recognise the new role. Opt in
    with ``OPENAI_USE_DEVELOPER_ROLE=1`` once your stack supports it.
    """
    if not is_reasoning_model(model_or_deployment):
        return "system"
    raw = os.getenv("OPENAI_USE_DEVELOPER_ROLE", "")
    return "developer" if raw.strip().lower() in {"1", "true", "yes", "on"} else "system"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

from openai import AzureOpenAI
from model_compat import build_openai_chat_kwargs

client = AzureOpenAI(
    azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
    api_version=os.environ["OPENAI_API_VERSION"],
    api_key=os.environ["AZURE_OPENAI_API_KEY"],
)

kwargs = build_openai_chat_kwargs(
    model=os.environ["OPENAI_ENGINE"],
    max_tokens=4096,           # automatically becomes max_completion_tokens for GPT-5
    temperature=0.2,           # automatically dropped for GPT-5
    reasoning_effort="low",    # automatically dropped for GPT-4
)
response = client.chat.completions.create(
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": user_input},
    ],
    **kwargs,
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

import json
import requests
from model_compat import build_openai_chat_kwargs, get_system_role

deployment = os.environ["OPENAI_ENGINE"]
api_version = os.environ["OPENAI_API_VERSION"]
endpoint = (
    f"{os.environ['AZURE_OPENAI_ENDPOINT']}/openai/deployments/{deployment}"
    f"/chat/completions?api-version={api_version}"
)

payload = {
    "messages": [
        {"role": get_system_role(deployment), "content": system_prompt},
        {"role": "user", "content": user_prompt},
    ],
}
# Splat the kwargs into the payload, then strip the SDK-only ``model`` key.
payload.update(build_openai_chat_kwargs(
    model=deployment,
    max_tokens=800,
    temperature=0.7,
    top_p=0.95,
    reasoning_effort="low",
))
payload.pop("model", None)        # ``model`` is encoded in the URL for Azure
payload.pop("extra_body", None)   # already on the payload root

resp = requests.post(
    endpoint,
    headers={"Content-Type": "application/json", "api-key": api_key},
    data=json.dumps(payload),
    timeout=60,
)
resp.raise_for_status()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

"""LangChain-side compatibility shim for reasoning-class deployments."""

from __future__ import annotations

from typing import Any, List, Optional

from langchain_core.callbacks.manager import (
    AsyncCallbackManagerForLLMRun, CallbackManagerForLLMRun,
)
from langchain_core.messages import BaseMessage
from langchain_core.outputs import ChatResult
from langchain_openai import AzureChatOpenAI   # use ChatOpenAI for non-Azure

from model_compat import is_reasoning_model

class ReasoningSafeAzureChatOpenAI(AzureChatOpenAI):
    """``AzureChatOpenAI`` variant that hides parameters reasoning models reject.

    Reasoning models (GPT-5.x, o1/o3/o4) return HTTP 400 when a request
    payload carries ``stop``. LangChain's SQL helpers unconditionally bind it,
    so the unsupported parameter reaches the SDK regardless of how the caller
    configured the LLM. This subclass strips ``stop`` for reasoning
    deployments while forwarding it unchanged for legacy GPT-4 / GPT-3.5
    deployments - the behaviour is byte-identical to upstream LangChain
    for those models.
    """

    def _deployment_id(self) -> str:
        # ``langchain-openai`` >= 0.2 exposes ``azure_deployment``; older
        # releases use ``deployment_name``. Either may be set by the caller.
        return (
            getattr(self, "azure_deployment", None)
            or getattr(self, "deployment_name", None)
            or ""
        )

    def _generate(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> ChatResult:
        if is_reasoning_model(self._deployment_id()):
            stop = None
        return super()._generate(messages, stop=stop, run_manager=run_manager, **kwargs)

    async def _agenerate(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[AsyncCallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> ChatResult:
        if is_reasoning_model(self._deployment_id()):
            stop = None
        return await super()._agenerate(messages, stop=stop, run_manager=run_manager, **kwargs)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

from langchain_compat import ReasoningSafeAzureChatOpenAI
from model_compat import build_langchain_chat_kwargs

llm_kwargs = build_langchain_chat_kwargs(
    deployment_name=os.environ["OPENAI_ENGINE"],
    max_tokens=6000,
    temperature=0,
    reasoning_effort="low",
)
llm = ReasoningSafeAzureChatOpenAI(
    azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
    azure_deployment=os.environ["OPENAI_ENGINE"],
    openai_api_version=os.environ["OPENAI_API_VERSION"],
    api_key=os.environ["AZURE_OPENAI_API_KEY"],
    **llm_kwargs,
)

1
2

sql = chain.invoke({...})              # -> "I don't know"
result = db.run(sql)                   # -> sends "I don't know" to pyodbc

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

import re

_SQL_START_RE = re.compile(
    r"^\s*(?:WITH|SELECT|INSERT|UPDATE|DELETE|CREATE|DROP|ALTER|MERGE|EXEC|EXECUTE|TRUNCATE)\b",
    re.IGNORECASE,
)

def looks_like_sql(text: str) -> bool:
    """True only if ``text`` starts with a recognised SQL DML/DDL keyword."""
    if not text or not text.strip():
        return False
    return bool(_SQL_START_RE.match(text))

sql = extract_sql_query(chain.invoke({...}))
if not looks_like_sql(sql):
    logging.warning("SQL chain returned a non-SQL response: %r", sql[:200])
    return (
        "I couldn't form a SQL query for that question. "
        "Please rephrase or add more context."
    )
result = db.run(sql)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

import re

def extract_sql_query(text: str) -> str:
    """Strip markdown fences, leading prose, and trailing explanations."""
    # 1) Prefer SQL inside a markdown code fence.
    m = re.search(r"```(?:sql|SQL|Sql)?\s*\n(.*?)\n```", text, re.DOTALL)
    if m:
        text = m.group(1)
    text = text.strip()

    # 2) Drop any prose *before* the SQL by jumping to the first SQL keyword.
    m = re.search(
        r"(?im)^\s*(WITH|SELECT|INSERT|UPDATE|DELETE|CREATE|DROP|ALTER|MERGE|EXEC|EXECUTE|TRUNCATE)\b",
        text,
    )
    if m:
        text = text[m.start(1):]

    # 3) Cut at the first "Explanation:" / "Note:" / "This query..." marker.
    m = re.compile(
        r"(?im)^\s*(?:Explanation|Note|Notes|Here(?:'|\u2019)?s|"
        r"This\s+(?:query|SQL|statement|returns|counts|selects|will|gets|finds)|"
        r"The\s+(?:query|SQL|above|result|statement)|"
        r"Result|Results|Description|Output|Answer)\b[^\n]*"
    ).search(text)
    if m:
        text = text[: m.start()].rstrip()

    # 4) Drop any trailing fence that survived step 1.
    if text.endswith("```"):
        text = text[:-3].rstrip()
    return text.strip()

1
2
3
4
5
6
7
8

TASK_EFFORT = {
    "sql":                 "low",
    "structured_extract":  "low",
    "kg_cleaning":         "low",
    "rag_qa":              "medium",
    "vision":              "medium",
    "analytical":          "high",
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

# prompt_audit.py - minimal differential tester
import csv
from openai import AzureOpenAI
from model_compat import build_openai_chat_kwargs

LEGACY      = "gpt-4o"
REASONING   = "gpt-5.1"
client      = AzureOpenAI(
    azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
    api_version=os.environ["OPENAI_API_VERSION"],
    api_key=os.environ["AZURE_OPENAI_API_KEY"],
)

def run(model: str, system: str, user: str) -> str:
    kw = build_openai_chat_kwargs(
        model=model,
        max_tokens=4096,
        temperature=0.2,           # auto-dropped for reasoning
        reasoning_effort="medium", # auto-dropped for legacy
    )
    resp = client.chat.completions.create(
        messages=[
            {"role": "system", "content": system},
            {"role": "user",   "content": user},
        ],
        **kw,
    )
    return resp.choices[0].message.content or ""

with open("prompts.csv") as f_in, open("diff.tsv", "w", newline="") as f_out:
    writer = csv.writer(f_out, delimiter="\t")
    writer.writerow(["id", "legacy_first80", "reasoning_first80",
                     "len_legacy", "len_new", "identical"])
    for row in csv.DictReader(f_in):
        legacy = run(LEGACY,    row["system"], row["user"])
        new    = run(REASONING, row["system"], row["user"])
        writer.writerow([
            row["id"],
            legacy[:80].replace("\n", " "),
            new[:80].replace("\n", " "),
            len(legacy), len(new),
            legacy.strip() == new.strip(),
        ])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

resp = client.chat.completions.create(
    messages=[...],
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "scored_entity",
            "schema": {
                "type": "object",
                "properties": {
                    "name":  {"type": "string"},
                    "score": {"type": "number"},
                },
                "required": ["name", "score"],
                "additionalProperties": False,
            },
            "strict": True,
        },
    },
    **kw,
)

1
2

resp = client.chat.completions.create(messages=[...], n=5, **kw)
candidates = [c.message.content for c in resp.choices]

1
2
3
4
5

messages = [
    {"role": get_system_role(deployment), "content": role_card_for_assistant},
    {"role": "developer",                 "content": procedural_instructions},
    {"role": "user",                      "content": user_question},
]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

LITERAL_EXECUTION_HEADER = (
    "Execution mode: follow the instructions below literally and in order. "
    "Do not infer intent, skip, reorder, merge, or add steps. Honour the "
    "exact formatting, tone, and verbosity specified. If a step is "
    "ambiguous, respond with the literal interpretation and flag the "
    "ambiguity instead of guessing."
)

def apply_literal_execution(prompt: str) -> str:
    if LITERAL_EXECUTION_HEADER in prompt:
        return prompt
    return f"{LITERAL_EXECUTION_HEADER}\n\n{prompt}"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

import pytest
from model_compat import get_model_family, build_openai_chat_kwargs

@pytest.mark.parametrize("name,expected", [
    ("gpt-5.1", "reasoning"),
    ("gpt5", "reasoning"),
    ("gpt-5-prod-eu", "reasoning"),
    ("o3-mini", "reasoning"),
    ("o1", "reasoning"),
    ("gpt-4o", "legacy"),
    ("gpt-4", "legacy"),
    ("gpt-4-32k", "legacy"),
    ("gpt-35-turbo", "legacy"),
    ("",          "legacy"),    # unknown -> fail closed to legacy
    (None,        "legacy"),
])
def test_family(name, expected):
    assert get_model_family(name) == expected

def test_kwargs_for_reasoning_drops_temperature():
    kw = build_openai_chat_kwargs(
        model="gpt-5.1", max_tokens=1000, temperature=0.2, top_p=0.9,
        reasoning_effort="low",
    )
    assert "temperature" not in kw
    assert "top_p" not in kw
    assert kw["max_completion_tokens"] >= 4096   # floor applied
    assert kw["reasoning_effort"] == "low"

def test_kwargs_for_legacy_keeps_temperature():
    kw = build_openai_chat_kwargs(
        model="gpt-4o", max_tokens=1000, temperature=0.2, top_p=0.9,
    )
    assert kw["max_tokens"] == 1000
    assert kw["temperature"] == 0.2
    assert kw["top_p"] == 0.9
    assert "reasoning_effort" not in kw

1
2
3
4
5
6
7
8
9

# Does a deployment name classify correctly?
python -c "from model_compat import get_model_family; print(get_model_family('gpt-5.1'))"
# -> reasoning

# Does the LangChain LLM strip ``stop`` when the deployment is GPT-5.1?
python -c "
from langchain_compat import ReasoningSafeAzureChatOpenAI
import inspect; print(inspect.getsource(ReasoningSafeAzureChatOpenAI._generate))
"