使用 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 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="""你是一个有用的三明治店助理。
    你的目标是接受用户的订单。要简洁且友好。
    不要使用表情符号、特殊字符或 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 = 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 typing import AsyncIterator
from cartesia_tts import CartesiaTTS
from events import VoiceAgentEvent

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

    将代理响应转换为流式音频。
    """
    tts = CartesiaTTS()

    async for event in event_stream:
        # 处理代理响应块
        if event.type == "agent_chunk":
            # 将文本发送到 TTS 并流式传输音频
            async for audio_chunk in tts.stream(event.text):
                yield audio_chunk

async def voice_agent_pipeline(audio_stream: AsyncIterator[bytes]) -> AsyncIterator[bytes]:
    """
    完整语音代理管道：

    音频 → STT → 代理 → TTS → 音频
    """
    # 创建事件流
    stt_events = stt_stream(audio_stream)
    agent_events = agent_stream(stt_events)
    audio_output = tts_stream(agent_events)

    # 流式传输最终音频
    async for audio_chunk in audio_output:
        yield audio_chunk

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)

async def voice_agent_pipeline(audio_stream: AsyncIterator[bytes]) -> AsyncIterator[bytes]:
    try:
        async for audio_chunk in pipeline(audio_stream):
            yield audio_chunk
    except Exception as e:
        # 记录错误并向用户返回错误消息
        logger.error(f"语音代理错误：{e}")
        yield generate_error_audio("抱歉，出现了问题。请重试。")