Guida pratica: framework open-source per agenti e ElevenAgents

Nel nostro post precedente su Integrazione di agenti esterni con l'orchestrazione vocale di ElevenLabs, abbiamo spiegato come i team possono collegare la loro orchestrazione di agenti basata su testo a ElevenLabs tramite LLM personalizzato. Partendo da questa base, in questa guida ti mostriamo come i principali framework open-source per agenti possono essere adattati e utilizzati dietro l’interfaccia Custom LLM. Il risultato è un’architettura flessibile in cui la voce si integra su sistemi di agenti già 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: creazione della richiesta di generazione, estrazione della risposta testuale finale e riformattazione in formato Server-Sent Events (SSE) compatibile con OpenAI. ElevenLabs supporta sia il formato Completamenti chatche Risposte. Anche se questa guida tratta quattro framework molto diffusi, gli schemi si applicano a qualsiasi runtime che possa produrre output in streaming compatibile con OpenAI.

Impostazione generale

Gli esempi di questa sezione usano Python e FastAPI, ma qualsiasi stack che gestisce richieste HTTP POST e risposte SSE in streaming va bene. Quando l’orchestrazione vocale di ElevenLabs rileva la probabile fine di un turno, invia una richiesta di generazione all’endpoint Custom LLM configurato. In questa sezione vediamo i componenti principali di questo livello di traduzione, ovvero il ponte o proxy che permette all’orchestrazione vocale e al framework di agenti di comunicare.

Ogni framework può essere scelto dai clienti per familiarità o per la capacità di rispondere a esigenze specifiche. LlamaIndex, ad esempio, è nato per semplificare la Retrieval-Augmented Generation (RAG), mentre CrewAI è stato creato per automatizzare task definiti nell’era degli agenti. Obiettivi di progettazione diversi portano a strutture di risposta diverse, ognuna con le sue particolarità. Inviare i chunk in streaming man mano che l’LLM li genera, invece di aspettare la risposta completa, è fondamentale perché permette al modello Text-to-Speech (TTS) di iniziare a generare la voce prima, riducendo la latenza percepita. Ci concentriamo su quattro framework popolari: LangGraph, Google ADK, CrewAI e LlamaIndex.

Nota sul codice condiviso

Ogni framework deve inviare le risposte in chunk SSE compatibili con OpenAI. Ti mostriamo una piccola funzione di supporto usata negli esempi per costruire questi chunk.

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

    payload = {

Con queste basi, iniziamo con LangGraph.

LangGraph

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

    }

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

Per ogni richiesta di generazione, l'Agent di LangGraph riceve l'intera cronologia della conversazione, così può mantenere lo stato necessario internamente. LangGraph supporta la persistenza lato server tramite

Gestita la gestione dello stato, il prossimo punto specifico di LangGraph riguarda la modalità di streaming, dove LangGraph offre due opzioni, ognuna adatta a un caso d'uso diverso:

LangGraph modella gli agenti come grafi, dove i nodi rappresentano i singoli passaggi e gli archi definiscono il flusso di controllo tra di essi. La configurazione minima è semplice: si inizializza un modello di chat, si definiscono gli strumenti dell’agente e si 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, il LangGraph Agent riceve l’intera cronologia della conversazione, così può gestire lo stato necessario internamente. LangGraph supporta la persistenza lato server tramite Checkpoint, anche se qui non li trattiamo per mantenere l’implementazione essenziale.

Gestita la gestione dello stato, la prossima decisione specifica di LangGraph riguarda la modalità di streaming, dove LangGraph offre due opzioni, ognuna adatta a casi d’uso diversi:

• stream_mode="values" fornisce snapshot dello stato del grafo. È più semplice da implementare ma include lo stato completo del messaggio a ogni risposta, aumentando la latenza nelle conversazioni in tempo reale.
• parametri extra

Con messaggio e sessione pronti, puoi invocare il runner. Le chiamate agli strumenti e i risultati degli strumenti appaiono comunque come eventi interni ADK durante l’esecuzione, ma sono trattati come passaggi intermedi di orchestrazione e non come output rivolto all’utente. Questo elimina la necessità di un filtro manuale rispetto ai framework dove le chiamate agli strumenti appaiono come testo visibile all’utente.

Il gestore qui sotto è una versione semplificata che include la risoluzione della sessione e la logica get-or-create.

[2] Lo strumento viene eseguito e restituisce i dati (result="$24.99")

[3] Il modello produce la risposta usando il risultato (content="Costa $24.99")

Ora vediamo CrewAI, che per sua natura è più orientato ai task.

CrewAI

CrewAI è stato progettato per orchestrare workflow multi-agente attorno a task strutturati (ricerca, scrittura, sintesi) invece che su cicli di dialogo aperti. Gli agenti sono definiti con ruolo, obiettivo e backstory. L’esecuzione ruota attorno a oggetti Task, ognuno con una descrizione chiara e un output atteso.

    input = {"messages": req.messages}

    async def stream():

