From 39417a21ecc8d7ec853aa5627781f700bda08796 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Mon, 10 Aug 2026 03:57:50 -0400 Subject: [PATCH] =?UTF-8?q?feat(adapters):=20=E8=90=BD=E6=88=90=E7=BD=91?= =?UTF-8?q?=E5=85=B3=E9=80=82=E9=85=8D=E5=99=A8=EF=BC=8C=E5=8D=81=E4=B8=AA?= =?UTF-8?q?=E6=A8=A1=E5=9D=97=E5=85=A8=E9=83=A8=E6=9C=89=E5=86=85=E5=AE=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit design 0012（待确认）定六条：收一个已经装配好的客户端而不自己装配（治理参数按 §1.5 不归本库 管，而且项目常要在多个用途间共享同一个限流器和缓存）；同时收那份配置只为算模型身份（客户端 把源列表与 scope 收在内部不公开）；内容块按顺序拼成一个字符串不加分隔符；绑定里只有网关认得 的两个键往下传、其余留在参数快照里且不报错；网关异常原样穿出去不翻译不重试；空串的调用标识 映射成空值。 参数快照不用网关内部那个 build_model_fingerprint：它不在 __all__ 里（用它就得从子模块 import， 他们重排一次我们就断），而且它是为缓存键设计的、按模型名去重。续跑守卫怕的是「配置变了而我 没发现」，所以宁可更严——改一个源名也报出来，那意味着这次运行打的可能是另一个端点。 tests/integration/ 这一层第一次有内容：用的是网关真实的 GatewaySettings / LLMResponse / 异常类型，装配守卫也真的跑了，只把「真的发出去」那一下换成受控替身。没装 polyloop[gateway] 时整份文件跳过——一个因为可选依赖没装而常年红的套件会训练所有人忽略红。 环境：从 ~/Projects/PolyGateway（活版本 1.1.2，比 reference/ 那份 1.1.1 新）复制一份装进 conda 环境。没有配私有源，polygateway 不在任何可达的 index 上。 --- .../design/0012-gateway-model-client.md | 14 +- src/polyloop/adapters/__init__.py | 140 +++++++++++- .../integration/test_gateway_model_client.py | 209 ++++++++++++++++++ 3 files changed, 355 insertions(+), 8 deletions(-) create mode 100644 tests/integration/test_gateway_model_client.py diff --git a/research-wiki/design/0012-gateway-model-client.md b/research-wiki/design/0012-gateway-model-client.md index 2716c98..91802d9 100644 --- a/research-wiki/design/0012-gateway-model-client.md +++ b/research-wiki/design/0012-gateway-model-client.md @@ -34,12 +34,16 @@ GatewayModelClient(client=..., settings=...) `parameters()` 要回答「这次运行用的是哪个模型配置」，续跑时逐字段比对。但**客户端不公开 它的源列表与 scope**（构造时收下，只留在内部），所以从客户端本身问不出这个答案。 -于是适配器同时收那份 `GatewaySettings`，用**网关自己的** `build_model_fingerprint(sources)` -算指纹。那个函数是它的公开函数，语义是「本 scope 会用哪些（模型、请求形态）组合」，并且把 -采样参数与推理开关也算进去——把 temperature 从 0 改成 1 之后重启，指纹会变。 +于是适配器同时收那份 `GatewaySettings`，从它的 `scope` 与 `sources` 里读出来。每个源报四样： +源名、供应商、模型名，以及会改变请求体的那两项——恒定采样参数与推理开关。 -**用它而不是我们自己拼一串**，因为模型身份怎么算是网关的事：他们哪天认为某个新字段也该参与 -身份，改在他们那里，我们跟着变。自己拼的话，那个定义会和他们的悄悄分叉。 +**不用网关内部那个 `build_model_fingerprint`。** 它确实算的是「模型身份」，但它不在网关的 +`__all__` 里，用它就得从子模块 import，而那是它的内部布局、他们重排一次我们就断。更要紧的是 +**它是为缓存键设计的**：它按模型名去重（多个源同一个模型算一份），因为缓存怕的是「不同配置 +读到同一份缓存」。续跑守卫怕的是另一件事——「配置变了而我没发现」，所以宁可更严：改一个源名 +也该让它报出来，因为那意味着这次运行打的可能是另一个端点。 + +读的全是 `SourceConfig` 的公开字段，那个类在网关的 `__all__` 里。 **残留风险照实认下：客户端与配置必须真的是同一对。** 传一个客户端加另一份配置，指纹会说谎， 而续跑守卫就白设了。库验不了这件事——客户端不公开它是按哪份配置装的。这条写进那个类的 diff --git a/src/polyloop/adapters/__init__.py b/src/polyloop/adapters/__init__.py index 39d695a..de13a6a 100644 --- a/src/polyloop/adapters/__init__.py +++ b/src/polyloop/adapters/__init__.py @@ -1,5 +1,139 @@ -"""PolyGateway 模型适配器。**须由使用者显式 import。** +"""模型调用接缝的网关适配器。 -**这是全库唯一允许 import `polygateway` 的地方。** 不进 `polyloop/__init__.py`：一个 -顺手提供的默认模型客户端会让每个进程在 import 时把网关连同它的 provider 目录一起拉起来。 +**这个模块公开，但必须显式 import**，不进 `polyloop/__init__.py`（`0003` 决策八第 9 条）： +顺手导出它会让每个进程在 import 本库时把网关连同它的 provider 目录一起拉起来。它也是本库里 +**唯一**允许 import 网关的地方，由 `pyproject.toml` 的一条 import 契约断言。 + +装它要装 `polyloop[gateway]` 那个 extra。缝两边怎么对上，定在 +`research-wiki/design/0012-gateway-model-client.md`。 + +**这里不写重试、限流、熔断、缓存、遥测**（`CLAUDE.md` §1.5）。网关内部已经有这些，外面再套 +一层会让两套预算重叠计费。缺能力就给网关提 PR。 """ + +import json +from collections.abc import Mapping, Sequence + +from polygateway import GatewayClient, GatewaySettings, SourceConfig + +from polyloop.ports import ModelCall +from polyloop.types import ContentBlock, Message, ModelReply, TextBlock + +#: 绑定里能被网关认下的那几个键。其余的键留在参数快照里，不往下传。 +_FORWARDED_BINDING_KEYS = ("session_id", "parent_call_id") + + +class GatewayModelClient: + """把本库的一次模型调用翻译成网关的一次治理调用。 + + **收一个已经装配好的客户端，不自己装配。** 网关的装配入口收十几个治理参数（限流器、熔断 + 器、缓存后端、遥测记录器……），那些按 `CLAUDE.md` §1.5 不归本库管；而且项目常常要在多个 + 用途之间共享同一个限流器和缓存，那只有它自己装配才做得到。 + + **同时收那份配置，只为算出可复现的模型身份。** 客户端把源列表与 scope 收在内部、不公开， + 所以从它本身问不出「这次运行用的是哪个模型配置」。 + + **调用方要保证这两个参数是它真的配对使用的那一对。** 传一个客户端加另一份配置，参数快照 + 会说谎，而续跑守卫就白设了——库验不了这件事，客户端不公开它是按哪份配置装的。 + + 它满足 `polyloop.ports.ModelClient`，但不显式继承那个 Protocol：结构化子类型不需要继承。 + """ + + __slots__ = ("_client", "_settings") + + def __init__(self, *, client: GatewayClient, settings: GatewaySettings) -> None: + self._client = client + self._settings = settings + + def __repr__(self) -> str: + return f"GatewayModelClient(scope={self._settings.scope!r})" + + async def call(self, call: ModelCall) -> ModelReply: + """发一次调用，把响应收成三个字段。 + + **网关的异常原样穿出去**，不捕获、不翻译、不重试。上一层已经定了模型调用失败怎么处置 + （记一条带失败说明的结果记录、记一步、以模型调用失败收尾），而失败说明取的是异常的 + 类名与文本——网关的异常类名本身就是最有用的那部分信息，翻译成我们自己的名字只会把它 + 盖掉。重试尤其不能做：网关内部已经有重试、退避、换源、熔断。 + """ + response = await self._client.chat( + [_as_gateway_message(message) for message in call.messages], + **_forwarded_binding(call.binding), + ) + return ModelReply( + # 网关那边这个字段的类型是 `str`，而本库这边**绝不为空串**——空串是个看起来合法的 + # 键，连表时静默匹配不上，而空值至少能被显式筛出来。 + call_id=response.call_id or None, + content=response.content, + thinking=response.thinking, + ) + + def parameters(self) -> Mapping[str, str]: + """上报这次装配的模型身份。 + + 每个源报四样：源名、供应商、模型名，以及会改变请求体的那两项（恒定采样参数、推理开关）。 + 把采样参数算进去是必须的——temperature 从 0 改成 1 之后续跑，模型的行为变了而轨迹上 + 看不出来。 + + **比网关自己那个缓存指纹更严**：那个按模型名去重，因为缓存怕的是「不同配置读到同一份 + 缓存」；续跑守卫怕的是另一件事——「配置变了而我没发现」，所以改一个源名也该报出来， + 那意味着这次运行打的可能是另一个端点。 + """ + return { + "scope": self._settings.scope, + "sources": _describe_sources(self._settings.sources), + } + + +def _as_gateway_message(message: Message) -> dict[str, object]: + """一条消息翻译成网关那边的形状。 + + **内容块按顺序拼成一个字符串，不加分隔符**——块之间本来就没有分隔符这个概念，加了就是往 + 模型看见的文字里塞东西。第一版只有文本块；将来有图片块时这里改成多模态数组，那是加分支 + 不是改签名，正是 `0003` 决策六把内容定成序列而不是裸字符串换来的。 + """ + return { + "role": message.role.value, + "content": "".join(_block_text(block) for block in message.content), + } + + +def _block_text(block: ContentBlock) -> str: + if isinstance(block, TextBlock): + return block.text + raise TypeError( + f"这个适配器还不认得内容块类型 {type(block).__name__}。" + "加一种块就要在这里加一条对应的翻译，漏了的话那一块会静默地不进请求" + ) + + +def _forwarded_binding(binding: Mapping[str, str]) -> dict[str, str]: + """绑定里网关认得的那几个键。 + + **其余的键不往下传，也不报错。** 绑定是项目自己的坐标（某个下游有五维），而网关只有两个 + 槽位放得下这类东西。不报错是因为那些键**已经被记下来了**——绑定的全部键值都进运行开始 + 记录的参数快照（`0006` 决策三），续跑时逐字段比对。报错等于要求项目为了适配一个网关而 + 裁剪自己的坐标系，而绑定同时是续跑守卫的输入，改它会让所有在跑的运行续不上。 + """ + return {key: binding[key] for key in _FORWARDED_BINDING_KEYS if key in binding} + + +def _describe_sources(sources: Sequence[SourceConfig]) -> str: + """把源列表压成一个确定的字符串。 + + 按源名排序，免得配置文件里换个顺序就报出一次假的漂移。 + """ + described = [ + { + "name": source.name, + "provider": source.provider, + "model": source.model, + "extra_body": dict(source.extra_body), + "enable_thinking": source.enable_thinking, + } + for source in sorted(sources, key=lambda source: source.name) + ] + return json.dumps(described, sort_keys=True, ensure_ascii=False) + + +__all__ = ["GatewayModelClient"] diff --git a/tests/integration/test_gateway_model_client.py b/tests/integration/test_gateway_model_client.py new file mode 100644 index 0000000..baa0693 --- /dev/null +++ b/tests/integration/test_gateway_model_client.py @@ -0,0 +1,209 @@ +"""网关适配器：把本库的一次模型调用翻译成网关的一次治理调用。 + +**这一层叫 integration，因为它连的是真的 PolyGateway**（`CLAUDE.md` §1.9 的分层判据是「依赖 +什么」）：这里用的 `GatewaySettings`、`SourceConfig`、`LLMResponse` 全是网关真实的类型，装配 +守卫也真的跑了。它**不打真实模型**——那是 e2e 的事，所以客户端那一下换成一个受控替身。 + +网关不在公共源上，装它要 `polyloop[gateway]` 那个 extra。没装的时候整份文件跳过，而不是让 +`make ci` 红——一个因为可选依赖没装而常年红的套件会训练所有人忽略红。 +""" + +from collections.abc import Mapping + +import pytest + +polygateway = pytest.importorskip( + "polygateway", reason="没装 polyloop[gateway]，网关适配器这一层跳过" +) + +from polygateway import GatewaySettings, LLMResponse # noqa: E402 +from polygateway.errors import AllSourcesExhausted # noqa: E402 + +from polyloop.adapters import GatewayModelClient # noqa: E402 +from polyloop.ports import ModelCall # noqa: E402 +from polyloop.types import Message, Role, TextBlock # noqa: E402 + +pytestmark = pytest.mark.integration + +#: 一份最小但能过网关全部装配守卫的配置。 +#: 键的形状是 `{SCOPE}__{PROVIDER}__{序号}__{字段}`，源名由网关拼成 `{provider}_{序号}`。 +_ENV = { + "LLM__OPENAI__1__BASE_URL": "https://example.invalid/v1", + "LLM__OPENAI__1__API_KEY": "sk-test", + "LLM__OPENAI__1__MODEL": "test-model", + "LLM__OPENAI__1__TIMEOUT_S": "30", + # 治理参数网关一律要求显式声明，不给默认——配错的代价它自己承担，这里照最小值填。 + "LLM__RETRY__MAX_ATTEMPTS": "1", + "LLM__RETRY__BACKOFF_BASE_S": "0.1", + "LLM__RETRY__BACKOFF_MAX_S": "1", + "LLM__BREAKER__FAIL_THRESHOLD": "5", + "LLM__BREAKER__COOLDOWN_S": "10", + "PGW_CACHE_BACKEND": "none", + "PGW_TELEMETRY_BACKEND": "none", +} + + +def _settings(**overrides: str) -> GatewaySettings: + return GatewaySettings.from_env(scope="LLM", env={**_ENV, **overrides}, env_file="") + + +class _StubClient: + """替身客户端：记下收到什么，按脚本返回或抛出。 + + 只替掉「真的发出去」那一下——配置、装配守卫、响应类型都还是网关真的那套。 + """ + + def __init__(self, response: LLMResponse | BaseException) -> None: + self._response = response + self.calls: list[tuple[list[dict], dict]] = [] + + async def chat(self, messages, **kwargs): + self.calls.append((messages, kwargs)) + if isinstance(self._response, BaseException): + raise self._response + return self._response + + +def _response(*, call_id: str = "gw-1", content: str = "答案", thinking: str = "想了想"): + return LLMResponse( + content=content, + thinking=thinking, + model="test-model", + provider="openai", + prompt_tokens=10, + completion_tokens=5, + latency_ms=123, + ttft_ms=12.0, + max_inter_token_ms=3.0, + cache_hit=False, + call_id=call_id, + ) + + +def _call(*, messages=None, binding: Mapping[str, str] | None = None) -> ModelCall: + return ModelCall( + messages=messages + or ( + Message(role=Role.SYSTEM, content=(TextBlock(text="你是助手"),)), + Message(role=Role.USER, content=(TextBlock(text="数到三"),)), + ), + call_index=0, + run_id="r1", + result_id="r1#model#0", + binding=dict(binding or {}), + ) + + +async def test_messages_are_translated_to_role_content_dicts() -> None: + stub = _StubClient(_response()) + client = GatewayModelClient(client=stub, settings=_settings()) + + await client.call(_call()) + + ((messages, _),) = stub.calls + assert messages == [ + {"role": "system", "content": "你是助手"}, + {"role": "user", "content": "数到三"}, + ] + + +async def test_content_blocks_are_joined_without_a_separator() -> None: + """块之间本来就没有分隔符这个概念，加了就是往模型看见的文字里塞东西。""" + stub = _StubClient(_response()) + client = GatewayModelClient(client=stub, settings=_settings()) + message = Message(role=Role.USER, content=(TextBlock(text="前半"), TextBlock(text="后半"))) + + await client.call(_call(messages=(message,))) + + ((messages, _),) = stub.calls + assert messages == [{"role": "user", "content": "前半后半"}] + + +async def test_the_reply_keeps_only_the_three_fields_this_library_records() -> None: + """响应有二十来个字段，本库只取三个，其余留在网关的账目里靠调用标识连过去。""" + stub = _StubClient(_response(call_id="gw-7", content="内容", thinking="推理")) + client = GatewayModelClient(client=stub, settings=_settings()) + + reply = await client.call(_call()) + + assert reply.call_id == "gw-7" + assert reply.content == "内容" + assert reply.thinking == "推理" + + +async def test_an_empty_call_id_becomes_no_call_id() -> None: + """空串是个看起来合法的键，连表时静默匹配不上，而空值至少能被显式筛出来。""" + stub = _StubClient(_response(call_id="")) + client = GatewayModelClient(client=stub, settings=_settings()) + + assert (await client.call(_call())).call_id is None + + +async def test_only_the_binding_keys_the_gateway_has_slots_for_are_forwarded() -> None: + """其余的键不往下传也不报错——它们已经进了参数快照，网关那边只是没有格子放。 + + 报错等于要求项目为了适配一个网关而裁剪自己的坐标系，而绑定同时是续跑守卫的输入。 + """ + stub = _StubClient(_response()) + client = GatewayModelClient(client=stub, settings=_settings()) + + await client.call(_call(binding={"session_id": "s1", "book": "b7", "task": "t3"})) + + ((_, kwargs),) = stub.calls + assert kwargs == {"session_id": "s1"} + + +async def test_gateway_errors_propagate_untranslated() -> None: + """网关的异常类名本身就是最有用的那部分信息，翻译成我们自己的名字只会把它盖掉。 + + 上一层已经定了怎么处置：记一条带失败说明的结果记录、记一步、以模型调用失败收尾。 + """ + stub = _StubClient( + AllSourcesExhausted(scope="llm", reason="retry_exhausted", retry_after_s=1.0) + ) + client = GatewayModelClient(client=stub, settings=_settings()) + + with pytest.raises(AllSourcesExhausted): + await client.call(_call()) + + +async def test_cancellation_passes_straight_through() -> None: + import asyncio + + stub = _StubClient(asyncio.CancelledError()) + client = GatewayModelClient(client=stub, settings=_settings()) + + with pytest.raises(asyncio.CancelledError): + await client.call(_call()) + + +def test_the_parameters_describe_the_model_configuration() -> None: + client = GatewayModelClient(client=_StubClient(_response()), settings=_settings()) + + parameters = client.parameters() + + assert parameters["scope"] == "llm" # 网关自己把 scope 收成小写 + assert "test-model" in parameters["sources"] + assert "openai" in parameters["sources"] + + +def test_changing_the_model_changes_the_parameters() -> None: + """换了模型而快照不变的话，续跑守卫就漏掉了最该拦的那一种改动。""" + before = GatewayModelClient(client=_StubClient(_response()), settings=_settings()).parameters() + after = GatewayModelClient( + client=_StubClient(_response()), + settings=_settings(**{"LLM__OPENAI__1__MODEL": "another-model"}), + ).parameters() + + assert before["sources"] != after["sources"] + + +def test_changing_a_sampling_parameter_changes_the_parameters() -> None: + """temperature 从 0 改成 1 之后续跑，模型的行为变了而轨迹上看不出来——除非它进快照。""" + before = GatewayModelClient(client=_StubClient(_response()), settings=_settings()).parameters() + after = GatewayModelClient( + client=_StubClient(_response()), + settings=_settings(**{"LLM__OPENAI__1__EXTRA_BODY": '{"temperature": 1}'}), + ).parameters() + + assert before["sources"] != after["sources"]