Agent Graph Reference
Out of scope for the main migration workflow. Read this only after a single-agent migration works end-to-end. The main
SKILL.mdworkflow stops at single-agent because multi-agent orchestration is a meaningful jump in complexity and is still evolving in the SDK.
Python is still the richer surface.
launchdarkly-server-sdk-ai(Python) has the fully-documented graph API used in the traversal pattern below.@launchdarkly/server-sdk-ai(Node) exposes Agent Graph Definitions and graph metric tracking — consult the js-core source for the current Node API shape before wiring Node graph code; the Python pattern in this doc is canonical.
What an agent graph is
An agent graph is a directed graph where each node is its own config (with its own instructions, model, parameters, and tools) and each edge carries routing metadata for handoffs. A supervisor node routes incoming requests to worker nodes based on the supervisor's output; worker nodes may themselves route to other workers or terminate. The graph lives in LaunchDarkly — both its topology and each node's config are managed as versioned resources and can be changed at runtime without redeploying.
Why use it:
- Add or remove an agent by editing LaunchDarkly — no redeploy
- A/B test routing strategies — target different graph topologies to different users
- Roll out a new worker node to a percentage of traffic
- Guardrail a specific path — attach a judge at a terminal node
SDK surface (Python, from main)
The current Python API (verified against launchdarkly-server-sdk-ai main branch) exposes these methods on LDAIClient:
def agent_graph(self, key: str, context: Context) -> AgentGraphDefinition:
"""Retrieve an agent graph by key."""
async def create_agent_graph(
self,
key: str,
context: Context,
tools: Optional[ToolRegistry] = None,
default_ai_provider: Optional[str] = None,
) -> Optional[ManagedAgentGraph]:
"""Experimental — not production-ready. Returns a managed graph that can be invoked directly."""Use agent_graph for read-only traversal (you drive the loop). create_agent_graph + ManagedAgentGraph.run is experimental and carries explicit production-not-ready warnings in the SDK source. Stick with agent_graph for now.
AgentGraphDefinition
graph_def: AgentGraphDefinition = ai_client.agent_graph("support-flow", context)
graph_def.is_enabled() -> bool
graph_def.root() -> Optional[AgentGraphNode]
graph_def.traverse(fn, execution_context=None) # callback over nodes from root
graph_def.reverse_traverse(fn, execution_context=None) # callback over nodes from terminals
graph_def.get_node(key: str) -> Optional[AgentGraphNode]
graph_def.get_child_nodes(node_key: str) -> List[AgentGraphNode]
graph_def.get_parent_nodes(node_key: str) -> List[AgentGraphNode]
graph_def.terminal_nodes() -> List[AgentGraphNode]
graph_def.get_tracker() -> Optional[AIGraphTracker]AgentGraphNode
node.get_key() -> str
node.get_config() -> AIAgentConfig # the same shape as agent_config() returns
node.get_edges() -> List[Edge]
node.is_terminal() -> boolEdge
@dataclass
class Edge:
key: str
source_config: str
target_config: str
handoff: Optional[dict] # arbitrary dict; typically has a 'route' keyAIGraphTracker
tracker.track_invocation_success() -> None
tracker.track_invocation_failure() -> None
tracker.track_duration(duration: int) -> None # milliseconds, graph-level total
tracker.track_total_tokens(tokens: TokenUsage) -> None
tracker.track_path(path: List[str]) -> None # e.g. ["supervisor", "security", "support"]
tracker.track_redirect(source_key: str, redirected_target: str) -> None
tracker.track_handoff_success(source_key: str, target_key: str) -> None
tracker.track_handoff_failure(source_key: str, target_key: str) -> NoneThings that are NOT on the graph tracker:
track_node_invocation— not a public method. Usetrack_path(execution_path)at the end of traversal instead.track_tool_call(node_key, tool_name)— graph-level tool-call tracking does not exist. Track per-node tool calls vianode_tracker.track_tool_call(tool_name)on each node's tracker (obtained vianode.get_config().create_tracker()). Trackers returned via a graph traversal are automatically bound to the right graph key — do not passgraph_keyas a keyword.track_judge_response— does not exist onAIGraphTracker. Record judge results at the config level viaLDAIConfigTracker.track_judge_result(result)instead.- No
track_request(), notrack_duration()per call — usetrack_duration(total_ms)once per traversal.
If you see older devrel-agents-tutorial code that calls track_node_invocation, track_tool_call, or pokes graph_tracker._ld_client.track(...) directly, that code targets an earlier API shape and needs updating. A PR is in flight against launchdarkly-labs/devrel-agents-tutorial to align the tutorial with the current SDK.
Canonical traversal pattern
from ldai.client import LDAIClient
from ldai.tracker import TokenUsage
async def execute_graph(ai_client: LDAIClient, graph_key: str, context, user_input: str):
graph = ai_client.agent_graph(graph_key, context)
if not graph.is_enabled():
raise ValueError(f"Agent graph '{graph_key}' is not enabled")
# Build a lookup from node key to AgentGraphNode so we can follow edges.
nodes: dict[str, object] = {}
graph.reverse_traverse(lambda node, _: nodes.update({node.get_key(): node}), {})
graph_tracker = graph.get_tracker()
start = time.time()
execution_path = []
shared_ctx = {"user_input": user_input, "final_response": "", "tool_calls": [],
"total_input_tokens": 0, "total_output_tokens": 0}
current_node = graph.root()
if not current_node:
raise ValueError("Graph has no root node")
prev_node_key = None
visited = set()
MAX_HOPS = 10
hop_count = 0
try:
while current_node:
node_key = current_node.get_key()
if node_key in visited:
raise ValueError(f"Cycle detected at {node_key}")
visited.add(node_key)
hop_count += 1
if hop_count > MAX_HOPS:
raise ValueError(f"Max hops exceeded: {hop_count}")
config = current_node.get_config()
execution_path.append(node_key)
# Track handoff into this node
if graph_tracker and prev_node_key:
graph_tracker.track_handoff_success(prev_node_key, node_key)
# Compute valid routes from outgoing edges
edges = current_node.get_edges()
valid_routes = [
(edge.handoff or {}).get("route")
for edge in edges
if (edge.handoff or {}).get("route")
]
# Execute this node — uses your existing agent-mode wiring
result = await run_node(config, shared_ctx, valid_routes=valid_routes)
# Per-node tool-call tracking lives on the node's config tracker.
# Create one tracker per node execution (fresh runId) and reuse it
# for every tracking call inside that node.
if result.get("tool_calls"):
node_tracker = config.create_tracker()
for tool_name in result["tool_calls"]:
node_tracker.track_tool_call(tool_name)
# Merge node result into shared context
update_shared_ctx(shared_ctx, result)
# Terminal?
if not edges or current_node.is_terminal():
break
# Pick next node by matching the node's routing_decision to an edge handoff
next_node = select_next_node(edges, result, nodes, graph_tracker, source_key=node_key)
prev_node_key = node_key
current_node = next_node
# Graph-level metrics
if graph_tracker:
graph_tracker.track_path(execution_path)
graph_tracker.track_duration(int((time.time() - start) * 1000))
if shared_ctx["total_input_tokens"] or shared_ctx["total_output_tokens"]:
graph_tracker.track_total_tokens(TokenUsage(
input=shared_ctx["total_input_tokens"],
output=shared_ctx["total_output_tokens"],
total=shared_ctx["total_input_tokens"] + shared_ctx["total_output_tokens"],
))
graph_tracker.track_invocation_success()
except Exception:
if graph_tracker:
graph_tracker.track_invocation_failure()
raise
return shared_ctx
def select_next_node(edges, result, nodes, graph_tracker, source_key: str):
routing = result.get("routing_decision", "").lower().strip() if result.get("routing_decision") else None
route_map = {
(edge.handoff or {}).get("route", "").lower().strip(): edge.target_config
for edge in edges
if (edge.handoff or {}).get("route")
}
if routing and routing in route_map:
return nodes.get(route_map[routing])
if routing:
# Unrecognized route — signal failure with source + attempted target
if graph_tracker:
graph_tracker.track_handoff_failure(source_key, routing)
# Fallback: first edge
if edges:
return nodes.get(edges[0].target_config)
return NoneMigrating a multi-agent app to graphs
Do this in phases, not one big bang:
- Pick one worker node to migrate first. Use the single-agent skill workflow on that worker in isolation — extract, wrap, tools, tracking, evals. Leave the rest of the multi-agent app hardcoded.
- Confirm the wrapped worker runs in production for the traffic it serves today, with metrics flowing in the Monitoring tab.
- Migrate the supervisor the same way — single-agent workflow — but keep its routing logic hardcoded initially (a big if/elif over the other workers).
- Create the agent graph in LaunchDarkly via the UI. Define nodes (one per worker + supervisor) and edges (with
handoff.routemetadata). - Replace the hardcoded router with the traversal pattern above. Call
ai_client.agent_graph(...)instead of assembling the pipeline by hand. - Verify the Monitoring tab shows the graph-level metrics (
track_path,track_duration,track_total_tokens, handoff success/failure counts) in addition to the per-node metrics. - Only then start moving routing decisions into LaunchDarkly edges and using targeting to change the graph topology per user segment.
Each phase is reversible. If something breaks at phase 5, the supervisor can fall back to the hardcoded router while the graph issue is fixed.
Limitations to know about
- Python has the canonical surface. The Python traversal pattern above is what this doc covers in full. For Node graphs, consult the
@launchdarkly/server-sdk-aisource for the current API. create_agent_graphis experimental. Do not build production features onManagedAgentGraph.run. Use the traversal pattern above.- Graph tracker is less granular than the config tracker. If you want per-node duration or per-node token breakdowns, obtain a per-node tracker via
node.get_config().create_tracker()— the graph tracker handles totals only. - Cycles must be caught in your code. The SDK does not stop cycle traversal automatically; track
visitedandhop_countyourself. - Fallback shape. There is no
AIAgentGraphDefault. Each node'sAIAgentConfigstill takes anAIAgentConfigDefault, but the graph itself has no aggregate fallback. Ifagent_graphfails, handle it at the app level — typically by falling back to the hardcoded pre-migration pipeline.
Resources
- Python SDK source: https://github.com/launchdarkly/python-server-sdk-ai
packages/sdk/server-ai/src/ldai/agent_graph/__init__.py—AgentGraphDefinitionandAgentGraphNodepackages/sdk/server-ai/src/ldai/tracker.py—AIGraphTracker(near the bottom of the file)packages/sdk/server-ai/src/ldai/client.py—LDAIClient.agent_graphandcreate_agent_graph
- Node SDK source: https://github.com/launchdarkly/js-core/tree/main/packages/sdk/server-ai
- SDK CHANGELOGs (for per-release breaking changes and the version each method landed in):
- Devrel reference implementation (Python, after PR alignment): https://github.com/launchdarkly-labs/devrel-agents-tutorial on the
tutorial/agent-graphsbranch