Workflows como código

Información general

Un flujo de trabajo se almacena como un grafo de nodos y aristas. Escribir ese grafo a mano implica que debes mantener coherentes por tu cuenta los ID de las aristas, las listas edge_order y los objetos de condición. El constructor Workflow de los SDK de Python y JavaScript te permite describir el mismo grafo mediante llamadas a métodos encadenadas. Después genera el valor workflow que agents.create() y agents.update() ya aceptan.

Úsalo para:

• Crear un flujo de trabajo en código y revisarlo como cualquier otro cambio de código.
• Cargar el flujo de trabajo de un agente, modificar un nodo y volver a enviarlo sin alterar el resto del grafo.
• Guardar flujos de trabajo como JSON en el control de versiones y validarlos en CI.

Requisitos

• Python: el paquete elevenlabs. El constructor se importa desde elevenlabs.agent_workflows.
• JavaScript: el paquete @elevenlabs/elevenlabs-js. El constructor se exporta desde la raíz del paquete.

Crear un flujo de trabajo

El objeto Workflow representa el nodo inicial. Cada verbo añade un nodo y lo devuelve, por lo que puedes encadenar desde él el siguiente paso:

Los métodos de enrutamiento deciden qué arista lleva al siguiente nodo:

Los verbos llamados directamente en el flujo de trabajo se ramifican desde el nodo inicial con una arista incondicional. Este ejemplo dirige a quienes llaman a una consulta de pedidos, vuelve a intentar la identificación si la consulta falla y transfiere a quienes son VIP o están frustrados:

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() valida todo el grafo y devuelve el modelo de solicitud. También:

• Asigna al nodo inicial el ID start_node, que requiere la API.
• Genera ID de arista con el formato source__target.
• Rellena el edge_order de cada nodo en el orden en que añadiste las aristas.
• Posiciona cualquier nodo que no tenga posición para que el grafo se organice de izquierda a derecha en el editor visual.

Escribir un flujo de trabajo como una sola cadena

up() devuelve el nodo desde el que se encadenó el actual, para que puedas describir un flujo de trabajo completo en una expresión en lugar de guardar cada nodo en una variable. Dentro de una sola cadena, usa go_to() / goTo() con un ID de nodo, ya que no hay ninguna variable que pasar:

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

Para un identificador de nodo de wf.node(id), up() devuelve el nodo del que procede la primera arista que entra en él. Para el nodo inicial, devuelve el nodo inicial.

Opciones de nodo

Los ajustes opcionales se indican como argumentos con nombre en Python y en un objeto de opciones final en JavaScript. Sus nombres coinciden con los campos del nodo en la API:

Cada verbo también acepta position si quieres colocar el nodo manualmente. Workflow(prevent_subagent_loops=True) en Python, o new Workflow({ preventSubagentLoops: true }) en JavaScript, evita que la conversación vuelva a entrar en un subagente que ya ha visitado.

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

Destinos de transferencia

Condiciones

when() acepta una cadena simple, un Condition o un Expr:

• Una cadena es una condición de LLM: el LLM decide si se cumple.
• Condition.llm(text, label=...) es lo mismo, con una etiqueta que se muestra en la arista del editor visual. Usa Condition.expr(expr, label=...) para etiquetar una expresión.
• Un Expr se evalúa con variables dinámicas, sin llamar al LLM.

Crea expresiones a partir de Expr.variable(name) y Expr.literal(value). Compáralas con eq, neq, gt, gte, lt y lte, y realiza operaciones aritméticas con add, sub, mul y div. Combínalas con all_of / allOf, any_of / anyOf e if_else / ifElse. Los valores simples se envuelven en Expr.literal() automáticamente. Expr.llm(description) pide al LLM que extraiga un valor de la conversación; pasa type y enum para obtener una cadena, un número o uno de un conjunto fijo en lugar de 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")

Bucles y aristas hacia atrás

