Connection error in OpenAILLMService #1108
Comments
|
@markbackman Can this be linked to 8c0ecb8? |
I don't think so. We've tested this extensively since that change. Is this issue with OpenAI or OpenAI Realtime? Can you repro using one of the dynamic examples? |
|
@markbackman Yes I just reproduced with https://github.com/pipecat-ai/pipecat-flows/blob/main/examples/dynamic/insurance_openai.py. Same as before, sometimes it works for a couple messages, sometimes I get the error right away. Btw I had to comment Full code async def run_bot_livekit(self, room):
(url, token) = await configure_livekit(room["name"])
self.piplet_call_id = room["sid"]
logger.info(f"Running new bot for call: piplet_call_id - {room['sid']}")
transport = LiveKitTransport(
url=url,
token=token,
room_name=room["name"],
params=LiveKitParams(
audio_in_channels=1,
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
vad_enabled=True,
vad_audio_passthrough=True,
),
)
llm = OpenAILLMService(api_key=self.openaiApiKey, model="gpt-4o")
stt = DeepgramSTTService(
api_key=self.deepgramApiKey,
audio_passthrough=True,
live_options=LiveOptions(
language=Language.FR,
),
)
tts = CartesiaTTSService(
api_key=Config.CARTESIA_API_KEY,
voice_id="34a34fd0-157b-4470-aaea-db59a8efcfb9",
model="sonic",
params=CartesiaTTSService.InputParams(language=Language.FR),
)
context = OpenAILLMContext()
context_aggregator = llm.create_context_aggregator(context)
pipeline = Pipeline(
[
transport.input(), # Websocket input from client
stt, # Speech-To-Text
context_aggregator.user(),
llm, # LLM
tts, # Text-To-Speech
transport.output(), # Websocket output to client
context_aggregator.assistant(),
]
)
class InsuranceQuote(TypedDict):
monthly_premium: float
coverage_amount: int
deductible: int
class AgeCollectionResult(FlowResult):
age: int
class MaritalStatusResult(FlowResult):
marital_status: str
class QuoteCalculationResult(FlowResult, InsuranceQuote):
pass
class CoverageUpdateResult(FlowResult, InsuranceQuote):
pass
# Simulated insurance data
INSURANCE_RATES = {
"young_single": {"base_rate": 150, "risk_multiplier": 1.5},
"young_married": {"base_rate": 130, "risk_multiplier": 1.3},
"adult_single": {"base_rate": 100, "risk_multiplier": 1.0},
"adult_married": {"base_rate": 90, "risk_multiplier": 0.9},
}
# Function handlers
async def collect_age(args: FlowArgs) -> AgeCollectionResult:
"""Process age collection."""
age = args["age"]
logger.debug(f"collect_age handler executing with age: {age}")
return AgeCollectionResult(age=age)
async def collect_marital_status(args: FlowArgs) -> MaritalStatusResult:
"""Process marital status collection."""
status = args["marital_status"]
logger.debug(
f"collect_marital_status handler executing with status: {status}"
)
return MaritalStatusResult(marital_status=status)
async def calculate_quote(args: FlowArgs) -> QuoteCalculationResult:
"""Calculate insurance quote based on age and marital status."""
age = args["age"]
marital_status = args["marital_status"]
logger.debug(
f"calculate_quote handler executing with age: {age}, status: {marital_status}"
)
# Determine rate category
age_category = "young" if age < 25 else "adult"
rate_key = f"{age_category}_{marital_status}"
rates = INSURANCE_RATES.get(rate_key, INSURANCE_RATES["adult_single"])
# Calculate quote
monthly_premium = rates["base_rate"] * rates["risk_multiplier"]
return {
"monthly_premium": monthly_premium,
"coverage_amount": 250000,
"deductible": 1000,
}
async def update_coverage(args: FlowArgs) -> CoverageUpdateResult:
"""Update coverage options and recalculate premium."""
coverage_amount = args["coverage_amount"]
deductible = args["deductible"]
logger.debug(
f"update_coverage handler executing with amount: {coverage_amount}, deductible: {deductible}"
)
# Calculate adjusted quote
monthly_premium = (coverage_amount / 250000) * 100
if deductible > 1000:
monthly_premium *= 0.9 # 10% discount for higher deductible
return {
"monthly_premium": monthly_premium,
"coverage_amount": coverage_amount,
"deductible": deductible,
}
async def end_quote() -> FlowResult:
"""Handle quote completion."""
logger.debug("end_quote handler executing")
return {"status": "completed"}
# Transition callbacks and handlers
async def handle_age_collection(args: Dict, flow_manager: FlowManager):
flow_manager.state["age"] = args["age"]
await flow_manager.set_node("marital_status", create_marital_status_node())
async def handle_marital_status_collection(
args: Dict, flow_manager: FlowManager
):
flow_manager.state["marital_status"] = args["marital_status"]
await flow_manager.set_node(
"quote_calculation",
create_quote_calculation_node(
flow_manager.state["age"], flow_manager.state["marital_status"]
),
)
async def handle_quote_calculation(args: Dict, flow_manager: FlowManager):
quote = await calculate_quote(args)
flow_manager.state["quote"] = quote
await flow_manager.set_node(
"quote_results", create_quote_results_node(quote)
)
async def handle_end_quote(_: Dict, flow_manager: FlowManager):
await flow_manager.set_node("end", create_end_node())
# Node configurations
def create_initial_node() -> NodeConfig:
"""Create the initial node asking for age."""
return {
"role_messages": [
{
"role": "system",
"content": """You are a friendly insurance agent. Your responses will be
converted to audio, so avoid special characters. Always use
the available functions to progress the conversation naturally.""",
}
],
"task_messages": [
{
"role": "system",
"content": "Start by asking for the customer's age.",
}
],
"functions": [
{
"type": "function",
"function": {
"name": "collect_age",
# "handler": collect_age,
"description": "Record customer's age",
"parameters": {
"type": "object",
"properties": {"age": {"type": "integer"}},
"required": ["age"],
},
# "transition_callback": handle_age_collection,
},
}
],
}
def create_marital_status_node() -> NodeConfig:
"""Create node for collecting marital status."""
return {
"task_messages": [
{
"role": "system",
"content": "Ask about the customer's marital status for premium calculation.",
}
],
"functions": [
{
"type": "function",
"function": {
"name": "collect_marital_status",
"handler": collect_marital_status,
"description": "Record marital status",
"parameters": {
"type": "object",
"properties": {
"marital_status": {
"type": "string",
"enum": ["single", "married"],
}
},
"required": ["marital_status"],
},
"transition_callback": handle_marital_status_collection,
},
}
],
}
def create_quote_calculation_node(age: int, marital_status: str) -> NodeConfig:
"""Create node for calculating initial quote."""
return {
"task_messages": [
{
"role": "system",
"content": (
f"Calculate a quote for {age} year old {marital_status} customer. "
"First, call calculate_quote with their information. "
"Then explain the quote details and ask if they'd like to adjust coverage."
),
}
],
"functions": [
{
"type": "function",
"function": {
"name": "calculate_quote",
"handler": calculate_quote,
"description": "Calculate initial insurance quote",
"parameters": {
"type": "object",
"properties": {
"age": {"type": "integer"},
"marital_status": {
"type": "string",
"enum": ["single", "married"],
},
},
"required": ["age", "marital_status"],
},
"transition_callback": handle_quote_calculation,
},
}
],
}
def create_quote_results_node(
quote: Union[QuoteCalculationResult, CoverageUpdateResult],
) -> NodeConfig:
"""Create node for showing quote and adjustment options."""
return {
"task_messages": [
{
"role": "system",
"content": (
f"Quote details:\n"
f"Monthly Premium: ${quote['monthly_premium']:.2f}\n"
f"Coverage Amount: ${quote['coverage_amount']:,}\n"
f"Deductible: ${quote['deductible']:,}\n\n"
"Explain these quote details to the customer. When they request changes, "
"use update_coverage to recalculate their quote. Explain how their "
"changes affected the premium and compare it to their previous quote. "
"Ask if they'd like to make any other adjustments or if they're ready "
"to end the quote process."
),
}
],
"functions": [
{
"type": "function",
"function": {
"name": "update_coverage",
"handler": update_coverage,
"description": "Recalculate quote with new coverage options",
"parameters": {
"type": "object",
"properties": {
"coverage_amount": {"type": "integer"},
"deductible": {"type": "integer"},
},
"required": ["coverage_amount", "deductible"],
},
},
},
{
"type": "function",
"function": {
"name": "end_quote",
"handler": end_quote,
"description": "Complete the quote process",
"parameters": {"type": "object", "properties": {}},
"transition_callback": handle_end_quote,
},
},
],
}
def create_end_node() -> NodeConfig:
"""Create the final node."""
return {
"task_messages": [
{
"role": "system",
"content": (
"Thank the customer for their time and end the conversation. "
"Mention that a representative will contact them about the quote."
),
}
],
"functions": [],
"post_actions": [{"type": "end_conversation"}],
}
task = PipelineTask(
pipeline,
params=PipelineParams(allow_interruptions=True),
)
flow_manager = FlowManager(
task=task, llm=llm, context_aggregator=context_aggregator, tts=tts
)
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
# await transport.capture_participant_transcription(participant["id"])
logger.debug("Initializing flow")
await flow_manager.initialize()
logger.debug("Setting initial node")
await flow_manager.set_node("initial", create_initial_node())
runner = PipelineRunner(handle_sigint=False)
await runner.run(task)Error trace |
|
What version of Flows are you using? On |
|
@markbackman on my side, I am not using Flows. What is breaking change you mentioned above ? |
|
@KevGTL I just released Flows 0.0.12: https://github.com/pipecat-ai/pipecat-flows/releases/tag/v0.0.12. Try updating to it to see if it solves your issues. Also, you need to include handlers and especially the |
There's a specific breaking change in the Flows code. If you're not using Flows, it won't affect you. I see no issues in connecting to OpenAI via the OpenAILLMService. |
|
I reverted to version 0.0.53 and still seeing the issue.. 15:04:42.965 | ERROR | pipecat.processors.frame_processor:__input_frame_task_handler:321 - OpenAILLMService#0: Uncaught exception peer closed connection without sending complete message body (incomplete chunked read) |
|
@Raphyo What's your repro case? Can you hit this with the 14-function-calling.py example? If so, how frequently does this occur? |
|
I am currently using FastAPIWebsocketTransport so it would require some work to try on 14-function-calling.py example |
|
Is this issue impacting anyone? I still can't repro. OpenAI works fine and I've tested a lot over the last week with it. |
|
@markbackman Yes I still get errors on GPT-4o, but not on other models. I think it's an OpenAI infra problem, we can close this issue I guess. |
Description
When running pipecat dynamic flows with a Livekit transport for a voice call over webrtc, I get errors on LLM calls to OpenAI. It started happening today, but reverting to pipecat-ai 0.0.53 doesn't fix it.
Environment
pipecat-ai version: 0.0.54
python version: 3.11
OS: MacOS 15.0.1
Repro steps
It doesn't happen on every single llm call, but i usually get the error very quickly during a conversation (during the first 3 or 4 messages).
Logs
The text was updated successfully, but these errors were encountered: