实用指南：开源智能体框架与 ElevenAgents

在我们之前关于集成外部智能体与 ElevenLabs 语音编排的文章中，我们介绍了团队如何通过自定义 LLM将现有文本智能体编排接入 ElevenLabs。在此基础上，本指南演示了主流开源智能体框架如何适配并部署到 Custom LLM 接口后端。这样可以在不影响状态管理、工具编排或应用控制的前提下，将语音能力叠加到成熟的智能体系统上。无论使用哪种框架，整体流程都分为三步：创建生成请求、提取最终文本响应，并将其转换为 OpenAI 兼容的 Server-Sent Events（SSE）格式。ElevenLabs 支持Chat Completions和回复两种格式。本指南涵盖了四个主流框架，但这些方法适用于任何能输出 OpenAI 兼容流式结果的运行环境。

通用设置

本节示例使用 Python 和 FastAPI，任何能处理 HTTP POST 请求和 SSE 流式响应的技术栈都适用。当 ElevenLabs 语音编排检测到可能的轮次结束时，会向配置的 Custom LLM 端点发起生成请求。本节将介绍该转换层的核心组件，即让语音编排与智能体框架“对话”的桥接或代理。

每个框架的选择，通常是因为用户熟悉或能满足特定需求。例如 LlamaIndex 最初为简化 RAG（检索增强生成）而开发，CrewAI 则专注于自动化任务。不同设计目标导致响应结构不同，需要分别处理。将 LLM 生成的内容以流式分块推送，而不是等待完整轮次，有助于 TTS（文本转语音）模型更早开始生成语音，从而降低延迟。我们重点介绍四个流行框架：LangGraph、Google ADK、CrewAI 和 LlamaIndex。

关于通用代码的说明

每个框架都需以 OpenAI 兼容的 SSE 分块流式返回响应。这里提供一个在各示例中通用的小工具函数，用于构建这些分块。

def sse_chunk(response_id: str, delta: dict, finish_reason=None) -> str:

    payload = {

有了这个基础，接下来介绍 LangGraph。

LangGraph

LangGraph 将智能体建模为图结构，节点代表每一步，边定义各步骤间的控制流。最基本的流程很简单：初始化聊天模型，定义智能体工具，创建智能体图运行时。

    }

    return f"data: {json.dumps(payload)}\n\n"

每次生成请求时，LangGraph Agent 会接收完整对话历史，从而在内部维护所需状态。LangGraph 支持通过

状态管理完成后，下一个 LangGraph 特有的决策点是流式模式。LangGraph 提供两种流式选项，适用于不同场景：

LangGraph 将智能体建模为图结构，节点代表单步操作，边定义控制流。最小化配置很简单：初始化聊天模型、定义工具、创建智能体图运行时。

from langchain.agents import create_agent

from langchain_openai import ChatOpenAI

from langchain_core.tools import tool

llm = ChatOpenAI(

    model=model_id,

    api_key=os.getenv("OPENAI_API_KEY"),

)

agent = create_agent(

    llm,
    tools=tool_list,
    system_prompt=system_prompt,
)

每次生成请求，LangGraph Agent 都会收到完整对话历史，从而在内部维护所需状态。LangGraph 支持通过检查点进行服务端持久化，但本例为简化实现未涉及。

状态管理完成后，LangGraph 的下一个关键点是流式模式，提供两种选择，适用于不同场景：

• stream_mode="values" 返回图状态快照，易于实现，但每次响应都包含完整消息状态，实时对话时延更高。
• 额外参数

准备好消息和会话后，即可调用 runner。工具调用和结果会作为 ADK 内部事件出现，但仅作为编排步骤，不会直接输出给用户。相比工具调用会作为文本输出的框架，这样无需手动过滤。

下面的 handler 是包含会话查找和创建逻辑的简化实现。

[2] 工具执行并返回数据（result="$24.99"）

[3] 模型用结果生成响应（content="It costs $24.99"）

接下来介绍 CrewAI，其设计更侧重任务。

CrewAI

CrewAI 旨在围绕结构化任务（调研、写作、总结）编排多智能体 workflow，而非开放式对话循环。每个智能体有角色、目标和背景设定。执行以 Task 对象为核心，每个任务都有明确描述和预期输出。

    input = {"messages": req.messages}

    async def stream():

