wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: type fixes (and a few latent bug fixes) in 4 LLM adapters

Browse Source 
Same shape of fix we applied to anthropic_adapter.py earlier — these
adapters do dict-style mutation on values typed as
ChatCompletionMessageParam (a union of TypedDicts) or against Optional
fields. Apply boundary casts (`cast(dict[str, Any], ...)` for the
mutation block, cast back to the TypedDict at return sites). Most
changes are pure typing (rename + cast); a handful in gemini and
openai_realtime are small defensive bug fixes for code paths that
were latently broken by Optional fields slipping through:

perplexity_adapter.py — pure typing. Cast the deepcopied messages to
`list[dict[str, Any]]` for the role-merging / system-conversion /
trailing-assistant-removal transformations and cast back to
ChatCompletionMessageParam at the return.

bedrock_adapter.py — pure typing. Cast the message to
`dict[str, Any]` at the top of `_from_standard_message` for the
tool-result / tool-use / image-content transformations. Cast the
constructed dict at the return site of `get_llm_invocation_params`.

gemini_adapter.py — typing + several None guards on Content.parts and
related Optional fields. Each guard turns a latent
`TypeError`/`AttributeError` (when the type-system-allowed None
showed up at runtime) into a defensive skip — the type annotations
say these can be None and we now handle that.

open_ai_realtime_adapter.py:
- Typing: cast the deepcopied messages, cast back where needed.
- LLMSpecificMessage handling: previously the function would crash on
  the first `.get()` call if any LLMSpecificMessage was in the list.
  Filter them out and document the limitation — this adapter's
  pack-into-single-text-message strategy doesn't compose with opaque
  per-provider payloads.
- Real bug fix: `events.ConversationItem` is a Pydantic BaseModel,
  not a TypedDict. The bulk-packing path was constructing a raw dict
  where a ConversationItem was expected. Replaced with proper
  constructor calls (matches what the single-user-message path
  already does).
- Real bug fix: `_from_universal_context_message` was declared
  `-> events.ConversationItem` but on the unhandled-message
  fallthrough it logged and returned None implicitly. Raise
  ValueError so the violation is loud, not silent.

Removes 4 newly-clean files from the pyright ignore list:
adapters/services/{perplexity,bedrock,gemini,open_ai_realtime}_adapter.py.

Net: -95 pyright errors (full-config: 775 -> 680).
This commit is contained in:
Paul Kompfner
2026-04-28 17:15:35 -04:00
parent ef226c8a8e
commit 5e24027fd5
5 changed files with 112 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
pyrightconfig.json
View File

