Bolna

To add support for a new telephony provider (e.g., Vonage, Telnyx), ensure the provider supports bi-directional streaming and follow these four steps:

1. Verify Streaming Support: Confirm the provider supports bi-directional streaming.
2. Implement Input Handler: Create a telephony-specific input handler file in bolna/input_handlers/telephony_providers. This file must contain custom functions that extend the telephony.py class and define how event packets from the provider are ingested.
3. Implement Output Handler: Create a telephony-specific output handler file in bolna/output_handlers/telephony_providers. This file must extend the telephony.py class and handle converting audio from the synthesizer class into a supported format for streaming over the provider's websocket.
4. Create a Dedicated Server: Write a dedicated server to initiate calls over websockets. You can use local_setup/telephony_server/twilio_api_server.py as a reference implementation.

A GraphAgentConfig defines a conversation flow using a directed graph of GraphNodes and GraphEdges.

Nodes

Nodes can be of type NodeType.LLM (standard conversation) or NodeType.ROUTER (silent dispatch).

• Router Nodes: Must NOT have a prompt or static_message. They must have at least one UNCONDITIONAL edge to ensure the conversation always advances.
• LLM Nodes: Contain a prompt, static_message, and edges.

Edges

Edges define transitions between nodes using several EdgeConditionTypes:

1. expression: Uses ExpressionGroup (logic + conditions) to evaluate variables (e.g., recipient_data.age).
2. event: Triggers based on a CallEvent name.
3. llm: The LLM decides the transition based on a condition description.
4. unconditional: Always triggers.

Edges also support priority to resolve conflicts between multiple valid transitions.

# Conceptual structure of a Graph Agent
agent_config = GraphAgentConfig(
    model="gpt-4o",
    agent_information="You are a helpful assistant.",
    nodes=[
        GraphNode(id="start", node_type="llm", prompt="Hello!", edges=[...]),
        GraphNode(id="router_1", node_type="router", edges=[...])
    ],
    current_node_id="start"
)

To run Bolna locally, use the provided docker-compose.yml file. The setup orchestrates several services including the main Bolna application, Redis for persistent storage, ngrok for local tunneling, and telephony servers (Twilio and Plivo).

Key Services

• bolna-app: The core Bolna service. It builds from dockerfiles/bolna_server.Dockerfile and exposes port 5001. It requires a .env file for environment variables and mounts local AWS credentials for cloud integration.
• redis: Used as persistent storage, running on port 6379.
• ngrok: Provides local tunneling. It requires a local ngrok-config.yml to be mounted to /etc/ngrok.yml inside the container. It exposes port 4040 for the dashboard.
• twilio-app: The Twilio telephony server, building from dockerfiles/twilio_server.Dockerfile and exposing port 8001.
• plivo-app: The Plivo telephony server, building from dockerfiles/plivo_server.Dockerfile and exposing port 8002.

Required Files and Volumes

Before running docker-compose up, ensure the following are present:

1. A .env file in the same directory as the compose file.
2. An ngrok-config.yml file for ngrok configuration.
3. An agent_data directory located one level above the compose file (../agent_data) to persist agent information.
4. Valid AWS credentials at $HOME/.aws/credentials and $HOME/.aws/config if using AWS services.

# Example command to start the stack
docker-compose up

If you do not require audio processing, you can create a text-only pipeline by setting enable_textual_input=True in assistant.add_task() and omitting the transcriber and synthesizer configurations.

import asyncio
from bolna.assistant import Assistant
from bolna.models import LlmAgent, SimpleLlmAgent


async def main():
    assistant = Assistant(name="text_only_agent")

    llm_agent = LlmAgent(
        agent_type="simple_llm_agent",
        agent_flow_type="streaming",
        llm_config=SimpleLlmAgent(
            provider="openai",
            model="gpt-4o-mini",
            temperature=0.2,
        ),
    )

    # No transcriber/synthesizer; enable a text-only pipeline
    assistant.add_task(
        task_type="conversation",
        llm_agent=llm_agent,
        enable_textual_input=True,
    )

    async for chunk in assistant.execute():
        print(chunk)


if __name__ == "__main__":
    asyncio.run(main())

You can build and run a voice agent directly in Python without needing the local telephony setup. This involves configuring a Transcriber (ASR), an LlmAgent (LLM), and a Synthesizer (TTS), then adding them as a conversation task to an Assistant instance.

assistant.execute() returns an async generator that yields per-task result dictionaries (event-like chunks).

import asyncio
from bolna.assistant import Assistant
from bolna.models import (
    Transcriber,
    Synthesizer,
    ElevenLabsConfig,
    LlmAgent,
    SimpleLlmAgent,
)


async def main():
    assistant = Assistant(name="demo_agent")

    # Configure audio input (ASR)
    transcriber = Transcriber(provider="deepgram", model="nova-2", stream=True, language="en")

    # Configure LLM
    llm_agent = LlmAgent(
        agent_type="simple_llm_agent",
        agent_flow_type="streaming",
        llm_config=SimpleLlmAgent(
            provider="openai",
            model="gpt-4o-mini",
            temperature=0.3,
        ),
    )

    # Configure audio output (TTS)
    synthesizer = Synthesizer(
        provider="elevenlabs",
        provider_config=ElevenLabsConfig(voice="George", voice_id="JBFqnCBsd6RMkjVDRZzb", model="eleven_turbo_v2_5"),
        stream=True,
        audio_format="wav",
    )

    # Build a single coherent pipeline: transcriber -> llm -> synthesizer
    assistant.add_task(
        task_type="conversation",
        llm_agent=llm_agent,
        transcriber=transcriber,
        synthesizer=synthesizer,
        enable_textual_input=False,
    )

    # Stream results
    async for chunk in assistant.execute():
        print(chunk)


if __name__ == "__main__":
    asyncio.run(main())