使用 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:
        # 接收并生成转录事件
        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"Added {quantity} x {item} to the order."

def confirm_order(order_summary: str) -> str:
    """与客户确认最终订单。"""
    return f"Order confirmed: {order_summary}. Sending to kitchen."

# 使用工具和记忆创建代理
agent = create_agent(
    model="anthropic:claude-haiku-4-5",  # 选择你的模型
    tools=[add_to_order, confirm_order],
    system_prompt="""You are a helpful sandwich shop assistant.
    Your goal is to take the user's order. Be concise and friendly.
    Do NOT use emojis, special characters, or markdown.
    Your responses will be read by a text-to-speech engine.""",
    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 = agent.astream(
                {"messages": [HumanMessage(content=event.transcript)]},
                {"configurable": {"thread_id": thread_id}},
                stream_mode="messages",
            )

            # 在到达时生成代理响应块
            async for message, _ in stream:
                if message.text:
                    yield AgentChunkEvent.create(message.text)

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(): 从 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 生成音频字节。"""
        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)