repository·main·Indexed 20 days ago
https://github.com/videosdk-live/agentsAn open-source Python framework for building real-time, low-latency voice and multimodal AI agents that participate in VideoSDK video meetings. The framework abstracts audio/video streaming, STT/LLM/TTS orchestration, and turn detection, supporting Cascade, Realtime, and Hybrid pipeline modes. It includes capabilities for custom Agent classes, function tools via @function_tool, and integration with tools like LangGraph and Slack.
The Agent-to-Agent (A2A) protocol allows multiple agents to coordinate and interact within the same room. An orchestrator can manage multiple agent sessions concurrently.
register_a2a(): Registers an agent for discovery by other agents.unregister_a2a(): Removes an agent from discovery.asyncio.create_task to run multiple agent sessions (e.g., a CustomerServiceAgent and a LoanAgent) in the same room.All agents are constructed using a single Pipeline class. Depending on which components (STT, LLM, TTS, VAD, etc.) you provide to the Pipeline constructor, the framework automatically selects one of three execution modes:
STT → LLM → TTS. This is best when you need specific control over each stage (e.g., a specific STT provider or a custom TTS voice).By passing components into the Pipeline, the framework handles the wiring and optimal execution automatically.
# Example of Cascade Mode
async def start_session(context: JobContext):
pipeline = Pipeline(
stt=DeepgramSTT(),
llm=GoogleLLM(),
tts=CartesiaTTS(),
vad=SileroVAD(),
turn_detector=TurnDetector(),
)
session = AgentSession(agent=MyAgent(), pipeline=pipeline)
await session.start(wait_for_participant=True, run_until_shutdown=True)
# Example of Realtime Mode
async def start_session(context: JobContext):
pipeline = Pipeline(
llm=GeminiRealtime(
model="gemini-3.1-flash-live-preview",
config=GeminiLiveConfig(voice="Leda", response_modalities=["AUDIO"]),
)
)
session = AgentSession(agent=MyAgent(), pipeline=pipeline)
await session.start(wait_for_participant=True, run_until_shutdown=True)Follow these patterns for consistent development:
@function_tool decorator on standalone functions or agent methods to make them available to the LLM.EventEmitter. Always listen for the "error" event to handle failures gracefully.async aclose() in your components to ensure proper resource cleanup.RealtimeBaseModel. The Pipeline will automatically wrap these in a RealtimeLLMAdapter.@function_tool
def my_agent_tool(param: str):
"""Docstring for the tool"""
return f"Processed {param}"When building with or extending the VideoSDK AI Agents framework, adhere to these technical conventions:
videosdk.plugins.* namespace.STT, LLM, TTS, VAD, EOU, and RealtimeBaseModel are located in videosdk.agents.@function_tool decorator to define both internal and external tools for the agent.@pipeline.on("event_name") decorator pattern to intercept pipeline stages. Supported event names include "stt", "llm", "tts", and others.The Pipeline class is the central orchestration component of the VideoSDK AI Agents framework. It automatically detects and operates in one of three modes based on the components you provide:
VAD → STT → Turn Detector → LLM → TTS. This mode is ideal for mixing and matching different providers (e.g., using AssemblyAI for STT and Anthropic for LLM).When using the OpenAI plugin, be aware of the following technical behaviors:
For GPT-5 and reasoning models (o-series), the plugin applies specific logic:
developer role instead of the standard system role.max_completion_tokens instead of max_tokens.temperature and top_p parameters (as they may not be supported by reasoning models).wss://api.openai.com/v1/realtime.The VideoSDK AI Agents framework uses a plug-and-play architecture where third-party AI services are integrated via interchangeable plugin packages.
Plugins implement base classes for specific roles in an AI pipeline. You can swap providers (e.g., switching from OpenAI to Anthropic) by simply changing the class you instantiate in your Pipeline configuration; no other code changes are required.
# Cascade Mode Example
from videosdk.agents import Pipeline
from videosdk.plugins.deepgram import DeepgramSTT
from videosdk.plugins.openai import OpenAILLM
from videosdk.plugins.elevenlabs import ElevenLabsTTS
from videosdk.plugins.silero import SileroVAD
from videosdk.plugins.turn_detector import TurnDetector
pipeline = Pipeline(
stt=DeepgramSTT(),
llm=OpenAILLM(),
tts=ElevenLabsTTS(),
vad=SileroVAD(),
turn_detector=TurnDetector()
)The Voice Blog Writer operates using a LangGraph pipeline that transitions through several stages once the user provides the Topic, Audience, and Tone.
The execution flow follows this sequence:
coordinator → planner → write_sections (consisting of 4 sequential Gemini calls) → compiler → synthesizer.
{title-slug}_{datetime}.md.LangGraphLLM(output_node="synthesizer_node"), the agent ensures that only the final spoken line from the synthesizer reaches the Text-to-Speech (TTS) engine, preventing intermediate graph steps from being spoken aloud.# Example of controlling which node's output is sent to TTS
LangGraphLLM(output_node="synthesizer_node")start_of_speech / end_of_speech) that drive the rest of the agent's pipeline (e.g., triggering an LLM response).The pre-built N8N workflow (customer_followup_agent.json) consists of three core functional nodes:
call-answered or call-hangup) from VideoSDK to allow the workflow to react when a call concludes.