Guide pratique : frameworks d'agents open source et ElevenAgents

Dans notre précédent article sur l’intégration d’agents externes à l’orchestration vocale d’ElevenLabs, nous avons expliqué comment les équipes peuvent connecter leur orchestration d’agents textuels existante à ElevenLabs via le Custom LLM. Dans la continuité, ce guide montre comment adapter et déployer les principaux frameworks d’agents open source derrière l’interface Custom LLM. Il en résulte une architecture flexible qui ajoute la voix à des systèmes d’agents matures sans compromettre la gestion de l’état, l’orchestration des outils ni le contrôle propre à l’application. Quel que soit le framework, nous suivons le même schéma en trois étapes : créer une requête de génération, extraire la réponse textuelle finale et la reformater au format Server-Sent Events (SSE) compatible avec OpenAI. ElevenLabs prend en charge les formats Chat Completions et Responses. Bien que ce guide couvre quatre frameworks largement adoptés, ces schémas s’appliquent à tout environnement d’exécution capable de produire une sortie de streaming compatible avec OpenAI.

Configuration générale

Les exemples de cette section utilisent Python et FastAPI, mais toute stack capable de traiter des requêtes HTTP POST et de diffuser des réponses SSE en streaming convient. Lorsque l’orchestration vocale d’ElevenLabs détecte une fin de tour probable, elle envoie une requête de génération au point de terminaison Custom LLM configuré. Cette section présente les composants essentiels de cette couche de traduction : le pont ou proxy qui permet à l’orchestration vocale et au framework d’agents de parler le même langage.

Naturellement, les clients peuvent choisir chaque framework par familiarité ou pour répondre à un besoin précis. LlamaIndex, par exemple, a été développé à l’origine pour simplifier la mise en place de la génération augmentée par récupération (RAG), tandis que CrewAI a été conçu pour automatiser des tâches définies à l’ère des agents. Des objectifs de conception différents produisent des structures de réponse différentes, qui nécessitent chacune un traitement spécifique. Diffuser les fragments à mesure que le LLM les génère, plutôt que d’attendre la fin du tour, est essentiel : le modèle Text-to-Speech (TTS) peut ainsi commencer à générer la parole plus tôt, ce qui réduit la latence perçue. Nous nous concentrons sur quatre frameworks populaires : LangGraph, Google ADK, CrewAI et LlamaIndex.

À propos du code partagé

Chaque framework doit diffuser les réponses sous forme de fragments SSE compatibles avec OpenAI. Nous introduisons une petite fonction d’assistance utilisée dans tous les exemples pour construire ces fragments.

def sse_chunk(response_id: str, delta: dict, finish_reason=None) -> str:
    payload = {
        "id": response_id,
        "object": "chat.completion.chunk",
        "choices": [{"index": 0, "delta": delta, "finish_reason": finish_reason}],
    }
    return f"data: {json.dumps(payload)}\n\n"

Ces bases posées, commençons par LangGraph. 

LangGraph

LangGraph modélise les agents sous forme de graphes, où les nœuds représentent les étapes individuelles et les arêtes définissent le flux de contrôle entre elles. La configuration minimale est simple : initialiser un modèle de chat, définir les outils de l’agent et créer l’environnement d’exécution du graphe d’agents.

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,
)

Pour chaque requête de génération, l’agent LangGraph reçoit l’historique complet de la conversation, ce qui lui permet de conserver l’état nécessaire en interne. LangGraph prend en charge la persistance côté serveur via les Checkpoints, que nous n’abordons toutefois pas ici afin de limiter l’implémentation au minimum.

Une fois la gestion de l’état assurée, le prochain choix propre à LangGraph concerne le mode de streaming. LangGraph propose deux options, chacune adaptée à un cas d’usage distinct :

• stream_mode="values" fournit des instantanés de l’état du graphe. Plus simple à implémenter, ce mode inclut toutefois un état de message plus complet dans chaque réponse, ce qui augmente la latence des flux conversationnels en temps réel.
• stream_mode="messages" diffuse des fragments de messages incrémentiels depuis le modèle. Ce mode est généralement préférable pour les interactions vocales en temps réel, car il réduit le délai avant le premier audio dans la couche d’orchestration d’ElevenLabs.

Plus précisément, l’implémentation en mode messages de la boucle d’agent comprend des étapes intermédiaires, telles que les mises à jour d’appel d’outil, qui ne doivent pas être prononcées. Le proxy les filtre et ne transmet que le texte de réponse destiné à l’utilisateur à la couche TTS. Voici un exemple de tour utilisant un outil.

