> ## Documentation Index
> Fetch the complete documentation index at: https://docs.tazolabs.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Speculative tool calling

> Build a voice-agent backend that stops talking and calls the right function while the caller is still finishing their sentence.

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](https://docs.tazolabs.com/sdk/python/overview.md) and two
[streaming sessions](https://docs.tazolabs.com/sdk/python/streaming.md).

```mermaid theme={null}
flowchart LR
    mic["caller audio"] --> fan["fan out"]
    fan --> a["Session 1 · chunk scope, every 1000 ms<br/>Is the caller talking?"]
    fan --> b["Session 2 · turn scope, every 500 ms<br/>Which function should be called?"]
    a --> s["Talking ≥ 70%<br/>→ stop_talking()"]
    b --> g{"same option ≥ 70%<br/>on 2 updates in a row?"}
    g -->|send_offer_details_to_mobile| t1["send_offer_details_to_mobile()"]
    g -->|end_call| t2["end_call()"]
```

## 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

```bash theme={null}
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](https://docs.tazolabs.com/sdk/python/streaming.md) 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.

```python theme={null}
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.

```python theme={null}
@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](#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.

```python theme={null}
@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.

```python theme={null}
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:

```python theme={null}
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](https://docs.tazolabs.com/guides/limits.md). |
| 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](#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.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.