bot.py

"""
AWS Teaching Voice Agent with AWS Bedrock and LangSmith Tracing

This bot is an AI-powered AWS instructor that helps students learn
Amazon Web Services through voice conversations.

Tech stack:
- STT: OpenAI Whisper API
- LLM: AWS Bedrock (Claude 3.5 Haiku)
- TTS: OpenAI TTS
- Transport: Daily WebRTC
- Tracing: LangSmith via OpenTelemetry
"""
import asyncio
import os
import sys
import argparse
import uuid
import aiohttp
from dotenv import load_dotenv
from loguru import logger

load_dotenv(override=True)

# Check if tracing is enabled
IS_TRACING_ENABLED = bool(os.getenv("LANGSMITH_API_KEY"))

# Setup OpenTelemetry tracing for LangSmith BEFORE importing pipecat services
if IS_TRACING_ENABLED:
    from opentelemetry import trace
    from opentelemetry.sdk.trace import TracerProvider
    from opentelemetry.sdk.trace.export import BatchSpanProcessor
    from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
    from pipecat.utils.tracing.setup import setup_tracing
    
    langsmith_api_key = os.getenv("LANGSMITH_API_KEY")
    project_name = os.getenv("LANGSMITH_PROJECT", "aws-teaching-voice-agent")
    
    # Set environment variables for OpenTelemetry (LangSmith recommended approach)
    os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = "https://api.smith.langchain.com/otel"
    os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"x-api-key={langsmith_api_key},Langsmith-Project={project_name}"
    
    # Create OTLP exporter for LangSmith
    otlp_exporter = OTLPSpanExporter(
        endpoint="https://api.smith.langchain.com/otel/v1/traces",
        headers={
            "x-api-key": langsmith_api_key,
            "Langsmith-Project": project_name,
        },
    )
    
    # Set up tracing with the exporter
    setup_tracing(
        service_name=project_name,
        exporter=otlp_exporter,
        console_export=True,  # Enable console output for debugging
    )
    logger.info(f"OpenTelemetry tracing initialized for LangSmith project: {project_name}")

from pipecat.audio.vad.silero import SileroVADAnalyzer, VADParams
from pipecat.frames.frames import TTSSpeakFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.openai.stt import OpenAISTTService
from pipecat.services.openai.tts import OpenAITTSService
from pipecat.services.aws.llm import AWSBedrockLLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport

# Import custom detailed tracing observer
from tracing_observer import create_detailed_tracing_observer

# Configure logging
logger.remove(0)
logger.add(sys.stderr, level="DEBUG")

# AWS Teaching Assistant system prompt
SYSTEM_PROMPT = """You are a professional AWS instructor who helps students learn and effectively use Amazon Web Services through conversational teaching.

Your expertise includes:
- Explaining AWS services such as EC2, S3, Lambda, DynamoDB, Bedrock, SageMaker, and many others
- Guiding cloud architecture design and best practices
- Preparing students for AWS Certifications like Solutions Architect, Developer, and Machine Learning Specialty
- Answering questions about pricing, billing, and cost optimization on AWS
- Providing step-by-step hands-on guidance for specific use cases

Teaching style:
- Keep responses concise and clear since this is a voice conversation
- Use real-world examples to illustrate complex concepts
- Break down technical jargon into simple terms when needed
- Encourage questions and hands-on practice
- If you don't understand a question, politely ask for clarification

You are ready to help students from beginners to experts advance their AWS knowledge."""