与 LangGraph 和 ADK 的智能体循环不同，CrewAI 通常每次请求都构建 Task 和 Crew，定义本轮对话的工作单元。通过占位符注入历史轮次，将对话上下文传递到下一个任务。每次请求会用 {crew_chat_messages} 变量填充当前对话历史，并在执行时插入到任务描述中。为生成简洁、适合语音播报的文本，还会显式过滤中间追踪内容（Thought、Action、Action Input、Observation），只输出最终答案。

下面的 handler 集成了按请求构建任务、历史插入、Crew 级流式输出、追踪过滤和结果格式化。

        async for message_chunk, metadata in agent.astream(input, stream_mode="messages"):

            # 只转发模型文本分块，跳过工具更新和非文本事件。

接下来介绍 LlamaIndex，其采用原生事件驱动流式模型。

LlamaIndex

与前述框架不同，LlamaIndex 主要用于将 LLM 连接到外部数据源（文档库、索引、检索流程）。其智能体层 FunctionAgent 基于此基础，专注于结构化上下文的检索和推理，而非开放对话或任务执行。

            if not content:

                continue

为保持对话连贯性，代理会将输入消息转为 LlamaIndex 聊天消息，并拆分为最新用户消息（user_msg）和历史消息（chat_history）。每个 AgentStream 事件的 event.delta 字段包含下一个文本片段，可直接映射为 OpenAI 风格的 delta.content 块。非空 delta 可直接转发，是本指南中最直接的流式桥接方式。流中既有编排事件（工具调用、结果），也有语音事件（助手文本 delta）。为保证语音输出简洁，代理只保留 AgentStream 事件并跳过空 delta。

为保证对话连贯性，代理会将收到的消息转换为 LlamaIndex 聊天消息，然后拆分为最新用户发言（user_msg）和历史对话（chat_history）。每个 AgentStream 事件的 event.delta 字段包含下一个文本片段，可直接映射为 OpenAI 风格的 delta.content 块。非空 delta 可直接转发，因此这是本指南中最直接的流式桥接方式。流中包含编排事件（工具调用、结果）和语音事件（助手文本 delta）。为保证语音输出简洁，代理只保留 AgentStream 事件并跳过空 delta。

[1] AgentStream（delta=''）       ← 忽略

这种分离方式能让中间工具逻辑不出现在语音输出中，同时保证低延迟的增量语音。下面的处理器整合了这些步骤。

            yield sse_chunk(response_id, {"content": content})

LlamaIndex 对端到端对话运行模式的约束比内置编排层更重的框架要少。实际生产部署时，通常需要用户自行实现会话管理、回复规范、工具编排和追踪。

相比内置编排更重的框架，LlamaIndex 对端到端对话运行模式要求更灵活。实际部署时，通常需要用户自行实现会话管理、回复约束、工具编排和追踪。

总结

本指南中的每个框架都通过相同方式接入 ElevenLabs：接受 OpenAI 风格的 Completions 或 Responses 请求，并以 SSE 流式返回数据块。这样，团队只需少量改动即可在现有智能体实现上叠加语音编排，既保留已有成果，又能实现实时

如果你已经在用开源框架运行智能体，并希望支持语音，可以试试这种方式，欢迎反馈体验。

接下来介绍 Google Agent Development Kit（ADK）的细节。

Google ADK

Google 的 ADK 用少量核心组件（Agent、Runner、SessionService）抽象了运行循环。ADK 的 Runner 介于 HTTP 层和智能体定义之间，负责消息路由、工具编排、会话生命周期和事件流。

from google.adk.agents import Agent

from google.adk.runners import Runner

from google.adk.agents.run_config import RunConfig, StreamingMode

from google.adk.sessions import InMemorySessionService

from google.genai import types as genai_types

agent = Agent(

    name=name,

    model=model,

    instruction=instruction,

    tools=[tool_list],

)

session_service = InMemorySessionService()

runner = Runner(

agent=agent,

app_name=app_name,

session_service=session_service

)

初始化 agent、会话后端和 runner 后，代理会为每个请求查找或创建 ADK 会话。在 ADK 中，session_id 控制记忆持久化：多轮复用同一 session_id 可自动保留历史、工具调用和响应。由于对话标识在 ElevenLabs 上游，代理需显式处理映射。通过为生成请求传递正确标识，SDK 可内部处理上下文。会话初始化时通过额外参数传递到请求体。

准备好消息和会话后，即可调用 runner。工具调用和结果会作为 ADK 内部事件出现，但仅作为编排步骤，不会作为用户可见输出。相比工具调用以文本形式出现的框架，这里无需手动过滤。

