Graph

Graph provides a deterministic, dependency-based orchestration pattern. You define which agents execute in which order, and the output of one agent becomes the input to the next.

Basic example

from elsai import Agent
from elsai.multiagent import Graph

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

# Build the graph
graph = Graph()
graph.add_node(extractor)
graph.add_node(analyst)
graph.add_node(reporter)

# Define execution order via edges
graph.add_edge(extractor, analyst)    # analyst runs after extractor
graph.add_edge(analyst, reporter)     # reporter runs after analyst

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

Parallel execution

Nodes without dependencies on each other run concurrently:

graph = Graph()
graph.add_node(summariser)
graph.add_node(sentiment_analyser)
graph.add_node(keyword_extractor)
graph.add_node(report_writer)

# All three run in parallel, then report_writer combines results
graph.add_edge(summariser, report_writer)
graph.add_edge(sentiment_analyser, report_writer)
graph.add_edge(keyword_extractor, report_writer)

result = graph("Analyse this customer feedback: ...")

Entry points

Nodes with no incoming edges are entry points and receive the initial input directly:

graph = Graph()
graph.add_node(agent_a)  # entry point → receives initial input
graph.add_node(agent_b)  # entry point → also receives initial input (parallel)
graph.add_node(agent_c)
graph.add_edge(agent_a, agent_c)
graph.add_edge(agent_b, agent_c)
# agent_c receives combined outputs from agent_a and agent_b

Accessing node results

result = graph("Do a complex task")

for node_name, node_result in result.node_results.items():
    print(f"{node_name}: {node_result.output}")

Graph state

Share state across all nodes in the graph:

from elsai.agent.state import AgentState

shared_state = AgentState({"session_id": "abc123", "budget": 1000})
graph = Graph(state=shared_state)

Hooks on graphs

from elsai.hooks import BeforeNodeCallEvent, AfterNodeCallEvent

def log_node(event: BeforeNodeCallEvent) -> None:
    print(f"Starting node: {event.node.name}")

graph = Graph(hooks=[log_node])

Async execution

import asyncio

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

asyncio.run(main())

Cyclic graphs

Graphs can contain cycles for iterative refinement:

graph = Graph()
graph.add_node(drafter)
graph.add_node(critic)

graph.add_edge(drafter, critic)
graph.add_edge(critic, drafter)  # critic sends feedback back to drafter

# Use a termination condition to avoid infinite loops
graph = Graph(max_iterations=3)

GraphResult

result = graph("...")

print(result.output)          # Final combined output
print(result.status)          # "completed" | "failed" | "interrupted"
print(result.execution_time)  # Total execution time in ms
print(result.node_results)    # Dict of node_name → NodeResult
print(result.metrics)         # Aggregated token usage