La API permite una arista entre dos nodos cualesquiera, en cualquiera de las dos direcciones. Para que la conversación pueda volver por donde vino, asigna a la arista una condición hacia atrás con back:

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

go_to() / goTo() se encarga del caso habitual por ti. Cuando el nodo de destino ya tiene una arista que lleva hasta aquí, la nueva arista se convierte en la condición hacia atrás de esa arista. En el primer ejemplo, failed.go_to(identify) se envía como la condición hacia atrás de la arista identify → lookup_order. Cualquier otra segunda arista entre los mismos dos nodos, y cualquier arista de un nodo hacia sí mismo, no supera la validación.

Editar un agente existente

Workflow.from_agent() / Workflow.fromAgent() carga el flujo de trabajo de un agente devuelto por agents.get(). Se conservan los ID de nodo, los ID de arista, las posiciones del editor y el orden de las aristas, por lo que al volver a enviarlo solo se modifica lo que has cambiado.

insert_between() / insertBetween() pasa a tu función una rama que contiene la condición de la arista sustituida. La función devuelve el último nodo que añadió, que después se conecta al destino original. El método devuelve ese nodo, o None / undefined si no hay ninguna arista entre los dos nodos.

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

Guardar flujos de trabajo en el control de versiones

to_json_string() / toJSONString() escribe el flujo de trabajo en el formato de transmisión de la API: claves en snake_case, ordenadas en todos los niveles y con sangría de 2 espacios. Tiene la misma estructura que usan la API REST y la CLI, y ambos SDK escriben archivos idénticos byte a byte para el mismo flujo de trabajo, por lo que las diferencias se mantienen pequeñas. Workflow.from_json() / Workflow.fromJSON() lo vuelve a leer y también acepta una configuración de agente completa con una clave workflow, como un archivo del directorio agent_configs/ de la 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 configuración habitual:

Visualizar un flujo de trabajo

La CLI dibuja un archivo de flujo de trabajo como un diagrama de flujo para que puedas comprobar lo que ha creado tu código sin abrir el panel. Acepta un archivo de flujo de trabajo independiente escrito por to_json_string() / toJSONString(), o una configuración de agente con una clave workflow:

elevenlabs agents workflow diagram workflows/support.json

De forma predeterminada, dibuja cuadros y flechas en la terminal, con cada condición de arista numerada en su arista y escrita en una leyenda debajo. Usa --format mermaid para mostrar en su lugar el código fuente de Mermaid, o --format mermaid-md para envolverlo en un bloque de código delimitado que GitHub renderiza, por ejemplo, en un comentario de pull request:

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

El comando solo lee archivos locales, por lo que no necesita una clave de API y puede ejecutarse en CI.

Validación

Las llamadas encadenadas nunca fallan por sí solas. build(), to_json() / toJSON() y to_json_string() / toJSONString() comprueban todo el grafo y generan o lanzan un único WorkflowValidationError cuya lista issues enumera cada problema, cada uno con un code, un message y el node_id / nodeId al que se refiere.

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)

Los errores incluyen:

• Dos aristas entre el mismo par de nodos o una arista de un nodo hacia sí mismo.
• Una arista que entra en el nodo inicial o sale de un nodo final.
• Una arista desde un nodo de herramienta que no es de éxito ni de error, o una arista de éxito o error desde cualquier otro tipo de nodo.
• Dos nodos con el mismo ID o una arista hacia un nodo que no existe.

Los nodos a los que no se puede llegar desde el nodo inicial se notifican como advertencias, mediante on_warning / onWarning, y no detienen la creación.

Tipos de nodo más recientes que tu SDK

Un flujo de trabajo cargado desde la API o desde un archivo puede contener tipos de nodo que tu versión del SDK aún no conoce. Se conservan sin cambios al pasar por from_json() / fromJSON() y to_json_string() / toJSONString(), y build() informa de cada uno como una advertencia. El modelo de solicitud tipado no puede incluirlos, así que actualiza el SDK o envía directamente el formato de transmisión:

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