Speculative tool calling

A voice agent usually waits for a transcript, asks an LLM what to do, and only then acts. That adds seconds of dead air after every turn. With Tazo you can decide from the audio itself, while the caller is still speaking: stop the agent the moment they interrupt, and call the right function as soon as their answer is clear. This cookbook builds the backend for a bank agent that offers a credit card upgrade. There is no UI. The backend takes a stream of caller audio and calls any of three functions:

Function	Called when
`stop_talking()`	The caller starts speaking over the agent.
`send_offer_details_to_mobile()`	The caller finished and clearly agreed.
`end_call(emotion, interest)`	The caller finished and clearly declined.

It uses the Tazo Python SDK and two streaming sessions.

Why two sessions

A streaming session accepts exactly one question, and the two questions need different settings:

Is the caller talking? looks at new audio only (audio_scope="chunk"), so it answers “right now” rather than “at some point in this turn”.
Which function should be called? looks at the whole turn so far (audio_scope="turn"), because a “yes” only means something with the words around it.

The backend sends every audio frame to both sessions.

Setup

python -m pip install tazo-sdk
export TAZO_API_KEY='your-key'

If you have a recorded .wav file instead of a live microphone, the SDK’s wav_chunks reads it and yields 100 ms frames of audio, which is how the example below replays a call. For a live call, yield frames from your telephony or microphone stream the same way. The streaming protocol has its own wire format, which the SDK handles for you; see streaming if you need the details.

Define the questions

The criteria are the real decision logic: each option says in plain language when it should be chosen. NO_TOOL is the safe default for silence, noise, mid-sentence pauses and unclear replies.

import asyncio
from collections.abc import AsyncIterator, Awaitable, Callable
from dataclasses import dataclass, field
from pathlib import Path

from tazo import AsyncTazo, wav_chunks
from tazo.types import ChoiceAnswer, DecisionEvent

THRESHOLD = 0.7  # minimum probability for an option to count
ACTION_STREAK = 2  # consecutive updates that must agree before a function runs
STOP_TO_ACTION_GAP_S = 2.0  # pause between stop_talking() and the function

CONTEXT = (
    "A bank agent is phoning a customer to offer an upgrade of their credit card to the "
    "Indus Platinum Rewards card. Only the caller's microphone audio is streamed to you; "
    "the agent's own voice is not included. Choose the one action the application should "
    "take, judged only from the caller's voice and words."
)

SPEAKING = {
    "speaking": {
        "type": "choice",
        "instructions": "Is the caller speaking in this audio right now?",
        "criteria": {
            "yes": "A person is clearly speaking: real words, a reply or a spoken interruption. Not coughs, clicks or background sounds.",
            "noise": "Non-speech sound such as background noise, music, a cough or typing, with no clear words from the caller.",
            "no": "Silence or near silence, with no speech and no notable noise.",
        },
    }
}

ACTION = {
    "action": {
        "type": "choice",
        "instructions": "Which action should the application take right now, based on what the caller has said so far?",
        "criteria": {
            "NO_TOOL": "Do nothing. No speech yet, only noise or silence, or the caller is mid-sentence or unclear.",
            "send_offer_details_to_mobile": "The caller has finished speaking and clearly agreed to the offer, for example yes, sure, okay or go ahead.",
            "end_call": "The caller has finished speaking and clearly declined, for example no, not interested or please do not call again.",
        },
    }
}

Define the functions

The controller calls these. Replace the bodies with your telephony and CRM code.

@dataclass
class Tools:
    """The functions the agent can call. Replace the bodies with your own telephony and CRM code."""

    stop_talking: Callable[[], Awaitable[None]]
    send_offer_details_to_mobile: Callable[[], Awaitable[None]]
    end_call: Callable[[str, str], Awaitable[None]]

Turn answers into function calls

Two rules keep this from firing early:

Threshold. An option only counts at 70% or more.
Streak. The same function option must win on 2 updates in a row, so a single noisy update never triggers a call.

In our design, once the caller starts speaking we want the agent to stop talking, because we do not want it to talk over them. The controller runs stop_talking() to enable this. Additionally, if the Action stream returns a clear yes or no at the same time as stop_talking() fires, the controller waits 2 seconds after stop_talking() before calling the function. The pause makes sure the caller has finished talking, so the agent is more likely to respond when the caller is not speaking, and the call does not end abruptly. A better approach however is to confirm the caller has stopped using the Speaking stream instead of a fixed delay. Another assumption we make in this example, as covered in Limitations: there is no re-ask step. If the caller says something unclear, NO_TOOL wins and the agent keeps waiting. A simple fix that reuses the existing streams: once the caller has started and then finished speaking, start a timeout. If the Action stream still has not produced a clear yes or no when it expires, have the agent repeat the offer, or ask whether the caller understood it and whether they are interested.

