使用 LangChain 构建语音代理

from typing import AsyncIterator
import asyncio
from assemblyai_stt import AssemblyAISTT
from events import VoiceAgentEvent

async def stt_stream(
    audio_stream: AsyncIterator[bytes],
) -> AsyncIterator[VoiceAgentEvent]:
    """
    转换流：音频（字节） → 语音事件（VoiceAgentEvent）

    使用生产者-消费者模式：
    - 生产者：读取音频块并发送到 AssemblyAI
    - 消费者：接收来自 AssemblyAI 的转录事件
    """
    stt = AssemblyAISTT(sample_rate=16000)

    async def send_audio():
        """后台任务，将音频块泵送到 AssemblyAI。"""
        try:
            async for audio_chunk in audio_stream:
                await stt.send_audio(audio_chunk)
        finally:
            # 当音频流结束时发出完成信号
            await stt.close()

    # 在后台启动音频发送
    send_task = asyncio.create_task(send_audio())

    try:
        # 接收并 yield 到达的转录事件
        async for event in stt.receive_events():
            yield event
    finally:
        # 清理
        with contextlib.suppress(asyncio.CancelledError):
            send_task.cancel()
            await send_task
        await stt.close()

from langchain_core.utils.uuid import uuid7
from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver

# 定义代理工具
def add_to_order(item: str, quantity: int) -> str:
    """向客户的三明治订单中添加一项物品。"""
    return f"已将 {quantity} 份 {item} 添加到订单中。"

def confirm_order(order_summary: str) -> str:
    """与客户确认最终订单。"""
    return f"订单已确认：{order_summary}。正在发送至厨房。"

# 创建带工具和记忆的代理
agent = create_agent(
    model="google_genai:gemini-3.6-flash",  # 选择你的模型
    tools=[add_to_order, confirm_order],
    system_prompt="""你是一家乐于助人的三明治店助手。
你的目标是接下用户的订单。请保持简洁和友好。
不要使用表情符号、特殊字符或 Markdown。
你的回复将由文本转语音引擎朗读。""",
    checkpointer=InMemorySaver(),
)

async def agent_stream(
    event_stream: AsyncIterator[VoiceAgentEvent],
) -> AsyncIterator[VoiceAgentEvent]:
    """
    转换流：语音事件 → 语音事件（包含代理响应）

    传递所有上游事件，并在处理 STT 转录时添加 agent_chunk 事件。
    """
    # 为对话记忆生成唯一的线程 ID
    thread_id = str(uuid7())

    async for event in event_stream:
        # 传递所有上游事件
        yield event

        # 通过代理处理最终转录
        if event.type == "stt_output":
            # 流式传输带有对话上下文的代理响应
            stream = await agent.astream_events(
                {"messages": [HumanMessage(content=event.transcript)]},
                {"configurable": {"thread_id": thread_id}},
                version="v3",
            )

            # 当代理响应块到达时 yield
            async for message in stream.messages:
                async for token in message.text:
                    yield AgentChunkEvent.create(token)

from cartesia_tts import CartesiaTTS
from utils import merge_async_iters

async def tts_stream(
    event_stream: AsyncIterator[VoiceAgentEvent],
) -> AsyncIterator[VoiceAgentEvent]:
    """
    转换流：语音事件 → 语音事件（包含音频）

    合并两个并发流：
    1. process_upstream()：传递事件并将文本发送到 Cartesia
    2. tts.receive_events()：yield 来自 Cartesia 的音频块
    """
    tts = CartesiaTTS()

    async def process_upstream() -> AsyncIterator[VoiceAgentEvent]:
        """处理上游事件并将代理文本发送到 Cartesia。"""
        async for event in event_stream:
            # 传递所有事件
            yield event
            # 将代理文本发送到 Cartesia 进行合成
            if event.type == "agent_chunk":
                await tts.send_text(event.text)

    try:
        # 合并上游事件和 TTS 音频事件
        # 两个流并发运行
        async for event in merge_async_iters(
            process_upstream(),
            tts.receive_events()
        ):
            yield event
    finally:
        await tts.close()

export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="..."

import getpass
import os

os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = getpass.getpass()

from langchain_core.runnables import RunnableGenerator

pipeline = (
    RunnableGenerator(stt_stream)      # 音频 → STT 事件
    | RunnableGenerator(agent_stream)  # STT 事件 → 代理事件
    | RunnableGenerator(tts_stream)    # 代理事件 → TTS 音频
)

# 在 WebSocket 端点中使用
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()

    async def websocket_audio_stream():
        """从 WebSocket yield 音频字节。"""
        while True:
            data = await websocket.receive_bytes()
            yield data

    # 通过管道转换音频
    output_stream = pipeline.atransform(websocket_audio_stream())

    # 将 TTS 音频发送回客户端
    async for event in output_stream:
        if event.type == "tts_chunk":
            await websocket.send_bytes(event.audio)