Guida pratica: framework open-source per agenti e ElevenAgents

Nel nostro precedente articolo su L'integrazione di agenti esterni con l'orchestrazione vocale di ElevenLabs, abbiamo illustrato come i team possano collegare a ElevenLabs l'orchestrazione dei propri agenti basati su testo tramite LLM personalizzato. Partendo da queste basi, questa guida mostra come adattare e distribuire i principali framework open source per agenti dietro l'interfaccia del Custom LLM. Il risultato è un'architettura flessibile in cui la voce si integra con sistemi di agenti maturi senza compromettere la gestione dello stato, l'orchestrazione degli strumenti o il controllo specifico dell'applicazione. In tutti i framework seguiamo lo stesso schema in tre passaggi: creiamo una richiesta di generazione, estraiamo la risposta testuale finale e la riformattiamo nel formato Server-Sent Events (SSE) compatibile con OpenAI. ElevenLabs supporta entrambi i formati Chat Completions e Responses. Sebbene questa guida tratti quattro framework molto diffusi, questi modelli si applicano a qualsiasi runtime in grado di produrre output in streaming compatibile con OpenAI.

Configurazione generale

Gli esempi di questa sezione usano Python e FastAPI, ma va bene qualsiasi stack che gestisca richieste HTTP POST e risposte SSE in streaming. Quando l'orchestrazione vocale di ElevenLabs rileva una probabile fine del turno, invia una richiesta di generazione all'endpoint Custom LLM configurato. Questa sezione illustra i componenti principali di questo livello di traduzione: il bridge o proxy che fa parlare la stessa lingua all'orchestrazione vocale e al framework per agenti.

Naturalmente, i clienti possono scegliere ciascun framework per familiarità generale o per la sua capacità di soddisfare uno scopo specifico. LlamaIndex, per esempio, è stato sviluppato originariamente per semplificare la configurazione della Retrieval-Augmented Generation (RAG), mentre CrewAI è stato creato per automatizzare attività definite nell'era degli agenti. Obiettivi progettuali diversi producono strutture di risposta diverse, e ciascuna richiede una gestione specifica. Trasmettere i chunk man mano che l'LLM li genera, anziché attendere un turno completo, è fondamentale perché consente al modello Text-to-Speech (TTS) di iniziare prima a generare il parlato, riducendo così la latenza percepita. Ci concentriamo sui quattro framework più diffusi: LangGraph, Google ADK, CrewAI e LlamaIndex.

Una nota sul codice condiviso

Ogni framework deve trasmettere le risposte come chunk SSE compatibili con OpenAI. Introduciamo una piccola funzione di supporto, usata in tutti gli esempi, per creare questi chunk.

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"

Stabilite queste basi, iniziamo con LangGraph. 

LangGraph

LangGraph modella gli agenti come grafi, in cui i nodi rappresentano i singoli passaggi e gli archi definiscono il flusso di controllo tra di essi. La configurazione minima è semplice: inizializza un modello chat, definisci gli strumenti dell'agente e crea il runtime del grafo dell'agente.

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

Per ogni richiesta di generazione, l'agente LangGraph riceve l'intera cronologia della conversazione, che gli consente di mantenere internamente lo stato necessario. LangGraph supporta la persistenza lato server tramite i checkpoint, che tuttavia non trattiamo qui per mantenere l'implementazione essenziale.

Una volta gestito lo stato, il successivo punto decisionale specifico di LangGraph è la modalità di streaming, per la quale LangGraph offre due opzioni, ciascuna adatta a un caso d'uso diverso:

• stream_mode="values" fornisce snapshot dello stato del grafo. È più semplice da implementare, ma include uno stato dei messaggi più completo in ogni risposta, aumentando la latenza nei flussi conversazionali in tempo reale.
• stream_mode="messages" trasmette dal modello chunk incrementali di messaggi. In genere è preferibile per le interazioni vocali in tempo reale, perché riduce il tempo necessario per il primo audio nel livello di orchestrazione di ElevenLabs.

Più nello specifico, l'implementazione messages del loop dell'agente include passaggi intermedi, come gli aggiornamenti di tool calling, che non dovrebbero essere pronunciati. Il proxy li filtra e passa al livello TTS soltanto il testo della risposta destinato all'utente. Ecco un esempio di turno con strumenti abilitati.

[1] Il modello decide di chiamare uno strumento (tool_calls=["get_price"])
[2] Lo strumento viene eseguito e restituisce dati (result="$24.99") 
[3] Il modello produce una risposta usando il risultato (content="Costa $24.99") 

Naturalmente, nello stream SSE devono essere inoltrati soltanto i chunk del passaggio 3. In pratica, questo filtraggio viene gestito nel loop di streaming da due controlli: uno conserva soltanto gli eventi langgraph_node == "model", l'altro ignora i contenuti vuoti. Insieme, questi controlli assicurano che a ElevenLabs venga inoltrato come SSE soltanto il testo dell'assistente destinato all'utente. Riunendo questi concetti, proponiamo un'implementazione leggera del proxy per le richieste.

@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")

Questo assicura che a ElevenLabs vengano inoltrati soltanto i chunk del modello destinati all'utente. Poiché LangGraph rende visibile l'esecuzione interna degli strumenti tramite lo stream di stato, il filtraggio è esplicito e controllato dal proxy. 

Vediamo ora le particolarità dell'uso dell'Agent Development Kit (ADK) di Google

Google ADK

L'ADK di Google astrae il loop del runtime dietro alcune primitive fondamentali: Agent, Runner e SessionService. Il Runner di ADK si colloca tra il livello HTTP e la definizione dell'agente. Gestisce il routing dei messaggi, l'orchestrazione degli strumenti, il ciclo di vita delle sessioni e lo streaming degli eventi. 

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
)

