From de7cae83e4a7db0e540c49bf4a966137679ff9ca Mon Sep 17 00:00:00 2001 From: worldmozara Date: Sat, 1 Aug 2026 00:34:15 +0800 Subject: [PATCH] core/agent: restore OpenAI responses dispatch and provider_tools --- src/plyngent/agent/chat.py | 5 + src/plyngent/agent/compact.py | 19 +- src/plyngent/agent/loop.py | 53 +++- src/plyngent/agent/responses_dispatch.py | 111 +++++++ src/plyngent/cli/state.py | 2 +- tests/test_agent/test_responses_dispatch.py | 316 ++++++++++++++++++++ 6 files changed, 489 insertions(+), 17 deletions(-) create mode 100644 src/plyngent/agent/responses_dispatch.py create mode 100644 tests/test_agent/test_responses_dispatch.py diff --git a/src/plyngent/agent/chat.py b/src/plyngent/agent/chat.py index 2e7d992..bd97956 100644 --- a/src/plyngent/agent/chat.py +++ b/src/plyngent/agent/chat.py @@ -172,6 +172,7 @@ class ChatAgent: todo_nag_strategy: TodoNagStrategy directive_reminder_tokens: int directive_reminder_text: str | None + provider_tools: list[dict[str, object]] messages: list[AnyChatMessage] session_usage: TokenUsage last_turn_usage: TokenUsage @@ -205,6 +206,7 @@ class ChatAgent: directive_reminder_text: str | None = None, peak_prompt_tokens: int = 0, reminder_last_band: int = 0, + provider_tools: Sequence[dict[str, object]] | None = None, ) -> None: self.client = client self.model = model @@ -224,6 +226,7 @@ class ChatAgent: self.directive_reminder_tokens = max(0, int(directive_reminder_tokens)) text = directive_reminder_text.strip() if directive_reminder_text else "" self.directive_reminder_text = text or None + self.provider_tools = [dict(item) for item in provider_tools] if provider_tools else [] self.messages = list(messages) if messages is not None else [] self.session_usage = TokenUsage() self.last_turn_usage = TokenUsage() @@ -456,6 +459,7 @@ class ChatAgent: directive_reminder_text=self.directive_reminder_text, reminder_last_band=self.reminder_last_band, on_reminder_band=self._on_reminder_band, + provider_tools=self.provider_tools or None, ): if isinstance(event, UsageEvent): turn_rounds += 1 @@ -589,6 +593,7 @@ class ChatAgent: todo_nag_strategy="none", directive_reminder_tokens=0, messages=history, + provider_tools=self.provider_tools or None, ) async for event in aside.run(text): yield event diff --git a/src/plyngent/agent/compact.py b/src/plyngent/agent/compact.py index dcb03da..871ad91 100644 --- a/src/plyngent/agent/compact.py +++ b/src/plyngent/agent/compact.py @@ -1,6 +1,6 @@ from __future__ import annotations -from typing import TYPE_CHECKING, cast +from typing import TYPE_CHECKING, Any, cast from msgspec import UNSET @@ -143,11 +143,24 @@ async def summarize_messages( model=model, temperature=temperature if temperature is not None else UNSET, ) + kind = getattr(client, "kind", "chat_completions") response: ChatCompletionResponse - if getattr(client, "kind", "chat_completions") == "chat_completions": + if kind == "chat_completions": response = await cast("OpenAICompatibleClient", client).chat_completions(param, stream=False) + elif kind == "responses": + from .responses_dispatch import dispatch_responses + + response = cast( + "ChatCompletionResponse", + await dispatch_responses( + cast("Any", client), + param, + provider_tools=None, + stream=False, + ), + ) else: - msg = f"summarization for client.kind={client.kind!r} is not implemented" + msg = f"summarization for client.kind={kind!r} is not implemented" raise NotImplementedError(msg) if not response.choices: msg = "summarization response contained no choices" diff --git a/src/plyngent/agent/loop.py b/src/plyngent/agent/loop.py index 50213c6..25cf30b 100644 --- a/src/plyngent/agent/loop.py +++ b/src/plyngent/agent/loop.py @@ -3,7 +3,7 @@ from __future__ import annotations import asyncio import inspect from collections.abc import AsyncIterator -from typing import TYPE_CHECKING, cast +from typing import TYPE_CHECKING, Any, cast import msgspec from msgspec import UNSET @@ -54,6 +54,7 @@ from .usage import resolve_round_usage, token_usage_from_api if TYPE_CHECKING: from collections.abc import AsyncIterator, Awaitable, Callable, Sequence + from plyngent.lmproto.openai.client import OpenAIClient from plyngent.lmproto.openai_compatible.model import AnyChatMessage, AnyToolItem from .todo_nag import TodoNagStrategy @@ -193,25 +194,37 @@ async def _dispatch_chat_completions( param: ChatCompletionsParam, *, stream: bool, + provider_tools: Sequence[dict[str, Any]] | None = None, ) -> ChatCompletionResponse | AsyncIterator[ChatCompletionChunk]: kind = getattr(client, "kind", "chat_completions") - if kind != "chat_completions": - if kind == "messages": - msg = "Anthropic /messages dispatch is not wired in the agent loop yet" - else: - msg = "OpenAI /responses dispatch is not wired in the agent loop yet" + if kind == "chat_completions": + cc = cast("OpenAICompatibleClient", client) + return await cc.chat_completions(param, stream=stream) + if kind == "responses": + from .responses_dispatch import dispatch_responses + + return await dispatch_responses( + cast("OpenAIClient", client), + param, + provider_tools=provider_tools, + stream=stream, + ) + if kind == "messages": + msg = "Anthropic /messages dispatch is not wired in the agent loop yet" raise NotImplementedError(msg) - cc = cast("OpenAICompatibleClient", client) - return await cc.chat_completions(param, stream=stream) + msg = f"client kind {kind!r} is not supported by the agent loop" + raise NotImplementedError(msg) async def _non_stream_round( client: AnyLLMClient, param: ChatCompletionsParam, + *, + provider_tools: Sequence[dict[str, Any]] | None = None, ) -> AsyncIterator[AgentEvent]: response = cast( "ChatCompletionResponse", - await _dispatch_chat_completions(client, param, stream=False), + await _dispatch_chat_completions(client, param, stream=False, provider_tools=provider_tools), ) if not response.choices: msg = "chat completion response contained no choices" @@ -238,6 +251,8 @@ async def _non_stream_round( async def _stream_round( client: AnyLLMClient, param: ChatCompletionsParam, + *, + provider_tools: Sequence[dict[str, Any]] | None = None, ) -> AsyncIterator[AgentEvent]: """Stream one completion; yield text deltas as chunks arrive, then assistant. @@ -252,7 +267,7 @@ async def _stream_round( ) stream = cast( "AsyncIterator[ChatCompletionChunk]", - await _dispatch_chat_completions(client, stream_param, stream=True), + await _dispatch_chat_completions(client, stream_param, stream=True, provider_tools=provider_tools), ) content_parts: list[str] = [] reasoning_parts: list[str] = [] @@ -317,15 +332,16 @@ async def _assistant_round( messages: list[AnyChatMessage], *, stream: bool, + provider_tools: Sequence[dict[str, Any]] | None = None, ) -> AsyncIterator[AgentEvent]: """One model turn: yield events and append the assistant message to ``messages``.""" if stream: - async for event in _stream_round(client, param): + async for event in _stream_round(client, param, provider_tools=provider_tools): if isinstance(event, AssistantMessageEvent): messages.append(event.message) yield event return - async for event in _non_stream_round(client, param): + async for event in _non_stream_round(client, param, provider_tools=provider_tools): if isinstance(event, AssistantMessageEvent): messages.append(event.message) yield event @@ -390,6 +406,7 @@ async def run_chat_loop( # noqa: C901, PLR0912 — multi-phase tool loop directive_reminder_text: str | None = None, reminder_last_band: int = 0, on_reminder_band: Callable[[int], Awaitable[None] | None] | None = None, + provider_tools: Sequence[dict[str, Any]] | None = None, ) -> AsyncIterator[AgentEvent]: """Multi-round chat/tool loop; mutates ``messages`` in place and yields events. @@ -398,6 +415,10 @@ async def run_chat_loop( # noqa: C901, PLR0912 — multi-phase tool loop Multiple tool calls in one round run in parallel when ``parallel_tools``. Request payloads may shrink older tool results when over ``max_context_tokens``. + *provider_tools* are hosted/provider-side tools (OpenAI Responses only, e.g. + ``web_search``) merged into the request as opaque dicts; never executed by the + :class:`ToolRegistry`. Ignored by non-Responses clients. + When *todo_stack* is set and still needs review after a natural stop (open items, or non-empty stack untouched this turn), injects a review nag (channel from *todo_nag_strategy*) once so the model reconciles unfinished work. @@ -442,7 +463,13 @@ async def run_chat_loop( # noqa: C901, PLR0912 — multi-phase tool loop pre_len = len(messages) last_usage_event: UsageEvent | None = None - async for event in _assistant_round(client, param, messages, stream=stream): + async for event in _assistant_round( + client, + param, + messages, + stream=stream, + provider_tools=provider_tools, + ): if isinstance(event, UsageEvent): # Next rounds scale char-estimates by real/resolved prompt size. prompt_tokens_hint = event.usage.prompt_tokens diff --git a/src/plyngent/agent/responses_dispatch.py b/src/plyngent/agent/responses_dispatch.py new file mode 100644 index 0000000..e2fd0a8 --- /dev/null +++ b/src/plyngent/agent/responses_dispatch.py @@ -0,0 +1,111 @@ +"""OpenAI Responses dispatch: present ``/responses`` as chat completions to the loop. + +Agent memory and events stay chat-completions-shaped; only the transport uses +``POST /responses``. DeepSeek / openai-compatible paths never enter this module. + +This restores the dispatch that lived on the removed ``ResponsesChatClient`` +wrapper, now as plain functions called by the kind-based switch in the loop. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +import msgspec +from msgspec import UNSET + +from plyngent.agent.responses_bridge import ( + chat_param_to_responses_kwargs, + finish_reason_chunk, + reasoning_delta_chunk, + response_to_assistant_message, + response_to_chat_completion, + responses_status_to_finish_reason, + text_delta_chunk, + tool_call_chunks_from_response, + usage_chunk_from_response, +) +from plyngent.lmproto.openai.model import Response as ResponseModel +from plyngent.lmproto.openai.model import ResponsesCreateParam + +if TYPE_CHECKING: + from collections.abc import AsyncIterator, Sequence + + from plyngent.lmproto.openai.client import OpenAIClient + from plyngent.lmproto.openai_compatible.model import ( + ChatCompletionChunk, + ChatCompletionResponse, + ChatCompletionsParam, + ) + +# Responses stream event types that carry reasoning text deltas. +_REASONING_DELTA_TYPES = {"response.reasoning_summary_text.delta", "response.reasoning_text.delta"} + + +async def _stream_as_chat_chunks( + client: OpenAIClient, + create: ResponsesCreateParam, + *, + model: str, +) -> AsyncIterator[ChatCompletionChunk]: + """Yield chat-completions chunks synthesized from a Responses SSE stream. + + Text / reasoning deltas stream as they arrive; tool calls, finish_reason, + and usage are emitted once on ``response.completed`` (the full Response is + needed to decode function calls + usage). A stream that ends without + ``response.completed`` yields nothing here — the loop treats that as a + missing terminal signal. + """ + stream = await client.responses(create, stream=True) + final: ResponseModel | None = None + async for event in stream: + etype = event.type + if etype == "response.output_text.delta" and isinstance(event.delta, str) and event.delta: + yield text_delta_chunk(model=model, content=event.delta) + continue + if etype in _REASONING_DELTA_TYPES and isinstance(event.delta, str) and event.delta: + yield reasoning_delta_chunk(model=model, content=event.delta) + continue + if etype == "response.completed" and event.response is not UNSET: + try: + final = msgspec.convert(event.response, ResponseModel) + except TypeError, ValueError, msgspec.ValidationError: + final = None + + if final is None: + # Stream ended without response.completed — signal missing terminal. + # The loop will treat empty + no finish_reason as a glitch. + return + + assistant = response_to_assistant_message(final) + has_tools = assistant.tool_calls is not UNSET and bool(assistant.tool_calls) + finish = responses_status_to_finish_reason(final, has_tool_calls=has_tools) + for chunk in tool_call_chunks_from_response(final, model=model): + yield chunk + yield finish_reason_chunk(model=model, finish_reason=finish) + usage = usage_chunk_from_response(final, model=model) + if usage is not None: + yield usage + + +async def dispatch_responses( + client: OpenAIClient, + param: ChatCompletionsParam, + *, + provider_tools: Sequence[dict[str, Any]] | None = None, + stream: bool = False, +) -> ChatCompletionResponse | AsyncIterator[ChatCompletionChunk]: + """Run one Responses turn and return a chat-completions-shaped result. + + *provider_tools* are hosted tools (web_search, file_search, …) as opaque + dicts merged after local function tools; never executed by the registry. + """ + kwargs = chat_param_to_responses_kwargs(param, provider_tools=provider_tools) + create = ResponsesCreateParam(**kwargs) + if stream: + return _stream_as_chat_chunks(client, create, model=param.model) + response = await client.responses(create, stream=False) + return response_to_chat_completion(response) + + +__all__ = ["dispatch_responses"] diff --git a/src/plyngent/cli/state.py b/src/plyngent/cli/state.py index 9fcf9e9..cbf4eac 100644 --- a/src/plyngent/cli/state.py +++ b/src/plyngent/cli/state.py @@ -75,7 +75,6 @@ class ReplState: _remote_models_error: str | None = field(default=None, init=False, repr=False) def __post_init__(self) -> None: - # DeepSeek client uses a compatible but distinct param type; treat as ChatClient. self.client = create_client(self.provider, model=self.model) self.workspace = Path(self.workspace).expanduser().resolve() self.instance_state.workspace_root = self.workspace @@ -266,6 +265,7 @@ class ReplState: directive_reminder_text=agent_cfg.directive_reminder_text or None, peak_prompt_tokens=peak, reminder_last_band=band, + provider_tools=self.effective_provider.provider_tools or None, ) if last_req is not None and not last_req.is_zero(): agent.last_request_usage = last_req diff --git a/tests/test_agent/test_responses_dispatch.py b/tests/test_agent/test_responses_dispatch.py new file mode 100644 index 0000000..01f7338 --- /dev/null +++ b/tests/test_agent/test_responses_dispatch.py @@ -0,0 +1,316 @@ +"""Tests for kind-based OpenAI Responses dispatch in the agent loop.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, Literal, cast, overload + +import pytest +from msgspec import UNSET + +from plyngent.agent import ( + AssistantMessageEvent, + TextDeltaEvent, + ToolCallEvent, + ToolRegistry, + ToolResultEvent, + UsageEvent, + run_chat_loop, + tool, +) +from plyngent.agent.responses_bridge import chat_param_to_responses_kwargs +from plyngent.agent.responses_dispatch import dispatch_responses +from plyngent.lmproto.openai.model import Response, ResponsesCreateParam, ResponseStreamEvent +from plyngent.lmproto.openai_compatible.model import ( + AssistantChatMessage, + ChatCompletionsParam, + SystemChatMessage, + ToolFunction, + ToolFunctionItem, + UserChatMessage, +) + +if TYPE_CHECKING: + from collections.abc import AsyncIterator, Sequence + + +def _completed_response( + *, + text: str = "hello", + tool_calls: list[dict[str, Any]] | None = None, + usage: dict[str, Any] | None = None, + status: str = "completed", +) -> Response: + output: list[dict[str, Any]] = [] + if text: + output.append( + { + "id": "msg_1", + "type": "message", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": text, "annotations": []}], + } + ) + if tool_calls: + output.extend(tool_calls) + return Response( + id="resp_test", + created_at=1, + model="gpt-test", + status=cast("Any", status), + output=output, + usage=usage if usage is not None else {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15}, + ) + + +class ScriptedResponsesClient: + """Fake OpenAIClient that scripts ``responses()`` results.""" + + kind: str = "responses" + calls: list[ResponsesCreateParam] + _non_stream: list[Response] + _stream_events: list[list[ResponseStreamEvent]] + + def __init__( + self, + *, + non_stream: Sequence[Response] | None = None, + stream_events: Sequence[Sequence[ResponseStreamEvent]] | None = None, + ) -> None: + self.calls = [] + self._non_stream = list(non_stream or ()) + self._stream_events = [list(events) for events in (stream_events or ())] + + @overload + async def responses(self, param: ResponsesCreateParam, *, stream: Literal[False] = False) -> Response: ... + + @overload + async def responses( + self, param: ResponsesCreateParam, *, stream: Literal[True] + ) -> AsyncIterator[ResponseStreamEvent]: ... + + async def responses( + self, param: ResponsesCreateParam, *, stream: bool = False + ) -> Response | AsyncIterator[ResponseStreamEvent]: + self.calls.append(param) + if stream: + if not self._stream_events: + msg = "no more scripted stream events" + raise RuntimeError(msg) + events = self._stream_events.pop(0) + + async def _gen() -> AsyncIterator[ResponseStreamEvent]: + for event in events: + yield event + + return _gen() + if not self._non_stream: + msg = "no more scripted responses" + raise RuntimeError(msg) + return self._non_stream.pop(0) + + +async def test_dispatch_responses_non_stream_text() -> None: + client = ScriptedResponsesClient(non_stream=[_completed_response(text="hi there")]) + param = ChatCompletionsParam( + model="gpt-test", + messages=[UserChatMessage(content="hello")], + ) + result = await dispatch_responses(cast("Any", client), param, stream=False) + assert result.choices[0].message.content == "hi there" + assert result.choices[0].finish_reason == "stop" + assert len(client.calls) == 1 + assert client.calls[0].store is False + + +async def test_dispatch_responses_merges_provider_tools() -> None: + client = ScriptedResponsesClient(non_stream=[_completed_response()]) + param = ChatCompletionsParam( + model="gpt-test", + messages=[UserChatMessage(content="search")], + tools=[ + ToolFunctionItem( + function=ToolFunction( + name="read_file", + description="Read a file", + parameters={"type": "object", "properties": {}}, + ) + ) + ], + ) + _ = await dispatch_responses( + cast("Any", client), + param, + provider_tools=[{"type": "web_search"}], + stream=False, + ) + create = client.calls[0] + assert create.tools is not UNSET + tools = cast("list[Any]", create.tools) + assert len(tools) == 2 + # Local function tool first, hosted tool second. + names_or_types: list[str] = [] + for item in tools: + if isinstance(item, dict): + names_or_types.append(str(item.get("type", ""))) + else: + names_or_types.append(getattr(item, "name", "")) + assert "read_file" in names_or_types + assert "web_search" in names_or_types + + +async def test_dispatch_responses_stream_text_and_usage() -> None: + final = _completed_response(text="streamed") + # Final response body is carried on the completed event as a dict. + import msgspec + + final_dict = msgspec.to_builtins(final) + events = [ + ResponseStreamEvent(type="response.output_text.delta", delta="stream"), + ResponseStreamEvent(type="response.output_text.delta", delta="ed"), + ResponseStreamEvent(type="response.completed", response=final_dict), + ] + client = ScriptedResponsesClient(stream_events=[events]) + param = ChatCompletionsParam( + model="gpt-test", + messages=[UserChatMessage(content="hi")], + ) + stream = await dispatch_responses(cast("Any", client), param, stream=True) + chunks = [chunk async for chunk in cast("Any", stream)] + # text deltas + finish + usage + texts = [c.choices[0].delta.content for c in chunks if c.choices and isinstance(c.choices[0].delta.content, str)] + assert texts == ["stream", "ed"] + assert any(c.choices and c.choices[0].finish_reason == "stop" for c in chunks) + assert any(token_usage_present(c) for c in chunks) + + +def token_usage_present(chunk: object) -> bool: + usage = getattr(chunk, "usage", UNSET) + return usage is not UNSET and usage is not None + + +async def test_run_chat_loop_responses_kind_text() -> None: + client = ScriptedResponsesClient(non_stream=[_completed_response(text="done")]) + messages: list[Any] = [UserChatMessage(content="go")] + events = [ + event + async for event in run_chat_loop( + cast("Any", client), + messages, + model="gpt-test", + stream=False, + ) + ] + assert any(isinstance(e, TextDeltaEvent) and e.content == "done" for e in events) + assert any(isinstance(e, AssistantMessageEvent) for e in events) + assert any(isinstance(e, UsageEvent) for e in events) + assert isinstance(messages[-1], AssistantChatMessage) + assert messages[-1].content == "done" + + +async def test_run_chat_loop_responses_kind_tool_round() -> None: + tool_call_item = { + "type": "function_call", + "call_id": "call_1", + "name": "add", + "arguments": '{"a": 1, "b": 2}', + "status": "completed", + } + first = _completed_response(text="", tool_calls=[tool_call_item]) + second = _completed_response(text="sum is 3") + client = ScriptedResponsesClient(non_stream=[first, second]) + + @tool(register=False) + def add(a: int, b: int) -> str: + return str(a + b) + + registry = ToolRegistry([add]) + + messages: list[Any] = [UserChatMessage(content="add 1 and 2")] + events = [ + event + async for event in run_chat_loop( + cast("Any", client), + messages, + model="gpt-test", + tools=registry, + stream=False, + ) + ] + assert any(isinstance(e, ToolCallEvent) for e in events) + assert any(isinstance(e, ToolResultEvent) and "3" in e.message.content for e in events) + assert any(isinstance(e, AssistantMessageEvent) and e.message.content == "sum is 3" for e in events) + assert len(client.calls) == 2 + # Second turn must include the function_call_output from the tool result. + from plyngent.lmproto.openai.model import ResponseFunctionToolCallOutput + + second_input = client.calls[1].input + assert isinstance(second_input, list) + assert any( + isinstance(item, ResponseFunctionToolCallOutput) + or (isinstance(item, dict) and item.get("type") == "function_call_output") + for item in second_input + ) + + +async def test_run_chat_loop_responses_provider_tools_passed() -> None: + client = ScriptedResponsesClient(non_stream=[_completed_response(text="ok")]) + messages: list[Any] = [UserChatMessage(content="hi")] + _ = [ + event + async for event in run_chat_loop( + cast("Any", client), + messages, + model="gpt-test", + stream=False, + provider_tools=[{"type": "web_search"}], + ) + ] + create = client.calls[0] + assert create.tools is not UNSET + tools = cast("list[Any]", create.tools) + assert any(isinstance(t, dict) and t.get("type") == "web_search" for t in tools) + + +async def test_dispatch_stream_missing_completed_yields_no_terminal() -> None: + """Stream without response.completed produces only early deltas (if any).""" + events = [ + ResponseStreamEvent(type="response.output_text.delta", delta="partial"), + # no completed event + ] + client = ScriptedResponsesClient(stream_events=[events]) + param = ChatCompletionsParam(model="gpt-test", messages=[UserChatMessage(content="x")]) + stream = await dispatch_responses(cast("Any", client), param, stream=True) + chunks = [chunk async for chunk in cast("Any", stream)] + assert len(chunks) == 1 # only the text delta + assert chunks[0].choices[0].delta.content == "partial" + assert chunks[0].choices[0].finish_reason is None or chunks[0].choices[0].finish_reason is UNSET + + +def test_chat_param_to_responses_kwargs_system_and_tools() -> None: + param = ChatCompletionsParam( + model="gpt-test", + messages=[ + SystemChatMessage(content="Be brief."), + UserChatMessage(content="hi"), + ], + tools=[ToolFunctionItem(function=ToolFunction(name="f", description="d", parameters={"type": "object"}))], + ) + kwargs = chat_param_to_responses_kwargs(param, provider_tools=[{"type": "web_search"}]) + assert kwargs["instructions"] == "Be brief." + assert kwargs["store"] is False + assert len(kwargs["tools"]) == 2 + + +async def test_unknown_kind_not_implemented() -> None: + class FakeOther: + kind = "weird" + + with pytest.raises(NotImplementedError, match="weird"): + async for _ in run_chat_loop( + cast("Any", FakeOther()), + [UserChatMessage(content="hi")], + model="x", + stream=False, + ): + pass