A differenza del modello agent-loop usato in LangGraph e ADK, CrewAI di solito costruisce Task e Crew per ogni richiesta, così da definire l’unità di lavoro per quel turno nella conversazione. Manteniamo il contesto conversazionale inserendo i turni precedenti nel task successivo tramite un placeholder. La variabile {crew_chat_messages} viene popolata a ogni richiesta con la cronologia della conversazione, poi interpolata nella descrizione del task al momento dell’esecuzione. Puntiamo inoltre a produrre testo pulito e pronto per la voce, filtrando esplicitamente pattern intermedi (Thought, Action, Action Input, Observation) ed emettendo solo la risposta finale.

Il gestore qui sotto unisce la costruzione del task per richiesta, l’inserimento della cronologia, lo streaming a livello Crew, il filtraggio delle tracce e la formattazione dell’output.

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

            # Inoltra solo i chunk di testo del modello; salta aggiornamenti tool e eventi non testuali.

Ora vediamo LlamaIndex, che segue un approccio diverso puntando su un modello di streaming nativo guidato dagli eventi.

LlamaIndex

A differenza degli altri framework trattati in questo articolo, LlamaIndex è stato progettato per collegare LLM a fonti dati esterne (document store, indici, pipeline di retrieval). Il suo layer agent, FunctionAgent, si basa su questa struttura per recuperare e ragionare su contesti strutturati, invece che su dialoghi aperti o esecuzione di task.

            if not content:

                continue

Per mantenere la continuità della conversazione, il proxy trasforma i messaggi in arrivo in chat message di LlamaIndex, poi li divide nell’ultimo turno utente (user_msg) e nei turni precedenti (chat_history). Il campo event.delta di ogni evento AgentStream contiene il prossimo frammento di testo, che corrisponde direttamente a 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 l’output vocale pulito, il proxy mantiene solo gli eventi AgentStream e salta i delta vuoti.

Per mantenere la continuità della conversazione, il proxy trasforma i messaggi in arrivo in messaggi chat di LlamaIndex, poi li divide nell’ultimo turno utente (user_msg) e nei turni precedenti (chat_history). Ogni campo event.delta di AgentStream contiene il prossimo frammento di testo, che corrisponde direttamente a 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 pulita l’uscita vocale, il proxy conserva solo gli eventi AgentStream e salta i delta vuoti.

[1] AgentStream (delta='')       ← ignorato

Questa separazione tiene le meccaniche intermedie degli strumenti fuori dall’output vocale, mantenendo però una voce incrementale a bassa latenza. L’handler qui sotto unisce questi passaggi.

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

LlamaIndex è meno prescrittivo sui pattern di runtime conversazionale end-to-end rispetto ai framework con orchestrazione più pesante. Per l’uso in produzione, di solito è necessario che i clienti implementino la gestione delle sessioni, i guardrail sulle risposte, l’orchestrazione degli strumenti e il tracing.

LlamaIndex è meno rigido sui pattern runtime conversazionali end-to-end rispetto ai framework con orchestrazione integrata più pesante. Per l’uso in produzione, di solito è necessario che i clienti implementino la gestione delle sessioni, le regole di sicurezza delle risposte, l’orchestrazione degli strumenti e il tracing.

Conclusione

Ogni framework in questa guida si collega a ElevenLabs tramite lo stesso contratto: accetta una richiesta di Completions o Responses in stile OpenAI e restituisce in streaming chunk SSE. Questo permette ai team di aggiungere l’orchestrazione vocale sopra un’implementazione di agent già esistente con modifiche minime, mantenendo ciò che hanno già costruito e abilitando l’audio in tempo reale

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

Ora vediamo le particolarità di Google Agent Development Kit (ADK)

Google ADK

L’ADK di Google astrae il ciclo runtime dietro pochi elementi principali: Agent, Runner e SessionService. Il Runner di ADK si trova tra il layer HTTP e la definizione dell’agente. Gestisce l’instradamento 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

)

Con agente, backend di sessione e runner inizializzati, 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 mantiene automaticamente cronologia, chiamate agli strumenti e risposte precedenti. Poiché l’identità della conversazione è gestita a monte da ElevenLabs, il proxy si occupa di questa mappatura in modo esplicito. Passando l’identificativo corretto nella richiesta di generazione, l’SDK può gestire il contesto precedente internamente. L’identificativo arbitrario viene passato all’avvio della conversazione tramite parametri extra inclusi nel body della richiesta.

Con messaggio e sessione pronti, si può invocare il runner. Le chiamate agli strumenti e i risultati appaiono comunque come eventi interni ADK durante l’esecuzione, ma sono trattati come passaggi intermedi di orchestrazione e non come output rivolto all’utente. Questo elimina la necessità di un filtro manuale rispetto ai framework dove le chiamate agli strumenti appaiono come testo visibile all’utente.