@@ -7,14 +7,10 @@
"ignore": [ "ignore": [
"tests", "tests",
"src/pipecat/adapters/services/aws_nova_sonic_adapter.py", "src/pipecat/adapters/services/aws_nova_sonic_adapter.py",
"src/pipecat/adapters/services/bedrock_adapter.py",
"src/pipecat/adapters/services/gemini_adapter.py",
"src/pipecat/adapters/services/grok_realtime_adapter.py", "src/pipecat/adapters/services/grok_realtime_adapter.py",
"src/pipecat/adapters/services/inworld_realtime_adapter.py", "src/pipecat/adapters/services/inworld_realtime_adapter.py",
"src/pipecat/adapters/services/open_ai_adapter.py", "src/pipecat/adapters/services/open_ai_adapter.py",
"src/pipecat/adapters/services/open_ai_realtime_adapter.py",
"src/pipecat/adapters/services/open_ai_responses_adapter.py", "src/pipecat/adapters/services/open_ai_responses_adapter.py",
"src/pipecat/adapters/services/perplexity_adapter.py",
"src/pipecat/audio/dtmf/utils.py", "src/pipecat/audio/dtmf/utils.py",
"src/pipecat/audio/filters/aic_filter.py", "src/pipecat/audio/filters/aic_filter.py",
"src/pipecat/audio/filters/krisp_viva_filter.py", "src/pipecat/audio/filters/krisp_viva_filter.py",

60
src/pipecat/adapters/services/bedrock_adapter.py
View File

@@ -10,7 +10,7 @@ import base64
import copy import copy
import json import json
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any, TypedDict from typing import Any, TypedDict, cast
from loguru import logger from loguru import logger
@@ -68,16 +68,19 @@ class AWSBedrockLLMAdapter(BaseLLMAdapter[AWSBedrockLLMInvocationParams]):
system_instruction, system_instruction,
discard_context_system=True, discard_context_system=True,
) )
return { return cast(
"system": [{"text": effective_system}] if effective_system else None, AWSBedrockLLMInvocationParams,
"messages": converted.messages, {
# NOTE: LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN) "system": [{"text": effective_system}] if effective_system else None,
"tools": self.from_standard_tools(context.tools) or [], "messages": converted.messages,
# To avoid refactoring in AWSBedrockLLMService, we just pass through tool_choice. # NOTE: LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN)
# Eventually (when we don't have to maintain the non-LLMContext code path) we should do "tools": self.from_standard_tools(context.tools) or [],
# the conversion to Bedrock's expected format here rather than in AWSBedrockLLMService. # To avoid refactoring in AWSBedrockLLMService, we just pass through tool_choice.
"tool_choice": context.tool_choice, # Eventually (when we don't have to maintain the non-LLMContext code path) we should do
} # the conversion to Bedrock's expected format here rather than in AWSBedrockLLMService.
"tool_choice": context.tool_choice,
},
)
def get_messages_for_logging(self, context) -> list[dict[str, Any]]: def get_messages_for_logging(self, context) -> list[dict[str, Any]]:
"""Get messages from a universal LLM context in a format ready for logging about AWS Bedrock. """Get messages from a universal LLM context in a format ready for logging about AWS Bedrock.
@@ -213,35 +216,36 @@ class AWSBedrockLLMAdapter(BaseLLMAdapter[AWSBedrockLLMInvocationParams]):
] ]
} }
""" """
message = copy.deepcopy(message) # ChatCompletionMessageParam (input) and the dict shape Bedrock expects
if message["role"] == "tool": # are different — work with the deepcopied message as a plain dict for
# the transformations below.
msg = cast(dict[str, Any], copy.deepcopy(message))
if msg["role"] == "tool":
# Try to parse the content as JSON if it looks like JSON # Try to parse the content as JSON if it looks like JSON
try: try:
if message["content"].strip().startswith("{") and message[ if msg["content"].strip().startswith("{") and msg["content"].strip().endswith("}"):
"content" content_json = json.loads(msg["content"])
].strip().endswith("}"):
content_json = json.loads(message["content"])
tool_result_content = [{"json": content_json}] tool_result_content = [{"json": content_json}]
else: else:
tool_result_content = [{"text": message["content"]}] tool_result_content = [{"text": msg["content"]}]
except (json.JSONDecodeError, ValueError, AttributeError): except (json.JSONDecodeError, ValueError, AttributeError):
tool_result_content = [{"text": message["content"]}] tool_result_content = [{"text": msg["content"]}]
return { return {
"role": "user", "role": "user",
"content": [ "content": [
{ {
"toolResult": { "toolResult": {
"toolUseId": message["tool_call_id"], "toolUseId": msg["tool_call_id"],
"content": tool_result_content, "content": tool_result_content,
}, },
}, },
], ],
} }
if message.get("tool_calls"): if msg.get("tool_calls"):
tc = message["tool_calls"] tc = msg["tool_calls"]
ret = {"role": "assistant", "content": []} ret: dict[str, Any] = {"role": "assistant", "content": []}
for tool_call in tc: for tool_call in tc:
function = tool_call["function"] function = tool_call["function"]
arguments = json.loads(function["arguments"]) arguments = json.loads(function["arguments"])
@@ -256,12 +260,12 @@ class AWSBedrockLLMAdapter(BaseLLMAdapter[AWSBedrockLLMInvocationParams]):
return ret return ret
# Handle text content # Handle text content
content = message.get("content") content = msg.get("content")
if isinstance(content, str): if isinstance(content, str):
if content == "": if content == "":
return {"role": message["role"], "content": [{"text": "(empty)"}]} return {"role": msg["role"], "content": [{"text": "(empty)"}]}
else: else:
return {"role": message["role"], "content": [{"text": content}]} return {"role": msg["role"], "content": [{"text": content}]}
elif isinstance(content, list): elif isinstance(content, list):
new_content = [] new_content = []
for item in content: for item in content:
@@ -300,9 +304,9 @@ class AWSBedrockLLMAdapter(BaseLLMAdapter[AWSBedrockLLMInvocationParams]):
# Move image before the first text # Move image before the first text
image_item = new_content.pop(img_idx) image_item = new_content.pop(img_idx)
new_content.insert(first_txt_idx, image_item) new_content.insert(first_txt_idx, image_item)
return {"role": message["role"], "content": new_content} return {"role": msg["role"], "content": new_content}
return message return msg
@staticmethod @staticmethod
def _to_bedrock_function_format(function: FunctionSchema) -> dict[str, Any]: def _to_bedrock_function_format(function: FunctionSchema) -> dict[str, Any]:

