Graphs for agents and knowledge: state machines, DAGs and knowledge graphs
Three kinds of graphs show up in AI systems. An agent's control flow is a state machine: nodes are steps, edges are transitions, and loops are allowed, which is what LangGraph models. A plan of tasks with dependencies is a DAG, run in topological order so independent tasks can run in parallel, and a knowledge graph stores entities and their relationships so you can answer questions that need several hops.
State machines for agents
An agent loop is a graph with a cycle: plan, call a tool, observe, decide whether to continue. Writing it as explicit states and transitions gives you things a free loop doesn't: a maximum number of steps, a known place to pause for human approval, and a saved state you can resume from after a failure. See agent state and recovery.
DAGs for task plans
When a task splits into steps with dependencies, a DAG says what can run when. Kahn's algorithm processes nodes with no remaining dependencies, layer by layer, and detects cycles when nodes are left over.
from collections import deque
def run_dag(deps: dict[str, list[str]], tasks: dict) -> dict:
indegree = {node: len(parents) for node, parents in deps.items()}
children = {node: [] for node in deps}
for node, parents in deps.items():
for parent in parents:
children[parent].append(node)
ready = deque(sorted(n for n, d in indegree.items() if d == 0))
results, layers = {}, []
while ready:
layer = list(ready)
ready.clear()
layers.append(layer)
for node in layer:
results[node] = tasks[node]({p: results[p] for p in deps[node]})
for child in children[node]:
indegree[child] -= 1
if indegree[child] == 0:
ready.append(child)
if len(results) != len(deps):
raise ValueError(f"cycle among: {sorted(set(deps) - set(results))}")
print("layers:", layers)
return results
deps = {
"fetch_orders": [],
"fetch_tickets": [],
"summarise_orders": ["fetch_orders"],
"summarise_tickets": ["fetch_tickets"],
"report": ["summarise_orders", "summarise_tickets"],
}
tasks = {
"fetch_orders": lambda _: ["order 1", "order 2"],
"fetch_tickets": lambda _: ["ticket 9"],
"summarise_orders": lambda r: f"{len(r['fetch_orders'])} orders",
"summarise_tickets": lambda r: f"{len(r['fetch_tickets'])} tickets",
"report": lambda r: f"report: {r['summarise_orders']}, {r['summarise_tickets']}",
}
print(run_dag(deps, tasks)["report"])
try:
run_dag({"a": ["b"], "b": ["a"]}, {})
except ValueError as error:
print("error:", error)
The two fetches have no dependencies, so they're in the same layer and could run in parallel. In a real system, each task might be an LLM call or a tool call, and the executor is where you add retries, timeouts and logging per node.
Knowledge graphs
A knowledge graph stores triples: (Acme, supplies, Contoso), (Contoso, located in, Pune). A question like "which suppliers of our Pune customers had late deliveries?" needs several hops across relationships, which vector search over chunks handles poorly because no single chunk contains the answer. GraphRAG approaches extract entities and relationships with an LLM and query the graph alongside vector search. Building and maintaining the graph costs real effort, so start with it only when your questions need relationships. See GraphRAG.
Which graph you need
| Question | Structure |
|---|---|
| What happens next, and can it loop? | State machine |
| What can run now, given dependencies? | DAG with topological order |
| How are these entities related? | Knowledge graph |
| Which tool calls happened, in what order? | Trace (a tree of spans) |
Check your understanding
Add a step that depends on report and on fetch_orders, and predict which layer it lands in before running it.
From the floor
In the AI data structures lecture, I started with the graph as one of the classic structures everyone already knows, and the same structure is what agent frameworks are built on: when you design an agent as explicit steps and transitions instead of an open loop, you can see where it can fail and where a person needs to approve before it continues.