core/agent: restore OpenAI responses dispatch and provider_tools

This commit is contained in:
2026-08-01 00:34:15 +08:00
parent ad58d6cdc6
commit de7cae83e4
6 changed files with 489 additions and 17 deletions
+5
View File
@@ -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
+16 -3
View File
@@ -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"
+40 -13
View File
@@ -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
+111
View File
@@ -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"]
+1 -1
View File
@@ -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
+316
View File
@@ -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