Il seguente handler è una versione semplificata che include la risoluzione della sessione e la logica get-or-create.

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

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

    # In produzione, usa un identificativo stabile dal tuo sistema a monte.

    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 modalità SSE, ADK emette eventi parziali (incrementali) e finali (completi).

            # Inoltrare solo gli eventi parziali evita di duplicare il testo completo.
            # Nota: lo streaming SSE è sperimentale in ADK. In produzione, gestisci

            # entrambi i tipi di evento nel caso il backend del modello non emetta i parziali.

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

Ora vediamo CrewAI, che per sua natura è più orientato ai task.

CrewAI

CrewAI è stato progettato per orchestrare workflow multi-agente su task strutturati (ricerca, scrittura, sintesi) invece che su dialoghi aperti. Gli agenti sono definiti con ruolo, obiettivo e backstory. L’esecuzione ruota attorno a oggetti Task, ognuno con una descrizione chiara e un output atteso.

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 agent-loop usato in LangGraph e ADK, CrewAI di solito costruisce Task e Crew per ogni richiesta, così da definire l’unità di lavoro per quel turno di conversazione. Manteniamo il contesto della conversazione inserendo i turni precedenti nel task successivo tramite una variabile. La variabile {crew_chat_messages} viene popolata a ogni richiesta con la cronologia della conversazione e poi inserita nella descrizione del task al momento dell’esecuzione. Inoltre, per produrre testo pulito e pronto per la voce, filtriamo esplicitamente i pattern intermedi di tracing (Thought, Action, Action Input, Observation) e inviamo solo il testo della risposta finale.

L’handler qui sotto unisce la costruzione del task per richiesta, l’inserimento della cronologia, lo streaming a livello Crew, il filtraggio dei trace e la formattazione dell’output.

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

async def chat_completions(req: ChatCompletionRequest):

    # Task e Crew vengono creati per ogni richiesta (non all’avvio).

    task = Task(

        description=(

            "Cronologia conversazione:\n{crew_chat_messages}\n\n"

            "Rispondi all’ultimo messaggio dell’utente."

        ),

        expected_output=expected_output,

        agent=store_agent,

    )

    # stream=True restituisce CrewStreamingOutput invece di un singolo 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:

            # Salta eventi non testuali (es. chiamate tool).

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

                continue

            # Inoltra solo il testo dopo il marker "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

            # Pulisci eventuali artefatti markdown in coda dall’output CrewAI.

            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 per gestire risposte brevi senza marker "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"

Ora vediamo LlamaIndex, che segue un approccio diverso puntando su un modello di streaming nativo guidato dagli eventi.

LlamaIndex

A differenza degli altri framework trattati in questo post, LlamaIndex è stato progettato per collegare gli LLM a fonti dati esterne (document store, indici, pipeline di retrieval). Il suo layer agent, FunctionAgent, si basa su questa struttura per recuperare e ragionare su contesti strutturati, più che per dialoghi aperti o esecuzione di task.

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 mantenere la continuità della conversazione, il proxy trasforma i messaggi in ingresso in chat message di LlamaIndex, poi li divide nell’ultimo turno utente (user_msg) e nei turni precedenti (chat_history). Ogni campo event.delta di AgentStream contiene il prossimo frammento di testo, che corrisponde direttamente a un chunk delta.content in stile OpenAI. I delta non vuoti possono essere inoltrati così come sono, rendendo questo il bridge di streaming più semplice della guida. Lo stream contiene sia eventi di orchestrazione (chiamate tool, risultati) sia eventi vocali (delta di testo dell’assistente). Per mantenere pulita la voce, il proxy tiene solo gli eventi AgentStream e salta 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 tiene fuori dall’output vocale la meccanica intermedia degli strumenti, mantenendo uno speech incrementale a bassa latenza. L’handler qui sotto mette insieme questi passaggi.

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

async def chat_completions(req: ChatCompletionRequest):

    # Si assume che l’ultimo messaggio sia sempre un turno utente con contenuto stringa.

    # In produzione, aggiungi controlli difensivi su ruolo/contenuto per payload non testuali.

    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 sui pattern runtime conversazionali end-to-end rispetto ai framework con orchestrazione più strutturata. Per l’uso in produzione, di solito serve che i clienti implementino la gestione delle sessioni, i guardrail sulle risposte, l’orchestrazione degli strumenti e il tracing.

Conclusione

Ogni framework di questa guida si collega a ElevenLabs tramite lo stesso schema: accetta una richiesta Completions o Responses in stile OpenAI e restituisce chunk SSE in streaming. Così puoi aggiungere l’orchestrazione vocale sopra una implementazione agent già esistente con modifiche minime, mantenendo ciò che hai già costruito e abilitando l’IA conversazionale in tempo reale. Questa modularità è un principio fondamentale della piattaforma ElevenAgents. Che tu stia estendendo un agente esistente o costruendo da zero in ottica voice-native, l’orchestrazione vocale di ElevenAgents è pensata per adattarsi alle tue esigenze.

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