ThunderPhone 2.0 正式上線。全程自助，每分鐘 2 美分起。查看公告
立即開始

Connect tools & data

從 Pipecat 使用 ThunderPhone

在 Pipecat 管線中將 ThunderPhone 語音智慧體作為語音對語音服務執行，並使用你自己的傳輸與電話服務。

Pipecat 是一個開放原始碼框架，可透過可組合的服務建置語音智慧體。ThunderPhone 可作為 語音對語音 LLM 服務整合：Pipecat 傳送來電者音訊，ThunderPhone 回傳智慧體的語音、逐字稿與函式呼叫，而 Pipecat 的傳輸層（Daily、LiveKit、Twilio、WebRTC、本機麥克風）負責傳遞音訊。語音辨識、語言模型、語音、輪流對話、47 種語言與工具皆在 ThunderPhone 上執行。

以此方式撥打的通話會顯示於通話紀錄，並會和其他即時通話一樣，依你的產品每分鐘費率計費。不需要訂閱，也不涉及 ThunderPhone 電話號碼。

安裝

pip install pipecat-thunderphone
export THUNDERPHONE_API_KEY=sk_live_...   # a secret API key

此套件封裝 Pipecat 的 OpenAI Realtime 服務，因為 ThunderPhone 的 Realtime WebSocket 採用相同通訊協定。它需要 pipecat-ai 1.8 或更新版本。

執行已儲存的智慧體

智慧體執行的所有內容（提示詞、語音、引擎、語言、工具、問候語、靜默確認）皆在 ThunderPhone 中設定。管線僅負責傳遞音訊。

from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.worker import PipelineParams, PipelineWorker
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat_thunderphone import ThunderPhoneRealtimeLLMService
 
llm = ThunderPhoneRealtimeLLMService(agent_id=12)
 
context = LLMContext()
aggregators = LLMContextAggregatorPair(context)
pipeline = Pipeline([
    transport.input(),
    aggregators.user(),
    llm,
    aggregators.assistant(),
    transport.output(),
])
worker = PipelineWorker(pipeline, params=PipelineParams(allow_interruptions=True))

已儲存的智慧體會依自身排程開啟通話，因此服務不會向 Pipecat 請求開場回應。若仍希望這麼做，請傳入 greet_on_connect=True。

直接設定工作階段

不使用 agent_id 時，指示與工具會和 OpenAI 一樣來自 Pipecat 內容。product 用於選擇引擎，voice 用於選擇 ThunderPhone 語音。

from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.services.llm_service import FunctionCallParams
 
llm = ThunderPhoneRealtimeLLMService(product="bolt", voice="olivia", language="es")
 
async def check_availability(params: FunctionCallParams):
    slots = await calendar.free_slots(params.arguments["date"])
    await params.result_callback({"slots": slots})
 
llm.register_function("check_availability", check_availability)
 
context = LLMContext(
    messages=[{"role": "system", "content": "You are Acme Dental's receptionist."}],
    tools=ToolsSchema(standard_tools=[
        FunctionSchema(
            name="check_availability",
            description="Free appointment slots on a date",
            properties={"date": {"type": "string"}},
            required=["date"],
        )
    ]),
)

直接設定的工作階段由用戶端主導：由於 Pipecat 在收到內容時會請求回應，智慧體會先開口；而在靜默期間，除非你附加訊息或請求另一個回應，否則會保持安靜。

選項

引數說明
api_key密鑰（sk_live_...）。預設為 THUNDERPHONE_API_KEY。
agent_id執行已儲存的智慧體。不可與 product 和 voice 同時使用。
product內嵌工作階段使用的引擎：spark、bolt 或 storm。
voice內嵌工作階段使用的 ThunderPhone 語音名稱。
language內嵌工作階段的主要語言提示，例如 es。
from_number、to_number當管線前端接入電話線路時，要記錄於通話中的號碼。
live_transcripts在來電者話語尚未結束時串流逐字稿片段（需額外計費）。
greet_on_connect工作階段準備就緒後立即要求第一則回應。
end_task_on_call_endedThunderPhone 結束通話時推送 EndWorkerFrame（預設啟用）。

事件

ThunderPhone 在即時通訊協定之上新增平台事件。服務會將這些事件傳送至處理常式：

@llm.event_handler("on_call_ended")
async def on_call_ended(service, event):
    print("call ended:", event["reason"], "call id:", service.call_id)

工作階段啟用後，系統便會設定 service.call_id。通話結束後，使用它搭配 GET /v1/calls/{call_id} 取得錄音、逐字稿和評分。

限制

  • 回合偵測在伺服器端執行且一律啟用；不支援由 Pipecat 驅動的回合 （turn_detection=False）。
  • 註冊於服務上的函式僅會在內嵌工作階段中執行。已儲存的智慧體會在 ThunderPhone 上執行自己的工具。
  • 音訊在雙向皆為 24 kHz、16 位元單聲道 PCM。
  • 會忽略視訊影格。