Designing Agent Graphs
In Hive, an agent is not a linear script but a Directed Graph. This architecture allows you to build complex, self-correcting workflows where nodes represent specific units of work (LLM calls, tool usage, or Python functions) and edges define the conditional logic that governs the flow of execution.
Graph Architecture
A Hive graph consists of three primary components:
- Nodes: Discrete processing steps.
- Edges: Directed connections between nodes.
- Context: A shared state object passed between nodes throughout the execution.
Node Types
Hive provides several specialized node types to handle different aspects of agentic behavior:
| Node Type | Purpose | Best Used For | | :--- | :--- | :--- | | LLMNode | Generates text or structured data using a language model. | Summarization, planning, or content generation. | | FunctionNode | Executes arbitrary Python code. | Data transformation, API calls, or state updates. | | RouterNode | Evaluates logic to decide which outgoing edge to follow. | Dynamic branching based on LLM output or data values. | | EventLoopNode | Iteratively executes a task until a specific condition (verdict) is met. | Autonomous research, coding tasks, or multi-step tool usage. | | HITLNode | Pauses execution to wait for human approval or input. | High-stakes actions like making a purchase or deleting data. |
Edge Logic and Conditions
Edges determine the path an agent takes. Every edge is associated with a Condition.
- Always: The transition occurs immediately after the source node completes successfully.
- On Success / On Failure: Transitions based on the completion status of a
FunctionNode. - Custom Conditions: Used primarily with
RouterNodes, where the edge is followed only if the router's logic matches a specific key (e.g.,"needs_research"or"ready_to_submit").
Defining a Graph
Graphs are typically defined within an Agent class using a builder pattern. The AgentRuntime then takes this graph and manages its lifecycle.
Example: A Research and Review Graph
from framework.runtime.graph import Graph, Node, Edge
from framework.runtime.agent_runtime import create_agent_runtime
def build_research_graph():
graph = Graph(id="research_workflow")
# 1. Define Nodes
graph.add_node(Node(id="intake", type="llm_generate"))
graph.add_node(Node(id="researcher", type="event_loop"))
graph.add_node(Node(id="reviewer", type="router"))
graph.add_node(Node(id="finalize", type="function"))
# 2. Define Edges with conditions
graph.add_edge(Edge(source="intake", target="researcher", condition="always"))
# The researcher loops until completion, then moves to reviewer
graph.add_edge(Edge(source="researcher", target="reviewer", condition="on_success"))
# Conditional branching from the Router
graph.add_edge(Edge(source="reviewer", target="researcher", condition="needs_more_info"))
graph.add_edge(Edge(source="reviewer", target="finalize", condition="approved"))
# 3. Set Entry and Terminal points
graph.entry_node = "intake"
graph.terminal_nodes = ["finalize"]
return graph
The Event Loop Node
The EventLoopNode is the core of Hive’s autonomous capabilities. Unlike a standard node that runs once, the Event Loop continues until it reaches one of the following verdicts:
- ACCEPT: The task is complete; move to the next node.
- RETRY: The last step failed or was incomplete; try again (often with feedback).
- ESCALATE: The agent cannot complete the task and requires intervention.
- CONTINUE: The task is in progress; proceed to the next iteration.
This node type is essential for agents that use tools, as it allows the agent to observe tool output and decide if more actions are required.
Execution and State Management
When a graph runs, it maintains an Execution Path. This path is tracked in real-time and can be visualized in the Hive TUI.
The Agent Context
Every node has access to the context. This is a dictionary where nodes store their results and retrieve inputs from previous steps.
# Inside a FunctionNode or Tool
async def process_data(context: dict):
raw_data = context.get("research_output")
# Process the data...
context["final_report"] = formatted_report
return True
Runtime Configuration
To execute a graph, you initialize the AgentRuntime. This requires the graph definition, a tool registry, and a provider for the LLM.
runtime = create_agent_runtime(
graph=my_graph,
goal="Conduct deep research on AI safety",
llm=llm_provider,
tools=list(tool_registry.get_tools().values()),
storage_path="./agent_data"
)
await runtime.start()
Best Practices for Graph Design
- Keep Nodes Atomic: Each node should do one thing well. If a node is both fetching data and formatting it, consider splitting it into a
FunctionNodeand anLLMNode. - Explicit Exit Conditions: Always ensure your
RouterNodeshave a default path or cover all possible output keys to prevent the graph from "stalling." - Use Terminal Nodes: Explicitly define
terminal_nodesso the runtime knows when to successfully conclude the session and generate the final summary logs. - Leverage HITL for High-Stakes Nodes: If your graph contains nodes that interact with external production systems, insert a
HITLNodebefore the execution to ensure safety.