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

Add support for universal LLMContext to Anthropic LLM service

Browse Source
This commit is contained in:
Paul Kompfner
2025-09-03 10:58:41 -04:00
parent 2df231143a
commit 5c574eaad9
8 changed files with 394 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
examples/foundational/14b-function-calling-anthropic-video.py
View File

@@ -97,7 +97,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
llm = AnthropicLLMService(
api_key=os.getenv("ANTHROPIC_API_KEY"),
model="claude-3-7-sonnet-latest",
enable_prompt_caching_beta=True,
params=AnthropicLLMService.InputParams(enable_prompt_caching_beta=True),
)
llm.register_function("get_weather", get_weather)
llm.register_function("get_image", get_image)

100
examples/foundational/14z-function-calling-anthropic-universal-context.py
View File

@@ -5,6 +5,7 @@
#
import asyncio
import os
from dotenv import load_dotenv
@@ -20,25 +21,49 @@ from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.runner.utils import (
create_transport,
get_transport_client_id,
maybe_capture_participant_camera,
)
from pipecat.services.anthropic.llm import AnthropicLLMService
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams
from pipecat.transports.services.daily import DailyParams
load_dotenv(override=True)
# Global variable to store the client ID
client_id = ""
async def get_weather(params: FunctionCallParams):
location = params.arguments["location"]
await params.result_callback(f"The weather in {location} is currently 72 degrees and sunny.")
async def fetch_restaurant_recommendation(params: FunctionCallParams):
await params.result_callback({"name": "The Golden Dragon"})
async def get_image(params: FunctionCallParams):
question = params.arguments["question"]
logger.debug(f"Requesting image with user_id={client_id}, question={question}")
# Request the image frame
await params.llm.request_image_frame(
user_id=client_id,
function_name=params.function_name,
tool_call_id=params.tool_call_id,
text_content=question,
)
# Wait a short time for the frame to be processed
await asyncio.sleep(0.5)
# Return a result to complete the function call
await params.result_callback(
f"I've captured an image from your camera and I'm analyzing what you asked about: {question}"
)
# We store functions so objects (e.g. SileroVADAnalyzer) don't get
@@ -48,16 +73,13 @@ transport_params = {
"daily": lambda: DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
"twilio": lambda: FastAPIWebsocketParams(
audio_in_enabled=True,
audio_out_enabled=True,
video_in_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
"webrtc": lambda: TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
video_in_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
}
@@ -76,9 +98,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
llm = AnthropicLLMService(
api_key=os.getenv("ANTHROPIC_API_KEY"),
model="claude-3-7-sonnet-latest",
params=AnthropicLLMService.InputParams(enable_prompt_caching_beta=True),
)
llm.register_function("get_weather", get_weather)
llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
llm.register_function("get_image", get_image)
weather_function = FunctionSchema(
name="get_weather",
@@ -91,27 +114,44 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
},
required=["location"],
)
restaurant_function = FunctionSchema(
name="get_restaurant_recommendation",
description="Get a restaurant recommendation",
get_image_function = FunctionSchema(
name="get_image",
description="Get an image from the video stream.",
properties={
"location": {
"question": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"description": "The question that the user is asking about the image.",
}
},
required=["location"],
required=["question"],
)
tools = ToolsSchema(standard_tools=[weather_function, restaurant_function])
tools = ToolsSchema(standard_tools=[weather_function, get_image_function])
# todo: test with very short initial user message
system_prompt = """\
You are a helpful assistant who converses with a user and answers questions. Respond concisely to general questions.
# messages = [{"role": "system",
# "content": "You are a helpful assistant who can report the weather in any location in the universe. Respond concisely. Your response will be turned into speech so use only simple words and punctuation."},
# {"role": "user",
# "content": " Start the conversation by introducing yourself."}]
Your response will be turned into speech so use only simple words and punctuation.
messages = [{"role": "user", "content": "Say 'hello' to start the conversation."}]
You have access to two tools: get_weather and get_image.
You can respond to questions about the weather using the get_weather tool.
You can answer questions about the user's video stream using the get_image tool. Some examples of phrases that \
indicate you should use the get_image tool are:
- What do you see?
- What's in the video?
- Can you describe the video?
- Tell me about what you see.
- Tell me something interesting about what you see.
- What's happening in the video?
If you need to use a tool, simply use the tool. Do not tell the user the tool you are using. Be brief and concise.
"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": "Start the conversation by introducing yourself."},
]
context = LLMContext(messages, tools)
context_aggregator = LLMContextAggregatorPair(context)
@@ -119,8 +159,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
pipeline = Pipeline(
[
transport.input(), # Transport user input
stt,
context_aggregator.user(), # User spoken responses
stt, # STT
context_aggregator.user(), # User speech to text
llm, # LLM
tts, # TTS
transport.output(), # Transport bot output
@@ -139,7 +179,13 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info(f"Client connected")
logger.info(f"Client connected: {client}")
await maybe_capture_participant_camera(transport, client)
global client_id
client_id = get_transport_client_id(transport, client)
# Kick off the conversation.
await task.queue_frames([LLMRunFrame()])

3
src/pipecat/adapters/base_llm_adapter.py
View File

@@ -39,11 +39,12 @@ class BaseLLMAdapter(ABC, Generic[TLLMInvocationParams]):
"""
@abstractmethod
def get_llm_invocation_params(self, context: LLMContext) -> TLLMInvocationParams:
def get_llm_invocation_params(self, context: LLMContext, **kwargs) -> TLLMInvocationParams:
"""Get provider-specific LLM invocation parameters from a universal LLM context.
Args:
context: The LLM context containing messages, tools, etc.
**kwargs: Additional provider-specific arguments that subclasses can use.
Returns:
Provider-specific parameters for invoking the LLM.

261
src/pipecat/adapters/services/anthropic_adapter.py
View File

@@ -6,12 +6,25 @@
"""Anthropic LLM adapter for Pipecat."""
from typing import Any, Dict, List, TypedDict
import copy
import json
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, TypedDict
from anthropic import NOT_GIVEN, NotGiven
from anthropic.types.message_param import MessageParam
from anthropic.types.tool_union_param import ToolUnionParam
from loguru import logger
from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_context import (
LLMContext,
LLMContextMessage,
LLMSpecificMessage,
LLMStandardMessage,
)
class AnthropicLLMInvocationParams(TypedDict):
@@ -20,7 +33,9 @@ class AnthropicLLMInvocationParams(TypedDict):
This is a placeholder until support for universal LLMContext machinery is added for Anthropic.
"""
pass
system: str | NotGiven
messages: List[MessageParam]
tools: List[ToolUnionParam] | NotGiven
class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]):
@@ -30,20 +45,31 @@ class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]):
to the specific format required by Anthropic's Claude models for function calling.
"""
def get_llm_invocation_params(self, context: LLMContext) -> AnthropicLLMInvocationParams:
def get_llm_invocation_params(
self, context: LLMContext, enable_prompt_caching: bool
) -> AnthropicLLMInvocationParams:
"""Get Anthropic-specific LLM invocation parameters from a universal LLM context.
This is a placeholder until support for universal LLMContext machinery is added for Anthropic.
Args:
context: The LLM context containing messages, tools, etc.
enable_prompt_caching: Whether prompt caching should be enabled.
Returns:
Dictionary of parameters for invoking Anthropic's LLM API.
"""
raise NotImplementedError("Universal LLMContext is not yet supported for Anthropic.")
messages = self._from_universal_context_messages(self._get_messages(context))
return {
"system": messages.system,
"messages": self._with_cache_control_markers(messages.messages)
if enable_prompt_caching
else messages.messages,
# NOTE: LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN)
"tools": self.from_standard_tools(context.tools),
}
def get_messages_for_logging(self, context) -> List[Dict[str, Any]]:
def get_messages_for_logging(self, context: LLMContext) -> List[Dict[str, Any]]:
"""Get messages from a universal LLM context in a format ready for logging about Anthropic.
Removes or truncates sensitive data like image content for safe logging.
@@ -56,7 +82,228 @@ class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]):
Returns:
List of messages in a format ready for logging about Anthropic.
"""
raise NotImplementedError("Universal LLMContext is not yet supported for Anthropic.")
# Get messages in Anthropic's format
messages = self._from_universal_context_messages(self._get_messages(context)).messages
# Sanitize messages for logging
messages_for_logging = []
for message in messages:
msg = copy.deepcopy(message)
if "content" in msg:
if isinstance(msg["content"], list):
for item in msg["content"]:
if item["type"] == "image":
item["source"]["data"] = "..."
messages_for_logging.append(msg)
return messages_for_logging
def _get_messages(self, context: LLMContext) -> List[LLMContextMessage]:
return context.get_messages("anthropic")
@dataclass
class ConvertedMessages:
"""Container for Anthropic-formatted messages converted from universal context."""
messages: List[MessageParam]
system: str | NotGiven
def _from_universal_context_messages(
self, universal_context_messages: List[LLMContextMessage]
) -> ConvertedMessages:
system = NOT_GIVEN
messages = []
# first, map messages using self._from_universal_context_message(m)
try:
messages = [self._from_universal_context_message(m) for m in universal_context_messages]
except Exception as e:
logger.error(f"Error mapping messages: {e}")
# See if we should pull the system message out of our messages list.
if messages and messages[0]["role"] == "system":
if len(messages) == 1:
# If we have only have a system message in the list, all we can really do
# without introducing too much magic is change the role to "user".
messages[0]["role"] = "user"
else:
# If we have more than one message, we'll pull the system message out of the
# list.
system = messages[0]["content"]
messages.pop(0)
# Convert any subsequent "system"-role messages to "user"-role
# messages, as Anthropic doesn't support system input messages.
for message in messages:
if message["role"] == "system":
message["role"] = "user"
# Merge consecutive messages with the same role.
i = 0
while i < len(messages) - 1:
current_message = messages[i]
next_message = messages[i + 1]
if current_message["role"] == next_message["role"]:
# Convert content to list of dictionaries if it's a string
if isinstance(current_message["content"], str):
current_message["content"] = [
{"type": "text", "text": current_message["content"]}
]
if isinstance(next_message["content"], str):
next_message["content"] = [{"type": "text", "text": next_message["content"]}]
# Concatenate the content
current_message["content"].extend(next_message["content"])
# Remove the next message from the list
messages.pop(i + 1)
else:
i += 1
# Avoid empty content in messages
for message in messages:
if isinstance(message["content"], str) and message["content"] == "":
message["content"] = "(empty)"
elif isinstance(message["content"], list) and len(message["content"]) == 0:
message["content"] = [{"type": "text", "text": "(empty)"}]
return self.ConvertedMessages(messages=messages, system=system)
def _from_universal_context_message(self, message: LLMContextMessage) -> MessageParam:
if isinstance(message, LLMSpecificMessage):
return message.message
return self._from_standard_message(message)
def _from_standard_message(self, message: LLMStandardMessage) -> MessageParam:
"""Convert standard universal context message to Anthropic format.
Handles conversion of text content, tool calls, and tool results.
Empty text content is converted to "(empty)".
Args:
message: Message in standard universal context format.
Returns:
Message in Anthropic format.
Examples:
Input standard format::
{
"role": "assistant",
"tool_calls": [
{
"id": "123",
"function": {"name": "search", "arguments": '{"q": "test"}'}
}
]
}
Output Anthropic format::
{
"role": "assistant",
"content": [
{
"type": "tool_use",
"id": "123",
"name": "search",
"input": {"q": "test"}
}
]
}
"""
if message["role"] == "tool":
return {
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": message["tool_call_id"],
"content": message["content"],
},
],
}
if message.get("tool_calls"):
tc = message["tool_calls"]
ret = {"role": "assistant", "content": []}
for tool_call in tc:
function = tool_call["function"]
arguments = json.loads(function["arguments"])
new_tool_use = {
"type": "tool_use",
"id": tool_call["id"],
"name": function["name"],
"input": arguments,
}
ret["content"].append(new_tool_use)
return ret
content = message.get("content")
if isinstance(content, str):
# fix empty text
if content == "":
content = "(empty)"
elif isinstance(content, list):
for item in content:
# fix empty text
if item["type"] == "text" and item["text"] == "":
item["text"] = "(empty)"
# handle image_url -> image conversion
if item["type"] == "image_url":
item["type"] = "image"
item["source"] = {
"type": "base64",
"media_type": "image/jpeg",
"data": item["image_url"]["url"].split(",")[1],
}
del item["image_url"]
# In the case where there's a single image in the list (like what
# would result from a UserImageRawFrame), ensure that the image
# comes before text, as recommended by Anthropic docs
# (https://docs.anthropic.com/en/docs/build-with-claude/vision#example-one-image)
image_indices = [i for i, item in enumerate(content) if item["type"] == "image"]
text_indices = [i for i, item in enumerate(content) if item["type"] == "text"]
if len(image_indices) == 1 and text_indices:
img_idx = image_indices[0]
first_txt_idx = text_indices[0]
if img_idx > first_txt_idx:
# Move image before the first text
image_item = content.pop(img_idx)
content.insert(first_txt_idx, image_item)
return message
def _with_cache_control_markers(self, messages: List[MessageParam]) -> List[MessageParam]:
"""Add cache control markers to messages for prompt caching.
Args:
messages: List of messages in Anthropic format.
Returns:
List of messages with cache control markers added.
"""
def add_cache_control_marker(messages: List[MessageParam], negative_index: int):
if len(messages) > -(negative_index + 1) and messages[negative_index]["role"] == "user":
if isinstance(messages[negative_index]["content"], str):
messages[negative_index]["content"] = [
{"type": "text", "text": messages[negative_index]["content"]}
]
messages[negative_index]["content"][-1]["cache_control"] = {"type": "ephemeral"}
try:
messages_with_markers = copy.deepcopy(messages)
# Add cache control markers to the *last two* user messages. Why?
# - The marker at the last recent user message tells Anthropic to
# cache the prompt up to that point.
# - The marker at the second-to-last user message tells Anthropic
# to look up the cached prompt that goes up to that point (the
# point that *was* the last user message the previous turn).
# If we only added the marker to the last user message, we'd only
# ever be adding to the cache, never looking up from it.
add_cache_control_marker(messages_with_markers, -1)
add_cache_control_marker(messages_with_markers, -3)
return messages_with_markers
except Exception as e:
logger.error(f"Error adding cache control marker: {e}")
return messages_with_markers
@staticmethod
def _to_anthropic_function_format(function: FunctionSchema) -> Dict[str, Any]:

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

@@ -67,7 +67,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
return {
"system_instruction": messages.system_instruction,
"messages": messages.messages,
# NOTE; LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN)
# NOTE: LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN)
"tools": self.from_standard_tools(context.tools),
}
@@ -192,14 +192,14 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
def _from_standard_message(
self, message: LLMStandardMessage, already_have_system_instruction: bool
) -> Content | str:
"""Convert universal context message to Google Content object.
"""Convert standard universal context message to Google Content object.
Handles conversion of text, images, and function calls to Google's
format.
System instructions are returned as a plain string.
Args:
message: Message in universal context format.
message: Message in standard universal context format.
already_have_system_instruction: Whether we already have a system instruction
Returns:

80
src/pipecat/services/anthropic/llm.py
View File

@@ -24,7 +24,10 @@ from loguru import logger
from PIL import Image
from pydantic import BaseModel, Field
from pipecat.adapters.services.anthropic_adapter import AnthropicLLMAdapter
from pipecat.adapters.services.anthropic_adapter import (
AnthropicLLMAdapter,
AnthropicLLMInvocationParams,
)
from pipecat.frames.frames import (
ErrorFrame,
Frame,
@@ -215,18 +218,18 @@ class AnthropicLLMService(LLMService):
The LLM's response as a string, or None if no response is generated.
"""
messages = []
system = []
system = NOT_GIVEN
if isinstance(context, LLMContext):
# Future code will be something like this:
# adapter = self.get_llm_adapter()
# params: AnthropicLLMInvocationParams = adapter.get_llm_invocation_params(context)
# messages = params["messages"]
# system = params["system_instruction"]
raise NotImplementedError("Universal LLMContext is not yet supported for Anthropic.")
adapter: AnthropicLLMAdapter = self.get_llm_adapter()
params = adapter.get_llm_invocation_params(
context, enable_prompt_caching=self._settings["enable_prompt_caching_beta"]
)
messages = params["messages"]
system = params["system"]
else:
context = AnthropicLLMContext.upgrade_to_anthropic(context)
messages = context.messages
system = getattr(context, "system", None) or system_instruction
system = getattr(context, "system", None) or system_instruction or NOT_GIVEN
# LLM completion
response = await self._client.messages.create(
@@ -277,8 +280,31 @@ class AnthropicLLMService(LLMService):
assistant = AnthropicAssistantContextAggregator(context, params=assistant_params)
return AnthropicContextAggregatorPair(_user=user, _assistant=assistant)
def _get_llm_invocation_params(
self, context: OpenAILLMContext | LLMContext
) -> AnthropicLLMInvocationParams:
# Universal LLMContext
if isinstance(context, LLMContext):
adapter: AnthropicLLMAdapter = self.get_llm_adapter()
params = adapter.get_llm_invocation_params(
context, enable_prompt_caching=self._settings["enable_prompt_caching_beta"]
)
return params
# Anthropic-specific context
messages = (
context.get_messages_with_cache_control_markers()
if self._settings["enable_prompt_caching_beta"]
else context.messages
)
return AnthropicLLMInvocationParams(
system=context.system,
messages=messages,
tools=context.tools,
)
@traced_llm
async def _process_context(self, context: OpenAILLMContext):
async def _process_context(self, context: OpenAILLMContext | LLMContext):
# Usage tracking. We track the usage reported by Anthropic in prompt_tokens and
# completion_tokens. We also estimate the completion tokens from output text
# and use that estimate if we are interrupted, because we almost certainly won't
@@ -294,24 +320,22 @@ class AnthropicLLMService(LLMService):
await self.push_frame(LLMFullResponseStartFrame())
await self.start_processing_metrics()
params_from_context = self._get_llm_invocation_params(context)
if isinstance(context, LLMContext):
adapter = self.get_llm_adapter()
context_type_for_logging = "universal"
messages_for_logging = adapter.get_messages_for_logging(context)
else:
context_type_for_logging = "LLM-specific"
messages_for_logging = context.get_messages_for_logging()
logger.debug(
f"{self}: Generating chat [{context.system}] | {context.get_messages_for_logging()}"
f"{self}: Generating chat from {context_type_for_logging} context [{params_from_context['system']}] | {messages_for_logging}"
)
messages = context.messages
if self._settings["enable_prompt_caching_beta"]:
messages = context.get_messages_with_cache_control_markers()
api_call = self._client.messages.create
if self._settings["enable_prompt_caching_beta"]:
api_call = self._client.beta.prompt_caching.messages.create
await self.start_ttfb_metrics()
params = {
"tools": context.tools or [],
"system": context.system,
"messages": messages,
"model": self.model_name,
"max_tokens": self._settings["max_tokens"],
"stream": True,
@@ -320,9 +344,12 @@ class AnthropicLLMService(LLMService):
"top_p": self._settings["top_p"],
}
# Messages, system, tools
params.update(params_from_context)
params.update(self._settings["extra"])
response = await self._create_message_stream(api_call, params)
response = await self._create_message_stream(self._client.messages.create, params)
await self.stop_ttfb_metrics()
@@ -405,7 +432,10 @@ class AnthropicLLMService(LLMService):
prompt_tokens + cache_creation_input_tokens + cache_read_input_tokens
)
if total_input_tokens >= 1024:
context.turns_above_cache_threshold += 1
if hasattr(
context, "turns_above_cache_threshold"
): # LLMContext doesn't have this attribute
context.turns_above_cache_threshold += 1
await self.run_function_calls(function_calls)
@@ -451,7 +481,7 @@ class AnthropicLLMService(LLMService):
if isinstance(frame, OpenAILLMContextFrame):
context: "AnthropicLLMContext" = AnthropicLLMContext.upgrade_to_anthropic(frame.context)
elif isinstance(frame, LLMContextFrame):
raise NotImplementedError("Universal LLMContext is not yet supported for Anthropic.")
context = frame.context
elif isinstance(frame, LLMMessagesFrame):
context = AnthropicLLMContext.from_messages(frame.messages)
elif isinstance(frame, VisionImageRawFrame):

