실시간 대화

# 의존성: pip install "openai[realtime]"
import asyncio
import base64
from openai import AsyncOpenAI

# 공식 SDK 재사용은 두 곳만 맞추면 됩니다:
#   1) websocket_base_url을 AiHubMix 게이트웨이로 지정(OpenAI 기본 주소가 아님)
#   2) connect()에 model을 명시적으로 전달, 게이트웨이 핸드셰이크에 필수(대화에는 intent 불필요)
client = AsyncOpenAI(
    api_key="sk-***",  # AiHubMix API 키로 교체하십시오
    websocket_base_url="wss://aihubmix.com/v1",
)

async def main():
    async with client.realtime.connect(model="gpt-realtime-2.1") as conn:
        # 1) 세션 구성: 시스템 지시 + 음색 + VAD 켜기(자동 턴 감지)
        await conn.session.update(session={
            "type": "realtime",
            "instructions": "You are a helpful voice assistant. Keep answers concise.",
            "audio": {
                "input": {
                    "format": {"type": "audio/pcm", "rate": 24000},
                    "turn_detection": {"type": "server_vad"},
                },
                "output": {
                    "format": {"type": "audio/pcm", "rate": 24000},
                    "voice": "alloy",
                },
            },
        })

        # 2) 텍스트 메시지를 보내고 답변 요청(오디오 입력은 원시 예제의 append/commit 참조)
        await conn.conversation.item.create(item={
            "type": "message",
            "role": "user",
            "content": [{"type": "input_text", "text": "What is the capital of France?"}],
        })
        await conn.response.create()

        # 3) 이번 턴 답변 수신: 텍스트는 자막 델타, 음성은 오디오 델타
        async for event in conn:
            if event.type == "response.output_audio_transcript.delta":
                print(event.delta, end="", flush=True)  # 답변 텍스트(출력에 음성 포함 시)
            elif event.type == "response.output_audio_transcript.done":
                print()  # 이번 턴 전사 종료
            elif event.type == "response.output_text.delta":
                print(event.delta, end="", flush=True)  # 텍스트만 출력 시
            elif event.type == "response.output_text.done":
                print()
            elif event.type == "response.output_audio.delta":
                pass  # base64 PCM16 오디오 조각, 디코딩해 재생 가능
            elif event.type == "input_audio_buffer.speech_started":
                pass  # 사용자 방해: 재생을 멈추고 conn.conversation.item.truncate(...)로 잘라내기
            elif event.type == "response.done":
                print("\n[턴 완료]", event.response.usage)
                break
            elif event.type == "error":
                print("\n[오류]", event.to_dict())
                break

asyncio.run(main())

import asyncio
import base64
import json
import websockets

API_KEY = "sk-***"  # AiHubMix API 키로 교체하십시오
URL = "wss://aihubmix.com/v1/realtime?model=gpt-realtime-2.1"

async def main():
    # websockets >= 13은 additional_headers, 구버전은 extra_headers
    async with websockets.connect(
        URL, additional_headers={"Authorization": f"Bearer {API_KEY}"}
    ) as ws:
        # 1) 세션 구성(VAD 끄기, 답변 시점을 직접 제어)
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "type": "realtime",
                "instructions": "You are a helpful voice assistant.",
                "audio": {
                    "input": {
                        "format": {"type": "audio/pcm", "rate": 24000},
                        "turn_detection": None,
                    },
                    "output": {
                        "format": {"type": "audio/pcm", "rate": 24000},
                        "voice": "alloy",
                    },
                },
            },
        }))

        # 2) 로컬 PCM16 / 24kHz / 모노 원본 오디오를 읽어 조각으로 보내고 commit + 답변 요청
        async def send_audio():
            with open("audio_pcm16_24k.raw", "rb") as f:
                pcm = f.read()
            chunk = 24000 * 2 // 10  # 100ms = 샘플 레이트 x 2바이트 / 10
            for i in range(0, len(pcm), chunk):
                await ws.send(json.dumps({
                    "type": "input_audio_buffer.append",
                    "audio": base64.b64encode(pcm[i:i + chunk]).decode(),
                }))
                await asyncio.sleep(0.1)  # 실시간 속도 모사
            # VAD를 끈 경우 직접 커밋하고 답변 요청
            await ws.send(json.dumps({"type": "input_audio_buffer.commit"}))
            await ws.send(json.dumps({"type": "response.create"}))

        asyncio.create_task(send_audio())

        # 3) 답변 수신
        async for msg in ws:
            evt = json.loads(msg)
            etype = evt.get("type", "")
            if etype == "response.output_audio_transcript.delta":
                print(evt.get("delta", ""), end="", flush=True)  # 답변 텍스트(출력에 음성 포함 시)
            elif etype == "response.output_audio_transcript.done":
                print()  # 이번 턴 전사 종료
            elif etype == "response.output_text.delta":
                print(evt.get("delta", ""), end="", flush=True)  # 텍스트만 출력 시
            elif etype == "input_audio_buffer.speech_started":
                pass  # 사용자 방해: 재생을 멈추고 conversation.item.truncate 전송
            elif etype == "response.output_audio.delta":
                pass  # base64 PCM16 오디오 조각, 디코딩해 재생 가능
            elif etype == "response.done":
                print("\n[턴 완료]", evt.get("response", {}).get("usage"))
                break
            elif etype == "error":
                print("\n[오류]", evt.get("error"))
                break

asyncio.run(main())

# 의존성: pip install openai-agents
import asyncio
from agents.realtime import (
    OpenAIRealtimeWebSocketModel,
    RealtimeAgent,
    RealtimeSession,
)

agent = RealtimeAgent(
    name="Assistant",
    instructions="You are a helpful voice assistant. Keep answers concise.",
)

async def main():
    session = RealtimeSession(
        OpenAIRealtimeWebSocketModel(),
        agent,
        None,
        model_config={
            "api_key": "sk-***",  # AiHubMix API 키로 교체하십시오
            "url": "wss://aihubmix.com/v1/realtime?model=gpt-realtime-2.1",
        },
        run_config={
            "model_settings": {
                "modalities": ["audio"],
                "output_audio_format": "pcm16",
                # 반드시 꺼야 함: 기본으로 켜지는 입력 오디오 전사는 본 API가 지원하지 않아 그대로 두면 세션이 1008로 닫힙니다
                "input_audio_transcription": None,
            },
        },
    )
    async with session:
        await session.send_message("What is the capital of France?")
        async for event in session:
            if getattr(event, "type", "") == "raw_model_event":
                data = getattr(event, "data", None)
                if getattr(data, "type", "") == "transcript_delta":
                    print(getattr(data, "delta", ""), end="", flush=True)
                elif getattr(data, "type", "") == "turn_ended":
                    print("\n[턴 완료]")
                    return

asyncio.run(main())

# wscat으로 연결성과 인증을 빠르게 검증(먼저 npm i -g wscat 필요)
wscat -c "wss://aihubmix.com/v1/realtime?model=gpt-realtime-2.1" \
  -H "Authorization: Bearer $AIHUBMIX_API_KEY"

# 연결 후 session.update 한 프레임을 붙여넣고 텍스트 메시지 + response.create를 보내면 시작됩니다