下面的处理函数为简化实现，包含会话查找和创建逻辑。

@app.post("/chat/completions")

async def chat_completions(req: ChatCompletionRequest, request: Request):

    # 生产环境建议用上游系统的稳定标识。

    session_id = req.elevenlabs_extra_body.arbitrary_identifier

    session = await session_service.get_session(

        app_name="elevenlabs", user_id="user", session_id=session_id

    )

    if not session:

        session = await session_service.create_session(

            app_name="elevenlabs", user_id="user", session_id=session_id

        )

    user_text = next((m["content"] for m in reversed(req.messages) if m["role"] == "user"), "")

    content = genai_types.Content(role="user", parts=[genai_types.Part(text=user_text)])

   async def stream():

        response_id = f"chatcmpl-{uuid.uuid4().hex[:12]}"

        sent_role = False

        async for event in runner.run_async(

            user_id="user",

            session_id=session.id,

            new_message=content,

            run_config=RunConfig(streaming_mode=StreamingMode.SSE),

        ):

            if not event.content or not event.content.parts:

                continue

            # SSE 模式下，ADK 会发出部分（增量）和最终（完整）事件。

            # 只转发部分事件可避免重复完整文本。
            # 注意：ADK 的 SSE 流式为实验特性，生产环境需兼容

            # 两种事件类型，以防模型后端不发部分事件。

            if not getattr(event, "partial", False):

                continue

            text = "".join((getattr(p, "text", "") or "") for p in event.content.parts)

            if not text:

                continue

            if not sent_role:

                yield sse_chunk(response_id, {"role": "assistant"})

                sent_role = True

            yield sse_chunk(response_id, {"content": text})

        yield sse_chunk(response_id, {}, finish_reason="stop")

        yield "data: [DONE]\n\n"

    return StreamingResponse(stream(), media_type="text/event-stream")

接下来介绍更偏向任务驱动的 CrewAI。

CrewAI

CrewAI 设计用于围绕结构化任务（如调研、写作、总结）编排多智能体流程，而非开放式对话循环。每个智能体有角色、目标和背景设定。执行以 Task 对象为中心，每个任务有明确描述和预期输出。

from crewai import Agent, Task, Crew, Process, LLM

from crewai.tools import tool

from crewai.types.streaming import StreamChunkType

llm = LLM(

    model=model_id,

    api_key=os.getenv("OPENAI_API_KEY")

)

store_agent = Agent(

    role=role,

    goal=goal,

    backstory=backstory,

    tools=tools,

    llm=llm,

    verbose=False,

)

与 LangGraph 和 ADK 的智能体循环不同，CrewAI 通常每次请求都新建 Task 和 Crew，定义本轮对话的工作单元。通过占位符注入历史轮次，将上下文传递到下一个任务。{crew_chat_messages} 变量在每次请求时填充当前对话历史，并在执行时插入到任务描述中。为保证输出适合语音播放，会显式过滤中间追踪内容（Thought、Action、Action Input、Observation），只输出最终答案文本。

下面的处理函数整合了按请求构建任务、历史插入、Crew 级流式输出、追踪过滤和输出格式化。

@app.post("/chat/completions")

async def chat_completions(req: ChatCompletionRequest):

    # 每次请求动态组装 Task 和 Crew（非启动时）。

    task = Task(

        description=(

            "Conversation history:\n{crew_chat_messages}\n\n"

            "Respond to the user's latest message."

        ),

        expected_output=expected_output,

        agent=store_agent,

    )

    # stream=True 返回 CrewStreamingOutput 而非单一 CrewOutput。

    crew = Crew(

        agents=[store_agent],

        tasks=[task],

        process=Process.sequential,

        verbose=False,

        stream=True,

    )

    async def stream():

        response_id = f"chatcmpl-{uuid.uuid4().hex[:12]}"

        sent_role = False

        final_marker = "final answer:"

        marker_buffer = ""

        marker_found = False

        emitted_any_content = False

        streaming = await crew.kickoff_async(

            inputs={"crew_chat_messages": json.dumps(req.messages)}

        )

       async for chunk in streaming:

            # 跳过非文本事件（如工具调用）。

            if chunk.chunk_type != StreamChunkType.TEXT or not chunk.content:

                continue

            # 只在出现 "Final Answer:" 标记后转发文本

            if not marker_found:

                marker_buffer += chunk.content

                idx = marker_buffer.lower().find("final answer:")

                if idx == -1:

                    continue

                marker_found = True

                content = marker_buffer[idx + 13:].lstrip()

                marker_buffer = ""

            else:

                content = chunk.content

            # 清理 CrewAI 输出中多余的 markdown 符号。

            content = content.rstrip("`").rstrip()

            if not content:

                continue

            if not sent_role:

                yield sse_chunk(response_id, {"role": "assistant"})

                sent_role = True

            yield sse_chunk(response_id, {"content": content})

        # 若无 "Final Answer:" 标记，兜底处理短回复

        if not sent_role:

            raw = getattr(streaming, "result", None)

            fallback = (raw.raw if raw else marker_buffer).strip().rstrip("`").rstrip()

            if fallback:

                yield sse_chunk(response_id, {"role": "assistant"})

                yield sse_chunk(response_id, {"content": fallback})

        yield sse_chunk(response_id, {}, finish_reason="stop")

        yield "data: [DONE]\n\n"

