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

fix formatting

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-05-06 11:37:23 -07:00
parent 664111a3c9
commit a4b9db9e07
5 changed files with 85 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
examples/foundational/07m-interruptible-aws.py
View File

@@ -13,13 +13,13 @@ from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.services.aws.llm import BedrockLLMContext, BedrockLLMService
from pipecat.services.aws.stt import TranscribeSTTService
from pipecat.services.aws.tts import PollyTTSService
from pipecat.transcriptions.language import Language from pipecat.transcriptions.language import Language
from pipecat.transports.base_transport import TransportParams from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
from pipecat.services.aws.llm import BedrockLLMService, BedrockLLMContext
from pipecat.services.aws.stt import TranscribeSTTService
from pipecat.services.aws.tts import PollyTTSService
load_dotenv(override=True) load_dotenv(override=True)
@@ -42,28 +42,26 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
region="us-west-2", # only specific regions support generative TTS region="us-west-2", # only specific regions support generative TTS
voice_id="Joanna", voice_id="Joanna",
params=PollyTTSService.InputParams( params=PollyTTSService.InputParams(
engine="generative", engine="generative", language=Language.EN_US, rate="1.1"
language=Language.EN_US,
rate="1.1"
), ),
) )
llm = BedrockLLMService( llm = BedrockLLMService(
aws_region="us-west-2", aws_region="us-west-2",
model="us.anthropic.claude-3-5-haiku-20241022-v1:0", model="us.anthropic.claude-3-5-haiku-20241022-v1:0",
params=BedrockLLMService.InputParams( params=BedrockLLMService.InputParams(temperature=0.8, latency="optimized"),
temperature=0.8,
latency="optimized"
)
) )
messages = [ messages = [
{ {
"role": "system", "role": "system",
"content": [{"text": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way."}], "content": [
}, {
] "text": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way."
) }
],
},
]
context = BedrockLLMContext(messages) context = BedrockLLMContext(messages)
context_aggregator = llm.create_context_aggregator(context) context_aggregator = llm.create_context_aggregator(context)
@@ -77,8 +75,8 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
tts, # TTS tts, # TTS
transport.output(), # Transport bot output transport.output(), # Transport bot output
context_aggregator.assistant(), # Assistant spoken responses context_aggregator.assistant(), # Assistant spoken responses
] ]
) )
task = PipelineTask( task = PipelineTask(
pipeline, pipeline,
@@ -94,7 +92,9 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
async def on_client_connected(transport, client): async def on_client_connected(transport, client):
logger.info(f"Client connected") logger.info(f"Client connected")
# Kick off the conversation. # Kick off the conversation.
messages.append({"role": "user", "content": [{"text": "Please introduce yourself to the user."}]}) messages.append(
{"role": "user", "content": [{"text": "Please introduce yourself to the user."}]}
)
await task.queue_frames([context_aggregator.user().get_context_frame()]) await task.queue_frames([context_aggregator.user().get_context_frame()])
@transport.event_handler("on_client_disconnected") @transport.event_handler("on_client_disconnected")

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

@@ -24,7 +24,7 @@ class BedrockLLMAdapter(BaseLLMAdapter):
"properties": function.properties, "properties": function.properties,
"required": function.required, "required": function.required,
}, },
} },
} }
} }

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