12
src/pipecat/services/google/llm.py
View File

@@ -858,8 +858,7 @@ class GoogleLLMService(LLMService):
self, context: OpenAILLMContext
) -> AsyncIterator[GenerateContentResponse]:
logger.debug(
# f"{self}: Generating chat [{self._system_instruction}] | {context.get_messages_for_logging()}"
f"{self}: Generating chat from OpenAI context {context.get_messages_for_logging()}"
f"{self}: Generating chat from LLM-specific context [{context.system_message}] | {context.get_messages_for_logging()}"
)
params = GeminiLLMInvocationParams(
@@ -874,13 +873,12 @@ class GoogleLLMService(LLMService):
self, context: LLMContext
) -> AsyncIterator[GenerateContentResponse]:
adapter = self.get_llm_adapter()
logger.debug(
# f"{self}: Generating chat [{self._system_instruction}] | {context.get_messages_for_logging()}"
f"{self}: Generating chat from universal context {adapter.get_messages_for_logging(context)}"
)
params: GeminiLLMInvocationParams = adapter.get_llm_invocation_params(context)
logger.debug(
f"{self}: Generating chat from universal context [{params['system_instruction']}] | {adapter.get_messages_for_logging(context)}"
)
return await self._stream_content(params)
@traced_llm

2
src/pipecat/services/openai/base_llm.py
View File

@@ -279,7 +279,7 @@ class BaseOpenAILLMService(LLMService):
self, context: OpenAILLMContext
) -> AsyncStream[ChatCompletionChunk]:
logger.debug(
f"{self}: Generating chat from OpenAI context {context.get_messages_for_logging()}"
f"{self}: Generating chat from LLM-specific context {context.get_messages_for_logging()}"
)
messages: List[ChatCompletionMessageParam] = context.get_messages()