Graph

Graph provides deterministic, dependency-based orchestration. You define which agents execute in which order; the output of one agent becomes the input to the next. Build a graph with GraphBuilder, then call .build() to get a runnable Graph instance.

Graph execution

1. Build — Register agents as nodes and connect them with directed edges via GraphBuilder.
2. Resolve — Nodes with no incoming edges are entry points; they receive the initial prompt. Independent branches run in parallel.
3. Execute — Each node runs after its dependencies complete. A node's output is passed to downstream nodes.
4. Return — The framework aggregates per-node results, token usage, and timing into a GraphResult.

Usage

from elsai import Agent
from elsai.agent import AgentConfig
from elsai.multiagent import GraphBuilder

extractor = Agent(
    system_prompt="Extract key facts and data points from the given text.",
    config=AgentConfig(name="extractor"),
)
analyst = Agent(
    system_prompt="Analyse the extracted data and identify trends.",
    config=AgentConfig(name="analyst"),
)
reporter = Agent(
    system_prompt="Write a clear executive summary based on the analysis.",
    config=AgentConfig(name="reporter"),
)

builder = GraphBuilder()
builder.add_node(extractor, "extractor")
builder.add_node(analyst, "analyst")
builder.add_node(reporter, "reporter")
builder.add_edge("extractor", "analyst")
builder.add_edge("analyst", "reporter")

graph = builder.build()
result = graph("Analyse this quarterly sales data: [... data ...]")
print(result)

Parallel branches — nodes without dependencies on each other run concurrently:

builder = GraphBuilder()
builder.add_node(summariser, "summariser")
builder.add_node(sentiment_analyser, "sentiment")
builder.add_node(keyword_extractor, "keywords")
builder.add_node(report_writer, "report")

builder.add_edge("summariser", "report")
builder.add_edge("sentiment", "report")
builder.add_edge("keywords", "report")

graph = builder.build()

Multiple entry points — nodes with no incoming edges each receive the initial input; downstream nodes receive combined upstream outputs:

builder = GraphBuilder()
builder.add_node(agent_a, "a")
builder.add_node(agent_b, "b")
builder.add_node(agent_c, "c")
builder.add_edge("a", "c")
builder.add_edge("b", "c")

graph = builder.build()

Cyclic graphs — use cycles for iterative refinement; cap executions to avoid infinite loops:

builder = GraphBuilder()
builder.add_node(drafter, "drafter")
builder.add_node(critic, "critic")
builder.add_edge("drafter", "critic")
builder.add_edge("critic", "drafter")
builder.set_max_node_executions(3)

graph = builder.build()

Session persistence:

from elsai.session import FileSessionManager

builder = GraphBuilder()
# ... add nodes and edges ...
builder.set_session_manager(
    FileSessionManager(session_id="graph-run-1", storage_dir="./sessions")
)
graph = builder.build()

Hooks:

from elsai.hooks import BeforeNodeCallEvent, HookProvider

class NodeLogger(HookProvider):
    def register_hooks(self, registry):
        registry.add_callback(
            BeforeNodeCallEvent,
            lambda event: print(f"Starting node: {event.node_id}"),
        )

builder = GraphBuilder()
# ... add nodes and edges ...
builder.set_hook_providers([NodeLogger()])
graph = builder.build()

Async execution:

import asyncio

async def main():
    result = await graph.invoke_async("Process this data")
    print(result)

asyncio.run(main())

Configuration

Method	Description
add_node(executor, node_id=None)	Add an Agent (or MultiAgentBase) as a node. Returns GraphNode. Auto-generates node_id from the executor's id or name when omitted.
add_edge(source, target, condition=None)	Directed edge — target runs after source and receives its output. Pass node ID strings or GraphNode objects.
set_max_node_executions(n)	Cap total node executions (required for cyclic graphs).
set_session_manager(manager)	Persist graph state across invocations.
set_hook_providers(providers)	Register HookProvider instances for lifecycle events.
build()	Returns a runnable Graph instance.

GraphResult

result = graph("...")

print(result.status)                       # "completed" | "failed" | "interrupted"
print(result.execution_time)               # total execution time in ms
print(result.results)                      # dict[str, NodeResult] per node
print(result.accumulated_usage["totalTokens"])
print(result.accumulated_metrics["latencyMs"])

for node_id, node_result in result.results.items():
    print(f"{node_id}: {node_result.result}")

Each NodeResult exposes result, status, execution_time, and accumulated_usage.