diff --git a/pycodeloop/providers/generic.py b/pycodeloop/providers/generic.py index 721c00c..63334af 100644 --- a/pycodeloop/providers/generic.py +++ b/pycodeloop/providers/generic.py @@ -120,7 +120,7 @@ class _ConnectionSnapshot: timeout: float request_builder: RequestBuilder response_parser: ResponseParser - uses_default_parser: bool + supports_openai_sse: bool class GenericProvider(Provider): @@ -168,6 +168,13 @@ class GenericProvider(Provider): `message_content` rename per-message keys). `params` are extra static fields merged into every request body (e.g. `temperature`, `max_tokens`, vendor-specific flags). + + Streaming (OpenAI-style SSE) still works with `response_paths` — + the wire format is unchanged, only the *non-streaming* response's + JSON key paths differ. `response_shape: "anthropic"` is the + exception: that's a genuinely different SSE envelope `_stream()` + doesn't understand, so streaming is skipped for it in favor of one + blocking, fully-buffered `on_delta` call. """ name = "generic" @@ -186,6 +193,7 @@ def __init__( repetition_min_period: int = _REPETITION_MIN_PERIOD, repetition_max_period: int = _REPETITION_MAX_PERIOD, repetition_repeats: int = _REPETITION_REPEATS, + supports_openai_sse: bool = True, **kwargs, ) -> None: super().__init__(model=model, api_key=api_key, **kwargs) @@ -199,7 +207,7 @@ def __init__( self.repetition_min_period = repetition_min_period self.repetition_max_period = repetition_max_period self.repetition_repeats = repetition_repeats - self._uses_default_parser = response_parser is None + self._supports_openai_sse = supports_openai_sse self._config_path: Path | None = None self._lock = threading.Lock() @@ -222,8 +230,9 @@ def _build_from_json(cls, path: str | Path) -> GenericProvider: if not api_key and data.get("api_key_env"): api_key = os.environ.get(data["api_key_env"]) + response_shape = data.get("response_shape") response_parser = None - if data.get("response_shape") == "anthropic": + if response_shape == "anthropic": response_parser = anthropic_response elif "response_paths" in data: response_parser = response_parser_from_paths( @@ -244,6 +253,7 @@ def _build_from_json(cls, path: str | Path) -> GenericProvider: request_builder=request_builder, response_parser=response_parser, timeout=data.get("timeout", 60.0), + supports_openai_sse=response_shape != "anthropic", ) def reload(self) -> None: @@ -265,7 +275,7 @@ def reload(self) -> None: self.request_builder = fresh.request_builder self.response_parser = fresh.response_parser self.timeout = fresh.timeout - self._uses_default_parser = fresh._uses_default_parser + self._supports_openai_sse = fresh._supports_openai_sse @staticmethod def _default_request( @@ -292,7 +302,7 @@ def _snapshot_locked(self) -> _ConnectionSnapshot: timeout=self.timeout, request_builder=self.request_builder, response_parser=self.response_parser, - uses_default_parser=self._uses_default_parser, + supports_openai_sse=self._supports_openai_sse, ) def _headers(self, config: _ConnectionSnapshot) -> dict[str, str]: @@ -337,7 +347,7 @@ def complete( ) known_tools = {tool["name"] for tool in tools} - if on_delta is not None and config.uses_default_parser: + if on_delta is not None and config.supports_openai_sse: return self._stream(body, on_delta, known_tools, config) with self._open(body, config) as response: diff --git a/tests/providers/test_generic.py b/tests/providers/test_generic.py index cf310e8..c1d7d17 100644 --- a/tests/providers/test_generic.py +++ b/tests/providers/test_generic.py @@ -489,6 +489,65 @@ def test_custom_tool_call_paths(self): self.assertEqual(result.tool_calls[0].name, "read_file") self.assertEqual(result.tool_calls[0].arguments, {"path": "a.py"}) + def test_response_paths_config_still_streams(self): + path = self._write_config( + { + "url": "http://fake/answer", + "model": "my-model", + "response_paths": {"text": "result.answer"}, + } + ) + provider = GenericProvider.from_json(path) + + chunks = [ + {"choices": [{"delta": {"content": "hel"}}]}, + {"choices": [{"delta": {"content": "lo"}}]}, + {"choices": [{"delta": {}, "finish_reason": "stop"}]}, + ] + sse_body = ( + "".join(f"data: {json.dumps(c)}\n" for c in chunks) + + "data: [DONE]\n" + ).encode() + + deltas = [] + with mock.patch( + "pycodeloop.providers.generic.urllib.request.urlopen", + return_value=_FakeResponse(sse_body), + ): + result = provider.complete("sys", [], [], on_delta=deltas.append) + + self.assertEqual(deltas, ["hel", "lo"]) + self.assertEqual(result.text, "hello") + + def test_anthropic_response_shape_falls_back_to_a_single_on_delta_call( + self, + ): + path = self._write_config( + { + "url": "http://fake/answer", + "model": "my-model", + "response_shape": "anthropic", + } + ) + provider = GenericProvider.from_json(path) + + response_body = json.dumps( + { + "content": [{"type": "text", "text": "hello"}], + "usage": {"input_tokens": 3, "output_tokens": 1}, + } + ).encode() + + deltas = [] + with mock.patch( + "pycodeloop.providers.generic.urllib.request.urlopen", + return_value=_FakeResponse(response_body), + ): + result = provider.complete("sys", [], [], on_delta=deltas.append) + + self.assertEqual(deltas, ["hello"]) + self.assertEqual(result.text, "hello") + class TestGetProviderJsonDispatch(GenericProviderTestCase): def test_get_provider_loads_json_config_by_path(self):