Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 16 additions & 6 deletions pycodeloop/providers/generic.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -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"
Expand All @@ -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)
Expand All @@ -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()

Expand All @@ -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(
Expand All @@ -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:
Expand All @@ -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(
Expand All @@ -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]:
Expand Down Expand Up @@ -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:
Expand Down
59 changes: 59 additions & 0 deletions tests/providers/test_generic.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
Loading