[1] Le modèle décide d’appeler un outil (tool_calls=["get_price"])
[2] L’outil s’exécute et renvoie des données (result="$24.99") 
[3] Le modèle produit une réponse à partir du résultat (content="Cela coûte $24.99") 

Naturellement, seuls les fragments de l’étape 3 doivent être transmis dans le flux SSE. En pratique, deux vérifications conditionnelles assurent ce filtrage dans la boucle de streaming : l’une conserve uniquement les événements langgraph_node == "model", l’autre ignore les contenus vides. Ensemble, elles garantissent que seul le texte de l’assistant destiné à l’utilisateur est transmis à ElevenLabs au format SSE. Ces principes réunis, voici une implémentation légère du proxy de requêtes.

@app.post("/chat/completions")
async def chat_completions(req: ChatCompletionRequest):
    input = {"messages": req.messages}
    async def stream():
        response_id = f"chatcmpl-{uuid.uuid4().hex[:12]}"
        sent_role = False
        async for message_chunk, metadata in agent.astream(input, stream_mode="messages"):
            # Only forward model text chunks; skip tool updates and non-text events.
            if metadata.get("langgraph_node") != "model":
                continue
            content = getattr(message_chunk, "content", None)
            if not content:
                continue
            if not sent_role:
                yield sse_chunk(response_id, {"role": "assistant"})
                sent_role = True
            # Send incremental token-like chunks to ElevenLabs in OpenAI format.
            yield sse_chunk(response_id, {"content": content})
         # Signal natural completion before using the finish_reason: "stop" [DONE]
        yield sse_chunk(response_id, {}, finish_reason="stop")
        yield "data: [DONE]\n\n"
    return StreamingResponse(stream(), media_type="text/event-stream")

Ainsi, seuls les fragments du modèle destinés à l’utilisateur sont transmis à ElevenLabs. Comme LangGraph expose l’exécution interne de ses outils dans le flux d’état, le filtrage est explicite et contrôlé par le proxy. 

Examinons maintenant les spécificités de Google Agent Development Kit (ADK).

Google ADK

L’ADK de Google masque la boucle d’exécution derrière quelques primitives fondamentales : Agent, Runner et SessionService. Le Runner d’ADK se situe entre la couche HTTP et la définition de l’agent. Il gère le routage des messages, l’orchestration des outils, le cycle de vie des sessions et le streaming des événements. 

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
)

Une fois l’agent, le backend de session et le runner initialisés, le proxy résout ou crée une session ADK pour chaque requête entrante. Dans ADK, session_id contrôle la persistance de la mémoire : réutiliser le même session_id d’un tour à l’autre conserve automatiquement l’historique, les appels d’outils et les réponses précédentes. L’identité de la conversation étant gérée en amont dans ElevenLabs, le proxy effectue explicitement ce mappage. En transmettant l’identifiant approprié avec la requête de génération, le SDK peut gérer le contexte antérieur en interne. Nous transmettons l’identifiant arbitraire lors de l’initialisation de la conversation via les paramètres supplémentaires transmis dans le corps de la requête.  

Le message et la session préparés, le runner peut être appelé. Les appels d’outils et leurs résultats apparaissent toujours comme des événements ADK internes pendant l’exécution, mais ils sont traités comme des étapes d’orchestration intermédiaires plutôt que comme une sortie destinée à l’utilisateur. Il n’est donc pas nécessaire d’appliquer un filtre manuel, contrairement aux frameworks où les appels d’outils apparaissent comme du texte visible par l’utilisateur. 

Le gestionnaire ci-dessous est une implémentation simplifiée qui inclut directement la résolution de session et la logique de récupération ou de création.

@app.post("/chat/completions")
async def chat_completions(req: ChatCompletionRequest, request: Request):
    # In production, prefer a stable identifier from your upstream system.
    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
            # In SSE mode, ADK emits partial (incremental) and final (complete) events.
            # Forwarding only partial events avoids duplicating the full text.
            # Note: SSE streaming is experimental in ADK. For production, reconcile
            # both event types in case the model backend doesn't emit partials.
            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")

Voyons maintenant CrewAI, dont la conception est davantage centrée sur les tâches.

CrewAI

CrewAI a été conçu pour orchestrer des workflows multi-agents autour de tâches structurées (rechercher, rédiger, résumer), plutôt que de boucles de dialogue ouvertes. Les agents sont définis par un rôle, un objectif et un contexte. L’exécution s’articule autour d’objets Task, chacun doté d’une description claire et d’un résultat attendu. 

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,
)

