Intelligent Virtual Assistants: Translation with AI Agents

1class AIAgent:
2    def __init__(self, meeting_id: str, authToken: str, name: str):
3        # Initialize the AI agent with audio processing capabilities
4        self.audio_track = CustomAudioStreamTrack(
5            loop=self.loop,
6            handle_interruption=True
7        )
8        # Connect to OpenAI for intelligence
9        self.intelligence = OpenAIIntelligence(
10            loop=self.loop,
11            api_key=api_key,
12            base_url="api.openai.com",
13            input_audio_transcription=InputAudioTranscription(model="whisper-1"),
14            audio_track=self.audio_track
15        )
16

1# Extract the info for each participant
2participant_ids = list(self.participants_data.keys())
3p1 = self.participants_data[participant_ids[0]]
4p2 = self.participants_data[participant_ids[1]]
5
6# Build translator-specific instructions
7translator_instructions = f"""
8    You are a real-time translator bridging a conversation between:
9    - {p1['name']} (speaks {p1['lang']})
10    - {p2['name']} (speaks {p2['lang']})
11
12    You have to listen and speak those exactly word in different language
13    eg. when {p1['lang']} is spoken then say that exact in language {p2['lang']}
14    similar when {p2['lang']} is spoken then say that exact in language {p1['lang']}
15    Keep in account who speaks what and use 
16    NOTE - 
17    Your job is to translate, from one language to another, don't engage in any conversation
18"""
19

1async def add_audio_listener(self, stream: Stream):
2    while True:
3        try:
4            # Continuously process audio frames in real-time
5            frame = await stream.track.recv()      
6            audio_data = frame.to_ndarray()[0]
7            # Process and send to OpenAI for translation
8            await self.intelligence.send_audio_data(pcm_frame)
9        except Exception as e:
10            print("Audio processing error:", e)
11            break
12

1class CustomAudioStreamTrack(CustomAudioTrack):
2    def __init__(self, loop, handle_interruption: Optional[bool] = True):
3        super().__init__()
4        self.loop = loop
5        self._start = None
6        self._timestamp = 0
7        self.frame_buffer = []
8        # Audio configuration
9        self.sample_rate = 24000
10        self.channels = 1
11        self.sample_width = 2
12        # More audio processing setup...
13

1def on_meeting_joined(self, data):
2    print("Meeting Joined - Starting OpenAI connection")
3    asyncio.create_task(self.intelligence.connect())
4

1async def connect(self):
2    # Connect to OpenAI's real-time API
3    url = f"wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview"
4    self.ws = await self._http_session.ws_connect(
5        url=url,
6        headers={
7            "Authorization": f"Bearer {self.api_key}",
8            "OpenAI-Beta": "realtime=v1",
9        },
10    )
11

1def on_participant_joined(self, participant: Participant):
2    peer_name = participant.display_name
3    native_lang = participant.meta_data["preferredLanguage"]
4    self.participants_data[participant.id] = {
5        "name": peer_name,
6        "lang": native_lang
7    }
8

1# Dynamically tell OpenAI to use these instructions
2asyncio.create_task(self.intelligence.update_session_instructions(translator_instructions))
3