Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.pipecat.ai/llms.txt

Use this file to discover all available pages before exploring further.

So far you’ve built a single agent: one LLM, one context, one set of tools. But many problems are better served by several agents working together, each owning its own LLM. A greeter hands off to a support agent. A researcher runs in the background while the main agent keeps talking. A screen-driving agent acts on the UI while a voice agent converses. Giving each agent its own LLM keeps every context small and focused. Instead of one model juggling every instruction and tool, each agent reasons over just its own job. That’s cheaper, faster, and less prone to the model getting lost.

​
LLMWorker overview

Multi-agent systems often run several LLM-backed agents — a greeter, a support agent, a researcher — each with its own instructions, tools, and (optionally) its own conversation context. LLMWorker is the building block for each one. It extends PipelineWorker with everything you need to run an LLM-powered agent:
  • A pipeline with your LLM service, automatically built
  • Tool registration via the @tool decorator
  • Activation handling that injects messages and runs the LLM
To create an LLM agent, subclass LLMWorker so you can host @tool methods, then instantiate it with its own LLM service. Pass bridged=() so the agent receives frames from the bus: 
import os

from pipecat.services.openai.llm import OpenAILLMService
from pipecat.workers.llm import LLMWorker


class MyAgent(LLMWorker):
    """An LLM agent. ``@tool`` methods go here."""


def build_agent() -> MyAgent:
    llm = OpenAILLMService(
        api_key=os.environ["OPENAI_API_KEY"],
        settings=OpenAILLMService.Settings(
            system_instruction="You are a helpful assistant.",
        ),
    )
    return MyAgent("assistant", llm=llm, bridged=())
You never pass a bus= argument to the constructor — the worker gets its bus when you register it with runner.add_workers(...). The default pipeline is Pipeline([llm]), with tools from build_tools() automatically registered. When bridged=() is set, the framework wraps this pipeline with edge processors that connect it to the bus.

​
The @tool decorator

The @tool decorator marks a method as an LLM-callable tool. The framework automatically collects all @tool-decorated methods and registers them with the LLM service. 
from pipecat.services.llm_service import FunctionCallParams
from pipecat.workers.llm import tool


class MyAgent(LLMWorker):
    @tool
    async def get_weather(self, params: FunctionCallParams, city: str):
        """Get the current weather for a city.

        Args:
            city (str): The city name (e.g. 'San Francisco').
        """
        weather = await fetch_weather(city)
        await params.result_callback(weather)
The tool’s name comes from the method name. The docstring becomes the tool description. Parameter types and descriptions are extracted from the type annotations and the Args section in the docstring.

​
Tool options

The @tool decorator accepts options: 
@tool(cancel_on_interruption=False, timeout=60)
async def long_running_tool(self, params: FunctionCallParams, query: str):
    """A tool that takes a while.

    Args:
        query (str): The search query.
    """
    result = await expensive_search(query)
    await params.result_callback(result)
OptionDefaultDescription
cancel_on_interruptionTrueCancel the tool if the user interrupts
timeoutNoneMaximum execution time in seconds

​
Tool parameters

Every tool method receives self and params: FunctionCallParams as the first two arguments. Additional arguments are the tool’s parameters that the LLM fills in. The params object gives you access to:
  • params.result_callback(result) — return the result to the LLM
  • params.llm — the LLM service instance, useful for queuing frames

​
Returning results

Always call params.result_callback() to return the tool result to the LLM: 
@tool
async def lookup(self, params: FunctionCallParams, item: str):
    """Look up an item.

    Args:
        item (str): The item to look up.
    """
    data = await database.get(item)
    await params.result_callback({"found": True, "data": data})

​
Activation with messages

When an LLMWorker is activated, you can inject messages into its context. Pass an LLMWorkerActivationArgs via the args parameter: 
from pipecat.workers.llm import LLMWorkerActivationArgs

await self.activate_worker(
    "support",
    args=LLMWorkerActivationArgs(
        messages=[{"role": "developer", "content": "The user asked about pricing."}],
        run_llm=True,  # Run the LLM immediately after injection
    ),
)
The default on_activated() implementation:
  1. Sets the tools from build_tools()
  2. Injects the provided messages into the LLM context
  3. Runs the LLM if run_llm is True (the default when messages is set)

​
Managing context with LLMContextWorker

A plain LLMWorker runs an LLM but doesn’t manage conversation context on its own — it relies on context coming from elsewhere (for example, the main agent’s aggregators bridged in). When an agent needs to keep its own history, use LLMContextWorker. It extends LLMWorker with a built-in LLMContext and the user/assistant aggregator pair, building the pipeline as [user_aggregator, llm, assistant_aggregator] for you. 
from pipecat.workers.llm import LLMContextWorker


class AssistantAgent(LLMContextWorker):
    """An LLM agent that keeps its own conversation context."""


agent = AssistantAgent("assistant", llm=llm)  # gets its own context
Each LLMContextWorker gets its own context by default, so agents don’t see each other’s history. To give several agents a shared conversation, pass the same context= to each: 
from pipecat.processors.aggregators.llm_context import LLMContext

shared = LLMContext()
agent_a = AssistantAgent("agent_a", llm=llm_a, context=shared)
agent_b = AssistantAgent("agent_b", llm=llm_b, context=shared)
Access the managed aggregators via self.user_aggregator and self.assistant_aggregator.

​
Custom pipelines

If you need more control, you can pass a custom pipeline= to the LLMWorker constructor. For example, to add TTS to the agent’s own pipeline: 
from pipecat.pipeline.pipeline import Pipeline
from pipecat.services.cartesia.tts import CartesiaTTSService


class AgentWithTTS(LLMWorker):
    def __init__(self, name: str, *, llm, voice_id: str):
        tts = CartesiaTTSService(
            api_key=os.environ["CARTESIA_API_KEY"],
            settings=CartesiaTTSService.Settings(voice=voice_id),
        )
        super().__init__(name, llm=llm, pipeline=Pipeline([llm, tts]), bridged=())
This is how you give each agent its own voice — each agent adds its own TTS after the LLM, so a handoff sounds like a real transfer between distinct speakers.

​
What’s next

Now that your agents can run LLMs and call tools, here’s a powerful one: an agent that sees and drives the user’s screen.

Controlling the UI

A UIWorker that reads the screen and acts on it over a two-way RTVI interface