Fimeg/letta-server
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

fix: fix pywright lint errors caused by bad use of pydantic (#3170)

Browse Source
This commit is contained in:
Charles Packer
2025-07-06 12:58:21 -07:00
committed by GitHub
parent 72b472896d
commit bbcc948213
4 changed files with 68 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
letta/schemas/letta_base.py
View File

@@ -97,7 +97,7 @@ class LettaBase(BaseModel):
class OrmMetadataBase(LettaBase):
# metadata fields
created_by_id: Optional[str] = Field(None, description="The id of the user that made this object.")
last_updated_by_id: Optional[str] = Field(None, description="The id of the user that made this object.")
created_at: Optional[datetime] = Field(None, description="The timestamp when the object was created.")
updated_at: Optional[datetime] = Field(None, description="The timestamp when the object was last updated.")
created_by_id: Optional[str] = Field(default=None, description="The id of the user that made this object.")
last_updated_by_id: Optional[str] = Field(default=None, description="The id of the user that made this object.")
created_at: Optional[datetime] = Field(default=None, description="The timestamp when the object was created.")
updated_at: Optional[datetime] = Field(default=None, description="The timestamp when the object was last updated.")

22
letta/schemas/letta_message.py
View File

@@ -72,7 +72,7 @@ class SystemMessage(LettaMessage):
content (str): The message content sent by the system
"""
message_type: Literal[MessageType.system_message] = Field(MessageType.system_message, description="The type of the message.")
message_type: Literal[MessageType.system_message] = Field(default=MessageType.system_message, description="The type of the message.")
content: str = Field(..., description="The message content sent by the system")
@@ -87,7 +87,7 @@ class UserMessage(LettaMessage):
content (Union[str, List[LettaUserMessageContentUnion]]): The message content sent by the user (can be a string or an array of multi-modal content parts)
"""
message_type: Literal[MessageType.user_message] = Field(MessageType.user_message, description="The type of the message.")
message_type: Literal[MessageType.user_message] = Field(default=MessageType.user_message, description="The type of the message.")
content: Union[str, List[LettaUserMessageContentUnion]] = Field(
...,
description="The message content sent by the user (can be a string or an array of multi-modal content parts)",
@@ -109,7 +109,9 @@ class ReasoningMessage(LettaMessage):
signature (Optional[str]): The model-generated signature of the reasoning step
"""
message_type: Literal[MessageType.reasoning_message] = Field(MessageType.reasoning_message, description="The type of the message.")
message_type: Literal[MessageType.reasoning_message] = Field(
default=MessageType.reasoning_message, description="The type of the message."
)
source: Literal["reasoner_model", "non_reasoner_model"] = "non_reasoner_model"
reasoning: str
signature: Optional[str] = None
@@ -130,7 +132,7 @@ class HiddenReasoningMessage(LettaMessage):
"""
message_type: Literal[MessageType.hidden_reasoning_message] = Field(
MessageType.hidden_reasoning_message, description="The type of the message."
default=MessageType.hidden_reasoning_message, description="The type of the message."
)
state: Literal["redacted", "omitted"]
hidden_reasoning: Optional[str] = None
@@ -170,7 +172,9 @@ class ToolCallMessage(LettaMessage):
tool_call (Union[ToolCall, ToolCallDelta]): The tool call
"""
message_type: Literal[MessageType.tool_call_message] = Field(MessageType.tool_call_message, description="The type of the message.")
message_type: Literal[MessageType.tool_call_message] = Field(
default=MessageType.tool_call_message, description="The type of the message."
)
tool_call: Union[ToolCall, ToolCallDelta]
def model_dump(self, *args, **kwargs):
@@ -222,7 +226,9 @@ class ToolReturnMessage(LettaMessage):
stderr (Optional[List(str)]): Captured stderr from the tool invocation
"""
message_type: Literal[MessageType.tool_return_message] = Field(MessageType.tool_return_message, description="The type of the message.")
message_type: Literal[MessageType.tool_return_message] = Field(
default=MessageType.tool_return_message, description="The type of the message."
)
tool_return: str
status: Literal["success", "error"]
tool_call_id: str
@@ -241,7 +247,9 @@ class AssistantMessage(LettaMessage):
content (Union[str, List[LettaAssistantMessageContentUnion]]): The message content sent by the agent (can be a string or an array of content parts)
"""
message_type: Literal[MessageType.assistant_message] = Field(MessageType.assistant_message, description="The type of the message.")
message_type: Literal[MessageType.assistant_message] = Field(
default=MessageType.assistant_message, description="The type of the message."
)
content: Union[str, List[LettaAssistantMessageContentUnion]] = Field(
...,
description="The message content sent by the agent (can be a string or an array of content parts)",

30
letta/schemas/letta_message_content.py
View File

@@ -24,7 +24,7 @@ class MessageContent(BaseModel):
class TextContent(MessageContent):
type: Literal[MessageContentType.text] = Field(MessageContentType.text, description="The type of the message.")
type: Literal[MessageContentType.text] = Field(default=MessageContentType.text, description="The type of the message.")
text: str = Field(..., description="The text content of the message.")
@@ -44,27 +44,27 @@ class ImageSource(BaseModel):
class UrlImage(ImageSource):
type: Literal[ImageSourceType.url] = Field(ImageSourceType.url, description="The source type for the image.")
type: Literal[ImageSourceType.url] = Field(default=ImageSourceType.url, description="The source type for the image.")
url: str = Field(..., description="The URL of the image.")
class Base64Image(ImageSource):
type: Literal[ImageSourceType.base64] = Field(ImageSourceType.base64, description="The source type for the image.")
type: Literal[ImageSourceType.base64] = Field(default=ImageSourceType.base64, description="The source type for the image.")
media_type: str = Field(..., description="The media type for the image.")
data: str = Field(..., description="The base64 encoded image data.")
detail: Optional[str] = Field(
None,
default=None,
description="What level of detail to use when processing and understanding the image (low, high, or auto to let the model decide)",
)
class LettaImage(ImageSource):
type: Literal[ImageSourceType.letta] = Field(ImageSourceType.letta, description="The source type for the image.")
type: Literal[ImageSourceType.letta] = Field(default=ImageSourceType.letta, description="The source type for the image.")
file_id: str = Field(..., description="The unique identifier of the image file persisted in storage.")
media_type: Optional[str] = Field(None, description="The media type for the image.")
data: Optional[str] = Field(None, description="The base64 encoded image data.")
media_type: Optional[str] = Field(default=None, description="The media type for the image.")
data: Optional[str] = Field(default=None, description="The base64 encoded image data.")
detail: Optional[str] = Field(
None,
default=None,
description="What level of detail to use when processing and understanding the image (low, high, or auto to let the model decide)",
)
@@ -73,7 +73,7 @@ ImageSourceUnion = Annotated[Union[UrlImage, Base64Image, LettaImage], Field(dis
class ImageContent(MessageContent):
type: Literal[MessageContentType.image] = Field(MessageContentType.image, description="The type of the message.")
type: Literal[MessageContentType.image] = Field(default=MessageContentType.image, description="The type of the message.")
source: ImageSourceUnion = Field(..., description="The source of the image.")
@@ -164,7 +164,7 @@ def get_letta_assistant_message_content_union_str_json_schema():
class ToolCallContent(MessageContent):
type: Literal[MessageContentType.tool_call] = Field(
MessageContentType.tool_call, description="Indicates this content represents a tool call event."
default=MessageContentType.tool_call, description="Indicates this content represents a tool call event."
)
id: str = Field(..., description="A unique identifier for this specific tool call instance.")
name: str = Field(..., description="The name of the tool being called.")
@@ -175,7 +175,7 @@ class ToolCallContent(MessageContent):
class ToolReturnContent(MessageContent):
type: Literal[MessageContentType.tool_return] = Field(
MessageContentType.tool_return, description="Indicates this content represents a tool return event."
default=MessageContentType.tool_return, description="Indicates this content represents a tool return event."
)
tool_call_id: str = Field(..., description="References the ID of the ToolCallContent that initiated this tool call.")
content: str = Field(..., description="The content returned by the tool execution.")
@@ -184,23 +184,23 @@ class ToolReturnContent(MessageContent):
class ReasoningContent(MessageContent):
type: Literal[MessageContentType.reasoning] = Field(
MessageContentType.reasoning, description="Indicates this is a reasoning/intermediate step."
default=MessageContentType.reasoning, description="Indicates this is a reasoning/intermediate step."
)
is_native: bool = Field(..., description="Whether the reasoning content was generated by a reasoner model that processed this step.")
reasoning: str = Field(..., description="The intermediate reasoning or thought process content.")
signature: Optional[str] = Field(None, description="A unique identifier for this reasoning step.")
signature: Optional[str] = Field(default=None, description="A unique identifier for this reasoning step.")
class RedactedReasoningContent(MessageContent):
type: Literal[MessageContentType.redacted_reasoning] = Field(
MessageContentType.redacted_reasoning, description="Indicates this is a redacted thinking step."
default=MessageContentType.redacted_reasoning, description="Indicates this is a redacted thinking step."
)
data: str = Field(..., description="The redacted or filtered intermediate reasoning content.")
class OmittedReasoningContent(MessageContent):
type: Literal[MessageContentType.omitted_reasoning] = Field(
MessageContentType.omitted_reasoning, description="Indicates this is an omitted reasoning step."
default=MessageContentType.omitted_reasoning, description="Indicates this is an omitted reasoning step."
)
# NOTE: dropping because we don't track this kind of information for the other reasoning types
# tokens: int = Field(..., description="The reasoning token count for intermediate reasoning content.")

64
letta/schemas/message.py
View File

@@ -84,11 +84,11 @@ class MessageCreate(BaseModel):
description="The content of the message.",
json_schema_extra=get_letta_message_content_union_str_json_schema(),
)
name: Optional[str] = Field(None, description="The name of the participant.")
otid: Optional[str] = Field(None, description="The offline threading id associated with this message")
sender_id: Optional[str] = Field(None, description="The id of the sender of the message, can be an identity id or agent id")
batch_item_id: Optional[str] = Field(None, description="The id of the LLMBatchItem that this message is associated with")
group_id: Optional[str] = Field(None, description="The multi-agent group that the message was sent in")
name: Optional[str] = Field(default=None, description="The name of the participant.")
otid: Optional[str] = Field(default=None, description="The offline threading id associated with this message")
sender_id: Optional[str] = Field(default=None, description="The id of the sender of the message, can be an identity id or agent id")
batch_item_id: Optional[str] = Field(default=None, description="The id of the LLMBatchItem that this message is associated with")
group_id: Optional[str] = Field(default=None, description="The multi-agent group that the message was sent in")
def model_dump(self, to_orm: bool = False, **kwargs) -> Dict[str, Any]:
data = super().model_dump(**kwargs)
@@ -101,9 +101,9 @@ class MessageCreate(BaseModel):
class MessageUpdate(BaseModel):
"""Request to update a message"""
role: Optional[MessageRole] = Field(None, description="The role of the participant.")
role: Optional[MessageRole] = Field(default=None, description="The role of the participant.")
content: Optional[Union[str, List[LettaMessageContentUnion]]] = Field(
None,
default=None,
description="The content of the message.",
json_schema_extra=get_letta_message_content_union_str_json_schema(),
)
@@ -112,11 +112,11 @@ class MessageUpdate(BaseModel):
# agent_id: Optional[str] = Field(None, description="The unique identifier of the agent.")
# NOTE: we probably shouldn't allow updating the model field, otherwise this loses meaning
# model: Optional[str] = Field(None, description="The model used to make the function call.")
name: Optional[str] = Field(None, description="The name of the participant.")
name: Optional[str] = Field(default=None, description="The name of the participant.")
# NOTE: we probably shouldn't allow updating the created_at field, right?
# created_at: Optional[datetime] = Field(None, description="The time the message was created.")
tool_calls: Optional[List[OpenAIToolCall,]] = Field(None, description="The list of tool calls requested.")
tool_call_id: Optional[str] = Field(None, description="The id of the tool call.")
tool_calls: Optional[List[OpenAIToolCall,]] = Field(default=None, description="The list of tool calls requested.")
tool_call_id: Optional[str] = Field(default=None, description="The id of the tool call.")
def model_dump(self, to_orm: bool = False, **kwargs) -> Dict[str, Any]:
data = super().model_dump(**kwargs)
@@ -150,28 +150,28 @@ class Message(BaseMessage):
"""
id: str = BaseMessage.generate_id_field()
organization_id: Optional[str] = Field(None, description="The unique identifier of the organization.")
agent_id: Optional[str] = Field(None, description="The unique identifier of the agent.")
model: Optional[str] = Field(None, description="The model used to make the function call.")
organization_id: Optional[str] = Field(default=None, description="The unique identifier of the organization.")
agent_id: Optional[str] = Field(default=None, description="The unique identifier of the agent.")
model: Optional[str] = Field(default=None, description="The model used to make the function call.")
# Basic OpenAI-style fields
role: MessageRole = Field(..., description="The role of the participant.")
content: Optional[List[LettaMessageContentUnion]] = Field(None, description="The content of the message.")
content: Optional[List[LettaMessageContentUnion]] = Field(default=None, description="The content of the message.")
# NOTE: in OpenAI, this field is only used for roles 'user', 'assistant', and 'function' (now deprecated). 'tool' does not use it.
name: Optional[str] = Field(
None,
default=None,
description="For role user/assistant: the (optional) name of the participant. For role tool/function: the name of the function called.",
)
tool_calls: Optional[List[OpenAIToolCall]] = Field(
None, description="The list of tool calls requested. Only applicable for role assistant."
default=None, description="The list of tool calls requested. Only applicable for role assistant."
)
tool_call_id: Optional[str] = Field(None, description="The ID of the tool call. Only applicable for role tool.")
tool_call_id: Optional[str] = Field(default=None, description="The ID of the tool call. Only applicable for role tool.")
# Extras
step_id: Optional[str] = Field(None, description="The id of the step that this message was created in.")
otid: Optional[str] = Field(None, description="The offline threading id associated with this message")
tool_returns: Optional[List[ToolReturn]] = Field(None, description="Tool execution return information for prior tool calls")
group_id: Optional[str] = Field(None, description="The multi-agent group that the message was sent in")
sender_id: Optional[str] = Field(None, description="The id of the sender of the message, can be an identity id or agent id")
batch_item_id: Optional[str] = Field(None, description="The id of the LLMBatchItem that this message is associated with")
step_id: Optional[str] = Field(default=None, description="The id of the step that this message was created in.")
otid: Optional[str] = Field(default=None, description="The offline threading id associated with this message")
tool_returns: Optional[List[ToolReturn]] = Field(default=None, description="Tool execution return information for prior tool calls")
group_id: Optional[str] = Field(default=None, description="The multi-agent group that the message was sent in")
sender_id: Optional[str] = Field(default=None, description="The id of the sender of the message, can be an identity id or agent id")
batch_item_id: Optional[str] = Field(default=None, description="The id of the LLMBatchItem that this message is associated with")
# This overrides the optional base orm schema, created_at MUST exist on all messages objects
created_at: datetime = Field(default_factory=get_utc_time, description="The timestamp when the object was created.")
@@ -482,7 +482,9 @@ class Message(BaseMessage):
# TODO(caren) implicit support for only non-parts/list content types
if openai_message_dict["content"] is not None and type(openai_message_dict["content"]) is not str:
raise ValueError(f"Invalid content type: {type(openai_message_dict['content'])}")
content = [TextContent(text=openai_message_dict["content"])] if openai_message_dict["content"] else []
content: List[LettaMessageContentUnion] = (
[TextContent(text=openai_message_dict["content"])] if openai_message_dict["content"] else []
)
# TODO(caren) bad assumption here that "reasoning_content" always comes before "redacted_reasoning_content"
if "reasoning_content" in openai_message_dict and openai_message_dict["reasoning_content"]:
@@ -491,14 +493,16 @@ class Message(BaseMessage):
reasoning=openai_message_dict["reasoning_content"],
is_native=True,
signature=(
openai_message_dict["reasoning_content_signature"] if openai_message_dict["reasoning_content_signature"] else None
str(openai_message_dict["reasoning_content_signature"])
if "reasoning_content_signature" in openai_message_dict
else None
),
),
)
if "redacted_reasoning_content" in openai_message_dict and openai_message_dict["redacted_reasoning_content"]:
content.append(
RedactedReasoningContent(
data=openai_message_dict["redacted_reasoning_content"] if "redacted_reasoning_content" in openai_message_dict else None,
data=str(openai_message_dict["redacted_reasoning_content"]),
),
)
if "omitted_reasoning_content" in openai_message_dict and openai_message_dict["omitted_reasoning_content"]:
@@ -733,7 +737,7 @@ class Message(BaseMessage):
else:
warnings.warn(f"Using OpenAI with invalid 'name' field (name={self.name} role={self.role}).")
if parse_content_parts:
if parse_content_parts and self.content is not None:
for content in self.content:
if isinstance(content, ReasoningContent):
openai_message["reasoning_content"] = content.reasoning
@@ -819,7 +823,7 @@ class Message(BaseMessage):
}
content = []
# COT / reasoning / thinking
if len(self.content) > 1:
if self.content is not None and len(self.content) > 1:
for content_part in self.content:
if isinstance(content_part, ReasoningContent):
content.append(
@@ -1154,6 +1158,6 @@ class Message(BaseMessage):
class ToolReturn(BaseModel):
status: Literal["success", "error"] = Field(..., description="The status of the tool call")
stdout: Optional[List[str]] = Field(None, description="Captured stdout (e.g. prints, logs) from the tool invocation")
stderr: Optional[List[str]] = Field(None, description="Captured stderr from the tool invocation")
stdout: Optional[List[str]] = Field(default=None, description="Captured stdout (e.g. prints, logs) from the tool invocation")
stderr: Optional[List[str]] = Field(default=None, description="Captured stderr from the tool invocation")
# func_return: Optional[Any] = Field(None, description="The function return object")