Workflow come codice

Panoramica

Un workflow viene archiviato come un grafo di nodi e archi. Scrivere manualmente quel grafo significa dover mantenere coerenti gli ID degli archi, gli elenchi edge_order e gli oggetti delle condizioni. Il builder Workflow negli SDK Python e JavaScript ti consente di descrivere lo stesso grafo con chiamate a metodi concatenate. Produce quindi il valore workflow già accettato da agents.create() e agents.update().

Usalo per:

• Creare un workflow nel codice e revisionarlo come qualsiasi altra modifica al codice.
• Caricare il workflow di un agente, modificare un nodo e inviarlo di nuovo senza alterare il resto del grafo.
• Archiviare i workflow come JSON nel controllo di versione e convalidarli in CI.

Requisiti

• Python: il pacchetto elevenlabs. Il builder viene importato da elevenlabs.agent_workflows.
• JavaScript: il pacchetto @elevenlabs/elevenlabs-js. Il builder viene esportato dalla radice del pacchetto.

Crea un workflow

L’oggetto Workflow rappresenta il nodo iniziale. Ogni verbo aggiunge un nodo e lo restituisce, così puoi concatenare il passaggio successivo a partire da esso:

I metodi di instradamento stabiliscono quale arco porta al nodo successivo:

I verbi chiamati direttamente sul workflow si diramano dal nodo iniziale con un arco incondizionato. Questo esempio indirizza i chiamanti a una ricerca dell’ordine, riprova l’identificazione se la ricerca fallisce e trasferisce i chiamanti VIP o frustrati:

from elevenlabs import ElevenLabs
from elevenlabs.agent_workflows import Destination, Expr, Workflow
elevenlabs = ElevenLabs()
wf = Workflow()
triage = wf.converse("triage", "Greet the caller and find out why they're calling")
identify = triage.when("Caller wants to check an order").converse(
    "identify", "Ask for the order number and the email on the account"
)
lookup = identify.when("Caller gave an order number").tool("lookup_order", ["tool_get_order"])
lookup.on_result(
    success=lambda ok: ok.converse("resolve", "Explain the order status")
        .when("Caller is satisfied")
        .end(),
    failure=lambda failed: failed.go_to(identify),
)
triage.when(Expr.variable("tier").eq("gold")).transfer("vip", Destination.agent("agent_vip"))
triage.when("Caller wants a human").transfer("human", Destination.phone("+15551234567"))
elevenlabs.conversational_ai.agents.update("agent_id", workflow=wf.build())

build() convalida l’intero grafo e restituisce il modello della richiesta. Inoltre:

• Assegna al nodo iniziale l’ID start_node, richiesto dall’API.
• Genera gli ID degli archi nel formato source__target.
• Compila edge_order di ogni nodo nell’ordine in cui hai aggiunto gli archi.
• Posiziona ogni nodo senza posizione, in modo che il grafo sia disposto da sinistra a destra nell’editor visivo.

Scrivere un workflow come un’unica catena

up() restituisce il nodo da cui è stato concatenato quello corrente, così puoi descrivere un intero workflow in un’unica espressione invece di conservare ogni nodo in una variabile. All’interno di una singola catena, usa go_to() / goTo() con un ID nodo, poiché non c’è una variabile da passare:

wf = Workflow()
(
    wf.converse("triage", "Find out why the customer is calling")
    .when("Caller gave an order number")
    .tool("lookup_order", ["tool_get_order"])
    .on_result(
        success=lambda ok: ok.converse("resolve", "Explain the order status").when("Customer is satisfied").end(),
        failure=lambda failed: failed.go_to("triage"),
    )
    .up()
    .when(Expr.variable("tier").eq("gold"))
    .transfer("vip", Destination.agent("agent_vip_desk"))
)
elevenlabs.conversational_ai.agents.update("agent_id", workflow=wf.build())

Per un handle di nodo da wf.node(id), up() restituisce il nodo da cui proviene il primo arco in entrata. Per il nodo iniziale, restituisce il nodo iniziale.

Opzioni dei nodi