37
src/pipecat/adapters/services/gemini_adapter.py
View File

@@ -9,7 +9,7 @@
import base64 import base64
import json import json
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import Any, TypedDict from typing import Any, TypedDict, cast
from loguru import logger from loguru import logger
from openai import NotGiven from openai import NotGiven
@@ -154,9 +154,12 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
messages = self._from_universal_context_messages(self.get_messages(context)).messages messages = self._from_universal_context_messages(self.get_messages(context)).messages
# Sanitize messages for logging # Sanitize messages for logging
messages_for_logging = [] messages_for_logging: list[dict[str, Any]] = []
for message in messages: for message in messages:
obj = message.to_json_dict() # `to_json_dict()` returns `dict[str, object]`; treat as a plain
# dict for the value indexing/mutation below. The broad `except`
# below is the safety net if any item isn't shaped as expected.
obj: dict[str, Any] = cast(dict[str, Any], message.to_json_dict())
try: try:
if "parts" in obj: if "parts" in obj:
for part in obj["parts"]: for part in obj["parts"]:
@@ -274,7 +277,8 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
# Check if we only have function-related messages (no regular text) # Check if we only have function-related messages (no regular text)
effective_system = extracted_system or system_instruction effective_system = extracted_system or system_instruction
has_regular_messages = any( has_regular_messages = any(
len(msg.parts) == 1 msg.parts is not None
and len(msg.parts) == 1
and getattr(msg.parts[0], "text", None) and getattr(msg.parts[0], "text", None)
and not getattr(msg.parts[0], "function_call", None) and not getattr(msg.parts[0], "function_call", None)
and not getattr(msg.parts[0], "function_response", None) and not getattr(msg.parts[0], "function_response", None)
@@ -346,8 +350,11 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
parts=[Part(function_call=FunctionCall(name="search", args={"query": "test"}))] parts=[Part(function_call=FunctionCall(name="search", args={"query": "test"}))]
) )
""" """
role = message["role"] # ChatCompletionMessageParam (a union of TypedDicts) doesn't allow
content = message.get("content", []) # the dict-style key access used below; treat it as a plain dict.
msg = cast(dict[str, Any], message)
role = msg["role"]
content = msg.get("content", [])
# Convert non-initial system/developer messages to user role, # Convert non-initial system/developer messages to user role,
# as Gemini doesn't support these as input messages. # as Gemini doesn't support these as input messages.
@@ -359,8 +366,8 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
parts = [] parts = []
tool_call_id_to_name_mapping = {} tool_call_id_to_name_mapping = {}
if message.get("tool_calls"): if msg.get("tool_calls"):
for tc in message["tool_calls"]: for tc in msg["tool_calls"]:
id = tc["id"] id = tc["id"]
name = tc["function"]["name"] name = tc["function"]["name"]
tool_call_id_to_name_mapping[id] = name tool_call_id_to_name_mapping[id] = name
@@ -376,7 +383,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
elif role == "tool": elif role == "tool":
role = "user" role = "user"
try: try:
response = json.loads(message["content"]) response = json.loads(msg["content"])
if isinstance(response, dict): if isinstance(response, dict):
response_dict = response response_dict = response
else: else:
@@ -384,10 +391,10 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
except Exception as e: except Exception as e:
# Response might not be JSON-deserializable. # Response might not be JSON-deserializable.
# This occurs with a UserImageFrame, for example, where we get a plain "COMPLETED" string. # This occurs with a UserImageFrame, for example, where we get a plain "COMPLETED" string.
response_dict = {"value": message["content"]} response_dict = {"value": msg["content"]}
# Get function name from mapping using tool_call_id, or fallback # Get function name from mapping using tool_call_id, or fallback
tool_call_id = message.get("tool_call_id") tool_call_id = msg.get("tool_call_id")
function_name = "tool_call_result" # Default fallback function_name = "tool_call_result" # Default fallback
if tool_call_id and tool_call_id in params.tool_call_id_to_name_mapping: if tool_call_id and tool_call_id in params.tool_call_id_to_name_mapping:
function_name = params.tool_call_id_to_name_mapping[tool_call_id] function_name = params.tool_call_id_to_name_mapping[tool_call_id]
@@ -491,7 +498,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
def is_tool_call_message(msg: Content) -> bool: def is_tool_call_message(msg: Content) -> bool:
"""Check if message contains only function_call parts.""" """Check if message contains only function_call parts."""
return ( return bool(
msg.role == "model" msg.role == "model"
and msg.parts and msg.parts
and all(getattr(part, "function_call", None) for part in msg.parts) and all(getattr(part, "function_call", None) for part in msg.parts)
@@ -499,6 +506,8 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
def message_has_thought_signature(msg: Content) -> bool: def message_has_thought_signature(msg: Content) -> bool:
"""Check if any part in the message has a thought_signature.""" """Check if any part in the message has a thought_signature."""
if msg.parts is None:
return False
return any(getattr(part, "thought_signature", None) for part in msg.parts) return any(getattr(part, "thought_signature", None) for part in msg.parts)
merged_messages = [] merged_messages = []
@@ -564,6 +573,8 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
logger.debug(f"Thought signatures to apply: {len(thought_signature_dicts)}") logger.debug(f"Thought signatures to apply: {len(thought_signature_dicts)}")
for ts in thought_signature_dicts: for ts in thought_signature_dicts:
bookmark = ts.get("bookmark") bookmark = ts.get("bookmark")
if bookmark is None:
continue
if bookmark.get("function_call"): if bookmark.get("function_call"):
logger.trace(f" - To function call: {bookmark['function_call']}") logger.trace(f" - To function call: {bookmark['function_call']}")
elif bookmark.get("text"): elif bookmark.get("text"):
@@ -665,6 +676,8 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
if ( if (
hasattr(part, "inline_data") hasattr(part, "inline_data")
and part.inline_data and part.inline_data
and part.inline_data.data is not None
and bookmark_inline_data.data is not None
# Comparing length should be good enough for matching inline data, # Comparing length should be good enough for matching inline data,
# especially since we're already matching thought signatures in # especially since we're already matching thought signatures in
# strict message order. Comparing actual data is expensive. # strict message order. Comparing actual data is expensive.

61
src/pipecat/adapters/services/open_ai_realtime_adapter.py
View File

@@ -9,7 +9,7 @@
import copy import copy
import json import json
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any, TypedDict from typing import Any, TypedDict, cast
from loguru import logger from loguru import logger
@@ -81,7 +81,7 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
Returns: Returns:
List of messages in a format ready for logging about OpenAI Realtime. List of messages in a format ready for logging about OpenAI Realtime.
""" """
return self.get_messages(context, truncate_large_values=True) return cast(list[dict[str, Any]], self.get_messages(context, truncate_large_values=True))
@dataclass @dataclass
class ConvertedMessages: class ConvertedMessages:
@@ -101,12 +101,24 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
if not universal_context_messages: if not universal_context_messages:
return self.ConvertedMessages(messages=[]) return self.ConvertedMessages(messages=[])
messages = copy.deepcopy(universal_context_messages) # NOTE: This adapter does not yet handle ``LLMSpecificMessage`` — those
# are filtered out below. Other adapters (e.g. Anthropic) dispatch
# LLMSpecific items through a per-provider passthrough. For OpenAI
# Realtime, the strategy here packs a multi-message history into a
# single text message (see comment further down), which doesn't
# compose with opaque per-provider payloads. If/when this adapter
# adopts the per-message strategy, LLMSpecific items can flow
# through `_from_universal_context_message` like in other adapters.
messages: list[dict[str, Any]] = [
cast(dict[str, Any], m)
for m in copy.deepcopy(universal_context_messages)
if isinstance(m, dict)
]
system_instruction = None system_instruction = None
# If we have a "system" message as our first message, # If we have a "system" message as our first message,
# pull that out into session "instructions" # pull that out into session "instructions"
if messages[0].get("role") == "system": if messages and messages[0].get("role") == "system":
system = messages.pop(0) system = messages.pop(0)
content = system.get("content") content = system.get("content")
if isinstance(content, str): if isinstance(content, str):
@@ -124,7 +136,9 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
# If we have just a single "user" item, we can just send it normally # If we have just a single "user" item, we can just send it normally
if len(messages) == 1 and messages[0].get("role") == "user": if len(messages) == 1 and messages[0].get("role") == "user":
return self.ConvertedMessages( return self.ConvertedMessages(
messages=[self._from_universal_context_message(messages[0])], messages=[
self._from_universal_context_message(cast(LLMContextMessage, messages[0]))
],
system_instruction=system_instruction, system_instruction=system_instruction,
) )
@@ -142,18 +156,18 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
return self.ConvertedMessages( return self.ConvertedMessages(
messages=[ messages=[
{ events.ConversationItem(
"role": "user", role="user",
"type": "message", type="message",
"content": [ content=[
{ events.ItemContent(
"type": "input_text", type="input_text",
"text": "\n\n".join( text="\n\n".join(
[intro_text, json.dumps(messages, indent=2), trailing_text] [intro_text, json.dumps(messages, indent=2), trailing_text]
), ),
} )
], ],
} )
], ],
system_instruction=system_instruction, system_instruction=system_instruction,
) )
@@ -161,31 +175,34 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
def _from_universal_context_message( def _from_universal_context_message(
self, message: LLMContextMessage self, message: LLMContextMessage
) -> events.ConversationItem: ) -> events.ConversationItem:
if message.get("role") == "user": # NOTE: ``LLMSpecificMessage`` is not yet handled here — see the
content = message.get("content") # corresponding note in `_from_universal_context_messages`.
if isinstance(message.get("content"), list): msg = cast(dict[str, Any], message)
if msg.get("role") == "user":
content = msg.get("content")
if isinstance(content, list):
content = "" content = ""
for c in message.get("content"): for c in msg.get("content", []):
if c.get("type") == "text": if c.get("type") == "text":
content += " " + c.get("text") content += " " + c.get("text")
else: else:
logger.error( logger.error(
f"Unhandled content type in context message: {c.get('type')} - {message}" f"Unhandled content type in context message: {c.get('type')} - {msg}"
) )
return events.ConversationItem( return events.ConversationItem(
role="user", role="user",
type="message", type="message",
content=[events.ItemContent(type="input_text", text=content)], content=[events.ItemContent(type="input_text", text=content)],
) )
if message.get("role") == "assistant" and message.get("tool_calls"): if msg.get("role") == "assistant" and msg.get("tool_calls"):
tc = message.get("tool_calls")[0] tc = msg["tool_calls"][0]
return events.ConversationItem( return events.ConversationItem(
type="function_call", type="function_call",
call_id=tc["id"], call_id=tc["id"],
name=tc["function"]["name"], name=tc["function"]["name"],
arguments=tc["function"]["arguments"], arguments=tc["function"]["arguments"],
) )
logger.error(f"Unhandled message type in _from_universal_context_message: {message}") raise ValueError(f"Unhandled message type in _from_universal_context_message: {msg}")
@staticmethod @staticmethod
def _to_openai_realtime_function_format(function: FunctionSchema) -> dict[str, Any]: def _to_openai_realtime_function_format(function: FunctionSchema) -> dict[str, Any]:

27
src/pipecat/adapters/services/perplexity_adapter.py
View File

@@ -28,6 +28,7 @@ the messages are sent to Perplexity's API.
""" """
import copy import copy
from typing import Any, cast
from openai.types.chat import ChatCompletionMessageParam from openai.types.chat import ChatCompletionMessageParam
@@ -116,7 +117,11 @@ class PerplexityLLMAdapter(OpenAILLMAdapter):
if not messages: if not messages:
return messages return messages
messages = copy.deepcopy(messages) # ChatCompletionMessageParam is a union of TypedDicts; the
# transformations below mutate by key/index in ways those TypedDicts
# don't permit. Work against a plain-dict view for the duration of
# the transformation and cast back at the return site.
msgs: list[dict[str, Any]] = cast(list[dict[str, Any]], copy.deepcopy(messages))
# Note: "developer" → "user" conversion is handled by the parent adapter # Note: "developer" → "user" conversion is handled by the parent adapter
# via the convert_developer_to_user parameter. # via the convert_developer_to_user parameter.
@@ -125,10 +130,10 @@ class PerplexityLLMAdapter(OpenAILLMAdapter):
# Perplexity allows system messages at the start, but rejects them # Perplexity allows system messages at the start, but rejects them
# after any non-system message. # after any non-system message.
in_initial_system_block = True in_initial_system_block = True
for i in range(len(messages)): for i in range(len(msgs)):
if messages[i].get("role") == "system": if msgs[i].get("role") == "system":
if not in_initial_system_block: if not in_initial_system_block:
messages[i]["role"] = "user" msgs[i]["role"] = "user"
else: else:
in_initial_system_block = False in_initial_system_block = False
@@ -137,9 +142,9 @@ class PerplexityLLMAdapter(OpenAILLMAdapter):
# messages that violate Perplexity's strict alternation requirement. # messages that violate Perplexity's strict alternation requirement.
# Skip consecutive system messages at the start — Perplexity allows those. # Skip consecutive system messages at the start — Perplexity allows those.
i = 0 i = 0
while i < len(messages) - 1: while i < len(msgs) - 1:
current = messages[i] current = msgs[i]
next_msg = messages[i + 1] next_msg = msgs[i + 1]
if current["role"] == next_msg["role"] == "system": if current["role"] == next_msg["role"] == "system":
# Perplexity allows multiple initial system messages, don't merge # Perplexity allows multiple initial system messages, don't merge
i += 1 i += 1
@@ -154,7 +159,7 @@ class PerplexityLLMAdapter(OpenAILLMAdapter):
next_msg.get("content"), list next_msg.get("content"), list
): ):
current["content"].extend(next_msg["content"]) current["content"].extend(next_msg["content"])
messages.pop(i + 1) msgs.pop(i + 1)
else: else:
i += 1 i += 1
@@ -162,7 +167,7 @@ class PerplexityLLMAdapter(OpenAILLMAdapter):
# Perplexity requires the last message to be "user" or "tool". # Perplexity requires the last message to be "user" or "tool".
# OpenAI appears to silently ignore trailing assistant messages # OpenAI appears to silently ignore trailing assistant messages
# server-side, so removing them preserves equivalent behavior. # server-side, so removing them preserves equivalent behavior.
while messages and messages[-1].get("role") == "assistant": while msgs and msgs[-1].get("role") == "assistant":
messages.pop() msgs.pop()
return messages return cast(list[ChatCompletionMessageParam], msgs)