@dataclass
class CallController:
    tools: Tools
    agent_speaking: bool = True
    done: asyncio.Event = field(default_factory=asyncio.Event)
    _last_pick: str | None = None
    _count: int = 0

    async def on_speaking(self, event: DecisionEvent) -> None:
        answer = event.answers["speaking"]
        assert isinstance(answer, ChoiceAnswer)
        if self.agent_speaking and answer.probabilities.get("yes", 0) >= THRESHOLD:
            await self._stop_talking()

    async def on_action(self, event: DecisionEvent) -> None:
        answer = event.answers["action"]
        assert isinstance(answer, ChoiceAnswer)
        pick = max(answer.probabilities, key=answer.probabilities.__getitem__)
        confident = answer.probabilities[pick] >= THRESHOLD and pick != "NO_TOOL"
        key = pick if confident else None
        self._count = self._count + 1 if key is not None and key == self._last_pick else (1 if key else 0)
        self._last_pick = key
        if key is None or self._count < ACTION_STREAK or self.done.is_set():
            return
        self.done.set()
        # Runs stop_talking() first when the agent is still mid-sentence, then the function.
        if self.agent_speaking:
            await self._stop_talking()
            await asyncio.sleep(STOP_TO_ACTION_GAP_S)
        if key == "send_offer_details_to_mobile":
            await self.tools.send_offer_details_to_mobile()
        else:
            await self.tools.end_call("neutral", "not_interested")

    async def _stop_talking(self) -> None:
        self.agent_speaking = False
        await self.tools.stop_talking()

Run a call

run_call opens both sessions, sends each frame to both, and handles their events in parallel. The context managers authenticate and wait for session.ready. Once a function has run, the loop stops sending audio and closes both sessions. If the audio ends without a decision, the backend commits both sessions so Tazo returns a final answer for the turn.

async def run_call(frames: AsyncIterator[bytes], tools: Tools) -> None:
    """Fan one caller-audio stream out to two Tazo sessions and drive the tools from their answers."""
    controller = CallController(tools)
    async with AsyncTazo() as client:
        async with (
            client.decisions.stream(
                state=CONTEXT, questions=SPEAKING, audio_scope="chunk", evaluation_interval_ms=1000
            ) as speaking,
            client.decisions.stream(
                state=CONTEXT, questions=ACTION, audio_scope="turn", evaluation_interval_ms=500
            ) as action,
        ):
            finalized = {"speaking": asyncio.Event(), "action": asyncio.Event()}

            async def send() -> None:
                async for frame in frames:
                    await asyncio.gather(speaking.send_audio(frame), action.send_audio(frame))
                    if controller.done.is_set():
                        break
                else:
                    # The caller hung up without a decision: ask for the final answer.
                    await asyncio.gather(speaking.commit(), action.commit())
                    await asyncio.gather(*(event.wait() for event in finalized.values()))
                await asyncio.gather(speaking.finish(), action.finish())

            async def receive(name: str, session, handler) -> None:
                async for event in session:
                    if event.type in {"decision.updated", "decision.finalized"}:
                        await handler(event)
                    if event.type == "decision.finalized":
                        finalized[name].set()

            async with asyncio.TaskGroup() as group:
                group.create_task(send())
                group.create_task(receive("speaking", speaking, controller.on_speaking))
                group.create_task(receive("action", action, controller.on_action))

Try it with a recording of a caller:

async def paced_wav(path: Path) -> AsyncIterator[bytes]:
    for chunk in wav_chunks(path):
        await asyncio.sleep(len(chunk) / 32000)  # 32,000 bytes of audio per second
        yield chunk


async def main() -> None:
    async def stop_talking() -> None:
        print("stop_talking()")

    async def send_offer_details_to_mobile() -> None:
        print("send_offer_details_to_mobile()")

    async def end_call(emotion: str, interest: str) -> None:
        print(f"end_call(emotion={emotion}, interest={interest})")

    tools = Tools(stop_talking, send_offer_details_to_mobile, end_call)
    await run_call(paced_wav(Path("caller_says_yes.wav")), tools)


asyncio.run(main())

paced_wav sleeps for the duration of each frame to imitate a microphone. With a live call, frames already arrive in real time, so you would pass them straight in.

Tune it

Setting	Effect
`THRESHOLD`	Higher means fewer wrong calls, but slower triggers.
`ACTION_STREAK`	Higher means more consecutive agreement is needed before acting.
`evaluation_interval_ms`	How often Tazo re-evaluates. Each evaluation uses your account’s inference allowance.
Option descriptions	The most effective lever. Add examples from your own calls.

Limitations

end_call always uses emotion="neutral". Add a second session with a sentiment question if you want to pass the caller’s emotion through.
There is no re-ask step: if the caller says something unclear, NO_TOOL wins and the agent keeps waiting. In addition to the fix mentioned in Turn answers into function calls, you can also add more streams (questions) for better signals to steer the agent, which would result in a slightly more complex workflow yet better experience for the end user.
Sessions are not reconnected automatically. If one fails, the TaskGroup raises a StreamingError and you decide whether to retry the call.

​Why two sessions

​Setup

​Define the questions

​Define the functions

​Turn answers into function calls

​Run a call

​Tune it

​Limitations