Le impostazioni facoltative vanno inserite negli argomenti con nome in Python e in un oggetto opzioni finale in JavaScript. I loro nomi corrispondono ai campi del nodo nell’API:

Ogni verbo accetta anche position se vuoi posizionare tu stesso il nodo. Workflow(prevent_subagent_loops=True) in Python, oppure new Workflow({ preventSubagentLoops: true }) in JavaScript, impedisce alla conversazione di rientrare in un sottoagente già visitato.

triage = wf.converse(
    "triage",
    "Greet the caller and find out why they're calling",
    label="Triage",
    tool_ids=["tool_crm_search"],
    entry_behavior="generate_immediately",
)
triage.when("Caller is a VIP").transfer(
    "vip_desk",
    Destination.agent("agent_vip_concierge"),
    transfer_message="Connecting you to our VIP team",
    delay_ms=500,
)
identify.when("Caller refuses to share details").end(id="abandoned")

Destinazioni di trasferimento

Condizioni

when() accetta una stringa semplice, una Condition o un Expr:

• Una stringa è una condizione LLM: l’LLM decide se è soddisfatta.
• Condition.llm(text, label=...) è equivalente, con un’etichetta visualizzata sull’arco nell’editor visivo. Usa Condition.expr(expr, label=...) per etichettare un’espressione.
• Un Expr viene valutato rispetto alle variabili dinamiche, senza chiamare l’LLM.

Crea espressioni con Expr.variable(name) e Expr.literal(value). Confrontale con eq, neq, gt, gte, lt e lte, ed esegui operazioni aritmetiche con add, sub, mul e div. Combinale con all_of / allOf, any_of / anyOf e if_else / ifElse. I valori semplici vengono racchiusi automaticamente in Expr.literal(). Expr.llm(description) chiede all’LLM di estrarre un valore dalla conversazione; passa type ed enum per ottenere una stringa, un numero o uno di un insieme fisso invece di un booleano.

from elevenlabs.agent_workflows import Condition, Expr
is_vip = Expr.any_of(
    Expr.variable("tier").eq("gold"),
    Expr.all_of(Expr.variable("orders").gte(10), Expr.variable("churn_risk").eq(True)),
)
triage.when(Condition.expr(is_vip, label="VIP")).transfer("vip_desk", Destination.agent("agent_vip"))
big_refund = Expr.variable("unit_price").mul(Expr.variable("quantity")).gt(500)
resolve.when(big_refund).transfer("finance", Destination.phone("+15551234567"))
score = Expr.llm("How satisfied the caller sounds, from 1 to 5", type="integer")
resolve.when(score.lte(2)).converse("recovery", "Apologize and offer a discount code")

Cicli e archi all’indietro

L’API consente un arco tra due nodi qualsiasi, in entrambe le direzioni. Per consentire alla conversazione di tornare indietro, assegna all’arco una condizione all’indietro con back:

faq = triage.when("Caller has a general question", back="Question answered").converse(
    "faq", "Answer using the FAQ knowledge base"
)

go_to() / goTo() gestisce automaticamente il caso più comune. Quando il nodo di destinazione ha già un arco che porta qui, il nuovo arco diventa la condizione all’indietro di quell’arco. Nel primo esempio, failed.go_to(identify) viene inviato come condizione all’indietro dell’arco identify → lookup_order. Qualsiasi altro secondo arco tra gli stessi due nodi, e qualsiasi arco da un nodo a sé stesso, non supera la convalida.

Modifica un agente esistente

Workflow.from_agent() / Workflow.fromAgent() carica il workflow di un agente restituito da agents.get(). Gli ID dei nodi, gli ID degli archi, le posizioni nell’editor e l’ordine degli archi vengono mantenuti, quindi reinviandolo cambierà solo ciò che hai modificato.

insert_between() / insertBetween() passa alla tua funzione un ramo che conserva la condizione dell’arco sostituito. La funzione restituisce l’ultimo nodo aggiunto, che viene quindi collegato alla destinazione originale. Il metodo restituisce quel nodo, oppure None / undefined se non esiste alcun arco tra i due nodi.

