https://livekit.io logo
Join Slack
Powered by
Getting this error when transferring flow from one...
# ask-ai
g
Getting this error when transferring flow from one agent to another agent, Model : aws nova 2 sonic Realtime ERROR… livekit.…ugins.aws Error in _process_responses Traceback (most recent call last): File "/Users/yashrank/Documents/Work/mtm-livekit-real-time-ai-agent/ai-agent/src/venv/lib/python3.12/site-packages/livekit/agents/utils/log.py", line 17, in async_fn_logs return await fn(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Users/yashrank/Documents/Work/mtm-livekit-real-time-ai-agent/ai-agent/src/venv/lib/python3.12/site-packages/livekit/plugins/aws/experimental/realtime/realtime_model.py", line 1402, in _process_responses result = await output_stream.receive() ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Users/yashrank/Documents/Work/mtm-livekit-real-time-ai-agent/ai-agent/src/venv/lib/python3.12/site-packages/smithy_aws_event_stream/aio/__init__.py", line 136, in receive raise result.value # type: ignore ^^^^^^^^^^^^^^^^^^ aws_sdk_bedrock_runtime.models.ValidationException: Invalid input request, please fix your input and try again.
✅ 1
t
Thanks for reaching out, YASH. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
Thanks — I’m sorry you hit this error while transferring a flow between agents. I’m responding to the ValidationException you pasted (aws_sdk_bedrock_runtime.models.ValidationException: Invalid input request) that occurs inside the LiveKit AWS Nova Sonic Realtime plugin during 
_process_responses
. What this means • The Nova Sonic realtime model rejected the input it received when your transfer attempted to start the new agent’s reply. The model returned a ValidationException: “Invalid input request,” so the request payload sent to Bedrock/Realtime was malformed or used unsupported fields. What you can check and try next • Ensure you trigger the new agent’s reply with text input (not raw agent objects). ◦ Use 
session.generate_reply()
with 
instructions
or 
user_input
(string) to start the new agent’s intro instead of passing an agent instance or non-string payload. ◦ Example pattern:
Copy code
await self.session.generate_reply(
        instructions="Transfer to support agent and greet the user"
    )
• Confirm the model instance is Nova Sonic 2.0 and configured for text input: ◦ Use 
with_nova_sonic_2(...)
and set 
modalities="mixed"
so text input is accepted.
Copy code
session = AgentSession(
        llm=aws.realtime.RealtimeModel.with_nova_sonic_2(
            voice="tiffany",
            turn_detection="MEDIUM",
            modalities="mixed"
        ),
    )
• Verify the agent’s 
on_enter()
(or equivalent entry hook) calls 
generate_reply()
to produce the greeting. The new agent must actively call 
generate_reply()
to create the model request. • Validate the exact payload being sent to the Bedrock/Realtime API: ◦ Look for non-string fields, nested objects, or unexpected keys in the request body. ◦ Ensure fields expected by 
generate_reply()
are present and are strings (e.g., 
instructions
, 
user_input
). • Check AWS credentials/region: ◦ Make sure 
AWS_ACCESS_KEY_ID
, 
AWS_SECRET_ACCESS_KEY
, and 
AWS_REGION
are set (Bedrock requires a region). • Confirm you aren’t using Nova Sonic 1.0 (audio-only). Nova Sonic 1.x does not accept text input in the same way; Nova Sonic 2.0 supports text via 
generate_reply()
. If you’re still seeing the error, please share one of the following (so I can pinpoint the malformed field): • The snippet of code you use to perform the agent transfer (the function that returns the new Agent and the new agent’s 
on_enter()
), with any secrets redacted. • The exact 
generate_reply()
call arguments you pass when the transfer happens. • The raw request payload (or sanitized copy) that ends up being sent to the AWS realtime/BEDROCK API. I’ll look at that and give concrete fixes — for example, identifying which key in the payload is invalid and how to reformat it so Nova Sonic accepts it. Sources: Amazon Nova Sonic integration guide | LiveKit Documentation | TTS Comparison | LiveKit Documentation | AWS AI and LiveKit | LiveKit Documentation, with interactive elements
g
is there any mehtod to wait until agetn initiation is completed ?
t
Thanks — I’m responding to your question about waiting until an agent’s initiation is completed before continuing. Yes. You can reliably wait for an agent to finish initializing by using the session lifecycle and the connection steps: • Start the session and then await the connection. Call 
await session.start(...)
followed by `await ctx.connect()`; awaiting 
ctx.connect()
ensures the session/room connection and startup are complete before you proceed. • Listen for the 
agent_state_changed
event and wait until the agent enters the 
listening
state. When the agent emits 
listening
, initialization is finished and it’s ready to process input. Example sequence (conceptual): 1. 
await session.start(room=ctx.room, agent=YourAgent())
2. 
await ctx.connect()
# wait for session/room connection and initialization 3. Optionally, subscribe to 
agent_state_changed
and proceed when state == "listening" If you’d like, share which SDK/language you’re using (Python, JS, etc.) and I can show a small code example that awaits 
ctx.connect()
and/or waits for the 
agent_state_changed
event in that environment. Sources: Agent session | LiveKit Documentation | Gemini Realtime Agent with Live Vision | LiveKit Documentation, with interactive elements
g
python
t
Thanks — I’ll show a concise Python pattern you can use to wait until an agent’s initialization is complete (and safe to continue with the transfer). I’ll cover two reliable approaches you can use together: • Awaiting the connection (
ctx.connect()
) after 
session.start()
— ensures the session/room setup is established. • Waiting for the agent to emit the 
agent_state_changed
event and reach the 
listening
state — indicates the agent is fully ready. Example (conceptual Python; adapt names to your codebase):
Copy code
import asyncio

# Example callback that listens for agent state changes.
# Adjust event names to match your SDK if they differ.
async def wait_for_listening_state(ctx, timeout=10):
    """
    Wait until agent_state_changed -> 'listening' or timeout.
    Returns True if reached listening, False if timed out.
    """
    fut = asyncio.get_event_loop().create_future()

    def on_state_change(new_state):
        # If event delivers richer payload, inspect it and check for 'listening'
        if new_state == "listening":
            if not fut.done():
                fut.set_result(True)

    # subscribe to the event; replace with your actual subscription API
    unsubscribe = ctx.on("agent_state_changed", on_state_change)

    try:
        await asyncio.wait_for(fut, timeout=timeout)
        return True
    except asyncio.TimeoutError:
        return False
    finally:
        # clean up listener
        unsubscribe()

# Usage when starting a session and performing a transfer
async def start_and_wait_for_agent(session, ctx, agent_instance):
    # Start the session (creates AgentSession, etc.)
    await session.start(room=ctx.room, agent=agent_instance)

    # Ensure connection and initialization are established
    await ctx.connect()

    # Wait until agent transitions to 'listening' (or timeout)
    ready = await wait_for_listening_state(ctx, timeout=15)
    if not ready:
        raise RuntimeError("Agent did not enter listening state within timeout")

    # At this point the agent is initialized and ready
    # Safe to perform actions like transferring flows or calling generate_reply()
Practical notes and tips • Use both steps: 
await ctx.connect()
ensures the session/room connection is established; the 
agent_state_changed
-> 
listening
is the signal your agent is ready to process input. • Pick a reasonable timeout (e.g., 10–30s) and surface helpful errors if initialization fails. • If your SDK’s event payload provides an object (not a plain string), inspect the payload in 
on_state_change()
and check the appropriate field for 
"listening"
. • If you need to wait for the agent to finish its initial greeting (e.g., the initial 
generate_reply()
to complete), you can also: ◦ Have the agent call 
generate_reply()
in 
on_enter()
and emit/resolve a custom event when 
generate_reply()
returns, then await that event from the caller. ◦ Or listen for a model output/completion event if your SDK exposes one. If you’d like, paste the exact APIs your 
ctx
object exposes (how you subscribe to events and what the 
agent_state_changed
payload looks like) and I’ll adapt the snippet to your concrete SDK methods and event signatures. , with interactive elements
Open in Slack
PreviousNext
Powered by