@@ -135,7 +135,7 @@ class BedrockLLMContext(OpenAILLMContext):
""" """
role = obj.get("role") role = obj.get("role")
content = obj.get("content") content = obj.get("content")
if role == "assistant": if role == "assistant":
if isinstance(content, str): if isinstance(content, str):
return [{"role": role, "content": [{"type": "text", "text": content}]}] return [{"role": role, "content": [{"type": "text", "text": content}]}]
@@ -184,7 +184,7 @@ class BedrockLLMContext(OpenAILLMContext):
result_content = json.dumps(content_item["json"]) result_content = json.dumps(content_item["json"])
else: else:
result_content = tool_result["content"] result_content = tool_result["content"]
tool_items.append( tool_items.append(
{ {
"role": "tool", "role": "tool",
@@ -226,26 +226,28 @@ class BedrockLLMContext(OpenAILLMContext):
if message["role"] == "tool": if message["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["content"].strip().endswith('}'): if message["content"].strip().startswith("{") and message[
"content"
].strip().endswith("}"):
content_json = json.loads(message["content"]) 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": message["content"]}]
except: except:
tool_result_content = [{"text": message["content"]}] tool_result_content = [{"text": message["content"]}]
return { return {
"role": "user", "role": "user",
"content": [ "content": [
{ {
"toolResult": { "toolResult": {
"toolUseId": message["tool_call_id"], "toolUseId": message["tool_call_id"],
"content": tool_result_content "content": tool_result_content,
}, },
}, },
], ],
} }
if message.get("tool_calls"): if message.get("tool_calls"):
tc = message["tool_calls"] tc = message["tool_calls"]
ret = {"role": "assistant", "content": []} ret = {"role": "assistant", "content": []}
@@ -261,7 +263,7 @@ class BedrockLLMContext(OpenAILLMContext):
} }
ret["content"].append(new_tool_use) ret["content"].append(new_tool_use)
return ret return ret
# Handle text content # Handle text content
content = message.get("content") content = message.get("content")
if isinstance(content, str): if isinstance(content, str):
@@ -276,7 +278,7 @@ class BedrockLLMContext(OpenAILLMContext):
text_content = item["text"] if item["text"] != "" else "(empty)" text_content = item["text"] if item["text"] != "" else "(empty)"
new_content.append({"text": text_content}) new_content.append({"text": text_content})
return {"role": message["role"], "content": new_content} return {"role": message["role"], "content": new_content}
return message return message
def add_image_frame_message( def add_image_frame_message(
@@ -287,15 +289,7 @@ class BedrockLLMContext(OpenAILLMContext):
encoded_image = base64.b64encode(buffer.getvalue()).decode("utf-8") encoded_image = base64.b64encode(buffer.getvalue()).decode("utf-8")
# Image should be the first content block in the message # Image should be the first content block in the message
content = [ content = [{"type": "image", "format": "jpeg", "source": {"bytes": encoded_image}}]
{
"type": "image",
"format": "jpeg",
"source": {
"bytes": encoded_image
}
}
]
if text: if text:
content.append({"text": text}) content.append({"text": text})
self.add_message({"role": "user", "content": content}) self.add_message({"role": "user", "content": content})
@@ -309,9 +303,7 @@ class BedrockLLMContext(OpenAILLMContext):
# if the last message has just a content string, convert it to a list # if the last message has just a content string, convert it to a list
# in the proper format # in the proper format
if isinstance(self.messages[-1]["content"], str): if isinstance(self.messages[-1]["content"], str):
self.messages[-1]["content"] = [ self.messages[-1]["content"] = [{"text": self.messages[-1]["content"]}]
{"text": self.messages[-1]["content"]}
]
# if this message has just a content string, convert it to a list # if this message has just a content string, convert it to a list
# in the proper format # in the proper format
if isinstance(message["content"], str): if isinstance(message["content"], str):
@@ -326,7 +318,7 @@ class BedrockLLMContext(OpenAILLMContext):
logger.error(f"Error adding message: {e}") logger.error(f"Error adding message: {e}")
def _restructure_from_bedrock_messages(self): def _restructure_from_bedrock_messages(self):
"""Restructure messages in Bedrock format by handling system messages, """Restructure messages in Bedrock format by handling system messages,
merging consecutive messages with the same role, and ensuring proper content formatting. merging consecutive messages with the same role, and ensuring proper content formatting.
""" """
# Handle system message if present at the beginning # Handle system message if present at the beginning
@@ -338,7 +330,7 @@ class BedrockLLMContext(OpenAILLMContext):
system_content = self.messages.pop(0)["content"] system_content = self.messages.pop(0)["content"]
if isinstance(system_content, str): if isinstance(system_content, str):
system_content = [{"text": system_content}] system_content = [{"text": system_content}]
if self.system: if self.system:
if isinstance(self.system, str): if isinstance(self.system, str):
self.system = [{"text": self.system}] self.system = [{"text": self.system}]
@@ -366,7 +358,7 @@ class BedrockLLMContext(OpenAILLMContext):
merged_messages[-1]["content"].extend(msg["content"]) merged_messages[-1]["content"].extend(msg["content"])
else: else:
merged_messages.append(msg) merged_messages.append(msg)
self.messages.clear() self.messages.clear()
self.messages.extend(merged_messages) self.messages.extend(merged_messages)
@@ -452,7 +444,7 @@ class BedrockAssistantContextAggregator(LLMAssistantContextAggregator):
"toolUse": { "toolUse": {
"toolUseId": frame.tool_call_id, "toolUseId": frame.tool_call_id,
"name": frame.function_name, "name": frame.function_name,
"input": frame.arguments if frame.arguments else {} "input": frame.arguments if frame.arguments else {},
} }
} }
], ],
@@ -465,11 +457,7 @@ class BedrockAssistantContextAggregator(LLMAssistantContextAggregator):
{ {
"toolResult": { "toolResult": {
"toolUseId": frame.tool_call_id, "toolUseId": frame.tool_call_id,
"content": [ "content": [{"text": "IN_PROGRESS"}],
{
"text": "IN_PROGRESS"
}
],
} }
} }
], ],
@@ -517,9 +505,10 @@ class BedrockAssistantContextAggregator(LLMAssistantContextAggregator):
class BedrockLLMService(LLMService): class BedrockLLMService(LLMService):
"""This class implements inference with AWS Bedrock models including Amazon Nova and Anthropic Claude. """This class implements inference with AWS Bedrock models including Amazon Nova and Anthropic Claude.
Requires AWS credentials to be configured in the environment or through boto3 configuration. Requires AWS credentials to be configured in the environment or through boto3 configuration.
""" """
class InputParams(BaseModel): class InputParams(BaseModel):
max_tokens: Optional[int] = Field(default_factory=lambda: 4096, ge=1) max_tokens: Optional[int] = Field(default_factory=lambda: 4096, ge=1)
temperature: Optional[float] = Field(default_factory=lambda: 0.7, ge=0.0, le=1.0) temperature: Optional[float] = Field(default_factory=lambda: 0.7, ge=0.0, le=1.0)
@@ -541,34 +530,33 @@ class BedrockLLMService(LLMService):
**kwargs, **kwargs,
): ):
super().__init__(**kwargs) super().__init__(**kwargs)
# Initialize the Bedrock client # Initialize the Bedrock client
if not client_config: if not client_config:
client_config = Config( client_config = Config(
connect_timeout=300, # 5 minutes connect_timeout=300, # 5 minutes
read_timeout=300, # 5 minutes read_timeout=300, # 5 minutes
retries={'max_attempts': 3} retries={"max_attempts": 3},
) )
session = boto3.Session( session = boto3.Session(
aws_access_key_id=aws_access_key, aws_access_key_id=aws_access_key,
aws_secret_access_key=aws_secret_key, aws_secret_access_key=aws_secret_key,
aws_session_token=aws_session_token, aws_session_token=aws_session_token,
region_name=aws_region region_name=aws_region,
) )
self._client = session.client( self._client = session.client(service_name="bedrock-runtime", config=client_config)
service_name='bedrock-runtime',
config=client_config
)
self.set_model_name(model) self.set_model_name(model)
self._settings = { self._settings = {
"max_tokens": params.max_tokens, "max_tokens": params.max_tokens,
"temperature": params.temperature, "temperature": params.temperature,
"top_p": params.top_p, "top_p": params.top_p,
"latency": params.latency, "latency": params.latency,
"additional_model_request_fields": params.additional_model_request_fields if isinstance(params.additional_model_request_fields, dict) else {}, "additional_model_request_fields": params.additional_model_request_fields
if isinstance(params.additional_model_request_fields, dict)
else {},
} }
logger.info(f"Using AWS Bedrock model: {model}") logger.info(f"Using AWS Bedrock model: {model}")
def can_generate_metrics(self) -> bool: def can_generate_metrics(self) -> bool:
@@ -603,7 +591,7 @@ class BedrockLLMService(LLMService):
if isinstance(context, OpenAILLMContext) and not isinstance(context, BedrockLLMContext): if isinstance(context, OpenAILLMContext) and not isinstance(context, BedrockLLMContext):
context = BedrockLLMContext.from_openai_context(context) context = BedrockLLMContext.from_openai_context(context)
user = BedrockUserContextAggregator(context, **user_kwargs) user = BedrockUserContextAggregator(context, **user_kwargs)
assistant = BedrockAssistantContextAggregator(context, **assistant_kwargs) assistant = BedrockAssistantContextAggregator(context, **assistant_kwargs)
return BedrockContextAggregatorPair(_user=user, _assistant=assistant) return BedrockContextAggregatorPair(_user=user, _assistant=assistant)
@@ -626,31 +614,29 @@ class BedrockLLMService(LLMService):
# ) # )
await self.start_ttfb_metrics() await self.start_ttfb_metrics()
# Set up inference config # Set up inference config
inference_config = { inference_config = {
"maxTokens": self._settings["max_tokens"], "maxTokens": self._settings["max_tokens"],
"temperature": self._settings["temperature"], "temperature": self._settings["temperature"],
"topP": self._settings["top_p"], "topP": self._settings["top_p"],
} }
# Prepare request parameters # Prepare request parameters
request_params = { request_params = {
"modelId": self.model_name, "modelId": self.model_name,
"messages": context.messages, "messages": context.messages,
"inferenceConfig": inference_config, "inferenceConfig": inference_config,
"additionalModelRequestFields": self._settings["additional_model_request_fields"] "additionalModelRequestFields": self._settings["additional_model_request_fields"],
} }
# Add system message # Add system message
request_params["system"] = context.system request_params["system"] = context.system
# Add tools if present # Add tools if present
if context.tools: if context.tools:
tool_config = { tool_config = {"tools": context.tools}
"tools": context.tools
}
# Add tool_choice if specified # Add tool_choice if specified
if context.tool_choice: if context.tool_choice:
if context.tool_choice == "auto": if context.tool_choice == "auto":
@@ -658,32 +644,30 @@ class BedrockLLMService(LLMService):
elif context.tool_choice == "none": elif context.tool_choice == "none":
# Skip adding toolChoice for "none" # Skip adding toolChoice for "none"
pass pass
elif isinstance(context.tool_choice, dict) and "function" in context.tool_choice: elif (
isinstance(context.tool_choice, dict) and "function" in context.tool_choice
):
tool_config["toolChoice"] = { tool_config["toolChoice"] = {
"tool": { "tool": {"name": context.tool_choice["function"]["name"]}
"name": context.tool_choice["function"]["name"]
}
} }
request_params["toolConfig"] = tool_config request_params["toolConfig"] = tool_config
# Add performance config if latency is specified # Add performance config if latency is specified
if self._settings["latency"] in ["standard", "optimized"]: if self._settings["latency"] in ["standard", "optimized"]:
request_params["performanceConfig"] = { request_params["performanceConfig"] = {"latency": self._settings["latency"]}
"latency": self._settings["latency"]
}
logger.debug(f"Calling Bedrock model with: {request_params}") logger.debug(f"Calling Bedrock model with: {request_params}")
# Call Bedrock with streaming # Call Bedrock with streaming
response = self._client.converse_stream(**request_params) response = self._client.converse_stream(**request_params)
await self.stop_ttfb_metrics() await self.stop_ttfb_metrics()
# Process the streaming response # Process the streaming response
tool_use_block = None tool_use_block = None
json_accumulator = "" json_accumulator = ""
for event in response["stream"]: for event in response["stream"]:
# Handle text content # Handle text content
if "contentBlockDelta" in event: if "contentBlockDelta" in event:
@@ -694,18 +678,20 @@ class BedrockLLMService(LLMService):
elif "toolUse" in delta and "input" in delta["toolUse"]: elif "toolUse" in delta and "input" in delta["toolUse"]:
# Handle partial JSON for tool use # Handle partial JSON for tool use
json_accumulator += delta["toolUse"]["input"] json_accumulator += delta["toolUse"]["input"]
completion_tokens_estimate += self._estimate_tokens(delta["toolUse"]["input"]) completion_tokens_estimate += self._estimate_tokens(
delta["toolUse"]["input"]
)
# Handle tool use start # Handle tool use start
elif "contentBlockStart" in event: elif "contentBlockStart" in event:
content_block_start = event["contentBlockStart"]['start'] content_block_start = event["contentBlockStart"]["start"]
if "toolUse" in content_block_start: if "toolUse" in content_block_start:
tool_use_block = { tool_use_block = {
"id": content_block_start["toolUse"].get("toolUseId", ""), "id": content_block_start["toolUse"].get("toolUseId", ""),
"name": content_block_start["toolUse"].get("name", "") "name": content_block_start["toolUse"].get("name", ""),
} }
json_accumulator = "" json_accumulator = ""
# Handle message completion with tool use # Handle message completion with tool use
elif "messageStop" in event and "stopReason" in event["messageStop"]: elif "messageStop" in event and "stopReason" in event["messageStop"]:
if event["messageStop"]["stopReason"] == "tool_use" and tool_use_block: if event["messageStop"]["stopReason"] == "tool_use" and tool_use_block:
@@ -719,7 +705,7 @@ class BedrockLLMService(LLMService):
) )
except json.JSONDecodeError: except json.JSONDecodeError:
logger.error(f"Failed to parse tool arguments: {json_accumulator}") logger.error(f"Failed to parse tool arguments: {json_accumulator}")
# Handle usage metrics if available # Handle usage metrics if available
if "metadata" in event and "usage" in event["metadata"]: if "metadata" in event and "usage" in event["metadata"]:
usage = event["metadata"]["usage"] usage = event["metadata"]["usage"]
@@ -750,7 +736,7 @@ class BedrockLLMService(LLMService):
prompt_tokens=prompt_tokens, prompt_tokens=prompt_tokens,
completion_tokens=comp_tokens, completion_tokens=comp_tokens,
cache_read_input_tokens=cache_read_input_tokens, cache_read_input_tokens=cache_read_input_tokens,
cache_creation_input_tokens=cache_creation_input_tokens cache_creation_input_tokens=cache_creation_input_tokens,
) )
async def process_frame(self, frame: Frame, direction: FrameDirection): async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -783,7 +769,7 @@ class BedrockLLMService(LLMService):
prompt_tokens: int, prompt_tokens: int,
completion_tokens: int, completion_tokens: int,
cache_read_input_tokens: int, cache_read_input_tokens: int,
cache_creation_input_tokens: int cache_creation_input_tokens: int,
): ):
if prompt_tokens or completion_tokens: if prompt_tokens or completion_tokens:
tokens = LLMTokenUsage( tokens = LLMTokenUsage(
@@ -791,6 +777,6 @@ class BedrockLLMService(LLMService):
completion_tokens=completion_tokens, completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens, total_tokens=prompt_tokens + completion_tokens,
cache_read_input_tokens=cache_read_input_tokens, cache_read_input_tokens=cache_read_input_tokens,
cache_creation_input_tokens=cache_creation_input_tokens cache_creation_input_tokens=cache_creation_input_tokens,
) )
await self.start_llm_usage_metrics(tokens) await self.start_llm_usage_metrics(tokens)

4
src/pipecat/services/aws/stt.py
View File

@@ -19,7 +19,7 @@ from pipecat.frames.frames import (
Frame, Frame,
TranscriptionFrame, TranscriptionFrame,
InterimTranscriptionFrame, InterimTranscriptionFrame,
StartFrame StartFrame,
) )
from pipecat.services.ai_services import STTService from pipecat.services.ai_services import STTService
from pipecat.transcriptions.language import Language from pipecat.transcriptions.language import Language
@@ -597,4 +597,4 @@ class TranscribeSTTService(STTService):
except Exception as e: except Exception as e:
logger.error(f"Unexpected error in receive loop: {e}") logger.error(f"Unexpected error in receive loop: {e}")
finally: finally:
logger.debug("Receive loop ended") logger.debug("Receive loop ended")

6
src/pipecat/services/aws/tts.py
View File

@@ -17,7 +17,7 @@ from pipecat.frames.frames import (
Frame, Frame,
TTSAudioRawFrame, TTSAudioRawFrame,
TTSStartedFrame, TTSStartedFrame,
TTSStoppedFrame TTSStoppedFrame,
) )
from pipecat.services.ai_services import TTSService from pipecat.services.ai_services import TTSService
from pipecat.transcriptions.language import Language from pipecat.transcriptions.language import Language
@@ -187,7 +187,7 @@ class PollyTTSService(TTSService):
if self._settings["engine"] == "standard": if self._settings["engine"] == "standard":
if self._settings["pitch"]: if self._settings["pitch"]:
prosody_attrs.append(f"pitch='{self._settings['pitch']}'") prosody_attrs.append(f"pitch='{self._settings['pitch']}'")
if self._settings["rate"]: if self._settings["rate"]:
prosody_attrs.append(f"rate='{self._settings['rate']}'") prosody_attrs.append(f"rate='{self._settings['rate']}'")
if self._settings["volume"]: if self._settings["volume"]:
@@ -195,7 +195,7 @@ class PollyTTSService(TTSService):
# logger.warning("Prosody tags are not supported for generative engine. Ignoring.") # logger.warning("Prosody tags are not supported for generative engine. Ignoring.")
if prosody_attrs: if prosody_attrs:
ssml += f"<prosody {' '.join(prosody_attrs)}>" ssml += f"<prosody {' '.join(prosody_attrs)}>"
ssml += text ssml += text