Contrairement au modèle de boucle d’agent utilisé dans LangGraph et ADK, CrewAI construit généralement un Task et un Crew par requête afin de définir l’unité de travail correspondant à ce tour de conversation. Nous conservons le contexte conversationnel en injectant les tours précédents dans la tâche suivante via un espace réservé. La variable {crew_chat_messages} est alimentée à chaque requête avec l’historique courant de la conversation, puis interpolée dans la description de la tâche au moment de l’exécution. Nous cherchons également à produire un texte propre, prêt à être prononcé, en filtrant explicitement les motifs de traçage intermédiaires (Thought, Action, Action Input, Observation) et en n’émettant que le texte de la réponse finale. 

Le gestionnaire ci-dessous réunit la construction de tâches par requête, l’interpolation de l’historique, le streaming au niveau du Crew, le filtrage des traces et le formatage de la sortie. 

@app.post("/chat/completions")
async def chat_completions(req: ChatCompletionRequest):
    # Task and Crew are assembled per request (not at startup).	
    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 returns CrewStreamingOutput instead of a single 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:
            # Skip non-text events (e.g. tool calls).
            if chunk.chunk_type != StreamChunkType.TEXT or not chunk.content:
                continue
            # Only forward text after the "Final Answer:" marker
            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
            # Clean up any trailing markdown artifacts from CrewAI output.
            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})
        # Fallback to handle short responses without the "Final Answer:" marker
        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"

Examinons maintenant LlamaIndex, qui adopte une approche différente, centrée sur un modèle de streaming natif piloté par les événements.

LlamaIndex

Contrairement aux autres frameworks abordés dans cet article, LlamaIndex a été conçu pour connecter les LLM à des sources de données externes (référentiels de documents, index, pipelines de récupération). Sa couche d’agents, FunctionAgent, s’appuie sur cette base pour récupérer et analyser un contexte structuré, plutôt que pour gérer des dialogues ouverts ou exécuter des tâches.

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,
)

Pour préserver la continuité conversationnelle, le proxy transforme les messages entrants en messages de chat LlamaIndex, puis les sépare entre le dernier tour utilisateur (user_msg) et les tours précédents (chat_history). Le champ event.delta de chaque événement AgentStream contient le fragment de texte suivant, qui correspond directement à un fragment delta.content de style OpenAI. Les deltas non vides peuvent être transmis tels quels, ce qui en fait le pont de streaming le plus direct du guide. Le flux contient à la fois des événements d’orchestration (appels d’outils, résultats) et des événements de parole (deltas de texte de l’assistant). Pour préserver la clarté de la sortie vocale, le proxy ne conserve que les événements AgentStream et ignore les deltas vides.

[1] AgentStream (delta='')       ← ignoré
[2] ToolCall                     ← ignoré
[3] ToolCallResult               ← ignoré
[4] AgentStream (delta='Cela')   ← transmis ✓
[5] AgentStream (delta=' coûte') ← transmis ✓
[6] AgentStream (delta=' $49.99')← transmis ✓

Cette séparation écarte les mécanismes intermédiaires des outils de la sortie parlée, tout en préservant une génération incrémentielle à faible latence. Le gestionnaire prêt à l’emploi ci-dessous réunit ces étapes.

@app.post("/chat/completions")
async def chat_completions(req: ChatCompletionRequest):
    # This assumes the last message is always a user turn with string content.
    # For production, add defensive role/content handling for non-text payloads.
    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 impose moins de règles sur les schémas d’exécution conversationnelle de bout en bout que les frameworks dotés de couches d’orchestration intégrées plus complètes. Pour les déploiements en production, les clients doivent donc généralement implémenter la gestion des sessions, les garde-fous de réponse, l’orchestration des outils et le traçage.

Conclusion

Chaque framework présenté dans ce guide se connecte à ElevenLabs via le même contrat : accepter une requête Completions ou Responses de style OpenAI et renvoyer des fragments SSE en streaming. Les équipes peuvent ainsi ajouter l’orchestration vocale à une implémentation d’agent existante avec un minimum de modifications, préserver ce qu’elles ont déjà construit et activer une IA conversationnelle en temps réel. Cette modularité est un principe fondamental de la plateforme ElevenAgents. Qu’elles étendent un agent existant ou développent une solution native pour la voix dès le départ, les organisations bénéficient d’une orchestration vocale ElevenAgents conçue pour s’adapter à leur situation.

Si vous utilisez déjà un agent reposant sur un framework open source et souhaitez y activer la voix, testez cette approche et faites-nous part de votre avis.