async def run_voice_agent(room_url: str, token: str, conversation_id: str):
    """Run the voice agent with OpenTelemetry tracing to LangSmith."""
    
    async with aiohttp.ClientSession() as session:
        logger.info(f"Connecting to Daily room: {room_url}")

        # Set up Daily transport
        transport = DailyTransport(
            room_url,
            token,
            "AWS Instructor",
            DailyParams(
                audio_in_enabled=True,
                audio_out_enabled=True,
                camera_in_enabled=False,
                camera_out_enabled=False,
                vad_enabled=True,
                vad_analyzer=SileroVADAnalyzer(
                    params=VADParams(stop_secs=0.5)
                ),
                transcription_enabled=True,
            ),
        )

        # Initialize OpenAI STT service (Whisper API)
        stt = OpenAISTTService(
            api_key=os.getenv("OPENAI_API_KEY"),
            model="whisper-1",
        )

        # Initialize OpenAI TTS service
        tts = OpenAITTSService(
            api_key=os.getenv("OPENAI_API_KEY"),
            voice="nova",
            model="tts-1",
        )

        # Initialize AWS Bedrock LLM service (Claude 3.5 Haiku)
        llm = AWSBedrockLLMService(
            aws_access_key=os.getenv("AWS_ACCESS_KEY_ID"),
            aws_secret_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
            aws_session_token=os.getenv("AWS_SESSION_TOKEN"),
            aws_region=os.getenv("AWS_REGION", "us-east-1"),
            model="us.anthropic.claude-3-5-haiku-20241022-v1:0",
            params=AWSBedrockLLMService.InputParams(
                temperature=0.7,
            )
        )

        # Set up conversation context
        context = OpenAILLMContext(
            messages=[
                {
                    "role": "system",
                    "content": SYSTEM_PROMPT
                }
            ]
        )
        context_aggregator = llm.create_context_aggregator(context)

        # Build the processing pipeline
        pipeline = Pipeline(
            [
                transport.input(),
                stt,
                context_aggregator.user(),
                llm,
                tts,
                transport.output(),
                context_aggregator.assistant(),
            ]
        )

        # Create detailed tracing observer for input/output capture
        tracing_observer = create_detailed_tracing_observer() if IS_TRACING_ENABLED else None

        # Create task with tracing enabled
        task = PipelineTask(
            pipeline,
            params=PipelineParams(
                allow_interruptions=True,
                enable_metrics=True,           # Required for TTFB and processing metrics
                enable_usage_metrics=True,     # Required for token/character usage
            ),
            enable_tracing=IS_TRACING_ENABLED,      # Enable OpenTelemetry tracing
            enable_turn_tracking=IS_TRACING_ENABLED, # Track conversation turns
            conversation_id=conversation_id,         # Track conversation ID
            additional_span_attributes={
                "service.type": "voice-agent",
                "llm.provider": "aws-bedrock",
                "llm.model": "claude-3-5-haiku",
            },
            observers=[tracing_observer] if tracing_observer else None,
        )

        @transport.event_handler("on_first_participant_joined")
        async def on_first_participant_joined(transport, participant):
            logger.info(f"Participant joined: {participant['id']}")
            await transport.capture_participant_transcription(participant["id"])
            # Use TTSSpeakFrame instead of deprecated tts.say()
            await task.queue_frames([TTSSpeakFrame("Hello! I'm your AWS instructor. What AWS service would you like to learn about today?")])

        @transport.event_handler("on_participant_left")
        async def on_participant_left(transport, participant, reason):
            logger.info(f"Participant left: {participant}, reason: {reason}")
            await task.cancel()

        runner = PipelineRunner(handle_sigint=False)
        await runner.run(task)
        
        return {"conversation_id": conversation_id, "status": "completed"}


async def main(room_url: str, token: str):
    """Main entry point."""
    conversation_id = str(uuid.uuid4())
    logger.info(f"Starting AWS Teaching Voice Agent - Conversation ID: {conversation_id}")
    
    await run_voice_agent(room_url, token, conversation_id)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="AWS Teaching Voice Agent with Bedrock and LangSmith Tracing")
    parser.add_argument("-u", "--url", type=str, required=True, help="Daily room URL")
    parser.add_argument("-t", "--token", type=str, required=True, help="Daily room token")

    config = parser.parse_args()

    asyncio.run(main(config.url, config.token))