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
5 changes: 5 additions & 0 deletions architecture/models.md
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,11 @@ Defines the contract: sync/async chat, embeddings, image generation, `supports_*
- **`OpenAICompatibleClient`** — native httpx adapter for OpenAI-compatible endpoints (NIM, vLLM, etc.)
- **`AnthropicClient`** — native httpx adapter for the Anthropic Messages API

Assistant messages can carry opaque `ProviderReplayData` tagged by provider type. The model facade and MCP
preserve it through tool turns and retry checkpoints. `ChatCompletionRequest.message_replay_data` aligns
these envelopes with canonical message dictionaries; only the matching adapter interprets its payload.
`ChatMessage.to_dict()` excludes replay data from traces.

### Client Factory

`create_model_client` routes by provider type to the appropriate adapter. Optionally wraps with:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
from data_designer.engine.mcp.errors import DuplicateToolNameError, MCPConfigurationError, MCPToolError
from data_designer.engine.mcp.registry import MCPToolDefinition
from data_designer.engine.model_provider import MCPProviderRegistry
from data_designer.engine.models.clients.types import ChatCompletionResponse, ToolCall
from data_designer.engine.models.clients.types import ChatCompletionResponse, ProviderReplayData, ToolCall
from data_designer.engine.models.utils import ChatMessage
from data_designer.engine.secret_resolver import SecretResolver

Expand Down Expand Up @@ -170,12 +170,15 @@ def process_completion_response(
ChatMessage.as_assistant(
content=response_content,
reasoning_content=reasoning_content or None,
replay_data=message.replay_data,
)
]

# Has tool calls - execute and return all messages
tool_call_dicts = _convert_canonical_tool_calls_to_dicts(tool_calls)
assistant_message = self._build_assistant_tool_message(response_content, tool_call_dicts, reasoning_content)
assistant_message = self._build_assistant_tool_message(
response_content, tool_call_dicts, reasoning_content, replay_data=message.replay_data
)
tool_messages = self._execute_tool_calls_from_canonical(tool_calls)

return [assistant_message, *tool_messages]
Expand Down Expand Up @@ -215,12 +218,15 @@ def refuse_completion_response(
ChatMessage.as_assistant(
content=response_content,
reasoning_content=reasoning_content or None,
replay_data=message.replay_data,
)
]

# Build assistant message with tool calls (same as normal)
tool_call_dicts = _convert_canonical_tool_calls_to_dicts(tool_calls)
assistant_message = self._build_assistant_tool_message(response_content, tool_call_dicts, reasoning_content)
assistant_message = self._build_assistant_tool_message(
response_content, tool_call_dicts, reasoning_content, replay_data=message.replay_data
)

# Build refusal messages instead of executing tools
refusal = refusal_message or DEFAULT_TOOL_REFUSAL_MESSAGE
Expand All @@ -241,6 +247,7 @@ def _build_assistant_tool_message(
response: str | None,
tool_calls: list[dict[str, Any]],
reasoning_content: str | None = None,
replay_data: ProviderReplayData | None = None,
) -> ChatMessage:
"""Build the assistant message containing tool call requests."""
tool_calls_payload = [
Expand All @@ -255,6 +262,7 @@ def _build_assistant_tool_message(
content=response or "",
reasoning_content=reasoning_content or None,
tool_calls=tool_calls_payload,
replay_data=replay_data,
)

def _execute_tool_calls_from_canonical(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
ChatCompletionChoice,
ChatCompletionRequest,
ChatCompletionResponse,
ProviderReplayData,
ToolCall,
Usage,
)
Expand Down Expand Up @@ -69,7 +70,7 @@ def merge_system_parts(parts: list[str | list[dict[str, Any]]]) -> str | list[di

def build_anthropic_payload(request: ChatCompletionRequest) -> dict[str, Any]:
"""Build an Anthropic Messages API payload from a canonical request."""
system_parts, messages = translate_request_messages(request.messages)
system_parts, messages = translate_request_messages(request.messages, request.message_replay_data)

payload: dict[str, Any] = {
"model": request.model,
Expand Down Expand Up @@ -126,6 +127,7 @@ def parse_anthropic_response(response_json: dict[str, Any]) -> ChatCompletionRes
content="\n".join(text_parts) if text_parts else None,
reasoning_content="\n".join(thinking_parts) if thinking_parts else None,
tool_calls=tool_calls,
replay_data=ProviderReplayData(provider_type="anthropic", payload=content_blocks),
)

raw_usage = response_json.get("usage")
Expand All @@ -148,12 +150,13 @@ def parse_anthropic_response(response_json: dict[str, Any]) -> ChatCompletionRes

def translate_request_messages(
messages: list[dict[str, Any]],
message_replay_data: list[ProviderReplayData | None] | None = None,
) -> tuple[list[str | list[dict[str, Any]]], list[dict[str, Any]]]:
system_parts: list[str | list[dict[str, Any]]] = []
translated_messages: list[dict[str, Any]] = []
pending_tool_results: list[dict[str, Any]] = []

for msg in messages:
for index, msg in enumerate(messages):
role = msg.get("role")
if role == "system":
system_content = extract_system_content(msg.get("content"))
Expand All @@ -169,7 +172,8 @@ def translate_request_messages(
translated_messages.append({"role": "user", "content": pending_tool_results})
pending_tool_results = []

translated_messages.append(translate_non_tool_message(msg))
replay_data = message_replay_data[index] if message_replay_data else None
translated_messages.append(translate_non_tool_message(msg, replay_data))

if pending_tool_results:
translated_messages.append({"role": "user", "content": pending_tool_results})
Expand Down Expand Up @@ -204,11 +208,18 @@ def extract_system_content(content: Any) -> str | list[dict[str, Any]] | None:
return "\n".join(text_parts)


def translate_non_tool_message(msg: dict[str, Any]) -> dict[str, Any]:
def translate_non_tool_message(msg: dict[str, Any], replay_data: ProviderReplayData | None = None) -> dict[str, Any]:
role = msg.get("role")
if role not in {"user", "assistant"}:
raise ValueError(f"Anthropic adapter does not support message role {role!r}.")

# Signed thinking blocks must be replayed with the entire original turn in order.
if role == "assistant" and replay_data is not None and replay_data.provider_type == "anthropic":
content = replay_data.payload
if not isinstance(content, list) or not all(isinstance(block, dict) for block in content):
raise ValueError("Anthropic replay data must be a list of content blocks")
return {"role": role, "content": content}

content = msg.get("content")
if role == "assistant" and msg.get("tool_calls"):
translated_content = translate_content_blocks(content)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -48,12 +48,21 @@ class ToolCall:
arguments_json: str


@dataclass(frozen=True)
class ProviderReplayData:
"""Opaque provider-owned data required to replay an assistant message."""

provider_type: str
payload: object


@dataclass
class AssistantMessage:
content: str | None = None
reasoning_content: str | None = None
tool_calls: list[ToolCall] = field(default_factory=list)
images: list[ImagePayload] = field(default_factory=list)
replay_data: ProviderReplayData | None = None


@dataclass
Expand All @@ -74,6 +83,12 @@ class ChatCompletionRequest:
timeout: float | None = None
extra_body: dict[str, Any] | None = None
extra_headers: dict[str, str] | None = None
# Entries align with messages; an empty list means no replay data.
message_replay_data: list[ProviderReplayData | None] = field(default_factory=list)

def __post_init__(self) -> None:
if self.message_replay_data and len(self.message_replay_data) != len(self.messages):
raise ValueError("message_replay_data must match the number of messages")


@dataclass
Expand Down Expand Up @@ -151,7 +166,9 @@ class TransportKwargs:
- ``headers``: Extra HTTP headers to attach to the outgoing request.
"""

_META_FIELDS: ClassVar[frozenset[str]] = frozenset({"extra_body", "extra_headers", "timeout"})
_META_FIELDS: ClassVar[frozenset[str]] = frozenset(
{"extra_body", "extra_headers", "timeout", "message_replay_data"}
)

body: dict[str, Any]
headers: dict[str, str]
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
EmbeddingResponse,
ImageGenerationRequest,
ImageGenerationResponse,
ProviderReplayData,
Usage,
)
from data_designer.engine.models.errors import (
Expand Down Expand Up @@ -245,7 +246,9 @@ def completion(
if not allow_multiple_choices:
kwargs = self._drop_multi_choice_request_fields(kwargs)
try:
request = self._build_chat_completion_request(message_payloads, kwargs)
request = self._build_chat_completion_request(
message_payloads, kwargs, message_replay_data=[message.replay_data for message in messages]
)
response = self._client.completion(request)
logger.debug(
f"Received completion from model {self.model_name!r}",
Expand Down Expand Up @@ -282,7 +285,9 @@ async def acompletion(
if not allow_multiple_choices:
kwargs = self._drop_multi_choice_request_fields(kwargs)
try:
request = self._build_chat_completion_request(message_payloads, kwargs)
request = self._build_chat_completion_request(
message_payloads, kwargs, message_replay_data=[message.replay_data for message in messages]
)
response = await self._client.acompletion(request)
logger.debug(
f"Received completion from model {self.model_name!r}",
Expand Down Expand Up @@ -430,7 +435,13 @@ def generate(
# No tool calls remaining to process
response = (completion_response.message.content or "").strip()
reasoning_trace = completion_response.message.reasoning_content
messages.append(ChatMessage.as_assistant(content=response, reasoning_content=reasoning_trace or None))
messages.append(
ChatMessage.as_assistant(
content=response,
reasoning_content=reasoning_trace or None,
replay_data=completion_response.message.replay_data,
)
)
parse_attempts += 1

try:
Expand Down Expand Up @@ -549,7 +560,13 @@ async def agenerate(

response = (completion_response.message.content or "").strip()
reasoning_trace = completion_response.message.reasoning_content
messages.append(ChatMessage.as_assistant(content=response, reasoning_content=reasoning_trace or None))
messages.append(
ChatMessage.as_assistant(
content=response,
reasoning_content=reasoning_trace or None,
replay_data=completion_response.message.replay_data,
)
)
parse_attempts += 1

try:
Expand Down Expand Up @@ -827,10 +844,18 @@ def _get_mcp_facade(self, tool_alias: str | None) -> MCPFacade | None:
raise MCPConfigurationError(f"Tool alias {tool_alias!r} is not registered.") from exc

def _build_chat_completion_request(
self, messages: list[dict[str, Any]], kwargs: dict[str, Any]
self,
messages: list[dict[str, Any]],
kwargs: dict[str, Any],
*,
message_replay_data: list[ProviderReplayData | None],
) -> ChatCompletionRequest:
"""Build a ChatCompletionRequest from message payloads and consolidated kwargs."""
request_fields: dict[str, Any] = {"model": self.model_name, "messages": messages}
request_fields: dict[str, Any] = {
"model": self.model_name,
"messages": messages,
"message_replay_data": message_replay_data,
}
metadata: dict[str, Any] = {}

for key, value in kwargs.items():
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
from pydantic import BaseModel

if TYPE_CHECKING:
from data_designer.engine.models.clients.types import ChatCompletionResponse
from data_designer.engine.models.clients.types import ChatCompletionResponse, ProviderReplayData


class GenerationTruncationReason(str, Enum):
Expand Down Expand Up @@ -101,13 +101,15 @@ class ChatMessage:
Each tool call contains 'id', 'type', and 'function' keys.
tool_call_id: Optional ID linking a tool response to its corresponding
tool call. Required for messages with role='tool'.
replay_data: Opaque provider-owned data excluded from trace serialization.
"""

role: Literal["user", "assistant", "system", "tool"]
content: str | list[dict[str, Any]] = ""
reasoning_content: str | None = None
tool_calls: list[dict[str, Any]] = field(default_factory=list)
tool_call_id: str | None = None
replay_data: ProviderReplayData | None = None

def to_dict(self) -> dict[str, Any]:
"""Convert the message to a dictionary format for API calls.
Expand Down Expand Up @@ -139,13 +141,15 @@ def as_assistant(
content: str = "",
reasoning_content: str | None = None,
tool_calls: list[dict[str, Any]] | None = None,
replay_data: ProviderReplayData | None = None,
) -> ChatMessage:
"""Create an assistant message."""
return cls(
role="assistant",
content=content,
reasoning_content=reasoning_content,
tool_calls=tool_calls or [],
replay_data=replay_data,
)

@classmethod
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,12 @@
from data_designer.engine.mcp.facade import DEFAULT_TOOL_REFUSAL_MESSAGE, MCPFacade
from data_designer.engine.mcp.registry import MCPToolDefinition, MCPToolResult
from data_designer.engine.model_provider import MCPProviderRegistry
from data_designer.engine.models.clients.types import AssistantMessage, ChatCompletionResponse, ToolCall
from data_designer.engine.models.clients.types import (
AssistantMessage,
ChatCompletionResponse,
ProviderReplayData,
ToolCall,
)


def _make_response(
Expand Down Expand Up @@ -166,6 +171,26 @@ def test_process_completion_preserves_content(
assert messages[0].content == "Hello, I can help with that."


@pytest.mark.parametrize("method", ["process_completion_response", "refuse_completion_response"])
@pytest.mark.parametrize("tool_calls", [[], [ToolCall(id="call-1", name="lookup", arguments_json="{}")]])
def test_completion_preserves_opaque_replay_data(
stub_mcp_facade: MCPFacade, method: str, tool_calls: list[ToolCall], monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(
mcp_io,
"list_tools",
MagicMock(return_value=(MCPToolDefinition(name="lookup", description="Lookup", input_schema={}),)),
)
monkeypatch.setattr(mcp_io, "call_tools", MagicMock(return_value=[MCPToolResult(content="Result")]))
replay_data = ProviderReplayData(provider_type="future-provider", payload={"opaque": object()})
response = ChatCompletionResponse(
message=AssistantMessage(content="Answer", tool_calls=tool_calls, replay_data=replay_data)
)
messages = getattr(stub_mcp_facade, method)(response)
assert len(messages) == 1 + len(tool_calls)
assert messages[0].replay_data is replay_data


def test_process_completion_preserves_reasoning_content(
stub_mcp_facade: MCPFacade,
mock_completion_response_with_reasoning: ChatCompletionResponse,
Expand Down
Loading
Loading