Dopo aver inizializzato l'agente, il backend delle sessioni e il runner, il proxy risolve o crea una sessione ADK per ogni richiesta in arrivo. In ADK, session_id controlla la persistenza della memoria: riutilizzare lo stesso session_id tra i turni conserva automaticamente la cronologia, le chiamate agli strumenti e le risposte precedenti. Poiché l'identità della conversazione è gestita a monte in ElevenLabs, il proxy gestisce esplicitamente questa mappatura. Passando l'identificatore corretto per la richiesta di generazione, l'SDK può gestire internamente il contesto precedente. Durante l'avvio della conversazione passiamo l'identificatore arbitrario tramite parametri aggiuntivi passati al body della richiesta.  

Una volta preparati il messaggio e la sessione, è possibile invocare il runner. Durante l'esecuzione, le chiamate agli strumenti e i relativi risultati compaiono comunque come eventi ADK interni, ma vengono trattati come passaggi intermedi di orchestrazione anziché come output destinato all'utente. Questo elimina la necessità di un filtro manuale rispetto ai framework in cui le chiamate agli strumenti appaiono come testo visibile all'utente. 

L'handler seguente è un'implementazione semplificata che include inline la risoluzione della sessione e la logica di recupero o creazione.

@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")

Vediamo ora CrewAI, che per progettazione è più incentrato sulle attività.

CrewAI

CrewAI è stato progettato per orchestrare workflow multi-agente attorno ad attività strutturate (ricercare, scrivere, riassumere), anziché a loop di dialogo aperti. Gli agenti sono definiti da un ruolo, un obiettivo e un background. L'esecuzione è incentrata su oggetti Task, ciascuno con una descrizione chiara e un output previsto. 

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

A differenza del modello a loop dell'agente usato in LangGraph e ADK, CrewAI in genere costruisce un Task e un Crew per ogni richiesta, così da definire l'unità di lavoro per quel turno della conversazione. Manteniamo il contesto conversazionale inserendo i turni precedenti nell'attività successiva tramite un placeholder. La variabile {crew_chat_messages} viene popolata a ogni richiesta con la cronologia della conversazione in corso, quindi interpolata nella descrizione dell'attività al momento dell'esecuzione. Puntiamo inoltre a produrre testo pulito e pronto per il parlato, filtrando esplicitamente i pattern di tracciamento intermedi (Thought, Action, Action Input, Observation) ed emettendo soltanto il testo della risposta finale. 

L'handler seguente riunisce la costruzione dell'attività per richiesta, l'interpolazione della cronologia, lo streaming a livello di Crew, il filtraggio delle tracce e la formattazione dell'output. 

@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"

Vediamo ora LlamaIndex, che segue un approccio diverso incentrato su un modello di streaming nativo basato sugli eventi.

LlamaIndex

A differenza degli altri framework trattati in questo articolo, LlamaIndex è stato progettato per collegare gli LLM a fonti di dati esterne (archivi di documenti, indici, pipeline di retrieval). Il suo livello per agenti, FunctionAgent, si basa su queste fondamenta per recuperare informazioni e ragionare sul contesto strutturato, anziché per dialoghi aperti o l'esecuzione di attività.

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

Per preservare la continuità conversazionale, il proxy trasforma i messaggi in arrivo in messaggi chat di LlamaIndex, poi li divide nell'ultimo turno dell'utente (user_msg) e nei turni precedenti (chat_history). Il campo event.delta di ciascun evento AgentStream contiene il frammento di testo successivo, che viene mappato direttamente in un chunk delta.content in stile OpenAI. I delta non vuoti possono essere inoltrati così come sono, rendendo questo il bridge di streaming più diretto della guida. Lo stream contiene sia eventi di orchestrazione (chiamate agli strumenti, risultati) sia eventi vocali (delta di testo dell'assistente). Per mantenere pulito l'output vocale, il proxy conserva soltanto gli eventi AgentStream e ignora i delta vuoti.

[1] AgentStream (delta='')       ← ignorato
[2] ToolCall                     ← ignorato
[3] ToolCallResult               ← ignorato
[4] AgentStream (delta='It')     ← inoltrato ✓
[5] AgentStream (delta=' costs') ← inoltrato ✓
[6] AgentStream (delta=' $49.99')← inoltrato ✓

Questa separazione esclude dall'output pronunciato i meccanismi intermedi degli strumenti, preservando al contempo il parlato incrementale a bassa latenza. L'handler pronto all'uso qui sotto riunisce questi passaggi.

@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 è meno prescrittivo riguardo ai modelli di runtime conversazionale end-to-end rispetto ai framework con livelli di orchestrazione integrati più complessi. Per le distribuzioni in produzione, questo richiede in genere ai clienti di implementare la gestione delle sessioni, le misure di protezione delle risposte, l'orchestrazione degli strumenti e il tracing.

Conclusione

Ogni framework di questa guida si collega a ElevenLabs tramite lo stesso contratto: accettare una richiesta Completions o Responses in stile OpenAI e restituire chunk SSE in streaming. Questo permette ai team di aggiungere l'orchestrazione vocale a un'implementazione di agenti esistente con modifiche minime, preservando ciò che hanno già creato e abilitando l'IA conversazionale in tempo reale. Questa modularità è un principio fondamentale della piattaforma ElevenAgents. Che le organizzazioni stiano estendendo un agente esistente o creando da zero un'esperienza nativa per la voce, l'orchestrazione vocale di ElevenAgents è progettata per adattarsi alle loro esigenze.

Se stai già eseguendo un agente con un framework open source e vuoi abilitare la voce, prova questo approccio e facci sapere cosa ne pensi.