SambaNova

pip install "pipecat-ai[sambanova]"

from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.services.sambanova.llm import SambaNovaLLMService
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from openai.types.chat import ChatCompletionToolParam
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.services.llm_service import FunctionCallParams

# Configure service
llm = SambaNovaLLMService(
    api_key'your-sambanova-api-key',
    model='Llama-4-Maverick-17B-128E-Instruct',
    params=SambaNovaLLMService.InputParams(temperature=0.7, max_tokens=1024),
)

# Define function to call
async def fetch_weather(params: FunctionCallParams) -> Any:
    """Mock function that fetches the weather forcast from an API."""

    await params.result_callback({'conditions': 'nice', 'temperature': '20 Degrees Celsius'})

# Register function handlers
llm.register_function('get_current_weather', fetch_weather)

# Define weather function using standardized schema
weather_function = FunctionSchema(
    name='get_current_weather',
    description='Get the current weather',
    properties={
        'location': {
            'type': 'string',
            'description': 'The city and state.',
        },
        'format': {
            'type': 'string',
            'enum': ['celsius', 'fahrenheit'],
            'description': "The temperature unit to use. Infer this from the user's location.",
        },
    },
    required=['location', 'format'],
)

# Create tools schema
tools = ToolsSchema(standard_tools=[weather_function])

# Define system message
messages = [
    {
        'role': 'system',
        'content': 'You are a helpful LLM in a WebRTC call. '
        'Your goal is to demonstrate your capabilities of weather forecasting in a succinct way. '
        'Introduce yourself to the user and then wait for their question. '
        'Elaborate your response into a conversational answer in a creative and helpful way. '
        'Your output will be converted to audio so do not include special characters in your answer. '
        'Once the final answer has been provided, please stop, unless the user asks another question. ',
    },
]

# Create context with system message and tools
context = OpenAILLMContext(messages, tools)

# Context aggregator
context_aggregator = llm.create_context_aggregator(context)

# Create context aggregator for message handling
context_aggregator = llm.create_context_aggregator(context)

# Set up pipeline
pipeline = Pipeline(
    [
        transport.input(),
        stt,
        context_aggregator.user(),
        llm,
        tts,
        transport.output(),
        context_aggregator.assistant(),
    ]
)

# Create and configure task
task = PipelineTask(
    pipeline,
    params=PipelineParams(
        allow_interruptions=True,
        enable_metrics=True,
        enable_usage_metrics=True,
    ),
)