接下来介绍 LlamaIndex，其核心是原生事件驱动流式模型。

LlamaIndex

与前述框架不同，LlamaIndex 主要用于将 LLM 连接到外部数据源（文档库、索引、检索管道）。其智能体层 FunctionAgent 基于此，专注于结构化上下文的检索和推理，而非开放对话或任务执行。

from llama_index.llms.openai import OpenAI

from llama_index.core.agent.workflow import FunctionAgent, AgentStream

from llama_index.core.base.llms.types import ChatMessage, MessageRole

llm = OpenAI(

    model=model,

    api_key=os.getenv("OPENAI_API_KEY")

)

agent = FunctionAgent(

    tools=[list_inventory, get_item_price],

    llm=llm,

    system_prompt=system_prompt,

)

为保证对话连贯性，代理会将输入消息转换为 LlamaIndex 聊天消息，并拆分为最新用户轮次（user_msg）和历史轮次（chat_history）。每个 AgentStream 事件的 event.delta 字段包含下一个文本片段，可直接映射为 OpenAI 风格的 delta.content 分块。只需转发非空 delta，即可实现最简流式桥接。流中既有编排事件（工具调用、结果），也有语音事件（助手文本分块）。为保证语音输出简洁，代理只保留 AgentStream 事件，跳过空 delta。

[1] AgentStream（delta=''）       ← 忽略

[2] ToolCall                     ← 忽略

[3] ToolCallResult               ← 忽略

[4] AgentStream（delta='It'）     ← 转发 ✓

[5] AgentStream（delta=' costs'） ← 转发 ✓

[6] AgentStream（delta=' $49.99'）← 转发 ✓

这种分离方式可避免工具机制内容进入语音输出，同时保证低延迟增量语音。下面是整合上述步骤的处理函数。

@app.post("/chat/completions")

async def chat_completions(req: ChatCompletionRequest):

    # 假设最后一条消息总是用户文本。

    # 生产环境需防御性处理非文本消息。

    chat_history = [

        ChatMessage(role=MessageRole(m["role"]), content=m.get("content") or "")

        for m in req.messages

    ]

    user_text = chat_history.pop().content

    async def stream():

        response_id = f"chatcmpl-{uuid.uuid4().hex[:12]}"

        handler = agent.run(user_msg=user_text, chat_history=chat_history)

        async for event in handler.stream_events():

            if not isinstance(event, AgentStream):

                continue

            if not event.delta:

                continue

            yield sse_chunk(response_id, {"content": event.delta})

        yield sse_chunk(response_id, {}, finish_reason="stop")

        yield "data: [DONE]\n\n"

    return StreamingResponse(stream(), media_type="text/event-stream")

LlamaIndex 对端到端对话运行模式要求较少，未像其他框架那样内置编排。生产环境下，通常需用户自行实现会话管理、响应约束、工具编排和追踪。

总结

本指南中的每个框架都通过同一协议接入 ElevenLabs：接收 OpenAI 风格的 Completions 或 Responses 请求，并以 SSE 分块流式返回。这让团队可以在现有智能体实现基础上，轻松叠加语音编排，无需大幅改动，既保留已有成果，又能实现实时对话式 AI。这种模块化是 ElevenAgents 平台的核心理念。无论是扩展现有智能体，还是从零构建语音原生应用，ElevenAgent 的语音编排都能灵活适配。

如果你已经在用开源框架运行智能体，并希望接入语音，不妨试试这种方式，欢迎反馈体验。