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

# Runtime Hooks

> Intercept and process pipeline data at the STT, LLM, TTS, vision, and lifecycle stages.

Runtime hooks let you intercept and process data at each stage of the Pipeline without subclassing it. Register them with the `@pipeline.on()` decorator.

## Audio Processing Hooks

The `stt` and `tts` hooks replace the built-in component processing. Each receives an async iterator and yields processed results. Only one of each can be registered.

<Note>
  The `stt` and `tts` hooks fully replace the built-in component processing: you receive the raw stream and yield the processed result yourself.
</Note>

#### STT hook: clean up transcripts

A common use case is preprocessing the audio and normalizing the transcript before it reaches the LLM. This hook drops tiny audio chunks, then strips filler words from the transcribed text.

<CodeGroup>
  ```python title="main.py" Python theme={null}
  @pipeline.on("stt")
  async def stt_hook(audio_stream):
      # Drop very small audio chunks before transcribing
      async def filtered():
          async for audio in audio_stream:
              if len(audio) >= 300:
                  yield audio

      # Transcribe, then remove filler words from the text
      async for event in run_stt(filtered()):
          if event.data and event.data.text:
              text = re.sub(r"\b(uh|um|like)\b", "", event.data.text)
              event.data.text = " ".join(text.split())
          yield event
  ```
</CodeGroup>

#### TTS hook: fix pronunciation

A common use case is reshaping text so the voice reads it correctly. This hook spells out abbreviations so the TTS pronounces them as expected.

<CodeGroup>
  ```python title="main.py" Python theme={null}
  @pipeline.on("tts")
  async def tts_hook(text_stream):
      # Rewrite text before it is synthesized
      async def preprocess():
          async for text in text_stream:
              yield text.replace("AM", "A M").replace("PM", "P M")

      async for audio in run_tts(preprocess()):
          yield audio
  ```
</CodeGroup>

## Lifecycle Hooks

Lifecycle hooks are side-effect-only. Use them for logging, analytics, and triggering external actions. You can register multiple per event.

* `user_turn_start(transcript: str)`: user's final transcript is available.
* `user_turn_end()`: agent finished responding for the turn.
* `agent_turn_start()`: agent starts speaking.
* `agent_turn_end()`: agent finishes speaking.

<Tabs>
  <Tab title="user_turn_start">
    Fetch context for the turn. The final transcript is available here, so it is a good place to run a knowledge-base lookup before the LLM responds.

    <CodeGroup>
      ```python title="main.py" Python theme={null}
          @pipeline.on("user_turn_start")
          async def on_user_start(transcript: str):
              docs = await knowledge_base.search(transcript)
              agent.metadata["retrieved"] = docs
      ```
    </CodeGroup>
  </Tab>

  <Tab title="user_turn_end">
    Save the completed turn. Fires once the agent has finished responding, so the turn is complete. Use it to persist the exchange.

    <CodeGroup>
      ```python title="main.py" Python theme={null}
          @pipeline.on("user_turn_end")
          async def on_user_end():
              await db.save_turn(session_id, turn_count)
      ```
    </CodeGroup>
  </Tab>

  <Tab title="agent_turn_start">
    Show a speaking indicator. Fires when the agent starts speaking. Push a status update so the client UI can show the agent is talking.

    <CodeGroup>
      ```python title="main.py" Python theme={null}
          from zrt import PubSubPublishConfig

          @pipeline.on("agent_turn_start")
          async def on_agent_start():
              await agent.session.publish_message(
                  PubSubPublishConfig(topic="AGENT_STATUS", message="speaking")
              )
      ```
    </CodeGroup>
  </Tab>

  <Tab title="agent_turn_end">
    Clear the indicator. Fires when the agent finishes speaking. Use it to reset UI state or measure response latency.

    <CodeGroup>
      ```python title="main.py" Python theme={null}
          from zrt import PubSubPublishConfig

          @pipeline.on("agent_turn_end")
          async def on_agent_end():
              await agent.session.publish_message(
                  PubSubPublishConfig(topic="AGENT_STATUS", message="idle")
              )
      ```
    </CodeGroup>
  </Tab>
</Tabs>

## What's Next

<CardGroup cols={2}>
  <Card title="Pipeline Overview" icon="diagram-project" href="/build/configure-a-pipeline/overview">
    Review how the pipeline is configured before hooking into it.
  </Card>

  <Card title="Run Your Agent" icon="play" href="/build/run-the-runtime">
    Run the pipeline with an agent.
  </Card>
</CardGroup>

## References

<Tabs>
  <Tab title="Python">
    #### Examples

    <CardGroup cols={2}>
      <Card title="Pipeline Hooks" icon="github" href="https://github.com/ZeroRuntimeAI/zrt-python-sdk-examples/blob/main/features/pipeline_hooks.py">
        Hook into a cascade pipeline at runtime.
      </Card>
    </CardGroup>

    #### SDK Reference

    <CardGroup cols={2}>
      <Card title="Pipeline Hooks" icon="book" href="/api-reference/python/core/pipeline">
        `Pipeline Hooks` in the Python API reference.
      </Card>
    </CardGroup>
  </Tab>
</Tabs>