-
Notifications
You must be signed in to change notification settings - Fork 1.4k
Add VercelAIAdapter.dump_messages to convert Pydantic AI messages to Vercel AI messages
#3392
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Add VercelAIAdapter.dump_messages to convert Pydantic AI messages to Vercel AI messages
#3392
Conversation
|
there's a missing for loop in the method inplementstion, will push a fix tomorrow! |
…dumping and IsStr - add dump_messages to base adapter class
VercelAIAdapter.dump_messages to convert Pydantic AI messages to Vercel AI messages
| user_ui_parts.extend(_convert_user_prompt_part(part)) | ||
| elif isinstance(part, ToolReturnPart | RetryPromptPart): | ||
| # Tool returns/errors don't create separate UI parts | ||
| # They're merged into the tool call in the assistant message |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Only if the retry prompt part has a tool call ID, otherwise it should be a text message
| local_builtin_returns: dict[str, BuiltinToolReturnPart] = {} | ||
| for part in msg.parts: | ||
| if isinstance(part, BuiltinToolReturnPart): | ||
| local_builtin_returns[part.tool_call_id] = part |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This can be a single dict comprehension
| # Combine consecutive text parts by checking the last UI part | ||
| if ui_parts and isinstance(ui_parts[-1], TextUIPart): | ||
| last_text = ui_parts[-1] | ||
| ui_parts[-1] = last_text.model_copy(update={'text': last_text.text + part.content}) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why not just last_text.text += part.content?
| elif isinstance(part, BaseToolCallPart): | ||
| if isinstance(part, BuiltinToolCallPart): | ||
| prefixed_id = ( | ||
| f'{BUILTIN_TOOL_CALL_ID_PREFIX}|{part.provider_name or ""}|{part.tool_call_id}' |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We shouldn't need to make any special IDs, as we can store metadata on the part, right? And Vercel AI already has a native way of specifying provider_executed tool calls
| tool_call_id=prefixed_id, | ||
| input=part.args_as_json_str(), | ||
| state='input-available', | ||
| provider_executed=True, |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Shouldn't we still store the metadata in this case?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
yeahp we should, I'm not even sure this case is reachable but I should've included the metadata
| ) | ||
| else: | ||
| tool_return = tool_returns.get(part.tool_call_id) | ||
| tool_error = tool_errors.get(part.tool_call_id) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We could have a single dict, right?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
makes it more verbose though, when we start getting into second level keys accesses
|
|
||
| @classmethod | ||
| @abstractmethod | ||
| def dump_messages(cls, messages: Sequence[ModelMessage]) -> list[MessageT]: |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I still wanted this method exactly as defined, just not abstractmethod :) That way subclasses are not required to implemented it, but if it's called it'll still raise NotImplementedError.
| def _dump_response_message( # noqa: C901 | ||
| msg: ModelResponse, | ||
| tool_returns: dict[str, ToolReturnPart], | ||
| tool_errors: dict[str, RetryPromptPart], |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can we combine these 2 dicts into 1?
| ) | ||
| ) | ||
| elif isinstance(part, BaseToolCallPart): | ||
| if isinstance(part, BuiltinToolCallPart): |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Instead of if isinstance(part, BaseToolCallPart): and inside it 2 branches for Builtin vs not builtin, I'd rather have top-level branches for if isinstance(part, ToolCallPart): and if isinstance(part, BuiltinToolCallPart):
…ptPart without tool_name, split BaseToolCallPart branches
| ) | ||
| else: | ||
| tool_return = tool_returns.get(part.tool_call_id) | ||
| tool_error = tool_errors.get(part.tool_call_id) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
makes it more verbose though, when we start getting into second level keys accesses
| tool_call_id=prefixed_id, | ||
| input=part.args_as_json_str(), | ||
| state='input-available', | ||
| provider_executed=True, |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
yeahp we should, I'm not even sure this case is reachable but I should've included the metadata
| call_provider_metadata=call_provider_metadata, | ||
| ) | ||
| ) | ||
| else: # pragma: no cover |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Let's cover this branch as well
Summary
Implements a
dump_messages()@classmethod for the VercelAIAdapter to convert Pydantic AI messages to Vercel AI format, enabling export of agent conversation history from Pydantic AI to Vercel AI protocol.This is the reverse operation of the existing
load_messages()method.Implementation
dump_messages():ModelMessageobjects (ModelRequest/ModelResponse)UIMessageobjects in Vercel AI format_id_generator()function to generate message idsCases Covered
The implementation handles:
Design Considerations
Tool return handling: Tool returns in ModelRequest are used to determine the state of tool calls in ModelResponse messages, not emitted as separate user messages.
Text concatenation: Consecutive TextPart instances are concatenated. When interrupted by non-text parts (tools, files, reasoning), subsequent text is separated with
\n\n.Builtin tool IDs: Uses the existing
BUILTIN_TOOL_CALL_ID_PREFIXpattern for consistency with other adapters.Message grouping: System and user parts within a ModelRequest are split into separate UIMessage objects when needed.
Caveats
Tool call input reconstruction: When tool returns appear in ModelRequest without the original tool call in the same message history, the
inputfield is set to an empty object{}since the original arguments are not available.No perfect roundtrip: Due to timestamp generation and UUID assignment,
dump_messages(load_messages(ui_msgs))will not produce identical objects, but will preserve semantic equivalence.Builtin tool return location: The implementation checks both the same ModelResponse (for builtin tools) and subsequent ModelRequest messages (for regular tools) to find tool returns.
Tests
Added tests for the following cases in test_vercel_ai.py