Graph
Store the entities your agents see and how they relate, then walk, search, and read those relationships at any past time
The graph stores entities as nodes and relationships as edges, such as which host runs which service or which incident touched which component. You can walk relationships, start from the nodes closest to an embedding, and read the graph as it was at an earlier time. Build it with direct writes, or derive it from the messages on your log.
Quick example
This program records that a host runs two services, then reads the host's neighbors. It needs Laser Stack or LaserData Cloud.
import { Laser, graphNodeEntity } from "@laserdata/laser-sdk"
await using laser = await Laser.connectWithStream("iggy:[email protected]:8090", "ops")
const graph = laser.graph("kg")
for (const service of ["service:auth", "service:metrics"]) {
await graph.link("host:node-7", "runs", service)
}
const host = graphNodeEntity("host", "node-7")
const result = await graph.neighbors(host.id, "out", "runs", 1)
for (const node of result.nodes) {
for (const [key, value] of node.attrs) {
if (key === "value" && value.kind === "str") console.log(`${node.labels[0]}:${value.value}`)
}
}use laser_sdk::prelude::full::*;
#[tokio::main]
async fn main() -> Result<(), LaserError> {
let laser = Laser::connect_with_stream("iggy:[email protected]:8090", "ops").await?;
for service in ["service:auth", "service:metrics"] {
laser.graph("kg").link("host:node-7", "runs", service).await?;
}
let host = GraphNode::entity("host", "node-7").id;
let result = laser
.graph("kg")
.neighbors(host, EdgeDir::Out, Some("runs".to_owned()), 1)
.await?;
for node in &result.nodes {
let label = node.labels.first().map_or("entity", String::as_str);
if let Some((_, Value::Str(value))) = node.attrs.iter().find(|(key, _)| key == "value") {
println!("{label}:{value}");
}
}
Ok(())
}import asyncio
import laser_sdk as ls
async def main():
async with await ls.Laser.connect("iggy:[email protected]:8090", stream="ops") as laser:
graph = laser.graph("kg")
for service in ("service:auth", "service:metrics"):
await graph.link("host:node-7", "runs", service)
host_id = ls.node_id_content("host", "node-7")
result = await graph.neighbors(host_id, "out", "runs", 1)
for node in result.get("nodes", []):
print(f"{node['labels'][0]}:{dict(node['attrs']).get('value')}")
asyncio.run(main())link takes "label:value" strings and writes both nodes and the edge. A node's ID comes from its label and value, so every agent that links host:node-7 updates the same node, and repeating a link changes nothing. The reply includes the start node.
Change a fact
relink(from, relation, to) replaces a relationship that has one current value, such as which host a service runs on. It closes the old edges and writes the new one. unlink(from, relation, to) closes one edge. Closed edges stay in the history.
const closed = await laser.graph("kg").relink("service:auth", "runs_on", "host:node-9")let closed = laser.graph("kg").relink("service:auth", "runs_on", "host:node-9").await?;closed = await laser.graph("kg").relink("service:auth", "runs_on", "host:node-9")Walk several hops
The traversal builder picks a start, adds one hop per call, and chooses what to return. Start from node IDs with start_ids, from nodes that match a filter with start_match, or from the nodes closest to an embedding with start_nearest(embedding, k).
const incident = graphNodeEntity("incident", "INC-101")
const paths = await laser.graph("kg").startIds([incident.id]).out("affected").out("runs_on").returnPaths().fetch()let incident = GraphNode::entity("incident", "INC-101").id;
let paths = laser
.graph("kg")
.start_ids(vec![incident])
.out("affected")
.out("runs_on")
.return_paths()
.fetch()
.await?;incident_id = ls.node_id_content("incident", "INC-101")
paths = await (
laser.graph("kg").start_ids([incident_id]).out("affected").out("runs_on").return_paths().fetch()
)Read the graph at a past time
Add as_of(micros) (TypeScript: asOf, as a bigint) before a read to see only the edges that were valid at that time, in epoch microseconds. An edge closed by unlink or relink still shows up in a read from before it closed.
Build the graph from messages
Register a graph projection with an entity schema, and bind it to a topic. The deployment reads each record, extracts nodes and edges with JSON pointers, and writes them, so your services need no extraction code. Graph in depth shows the schema. Inside a session, linked_graph(name) (TypeScript: linkedGraph) records which session and agent wrote each fact.
Good to know
- The graph needs Laser Stack or LaserData Cloud. Check
capabilities.graph. Against Apache Iggy alone, every graph call fails with an unsupported error. - A traversal goes at most 8 hops, a reply holds at most 10,000 nodes and edges, and the default limit is 100.
- Reads are eventually consistent, so a write can take a moment to show up in a traversal.
- Graph names are scoped to the default stream, so
laser.graph("kg")addressesstream:<stream>/kg. - In Rust, each graph call consumes its handle, so call
laser.graph(name)for each operation. Rust also needs thegraphfeature.