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

Update run_inference to use the provided LLM configuration params

Browse Source
This commit is contained in:
Mark Backman
2025-12-09 13:33:05 -05:00
committed by Paul Kompfner
parent afa7573834
commit 21a55f6aae
6 changed files with 366 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog/3214.changed.md Normal file
View File

@@ -0,0 +1 @@
- Updated the `run_inference` methods in the LLM service classes (`AnthropicLLMService`, `AWSBedrockLLMService`, `GoogleLLMService`, and `OpenAILLMService` and its base classes) to use the provided LLM configuration parameters.

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

@@ -267,26 +267,41 @@ class AnthropicLLMService(LLMService):
"""
messages = []
system = NOT_GIVEN
tools = []
if isinstance(context, LLMContext):
adapter: AnthropicLLMAdapter = self.get_llm_adapter()
params = adapter.get_llm_invocation_params(
invocation_params = adapter.get_llm_invocation_params(
context, enable_prompt_caching=self._settings["enable_prompt_caching"]
)
messages = params["messages"]
system = params["system"]
messages = invocation_params["messages"]
system = invocation_params["system"]
tools = invocation_params["tools"]
else:
context = AnthropicLLMContext.upgrade_to_anthropic(context)
messages = context.messages
system = getattr(context, "system", NOT_GIVEN)
tools = context.tools or []
# Build params using the same method as streaming completions
params = {
"model": self.model_name,
"max_tokens": self._settings["max_tokens"],
"stream": False,
"temperature": self._settings["temperature"],
"top_k": self._settings["top_k"],
"top_p": self._settings["top_p"],
"messages": messages,
"system": system,
"tools": tools,
"betas": ["interleaved-thinking-2025-05-14"],
}
if self._settings["thinking"]:
params["thinking"] = self._settings["thinking"].model_dump(exclude_unset=True)
params.update(self._settings["extra"])
# LLM completion
response = await self._client.messages.create(
model=self.model_name,
messages=messages,
system=system,
max_tokens=8192,
stream=False,
)
response = await self._client.beta.messages.create(**params)
return response.content[0].text

8
src/pipecat/services/aws/llm.py
View File

@@ -840,15 +840,13 @@ class AWSBedrockLLMService(LLMService):
messages = context.messages
system = getattr(context, "system", None) # [{"text": "system message"}]
# Determine if we're using Claude or Nova based on model ID
model_id = self.model_name
# Prepare request parameters
# Prepare request parameters using the same method as streaming
inference_config = self._build_inference_config()
request_params = {
"modelId": model_id,
"modelId": self.model_name,
"messages": messages,
"additionalModelRequestFields": self._settings["additional_model_request_fields"],
}
if inference_config:

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

@@ -798,17 +798,25 @@ class GoogleLLMService(LLMService):
"""
messages = []
system = []
tools = []
if isinstance(context, LLMContext):
adapter = self.get_llm_adapter()
params: GeminiLLMInvocationParams = adapter.get_llm_invocation_params(context)
messages = params["messages"]
system = params["system_instruction"]
tools = params["tools"]
else:
context = GoogleLLMContext.upgrade_to_google(context)
messages = context.messages
system = getattr(context, "system_message", None)
tools = context.tools or []
generation_config = GenerateContentConfig(system_instruction=system)
# Build generation config using the same method as streaming
generation_params = self._build_generation_params(
system_instruction=system, tools=tools if tools else None
)
generation_config = GenerateContentConfig(**generation_params)
# Use the new google-genai client's async method
response = await self._client.aio.models.generate_content(
@@ -825,6 +833,48 @@ class GoogleLLMService(LLMService):
return None
def _build_generation_params(
self,
system_instruction: Optional[str] = None,
tools: Optional[List] = None,
tool_config: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""Build generation parameters for Google AI API.
Args:
system_instruction: Optional system instruction to use.
tools: Optional list of tools to include.
tool_config: Optional tool configuration.
Returns:
Dictionary of generation parameters with None values filtered out.
"""
# Filter out None values and create GenerationContentConfig
generation_params = {
k: v
for k, v in {
"system_instruction": system_instruction,
"temperature": self._settings["temperature"],
"top_p": self._settings["top_p"],
"top_k": self._settings["top_k"],
"max_output_tokens": self._settings["max_tokens"],
"tools": tools,
"tool_config": tool_config,
}.items()
if v is not None
}
# Add thinking parameters if configured
if self._settings["thinking"]:
generation_params["thinking_config"] = self._settings["thinking"].model_dump(
exclude_unset=True
)
if self._settings["extra"]:
generation_params.update(self._settings["extra"])
return generation_params
def _maybe_unset_thinking_budget(self, generation_params: Dict[str, Any]):
try:
# There's no way to introspect on model capabilities, so
@@ -862,36 +912,15 @@ class GoogleLLMService(LLMService):
if self._tool_config:
tool_config = self._tool_config
# Filter out None values and create GenerationContentConfig
generation_params = {
k: v
for k, v in {
"system_instruction": self._system_instruction,
"temperature": self._settings["temperature"],
"top_p": self._settings["top_p"],
"top_k": self._settings["top_k"],
"max_output_tokens": self._settings["max_tokens"],
"tools": tools,
"tool_config": tool_config,
}.items()
if v is not None
}
# Add thinking parameters if configured
if self._settings["thinking"]:
generation_params["thinking_config"] = self._settings["thinking"].model_dump(
exclude_unset=True
)
if self._settings["extra"]:
generation_params.update(self._settings["extra"])
# Build generation parameters
generation_params = self._build_generation_params(
system_instruction=self._system_instruction, tools=tools, tool_config=tool_config
)
# possibly modify generation_params (in place) to set thinking to off by default
self._maybe_unset_thinking_budget(generation_params)
generation_config = (
GenerateContentConfig(**generation_params) if generation_params else None
)
generation_config = GenerateContentConfig(**generation_params)
await self.start_ttfb_metrics()
return await self._client.aio.models.generate_content_stream(
@@ -1166,6 +1195,14 @@ class GoogleLLMService(LLMService):
# Do nothing - we're shutting down anyway
pass
async def _update_settings(self, settings):
"""Override to handle ThinkingConfig validation."""
# Convert thinking dict to ThinkingConfig if needed
if "thinking" in settings and isinstance(settings["thinking"], dict):
settings = dict(settings) # Make a copy to avoid modifying the original
settings["thinking"] = self.ThinkingConfig(**settings["thinking"])
await super()._update_settings(settings)
def create_context_aggregator(
self,
context: OpenAILLMContext,

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

@@ -276,17 +276,23 @@ class BaseOpenAILLMService(LLMService):
"""
if isinstance(context, LLMContext):
adapter = self.get_llm_adapter()
params: OpenAILLMInvocationParams = adapter.get_llm_invocation_params(context)
messages = params["messages"]
invocation_params: OpenAILLMInvocationParams = adapter.get_llm_invocation_params(
context
)
else:
messages = context.messages
invocation_params = OpenAILLMInvocationParams(
messages=context.messages, tools=context.tools, tool_choice=context.tool_choice
)
# Build params using the same method as streaming completions
params = self.build_chat_completion_params(invocation_params)
# Override for non-streaming
params["stream"] = False
params.pop("stream_options", None)
# LLM completion
response = await self._client.chat.completions.create(
model=self.model_name,
messages=messages,
stream=False,
)
response = await self._client.chat.completions.create(**params)
return response.choices[0].message.content

274
tests/test_run_inference.py
View File

@@ -19,9 +19,14 @@ from pipecat.services.openai.llm import OpenAILLMService
@pytest.mark.asyncio
async def test_openai_run_inference_with_llm_context():
"""Test run_inference with LLMContext returns expected response."""
# Create service with mocked client
# Create service with mocked client and specific parameters
with patch.object(OpenAILLMService, "create_client"):
service = OpenAILLMService(model="gpt-4")
from pipecat.services.openai.base_llm import BaseOpenAILLMService
params = BaseOpenAILLMService.InputParams(
temperature=0.7, max_tokens=100, frequency_penalty=0.5, seed=42
)
service = OpenAILLMService(model="gpt-4", params=params)
service._client = AsyncMock()
# Setup mocks
@@ -51,8 +56,73 @@ async def test_openai_run_inference_with_llm_context():
mock_adapter.get_llm_invocation_params.assert_called_once_with(mock_context)
service._client.chat.completions.create.assert_called_once_with(
model="gpt-4",
messages=test_messages,
stream=False,
frequency_penalty=0.5,
presence_penalty=OPENAI_NOT_GIVEN,
seed=42,
temperature=0.7,
top_p=OPENAI_NOT_GIVEN,
max_tokens=100,
max_completion_tokens=OPENAI_NOT_GIVEN,
service_tier=OPENAI_NOT_GIVEN,
messages=test_messages,
tools=OPENAI_NOT_GIVEN,
tool_choice=OPENAI_NOT_GIVEN,
)
@pytest.mark.asyncio
async def test_openai_run_inference_with_openai_llm_context():
"""Test run_inference with OpenAILLMContext returns expected response."""
# Create service with mocked client and specific parameters
with patch.object(OpenAILLMService, "create_client"):
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.openai.base_llm import BaseOpenAILLMService
params = BaseOpenAILLMService.InputParams(
temperature=0.8, max_completion_tokens=150, presence_penalty=0.3, top_p=0.9
)
service = OpenAILLMService(model="gpt-4", params=params)
service._client = AsyncMock()
# Create OpenAILLMContext
context = OpenAILLMContext(
messages=[
{"role": "system", "content": "You are a helpful assistant"},
{"role": "user", "content": "Hello, world!"},
],
tools=OPENAI_NOT_GIVEN,
tool_choice=OPENAI_NOT_GIVEN,
)
# Mock response
mock_response = MagicMock()
mock_response.choices = [MagicMock()]
mock_response.choices[0].message.content = "Hello! How can I help you today?"
service._client.chat.completions.create.return_value = mock_response
# Execute
result = await service.run_inference(context)
# Verify
assert result == "Hello! How can I help you today?"
service._client.chat.completions.create.assert_called_once_with(
model="gpt-4",
stream=False,
frequency_penalty=OPENAI_NOT_GIVEN,
presence_penalty=0.3,
seed=OPENAI_NOT_GIVEN,
temperature=0.8,
top_p=0.9,
max_tokens=OPENAI_NOT_GIVEN,
max_completion_tokens=150,
service_tier=OPENAI_NOT_GIVEN,
messages=[
{"role": "system", "content": "You are a helpful assistant"},
{"role": "user", "content": "Hello, world!"},
],
tools=OPENAI_NOT_GIVEN,
tool_choice=OPENAI_NOT_GIVEN,
)
@@ -78,8 +148,13 @@ async def test_openai_run_inference_client_exception():
@pytest.mark.asyncio
async def test_anthropic_run_inference_with_llm_context():
"""Test run_inference with LLMContext returns expected response for Anthropic."""
# Create service with mocked client
service = AnthropicLLMService(api_key="test-key", model="claude-3-sonnet-20240229")
# Create service with mocked client and specific parameters
from pipecat.services.anthropic.llm import AnthropicLLMService
params = AnthropicLLMService.InputParams(max_tokens=2048, temperature=0.6, top_k=50, top_p=0.95)
service = AnthropicLLMService(
api_key="test-key", model="claude-3-sonnet-20240229", params=params
)
service._client = AsyncMock()
# Setup mocks
@@ -96,7 +171,7 @@ async def test_anthropic_run_inference_with_llm_context():
mock_response = MagicMock()
mock_response.content = [MagicMock()]
mock_response.content[0].text = "Hello! How can I help you today?"
service._client.messages.create.return_value = mock_response
service._client.beta.messages.create.return_value = mock_response
# Execute
result = await service.run_inference(mock_context)
@@ -107,12 +182,65 @@ async def test_anthropic_run_inference_with_llm_context():
mock_adapter.get_llm_invocation_params.assert_called_once_with(
mock_context, enable_prompt_caching=False
)
service._client.messages.create.assert_called_once_with(
service._client.beta.messages.create.assert_called_once_with(
model="claude-3-sonnet-20240229",
max_tokens=2048,
stream=False,
temperature=0.6,
top_k=50,
top_p=0.95,
messages=test_messages,
system=test_system,
max_tokens=8192,
tools=[],
betas=["interleaved-thinking-2025-05-14"],
)
@pytest.mark.asyncio
async def test_anthropic_run_inference_with_openai_llm_context():
"""Test run_inference with OpenAILLMContext returns expected response for Anthropic."""
# Create service with mocked client and specific parameters
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.anthropic.llm import AnthropicLLMService
params = AnthropicLLMService.InputParams(max_tokens=1024, temperature=0.7, top_k=40, top_p=0.9)
service = AnthropicLLMService(
api_key="test-key", model="claude-3-sonnet-20240229", params=params
)
service._client = AsyncMock()
# Create OpenAILLMContext
context = OpenAILLMContext(
messages=[
{"role": "system", "content": "You are a helpful assistant"},
{"role": "user", "content": "Hello, world!"},
],
tools=NOT_GIVEN,
tool_choice=NOT_GIVEN,
)
# Mock response
mock_response = MagicMock()
mock_response.content = [MagicMock()]
mock_response.content[0].text = "Hello! How can I help you today?"
service._client.beta.messages.create.return_value = mock_response
# Execute
result = await service.run_inference(context)
# Verify
assert result == "Hello! How can I help you today?"
service._client.beta.messages.create.assert_called_once_with(
model="claude-3-sonnet-20240229",
max_tokens=1024,
stream=False,
temperature=0.7,
top_k=40,
top_p=0.9,
messages=[{"role": "user", "content": "Hello, world!"}],
system="You are a helpful assistant",
tools=[],
betas=["interleaved-thinking-2025-05-14"],
)
@@ -128,7 +256,7 @@ async def test_anthropic_run_inference_client_exception():
messages=[], system="Test system", tools=[]
)
service.get_llm_adapter = MagicMock(return_value=mock_adapter)
service._client.messages.create.side_effect = Exception("Anthropic API Error")
service._client.beta.messages.create.side_effect = Exception("Anthropic API Error")
with pytest.raises(Exception, match="Anthropic API Error"):
await service.run_inference(mock_context)
@@ -193,11 +321,69 @@ async def test_google_run_inference_client_exception():
await service.run_inference(mock_context)
@pytest.mark.asyncio
async def test_google_run_inference_with_openai_llm_context():
"""Test run_inference with OpenAILLMContext returns expected response for Google."""
# Create service with mocked client and specific parameters
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
params = GoogleLLMService.InputParams(max_tokens=256, temperature=0.4, top_k=30, top_p=0.75)
service = GoogleLLMService(api_key="test-key", model="gemini-2.0-flash", params=params)
service._client = AsyncMock()
# Create OpenAILLMContext
context = OpenAILLMContext(
messages=[
{"role": "system", "content": "You are a helpful assistant"},
{"role": "user", "content": "Hello, world!"},
],
tools=NOT_GIVEN,
tool_choice=NOT_GIVEN,
)
# Mock response
mock_response = MagicMock()
mock_response.candidates = [MagicMock()]
mock_response.candidates[0].content = MagicMock()
mock_response.candidates[0].content.parts = [MagicMock()]
mock_response.candidates[0].content.parts[0].text = "Hello! How can I help you today?"
service._client.aio = AsyncMock()
service._client.aio.models = AsyncMock()
service._client.aio.models.generate_content = AsyncMock(return_value=mock_response)
# Execute
result = await service.run_inference(context)
# Verify
assert result == "Hello! How can I help you today?"
# Verify the call includes configured parameters
call_kwargs = service._client.aio.models.generate_content.call_args.kwargs
assert call_kwargs["model"] == "gemini-2.0-flash"
# Contents is a Google Content object, so check its structure
contents = call_kwargs["contents"]
assert len(contents) == 1
assert contents[0].role == "user"
assert len(contents[0].parts) == 1
assert contents[0].parts[0].text == "Hello, world!"
assert "config" in call_kwargs
config = call_kwargs["config"]
# Config is a GenerateContentConfig object, so access attributes
assert config.system_instruction == "You are a helpful assistant"
assert config.temperature == 0.4
assert config.top_k == 30
assert config.top_p == 0.75
assert config.max_output_tokens == 256
@pytest.mark.asyncio
async def test_aws_bedrock_run_inference_with_llm_context():
"""Test run_inference with LLMContext returns expected response for AWS Bedrock."""
# Create service and patch the session client method
service = AWSBedrockLLMService(model="anthropic.claude-3-sonnet-20240229-v1:0")
# Create service with specific parameters
from pipecat.services.aws.llm import AWSBedrockLLMService
params = AWSBedrockLLMService.InputParams(max_tokens=1024, temperature=0.5, top_p=0.85)
service = AWSBedrockLLMService(model="anthropic.claude-3-sonnet-20240229-v1:0", params=params)
# Setup mocks
mock_context = MagicMock(spec=LLMContext)
@@ -217,9 +403,6 @@ async def test_aws_bedrock_run_inference_with_llm_context():
mock_client.converse.return_value = mock_response
# Patch the _aws_session.client method to be an async context manager
async def mock_client_cm(*args, **kwargs):
return mock_client
mock_context_manager = AsyncMock()
mock_context_manager.__aenter__ = AsyncMock(return_value=mock_client)
mock_context_manager.__aexit__ = AsyncMock(return_value=None)
@@ -232,7 +415,68 @@ async def test_aws_bedrock_run_inference_with_llm_context():
assert result == "Hello! How can I help you today?"
service.get_llm_adapter.assert_called_once()
mock_adapter.get_llm_invocation_params.assert_called_once_with(mock_context)
mock_client.converse.assert_called_once()
# Verify the call includes configured parameters
call_kwargs = mock_client.converse.call_args.kwargs
assert call_kwargs["modelId"] == "anthropic.claude-3-sonnet-20240229-v1:0"
assert call_kwargs["messages"] == test_messages
assert call_kwargs["system"] == test_system
assert call_kwargs["additionalModelRequestFields"] == {}
assert "inferenceConfig" in call_kwargs
assert call_kwargs["inferenceConfig"]["maxTokens"] == 1024
assert call_kwargs["inferenceConfig"]["temperature"] == 0.5
assert call_kwargs["inferenceConfig"]["topP"] == 0.85
@pytest.mark.asyncio
async def test_aws_bedrock_run_inference_with_openai_llm_context():
"""Test run_inference with OpenAILLMContext returns expected response for AWS Bedrock."""
# Create service with specific parameters
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.aws.llm import AWSBedrockLLMService
params = AWSBedrockLLMService.InputParams(max_tokens=512, temperature=0.8, top_p=0.95)
service = AWSBedrockLLMService(model="anthropic.claude-3-sonnet-20240229-v1:0", params=params)
# Create OpenAILLMContext
context = OpenAILLMContext(
messages=[
{"role": "system", "content": "You are a helpful assistant"},
{"role": "user", "content": "Hello, world!"},
],
tools=NOT_GIVEN,
tool_choice=NOT_GIVEN,
)
# Mock the client and response
mock_client = AsyncMock()
mock_response = {
"output": {"message": {"content": [{"text": "Hello! How can I help you today?"}]}}
}
mock_client.converse.return_value = mock_response
# Patch the _aws_session.client method to be an async context manager
mock_context_manager = AsyncMock()
mock_context_manager.__aenter__ = AsyncMock(return_value=mock_client)
mock_context_manager.__aexit__ = AsyncMock(return_value=None)
with patch.object(service._aws_session, "client", return_value=mock_context_manager):
# Execute
result = await service.run_inference(context)
# Verify
assert result == "Hello! How can I help you today?"
# Verify the call includes configured parameters
call_kwargs = mock_client.converse.call_args.kwargs
assert call_kwargs["modelId"] == "anthropic.claude-3-sonnet-20240229-v1:0"
assert call_kwargs["messages"] == [{"role": "user", "content": [{"text": "Hello, world!"}]}]
assert call_kwargs["system"] == [{"text": "You are a helpful assistant"}]
assert call_kwargs["additionalModelRequestFields"] == {}
assert "inferenceConfig" in call_kwargs
assert call_kwargs["inferenceConfig"]["maxTokens"] == 512
assert call_kwargs["inferenceConfig"]["temperature"] == 0.8
assert call_kwargs["inferenceConfig"]["topP"] == 0.95
@pytest.mark.asyncio