agent = elevenlabs.conversational_ai.agents.get("agent_id")
wf = Workflow.from_agent(agent)
wf.node("resolve").update(prompt="Explain the status, then offer a discount code")
verify = wf.insert_between(
    "identify",
    "lookup_order",
    lambda start: start.converse("verify", "Verify the caller's identity with a one-time code"),
)
if verify is not None:
    verify.when("Caller can't verify").end(id="unverified")
wf.remove("vip")
elevenlabs.conversational_ai.agents.update("agent_id", workflow=wf.build())

Archivia i workflow nel controllo di versione

to_json_string() / toJSONString() scrive il workflow nel formato wire dell’API: chiavi snake_case, ordinate a ogni livello, con rientro di 2 spazi. Ha la stessa struttura usata dall’API REST e dalla CLI, ed entrambi gli SDK scrivono file identici byte per byte per lo stesso workflow, così le differenze restano minime. Workflow.from_json() / Workflow.fromJSON() lo legge nuovamente e accetta anche una configurazione completa dell’agente con una chiave workflow, come un file nella directory agent_configs/ della CLI.

from pathlib import Path
from elevenlabs.agent_workflows import Workflow
Path("workflows/support.json").write_text(wf.to_json_string())
stored = Workflow.from_json(Path("workflows/support.json").read_text())
from_cli = Workflow.from_json(Path("agent_configs/support.json").read_text())

Una configurazione tipica:

Visualizza un workflow

La CLI disegna un file di workflow come diagramma di flusso, così puoi verificare ciò che ha creato il tuo codice senza aprire la dashboard. Accetta un file di workflow semplice scritto da to_json_string() / toJSONString(), oppure una configurazione dell’agente con una chiave workflow:

elevenlabs agents workflow diagram workflows/support.json

Per impostazione predefinita, disegna riquadri e frecce nel terminale, con ogni condizione dell’arco numerata sul relativo arco e riportata in una legenda sottostante. Usa invece --format mermaid per stampare il sorgente Mermaid, oppure --format mermaid-md per racchiuderlo in un blocco di codice delimitato che GitHub visualizza, ad esempio in un commento a una pull request:

elevenlabs agents workflow diagram workflows/support.json --format mermaid-md

Il comando legge solo file locali, quindi non richiede una chiave API e può essere eseguito in CI.

Convalida

Le chiamate concatenate non falliscono mai da sole. build(), to_json() / toJSON() e to_json_string() / toJSONString() verificano l’intero grafo e sollevano o generano un singolo WorkflowValidationError, il cui elenco issues riporta ogni problema, ciascuno con un code, un message e il node_id / nodeId a cui si riferisce.

from elevenlabs.agent_workflows import WorkflowValidationError
try:
    workflow = wf.build(on_warning=lambda issue: print("warning:", issue.message))
except WorkflowValidationError as err:
    for issue in err.issues:
        print(issue.code, issue.message)

Gli errori includono:

• Due archi tra la stessa coppia di nodi, oppure un arco da un nodo a sé stesso.
• Un arco verso il nodo iniziale o in uscita da un nodo finale.
• Un arco da un nodo strumento che non sia un arco di successo o fallimento, oppure un arco di successo o fallimento da qualsiasi altro tipo di nodo.
• Due nodi con lo stesso ID, oppure un arco verso un nodo inesistente.

I nodi che non possono essere raggiunti dal nodo iniziale vengono segnalati come avvisi, tramite on_warning / onWarning, e non interrompono la creazione.

Tipi di nodo più recenti del tuo SDK

Un workflow caricato dall’API o da un file può contenere tipi di nodo che la tua versione dell’SDK non conosce ancora. Vengono mantenuti invariati tramite from_json() / fromJSON() e to_json_string() / toJSONString(), e build() segnala ciascuno di essi come avviso. Il modello di richiesta tipizzato non può includerli, quindi aggiorna l’SDK oppure invia direttamente il formato wire:

elevenlabs.conversational_ai.agents.update(
    "agent_id",
    request_options={"additional_body_parameters": {"workflow": wf.to_json()}},
)